Tributary

What you can do with one query

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

CREATE TABLE orders_snapshot AS
SELECT
  partition,
  "offset",
  json_extract_string(message::VARCHAR, '$.user_id') AS user_id,
  json_extract_string(message::VARCHAR, '$.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. Any librdkafka configuration option — security.protocol, sasl.username, auto.offset.reset, ssl.ca.location — passes through as a named parameter.

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. If a setting works in the Kafka consumer config reference, it works here.

A topic with N partitions is scanned by N parallel consumers, and DuckDB pipelines the resulting Arrow batches into the rest of the query plan. The output schema is fixed: (topic VARCHAR, partition INTEGER, offset BIGINT, message BLOB). The message column is intentionally a BLOB — Tributary doesn’t try to guess your wire format. Cast it (message::VARCHAR) and use DuckDB’s JSON functions, or pair with the avro / protobuf community extensions, depending on what’s on your topic.

Securing credentials

Authentication parameters are plain SQL values, so anything passed inline appears in the cached query plan and EXPLAIN output. Source secrets from the environment via getenv() and bind them in:

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"      := :k_user,
  "sasl.password"      := :k_pass);

getenv runs at parse time, so the literal credential never appears in any cached plan. See Kafka security: SASL and SSL for the upstream guidance on which mechanism fits your cluster.

Scoping the scan

tributary_scan_topic defaults to reading the topic’s current contents — fine for a low-volume topic or a one-off snapshot, but expensive on a high-throughput production topic. Three ways to scope:

• Filter on partition / offset in a WHERE clause. DuckDB pushes these down so partitions you don’t need aren’t consumed at all.
• Use auto.offset.reset := 'latest' so a fresh consumer group starts at the tail of the topic instead of the head.
• Set enable.auto.commit := 'false' for ad-hoc analytics so the read doesn’t advance any production consumer-group offsets you might inadvertently share.

Compared to alternatives

• Kafka Connect — purpose-built for sustained, structured Kafka↔system pipelines. Reach for that when you need delivery guarantees, schema-registry integration, 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 payload as a BLOB. See tributary_scan_topic for the full signature and parameter table.

Decode JSON messages on the way in

Most topics carry JSON. Cast the message column to VARCHAR and use DuckDB’s JSON functions:

SELECT
  partition,
  "offset",
  json_extract_string(message::VARCHAR, '$.user_id') AS user_id,
  json_extract_string(message::VARCHAR, '$.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.

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

Filter while consuming

DuckDB pushes predicates down into the partition-parallel scan — partitions you don’t need aren’t consumed at all:

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

For time-bounded scans, decode the timestamp from the message body and filter there, or use auto.offset.reset := 'latest' so a fresh consumer group starts at the tail.

Snapshot a topic into a local table

CREATE 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(k.message::VARCHAR, '$.user_id') AS user_id,
    json_extract_string(k.message::VARCHAR, '$.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:

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"      := :k_user,
  "sasl.password"      := :k_pass);

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.

Diagnostics

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

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