Lateral Flatten¶

Part of Streaming Transformations. See all pages in this section.

A PIPE transformation can expand a single streamed record into multiple rows by flattening an array inside it. A record carrying an array of three elements lands as three rows in the target table, one per element. This is useful when a producer batches several readings, line items, or events into one record but the target table has one row per reading.

LATERAL FLATTEN is supported both before and after any dimension join, and both may be used in the same pipe.

After the dimension join. Add the flatten to the FROM clause following any joins. This is the simpler form and is also what you use when the PIPE has no dimension join at all.

CREATE OR REPLACE PIPE readings_pipe
  AS COPY INTO readings (device_id, reading)
  FROM (
    SELECT s.$1:device_id::STRING,
           f.value:reading::FLOAT
    FROM TABLE(DATA_SOURCE(TYPE => 'STREAMING')) s
       , LATERAL FLATTEN(input => s.$1:signals) f
  );

Before the dimension join. When the array supplies the key you join on, the flatten must happen first. Wrap the source and the flatten in a derived table, then join to that.

CREATE OR REPLACE PIPE readings_pipe
  AS COPY INTO readings (device_id, category_name, reading)
  FROM (
    SELECT x.device_id,
           c.category_name,
           x.reading
    FROM ( SELECT s.$1:device_id::STRING  AS device_id,
                  f.value:category_id::INT AS category_id,
                  f.value:reading::FLOAT   AS reading
           FROM TABLE(DATA_SOURCE(TYPE => 'STREAMING')) s
              , LATERAL FLATTEN(input => s.$1:signals) f
         ) x
         LEFT OUTER JOIN dim_category c ON x.category_id = c.category_id
  );

The derived table is required in this position. Writing the flatten directly ahead of the join is rejected.

Note that the aliases inside a derived table are local to it. In the example above, s and f cannot be referenced from the outer SELECT, the ON clause, or the WHERE clause; only the columns projected by the inner SELECT, qualified with x, are visible. Any column used in the join condition must therefore be projected out of the derived table.

Empty, absent, and NULL arrays¶

By default, FLATTEN produces no rows for an input that is an empty array, is NULL, or is absent from the record. Such a record is not written to the target table, no error is raised, and no row is added to an error table. The record is silently dropped.

Specify OUTER => TRUE to keep it instead. The record then produces one row, with the flattened columns set to NULL:

-- OUTER => TRUE keeps records with an empty, NULL, or absent array
CREATE OR REPLACE PIPE readings_pipe
  AS COPY INTO readings (device_id, reading)
  FROM (
    SELECT s.$1:device_id::STRING,
           f.value:reading::FLOAT
    FROM TABLE(DATA_SOURCE(TYPE => 'STREAMING')) s
       , LATERAL FLATTEN(input => s.$1:signals, OUTER => TRUE) f
);

OUTER => TRUE is to FLATTEN what LEFT OUTER is to a join: it preserves the input row when there is nothing to match.

This is particularly important for a flatten placed after a LEFT OUTER JOIN over an array column of the dimension table. A record that matches no dimension row has a NULL array to flatten, so with the default the record is discarded — which defeats the purpose of the outer join:

-- OUTER => TRUE keeps records that matched no dimension row
CREATE OR REPLACE PIPE tagged_events_pipe
  AS COPY INTO tagged_events (device_id, category_name, tag)
  FROM (
    SELECT s.$1:device_id::STRING,
           c.category_name,
           t.value::STRING
    FROM TABLE(DATA_SOURCE(TYPE => 'STREAMING')) s
         LEFT OUTER JOIN dim_category c ON s.$1:category_id::INT = c.category_id
   , LATERAL FLATTEN(input => c.tags, OUTER => TRUE) t
  );

Decide per flatten which behavior you want. Dropping a record that carries no array elements is often correct; dropping a record because its dimension lookup failed usually is not.

Filtering flattened rows¶

A WHERE clause may filter on flattened values, and is applied after expansion:

CREATE OR REPLACE PIPE sensor_tags_pipe
  AS COPY INTO sensor_tags (device_id, tag)
  FROM (
    SELECT s.$1:device_id::STRING,
           t.value::STRING
    FROM TABLE(DATA_SOURCE(TYPE => 'STREAMING')) s
       , LATERAL FLATTEN(input => s.$1:tags) t
    WHERE t.value::STRING = 'sensor'  -- applied after expansion, to each element
  );

Each element that does not satisfy the predicate is removed individually, so a record can land fewer rows than its array has elements, or none at all.

Sizing¶

Flattening multiplies the rows written per record. A PIPE flattening a 50-element array writes 50 times the rows of a non-expanding pipe at the same record rate, and two flattens in one pipe multiply. Take the expansion factor into account when sizing throughput and when interpreting row counts, which will not match the number of records streamed.

Usage notes¶

With LATERAL FLATTEN, rows_parsed in the SDK function getChannelStatus() counts rows after expansion, so it equals rows_inserted rather than the number of records read.

Limitations¶

FLATTEN currently cannot be combined with ERROR_LOGGING on the target table.
A flatten before a dimension join must be wrapped in a derived table, as described above.
LATERAL FLATTEN counts as a lateral join, not a dimension table. It does not count toward the limit of four dimension tables, so a pipe may join the maximum number of dimension tables and still expand rows.
Transformation pipes cannot be altered. To add or remove a flatten, drop and recreate the pipe.