Cloud

CREATE TABLE

The CREATE TABLE statement maps a Redpanda topic to a SQL table through a catalog. After creating the table, you can query the topic using standard SQL.

You must first create a Redpanda catalog connection before creating tables. CREATE TABLE in Redpanda SQL maps Redpanda topics to SQL tables and does not create standalone tables with user-defined schemas.

Syntax

CREATE TABLE [IF NOT EXISTS] catalog_name=>table_name
WITH (option = 'value' [, ...]);

  • catalog_name: Name of an existing Redpanda catalog.

  • table_name: Name for the new table.

  • IF NOT EXISTS: Optional. Prevents an error if a table with the same name already exists in the catalog.

Options

Option Type Required Description

topic

STRING

Yes

Name of the Redpanda topic to map to this table.

schema_subject

STRING

No

Schema Registry subject name to use for deserializing topic data. Defaults to the topic-name strategy (<topic>-value).

schema_lookup_policy

STRING

No

How to resolve the schema version. LATEST is the only supported value.

error_handling_policy

STRING

No

How to handle records that fail deserialization.

  • FAIL (default): Raises an error.

  • FILL_NULL: Replaces failed fields with NULL.

  • DROP_RECORD: Skips the record.

struct_mapping_policy

STRING

No

How to map nested structures from the topic schema to SQL columns.

  • COMPOUND (default): Maps each nested structure to a user-defined type with named fields, queryable using (column).field_name syntax. Cyclic types are not supported in COMPOUND mode. Use JSON for recursive schemas. See row for the field-access syntax.

  • JSON: Stores each nested structure as a JSON value. Required for recursive (cyclic) types.

output_schema_message_full_name

STRING

No

Full Protobuf message name for the value schema. Defaults to the first message declared in the schema. Set this when the topic’s records use a message type other than the first declared one; otherwise decoding fails with a message-type mismatch error.

confluent_wire_protocol

STRING

No

Whether records on the topic are encoded with the Confluent Schema Registry wire format (a magic byte followed by a 4-byte schema ID before the payload).

  • 'true' (default): Records carry the Confluent wire-format prefix. Use this for topics whose values were produced by a Schema-Registry-aware client.

  • 'false': Records are raw Protobuf or Avro without the wire-format prefix.

Only valid when schema_lookup_policy = 'LATEST'.

key_decode_mode

STRING

No

How to interpret record keys. Applies to both live topic records and Iceberg-committed records.

  • binary (default): Exposes the key as raw bytea.

  • string: Decodes the key as a UTF-8 string. Invalid bytes are replaced with the Unicode replacement character (U+FFFD).

  • schema_latest: Decodes every key with the latest schema registered for the key subject.

  • schema_id_prefix: Decodes each key using the schema ID embedded in the record’s Confluent wire-format header.

Fixed for the table’s lifetime. ALTER cannot change it; recreate the table instead.

key_schema_subject

STRING

No

Schema Registry subject for the key schema. Defaults to the topic-name strategy (<topic>-key). Override when the producer does not use the topic-name strategy.

key_schema_message_full_name

STRING

No

Full Protobuf message name for the key schema. Defaults to the first message declared in the schema. Set this when the topic’s key records use a message type other than the first declared one; otherwise decoding fails with a message-type mismatch error.

key_confluent_wire_protocol

STRING

No

Whether record keys carry the Confluent Schema Registry wire-format prefix. Only valid when key_decode_mode = 'schema_latest'.

  • 'false' (default): Keys are raw encoded bytes without the prefix.

  • 'true': Keys carry the wire-format prefix.

The schema_id_prefix mode always requires the prefix, so this option does not apply.

header_value_type

STRING

No

How to store record header values.

  • binary (default): Exposes header values as raw bytea.

  • string: Decodes header values as UTF-8 strings. Invalid bytes are replaced with U+FFFD.

Header keys are always text.

Fixed for the table’s lifetime. ALTER cannot change it; recreate the table instead.

Auto-added columns

Every catalog-mapped table includes two struct columns in addition to the columns derived from the topic’s schema. Redpanda SQL adds these columns to both Kafka-backed and Iceberg-backed tables. The names redpanda and redpanda_raw are reserved. A topic schema cannot define columns with these names.

REFRESH also rejects a schema whose declared type names begin with __redpanda_, a prefix reserved for Redpanda’s internal metadata types. For JSON schemas, a top-level property named __json_root is reserved and rejected as well.

redpanda

Contains Kafka record metadata. Always present on every row.

Field Type Nullable Description

partition

int

No

Partition the record was read from.

offset

bigint

No

Offset of the record within its partition.

timestamp

timestamp with time zone

Yes

Record timestamp.

headers

Array of struct {key TEXT, value BYTEA}

Yes

Record headers, as an array where each element is a struct of header name and value bytes.

key

bytea

Yes

Record key bytes.

timestamp_type

int

Yes

Kafka timestamp type code. 0 for CreateTime, 1 for LogAppendTime. NULL when not available.

By default, key is raw bytea and each header value is raw bytea. To decode the key, set key_decode_mode (the schema-mode options key_schema_subject, key_schema_message_full_name, and key_confluent_wire_protocol only configure decoding when key_decode_mode uses a schema mode). To decode header values, set header_value_type. See Options:

  • When you decode the key with a schema mode, key becomes a struct that you access with ((redpanda).key).field_name.

  • When key_decode_mode = 'string', key is text.

  • When header_value_type = 'string', each header value is text. Header keys are always text.

A NULL key stays NULL. A key that fails to decode follows the table’s error_handling_policy. Run DESCRIBE TABLE <catalog>⇒<table> to see the decoded key type.

The key and value can use different schemas and formats (for example, an Avro key with a Protobuf value), and key and value field names can overlap without conflict.

redpanda_raw

Populated only when error_handling_policy = 'FILL_NULL' and a record fails to decode. In all other cases, redpanda_raw is NULL.

Use redpanda_raw as a dead-letter pattern. Rows whose value fails schema deserialization remain queryable, with the malformed payload preserved for inspection or reprocessing.

Field Type Nullable Description

key

bytea

Yes

Raw record key bytes.

value

bytea

Yes

Raw record value bytes that failed to decode.

Examples

Map a topic to a table

Map the transactions topic to a table through default_redpanda_catalog:

CREATE TABLE default_redpanda_catalog=>transactions
WITH (
  topic = 'transactions',
  schema_subject = 'transactions-value'
);

Create a table from a multi-message Protobuf schema

When the Protobuf schema for the topic defines more than one message, specify the message to use with output_schema_message_full_name:

CREATE TABLE default_redpanda_catalog=>orders
WITH (
  topic = 'orders',
  schema_subject = 'orders-value',
  output_schema_message_full_name = 'com.example.orders.Order'
);

Create a table with error handling

Map a topic and skip records that fail deserialization:

CREATE TABLE IF NOT EXISTS default_redpanda_catalog=>sensor_readings
WITH (
  topic = 'sensor-data',
  schema_subject = 'sensor-data-value',
  error_handling_policy = 'DROP_RECORD'
);

Decode record keys and headers

Map a topic, decoding keys with the schema ID embedded in each record and storing header values as strings:

CREATE TABLE default_redpanda_catalog=>orders
WITH (
  topic = 'orders',
  schema_subject = 'orders-value',
  key_decode_mode = 'schema_id_prefix',
  header_value_type = 'string'
);