CREATE TABLE: Kafka source table

In Materialize, you can create read-only tables from Kafka/Redpanda sources created using the new syntax.

Syntax

CREATE TABLE [IF NOT EXISTS] <table_name>
FROM SOURCE <src_name>
FORMAT AVRO
    USING CONFLUENT SCHEMA REGISTRY CONNECTION <csr_connection_name>
      [KEY STRATEGY <key_strategy>]
      [VALUE STRATEGY <value_strategy>]
  | USING AWS GLUE SCHEMA REGISTRY CONNECTION <glue_connection_name> (
      SCHEMA NAME = '<schema_name>'
    )
[INCLUDE
    KEY [AS <name>]
  | PARTITION [AS <name>]
  | OFFSET [AS <name>]
  | TIMESTAMP [AS <name>]
  | HEADERS [AS <name>]
  | HEADER '<key>' AS <name> [BYTES]
  [, ...]
]
[ENVELOPE
    NONE
  | DEBEZIUM
  | UPSERT [ ( VALUE DECODING ERRORS = INLINE [AS <name>] ) ]
];

CREATE TABLE [IF NOT EXISTS] <table_name>
FROM SOURCE <src_name>
FORMAT JSON
[INCLUDE
    PARTITION [AS <name>]
  | OFFSET [AS <name>]
  | TIMESTAMP [AS <name>]
  | HEADERS [AS <name>]
  | HEADER '<key>' AS <name> [BYTES]
  [, ...]
]
[ENVELOPE NONE];

CREATE TABLE [IF NOT EXISTS] <table_name>
FROM SOURCE <src_name>
FORMAT TEXT | BYTES
[INCLUDE
    PARTITION [AS <name>]
  | OFFSET [AS <name>]
  | TIMESTAMP [AS <name>]
  | HEADERS [AS <name>]
  | HEADER '<key>' AS <name> [BYTES]
  [, ...]
]
[ENVELOPE NONE];

CREATE TABLE [IF NOT EXISTS] <table_name>
FROM SOURCE <src_name>
FORMAT CSV WITH <n> COLUMNS [DELIMITED BY <char>]
[INCLUDE
    PARTITION [AS <name>]
  | OFFSET [AS <name>]
  | TIMESTAMP [AS <name>]
  | HEADERS [AS <name>]
  | HEADER '<key>' AS <name> [BYTES]
  [, ...]
]
[ENVELOPE NONE];

CREATE TABLE [IF NOT EXISTS] <table_name>
FROM SOURCE <src_name>
FORMAT PROTOBUF USING CONFLUENT SCHEMA REGISTRY CONNECTION <csr_connection_name>
  | FORMAT PROTOBUF MESSAGE '<message_name>' USING SCHEMA '<schema_bytes>'
[INCLUDE
    KEY [AS <name>]
  | PARTITION [AS <name>]
  | OFFSET [AS <name>]
  | TIMESTAMP [AS <name>]
  | HEADERS [AS <name>]
  | HEADER '<key>' AS <name> [BYTES]
  [, ...]
]
[ENVELOPE
    NONE
  | UPSERT [ ( VALUE DECODING ERRORS = INLINE [AS <name>] ) ]
];

CREATE TABLE [IF NOT EXISTS] <table_name>
FROM SOURCE <src_name>
KEY FORMAT <key_format> VALUE FORMAT <value_format>
-- <key_format> and <value_format> can be:
-- AVRO USING CONFLUENT SCHEMA REGISTRY CONNECTION <conn_name>
--     [KEY STRATEGY <strategy>]
--     [VALUE STRATEGY <strategy>]
-- | AVRO USING AWS GLUE SCHEMA REGISTRY CONNECTION <glue_conn_name> (SCHEMA NAME = '<schema_name>')
-- | CSV WITH <num> COLUMNS DELIMITED BY <char>
-- | JSON | TEXT | BYTES
-- | PROTOBUF USING CONFLUENT SCHEMA REGISTRY CONNECTION <conn_name>
-- | PROTOBUF MESSAGE '<message_name>' USING SCHEMA '<schema_bytes>'
[INCLUDE
    KEY [AS <name>]
  | PARTITION [AS <name>]
  | OFFSET [AS <name>]
  | TIMESTAMP [AS <name>]
  | HEADERS [AS <name>]
  | HEADER '<key>' AS <name> [BYTES]
  [, ...]
]
[ENVELOPE
    NONE
  | DEBEZIUM
  | UPSERT [(VALUE DECODING ERRORS = INLINE [AS name])]
];

Envelopes

In addition to determining how to decode incoming records, Materialize also needs to understand how to interpret them. Whether a new record inserts, updates, or deletes existing data in Materialize depends on the ENVELOPE specified. For where the ENVELOPE clause goes, see Syntax.

Append-only envelope

Syntax: ENVELOPE NONE

The append-only envelope treats all records as inserts. This is the default envelope, if no envelope is specified.

Upsert envelope

Syntax: ENVELOPE UPSERT

The upsert envelope uses the standard key-value convention to support inserts, updates, and deletes within Materialize. It treats all records as having a key and a value:

• If the key does not match a preexisting record, it inserts the record’s key and value.

• If the key matches a preexisting record and the value is non-null, Materialize updates the existing record with the new value.

• If the key matches a preexisting record and the value is null, Materialize deletes the record.

Null keys

If a message with a NULL key is detected, Materialize sets the source into an error state. To recover an errored source, you must produce a record with a NULL value and a NULL key to the topic, to force a retraction.

As an example, you can use kcat to produce an empty message:

echo ":" | kcat -b $BROKER -t $TOPIC -Z -K: \
  -X security.protocol=SASL_SSL \
  -X sasl.mechanisms=SCRAM-SHA-256 \
  -X sasl.username=$KAFKA_USERNAME \
  -X sasl.password=$KAFKA_PASSWORD

Value decoding errors

By default, if an error happens while decoding the value of a message for a specific key, Materialize sets the source into an error state. You can configure the source to continue ingesting data in the presence of value decoding errors using the VALUE DECODING ERRORS = INLINE option:

ENVELOPE UPSERT (VALUE DECODING ERRORS = INLINE)

When this option is specified the source will include an additional column named error with type record(description: text).

This column and all value columns will be nullable, such that if the most recent value for the given Kafka message key cannot be decoded, this error column will contain the error message. If the most recent value for a key has been successfully decoded, this column will be NULL.

To use an alternative name for the error column, use INLINE AS .. to specify the column name to use:

ENVELOPE UPSERT (VALUE DECODING ERRORS = (INLINE AS my_error_col))

It might be convenient to implement a parsing view on top of your Kafka upsert source that excludes keys with decoding errors:

CREATE VIEW kafka_upsert_parsed
SELECT *
FROM kafka_upsert
WHERE error IS NULL;

Debezium envelope

Syntax: ENVELOPE DEBEZIUM

Materialize provides a dedicated envelope (ENVELOPE DEBEZIUM) to decode Kafka messages produced by Debezium.

Any materialized view defined on top of a Debezium source will be incrementally updated as new change events stream in through Kafka, as a result of INSERT, UPDATE and DELETE operations in the original database.

This envelope treats all records as change events with a diff structure that indicates whether each record should be interpreted as an insert, update or delete within Materialize:

Truncation

The Debezium envelope does not support upstream truncate events.

Debezium metadata

The envelope exposes the before and after value fields from change events.

Duplicate handling

Debezium may produce duplicate records if the connector is interrupted. Materialize makes a best-effort attempt to detect and filter out duplicates.

Details

Avro

Schema registries

Materialize can retrieve Avro schemas from either of two schema registries, selected by the USING clause:

• Confluent Schema Registry (USING CONFLUENT SCHEMA REGISTRY): schemas are looked up by topic using the TopicNameStrategy, and the message’s embedded schema ID resolves the writer schema at decode time.

• AWS Glue Schema Registry (USING AWS GLUE SCHEMA REGISTRY) (feature in private preview) : schemas are looked up by the SCHEMA NAME you provide, and the message’s embedded schema version ID resolves the writer schema at decode time. Each FORMAT AVRO USING AWS GLUE clause resolves a single schema. To decode keys and values from different schemas (for example, under ENVELOPE UPSERT or ENVELOPE DEBEZIUM), you must specify KEY FORMAT ... VALUE FORMAT ... explicitly.

Schema versioning

The schema is resolved when the source or table is created. With Confluent Schema Registry, the latest schema is retrieved using the TopicNameStrategy strategy. With AWS Glue Schema Registry, the latest version of the schema named by SCHEMA NAME is retrieved.

Schema evolution

As long as the writer schema changes in a compatible way, Materialize will continue using the original reader schema definition by mapping values from the new to the old schema version. This applies to both Confluent Schema Registry and AWS Glue Schema Registry.

To pick up the new version of the writer schema, the approach depends on the syntax you used:

• Legacy syntax (CREATE SOURCE ... FORMAT AVRO ...): you need to drop and recreate the source, which incurs downtime.
• New syntax (CREATE SOURCE plus CREATE TABLE ... FROM SOURCE): you can create a new table that reads the evolved schema and cut over without downtime. See Handle upstream schema changes with zero downtime.

Name collision

To avoid case-sensitivity conflicts with Materialize identifiers, we recommend double-quoting all field names when working with Avro-formatted sources.

Supported types

Materialize supports all Avro types, except for recursive types and union types in arrays.

JSON

If your JSON messages have a consistent shape, we recommend creating a parsing view that maps the individual fields to columns with the required data types:

-- extract jsonb into typed columns
CREATE VIEW my_typed_source AS
  SELECT
    (data->>'field1')::boolean AS field_1,
    (data->>'field2')::int AS field_2,
    (data->>'field3')::float AS field_3
  FROM my_jsonb_source;

To avoid doing this task manually, you can use this JSON parsing widget.

Schema registry integration

Retrieving schemas from a schema registry is not supported yet for JSON-formatted sources. This means that Materialize cannot decode messages serialized using the JSON Schema serialization format (JSON_SR).

Protobuf

Unlike Avro, Protobuf does not serialize a schema with the message, so Materialize expects:

• A FileDescriptorSet that encodes the Protobuf message schema. You can generate the FileDescriptorSet with protoc, for example:

  protoc --include_imports --descriptor_set_out=SCHEMA billing.proto
  

• A top-level message name and its package name, so Materialize knows which message from the FileDescriptorSet is the top-level message to decode, in the following format:

  <package name>.<top-level message>
  

  For example, if the FileDescriptorSet were from a .proto file in the billing package, and the top-level message was called Batch, the message_name value would be billing.Batch.

Schema versioning

The latest schema is retrieved using the TopicNameStrategy strategy at the time the source or table is created.

Schema evolution

As long as the .proto schema definition changes in a compatible way, Materialize will continue using the original schema definition by mapping values from the new to the old schema version. To pick up the new version of the schema with the legacy syntax (CREATE SOURCE ... FORMAT PROTOBUF ...), you need to drop and recreate the source. With the new syntax (CREATE SOURCE plus CREATE TABLE ... FROM SOURCE), you can instead create a new table that reads the evolved schema and cut over without downtime, following the approach in Handle upstream schema changes with zero downtime.

Supported types

Materialize supports all well-known Protobuf types from the proto2 and proto3 specs, except for recursive Struct values and map types.

Multiple message schemas

When using a schema registry with Protobuf sources, the registered schemas must contain exactly one Message definition.

Exposing source metadata

In addition to the message value, Materialize can expose the message key, headers and other source metadata fields to SQL through the INCLUDE clause. For where the INCLUDE clause goes, see Syntax.

Key

The message key is exposed via the INCLUDE KEY option. Composite keys are also supported.

INCLUDE KEY AS renamed_id

Note that:

• This option requires specifying the key and value encodings explicitly using the KEY FORMAT ... VALUE FORMAT syntax.

• The UPSERT envelope always includes keys.

• The DEBEZIUM envelope is incompatible with this option.

Headers

Message headers can be retained in Materialize and exposed as part of the source data.

Note that:

• The DEBEZIUM envelope is incompatible with this option.

All headers

All of a message’s headers can be exposed using INCLUDE HEADERS, followed by an AS <header_col>.

This introduces column with the name specified or headers if none was specified. The column has the type record(key: text, value: bytea?) list, i.e. a list of records containing key-value pairs, where the keys are text and the values are nullable byteas.

INCLUDE HEADERS

To simplify turning the headers column into a map (so individual headers can be searched), you can use the map_build function:

SELECT
    id,
    seller,
    item,
    convert_from(map_build(headers)->'client_id', 'utf-8') AS client_id,
    map_build(headers)->'encryption_key' AS encryption_key,
FROM kafka_metadata;

 id | seller |        item        | client_id |    encryption_key
----+--------+--------------------+-----------+----------------------
  2 |   1592 | Custom Art         |        23 | \x796f75207769736821
  3 |   1411 | City Bar Crawl     |        42 | \x796f75207769736821

Individual headers

Individual message headers can be exposed via the INCLUDE HEADER key AS name option.

The bytea value of the header is automatically parsed into an UTF-8 string. To expose the raw bytea instead, the BYTES option can be used.

INCLUDE HEADER 'c_id' AS client_id, HEADER 'key' AS encryption_key BYTES

Headers can be queried as any other column in the source:

SELECT
    id,
    seller,
    item,
    client_id::numeric,
    encryption_key
FROM kafka_metadata;

 id | seller |        item        | client_id |    encryption_key
----+--------+--------------------+-----------+----------------------
  2 |   1592 | Custom Art         |        23 | \x796f75207769736821
  3 |   1411 | City Bar Crawl     |        42 | \x796f75207769736821

Note that:

• Messages that do not contain all header keys as specified in the source DDL will cause an error that prevents further querying the source.

• Header values containing badly formed UTF-8 strings will cause an error in the source that prevents querying it, unless the BYTES option is specified.

Partition, offset, timestamp

These metadata fields are exposed via the INCLUDE PARTITION, INCLUDE OFFSET and INCLUDE TIMESTAMP options.

INCLUDE PARTITION, OFFSET, TIMESTAMP AS ts

SELECT "offset" FROM kafka_metadata WHERE ts > '2021-01-01';

offset
------
15
14
13

Excluding or recasting a field

A Kafka table’s columns are determined entirely by its format (for Avro, the reader schema). To exclude or recast a field, project or cast it in a view on top of the table.

DDL transaction block

For performance, when issuing multiple CREATE TABLE FROM SOURCE... statements, use within a transaction block.

Source-populated tables and snapshotting

Creating the tables from sources starts the snapshotting process. Snapshotting syncs the currently available data into Materialize. Because the initial snapshot is persisted in the storage layer atomically (i.e., at the same ingestion timestamp), you are not able to query the table until snapshotting is complete.

Handling table schema changes

The use of CREATE SOURCE (new syntax) with CREATE TABLE FROM SOURCE allows for the handling of the upstream schema changes, specifically adding or dropping columns, without downtime. For details, see Kafka: Handling upstream schema changes with zero downtime.

Privileges

The privileges required to execute this statement are:

• CREATE privileges on the containing schema.
• USAGE privileges on all types used in the table definition.
• USAGE privileges on the schemas that all types in the statement are contained in.

Examples

Create a table

To create a read-only table from a Kafka source, use the CREATE TABLE ... FROM SOURCE statement. Unlike other source types, a Kafka source exposes a single topic, so the statement does not take a (REFERENCE ...) clause. The following example creates a read-only table orders that decodes the topic as JSON.

/* This example assumes:
  - In Materialize:
    - You have defined a connection to the upstream Kafka broker.
    - You have used the connection to create a source.

   For example (substitute with your configuration):
      CREATE CONNECTION kafka_connection TO KAFKA (
        BROKER '<broker>',            -- substitute
        SECURITY PROTOCOL = SASL_SSL,
        SASL MECHANISMS = 'SCRAM-SHA-256',
        SASL USERNAME = '<user>',     -- substitute
        SASL PASSWORD = SECRET kafka_password
      );

      CREATE SOURCE orders_src
      FROM KAFKA CONNECTION kafka_connection (
        TOPIC 'orders'                -- substitute
      );
*/

CREATE TABLE orders
FROM SOURCE orders_src
FORMAT JSON
;

Creating the tables from sources starts the snapshotting process. Snapshotting syncs the currently available data into Materialize. Because the initial snapshot is persisted in the storage layer atomically (i.e., at the same ingestion timestamp), you are not able to query the table until snapshotting is complete.

To verify that the table has been created, you can run SHOW TABLES to list all tables in the current schema:

SHOW TABLES;

The results should include the orders table:

| name    | comment |
| ------- | ------- |
| orders  |         |

Inspect the table columns using the SHOW COLUMNS command:

SHOW COLUMNS FROM orders;

Because the table decodes the topic as JSON, the message value is exposed as a single data column of type jsonb:

| name | nullable | type  | comment |
| ---- | -------- | ----- | ------- |
| data | false    | jsonb |         |

Once the snapshotting process completes and the table is in the running state, you can query the table. For JSON-formatted data, use the jsonb operators to extract individual fields:

SELECT
    (data->>'id')::bigint AS id,
    data->>'item' AS item,
    (data->>'quantity')::bigint AS quantity
FROM orders
ORDER BY id;

| id | item   | quantity |
| -- | ------ | -------- |
| 1  | widget | 5        |
| 2  | gadget | 2        |

Related pages

• CREATE SOURCE: Kafka/Redpanda (New Syntax)
• DROP TABLE