Skip to content
DuckDB Extensions All extensions Tributary

Reading functions in the Tributary DuckDB extension

Function category

Reading

1 function

Stream messages from Kafka topics into DuckDB tables. Multi-partition topics are scanned in parallel; the result is `(topic, partition, offset, message)` rows you decode in SQL.

tributary_scan_topic

Table function Reading
Signature
tributary_scan_topic( topic_name: VARCHAR, ... := VARCHAR, bootstrap.servers := VARCHAR, group.id := VARCHAR, sasl.mechanism := VARCHAR, sasl.password := VARCHAR, sasl.username := VARCHAR, security.protocol := VARCHAR ) → TABLE
Arguments
Argument topic_name Type VARCHAR Mode Positional Description Kafka topic to read.
Argument bootstrap.servers Type VARCHAR Mode Named Description Comma-separated host:port list of brokers to bootstrap from. Required.
Argument group.id Type VARCHAR Mode Named Description Consumer group id. See Kafka consumer configs.
Argument security.protocol Type VARCHAR Mode Named Description PLAINTEXT, SSL, SASL_PLAINTEXT, or SASL_SSL. See Kafka security.
Argument sasl.mechanism Type VARCHAR Mode Named Description SASL mechanism — PLAIN, SCRAM-SHA-256, SCRAM-SHA-512, etc. See Kafka SASL.
Argument sasl.username Type VARCHAR Mode Named Description SASL username (PLAIN / SCRAM).
Argument sasl.password Type VARCHAR Mode Named Description SASL password (PLAIN / SCRAM). Pass via getenv() and a SQL variable rather than inlining.
Argument ... Type VARCHAR Mode Named Description Varargs Any other librdkafka configuration name — auto.offset.reset, enable.auto.commit, ssl.ca.location, etc. — passes through unchanged. Tributary doesn't validate keys; misspellings surface as librdkafka errors at scan time.
Returns

A table with the following columns:

Column topic Type VARCHAR Description Topic the message was read from.
Column partition Type INTEGER Description Kafka partition id the message lives in.
Column offset Type BIGINT Description Per-partition offset of the message.
Column message Type BLOB Description Raw message payload as a BLOB. Cast or parse it with DuckDB's JSON / Avro / decode utilities depending on your topic's serialization.
Description

Read messages from a Kafka topic as DuckDB rows. Multi-partition topics are scanned in parallel by partition. Each row is (topic, partition, offset, message); the message is a BLOB you decode in SQL based on your serialization (JSON, Avro, Protobuf, etc.).

For a long-running topic this scans the entire current contents — pair with a WHERE clause on timestamp / partition / offset for finite slices, or with tributary_metadata first to plan the scan.

1 Snapshot a topic into a local table 
CREATE TABLE orders_snapshot AS
SELECT *
FROM tributary_scan_topic('orders',
                          "bootstrap.servers" := 'kafka:9092');
2 Decode JSON messages on the way in 
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');
3 SASL-authenticated consume against a managed Kafka 
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);
Related functions