Stream filtering¶

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

A WHERE clause filters records from the stream before they are written to the target table. Records that do not satisfy the predicate are silently discarded and do not appear in load history.

WHERE clause syntax¶

The predicate may combine any of the following:

Element	Supported syntax
Comparison	`=`, `!=` (`<>`), `<`, `<=`, `>`, `>=`
Range	`BETWEEN <expr> AND <expr>`
Membership	`IN ( <value_list> )`
Null check	`IS NULL`, `IS NOT NULL`
Pattern match	`LIKE`, `ILIKE`
Logical	`AND`, `OR`, `NOT`
Built-in scalar functions	Yes (e.g., `UPPER`, `TRY_CAST`, `IFF`, `COALESCE`)
Metadata columns	Yes (e.g., `METADATA$FILENAME`, `METADATA$START_SCAN_TIME`)
Flattened element columns	Yes (e.g., `f.value`, when a `LATERAL FLATTEN` is present in the `FROM` clause)

Stream column references in the WHERE clause use positional $n notation (for example, s.$1:field_name) or a column alias defined in the SELECT list. Predicates may also reference columns from dimension tables joined in the same subquery.

Not supported in WHERE:

Subqueries
Python, Java, and Scala language handlers, along with UDAFs, UDTFs, external functions, and Cortex AI functions
Window functions
FLATTEN in the predicate itself

Example — Filter by a stream field value¶

Ingest only records where the region is 'EU':

CREATE OR REPLACE PIPE eu_orders_pipe
  AS COPY INTO orders (order_id, customer_id, amount)
  FROM (
    SELECT
      $1:order_id::STRING,
      $1:customer_id::STRING,
      $1:amount::NUMBER
    FROM TABLE(DATA_SOURCE(TYPE => 'STREAMING'))
    WHERE $1:region::STRING = 'EU'
  );

Example — Compound filter with a metadata column¶

Ingest only click and purchase events that arrive on or after a specific date:

CREATE OR REPLACE PIPE recent_events_pipe
  AS COPY INTO events (ts, event_type, payload)
  FROM (
    SELECT
      $1:ts::TIMESTAMP_NTZ,
      $1:event_type::STRING,
      $1
    FROM TABLE(DATA_SOURCE(TYPE => 'STREAMING'))
    WHERE $1:event_type::STRING IN ('click', 'purchase')
      AND METADATA$START_SCAN_TIME >= '2025-01-01'::TIMESTAMP_NTZ
  );

Usage notes¶

The per-channel offset token advances for every processed row, including rows excluded by the WHERE predicate. This may be surprising because get_latest_committed_offset_token(), the SDK method used to retrieve it, predates Streaming Transformations (API reference).