From 3f3af66c47c6ec08d177df9c88f337217470b15b Mon Sep 17 00:00:00 2001 From: Alan Yu Date: Tue, 4 Aug 2026 17:20:50 +0800 Subject: [PATCH 1/2] add document for consume_schema_strategy --- docs/kafka-schema-registry.md | 4 +- docs/shared/kafka-external-stream-read.md | 69 +++++++++++++- docs/shared/kafka-external-stream.md | 106 +++++++++++++++++++++- docs/sql-create-external-stream.md | 7 +- 4 files changed, 181 insertions(+), 5 deletions(-) diff --git a/docs/kafka-schema-registry.md b/docs/kafka-schema-registry.md index 99275d577..b60dd032e 100644 --- a/docs/kafka-schema-registry.md +++ b/docs/kafka-schema-registry.md @@ -18,6 +18,7 @@ CREATE EXTERNAL STREAM my_stream ( data_format = 'Avro', -- or 'ProtobufSingle' subject_name_strategy = '..', schema_subject_name = '..', + consume_schema_strategy = '..', kafka_schema_registry_url = 'http://url.to/my/schema/registry', kafka_schema_registry_credentials = 'API_KEY:API_SECRET', kafka_schema_registry_skip_cert_check = [true|false], @@ -52,8 +53,9 @@ The `subject_name_strategy` determines how the stream looks up schemas in the re The schema subject specified in external stream is used in the following cases: 1. Auto inference columns name and type from schema when no column definition in create DDL. 2. Encode Timeplus data and write to Kafka. +3. **Filter messages by schema during reads** — when `consume_schema_strategy='single'` (default), only messages matching the schema identified by `schema_subject_name` and `subject_name_strategy` are consumed. Non-matching messages are silently skipped. For details, see [consume_schema_strategy](/kafka-source#consume_schema_strategy). -In reading from Kafka, the schema subject settings are ignored. The schema ID is get directly from each Kafka record and decoded with corresponding schema. The decoded messages are then converted to external stream rows. If the column name is not found in the decoded message keys, the default value of column type is filled. +For `consume_schema_strategy='all'`, the schema subject is not used for filtering; the schema ID is read directly from each Kafka record's Confluent header, the corresponding schema is fetched from the registry (cached), and all messages are decoded and mapped to stream columns by name. ::: ## Write Messages in Avro Schema{#write} diff --git a/docs/shared/kafka-external-stream-read.md b/docs/shared/kafka-external-stream-read.md index c398fc28c..b122d6ef8 100644 --- a/docs/shared/kafka-external-stream-read.md +++ b/docs/shared/kafka-external-stream-read.md @@ -136,7 +136,74 @@ SETTINGS data_format='TSV'; ### Read Avro or Protobuf Messages -To read Avro-encoded / Protobuf-encoded Kafka message, please refer to [Avro Schema](/data-formats#avro), [Protobuf Schema](/data-formats#protobuf) and [Schema Registry](/kafka-schema-registry) for details. +To read Avro-encoded or Protobuf-encoded Kafka messages, please refer to [Avro Schema](/data-formats#avro), [Protobuf Schema](/data-formats#protobuf) and [Schema Registry](/kafka-schema-registry) for details. + +#### Handling Multi-Schema Topics + +Kafka topics may contain messages encoded with different Avro or Protobuf schemas (multiple schema IDs) via Confluent Schema Registry's `RecordNameStrategy` or `TopicRecordNameStrategy`. The `consume_schema_strategy` setting controls how these messages are processed. + +**`single` mode (default):** Only consume messages matching a specific schema. Non-matching messages are silently skipped. + +```sql +CREATE EXTERNAL STREAM user_events ( + user_id string, + event_type string, + created_at datetime64(3) +) SETTINGS + type='kafka', + brokers='localhost:9092', + topic='multi_schema_topic', + data_format='Avro', + kafka_schema_registry_url='http://localhost:8081', + consume_schema_strategy='single', + subject_name_strategy='RecordNameStrategy', + schema_subject_name='com.example.avro.UserEvent'; +``` + +**`all` mode:** Decode all messages regardless of schema ID. Fields are mapped to stream columns by name. Missing fields get default/null values. + +```sql +CREATE EXTERNAL STREAM all_events ( + user_id string, + event_type string, + created_at datetime64(3) +) SETTINGS + type='kafka', + brokers='localhost:9092', + topic='multi_schema_topic', + data_format='Avro', + kafka_schema_registry_url='http://localhost:8081', + consume_schema_strategy='all'; +``` + +**`raw` mode:** Strip the 5-byte Confluent header and store the raw payload as a single String column. No schema registry lookup or deserialization occurs. Useful for routing messages to different streams via materialized views. + +```sql +CREATE EXTERNAL STREAM raw_events (raw string) SETTINGS + type='kafka', + brokers='localhost:9092', + topic='multi_schema_topic', + data_format='Avro', + kafka_schema_registry_url='http://localhost:8081', + consume_schema_strategy='raw'; + +-- Route to typed streams via MVs +CREATE STREAM user_events ( + user_id string, + event_type string, + created_at datetime64(3) +); + +CREATE MATERIALIZED VIEW route_user_events INTO user_events AS +SELECT + json_extract_string(raw, 'user_id') AS user_id, + json_extract_string(raw, 'event_type') AS event_type, + json_extract_string(raw, 'created_at') AS created_at +FROM raw_events +WHERE json_extract_string(raw, '_schema_type') = 'user_event'; +``` + +For more details on `consume_schema_strategy`, see the [Kafka External Stream settings](/kafka-source#consume_schema_strategy). ### Access Kafka Message Metadata diff --git a/docs/shared/kafka-external-stream.md b/docs/shared/kafka-external-stream.md index 3033831b7..d4d1420d1 100644 --- a/docs/shared/kafka-external-stream.md +++ b/docs/shared/kafka-external-stream.md @@ -29,7 +29,10 @@ SETTINGS ssl_ca_pem='..', skip_ssl_cert_check=.., properties='..', - named_collection='..'; + named_collection='..', + subject_name_strategy='..', + schema_subject_name='..', + consume_schema_strategy='..'; ``` ### Settings @@ -211,3 +214,104 @@ SETTINGS ``` For more detailed syntax of named collection, please refer to the [Named Collection](/named-collection) documentation. + +#### subject_name_strategy + +Determines how the stream looks up schemas in the Kafka Schema Registry. Supported values: + +| Strategy | Behavior | Derived Subject Name | +| :--- | :--- | :--- | +| `TopicNameStrategy` | **Default.** Assumes one schema per topic. `schema_subject_name` is ignored. | `-value` | +| `RecordNameStrategy` | Supports mixed schemas in one topic. `schema_subject_name` is required. | `schema_subject_name` (fully qualified record name) | +| `TopicRecordNameStrategy` | Scopes record names to a specific topic. `schema_subject_name` is required. | `-` | + +For more details, see [Kafka Schema Registry](/kafka-schema-registry). + +#### schema_subject_name + +Specifies the subject name for schema lookups in the Schema Registry. Required when `subject_name_strategy` is set to `RecordNameStrategy` or `TopicRecordNameStrategy`. Typically the fully qualified record name, e.g. `com.example.avro.UserRecord`. + +#### consume_schema_strategy + +Controls how Kafka messages with different Confluent Schema Registry schema IDs are processed. This setting is useful when a Kafka topic contains messages encoded with multiple Avro or Protobuf schemas (e.g., via `RecordNameStrategy` or `TopicRecordNameStrategy`). + +Supported values: + +| Value | Default | Behavior | +| :--- | :--- | :--- | +| `single` | **Yes** | Only consume messages matching the schema identified by `schema_subject_name` and `subject_name_strategy`. Non-matching messages are silently skipped. `schema_subject_name` must be set. | +| `all` | No | Decode **all** messages by their schema ID, mapping fields to stream columns by name. Missing fields get default/null values. This preserves the pre-v3.4 behavior. | +| `raw` | No | Strip the 5-byte Confluent header (magic byte + schema ID) and store the remaining payload bytes in a single `String` column. No schema registry lookup or deserialization occurs. The external stream must have exactly one physical column of type `String`. | + +##### `single` mode (Default) + +When a topic contains multiple unrelated record types and you only care about one, use `single` mode. This is the safest default — each stream consumes exactly one schema. + +```sql +CREATE EXTERNAL STREAM user_events ( + user_id string, + event_type string, + created_at datetime64(3) +) SETTINGS + type='kafka', + brokers='localhost:9092', + topic='multi_schema_topic', + data_format='Avro', + kafka_schema_registry_url='http://localhost:8081', + consume_schema_strategy='single', + subject_name_strategy='RecordNameStrategy', + schema_subject_name='com.example.avro.UserEvent'; +``` + +Streaming queries against this external stream will only return rows where the message schema matches `com.example.avro.UserEvent`. Messages with other schema IDs are silently skipped. + +##### `all` mode + +When a topic has multiple related schemas that share field names, and you want all records in a single stream: + +```sql +CREATE EXTERNAL STREAM all_events ( + user_id string, + event_type string, + created_at datetime64(3) +) SETTINGS + type='kafka', + brokers='localhost:9092', + topic='multi_schema_topic', + data_format='Avro', + kafka_schema_registry_url='http://localhost:8081', + consume_schema_strategy='all'; +``` + +Each message is decoded by its schema ID, and fields are mapped to stream columns by name. Fields not present in a message get default/null values. + +##### `raw` mode + +When you want to preserve the raw bytes for later processing, route to different streams via materialized views, or handle decoding in application logic: + +```sql +CREATE EXTERNAL STREAM raw_events (raw string) SETTINGS + type='kafka', + brokers='localhost:9092', + topic='multi_schema_topic', + data_format='Avro', + kafka_schema_registry_url='http://localhost:8081', + consume_schema_strategy='raw'; + +-- Route to typed streams via materialized views +CREATE STREAM user_events (user_id string, event_type string, created_at datetime64(3)); + +CREATE MATERIALIZED VIEW route_user_events INTO user_events AS +SELECT + json_extract_string(raw, 'user_id') AS user_id, + json_extract_string(raw, 'event_type') AS event_type, + json_extract_string(raw, 'created_at') AS created_at +FROM raw_events +WHERE json_extract_string(raw, 'type') = 'user_event'; +``` + +In `raw` mode, the virtual columns (`_tp_message_key`, `_tp_message_headers`, `_tp_time`, `_tp_sn`, `_tp_shard`) are still populated as transport-level metadata. + +:::info Migration Note +Users who previously relied on `TopicNameStrategy` to consume all messages (including those from mixed-schema topics) must now explicitly set `consume_schema_strategy='all'` to preserve the old behavior. The default `single` mode only consumes messages matching the specified schema subject. +::: diff --git a/docs/sql-create-external-stream.md b/docs/sql-create-external-stream.md index 6bc5948d4..33c6a5ab6 100644 --- a/docs/sql-create-external-stream.md +++ b/docs/sql-create-external-stream.md @@ -18,10 +18,13 @@ SETTINGS type='kafka', kafka_schema_registry_url='..', kafka_schema_registry_credentials='..', ssl_ca_cert_file='..', - ss_ca_pem='..', + ssl_ca_pem='..', skip_ssl_cert_check=.., properties='..', - config_file='..' + config_file='..', + subject_name_strategy='..', + schema_subject_name='..', + consume_schema_strategy='..' ``` Please check the [Kafka External Stream](/kafka-source) for more details about the settings, and [this doc](/tutorial-sql-connect-kafka) for examples to connect to various Kafka API compatible message platforms. From ed84b6b4d42e101d9284b8f90513e9a011c55cc2 Mon Sep 17 00:00:00 2001 From: Alan Yu Date: Thu, 6 Aug 2026 12:00:36 +0800 Subject: [PATCH 2/2] fix comment --- docs/shared/kafka-external-stream-read.md | 8 +-- docs/shared/kafka-external-stream.md | 79 +---------------------- 2 files changed, 7 insertions(+), 80 deletions(-) diff --git a/docs/shared/kafka-external-stream-read.md b/docs/shared/kafka-external-stream-read.md index b122d6ef8..efec6f448 100644 --- a/docs/shared/kafka-external-stream-read.md +++ b/docs/shared/kafka-external-stream-read.md @@ -196,11 +196,11 @@ CREATE STREAM user_events ( CREATE MATERIALIZED VIEW route_user_events INTO user_events AS SELECT - json_extract_string(raw, 'user_id') AS user_id, - json_extract_string(raw, 'event_type') AS event_type, - json_extract_string(raw, 'created_at') AS created_at + raw:user_id AS user_id, + raw:event_type AS event_type, + raw:created_at AS created_at FROM raw_events -WHERE json_extract_string(raw, '_schema_type') = 'user_event'; +WHERE raw:type = 'user_event'; ``` For more details on `consume_schema_strategy`, see the [Kafka External Stream settings](/kafka-source#consume_schema_strategy). diff --git a/docs/shared/kafka-external-stream.md b/docs/shared/kafka-external-stream.md index d4d1420d1..bdf3fc238 100644 --- a/docs/shared/kafka-external-stream.md +++ b/docs/shared/kafka-external-stream.md @@ -239,79 +239,6 @@ Supported values: | Value | Default | Behavior | | :--- | :--- | :--- | -| `single` | **Yes** | Only consume messages matching the schema identified by `schema_subject_name` and `subject_name_strategy`. Non-matching messages are silently skipped. `schema_subject_name` must be set. | -| `all` | No | Decode **all** messages by their schema ID, mapping fields to stream columns by name. Missing fields get default/null values. This preserves the pre-v3.4 behavior. | -| `raw` | No | Strip the 5-byte Confluent header (magic byte + schema ID) and store the remaining payload bytes in a single `String` column. No schema registry lookup or deserialization occurs. The external stream must have exactly one physical column of type `String`. | - -##### `single` mode (Default) - -When a topic contains multiple unrelated record types and you only care about one, use `single` mode. This is the safest default — each stream consumes exactly one schema. - -```sql -CREATE EXTERNAL STREAM user_events ( - user_id string, - event_type string, - created_at datetime64(3) -) SETTINGS - type='kafka', - brokers='localhost:9092', - topic='multi_schema_topic', - data_format='Avro', - kafka_schema_registry_url='http://localhost:8081', - consume_schema_strategy='single', - subject_name_strategy='RecordNameStrategy', - schema_subject_name='com.example.avro.UserEvent'; -``` - -Streaming queries against this external stream will only return rows where the message schema matches `com.example.avro.UserEvent`. Messages with other schema IDs are silently skipped. - -##### `all` mode - -When a topic has multiple related schemas that share field names, and you want all records in a single stream: - -```sql -CREATE EXTERNAL STREAM all_events ( - user_id string, - event_type string, - created_at datetime64(3) -) SETTINGS - type='kafka', - brokers='localhost:9092', - topic='multi_schema_topic', - data_format='Avro', - kafka_schema_registry_url='http://localhost:8081', - consume_schema_strategy='all'; -``` - -Each message is decoded by its schema ID, and fields are mapped to stream columns by name. Fields not present in a message get default/null values. - -##### `raw` mode - -When you want to preserve the raw bytes for later processing, route to different streams via materialized views, or handle decoding in application logic: - -```sql -CREATE EXTERNAL STREAM raw_events (raw string) SETTINGS - type='kafka', - brokers='localhost:9092', - topic='multi_schema_topic', - data_format='Avro', - kafka_schema_registry_url='http://localhost:8081', - consume_schema_strategy='raw'; - --- Route to typed streams via materialized views -CREATE STREAM user_events (user_id string, event_type string, created_at datetime64(3)); - -CREATE MATERIALIZED VIEW route_user_events INTO user_events AS -SELECT - json_extract_string(raw, 'user_id') AS user_id, - json_extract_string(raw, 'event_type') AS event_type, - json_extract_string(raw, 'created_at') AS created_at -FROM raw_events -WHERE json_extract_string(raw, 'type') = 'user_event'; -``` - -In `raw` mode, the virtual columns (`_tp_message_key`, `_tp_message_headers`, `_tp_time`, `_tp_sn`, `_tp_shard`) are still populated as transport-level metadata. - -:::info Migration Note -Users who previously relied on `TopicNameStrategy` to consume all messages (including those from mixed-schema topics) must now explicitly set `consume_schema_strategy='all'` to preserve the old behavior. The default `single` mode only consumes messages matching the specified schema subject. -::: +| `single` | **Yes** | Only consume messages matching the schema identified by `schema_subject_name` and `subject_name_strategy`. Non-matching messages are silently skipped. | +| `all` | No | Decode **all** messages by their schema ID, mapping fields to stream columns by name. Missing fields get default/null values. | +| `raw` | No | Decode messages by their schema ID and stringy the result into JSON format text. No schema registry lookup or deserialization occurs. The external stream must have exactly one physical column of type `String`. |