SQLFlow: DuckDB for Streaming Data.

Quickstart | Tutorials | !Docker Pulls | 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

One click runs SQLFlow on Render: a webhook that accepts metrics, a one-minute aggregation into Postgres, and an HTTP API that reads them back. It asks for one secret and is a paid deploy. See render/.

Quick Start (Getting Started in 5 Minutes)

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

make sqlflow

1. Validate a pipeline against test data, without a broker. dev invoke runs the

./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}

1. Start Kafka locally:

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

1. Publish test messages. The publisher is a Python script. [uv][uv] reads

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

[uv]: https://docs.astral.sh/uv/

1. Start a Kafka consumer to watch the output:

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

1. 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.27.3+, 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 | | serve | Serve a config's named SQL datasets over HTTP | | 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 serve

Serves named SQL over HTTP. A serve config declares datasets, each a name and a fixed SQL statement with typed parameters. serve attaches the data through the config's commands, then answers each request by binding its parameters into that statement. Nothing in a request becomes SQL text.

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

| Flag | Default | Description | |---|---|---| | -c, --config | (required) | Path to the serve config, unless given positionally | | --pprof | false | Serve pprof on :6060 |

A serve config is its own file, with commands and serve and no pipeline. sqlflow validate checks it against the serve schema, and sqlflow config example --serve prints every key. This one serves a Postgres view:

commands:
  - name: pin the session timezone
    sql: SET TimeZone='UTC';
  - name: load postgres
    sql: |
      INSTALL postgres;
      LOAD postgres;
  - name: attach postgres read-only
    sql: ATTACH '{{ SQLFLOW_POSTGRES_URI }}' AS pg (TYPE POSTGRES, READ_ONLY);
serve:
  http:
    addr: "0.0.0.0:8080"
    cors:
      allowed_origins: [https://example.com]
  clients:
    - name: demo-page
      id: "{{ SQLFLOW_SERVE_CLIENT_ID }}"
  limits:
    max_rows: 10000
    timeout_seconds: 10
  pool:
    # Requests answered at once. Each session is a backend session; a
    # concurrent query costs a few MiB. Omit for the default, 4.
    size: 4
  metrics:
    # Serve GET /metrics on this listener, without a client id. Off by default:
    # the listener is public and the labels name every dataset.
    enabled: false
  datasets:
    - name: posts_by_lang
      description: Posts per bucket per language.
      params:
        - {name: since, type: timestamp}
        - {name: lang, type: string}
      grains:
        1h:
          sql: |
            SELECT bucket, lang, posts FROM pg.posts_per_hour_by_lang
            WHERE bucket >= coalesce($since, now() - INTERVAL '7 days')
              AND lang = coalesce($lang, lang)
            ORDER BY bucket, lang

Parameters are string, integer, or timestamp. A timestamp is RFC 3339 with an offset; encode a + as %2B in a URL. An absent parameter binds NULL, so default it in SQL with coalesce. Every statement must use every declared parameter. A dataset has either sql or grains, and a request names a grain with ?grain=.

An integer param may declare min, max, or both. A request outside them is 400 invalid_param, and the message names the bounds. Nothing is clamped: a clamp would answer 200 with less than the caller asked for. /v1/datasets lists the bounds with the param.

params:
        - {name: top, type: integer, min: 1, max: 20}

A dataset with grains can declare a time range instead, and let the server choose the grain:

- name: posts_by_lang
      params:
        - {name: since, type: timestamp}
        - {name: until, type: timestamp}
        - {name: lang, type: string}
      range: {since: since, until: until, default: 24h}
      grains:
        5m:
          max_range: 1d
          sql: |
            SELECT bucket, lang, posts FROM pg.posts_per_5m_by_lang
            WHERE bucket >= $since AND bucket < $until
              AND lang = coalesce($lang, lang)
        1h:
          max_range: 14d
          sql: ...
        1d:
          max_range: 365d
          sql: ...

range names the two timestamp params that bound the range, and default is the width a request gets without since. Each grain's max_range is the widest range it serves. For each request:

1. until defaults to the time the request arrived, and since to until

1. Without grain, the server picks the grain with the narrowest

1. The server binds the resolved since and until, never NULL, so the

1. The response carries the grain it chose and the range it resolved:

A range wider than the named grain's max_range, or wider than every grain's, is 400 range_too_wide, and the message names the grains that fit. Nothing is cut from the left of a chart without saying so. Durations are a whole number and one unit: s, m, h or d.

Caching

Every request runs its query unless the dataset opts in:

serve:
  cache:
    max_mb: 16              # optional. Bounds the cache; enables nothing
  datasets:
    - name: pipeline_status
      cache: {ttl_seconds: 15}
      sql: ...
    - name: posts_by_lang
      cache: {ttl_seconds: 30}
      range: {since: since, until: until, default: 24h}
      grains:
        5m:
          bucket: 5m        # how wide one bucket of this grain is
          max_range: 1d
          sql: ... WHERE bucket >= $since AND bucket < $until
        1d:
          bucket: 1d
          cache: {ttl_seconds: 600}   # optional: this grain's own TTL
          max_range: 365d
          sql: ...

A cached answer is the encoded result, served for at most ttl_seconds after the query that produced it started. Nothing is invalidated, because serve is never told that the backend changed: after a backfill, wait one TTL or restart. max_mb is what every cached dataset shares, 16 by default. The least recently used answer goes first, after every expired one, and an answer over a quarter of the bound is returned and not kept.

A grain may carry a cache block of its own, and its ttl_seconds replaces the dataset's for that grain. Use it on the coarse grains. A year of days changes by one open bucket, and its rounded until only moves at midnight, so the TTL is the only thing that re-runs the widest query there is: at the dataset's 30 seconds, twice a minute, for a change no chart can show. The dataset's block is still the opt-in, and a grain's block without it is refused.

A dataset with a range declares bucket on every grain. since and until are rounded up to it, the statement binds the rounded values, and range echoes them, so every request inside one bucket-wide window is one question with one answer. A chart polled by a hundred tabs runs its query once per window, not once per tab.

The rounding returns the same rows only for SQL like the above: since and until compared with a column whose values sit on bucket boundaries, half-open, >= $since AND < $until. Two ways to break it, and serve cannot detect either, which is why caching is the author's claim and off by default:

• bucket <= $until admits the bucket that starts at the rounded until.
• Filtering a raw timestamp, event_at >= $since, drops the events between

sqlflow rollup serve writes bucket on every grain and SQL of the right shape. bucket must divide one day, so a week is refused: weeks start on a Monday and the epoch was a Thursday.

Concurrent requests for one question run one query. That query finishes even if every caller gives up, because the engine would have finished it anyway, and its answer is the next caller's hit.

A cached dataset's response says what happened:

"cache":"hit","age_ms":12400,"queued_ms":0,"elapsed_ms":0

cache is miss when this request ran the query, hit when the answer came from memory, and shared when it waited on a query another request had started. age_ms is how long ago that query started. queued_ms and elapsed_ms are this request's own, so a hit reports zero and not the original query's time. The request log carries cache too, and /v1/datasets lists each dataset's ttl_seconds and each grain's bucket.

Responses stay Cache-Control: no-store: the cache is inside serve, and nothing between it and the caller holds an answer. Each instance holds its own, so two behind a load balancer may answer up to one TTL apart.

With serve.metrics.enabled, a server with a cached dataset adds four metrics: sqlflow_serve_cache_requests_total by dataset and outcome, sqlflow_serve_cache_evictions_total by reason (size or expired), and the gauges sqlflow_serve_cache_bytes and sqlflow_serve_cache_entries. Expired answers are reclaimed when the next one is stored, so after a quiet spell the gauges still count them. sqlflow_serve_query_duration_seconds counts only requests that ran a query; a hit shows in sqlflow_serve_request_duration_seconds.

dev/config/serve/local.cached.yml caches one dataset and says why it does not cache the other.

Routes, all GET:

| Route | client_id | Returns | |---|---|---| | /healthz | none | 200 {"status":"ok"} when a session answers SELECT 1; 503 {"status":"busy"} when none is free, {"status":"unavailable"} when the query fails. HEAD is answered too, for monitors | | /metrics | none | Prometheus text, only when serve.metrics.enabled is set | | /v1/datasets | required | Every dataset: its params, and its SQL as written | | /v1/datasets/{name} | required | Rows |

$ curl 'localhost:8080/v1/datasets/posts_by_lang?client_id=<id>&grain=1h&lang=en'
{"dataset":"posts_by_lang","grain":"1h",
 "columns":[{"name":"bucket","type":"TIMESTAMP WITH TIME ZONE"},...],
 "rows":[{"bucket":"2026-09-10T00:00:00Z","lang":"en","posts":102340}],
 "row_count":1,"truncated":false,"queued_ms":0,"elapsed_ms":41}

A zoned timestamp is UTC. A decimal is a string of its exact digits. NaN and infinities are strings. truncated: true means max_rows cut the result.

elapsed_ms is the query. queued_ms is how long the request waited for a session, which is what rises when the pool is too small for the load.

Every route is GET, and /healthz also answers HEAD. A HEAD of a dataset would run its query, borrow a session and discard the rows, so it is refused with 405.

A client id identifies a caller. It does not authenticate one: a browser page ships it in plain sight, as an analytics snippet ships its site key. It names the caller in the request log, and deleting it cuts the caller off. Do not serve through it anything that the public may not read.

The id travels as ?client_id= and not in an Authorization header, for two reasons. A header named for authorization reads as a leaked credential to everyone who opens the page's source. A GET with no custom header is also a CORS simple request, so a browser sends it without a preflight. client_id is no dataset's param: it never reaches the SQL or the cache key, and a dataset cannot declare a param of that name.

Deprecated: this release still answers Authorization: Bearer when a request has no client_id, and still reads serve.auth.tokens, each token as a client whose id is the token. Both log a warning. The next release refuses both.

Every refusal is {"error": {"code", "message"}}:

| Status | Code | |---|---| | 400 | unknown_param, invalid_param, missing_grain, unknown_grain, range_too_wide | | 401 | unauthorized | | 404 | unknown_dataset, not_found | | 405 | method_not_allowed | | 500 | query_failed. DuckDB's error goes to the server log, with passwords redacted, never to the caller | | 504 | query_timeout |

At start, serve prepares every statement, so a missing table or a syntax error stops the process with user.sql.invalid rather than failing the first request. rate_limit is reserved: the config accepts the key, and a non-zero value stops the process with user.config.serve_reserved.

What to know before you deploy it:

• Attach READ_ONLY. serve runs the dataset SQL as written. A write in

• Aggregate in the backend. DuckDB pushes filters and projections into an

• Bound Postgres connections. An attachment opens up to

• A timeout stops reading, not always the query. At the deadline the

• A pool serves requests. serve.pool.size sessions answer at once, four

sqlflow rollup

Generates rollup tables in Postgres, the triggers that keep them current, and the serve datasets that read them, from one declaration. The pipeline keeps writing its finest grain; the database keeps every coarser grain as each write commits, and serve reads pre-aggregated rows.

sqlflow rollup install -c rollups.yml
sqlflow rollup run     -c rollups.yml [--metrics prometheus]
sqlflow rollup verify  -c rollups.yml [--since T]
sqlflow rollup test    -c rollups.yml --dsn DSN [--tests FILE] [--seed N] [--keep]
sqlflow rollup ddl     -c rollups.yml [--backend postgres]
sqlflow rollup serve   -c rollups.yml [--dataset NAME]
sqlflow rollup check   -c rollups.yml --serve FILE [--migration FILE]

| Command | Does | |---|---| | install | Creates the tables, functions and triggers the file declares, in one transaction, and records what it applied in sqlflow_rollup_state. Refuses a change that would corrupt stored rows. Fills nothing. | | run | Installs, then leads the file's rollups and fills every table install marks, one chunk at a time, beside a live pipeline. Several may run: one leads, the rest stand by. --metrics prometheus serves /metrics and /healthz on :8000. | | verify | Recomputes every table from the table it is built from and exits system.rollup.drift when a bucket differs, printing the first rows that do. It only reads. --since takes an RFC 3339 time or a duration such as 36h. | | test | Installs the file in a schema it creates on --dsn, writes a seeded workload and any fixture cases through the triggers, and exits user.config.rollup_test_failed when a table differs from its source or from a case's expected rows. Drops the schema unless --keep. | | ddl | Prints a migration: the same tables, functions and triggers, and a backfill in one transaction. For a team that applies SQL itself. Connects to nothing. | | serve | Prints the serve datasets, to paste into a serve file's datasets. | | check | Exits 10 when the serve file, or the migration when given, differs from what the declaration generates, or when a dataset could answer more rows than its max_rows. |

install, run and verify connect to store.postgres.dsn:

store:
  type: postgres
  postgres:
    dsn: "{{ SQLFLOW_POSTGRES_URI }}"

Run install in a deploy's entrypoints, before sqlflow serve prepares its statements, and run as its own process. install locks each source for its DDL only, a few milliseconds. run fills history newest day first. A chunk takes the triggers' own bucket locks without ever waiting on them: when a write holds one, the chunk steps back and tries again, so the pipeline keeps writing.

run also checks each table's newest two buckets every minute against the table it is built from. It counts drift in rollup_drift_buckets and logs the first rows that differ. Drift leaves /healthz alone: it belongs to the data, and a restart cannot fix it, so alert on the metric's rate. A disabled trigger, a hand edit or a write that bypassed the triggers causes drift. Rewrite the bucket's source rows to re-merge it.

With a turbostats block that sets report_to, run reports itself to TurboStats as sqlflow run and sqlflow serve do. The leader sends each table's newest bucket under one store id, and per rollup its verify and drift totals, backfill progress, least complete closed bucket and trigger cost. A file that reports declares at most 10 tables; split a larger deployment across files, one run each.

turbostats:
  id: "{{ SQLFLOW_TURBOSTATS_ID }}"
  report_to: "{{ SQLFLOW_TURBOSTATS_REPORT_TO }}"
  key: "{{ SQLFLOW_TURBOSTATS_KEY }}"

test checks a declaration before it ships. Point --dsn at a scratch Postgres where the team's migrations have run, never production: the command connects only there, never to store.postgres.dsn. It clones each source into a schema named sqlflow_test_ and 16 hex characters, installs the rollups there, and writes 20 batches per rollup across the 2026-03-08 daylight saving change, each in a session zone of UTC, America/New_York or Asia/Kolkata. Every table must equal a from-scratch GROUP BY of its source, and every bucket must lie on its grain's UTC boundary. --tests adds fixture cases for what those two rules cannot see, such as last and count_buckets; dev/config/rollups/bluesky.test.yml shows the format. The output starts with the seed, and --seed repeats a run. The role on --dsn needs CREATE on the database. Ctrl-C or SIGTERM ends a run and still drops its schema; a run killed with SIGKILL leaves it, so drop a leftover sqlflow_test_ schema with DROP SCHEMA ... CASCADE.

ddl and check suit a team that applies SQL through its own migration runner. install and run do it for you. dev/config/rollups/bluesky.yml is a complete declaration, and sqlflow validate checks a rollups file against its schema and rules.

Each grain is re-merged from the grain below it. When a statement writes the finer table, a statement-level trigger locks the coarse buckets it touched and recomputes them, in the writer's transaction. A minute written twice replaces its count at every grain instead of adding to it.

A rollup reads a table no rollup makes. To roll up a rollup's table further, add a grain or a dimension set to that rollup.

Writers to one rollup take turns: each trigger takes the rollup's lock first, so two writers never deadlock. Write the source once per transaction, as the Postgres sink does. A write left open in a transaction holds up every other write to its rollup until it ends. Each flush re-merges its buckets for every dimension value, so its cost grows with the values in a bucket, not the rows it wrote. Several processes writing one source pay that cost once each, and they take turns doing it.

Measures are sum, min, max, last and count_buckets, which counts the source buckets present, such as minutes observed. avg, gauge and histogram are reserved.

A sum is stored as a bigint. Add numeric: double to store a double precision, which keeps a fraction: a gauge of 0.73 summed as an integer is 1, and 0.25 is 0. For an average, keep a sum and a count and divide when you read. That is exact at every grain, and an average of averages is not.

last keeps the value of a bucket's latest finer bucket, which is what a gauge reads as: queue depth now. It takes the source column's type. A last needs a dimension set that keeps every source dimension, and a generated dataset cannot serve it, because the last value of the series folded into other belongs to none of them. Write that dataset in serve.yml. dev/config/rollups/metrics.yml declares both.

What to know before you deploy it:

• Postgres 15 or later, and a writer in READ COMMITTED, the default. The

• ddl's migration blocks the source's writers until it commits, and its

• A trigger error fails the writer's transaction. A pipeline writing the

• Deletes do not propagate. Rollups outlive a retention job on the source.
• Changes are additive. Adding a grain or a dimension set is a new

• A served dataset has at most one dimension, and folds it to its top

• A served dataset can be cached. cache_ttl_seconds: 30 on a serve

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. --serve prints the serve config's skeleton instead.

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.26.8

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 addr, by default 0.0.0.0:8001, and optionally validates an HMAC-SHA256 signature:

source:
  type: webhook
  webhook:
    addr: "0.0.0.0:8001" # optional, default 0.0.0.0:8001
    signature_type: hmac
    hmac:
      header: 'X-Hub-Signature-256'
      sig_key: 'sha256'
      secret: "{{ SQLFLOW_GITHUB_WEBHOOK_SECRET }}"
    max_body_bytes: 26214400 # optional, default 25 MiB
    max_connections: 64 # optional, default 64

A platform that assigns the port passes it through the template: addr: "0.0.0.0:{{ PORT }}". SQLFlow refuses at startup an addr that is not a host:port.

To send a signed event, compute an HMAC-SHA256 of the exact bytes of the body with the shared secret, hex-encode it, and put it in the configured header with the sha256= prefix:

export SQLFLOW_GITHUB_WEBHOOK_SECRET=shhh
body='{"action":"opened","number":1}'
sig="sha256=$(printf '%s' "$body" \
  | openssl dgst -sha256 -hmac "$SQLFLOW_GITHUB_WEBHOOK_SECRET" \
  | awk '{print $NF}')"
curl -s -X POST http://localhost:8001/events \
  -H 'Content-Type: application/json' \
  -H "X-Hub-Signature-256: $sig" \
  --data-binary "$body"
{"status":"received"}

The signature covers the raw bytes, so the body sent must be the body signed: --data-binary and printf '%s' keep it byte for byte, where -d and echo can alter whitespace or add a newline. GitHub signs the same way, so a repository webhook with the same secret is accepted as is.

ack decides what the 200 means. The default, on_receive, answers once the body is queued for the pipeline; a crash before the batch flushes loses an event the sender was told it had delivered. ack: after_flush answers once the batch holding the body has flushed to the sink and its position is committed, with {"status":"flushed"}; a flush that fails answers 503 {"detail":"Not flushed"}, and so does a pipeline that stops first, so the sender retries. The cost is the wait: a request waits for its batch to fill or for flush_interval_seconds, whichever comes first, so a lone sender waits the whole interval (measured p50 1.00s, p99 1.04s at 1s). A sender that hangs up keeps its place in the queue; nobody else is given its answer. A flush the sink cannot finish fails within a bound, so a sender always gets an answer: with a Kafka sink whose broker is unreachable, waiting senders get 503 after about 30s (the 20s record timeout, a 5s margin, and up to 5s to abort what is buffered). A failed flush ends the run and the process exits 12 (system.sink.unreachable), so run it under a supervisor that restarts it; until the broker is back it fails at start. after_flush is refused on a pipeline that windows, because a window holds a row for minutes; put the webhook in front of a Kafka topic and window in a second pipeline, as metering/ingest.yml does.

Responds 200 on accept, 400 for a missing signature, 403 for an invalid one, 400 {"detail":"Request body is not valid JSON"} for a body that is not one JSON value, 413 for a body over max_body_bytes, and 408 for a body that stalls. The body bound applies before the body is read and before the signature is checked, so an unsigned oversized request never holds memory past it. The default is 25 MiB, GitHub's payload ceiling.

A body that is not JSON is refused to its sender before it reaches the pipeline: every handler reads JSON, and one unreadable body in a batch would fail the batch for every other sender in it. An empty body and newline- delimited JSON (several values, one per line) are refused the same way; send several events as one array or object.

GET /healthz on the same address answers {"status":"ok"} while the source admits deliveries, and 503 once it is closing. It needs no signature and reads no body, so a platform's health check or an uptime monitor can call it, and HEAD works too. A queue that is full is backpressure, not failure, and still answers 200. Health checks are not counted in webhook_requests_total, which counts deliveries. The pipeline's own /healthz, on the metrics listener, is unchanged: this one exists because a platform that routes a single port to a service, as Render does, checks health on that port.

The listener holds at most max_connections open connections; past that, a new connection waits in the kernel backlog until one closes. Three fixed timeouts release a slot: 10s for a connection to send its request headers, 60s to send the body, and 60s of idle keep-alive. A delivery that is in and waiting on the pipeline is under no timeout.

sqlflow config validate checks the webhook block, including max_body_bytes and max_connections, against the same JSON Schema sqlflow run loads.

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

• Nested struct fields are unioned across the whole batch: a key inside an

• Types are promoted across the batch (int widens to double), JSON

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
    # key: customer  — optional; each record is keyed by that column's text,
    #   so rows with one value land on one partition. A null fails the batch.
    # 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, Iceberg and Postgres 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

• Window close predicates are evaluated at most one flush_interval_seconds

• DuckDB locks the file exclusively. One state file belongs to one running

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

... (README truncated for length)