add document for consume_schema_strategy

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}

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
+    raw:user_id AS user_id,
+    raw:event_type AS event_type,
+    raw:created_at AS created_at
+FROM raw_events
+WHERE raw: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
 

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,31 @@ 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. | `<topic>-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. | `<topic>-<schema_subject_name>` |
+
+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. |
+| `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`. |

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.