Reference¶

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

This page shows the complete CREATE PIPE syntax with every in-flight transformation enabled, and a full example that combines them with an insert-triggered materialized view. For a first pipe, start with Pipe Creation.

Full syntax¶

CREATE [ OR REPLACE ] PIPE <pipe_name>
  AS COPY INTO <target_table> ( <col1> [, <col2>, ...] )
  FROM (
    SELECT <select_list>                                        -- (1)
    FROM
      [ ( SELECT <inner_select_list>                            -- (2)
          FROM ]
            TABLE(DATA_SOURCE(TYPE => 'STREAMING')) <stream_alias>
      [     , LATERAL FLATTEN( input => <array_expr> [ , OUTER => TRUE ] ) <flat_alias>
        ) <source_alias> ]
      [ LEFT OUTER JOIN <dim_table> [ <alias> ] ON <join_condition> ] ...                                                             -- (3)
      [ , LATERAL FLATTEN( input => <dim_array_expr> [ , OUTER => TRUE ] ) <flat_alias2> ]                                             -- (4)
    [ WHERE <filter_predicate> ]                                -- (5)
  );

(1) <select_list> may call scalar UDFs — SQL, JavaScript, SECURE, IMMUTABLE, and MEMOIZABLE — including a UDF that returns an ARRAY, which (2) or (4) can then flatten. When the derived table in (2) is present, <select_list> can reference only the columns that the inner SELECT projects, addressed through <source_alias>.

(2) Optional derived table, used to expand a streamed record into multiple rows before the dimension join. The bracketed lines form a single unit: either include all of them — the opening ( SELECT <inner_select_list> FROM, the LATERAL FLATTEN line, and the closing ) <source_alias> — or omit all of them and select directly from TABLE(DATA_SOURCE(TYPE => 'STREAMING')) <stream_alias>.

(3) The LEFT OUTER JOIN clause may be repeated once for each dimension table. A maximum of 4 dimension tables per PIPE are supported. Dimension tables are initially limited to approximately 1 GB per table, ramping to approximately 10 GB per table.

(4) LATERAL FLATTEN after the dimension join, typically over an array column of a dimension table.

(5) Records that don’t satisfy the predicate are silently discarded before they land. The predicate can reference stream columns (positional $n notation), columns of the joined dimension tables, flattened values, and scalar UDFs. It can’t contain subqueries, Python, Java, and Scala language handlers, along with UDAFs, UDTFs, external functions, and Cortex AI functions, window functions, or FLATTEN. See Stream filtering.

Parameters:

<target_table> — Required name of the destination table. Interactive tables, standard Snowflake tables, and managed Iceberg tables are currently supported. ERROR_LOGGING must not be enabled on the target table of a PIPE that uses LATERAL FLATTEN. See Limitations.
<select_list> — Expressions written to the target columns, in the same order as the column list. May call scalar UDFs, including a UDF returning an ARRAY that is then flattened. See footnote (1).
<inner_select_list> — Expressions projected out of the derived table when flattening before a dimension join. Must include every column referenced outside it, including the column used in <join_condition>.
<stream_alias> — Required alias for the streaming source when dimension joins or LATERAL FLATTEN are present. Used to qualify stream column references in ON and SELECT clauses (for example, s.$1:user_id::INT).
<array_expr> — Expression yielding the array to expand, read from the streamed record. A path such as s.$1:items, or a UDF returning an ARRAY.
<flat_alias> — Alias for the FLATTEN result. Used to qualify its columns (VALUE, INDEX, KEY, PATH, THIS, SEQ), for example f.value::INT. Inside a derived table it is not visible outside it.
<source_alias> — Required alias for the derived table when flattening before a dimension join. Qualifies the expanded rows in the ON, SELECT, and WHERE clauses.
<dim_table> — Fully qualified name of the dimension (lookup) table. See Dimension table requirements.
<join_condition> — Pure (fully qualified) equi-join on the dimension table primary key. See ON clause requirements.
<dim_array_expr> — Expression yielding the array to expand after the join, typically an array column of the dimension table.
<flat_alias2> — Alias for the FLATTEN result after the join.
<filter_predicate> — Boolean expression applied to stream records before enrichment, or to the expanded rows when it references flattened values. See WHERE clause syntax.

Note

An explicit column list is required in both the COPY INTO target and the SELECT subquery. MATCH_BY_COLUMN_NAME is not supported when dimension joins are present.

Full example¶

The following example combines multiple features on one stream of batched device messages:

A SQL scalar UDF (site_timezone) in the SELECT list.
LATERAL FLATTEN of the readings array in a derived table, before the dimension join.
A LEFT OUTER JOIN with a dimension table (dim_tags) to add a readable name and unit.
A WHERE clause on a stream column and on a dimension key.
An insert-triggered materialized view (readings_per_tag) on the target table, created before the pipe.

-- Dimension table. A primary key with RELY is required.
CREATE OR REPLACE TABLE dim_tags (
  tag_id   STRING PRIMARY KEY RELY,
  tag_name STRING,
  unit     STRING
);

INSERT INTO dim_tags VALUES
  ('T100', 'spindle_temperature', 'degC'),
  ('T200', 'motor_current', 'A');

-- Scalar UDF used in the SELECT list.
CREATE OR REPLACE FUNCTION site_timezone(site STRING)
  RETURNS STRING
  IMMUTABLE
AS $$
  CASE site
    WHEN 'ber' THEN 'Europe/Berlin'
    WHEN 'sfo' THEN 'America/Los_Angeles'
    ELSE 'UTC'
  END
$$;

-- Target table.
CREATE OR REPLACE TABLE readings (
  device_id     STRING,
  local_ts      TIMESTAMP_TZ,
  tag_id        STRING,
  tag_name      STRING,
  unit          STRING,
  reading_value FLOAT
);

-- Insert-triggered materialized view. Create it before the pipe.
CREATE OR REPLACE MATERIALIZED VIEW readings_per_tag
  REFRESH USING INSERT ONLY
  AS
  SELECT tag_name, SUM(reading_value) AS total_value,
        COUNT(*) AS reading_count
  FROM readings
  GROUP BY tag_name;

-- Pipe with UDF, flatten, join and filter.
CREATE OR REPLACE PIPE readings_pipe
  AS COPY INTO readings (device_id, local_ts, tag_id, tag_name, unit, reading_value)
  FROM (
    SELECT
      s.device_id,
      CONVERT_TIMEZONE(site_timezone(s.site), s.event_ts),
      s.tag_id,
      t.tag_name,
      t.unit,
      s.reading_value
    FROM (
      SELECT
        e.$1:device_id::STRING AS device_id,
        e.$1:site::STRING      AS site,
        e.$1:ts::TIMESTAMP_NTZ AS event_ts,
        r.value:tag::STRING    AS tag_id,
        r.value:v::FLOAT       AS reading_value
      FROM TABLE(DATA_SOURCE(TYPE => 'STREAMING')) e,
           LATERAL FLATTEN(input => e.$1:readings) r) s
    LEFT OUTER JOIN dim_tags t
      ON s.tag_id = t.tag_id
    WHERE t.tag_id IS NOT NULL AND s.reading_value >= 0);

Each streamed record carries an array of readings:

{
  "device_id": "press-01",
  "site": "ber",
  "ts": "2026-10-09 08:00:00",
  "readings": [
    {"tag": "T100", "v": 71.4},
    {"tag": "T200", "v": 12.8},
    {"tag": "T999", "v": 3.0}
  ]
}

The pipe lands one row per known tag with its name, unit, and local time, and the view keeps the totals per tag up to date. The reading for the unknown tag T999 is filtered out.

Reference¶

Full syntax¶

Full example¶

See also¶