Tributary

What you can do with one query

The shortest path from “Kafka topic” to “queryable DuckDB table”:

CREATE OR REPLACE TABLE orders_snapshot AS
SELECT
  partition,
  "offset",
  json_extract_string(decode(message), '$.user_id') AS user_id,
  json_extract_string(decode(message), '$.amount')  AS amount
FROM tributary_scan_topic('orders',
  "bootstrap.servers" := 'kafka:9092');

tributary_scan_topic reads each partition in parallel, hands DuckDB the message bytes as a BLOB, and lets you decode them inline with whatever SQL fits your serialization. librdkafka’s client configuration options — security.protocol, sasl.username, ssl.ca.location, client.id — pass through as named parameters.

Architecture

Tributary is a thin DuckDB table-function wrapper around librdkafka, the standard C/C++ Kafka client. Every named parameter you pass to tributary_scan_topic or tributary_metadata is forwarded to librdkafka’s configuration as-is — there’s no Tributary-specific naming layer to learn. The accepted names are librdkafka’s global (client) properties from its CONFIGURATION reference, plus schema.registry.url. Topic-level properties such as auto.offset.reset are not accepted.

Each scan reads every non-empty partition from its first retained offset up to the high watermark it saw when the scan started, so the result is a finite snapshot of the topic even while producers keep writing. Partitions are separate scan tasks that DuckDB’s threads pick up in parallel. The scan doesn’t join a consumer group or commit offsets, so it never moves an application’s consumer-group position.

The output schema is (topic VARCHAR, partition INTEGER, offset BIGINT, key BLOB, message BLOB); a message without a key has a NULL key. The key and message columns are intentionally BLOB — Tributary doesn’t try to guess your wire format. Turn UTF-8 payloads into text with decode(message) and use DuckDB’s JSON functions, or pair with the avro / protobuf community extensions, depending on what’s on your topic. Don’t use message::VARCHAR for JSON: in DuckDB 2.0 a BLOB to VARCHAR cast escapes quotes and non-ASCII bytes ({\x22user_id\x22: ...}), which JSON functions reject. When the topic’s schemas live in a Schema Registry, Tributary can decode JSON-schema payloads for you — see Schema Registry.

Securing credentials

Authentication parameters are plain SQL values, so anything passed inline appears in the query text. Keep them in a tributary_cluster secret, which stores librdkafka options for one cluster and redacts the sensitive ones (sasl.username, sasl.password, ssl.key.password, key and certificate locations) in duckdb_secrets(). The secret’s scope is a kafka:// name, and a scan picks the secret up when its topic is written as kafka://<name>/<topic>:

CREATE SECRET prod_kafka (
  TYPE tributary_cluster,
  SCOPE 'kafka://prod',
  "bootstrap.servers" 'broker.example.com:9093',
  "security.protocol" 'SASL_SSL',
  "sasl.mechanism"    'SCRAM-SHA-512',
  "sasl.username"     getenv('KAFKA_USER'),
  "sasl.password"     getenv('KAFKA_PASSWORD')
);

SELECT * FROM tributary_scan_topic('kafka://prod/orders');

Named parameters passed to the scan override the secret’s values. A plain topic name (no kafka:// prefix) never matches a secret. tributary_metadata doesn’t read secrets; pass its connection options as named parameters.

Without a secret, source credentials from the environment and pass them with getvariable:

SET VARIABLE k_user = getenv('KAFKA_USER');
SET VARIABLE k_pass = getenv('KAFKA_PASSWORD');

SELECT *
FROM tributary_scan_topic('orders',
  "bootstrap.servers"  := 'broker.example.com:9093',
  "security.protocol"  := 'SASL_SSL',
  "sasl.mechanism"     := 'SCRAM-SHA-512',
  "sasl.username"      := getvariable('k_user'),
  "sasl.password"      := getvariable('k_pass'));

See Kafka security: SASL and SSL for the upstream guidance on which mechanism fits your cluster.

Scoping the scan

tributary_scan_topic always reads everything the topic currently retains — fine for a low-volume topic or a one-off snapshot, but expensive on a high-throughput production topic. There is no offset or time range to pass, and a WHERE on partition or offset is applied after the messages are read, so it narrows the result but not the amount consumed. To keep the cost down:

• Check the size first with tributary_metadata — partition counts tell you how wide the scan will be.
• Scan once, query many times. Materialize the topic with CREATE TABLE ... AS SELECT * FROM tributary_scan_topic(...) and run the exploratory queries against the local table.
• Keep retention in mind. The scan starts at the oldest retained offset, so a topic’s retention settings decide how far back it reaches.

Schema Registry

Tributary includes a client for a Confluent-compatible Schema Registry. Every function takes the registry URL as its first argument; credentials come from a tributary_schema_registry secret whose scope is that URL (basic_auth_username + basic_auth_password, or bearer_access_token):

CREATE SECRET registry (
  TYPE tributary_schema_registry,
  SCOPE 'https://registry.example.com',
  basic_auth_username 'svc-analytics',
  basic_auth_password getenv('REGISTRY_PASSWORD')
);

SELECT * FROM tributary_schema_registry_subjects('https://registry.example.com', false);
SELECT * FROM tributary_schema_registry_get_schema_latest('https://registry.example.com', 'orders-value');

• Browse: tributary_schema_registry_subjects, tributary_schema_registry_versions, and the get_schema_* lookups by latest version, version number, subject and id, or GUID.
• Change: tributary_schema_registry_register_schema, tributary_schema_registry_delete_subject, tributary_schema_registry_delete_subject_version.
• Decode while scanning: pass "schema.registry.url" := '...' to tributary_scan_topic (or put it in the tributary_cluster secret). Tributary looks up the latest schema of the <topic>-key and <topic>-value subjects; when one is a JSON schema, that column comes back decoded as JSON — named key and value — instead of a BLOB. A subject that doesn’t exist leaves its column as a BLOB. Only JSON schemas are supported: an Avro or Protobuf subject is an error.
• Encode: tributary_schema_registry_serialize_json encodes a JSON value with a JSON schema into the Schema Registry wire format for a topic, as a BLOB ready to hand to a producer.

Compared to alternatives

• Kafka Connect — purpose-built for sustained, structured Kafka↔system pipelines. Reach for that when you need delivery guarantees, Avro / Protobuf schema handling, or recurring sync. Reach for Tributary when you want SQL on demand.
• kcat + CSV → DuckDB — what people often did before this extension. Tributary collapses the file-in-the-middle and lets you push DuckDB predicates back into the consume.
• Python confluent-kafka + DataFrames — much heavier setup; you give up DuckDB’s vectorized execution against the streamed data and re-introduce row-format conversion at the DataFrame boundary.

Read a Kafka topic

The simplest pattern — point a query at a topic and treat it as a table:

SELECT *
FROM tributary_scan_topic('orders',
                          "bootstrap.servers" := 'kafka:9092');

Each row corresponds to one Kafka message, with metadata columns (topic, partition, offset) plus the raw key and payload (message) as BLOBs. The scan reads everything the topic retains, up to where each partition ended when the scan started. See tributary_scan_topic for the full signature and parameter table.

Decode JSON messages on the way in

Most topics carry JSON. Turn the message bytes into text with decode and use DuckDB’s JSON functions:

SELECT
  partition,
  "offset",
  json_extract_string(decode(message), '$.user_id') AS user_id,
  json_extract_string(decode(message), '$.event')   AS event
FROM tributary_scan_topic('events',
                          "bootstrap.servers" := 'kafka:9092');

offset is a reserved keyword in DuckDB (used by LIMIT … OFFSET …), so quote it as "offset" whenever you reference the column directly.

Use decode(message) rather than message::VARCHAR: in DuckDB 2.0 casting a BLOB to VARCHAR escapes quotes and non-ASCII bytes ({\x22user_id\x22: ...}), which the JSON functions reject.

For Avro / Protobuf payloads, pair with DuckDB’s avro / protobuf community extensions and decode the BLOB column the same way.

Filter a scan

Filter on the metadata columns like any other table:

SELECT *
FROM tributary_scan_topic('orders',
                          "bootstrap.servers" := 'kafka:9092')
WHERE partition = 0
  AND "offset" >= 1000000;

The filter runs after the messages are read: every partition is still consumed from its first retained offset, so this narrows the result, not the cost of the scan. For time-bounded slices, decode the timestamp from the message body and filter on it. If you’ll query the same slice repeatedly, snapshot it into a table first (below).

Snapshot a topic into a local table

CREATE OR REPLACE TABLE orders_snapshot AS
SELECT *
FROM tributary_scan_topic('orders',
                          "bootstrap.servers" := 'kafka:9092');

Useful for ad-hoc analytics over a finite slice of the stream — once it’s a DuckDB table, the rest of your query plan runs at full vectorized speed.

Join Kafka with local data

SELECT k.partition, k."offset", u.name, msg.event
FROM tributary_scan_topic('events',
                          "bootstrap.servers" := 'kafka:9092') k
CROSS JOIN LATERAL (
  SELECT
    json_extract_string(decode(k.message), '$.user_id') AS user_id,
    json_extract_string(decode(k.message), '$.event')   AS event
) msg
JOIN users u ON u.user_id = msg.user_id::BIGINT;

The classic enrichment pattern — Kafka stream on one side, reference data on the other, joined in a single SQL statement.

Authenticate against managed Kafka

Source credentials from the environment so they don’t appear in cached query plans:

CREATE SECRET prod_kafka (
  TYPE tributary_cluster,
  SCOPE 'kafka://prod',
  "bootstrap.servers" 'broker.example.com:9093',
  "security.protocol" 'SASL_SSL',
  "sasl.mechanism"    'SCRAM-SHA-512',
  "sasl.username"     getenv('KAFKA_USER'),
  "sasl.password"     getenv('KAFKA_PASSWORD')
);

-- The kafka://prod/ prefix picks up the secret
SELECT * FROM tributary_scan_topic('kafka://prod/orders');

The tributary_cluster secret holds the librdkafka options for one cluster and redacts the credentials in duckdb_secrets(). Named parameters passed to the scan override the secret’s values. To skip the secret, pass the same options as named parameters, using getvariable('k_user') for values loaded with SET VARIABLE k_user = getenv('KAFKA_USER').

Same pattern for SSL-only clusters: pass security.protocol := 'SSL' plus the relevant ssl.ca.location / ssl.certificate.location parameters. See Kafka SASL and SSL for the upstream config matrix.

Discover what’s on a cluster

-- List every topic the cluster exposes
SELECT unnest(topics).name AS topic_name
FROM tributary_metadata("bootstrap.servers" := 'kafka:9092');

-- Count partitions per topic
SELECT
  t.name                      AS topic,
  array_length(t.partitions)  AS partitions
FROM (
  SELECT unnest(topics) AS t
  FROM tributary_metadata("bootstrap.servers" := 'kafka:9092')
)
ORDER BY partitions DESC;

See tributary_metadata for the full output shape.

Decode JSON-schema topics through a Schema Registry

When producers write JSON-schema payloads through a Confluent-compatible Schema Registry, point the scan at the registry and the key and value come back as JSON:

SELECT partition, "offset", value->>'user_id' AS user_id
FROM tributary_scan_topic('orders',
                          "bootstrap.servers"   := 'kafka:9092',
                          "schema.registry.url" := 'https://registry.example.com');

Tributary uses the latest schema of the orders-key and orders-value subjects; a missing subject leaves that column as a BLOB named key or message. Only JSON schemas are decoded — an Avro or Protobuf subject is an error. Browse the registry itself with the tributary_schema_registry_* functions:

SELECT * FROM tributary_schema_registry_subjects('https://registry.example.com', false);

SELECT version, schema_type, schema
FROM tributary_schema_registry_get_schema_latest('https://registry.example.com', 'orders-value');

Registry credentials go in a tributary_schema_registry secret scoped to the registry URL. See Schema Registry in the technical details.

Diagnostics

SELECT tributary_version();   -- e.g. '20251020.01'

Use tributary_version when you need to confirm the loaded build against bug reports or release notes.