SQLFlow: DuckDB for Streaming Data.

Quickstart | Tutorials | | Documentation

SQLFlow is a stream processing engine that lets you define pipelines with just SQL. It consumes a stream, runs DuckDB SQL over each batch, and writes the result out. Think of it as a lightweight, single-binary Flink.

Sources: Kafka, WebSockets, webhooks.
Sinks: Kafka, ClickHouse, Iceberg, the console, or anything DuckDB can COPY to (PostgreSQL, S3, parquet, MotherDuck, DuckLake).
Built on DuckDB and Apache Arrow: ~900k messages/sec on a laptop, in about a quarter GiB of memory.
One Go binary, one Docker image, one YAML file per pipeline. No cluster.
Coverage and invariant matrix: what is tested, and separately, what is proven — regenerated from the suites on every push.

Quick Start (Getting Started in 5 Minutes)

Build a binary (see Installation for prebuilt binaries and Docker):

make sqlflow

Validate a pipeline against test data, without a broker. dev invoke runs the config's handler over a JSONL fixture and prints the result:

./bin/sqlflow dev invoke dev/config/examples/basic.agg.mem.yml dev/fixtures/simple.json

{"city":"New York","city_count":28672}
{"city":"Baltimore","city_count":28672}

Start Kafka locally:

docker-compose -f dev/kafka-single.yml up -d

Publish test messages. The publisher is a Python script. uv reads uv.lock and builds the environment on first run:

uv run python cmd/publish-test-data.py --num-messages=10000 --topic="input-simple-agg-mem"

Start a Kafka consumer to watch the output:

docker exec -it kafka1 kafka-console-consumer --bootstrap-server=kafka1:9092 --topic=output-simple-agg-mem

Run the pipeline:

./bin/sqlflow run -c dev/config/examples/basic.agg.mem.yml --max-msgs=10000

The consumer prints one row per city:

{"city":"San Francisco","city_count":177}
{"city":"New York","city_count":236}
{"city":"Miami","city_count":203}
{"city":"Baltimore","city_count":180}

Installation

sqlflow reaches DuckDB through the Arrow ADBC driver manager, which dlopens libduckdb at runtime. The binary is not standalone: wherever you run it, that shared library has to be present. SQLFLOW_DUCKDB_LIB points at it; without that variable sqlflow looks in /opt/homebrew/lib/libduckdb.dylib on macOS and /usr/local/lib/libduckdb.so on Linux.

The pinned DuckDB version lives in one place, the DUCKDB_VERSION file.

Docker (no libduckdb setup)

The published image bakes in a matching libduckdb.so and sets SQLFLOW_DUCKDB_LIB, so nothing else is needed. Images are multi-arch (linux/amd64, linux/arm64):

docker run --rm \
  -v $(pwd)/dev:/tmp/conf \
  turbolytics/sql-flow:latest \
  dev invoke /tmp/conf/config/examples/basic.agg.mem.yml /tmp/conf/fixtures/simple.json

To build the image from the repo, make sqlflow-image — it prints the tag it built, derived from git describe.

Prebuilt binary

Release binaries are published for linux and macOS on amd64 and arm64. Download the one matching your platform, then install a matching libduckdb:

chmod +x sqlflow_<version>_<os>_<arch>
./scripts/install-libduckdb.sh /usr/local/lib
export SQLFLOW_DUCKDB_LIB=/usr/local/lib/libduckdb.so
./sqlflow_<version>_<os>_<arch> version

scripts/install-libduckdb.sh always fetches the linux libduckdb.so, for the architecture it detects. It is for linux hosts and containers only. On macOS, brew install duckdb puts the library at the default path and no environment variable is needed.

From source

Requires Go 1.25+, a C toolchain (cgo is mandatory), and libduckdb.

# macOS
brew install duckdb
make sqlflow

# linux
./scripts/install-libduckdb.sh /usr/local/lib
export SQLFLOW_DUCKDB_LIB=/usr/local/lib/libduckdb.so
make sqlflow

The binary lands at bin/sqlflow.

How SQLFlow Works

A pipeline has three parts:

Source — Kafka, a WebSocket, or webhooks, modelled as a stream of messages.

Handler — DuckDB SQL executed over a batch of that stream: filter, aggregate, enrich, or drop data.

Sink — where the SQL result goes: Kafka, ClickHouse, Iceberg, the console, or — through the sqlcommand sink — anywhere DuckDB can write.

A config file names all three, plus optional commands run before the pipeline starts (attaching databases, creating tables) and optional tables the engine manages across the pipeline's lifetime:

SQLFlow Use-Cases

Streaming Data Transformations: Clean data and types and publish the new data (example config).
Stream Enrichment: Add data to an input stream and publish the new data (example config).
Data aggregation: Aggregate input data batches to decrease data volume (example config).
Tumbling Window Aggregation: Bucket data into arbitrary time windows (such as "hour" or "10 minutes") (example config).
Run SQL against the Bluesky Firehose: Execute SQL against any websocket source, such as the Bluesky firehose (example config).
Stream Data to Iceberg: Stream writes to an Iceberg catalog.
Stream Data to ClickHouse: Insert stream processing outputs into ClickHouse (example config).
Enrich Streams with Postgres Data: Query postgres during stream processing to enrich stream data.
Sink Kafka to Postgres: Insert stream processing outputs into postgres.

CLI Reference

sqlflow [command]

Command	Purpose
`run`	Run a pipeline against a live source
`validate`	Check a pipeline offline and report every fault at once
`dev invoke`	Run a pipeline's handler against a static file
`config validate`	Validate a config against the JSON Schema
`config example`	Print a commented example configuration
`tail`	Print every message from a config's source
`version`	Print version, commit and Go version

`sqlflow run`

Runs the pipeline: consume, batch, execute SQL, sink, commit offsets.

sqlflow run <config> [flags]
sqlflow run -c <config> [flags]

Flag	Default	Description
`-c`, `--config`	(required)	Path to the config file, unless given positionally
`--max-msgs`	`0`	Stop after N messages; `0` is unlimited. `--max-msgs-to-process` is an alias
`--metrics`	(off)	Metrics exporter. Only `prometheus` is supported; serves `/metrics` on `:8000`
`--stats-json`	(off)	Write final run stats as JSON to this path
`--pprof`	`false`	Serve pprof on `:6060`, and enable block/mutex profiling

--stats-json writes a small object, useful for CI assertions:

$ sqlflow run -c dev/config/examples/benchmark.structured.mem.yml \
    --max-msgs=2000 --stats-json=/tmp/stats.json
...
{"messages_consumed":2000,"num_errors":0}

`sqlflow validate`

Checks a pipeline without running it. validate reaches no broker and no sink, executes nothing from the commands block, and reports every fault it finds in one pass rather than stopping at the first.

sqlflow validate <config> [--json]

It renders the config's template, then checks the result against the config schema. Both halves of the template's variable use are reported, which is what makes a misspelled name obvious:

$ sqlflow validate pipeline.yml
pipeline.yml:8:49: error: [user.config.template_undefined] template variable
  SQLFLOW_AZURE_CONNECTION_STRING is not defined and renders as an empty
  string, but a similar name is supplied and never read
    did you mean: [SQLFLOW_AZURE_STORAGE_CONNECTION_STRING]
supplied but never read: [SQLFLOW_AZURE_STORAGE_CONNECTION_STRING]

A variable the config reads with no default filter is a required input. When it is unset, validate warns rather than failing, so a config still checks in CI where no secret is set. A missing name that closely resembles a supplied one is a different matter: that resemblance is evidence of a typo, and it fails.

--json emits the same report as a document, with a checks array beside the diagnostics. A check that could not run reports skipped with a reason and never reports pass, so a consumer can tell "checked and fine" from "not checked".

Exit codes follow the error taxonomy: 0 when the config is sound, 10 for a fault the user has to fix.

`sqlflow dev invoke`

Runs the config's commands, tables and handler over a JSONL fixture and prints the resulting rows to stdout, one JSON object per line. The sink is deliberately not exercised, so this is safe to run against a production config. This is the fastest way to iterate on SQL.

sqlflow dev invoke <config> <fixture>

`sqlflow config validate`

Renders the config's Jinja2 template, then validates the result against the JSON Schema.

$ sqlflow config validate dev/config/examples/basic.agg.mem.yml
dev/config/examples/basic.agg.mem.yml: valid

`sqlflow config example`

Prints a fully commented YAML skeleton generated from the schema — every key, its description, and the accepted enum values.

`sqlflow tail`

Connects a config's source and prints every message to stdout, with no handler or sink. Useful for confirming a source is configured correctly.

sqlflow tail -c <config>

`sqlflow version`

$ sqlflow version
sqlflow v1.0.4
commit: 55c3129
go:     go1.25.5

Configuration

A config is a YAML file rendered as a Jinja2 template before it is parsed. Every SQLFLOW_* environment variable is injected into the template context under its own name, which is how configs stay portable across environments:

brokers: [{{ SQLFLOW_KAFKA_BROKERS|default('localhost:9092') }}]

Two extra template variables are always defined: STATIC_ROOT (from SQLFLOW_STATIC_ROOT, default /tmp/sqlflow/static) and SQL_RESULTS_CACHE_DIR (from SQLFLOW_SQL_RESULTS_CACHE_DIR, default /tmp/sqlflow/resultscache).

Parsing is strict: an unknown key is an error rather than a silently ignored setting.

The top-level shape:

commands:   # optional: SQL run once, before the pipeline starts
tables:     # optional: tables created at startup, optionally window-managed
pipeline:   # required
  name:
  description:
  batch_size:              # required, >= 1
  flush_interval_seconds:  # optional; unset means only batch_size triggers a batch
  state:                   # optional; makes state durable, see Durable state
  on_error:                # optional
  source:                  # required
  handler:                 # required
  sink:                    # required

batch_size is how many messages accumulate before the handler runs. flush_interval_seconds bounds the wait: once the interval elapses a partial batch is executed anyway, so a low-traffic topic still makes progress. When --max-msgs ends a run, the final partial batch is executed on exit.

Sources

Kafka

source:
  type: kafka
  kafka:
    brokers: [localhost:9092]
    group_id: my-consumer-group
    auto_offset_reset: earliest   # or latest
    topics:
      - input-topic

Offsets are committed after the batch has been handled and the sink has flushed, and only up to the last message the pipeline actually processed, not to wherever the consumer has read ahead to. See Delivery guarantees.

SASL / TLS. Set security_protocol to one of PLAINTEXT, SSL, SASL_PLAINTEXT, SASL_SSL:

source:
  type: kafka
  kafka:
    brokers: [localhost:9093]
    group_id: test
    auto_offset_reset: earliest
    security_protocol: SASL_SSL
    ssl:
      ca_location: /certs/ca-cert.pem
      certificate_location: /certs/client-cert.pem
      key_location: /certs/client-key.pem
      key_password: testpass
      endpoint_identification_algorithm: 'none'   # disables hostname verification
    sasl:
      mechanism: PLAIN            # or SCRAM-SHA-256, SCRAM-SHA-512
      username: user
      password: bitnami
    topics:
      - input-sasl-tls-1

The same security_protocol / ssl / sasl block works on the Kafka sink. See kafka.sasl-tls.yml.

Two limits, both of which fail loudly rather than silently: GSSAPI is rejected, and an encrypted PEM key is rejected with instructions to convert it (openssl pkcs8 -topk8 -nocrypt). key_password only covers unencrypted PEMs.

Kafka metadata. A Kafka source exposes kafka_topic, kafka_partition and kafka_offset to InferredMemBatch handler SQL, if the SQL selects them.

WebSocket

source:
  type: websocket
  websocket:
    uri: wss://jetstream2.us-east.bsky.network/subscribe?wantedCollections=app.bsky.feed.post

Reconnects with backoff. See the bluesky examples.

Webhook

Listens for POST /events on 0.0.0.0:8001 (not configurable) and optionally validates an HMAC-SHA256 signature:

source:
  type: webhook
  webhook:
    signature_type: hmac
    hmac:
      header: 'X-Hub-Signature-256'
      sig_key: 'sha256'
      secret: "{{ SQLFLOW_GITHUB_WEBHOOK_SECRET }}"

Responds 200 on accept, 400 for a missing signature, 403 for an invalid one.

Known gotcha: the JSON Schema only enumerates kafka and websocket, so sqlflow config validate rejects a webhook config that sqlflow run accepts.

Handlers

All three handlers take sql. The batch is exposed to that SQL as a table.

`type`	Batch table	Notes
`handlers.InferredMemBatch`	`batch`	Schema inferred from the JSON, in memory
`handlers.InferredDiskBatch`	`batch`	Same, but buffered through disk via `read_json_auto`
`handlers.StructuredBatch`	the table named by `table`	Schema declared up front; fastest

Inferred handlers derive the Arrow schema from the messages themselves:

The column set comes from the first message in the batch. A top-level key that first appears in a later message is not a column.
Nested struct fields are unioned across the whole batch: a key inside an object that appears only in a later message still becomes a field, null in the rows that lack it. Without this, whether a nested field existed would depend on which message happened to arrive first.
Types are promoted across the batch (int widens to double), JSON arrays become lists (of scalars, structs, or further lists), and JSON string escapes are decoded. A value that cannot be promoted fails the batch.

handler:
  type: 'handlers.InferredMemBatch'
  sql: |
    SELECT properties.city as city, count(*) as city_count
    FROM batch
    GROUP BY city

InferredDiskBatch additionally accepts sql_results_cache_dir (default /tmp/sqlflow/resultscache). It stages fixed filenames there, so two pipelines must not share a cache directory.

StructuredBatch takes a table you declared in commands, and parses directly into that schema. This is the fastest handler — no inference, and a zero-copy Arrow ingest. The table is truncated at the start of every batch.

commands:
  - name: create source buffer table
    sql: |
      CREATE TABLE source (
        event STRING,
        properties STRUCT(city TEXT)
      );

pipeline:
  handler:
    type: "handlers.StructuredBatch"
    table: source
    sql: |
      SELECT properties.city as city, COUNT(*) as count
      FROM source
      GROUP BY properties.city

Sinks

sink.type is one of console, kafka, sqlcommand, iceberg, clickhouse, noop. An omitted type falls back to console.

# console — one JSON object per row on stdout
sink:
  type: console

# noop — discard (used for benchmarking)
sink:
  type: noop

# kafka — one JSON message per row; Flush blocks on broker acks
#         before offsets are committed
sink:
  type: kafka
  kafka:
    brokers: [localhost:9092]
    topic: output-topic
    # security_protocol / ssl / sasl as per the Kafka source

# clickhouse — the table must exist; columns are matched by name
sink:
  type: clickhouse
  clickhouse:
    dsn: clickhouse://default:@localhost:8123/default          # self-hosted, HTTP
    # dsn: clickhouses://default:<pw>@<host>.clickhouse.cloud:8443/default   # Cloud, TLS
    table: events

# iceberg — catalog resolved from .pyiceberg.yaml, exactly as pyiceberg does
sink:
  type: iceberg
  iceberg:
    catalog_name: default
    table_name: default.city_events

ClickHouse. The DSN scheme picks the protocol: clickhouse:// is HTTP on 8123, clickhouses:// is HTTP over TLS on 8443 (ClickHouse Cloud), tcp:// / natives:// the native protocol on 9000 / 9440. Scalars, DateTime, Date, Enum, LowCardinality, Nullable and Array(T) columns all map; UUID, IPv4 and Decimal accept their textual form as a string. Map and Tuple are not supported. Measured throughput is ~50k rows/sec into a local ClickHouse from one process.

Iceberg. SQL-backed catalogs only (sqlite://); a REST catalog is an error.

sqlcommand is the general escape hatch: the batch is exposed to your SQL as the table sqlflow_sink_batch, and DuckDB does the writing. This is how to reach PostgreSQL, S3, local parquet, MotherDuck and DuckLake:

sink:
  type: sqlcommand
  sqlcommand:
    substitutions:
      - var: $sqlflow_uuid
        type: uuid4        # the only supported substitution type
    sql: |
      COPY sqlflow_sink_batch
        TO '/tmp/sqlflow/out/$sqlflow_uuid.parquet'
      (FORMAT 'parquet');

sink.format.type: parquet is parsed and ignored.

Sink retries

ClickHouse and Iceberg flushes retry when the destination is not answering. Omit the block to accept the defaults. Set max_attempts: 1 to turn retrying off. The Kafka sink ignores this block: franz-go already retries a produce with its own backoff.

sink:
  type: clickhouse
  retry:
    max_attempts: 4           # total attempts, including the first
    initial_backoff_ms: 100   # doubles each attempt
    max_backoff_ms: 2000      # ceiling on the backoff
    deadline_seconds: 10      # bounds the whole ladder, not one attempt

The values shown are the defaults.

Keep deadline_seconds below pipeline.flush_interval_seconds. The retry runs inside the open state transaction, and a ladder that outlives the flush interval freezes the window clock.

The ladder retries only what another attempt could change. The error code decides:

Error	Retried	Why
Any `user.*` code, including `user.sink.encode_failed` and `user.sink.type_unsupported`	No	The config, SQL or a value is wrong. It fails identically every time.
`system.sink.write_failed`	No	The destination answered and refused the write.
`system.sink.unreachable`	Yes	The destination may come back.
An error with no code	Yes	A driver's timeout or reset arrives unclassified. The deadline bounds the cost.
Any other `system.*` code	Yes

A failure that is not retried keeps its own code and exit code. A retried failure that outlasts the ladder is reported as system.sink.unreachable, exit 12, and sink_retry_count_total counts each attempt after the first.

user.sink.encode_failed is the sink's client refusing a value before anything reaches the network, such as a timestamp string the ClickHouse driver's DateTime layout cannot parse. Cast or format the column in the handler SQL. The message names the column and quotes the value.

Error policies

pipeline.on_error.policy is RAISE (default), IGNORE, or DLQ. It is applied at both the handler.write and handler.invoke phases.

pipeline:
  on_error:
    policy: DLQ
    dlq:                 # a full sink definition, any sink type
      type: kafka
      kafka:
        brokers: [localhost:9092]
        topic: dlq-topic

DLQ without a dlq block is a startup error. DLQ records carry four string columns: error, message, phase (handler.write or handler.invoke) and timestamp. See kafka.dlq.yml. source.error.policy is parsed but unused; use pipeline.on_error.

Durable state

A pipeline that aggregates needs its state to survive a restart. Give it a file:

pipeline:
  state:
    path: /var/lib/sqlflow/state.db

DuckDB then runs on that file instead of in memory, and a sqlflow_offsets table lives there beside the tables your handler writes. Every batch commits the handler's writes and the Kafka offsets that produced them in one transaction. On startup the pipeline reads those offsets and resumes from them.

Without a state path, DuckDB runs in memory. A crash mid-window loses that window's aggregate while its offsets are already committed, so the consumer group reports no lag and a restart replays nothing. Set a state path for any pipeline with a tables block.

Two consequences worth knowing:

The state file is the source of truth. When it disagrees with the consumer group, the pipeline resumes from the file's offsets. It applies them while joining the group, so a restart works even while the previous process is still a member, which it is for the session timeout after a crash.
Window close predicates are evaluated at most one flush_interval_seconds behind the wall clock. A stateful pipeline holds one transaction open per batch and DuckDB's now() is the transaction's start time, so the pipeline commits on the flush tick even when idle to keep that clock moving.
DuckDB locks the file exclusively. One state file belongs to one running pipeline, and no other process can read it, not even read-only. Use the /stats endpoint to inspect a running pipeline.

Durable state costs throughput. See What durable state costs for the numbers and the batch size to use.

Delivery guarantees

SQLFlow gives two guarantees. Which one applies depends on where the data lands.

Pipeline state is exactly-once relative to offsets. With a state path set, the tables your handler writes and the offsets that produced them commit in a single transaction. A crash replays exactly the batches whose state did not commit, so a windowed aggregate is neither short nor double-counted across a restart.

External sinks are at-least-once. Kafka, ClickHouse, Iceberg and sqlcommand are flushed before that transaction commits. A crash in between replays the batch and the sink sees those rows twice. Committing first would move offsets past rows the sink never received, which loses them silently, so the duplicate is the deliberate choice. Use a sink that absorbs it: ReplacingMergeTree in ClickHouse, an upsert in sqlcommand, or a downstream dedupe on a key.

Window sinks are at-least-once for the same reason. A tumbling window is published before its rows are deleted, and that delete commits with the pipeline's next batch. A crash in between republishes the window.

Without a state path there is no state guarantee at all: handler state does not survive the process.

Graceful shutdown

SIGTERM and SIGINT drain the pipeline. The drain runs in order:

Stop consuming.
Write the batch it had buffered.
Run each table manager's final poll.
Commit state and offsets.
Exit 0.

A supervisor that stops a pipeline this way loses nothing. Skipping the drain loses no data either, because the buffered batch replays from the last committed offset. It costs that duplicate work, and it republishes any window that closed during shutdown.

The drain has no deadline. A sink that blocks holds the process open until the supervisor escalates to SIGKILL.

Tumbling windows

A table declared under tables.sql can carry a manager, which polls the table on an interval, publishes the closed windows to its own sink, and then deletes them. The handler SQL keeps the window table up to date with an upsert:

tables:
  sql:
    - name: agg_cities_count
      sql: |
        CREATE TABLE agg_cities_count (
          bucket TIMESTAMPTZ, city VARCHAR, count INT
        );
        CREATE UNIQUE INDEX daily_cities_count_idx ON agg_cities_count (bucket, city);
      manager:
        tumbling_window:
          poll_interval_seconds: 10      # optional, default 10
          collect_closed_windows_sql: |
            SELECT ... FROM agg_cities_count
            WHERE bucket < (now()::timestamptz - INTERVAL '60' SECOND)
          delete_closed_windows_sql: |
            DELETE FROM agg_cities_count
            WHERE bucket < (now()::timestamptz - INTERVAL '60' SECOND)
        sink:
          type: kafka
          kafka:
            brokers: [localhost:9092]
            topic: output-tumbling-window-1

Collect, write and flush happen before the delete, so a sink failure retries rather than dropping a window. A retry re-sends rows the sink already received, so give a window sink a key it can deduplicate on. One final poll runs on shutdown, against a current clock and with its delete committed, so a window that closes during shutdown is published once and not republished on the next start.

Windows close on wall-clock time, as written in your collect_closed_windows_sql. There is no event-time watermarking and no late-arrival policy: a message that arrives after its window closed lands in whichever window its own SQL puts it in.

State in a managed table is lost on a crash unless the pipeline sets state.path.

tumbling_window is currently the only manager type. See tumbling.window.yml and kafka.stateful.window.yml.

Metrics

--metrics prometheus serves /metrics on :8000. Nineteen instruments are exported, all under the meter name sqlflow except the two webhook ones.

The instrument name and the Prometheus series name differ: the exporter appends the unit, then _total for counters, and skips the unit when the name already contains it. Only the series name is queryable, so both are listed. A test asserts this table against the running exporter.

Row accounting. Four counts follow a row through the pipeline:

Instrument	Series	Type	Attrs
`message_count`	`message_count_messages_total`	counter	—
`handler_rows_read`	`handler_rows_read_total`	counter	—
`sink_rows_accepted`	`sink_rows_accepted_total`	counter	`sink`, `role`
`sink_rows_written`	`sink_rows_written_total`	counter	`sink`, `role`
`sink_flush_num_rows`	`sink_flush_num_rows`	gauge	—

Sink health:

Instrument	Series	Type	Attrs
`sink_flush_count`	`sink_flush_count_flushes_total`	counter	`result`
`sink_flush_latency`	`sink_flush_latency_seconds`	histogram	—
`sink_retry_count`	`sink_retry_count_total`	counter	`sink`

Pipeline:

Instrument	Series	Type	Attrs
`batch_processing_latency`	`batch_processing_latency_seconds`	histogram	—
`error_count`	`error_count_total`	counter	`class`, `domain`, `code`, `phase`
`phase_duration`	`phase_duration_seconds`	histogram	`phase`

phase_duration decomposes batch time. It carries the same six phases error_count does — handler.write, handler.invoke, sink.write, sink.flush, state.commit, handler.init — so one query says where the time went:

sum(rate(phase_duration_seconds_sum[5m])) by (phase)

Read it with error_count to tell slow from broken. phase_duration records whether or not the phase succeeded, which no other latency here does: sink_flush_latency and state_commit_latency record after their error returns, so a sink that takes thirty seconds to fail leaves them flat.

handler.write is measured around the whole message loop rather than each write, so it also carries the loop's own bookkeeping. Timing each write cost 4.1x on that path; see BenchmarkConsumeLoopWritePath.

Every latency histogram shares one set of bucket boundaries, from 100 microseconds to 60 seconds. The OTel SDK's defaults are millisecond-shaped and these instruments record seconds, so before this every histogram_quantile over them returned a number under five seconds and meant nothing.

Source:

Instrument	Series	Type	Attrs
`source_read_latency`	`source_read_latency_seconds`	histogram	—
`consumer_lag`	`consumer_lag_messages`	gauge	`topic`, `partition`

Reference tables, recorded once at startup for each table the handler SQL joins, with or without a state path:

Instrument	Series	Type	Attrs
`reference_table_rows`	`reference_table_rows`	gauge	`table`

Durable state, present only when the pipeline declares a state path. An absent series and an empty state are different facts, so a pipeline with state in memory reports nothing rather than zero:

Instrument	Series	Type	Attrs
`state_commit_count`	`state_commit_count_commits_total`	counter	`result`
`state_commit_latency`	`state_commit_latency_seconds`	histogram	—
`state_db_size_bytes`	`state_db_size_bytes`	gauge	—
`state_table_rows`	`state_table_rows`	gauge	`table`

Webhook source, under the meter sqlflow.sources.http and present only with a webhook source:

Instrument	Series	Type	Attrs
`webhook_requests_total`	`webhook_requests_total`	counter	`status_code`
`webhook_request_duration_seconds`	`webhook_request_duration_seconds`	histogram	`status_code`

$ sqlflow run -c <config> --metrics=prometheus &
$ curl -s localhost:8000/metrics | grep message_count
message_count_messages_total{otel_scope_name="sqlflow",...} 154635

consumer_lag is the one to alert on. It is the broker's high watermark minus the offset the pipeline has finished with, so it measures the work the pipeline still owes rather than what its consumer group has been told.

Reading the row counts

Each adjacent ratio isolates one kind of loss:

handler_rows_read over message_count — parse and DLQ loss. A message the handler rejected never reaches the SQL.
sink_rows_accepted over handler_rows_read — whatever the SQL does. A join that drops, a WHERE, a GROUP BY. Its meaning depends on the pipeline, which is why sqlflow reports the numbers and leaves the threshold to you.
sink_rows_written over sink_rows_accepted — delivery loss. A ratio that stays below 1 is a sink that is not draining.

The role attribute separates the pipeline sink from the DLQ and from a window manager's sink. Sum across roles and a rejected record counts as a delivered one.

Every ratio is a floor, not an equality. sqlflow is at-least-once: a crash between the flush and the offset commit replays the batch, and the sink writes those rows again, so a ratio can exceed 1 after a normal recovery. Alert on a ratio that is low. Never alert on one that is not exactly 1, or a healthy restart pages somebody.

The sink's buffer depth is sink_rows_accepted_total - sink_rows_written_total. There is no gauge for it: a counter difference gives the depth and a rate, and it cannot misreport itself the way a sink's own count can.

An enrichment pipeline joining a dimension table that never loaded shows up here as sink_rows_accepted collapsing against handler_rows_read — and at startup, as a warning naming the empty table.

Inspecting durable state

--metrics prometheus also serves /stats on :8000, which reports what is on disk right now:

$ curl -s localhost:8000/stats
{"state":{"path":"/var/lib/sqlflow/state.db","size_bytes":2109440,
          "tables":[{"table":"agg_city_count","rows":1440}],
          "offsets":[{"topic":"events","partition":0,"offset":98213}]}}

Reads come from a second connection, so a scrape never blocks a batch and never reports rows a rollback then erased. A pipeline with no state path serves no /stats.

Environment variables

Variable	Purpose
`SQLFLOW_DUCKDB_LIB`	Path to `libduckdb`. Defaults per-OS as described in Installation
`SQLFLOW_LOG_LEVEL`	Log level, default `INFO` (`DEBUG`, `INFO`, `WARN`/`WARNING`, `ERROR`)
`SQLFLOW_SQL_RESULTS_CACHE_DIR`	Staging dir for `InferredDiskBatch`, default `/tmp/sqlflow/resultscache`
`SQLFLOW_STATIC_ROOT`	`STATIC_ROOT` template variable, default `/tmp/sqlflow/static`
`PYICEBERG_HOME`, `PYICEBERG_CATALOG__*`	Iceberg catalog resolution, same as pyiceberg
`SQLFLOW_*`	Anything else is injected into the config template context under its own name

Behaviour notes

An empty batch produces no output rather than an error. A batch is legitimately empty when a fixture is empty or every message in it was rejected; the handlers return no table and the sink is not called.
A missing field and an explicit null are the same value. Both read as SQL NULL, so they aggregate into one group.
StructuredBatch truncates its table at the start of every batch.
DuckDB version. The engine loads whatever libduckdb you install; DUCKDB_VERSION pins what the image and benchmarks use.
Templating is gonja (Jinja2 for Go). Every example config renders under it, asserted by a test.

Benchmarks

Measured with make benchmark-container: Apple M1 Pro (10 cores, 32 GB), Docker 20.10.13, DuckDB v1.5.2, Go 1.25.5, single-partition Kafka (confluentinc/cp-kafka:7.3.2), 300,000 JSON messages aggregated into DuckDB. Every run uses a fresh topic and consumer group, so runs are hermetic.

Handler	`batch_size`	Throughput	Peak memory (container)	Peak working set
`handlers.StructuredBatch`	500	~305k msgs/sec	254 MiB	167 MiB
`handlers.StructuredBatch`	2000	~685k msgs/sec	255 MiB	172 MiB
`handlers.StructuredBatch`	5000	~927k msgs/sec	240 MiB	155 MiB
`handlers.InferredMemBatch`	500	~159k msgs/sec	256 MiB	171 MiB
`handlers.InferredMemBatch`	2000	~229k msgs/sec	264 MiB	181 MiB
`handlers.InferredMemBatch`	5000	~256k msgs/sec	255 MiB	171 MiB

Two memory figures, both sampled from the container's cgroup by the benchmark script: peak memory is everything the container is charged for (page cache and lazily-freed pages included) — the provisioning ceiling; peak working set is anonymous memory, comparable to RSS — what the engine actually holds. Memory is flat across handlers and batch sizes at roughly a quarter GiB, so throughput scales with batch size without buying it with memory.

What durable state costs

Setting pipeline.state.path puts every batch in a DuckDB transaction. That transaction is one fsync per batch, so its cost per message falls as batches grow. Same pipeline, same 300,000 messages, state in memory against state on disk:

`batch_size`	State in memory	Durable state	Cost
500	~95,600 msgs/sec	~51,100 msgs/sec	47%
2000	~184,600 msgs/sec	~130,600 msgs/sec	29%
5000	~201,600 msgs/sec	~171,500 msgs/sec	15%

Use a batch of at least 5000 for a stateful pipeline. Below 2000 the commit dominates and you pay for durability on every message instead of amortising it across a batch.

Memory is unchanged: peak working set stayed within 191-218 MiB across every run above, with and without state. Durability costs throughput, not memory.

state_commit_latency reports the per-batch commit time on the metrics endpoint, so you can see this cost on your own hardware rather than inferring it from the table.

Reproduce both arms:

make benchmark-container NUM_MESSAGES=300000 BATCH_SIZE=5000 \
    CONFIG=dev/config/examples/benchmark.stateful.mem.yml
STATE_PATH=/tmp/bench-state.db make benchmark-container NUM_MESSAGES=300000 BATCH_SIZE=5000 \
    CONFIG=dev/config/examples/benchmark.stateful.mem.yml

Delete the state file between runs. A second run resumes from the first run's offsets and consumes nothing, which reports a meaningless number rather than failing.

To reproduce:

make start-backing-services
make benchmark-container NUM_MESSAGES=300000 BATCH_SIZE=5000
make benchmark-container NUM_MESSAGES=300000 BATCH_SIZE=5000 \
    CONFIG=dev/config/examples/benchmark.inferred.mem.yml

Benchmarks must run inside the docker network

Docker Desktop's host→container port-forwarding caps Kafka fetches at roughly 10-15 MB/s. That starves the pipeline and understates throughput by about 10x — you will measure the NAT, not the engine. make benchmark-container builds a linux sqlflow and runs it on the same docker network as the broker, which is the only way to get a number that reflects the engine.

make benchmark runs the same workload from the host. It is fine for a quick smoke test, but do not quote its numbers.

Building release binaries

make release-binaries        # artifacts land in dist/

sqlflow cannot be cross-compiled the usual way, and it is worth understanding why before you try. The ADBC driver manager is a cgo package:

CGO_ENABLED=0 does not merely produce a degraded binary, it fails to compile: internal/duckdb/open.go:36:20: undefined: drivermgr.Driver.
CGO_ENABLED=1 GOOS=linux go build on a mac hands the C files to the host clang, which cannot target linux, and the build dies in runtime/cgo.

So each target needs a C toolchain for that target, and the matrix is built three different ways:

Target	How it is built	Host requirement
`linux/amd64`	`docker run --platform linux/amd64`	docker (+ binfmt/qemu if the host is arm64)
`linux/arm64`	`docker run --platform linux/arm64`	docker (+ binfmt/qemu if the host is amd64)
`darwin/arm64`	native `go build`	macOS + Xcode command line tools
`darwin/amd64`	`go build` with `-arch x86_64`	macOS + Xcode command line tools

A macOS host with docker produces all four. A linux host produces only the two linux targets — darwin binaries would need a macOS SDK and an osxcross-style toolchain, which this repo deliberately does not ship. Targets that cannot be built on the current host are reported as skipped, not faked. Windows is not a target.

The resulting binaries are dynamically linked and dlopen libduckdb; they are not standalone. See Installation.

Publishing the release image

make sqlflow-image builds for the host architecture only, which is fine for local testing and wrong for publishing. Releases go out through make release-image, which builds linux/amd64 and linux/arm64 and pushes both under one manifest:

git tag -a v1.0.4 -m "..." && git push origin v1.0.4
make test-image        # functional tests against the image
make release-image     # multi-arch build + push, tags latest too
make release-image-verify

Tag first: VERSION comes from git describe, so an untagged main yields v1.0.3-1-gabc1234 rather than a release version. The target refuses to run on a dirty tree or an untagged HEAD for that reason.

Variable	Default	Purpose
`RELEASE_PLATFORMS`	`linux/amd64,linux/arm64`	Architectures to build
`RELEASE_LATEST`	`1`	Also tag `latest`; set `0` when re-publishing an older tag
`RELEASE_OUTPUT`	`--push`	Set `--output=type=cacheonly` for a dry run that publishes nothing
`SQLFLOW_IMAGE`	`turbolytics/sql-flow:$(VERSION)`	Full image reference

make release-image-verify reads the registry back and runs the published image on each architecture: it fails unless the version tag carries both architectures, latest resolves to the identical manifest digest, and sqlflow version on each platform reports the tag. A single-arch publish, a latest left on an older release, or an emulated build that never actually ran all look fine locally and are only visible from outside.

The foreign architecture builds under QEMU emulation, so expect the amd64 go build to take several minutes on an arm64 host — CGO_ENABLED=1 is required for the ADBC driver manager, which rules out cross-compiling. Each platform fetches its own libduckdb, because scripts/install-libduckdb.sh branches on uname -m and sees the target architecture.

Publishing is a manual step from a workstation; CI builds and tests the image on every push but does not push to the registry.

Why this target exists: v1.0.0 was published by hand from a mac with a plain docker build, so it went out arm64-only and did not run on amd64 at all. docker tag + docker push has the same failure mode — it flattens a manifest list down to one architecture. To point an existing tag at another release, copy the manifest instead: docker buildx imagetools create -t turbolytics/sql-flow:latest turbolytics/sql-flow:v1.0.4.

Examples

Additional examples are available in the wiki: Tutorials. Every example config lives in dev/config/examples/.

Consume Bluesky Firehose

Running SQL against the Bluesky firehose is a single configuration file:

The following command starts a bluesky consumer and prints every post to stdout:

./bin/sqlflow run -c dev/config/examples/bluesky/bluesky.raw.stdout.yml

Checkout the configuration files here

Stream Kafka to Iceberg

The following configuration writes to an Iceberg table using a local SQLite catalog:

Initialize the SQLite iceberg catalog and test table. PYICEBERG_HOME points at the directory holding .pyiceberg.yaml, which defines the sqlflow_test catalog the example config expects:

PYICEBERG_HOME=$(pwd)/dev/config/iceberg uv run python cmd/setup-iceberg-local.py setup
created default.city_events
created default.bluesky_post_events
Catalog setup complete.

Start Kafka Locally

docker-compose -f dev/kafka-single.yml up -d

Publish Test Messages to Kafka

uv run python cmd/publish-test-data.py --num-messages=5000 --topic="input-kafka-mem-iceberg"

Run sqlflow, which reads from Kafka and writes to the iceberg table locally

PYICEBERG_HOME=$(pwd)/dev/config/iceberg \
  ./bin/sqlflow run -c dev/config/examples/kafka.mem.iceberg.yml --max-msgs=5000

Verify iceberg data was written by querying it with duckdb

$ duckdb -c "select count(*) from '/tmp/sqlflow/warehouse/default.db/city_events/data/*.parquet';"
┌──────────────┐
│ count_star() │
│    int64     │
├──────────────┤
│         5000 │
└──────────────┘

Upgrading from a pre-v1 release

SQLFlow was written in Python before v1. That implementation has been removed. v1 reads the same configuration files, so moving a pipeline means swapping the image or binary: the config, the templating and the DuckDB SQL carry over.

Seven things changed:

UDFs are not supported. Pre-v1 took Python UDFs through a udfs: block. Define the function in DuckDB instead — a macro, an extension, or an ATTACHed database that provides it. A udfs: block is now a hard error naming the functions, rather than a silent skip that surfaces later as an opaque binder error.
The command line is accepted as it was. run pipeline.yml --max-msgs-to-process=N works unchanged. -c and --max-msgs are the native spellings.
Console output is one JSON object per line rather than a Python list of dicts. Same rows, different rendering.
An empty batch produces no output. It used to raise.
StructuredBatch truncates its table every batch. It used to keep it.
Log format is zap's console format. SQLFLOW_LOG_LEVEL still accepts Python's level names.
DuckDB is whatever libduckdb you install, and the image pins 1.5.x. Pre-v1 pinned 1.3.1, so "same SQL, same result" is not guaranteed across that gap.

Published python-<sha> tags remain on Docker Hub. Reproducing one means checking out a commit from before the removal.

The sqlflow/ directory still exists, and holds test harness rather than an engine: settings.py, kafka.py, fixtures/ and logging.py. The image tests under tests/release and the dev scripts in cmd/ import them.

Development

make sqlflow        # build bin/sqlflow
make test-go        # build, vet, gofmt check, unit tests
make test-image     # build the image and run tests/release against it

tests/release and the coverage matrix are Python. Both run through uv, which builds the environment from uv.lock on first use. There is no pip install step, and no dependency is resolved at install time.

The matrix generator is scripts/coverage_matrix/, one module per concern with a test file each under tests/tooling/. Its own docstring lists them.

Coverage and invariant matrix — what is tested, and separately, what is proven. It is generated from the suites on every push and gates the merge.

What is committed is one status per feature and per invariant, under docs/coverage/status/. Adding a test inside a feature that is already covered changes none of it. The page is rendered from those files and the registries alone, so make coverage-page regenerates it in a second, with no Docker and no test run. Test names and counts are in the coverage report CI publishes on every run.

Contributing — the verification a pull request carries, including the ten-minute memory soak (make soak) required of any change that allocates per message or per request.

make test-go and make test-image are what CI runs on every push. Kafka-backed integration tests are deliberately excluded from test-go; they run from the dev stack. Backing services for local development:

make start-backing-services
make stop-backing-services

Contact Us

Like SQLFlow? Use SQLFlow? Feature Requests? Please let us know! danny@turbolytics.io

Name		Name	Last commit message	Last commit date
.claude/skills		.claude/skills
.github		.github
adrs		adrs
benchmark		benchmark
cmd		cmd
dev		dev
docs		docs
internal		internal
scripts		scripts
sqlflow		sqlflow
tests		tests
.dockerignore		.dockerignore
.gitattributes		.gitattributes
.gitignore		.gitignore
.python-version		.python-version
CHANGELOG.md		CHANGELOG.md
CLAUDE.md		CLAUDE.md
CONTRIBUTING.md		CONTRIBUTING.md
DUCKDB_VERSION		DUCKDB_VERSION
Dockerfile		Dockerfile
LICENSE		LICENSE
Makefile		Makefile
README.md		README.md
SECURITY.md		SECURITY.md
go.mod		go.mod
go.sum		go.sum
pyproject.toml		pyproject.toml
uv.lock		uv.lock

Latest commit

History

Folders and files

Repository files navigation