Tributary

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.