Skip to content

SQL: Query decoded keys and headers on Iceberg topics #658

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
kbatuigas wants to merge 4 commits into
base: main
Choose a base branch
from
Open

SQL: Query decoded keys and headers on Iceberg topics #658

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
d19a1e0
Add new decode options
kbatuigas Jul 31, 2026
44e2d5d
Draft querying decoded values section
kbatuigas Jul 31, 2026
a33ffc6
Add to What's new
kbatuigas Jul 31, 2026
8b959cf
Specify limitations per SME
kbatuigas Jul 31, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
6 changes: 6 additions & 0 deletions modules/get-started/pages/whats-new-cloud.adoc
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -6,6 +6,12 @@

This page lists new features added to Redpanda Cloud.

== August 2026

=== Query decoded keys and headers from Iceberg topics with SQL

Redpanda SQL can now read decoded record keys and header values when you query Iceberg-enabled topics, instead of only raw bytes. Set `key_decode_mode` and `header_value_type` on `CREATE TABLE` to expose the `redpanda` metadata column's `key` as a decoded struct or string and its header values as text. See xref:sql:query-data/query-iceberg-topics.adoc#query-decoded-keys-and-headers[Query decoded keys and headers].

== July 2026

=== Schema Registry contexts enabled by default
Expand Down
69 changes: 69 additions & 0 deletions modules/reference/pages/sql/sql-statements/create-table.adoc
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -70,13 +70,57 @@ a|Whether records on the topic are encoded with the https://docs.confluent.io/pl
* `'false'`: Records are raw Protobuf or Avro without the wire-format prefix.

Only valid when `schema_lookup_policy = 'LATEST'`.

|`key_decode_mode`
|STRING
|No
a|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. Required when the key schema defines more than one message. Defaults to the first message in the schema.
Comment on lines +91 to +94

@coderabbitai coderabbitai Bot Jul 31, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🎯 Functional Correctness | 🟠 Major | ⚡ Quick win

Resolve the contradictory Protobuf key message-name rule.

Line 94 says key_schema_message_full_name is required for multi-message schemas, then says it defaults to the first message. Choose one behavior and align it with the implementation; otherwise users may omit a required option or rely on an ambiguous message selection.

🤖 Prompt for AI Agents 
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@modules/reference/pages/sql/sql-statements/create-table.adoc` around lines 91
- 94, Resolve the contradictory description for key_schema_message_full_name in
the CREATE TABLE option reference by checking the implementation’s behavior and
documenting either that the option is required for schemas with multiple
messages or that it defaults to the first message. Update the column description
so it states only the implemented rule and removes the conflicting fallback or
requirement.


|`key_confluent_wire_protocol`
|STRING
|No
a|Whether record keys carry the Confluent Schema Registry wire-format prefix. Only valid when `key_decode_mode = 'schema_latest'`. The `schema_id_prefix` mode always requires the prefix.

* `'false'` (default): Keys are raw encoded bytes without the prefix.
* `'true'`: Keys carry the wire-format prefix.

|`header_value_type`
|STRING
|No
a|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]
== 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.
Expand Down Expand Up @@ -116,6 +160,16 @@ Contains Kafka record metadata. Always present on every row.
|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 them, set `key_decode_mode`, `key_schema_subject`, `key_confluent_wire_protocol`, or `header_value_type` (see <<options>>):

Comment on lines +163 to +164

@coderabbitai coderabbitai Bot Jul 31, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🎯 Functional Correctness | 🟡 Minor | ⚡ Quick win

Clarify which options actually enable decoding.

key_decode_mode selects the decoding mode; key_schema_subject, key_schema_message_full_name, and key_confluent_wire_protocol only provide schema-mode configuration. Reword this paragraph so users do not assume setting an auxiliary option alone changes the metadata type.

🤖 Prompt for AI Agents 
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@modules/reference/pages/sql/sql-statements/create-table.adoc` around lines
163 - 164, Revise the paragraph describing raw key and header values so it
states that key_decode_mode enables decoding, while key_schema_subject,
key_schema_message_full_name, and key_confluent_wire_protocol only configure
schema-mode decoding. Clarify that header_value_type controls header decoding,
and avoid implying auxiliary schema options alone change metadata types.

* 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`.
Expand Down Expand Up @@ -179,3 +233,18 @@ WITH (
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:

[source,sql]
----
CREATE TABLE default_redpanda_catalog=>orders
WITH (
topic = 'orders',
schema_subject = 'orders-value',
key_decode_mode = 'schema_id_prefix',
header_value_type = 'string'
);
----
45 changes: 45 additions & 0 deletions modules/sql/pages/query-data/query-iceberg-topics.adoc
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -100,6 +100,51 @@ Redpanda SQL plans the union internally, so you don't write a `UNION ALL`.

NOTE: Iceberg-committed data persists independently of Redpanda topic retention. Queries continue to return records past the Redpanda topic's retention window, provided they were committed to Iceberg first.

== Query decoded keys and headers

By default, the `redpanda` metadata column exposes the record `key` and header values as raw `bytea`. To query them as decoded values, set the key and header options when you map the table. For the full list, see xref:reference:sql/sql-statements/create-table.adoc#options[CREATE TABLE options].

For an Iceberg-enabled topic, the table's key mode must match how the topic encodes keys and headers into its Iceberg table, which is set by the topic's `redpanda.iceberg.mode` property. See xref:manage:iceberg/specify-iceberg-schema.adoc#configure-key-value-and-header-translation[Configure key, value, and header translation]. Set the `CREATE TABLE` options to match:

[cols="<55%,<45%",options="header"]
|===
|Topic `redpanda.iceberg.mode` |`CREATE TABLE` options

|`value_schema_id_prefix`
|Omit the key options. The key stays `bytea`.

|`key:mode=schema_id_prefix;value:mode=schema_id_prefix`
|`key_decode_mode = 'schema_id_prefix'`

|`key:mode=string;value:mode=schema_id_prefix`
|`key_decode_mode = 'string'`

|`headers:value_type=string;value:mode=schema_id_prefix`
|`header_value_type = 'string'`
|===

If the table's key mode doesn't match the topic's Iceberg encoding, the query fails at planning time with a type mismatch instead of returning incorrect data.

Redpanda SQL exposes decoded value fields as top-level columns, so a topic whose `redpanda.iceberg.mode` nests value fields under a `value` struct (`value:layout=nested`) is not supported. The Iceberg table's `value` column has no counterpart in the flattened topic schema, so the query fails at planning time with the `Kafka schema must be a name-superset of Iceberg schema` error described in <<handle-schema-differences>>.

Map the table with the matching options, then access the decoded key and header values in your query:

[source,sql]
----
CREATE TABLE default_redpanda_catalog=>orders WITH (
topic = 'orders',
schema_subject = 'orders-value',
key_decode_mode = 'schema_id_prefix'
);

SELECT ((redpanda).key).customer_id, (redpanda).headers
FROM default_redpanda_catalog=>orders;
Comment on lines +134 to +141

@coderabbitai coderabbitai Bot Jul 31, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🎯 Functional Correctness | 🟡 Minor | ⚡ Quick win

Make the example actually decode headers.

The CREATE TABLE statement configures only key_decode_mode, while header_value_type defaults to binary. Therefore (redpanda).headers still contains raw bytea values, despite the preceding text saying the example accesses decoded key and header values. Add header_value_type = 'string' or change the prose to describe headers as raw bytes.

🤖 Prompt for AI Agents 
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@modules/sql/pages/query-data/query-iceberg-topics.adoc` around lines 134 -
141, Update the CREATE TABLE example for default_redpanda_catalog=>orders to set
header_value_type to 'string', so the subsequent (redpanda).headers query
returns decoded string values and matches the surrounding description.

----

When you decode the key with a schema, `(redpanda).key` is a struct whose fields you access with `((redpanda).key).field_name`. With `key_decode_mode = 'string'`, `(redpanda).key` is `text`. Run `DESCRIBE TABLE default_redpanda_catalog=>orders` to see the decoded key type.

NOTE: `key_decode_mode` and `header_value_type` are fixed for the lifetime of the table. `ALTER` rejects a change to either. To change how keys or header values are decoded, drop and recreate the table.

== Handle schema differences

A topic schema can evolve over time. You might add or remove fields in your Schema Registry value subject as your application changes. Redpanda writes new records to the Iceberg table forward-only: it adds columns to the Iceberg table when the topic schema widens, but it does not drop columns from the Iceberg table when the topic schema narrows. As a result, the Iceberg table can carry columns the current topic schema doesn't have.
Expand Down
Loading