Stream enrichment¶

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

LEFT OUTER JOIN clauses join the incoming stream to dimension tables, annotating each ingested record with reference data — for example, a customer tier or a product category — without preprocessing in the upstream source.

Dimension table data is loaded into a per-pipe snapshot and refreshed periodically during active ingestion. Updates to a dimension table may be reflected with a delay of up to several minutes.

Dimension table requirements¶

The following requirements apply to each dimension table. Snowflake validates these conditions at pipe creation time; CREATE PIPE fails if they are not met.

Declaring a primary key with RELY¶

Primary key constraints on standard Snowflake tables are not enforced or maintained by Snowflake. The RELY property specifies that the constraint should be taken into account during query rewriting. Snowpipe Streaming extends this: RELY is also used at PIPE creation time to validate that a dimension table is a suitable join target, and at ingest time to guide the enrichment lookup.

If you have ensured that the data in the dimension table complies with the PRIMARY KEY constraint, set RELY to indicate that Snowflake should expect such data integrity. For standard tables, it is your responsibility to enforce RELY constraints; otherwise, you might risk unintended behavior and unexpected results. If the RELY property is set and the primary key values in the dimension table are not unique, enrichment results for affected records are undefined.

If the dimension table does not already have a primary key declared with RELY, add one before creating the PIPE:

-- Option A: primary key at table creation
CREATE TABLE customers (
  customer_id   STRING PRIMARY KEY RELY,
  customer_name STRING,
  tier          STRING
);

-- Option B: add a primary key to an existing table
ALTER TABLE customers
  ADD PRIMARY KEY (customer_id) RELY;

-- Option C: add a named primary key constraint to an existing table
ALTER TABLE customers
  ADD CONSTRAINT pk_customers PRIMARY KEY (customer_id) RELY;

ON clause requirements¶

Treat each dimension join as a key-value lookup: the ON clause matches on the dimension table’s primary key and returns at most one matching row per stream record.

1. The ON clause must be a pure equi-join that exactly covers the dimension table primary key: <stream_col> = <dim_pk_col>.
2. Composite primary keys are supported by combining equality conditions with AND: ON s.$1:k1::STRING = d.dim_pk_col1 AND s.$1:k2::INT = d.dim_pk_col2.
3. Stream column references use the positional $n notation qualified by the stream alias (e.g. s.$1:customer_id).
4. Extra predicates are not allowed in the ON clause. Additional conditions in the ON clause involving dimension or stream columns beyond the PK equality are rejected.
5. Expressions, scalar functions, and subqueries are not allowed in the ON clause.

JOIN semantics¶

LEFT OUTER JOIN semantics apply:

1. The ON clause must exactly cover the dimension table’s primary key.
2. If a matching row exists in the dimension table snapshot, its columns are added to the output record.
3. If no matching row exists, dimension table columns for that record are set to NULL. The record still ingests; it is not dropped. Use COALESCE in the SELECT list to substitute a default value.
4. Dimension table data is loaded from a per-pipe snapshot. The snapshot is refreshed periodically while the PIPE is actively ingesting. Staleness may be up to several minutes relative to the current state of the dimension table.
5. Snapshot refresh does not occur when the pipe is idle (no ingestion traffic for approximately 1.5 minutes). The snapshot is refreshed again when ingestion resumes.

SELECT list for enrichment¶

Example — Single dimension join (customer tier lookup)¶

-- Dimension table (PRIMARY KEY RELY required)
CREATE TABLE mydb.ref.customers (
  customer_id   STRING PRIMARY KEY RELY,
  customer_name STRING,
  tier          STRING
);

-- Pipe: enrich each order record with customer tier
CREATE OR REPLACE PIPE mydb.ingest.enriched_orders_pipe
  AS COPY INTO mydb.sales.orders_enriched (order_id, customer_id, amount, customer_tier)
  FROM (
    SELECT
      s.$1:order_id::STRING,
      s.$1:customer_id::STRING,
      s.$1:amount::NUMBER,
      COALESCE(c.tier, 'UNKNOWN') AS customer_tier
    FROM TABLE(DATA_SOURCE(TYPE => 'STREAMING')) s
    LEFT OUTER JOIN mydb.ref.customers c
      ON s.$1:customer_id::STRING = c.customer_id
  );

Example — Multiple dimension joins¶

Enrich each event with user name and city name from separate dimension tables:

CREATE OR REPLACE PIPE enriched_events_pipe
  AS COPY INTO enriched_events (event_id, user_id, city_id, user_name, city)
  FROM (
    SELECT
      e.$1:event_id::INT,
      e.$1:user_id::INT,
      e.$1:city_id::INT,
      u.user_name,
      c.city
    FROM TABLE(DATA_SOURCE(TYPE => 'STREAMING')) e
    LEFT OUTER JOIN dim_users  u ON e.$1:user_id::INT  = u.user_id
    LEFT OUTER JOIN dim_cities c ON e.$1:city_id::INT  = c.city_id
  );

Example — Combined filter and enrichment¶

Ingest only high-value orders from EU customers and enrich with customer tier:

CREATE OR REPLACE PIPE eu_vip_orders_pipe
  AS COPY INTO orders_enriched (order_id, customer_id, amount, tier)
  FROM (
    SELECT
      s.$1:order_id::STRING,
      s.$1:customer_id::STRING,
      s.$1:amount::NUMBER,
      COALESCE(c.tier, 'UNKNOWN') AS tier
    FROM TABLE(DATA_SOURCE(TYPE => 'STREAMING')) s
    LEFT OUTER JOIN mydb.ref.customers c
      ON s.$1:customer_id::STRING = c.customer_id
    WHERE s.$1:region::STRING = 'EU'
      AND s.$1:amount::NUMBER > 1000
  );

Usage notes¶

1. In pipes that include dimension joins, stream column references in the SELECT, ON, and WHERE clauses must be qualified with the stream alias (for example, s.$1:customer_id::STRING). Metadata columns such as METADATA$FILENAME and METADATA$START_SCAN_TIME may also be referenced in WHERE predicates.
2. Dimension snapshots are not refreshed while a pipe is idle. Ingestion is held until a valid snapshot is available; it resumes automatically once the snapshot is ready.
3. Streaming Transformations loads dimension table data into a per-pipe snapshot and refreshes it periodically while the PIPE is actively ingesting. Updates committed to the dimension table may not be visible immediately; staleness of up to several minutes is expected. Different ingest workers within the same pipe may briefly enrich against different snapshot versions.
4. If a dimension table is dropped or the PIPE owner role loses SELECT access after the pipe is created, the pipe suspends with an error. Restore access and resume the pipe to continue ingestion.
5. INNER JOIN isn’t supported. To keep only matched rows, use LEFT OUTER JOIN and WHERE on a dimension key with IS NOT NULL.

Limitations¶

1. Only equi-joins are supported; no extra predicates, no expressions, no subqueries.
2. Currently at most one matching row per stream record and dimension table is supported.
3. Currently up to 4 dimension tables per PIPE are supported.
4. Please refer to the Dimension table requirements section above for the requirements applicable to each dimension table.

See also¶

The Scenarios show real-world use cases that combine stream enrichment with other transformations. For the complete syntax with all features combined, see the Reference.

Previous: Stream filtering · Next: Custom scalar logic