From 5b067d71ddcd92a754e923a34ee559046d5bfb3e Mon Sep 17 00:00:00 2001 From: Valerie Chiang Date: Fri, 23 Oct 2026 10:20:49 -0400 Subject: [PATCH 1/4] doc: pattern for transactional downstream updates using Kafka Adds an Export data > Patterns section and a page describing how to gate downstream reads of a Kafka sink on the sink's progress frontier, joining data and progress records with the `materialize-sink-id` header. Part of SS-504. Co-Authored-By: Claude Opus 5.5 --- .../content/export-data/patterns/_index.md | 12 +++ .../transactional-downstream-updates-kafka.md | 91 +++++++++++++++++++ 2 files changed, 103 insertions(+) create mode 100644 doc/user/content/export-data/patterns/_index.md create mode 100644 doc/user/content/export-data/patterns/transactional-downstream-updates-kafka.md diff --git a/doc/user/content/export-data/patterns/_index.md b/doc/user/content/export-data/patterns/_index.md new file mode 100644 index 0000000000000..0a18304f02861 --- /dev/null +++ b/doc/user/content/export-data/patterns/_index.md @@ -0,0 +1,12 @@ +--- +title: "Patterns" +description: "Learn about common patterns for consuming data that Materialize sinks to external systems." +menu: + main: + parent: sink + identifier: sink-patterns + weight: 70 +--- + +The following section provides examples of common patterns for consuming the +data that Materialize writes to external systems: diff --git a/doc/user/content/export-data/patterns/transactional-downstream-updates-kafka.md b/doc/user/content/export-data/patterns/transactional-downstream-updates-kafka.md new file mode 100644 index 0000000000000..20e661570c7e2 --- /dev/null +++ b/doc/user/content/export-data/patterns/transactional-downstream-updates-kafka.md @@ -0,0 +1,91 @@ +--- +title: "Transactional downstream updates using Kafka" +description: "Expose only fully committed Materialize timestamps when loading a Kafka sink into another data store." +menu: + main: + parent: 'sink-patterns' +--- + +When you load a [Kafka sink](/sql/create-sink/kafka/) into another data store, +such as Snowflake or ClickHouse, the loader writes messages as they arrive. A +query against the downstream store can therefore see some, but not all, of the +changes Materialize made at a given timestamp. Use this pattern when downstream +readers must only see complete, transactionally consistent results. + +## How it works + +Materialize writes each batch of sink messages to Kafka in a transaction, +together with a progress record in the connection's [progress +topic](/sql/create-sink/kafka/#exactly-once-processing). The progress record +says which timestamps the sink has fully committed. A downstream view can join +the two topics and hide any row that is not yet covered by a progress record. + +| Where | What to read | Value | +|-------|--------------|-------| +| Data topic | `materialize-timestamp` header | The Materialize timestamp of the change, for example `1792700000000`. | +| Data topic | [`materialize-sink-id`](/sql/create-sink/kafka/#the-materialize-sink-id-header) header | The sink's ID, as shown in [`mz_sinks`](/sql/system-catalog/mz_catalog/#mz_sinks), for example `u123`. | +| Progress topic | Message key | `mz-sink-` followed by the sink's ID, for example `mz-sink-u123`. | +| Progress topic | Message value | JSON of the form `{"frontier":[1792700001000],"version":1}`. | + +The `frontier` is exclusive: the sink has committed every change with a +timestamp less than the frontier. To expose only committed data: + +1. Match each data message to the progress records whose key equals + `'mz-sink-' || `. +1. Take the latest frontier for that sink. +1. Keep the data messages where `materialize-timestamp` is less than that + frontier. + +The progress topic name is shown in the `sink_progress_topic` column of +[`mz_kafka_connections`](/sql/system-catalog/mz_catalog/#mz_kafka_connections). + +## Example: Snowflake + +This example is illustrative. It assumes the [Snowflake connector for +Kafka](https://docs.snowflake.com/en/user-guide/kafka-connector-overview) loads +the data topic into `raw_orders` and the progress topic into `raw_progress`. +The connector stores each message in a `RECORD_CONTENT` column and its headers +under `RECORD_METADATA:headers`. It keeps the message key in +`RECORD_METADATA:key` only when `key.converter` is set to +`org.apache.kafka.connect.storage.StringConverter`, so configure the progress +topic that way, and parse its value as JSON. + +```sql +CREATE VIEW committed_orders AS +WITH latest_progress AS ( + SELECT + RECORD_METADATA:key::string AS progress_key, + MAX(RECORD_CONTENT:frontier[0]::number) AS frontier + FROM raw_progress + GROUP BY 1 +) +SELECT o.RECORD_CONTENT AS order_record +FROM raw_orders AS o +JOIN latest_progress AS p + ON p.progress_key = + 'mz-sink-' || o.RECORD_METADATA:headers:"materialize-sink-id"::string +WHERE o.RECORD_METADATA:headers:"materialize-timestamp"::number < p.frontier; +``` + +Readers query `committed_orders` instead of `raw_orders`. + +## Considerations + +- The `materialize-sink-id` header is in private preview. See [The + `materialize-sink-id` header](/sql/create-sink/kafka/#the-materialize-sink-id-header) + for its current status. +- The progress topic is shared by every sink that uses the same Kafka + connection. Always filter progress records by key. +- Materialize writes a progress record each time the sink's frontier advances, + even when there is no new data. Expect the progress table to grow steadily. +- Downstream consumers must read only committed Kafka data + (`isolation.level=read_committed`). Otherwise they can load messages from + aborted transactions. See [End-to-end exactly-once + processing](/sql/create-sink/kafka/#end-to-end-exactly-once-processing). +- Header keys starting with `materialize-` are reserved for Materialize. + +## Related pages + +- [`CREATE SINK`: Kafka and Redpanda](/sql/create-sink/kafka/) +- [Exactly-once processing](/sql/create-sink/kafka/#exactly-once-processing) +- [Headers](/sql/create-sink/kafka/#headers) From 7fd3d72e0c9ab0904e90a0ffd7ed09f941c04abb Mon Sep 17 00:00:00 2001 From: Valerie Chiang Date: Fri, 23 Oct 2026 10:45:28 -0400 Subject: [PATCH 2/4] doc: note the materialize-sink-id header becomes GA in v26.48 Co-Authored-By: Claude Opus 5.5 --- .../patterns/transactional-downstream-updates-kafka.md | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/doc/user/content/export-data/patterns/transactional-downstream-updates-kafka.md b/doc/user/content/export-data/patterns/transactional-downstream-updates-kafka.md index 20e661570c7e2..415687bbfa1ba 100644 --- a/doc/user/content/export-data/patterns/transactional-downstream-updates-kafka.md +++ b/doc/user/content/export-data/patterns/transactional-downstream-updates-kafka.md @@ -71,9 +71,9 @@ Readers query `committed_orders` instead of `raw_orders`. ## Considerations -- The `materialize-sink-id` header is in private preview. See [The - `materialize-sink-id` header](/sql/create-sink/kafka/#the-materialize-sink-id-header) - for its current status. +- The `materialize-sink-id` header is in private preview until it becomes + generally available in v26.48. See [The `materialize-sink-id` + header](/sql/create-sink/kafka/#the-materialize-sink-id-header). - The progress topic is shared by every sink that uses the same Kafka connection. Always filter progress records by key. - Materialize writes a progress record each time the sink's frontier advances, From 64bfdbd2b20ffc0da53828bb55ef1ff1fdf79602 Mon Sep 17 00:00:00 2001 From: Valerie Chiang Date: Mon, 26 Oct 2026 08:02:45 -0700 Subject: [PATCH 3/4] doc: drop private preview note, sink-id header is GA in v26.48 Co-Authored-By: Claude Opus 5.5 --- .../patterns/transactional-downstream-updates-kafka.md | 3 --- 1 file changed, 3 deletions(-) diff --git a/doc/user/content/export-data/patterns/transactional-downstream-updates-kafka.md b/doc/user/content/export-data/patterns/transactional-downstream-updates-kafka.md index 415687bbfa1ba..83abbad9dac3b 100644 --- a/doc/user/content/export-data/patterns/transactional-downstream-updates-kafka.md +++ b/doc/user/content/export-data/patterns/transactional-downstream-updates-kafka.md @@ -71,9 +71,6 @@ Readers query `committed_orders` instead of `raw_orders`. ## Considerations -- The `materialize-sink-id` header is in private preview until it becomes - generally available in v26.48. See [The `materialize-sink-id` - header](/sql/create-sink/kafka/#the-materialize-sink-id-header). - The progress topic is shared by every sink that uses the same Kafka connection. Always filter progress records by key. - Materialize writes a progress record each time the sink's frontier advances, From dcb48bafecc0cf52b92a914e6d8f2e1827aafa31 Mon Sep 17 00:00:00 2001 From: Valerie Chiang Date: Thu, 29 Oct 2026 06:20:09 -0700 Subject: [PATCH 4/4] doc: expand Kafka transactional pattern into a step-by-step guide Address review: rephrase intro, and walk through creating the sink in Materialize, loading both topics into Snowflake, and building the committed-changes view. Co-Authored-By: Claude Opus 5.5 --- .../transactional-downstream-updates-kafka.md | 179 +++++++++++++++--- 1 file changed, 154 insertions(+), 25 deletions(-) diff --git a/doc/user/content/export-data/patterns/transactional-downstream-updates-kafka.md b/doc/user/content/export-data/patterns/transactional-downstream-updates-kafka.md index 83abbad9dac3b..65ceeb42327a2 100644 --- a/doc/user/content/export-data/patterns/transactional-downstream-updates-kafka.md +++ b/doc/user/content/export-data/patterns/transactional-downstream-updates-kafka.md @@ -6,11 +6,21 @@ menu: parent: 'sink-patterns' --- -When you load a [Kafka sink](/sql/create-sink/kafka/) into another data store, -such as Snowflake or ClickHouse, the loader writes messages as they arrive. A -query against the downstream store can therefore see some, but not all, of the -changes Materialize made at a given timestamp. Use this pattern when downstream -readers must only see complete, transactionally consistent results. +When you load messages from Kafka into another data store, such as Snowflake or +ClickHouse, the loader writes messages as they arrive. A query against the +downstream store can therefore see some, but not all, of the changes +Materialize made at a given timestamp. Use this pattern when downstream readers +must only see complete, transactionally consistent results. + +This guide walks you through the pattern end to end, using Snowflake as the +destination: + +1. Create a Kafka sink in Materialize. +1. Load the sink's data topic and progress topic into Snowflake. +1. Create a Snowflake view that exposes only committed changes. + +The same approach works for any destination that stores Kafka headers and +message keys alongside the message value. ## How it works @@ -36,50 +46,169 @@ timestamp less than the frontier. To expose only committed data: 1. Keep the data messages where `materialize-timestamp` is less than that frontier. -The progress topic name is shown in the `sink_progress_topic` column of -[`mz_kafka_connections`](/sql/system-catalog/mz_catalog/#mz_kafka_connections). +## Before you begin + +- A Kafka or Redpanda cluster that Materialize can write to. +- A Kafka Connect cluster with the [Snowflake connector for + Kafka](https://docs.snowflake.com/en/user-guide/kafka-connector-overview) + installed. +- A Snowflake database and schema, and a Snowflake user that the connector can + authenticate as. See [Installing and configuring the Kafka + connector](https://docs.snowflake.com/en/user-guide/kafka-connector-install). +- A table, view, or materialized view in Materialize to sink. This guide uses a + materialized view named `orders` with a unique key column `id`. + +## Step 1. Create a Kafka sink in Materialize + +1. Create a [connection](/sql/create-connection/#kafka) to your Kafka cluster. + You can optionally name the progress topic with the `PROGRESS TOPIC` option; + otherwise, Materialize uses a default name. + + ```mzsql + CREATE SECRET kafka_password AS ''; + + CREATE CONNECTION kafka_connection TO KAFKA ( + BROKER '', + SASL MECHANISMS = 'SCRAM-SHA-256', + SASL USERNAME = '', + SASL PASSWORD = SECRET kafka_password + ); + ``` -## Example: Snowflake +1. Create the sink. This guide uses the JSON format and the upsert envelope. -This example is illustrative. It assumes the [Snowflake connector for -Kafka](https://docs.snowflake.com/en/user-guide/kafka-connector-overview) loads -the data topic into `raw_orders` and the progress topic into `raw_progress`. -The connector stores each message in a `RECORD_CONTENT` column and its headers -under `RECORD_METADATA:headers`. It keeps the message key in -`RECORD_METADATA:key` only when `key.converter` is set to -`org.apache.kafka.connect.storage.StringConverter`, so configure the progress -topic that way, and parse its value as JSON. + ```mzsql + CREATE SINK orders_sink + FROM orders + INTO KAFKA CONNECTION kafka_connection (TOPIC 'orders') + KEY (id) + FORMAT JSON + ENVELOPE UPSERT; + ``` + + Materialize adds the `materialize-timestamp` and `materialize-sink-id` + headers to every message automatically. You don't need to set any options. + +1. Look up the sink's ID and the name of its progress topic. You need the + progress topic name to configure the Snowflake connector. + + ```mzsql + SELECT s.id AS sink_id, c.sink_progress_topic + FROM mz_sinks AS s + JOIN mz_kafka_connections AS c ON s.connection_id = c.id + WHERE s.name = 'orders_sink'; + ``` + + ``` + sink_id | sink_progress_topic + ---------+-------------------------------------- + u123 | _materialize-progress-- + ``` + +## Step 2. Load both topics into Snowflake + +Configure the Snowflake connector to load the data topic into a table named +`raw_orders` and the progress topic into a table named `raw_progress`. For +example: + +```json +{ + "name": "materialize_orders", + "config": { + "connector.class": "com.snowflake.kafka.connector.SnowflakeSinkConnector", + "topics": "orders,", + "snowflake.topic2table.map": "orders:raw_orders,:raw_progress", + "snowflake.url.name": ".snowflakecomputing.com:443", + "snowflake.user.name": "", + "snowflake.private.key": "", + "snowflake.database.name": "", + "snowflake.schema.name": "", + "key.converter": "org.apache.kafka.connect.storage.StringConverter", + "value.converter": "com.snowflake.kafka.connector.records.SnowflakeJsonConverter", + "consumer.override.isolation.level": "read_committed" + } +} +``` + +Each Snowflake table gets two columns: + +- `RECORD_CONTENT`: the message value, as a `VARIANT`. For the progress topic, + this is the JSON object containing `frontier`. +- `RECORD_METADATA`: the message metadata, as a `VARIANT`. The message key is + in `RECORD_METADATA:key` and the headers are in `RECORD_METADATA:headers`. + +Two settings in this configuration matter for the pattern: + +- `key.converter` must be `org.apache.kafka.connect.storage.StringConverter`. + Otherwise, the connector does not keep the progress topic's message key, and + you can't tell which sink a progress record belongs to. +- `consumer.override.isolation.level` must be `read_committed`, so that the + connector never loads messages from aborted Kafka transactions. Your Kafka + Connect worker must allow client configuration overrides + (`connector.client.config.override.policy`). Alternatively, set + `consumer.isolation.level=read_committed` in the worker configuration. + +## Step 3. Create a view of committed changes in Snowflake + +In Snowflake, create a view that keeps only the data messages that Materialize +has fully committed: ```sql CREATE VIEW committed_orders AS WITH latest_progress AS ( + -- The latest frontier for each sink that uses the progress topic. SELECT RECORD_METADATA:key::string AS progress_key, MAX(RECORD_CONTENT:frontier[0]::number) AS frontier FROM raw_progress GROUP BY 1 ) -SELECT o.RECORD_CONTENT AS order_record +SELECT + o.RECORD_METADATA:headers:"materialize-timestamp"::number AS mz_timestamp, + o.RECORD_CONTENT AS order_record FROM raw_orders AS o +-- Match each data message to the progress records for its sink. JOIN latest_progress AS p ON p.progress_key = 'mz-sink-' || o.RECORD_METADATA:headers:"materialize-sink-id"::string +-- Keep only the changes below the frontier. WHERE o.RECORD_METADATA:headers:"materialize-timestamp"::number < p.frontier; ``` -Readers query `committed_orders` instead of `raw_orders`. +The view works as follows: + +1. `latest_progress` finds the most recent frontier for each sink. Because the + progress topic is shared by every sink on the same Kafka connection, the + query groups by message key. +1. The join matches each order message to its sink's frontier, using the + `materialize-sink-id` header. +1. The `WHERE` clause drops any message whose timestamp is not yet below the + frontier. Those messages appear in the view once Materialize commits the + rest of their timestamp. + +## Step 4. Query the view + +Point downstream readers at `committed_orders` instead of `raw_orders`: + +```sql +SELECT * FROM committed_orders ORDER BY mz_timestamp DESC LIMIT 10; +``` + +Every row in the result belongs to a Materialize timestamp whose changes have +all been loaded into Snowflake. ## Considerations -- The progress topic is shared by every sink that uses the same Kafka - connection. Always filter progress records by key. +- `committed_orders` contains one row per change, not the current state of + `orders`. With the upsert envelope, a deletion is a message with a `null` + value (a tombstone). To build the current state, take the latest committed + row for each key and handle tombstones as configured by the connector's + `behavior.on.null.values` setting. - Materialize writes a progress record each time the sink's frontier advances, - even when there is no new data. Expect the progress table to grow steadily. -- Downstream consumers must read only committed Kafka data - (`isolation.level=read_committed`). Otherwise they can load messages from - aborted transactions. See [End-to-end exactly-once - processing](/sql/create-sink/kafka/#end-to-end-exactly-once-processing). + even when there is no new data. Expect `raw_progress` to grow steadily. - Header keys starting with `materialize-` are reserved for Materialize. +- For other guarantees you need end to end, see [End-to-end exactly-once + processing](/sql/create-sink/kafka/#end-to-end-exactly-once-processing). ## Related pages