### Windowed Pipeline Output Example

Source: https://turbolytics.io/docs/sqlflow/tutorials/bluesky-firehose

Example output from the windowed pipeline after the first full minute. Shows counts per kind for the bucket start time.

```json
{"count":2434,"iso_string":"2026-09-15T00:53:00","kind":"commit"}
{"count":18,"iso_string":"2026-09-15T00:53:00","kind":"account"}
{"count":15,"iso_string":"2026-09-15T00:53:00","kind":"identity"}
```

--------------------------------

### Start SQLFlow with Kafka and Postgres Join

Source: https://turbolytics.io/docs/sqlflow/tutorials/postgres-join

Starts SQLFlow with the Kafka-to-Postgres join configuration. The command attaches Postgres and initializes the Kafka source; the log output shows the consumer loop starting and SQLFlow idling for messages.

```bash
docker run -v $(pwd)/dev:/tmp/conf \
  -e SQLFLOW_POSTGRES_USERS_URI=postgresql://postgres:postgres@host.docker.internal:5432/testdb \
  -e SQLFLOW_KAFKA_BROKERS=host.docker.internal:29092 \
  turbolytics/sql-flow:v1.0.6 \
  run /tmp/conf/config/examples/kafka.postgres.join.yml
```

```log
INFO  Executing command step  {"name": "load extensions"}
INFO  Executing command step  {"name": "attach usersdb"}
INFO  initializing kafka source  {"topics": "[input-user-clicks-demo]", "group.id": "test", "auto.offset.reset": "earliest"}
INFO  consumer loop starting
```

--------------------------------

### Start Kafka with make start-backing-services

Source: https://turbolytics.io/docs/sqlflow/tutorials/ducklake-sink

Starts the local Kafka backing services required for the tutorial. Run this before publishing events.

```bash
make start-backing-services
```

--------------------------------

### healthz response for starting status

Source: https://turbolytics.io/docs/sqlflow/operations/running-in-production

Example response from GET /healthz showing a starting status with no commit yet.

```json
{"commit_age_seconds":7.94,"interval_seconds":3600,"reason":"no commit yet","status":"starting"}
```

--------------------------------

### Start Kafka with Docker Compose

Source: https://turbolytics.io/docs/sqlflow/tutorials/motherduck-sink

Starts a single-node Kafka instance using the provided docker-compose file.

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

--------------------------------

### Generate pipeline.yml from config example

Source: https://turbolytics.io/docs/sqlflow/introduction/configuration

Pipes the config reference output to a file to use as a starting point for editing.

```bash
./bin/sqlflow config example > pipeline.yml
```

--------------------------------

### Run SQLFlow Pipeline with Configuration File

Source: https://turbolytics.io/docs/sqlflow/tutorials/bluesky-firehose

Docker command to start the SQLFlow pipeline using the configuration file for the stdout example. Mounts the config directory and runs with a maximum of 200 messages.

```bash
docker run -v $(pwd)/dev/config/examples:/examples \
  turbolytics/sql-flow:v2026.09.14.1 \
  run /examples/bluesky/bluesky.kafka.transformed.yml --max-msgs=200
```

--------------------------------

### Start SQLFlow Pipeline

Source: https://turbolytics.io/docs/sqlflow/tutorials/postgres-sink

Starts the SQLFlow container with environment variables for Postgres and Kafka, followed by the expected initialization logs.

```bash
docker run -v $(pwd)/dev:/tmp/conf \
  -e SQLFLOW_POSTGRES_USERS_URI=postgresql://postgres:postgres@host.docker.internal:5432/testdb \
  -e SQLFLOW_KAFKA_BROKERS=host.docker.internal:29092 \
  turbolytics/sql-flow:v2026.09.14.1 \
  run /tmp/conf/config/examples/kafka.postgres.sink.yml
```

```text
INFO  serving http  {"addr": ":8000", "routes": ["/healthz"]}
INFO  initializing kafka source  {"topics": "[input-kafka-postgres-sink]", "group.id": "test", "auto.offset.reset": "earliest", ...}
INFO  consumer loop starting
```

--------------------------------

### Start Kafka consumer for output-bluesky-firehose topic

Source: https://turbolytics.io/docs/sqlflow/tutorials/bluesky-firehose

Start a Kafka console consumer on the output topic to see the stream of Bluesky messages. Run this before starting the SQLFlow pipeline.

```bash
docker exec -it kafka1 kafka-console-consumer --bootstrap-server=kafka1:9092 --topic=output-bluesky-firehose
```

--------------------------------

### Start SQLFlow in Docker

Source: https://turbolytics.io/docs/sqlflow/tutorials/s3-parquet-sink

Starts SQLFlow in Docker using the given configuration file and environment variables. The documented run measured 194,590 messages per second on an Apple M1 Pro with SQLFlow, Kafka, and MinIO on one Docker network.

```bash
docker run \
    -v $(pwd)/dev/config/examples:/examples \
    -e SQLFLOW_S3_ENDPOINT=host.docker.internal:9000 \
    -e SQLFLOW_KAFKA_BROKERS=host.docker.internal:29092 \
    -e SQLFLOW_BATCH_SIZE=5000 \
    -e SQLFLOW_SINK_BASEPATH=s3://sqlflow/test-sink-1 \
    turbolytics/sql-flow:v1.0.6 run /examples/s3.parquet.sink.yml --max-msgs=100000
```

--------------------------------

### Start Postgres container on Docker network

Source: https://turbolytics.io/docs/sqlflow/tutorials/tumbling-window-postgres

Creates a Docker network named 'sqlflow' and starts a Postgres 16 container named 'sqlflow-postgres' on that network with password 'postgres'. This is the first step to set up the database for the tutorial.

```bash
docker network create sqlflow
docker run -d --name sqlflow-postgres --network sqlflow \
  -e POSTGRES_PASSWORD=postgres postgres:16
```

--------------------------------

### Pipeline startup log output

Source: https://turbolytics.io/docs/sqlflow/tutorials/tumbling-window-postgres

Example log output from running the pipeline. The warning about not deriving reference tables is harmless because the handler is an INSERT, not a SELECT.

```log
INFO  Executing command step  {"name": "declare the post schema"}
INFO  Creating managed table  {"name": "posts_per_minute_by_lang"}
WARN  could not derive reference tables from handler sql  {"error": "handler sql did not parse: Only SELECT statements can be serialized to json!"}
INFO  consumer loop starting
INFO  starting watermark manager  {"table": "posts_per_minute_by_lang", "poll_interval": "10s"}
INFO  throughput  {"messages_consumed": 142, "total_throughput_per_second": 30.8}
```

--------------------------------

### Start Kafka console consumer

Source: https://turbolytics.io/docs/sqlflow/tutorials/intro

Starts a Kafka console consumer to view the SQLFlow output on the topic 'output-simple-agg-mem'.

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

--------------------------------

### Install SQLFlow and clone repository

Source: https://turbolytics.io/docs/sqlflow/tutorials/postgres-sink

Commands to clone the SQLFlow repository, pull the required Docker image (v2026.09.14.1 or later), and install dependencies with `uv sync`. SQLFlow itself needs no Python.

```bash
git clone https://github.com/turbolytics/sql-flow.git
cd sql-flow
docker pull turbolytics/sql-flow:v2026.09.14.1
uv sync
```

--------------------------------

### Start Kafka Consumer for Windowed Output

Source: https://turbolytics.io/docs/sqlflow/tutorials/bluesky-firehose

Docker command to start a Kafka console consumer for the output topic of the windowed pipeline.

```bash
docker exec -it kafka1 kafka-console-consumer --bootstrap-server=kafka1:9092 --topic=output-bluesky-post-kind-count-1-minute
```

--------------------------------

### Start SQLFlow Tumbling Window Pipeline

Source: https://turbolytics.io/docs/sqlflow/tutorials/bluesky-firehose

Docker command to start the SQLFlow tumbling window pipeline with the windowed configuration file. Sets the Kafka brokers environment variable.

```bash
docker run -e SQLFLOW_KAFKA_BROKERS=host.docker.internal:29092 \
  -v $(pwd)/dev/config/examples:/examples \
  turbolytics/sql-flow:v2026.09.14.1 \
  run /examples/bluesky/bluesky.kafka.windowed.yml
```

--------------------------------

### Run SQLFlow pipeline

Source: https://turbolytics.io/docs/sqlflow/tutorials/motherduck-sink

Starts the SQLFlow pipeline with the example config. --max-msgs stops the run after consuming that many messages; leaving it off runs as a daemon tailing the topic until stopped.

```bash
./bin/sqlflow run --metrics=prometheus dev/config/examples/kafka.motherduck.yml --max-msgs=1000000
```

--------------------------------

### Initialize SQLite Iceberg catalog and test table

Source: https://turbolytics.io/docs/sqlflow/tutorials/iceberg-sink

Runs the setup script to create the Iceberg catalog and test tables. Output shows the created tables and completion message. The catalog and warehouse land in /tmp/sqlflow/warehouse.

```bash
uv run python3 cmd/setup-iceberg-local.py setup
```

```text
created default.city_events
created default.bluesky_post_events
Catalog setup complete.
```

--------------------------------

### Install SQLFlow

Source: https://turbolytics.io/docs/sqlflow/tutorials/clickhouse-sink

Clones the SQLFlow repository and builds the binary, which lands at bin/sqlflow. Requires the local development toolchain. To run the container instead, substitute docker run turbolytics/sql-flow:v1.0.6 for ./bin/sqlflow in the run step.

```bash
git clone https://github.com/turbolytics/sql-flow.git
cd sql-flow
make sqlflow
```

--------------------------------

### Common Log Format example line

Source: https://turbolytics.io/docs/sqlflow/tutorials/log-reduction

Example of a single Common Log Format line from the NASA trace files, showing the format of each request.

```text
199.72.81.55 - - [01/Jul/1995:00:00:01 -0400] "GET /history/apollo/ HTTP/1.0" 200 6245
```

--------------------------------

### SQLFlow config reference output

Source: https://turbolytics.io/docs/sqlflow/introduction/configuration

Example output of the config reference command, showing the structure and comments for each option.

```yaml
# Main pipeline configuration.
pipeline:
  # Name of the pipeline.
  name: <string>
  # Description of the pipeline.
  description: <string>
  # Number of messages processed in a batch.
  batch_size: <integer>
  # Time interval to flush batches in seconds.
  flush_interval_seconds: <integer>
  # Longest a shutdown may take after SIGTERM. The final batch, the
  # managers' final poll and the state syncs share it. Absent means 30.
  # When it passes the process exits 15, and the next start replays what
  # was not written.
  drain_deadline_seconds: <integer>
  # Error handling strategy.
  on_error:
    # Defines how errors should be handled.
    policy: RAISE | IGNORE | DLQ
  ...
```

--------------------------------

### SQLFlow Stdout Output Example

Source: https://turbolytics.io/docs/sqlflow/tutorials/bluesky-firehose

Example output from the SQLFlow pipeline when writing to stdout. The console sink sorts keys, so 'kind' precedes 'time'.

```json
{"kind":"commit","time":"2026-09-15T01:04:48"}
{"kind":"commit","time":"2026-09-15T01:04:48"}
{"kind":"commit","time":"2026-09-15T01:04:48"}
{"kind":"commit","time":"2026-09-15T01:04:48"}
```

--------------------------------

### View SQLFlow logs

Source: https://turbolytics.io/docs/sqlflow/tutorials/tigerdata-enrich

Check the container logs to see the pipeline startup and idle state. The log shows the service attached and the consumer loop starting.

```bash
docker logs sqlflow-tiger
```

--------------------------------

### Install Python dependencies

Source: https://turbolytics.io/docs/sqlflow/tutorials/clickhouse-sink

Installs the repository's Python dependencies, which are used to publish sample data. SQLFlow itself does not need Python.

```bash
uv sync
```

--------------------------------

### Run SQLFlow pipeline

Source: https://turbolytics.io/docs/sqlflow/tutorials/clickhouse-sink

Runs SQLFlow with the example config, reading from Kafka and writing to ClickHouse in batches of 50k. The output shows throughput logs; three consecutive runs on an Apple M1 Pro measured 104,605, 107,014 and 105,742 messages per second, all writing 1,000,000 rows.

```bash
./bin/sqlflow run $(pwd)/dev/config/examples/kafka.clickhouse.yml --max-msgs=1000000
```

```text
INFO  consumer loop starting
INFO  throughput  {"messages_consumed": 500000, "total_throughput_per_second": 100023}
INFO  max messages consumed, stopping consumer loop
INFO  throughput  {"messages_consumed": 1000000, "total_throughput_per_second": 104605}
```

--------------------------------

### Install libduckdb on macOS and Linux

Source: https://turbolytics.io/docs/sqlflow/introduction/local-development

Install libduckdb via Homebrew on macOS or via the provided script on Linux. On Linux, set SQLFLOW_DUCKDB_LIB to the library path. Without it, SQLFlow uses default paths. The script always fetches the Linux library, so it won't work on macOS.

```bash
# macOS: installs to the default path, no environment variable needed
brew install duckdb

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

--------------------------------

### Full SQLFlow configuration (kafka.ducklake.yml)

Source: https://turbolytics.io/docs/sqlflow/tutorials/ducklake-sink

Complete example configuration for a SQLFlow pipeline that reads from Kafka, transforms data with SQL, and writes to DuckLake. Includes commands to install/attach DuckLake and create the target table, with a batch size template default of 1.

```yaml
commands:
  - name: install ducklake
    sql: |
      INSTALL ducklake;
      LOAD ducklake;

  - name: attach to ducklake
    sql: |
      ATTACH 'ducklake:my_ducklake.ducklake' AS my_ducklake;
      USE my_ducklake;

  - name: create events table if not exists
    sql: |
      CREATE TABLE IF NOT EXISTS my_ducklake.events (
        ip TEXT,
        event TEXT,
        properties_city TEXT,
        properties_country TEXT,
        timestamp TIMESTAMP,
        type TEXT,
        userId TEXT
      );

pipeline:
  batch_size: {{ SQLFLOW_BATCH_SIZE|default(1) }}
  source:
    type: kafka
    kafka:
      brokers: [{{ SQLFLOW_KAFKA_BROKERS|default('localhost:9092') }}]
      group_id: motherduck-sink-1
      auto_offset_reset: earliest
      topics:
        - {{ SQLFLOW_INPUT_TOPIC|default('input-user-clicks-ducklake') }}

  handler:
    type: 'handlers.InferredMemBatch'
    sql: |
      INSERT INTO my_ducklake.events
      SELECT
        ip,
        event,
        properties ->> 'city' AS properties_city,
        properties ->> 'country' AS properties_country,
        CAST(timestamp AS TIMESTAMP) AS timestamp,
        type,
        userId
      FROM batch;

  sink:
    type: noop
```

--------------------------------

### Start SQLFlow container

Source: https://turbolytics.io/docs/sqlflow/tutorials/tigerdata-enrich

Run SQLFlow in the background with the name sqlflow-tiger. The -d flag detaches, and SQLFLOW_KAFKA_BROKERS is set to host.docker.internal:29092 because the compose file advertises that listener for container access.

```bash
docker run -d --name sqlflow-tiger -v $(pwd)/dev:/tmp/conf \
  -e SQLFLOW_TIGERDATA_DSN \
  -e SQLFLOW_KAFKA_BROKERS=host.docker.internal:29092 \
  turbolytics/sql-flow:v1.2.0 \
  run /tmp/conf/config/examples/published/tigerdata.enrich.yml
```

--------------------------------

### Publish 200k test messages to Kafka

Source: https://turbolytics.io/docs/sqlflow/tutorials/postgres-sink

Run the publish-test-data.py script to send 200,000 user_action messages to the input-kafka-postgres-sink topic with the user_action fixture. This is the first step before starting SQLFlow.

```bash
uv run python3 cmd/publish-test-data.py --num-messages=200000 --topic="input-kafka-postgres-sink" --fixture=user_action
```

--------------------------------

### Run the SQL Flow pipeline with Docker

Source: https://turbolytics.io/docs/sqlflow/tutorials/tumbling-window-postgres

Clones the sql-flow repository and runs the Docker container to execute the Bluesky Postgres windowed example configuration. The command mounts the local dev directory to /tmp/conf and uses the turbolytics/sql-flow:v2026.09.14.1 image.

```bash
git clone https://github.com/turbolytics/sql-flow.git
cd sql-flow
docker run --rm --network sqlflow \
  -v $(pwd)/dev:/tmp/conf \
  turbolytics/sql-flow:v2026.09.14.1 \
  run /tmp/conf/config/examples/bluesky/bluesky.postgres.windowed.yml
```

--------------------------------

### Install SQLFlow: clone repo and pull Docker image

Source: https://turbolytics.io/docs/sqlflow/tutorials/postgres-join

Clones the SQLFlow repository, changes into the directory, and pulls the v1.0.6 Docker image. Requires v1.0.5 or later; the command pulls v1.0.6.

```bash
git clone https://github.com/turbolytics/sql-flow.git
cd sql-flow
docker pull turbolytics/sql-flow:v1.0.6
```

--------------------------------

### Start SQLFlow with batch size 200

Source: https://turbolytics.io/docs/sqlflow/tutorials/postgres-sink

Run SQLFlow in a Docker container with SQLFLOW_BATCH_SIZE=200, pointing to Postgres and Kafka. This is the second step; it consumes the 200k messages and inserts them into the user_action table. The command uses the turbolytics/sql-flow:v2026.09.14.1 image and the kafka.postgres.sink.yml config.

```bash
docker run -v $(pwd)/dev:/tmp/conf \
  -e SQLFLOW_BATCH_SIZE=200 \
  -e SQLFLOW_POSTGRES_USERS_URI=postgresql://postgres:postgres@host.docker.internal:5432/testdb \
  -e SQLFLOW_KAFKA_BROKERS=host.docker.internal:29092 \
  turbolytics/sql-flow:v2026.09.14.1 \
  run /tmp/conf/config/examples/kafka.postgres.sink.yml --max-msgs=200000
```

--------------------------------

### Path normalization example

Source: https://turbolytics.io/docs/sqlflow/tutorials/log-reduction

Shows how raw paths are normalized into route classes: the filename is dropped, two directory segments are kept, digit runs are folded to 'N', and the extension classifies the route. This collapses 21,214 distinct paths into 3,423 classes.

```text
/shuttle/missions/sts-73/mission-sts-73.html   ->  /shuttle/missions/*.html
/images/NASA-logosmall.gif                     ->  /images/*.gif
```

--------------------------------

### Example JSON message for batch table

Source: https://turbolytics.io/docs/sqlflow/introduction/configuration

Example JSON message that produces a batch table with inferred schema. Nested objects become DuckDB structs and arrays become lists.

```json
{"user_id":7,"ts":"2026-09-07T10:00:00","props":{"city":"NYC"},"tags":["a","b"]}
```

--------------------------------

### Run archive pipeline with sql-flow

Source: https://turbolytics.io/docs/sqlflow/tutorials/log-reduction

Runs the archive pipeline from the sql-flow repository. It reads the Kafka topic and writes large batches for Parquet object size. Requires the published image and the example config file.

```bash
docker run \
    -v $(pwd)/dev/config/examples:/examples \
    -e SQLFLOW_KAFKA_BROKERS=host.docker.internal:29092 \
    -e SQLFLOW_S3_ENDPOINT=host.docker.internal:9000 \
    turbolytics/sql-flow:v2026.09.19.1 run /examples/logs.archive.parquet.yml
```

--------------------------------

### Run rollup pipeline with sql-flow

Source: https://turbolytics.io/docs/sqlflow/tutorials/log-reduction

Runs the rollup pipeline from the sql-flow repository. It reads the same Kafka topic and writes to ClickHouse, shaped by its window. Requires the published image and the example config file.

```bash
docker run \
    -v $(pwd)/dev/config/examples:/examples \
    -e SQLFLOW_KAFKA_BROKERS=host.docker.internal:29092 \
    -e SQLFLOW_CLICKHOUSE_DSN=clickhouse://host.docker.internal:8123/test \
    turbolytics/sql-flow:v2026.09.19.1 run /examples/logs.rollup.clickhouse.yml
```

--------------------------------

### Publish Kafka Message

Source: https://turbolytics.io/docs/sqlflow/tutorials/postgres-sink

Starts the Kafka console producer and publishes a JSON message to the input topic.

```bash
docker exec -it kafka1 kafka-console-producer --bootstrap-server=kafka1:9092 --topic=input-kafka-postgres-sink
>
```

```json
>{"timestamp": "2025-01-18T12:34:56Z", "user_id": "u12345", "action": "button_click", "browser": "Chrome"}
```

--------------------------------

### Example fixture: five search events with city property

Source: https://turbolytics.io/docs/sqlflow/testing/unit-testing-sql

A small fixture with five NDJSON records used to test a city-count aggregation. Keep fixtures small enough to reason about manually.

```text
{"event":"search","properties":{"city":"New York"},"user":{"id":"u1"}}
{"event":"search","properties":{"city":"New York"},"user":{"id":"u2"}}
{"event":"search","properties":{"city":"New York"},"user":{"id":"u3"}}
{"event":"search","properties":{"city":"Baltimore"},"user":{"id":"u4"}}
{"event":"search","properties":{"city":"Baltimore"},"user":{"id":"u5"}}

```

--------------------------------

### Aggregate posts per bucket in Postgres

Source: https://turbolytics.io/docs/sqlflow/tutorials/tumbling-window-postgres

Rolls up the per-minute, per-language rows into per-bucket totals and shows the output. The first bucket is short because the pipeline started part way through it.

```sql
SELECT bucket, sum(posts) AS posts, count(*) AS languages
FROM posts_per_minute_by_lang
GROUP BY bucket ORDER BY bucket;
```

```text
         bucket         | posts | languages
------------------------+-------+-----------
 2026-09-15 00:44:00+00 |    54 |         6
 2026-09-15 00:45:00+00 |  2571 |        25
 2026-09-15 00:46:00+00 |  2447 |        26
 2026-09-15 00:47:00+00 |  2402 |        26
 2026-09-15 00:48:00+00 |  2376 |        24
 2026-09-15 00:49:00+00 |  2378 |        27
 2026-09-15 00:50:00+00 |  2624 |        18
 2026-09-15 00:51:00+00 |  2399 |        25
```

--------------------------------

### Query newest bucket from Postgres

Source: https://turbolytics.io/docs/sqlflow/tutorials/tumbling-window-postgres

Queries the newest bucket from the posts_per_minute_by_lang table and shows the resulting output. The bucket closes 60 seconds after its end, so the first full minute appears about three minutes after the pipeline starts.

```bash
docker exec sqlflow-postgres psql -U postgres -c "
SELECT bucket, lang, posts
FROM posts_per_minute_by_lang
WHERE bucket = (SELECT max(bucket) FROM posts_per_minute_by_lang)
ORDER BY posts DESC LIMIT 8;"
```

```text
         bucket         |  lang   | posts
------------------------+---------+-------
 2026-09-15 00:51:00+00 | en      |  1720
 2026-09-15 00:51:00+00 | unknown |   234
 2026-09-15 00:51:00+00 | ja      |   180
 2026-09-15 00:51:00+00 | pt      |   102
 2026-09-15 00:51:00+00 | es      |    43
 2026-09-15 00:51:00+00 | ne      |    32
 2026-09-15 00:51:00+00 | ko      |    30
 2026-09-15 00:51:00+00 | ja-JP   |    10
(8 rows)
```

--------------------------------

### Sample output of Bluesky firehose messages

Source: https://turbolytics.io/docs/sqlflow/tutorials/bluesky-firehose

Example of the raw Bluesky firehose messages that the Kafka consumer displays. This output shows the structure of the JSON payloads.

```json
...
{"commit":{"cid":"bafyreibrgt7sv6opy2jyd3mxfgxhu7xzyaciv77xba7q33ukjzloghe4r4","collection":"app.bsky.feed.post","operation":"create","record":{"$type":"app.bsky.feed.post","createdAt":"2026-09-13T20:40:44.645Z","langs":["en"],"reply":{"parent":{"cid":"bafyreiggjkiiz7m2n5sa5dduq3soeikvrri6yqlhcrfcjrzy6trewwd27q","uri":"at://did:plc:hjdxvrq7rh43z5umtt3mucxw/app.bsky.feed.post/3mvge7lmfrk2o"},"root":{"cid":"bafyreiggjkiiz7m2n5sa5dduq3soeikvrri6yqlhcrfcjrzy6trewwd27q","uri":"at://did:plc:hjdxvrq7rh43z5umtt3mucxw/app.bsky.feed.post/3mvge7lmfrk2o"}},"text":"The urgent care is important if you have more than 3 meximelts"},"rev":"3mvghejkr5c24","rkey":"3mvgheik5m22k"},"did":"did:plc:hvnzm4bp66p4b2lqelfwtlwz","kind":"commit","time_us":1789332046760134}
{"commit":{"cid":"bafyreiadonzloj2mudhnolm7lggqa2hyqy5mwvlq4zmq3em42cncm4y3km","collection":"app.bsky.feed.post","operation":"create","record":{"$type":"app.bsky.feed.post","createdAt":"2026-09-13T20:40:44.498Z","embed":{"$type":"app.bsky.embed.external","external":{"description":"YouTube video by Clash With Bao","thumb":{"$type":"blob","mimeType":"image/jpeg","ref":{"$link":"bafkreibvghszoauw2kbtoz5sqkh5on5dzh2tjxeub5rnxlnuipya3ira6u"},"size":82250},"title":"4th attempt at \"Arraignment\" for a 3rd degree \"FELONY\" By the Bexar County Clowns #cops, #court","uri":"https://youtube.com/shorts/XCMTnI9X8RU?is=kQtqH9Zc9uzr1h3k"}},"langs":["en"],"text":"Legendary Clash with Bao."},"rev":"3mvgheja5up2n","rkey":"3mvgheifl4k2d"},"did":"did:plc:ul4zc6gkcsupdwhwqpucipam","kind":"commit","time_us":1789332046784256}
...
```

--------------------------------

### Start Postgres container

Source: https://turbolytics.io/docs/sqlflow/tutorials/tumbling-window-postgres

Restarts the Postgres container after an outage. A supervisor restarts the process once Postgres is back; the recovery depends on where the pipeline state lives.

```bash
docker start sqlflow-postgres
```

--------------------------------

### Run SQLFlow against Kafka

Source: https://turbolytics.io/docs/sqlflow/introduction/local-development

Starts the SQLFlow engine locally to process events from Kafka. The config can be passed positionally or with -c/--config, and the message cap can be --max-msgs or --max-msgs-to-process; both spellings are accepted for backward compatibility.

```bash
SQLFLOW_KAFKA_BROKERS=localhost:9092 \
./bin/sqlflow run -c dev/config/examples/basic.agg.mem.yml --max-msgs=10000
```

--------------------------------

### SQLFlow pipeline configuration: kafka.postgres.join.yml

Source: https://turbolytics.io/docs/sqlflow/tutorials/postgres-join

This YAML configuration defines a SQLFlow pipeline that reads JSON from a Kafka topic, joins it with a PostgreSQL table, and writes the enriched result to the console. It includes commands to install and load the Postgres extension, attach the PostgreSQL database as read-only, and configure the Kafka source, SQL handler, and console sink. The handler uses a LEFT JOIN and a DuckDB struct literal to nest user fields under a single 'user' key.

```yaml
commands:
  - name: load extensions
    sql: |
      INSTALL postgres;
      LOAD postgres;

  - name: attach usersdb
    sql: |
      ATTACH '{{ SQLFLOW_POSTGRES_USERS_URI|default('postgresql://postgres:postgres@localhost:5432/testdb') }}' AS pgusersdb (TYPE POSTGRES, READ_ONLY);

pipeline:
  name: kafka-postgres-join
  description: "Joins Kafka stream with PostgreSQL data and outputs enriched results"
  batch_size: 1

  source:
    type: kafka
    kafka:
      brokers: [{{ SQLFLOW_KAFKA_BROKERS|default('localhost:9092') }}]
      group_id: {{ SQLFLOW_GROUP_ID|default('test') }}
      auto_offset_reset: earliest
      topics:
        - {{ SQLFLOW_INPUT_TOPIC|default('input-user-clicks-demo') }}

  handler:
    type: 'handlers.InferredMemBatch'
    sql: |
      SELECT
        action,
        CAST(timestamp as STRING) as timestamp,
        {
          "user_id": users.user_id,
          "first_name": first_name,
          "last_name": last_name,
          "age": age,
          "email": email,
          "last_login": CAST(last_login AS STRING),
          "is_active": is_active,
          "country": country,
          "preferred_language": preferred_language,
          "subscription_level": subscription_level,
          "marketing_opt_in": marketing_opt_in
        } as user
      FROM batch
      LEFT JOIN pgusersdb.users ON batch.user_id = users.user_id

  sink:
    type: console
```

--------------------------------

### Run SQLFlow with kafka.ducklake.yml config

Source: https://turbolytics.io/docs/sqlflow/tutorials/ducklake-sink

Starts SQLFlow to read 1M events from Kafka and write to DuckLake. The config fixes the consumer group to 'motherduck-sink-1' with no override; re-running without new events waits silently. Raising SQLFLOW_BATCH_SIZE trades latency for throughput.

```bash
SQLFLOW_BATCH_SIZE=1000 \
./bin/sqlflow run dev/config/examples/kafka.ducklake.yml --max-msgs=1000000
```

--------------------------------

### Invoke handler with dev invoke command

Source: https://turbolytics.io/docs/sqlflow/testing/unit-testing-sql

Runs the handler with a JSONL fixture using `sqlflow dev invoke`. It executes setup commands, builds the handler, feeds it every non-empty line of the fixture, invokes the batch once, and prints the resulting rows. It deliberately does not write to the sink.

```bash
sqlflow dev invoke <config> <fixture>

```

--------------------------------

### Define tigerdata.enrich.yml pipeline

Source: https://turbolytics.io/docs/sqlflow/tutorials/tigerdata-enrich

The complete YAML pipeline definition from dev/config/examples/published/tigerdata.enrich.yml. It installs the PostgreSQL extension, attaches the service, subscribes to Kafka, enriches with a LEFT JOIN, and inserts into a hypertable with ON CONFLICT DO NOTHING. Environment variables in {{ }} are resolved at startup; SQLFLOW_TIGERDATA_DSN is required.

```yaml
commands:
  - name: load the postgres extension
    sql: |
      INSTALL postgres;
      LOAD postgres;

  # One attach serves both directions: the join reads devices from it and
  # the sink inserts readings into it. The DSN is a required input with no
  # default on purpose; a production pipeline should not carry a bunk one.
  - name: attach the service
    sql: |
      ATTACH '{{ SQLFLOW_TIGERDATA_DSN }}' AS tiger (TYPE POSTGRES);

pipeline:
  name: tigerdata-enrich
  description: "Enrich a stream of readings with device metadata from the service, and land them in a hypertable"
  batch_size: {{ SQLFLOW_BATCH_SIZE|default('500') }}

  source:
    type: kafka
    kafka:
      brokers: [{{ SQLFLOW_KAFKA_BROKERS|default('localhost:9092') }}]
      group_id: {{ SQLFLOW_GROUP_ID|default('tigerdata-enrich') }}
      auto_offset_reset: earliest
      topics:
        - "{{ SQLFLOW_TOPIC|default('readings') }}"

  # LEFT JOIN, so a reading from a device the reference table does not know
  # still lands, with site and model NULL, rather than being dropped.
  handler:
    type: 'handlers.InferredMemBatch'
    sql: |
      SELECT
        CAST(batch.ts AS TIMESTAMPTZ) AS time,
        batch.device_id,
        devices.site,
        devices.model,
        batch.value
      FROM batch
      LEFT JOIN tiger.{{ SQLFLOW_TIGERDATA_SCHEMA|default('telemetry') }}.devices AS devices
        ON batch.device_id = devices.device_id

  # Delivery is at-least-once: a crash between this insert and the offset
  # commit replays the batch on restart. The primary key on (time,
  # device_id) plus ON CONFLICT DO NOTHING makes the replay a no-op.
  sink:
    type: sqlcommand
    sqlcommand:
      sql: |
        INSERT INTO tiger.{{ SQLFLOW_TIGERDATA_SCHEMA|default('telemetry') }}.readings
          (time, device_id, site, model, value)
        SELECT time, device_id, site, model, value FROM sqlflow_sink_batch
        ON CONFLICT DO NOTHING
```

--------------------------------

### RAISE error log output

Source: https://turbolytics.io/docs/sqlflow/operations/handling-errors

Example log output when the RAISE policy stops the pipeline on an error. Shows the coded error and its class, which a supervisor can use to distinguish permanent from retryable failures.

```text
ERROR  error writing message  {"error": "[user.data.malformed] invalid json: \"!invalid!\"",
                               "error.code": "user.data.malformed", "error.class": "user"}
```

--------------------------------

### SQLFlow pipeline configuration for Kafka to S3 Parquet sink

Source: https://turbolytics.io/docs/sqlflow/tutorials/s3-parquet-sink

This YAML configuration defines a SQLFlow pipeline that reads from a Kafka topic, transforms data with SQL, and writes to S3-compatible storage in Parquet format. It includes commands to install and configure DuckDB's S3 extensions, and uses environment variable substitutions for flexible configuration.

```yaml
commands:
  - name: Install Extensions
    sql: |
      INSTALL httpfs;
      LOAD httpfs;
  - name: Configure s3
    sql: |
      SET s3_region='{{ SQLFLOW_S3_REGION|default('us-east-1') }}';
      SET s3_url_style='path';
      SET s3_endpoint='{{ SQLFLOW_S3_ENDPOINT|default('localhost:9000') }}';
      SET s3_access_key_id='{{ SQLFLOW_S3_ACCESS_KEY_ID|default('minioadmin') }}';
      SET s3_secret_access_key='{{ SQLFLOW_S3_SECRET_ACCESS_KEY|default('minioadmin') }}';
      SET s3_use_ssl=false;

pipeline:
  batch_size: {{ SQLFLOW_BATCH_SIZE|default(1) }}

  source:
    type: kafka
    kafka:
      brokers: [{{ SQLFLOW_KAFKA_BROKERS|default('localhost:9092') }}]
      group_id: {{ SQLFLOW_GROUP_ID|default('test') }}
      auto_offset_reset: earliest
      topics:
        - {{ SQLFLOW_INPUT_TOPIC|default('input-s3-parquet-sink') }}

  handler:
    type: 'handlers.InferredMemBatch'
    sql: |
      SELECT
        CAST(timestamp as TIMESTAMP) as timestamp,
        user_id,
        action,
        browser
      FROM batch

  sink:
    type: sqlcommand
    sqlcommand:
      substitutions:
        - var: $sqlflow_uuid
          type: uuid4
      sql: |
        COPY sqlflow_sink_batch
          TO '{{ SQLFLOW_SINK_BASEPATH }}/$sqlflow_uuid.parquet'
          (FORMAT 'parquet');
```

--------------------------------

### Create telemetry schema, devices table, and readings hypertable

Source: https://turbolytics.io/docs/sqlflow/tutorials/tigerdata-enrich

Run this SQL in psql or the console's SQL editor to set up the telemetry schema, the devices reference table with seed data, and the readings hypertable. The readings table uses a primary key on (time, device_id) so the SQLFlow sink's ON CONFLICT DO NOTHING can safely deduplicate at-least-once deliveries. The site and model columns are nullable so readings from unknown devices are still accepted.

```sql
CREATE SCHEMA telemetry;
SET search_path TO telemetry;

CREATE TABLE devices (
  device_id TEXT PRIMARY KEY,
  site      TEXT NOT NULL,
  model     TEXT NOT NULL
);

INSERT INTO devices (device_id, site, model) VALUES
  ('d-100', 'oslo',   'tx-2'),
  ('d-200', 'lisbon', 'tx-2'),
  ('d-300', 'lisbon', 'tx-3');

CREATE TABLE readings (
  time      TIMESTAMPTZ      NOT NULL,
  device_id TEXT             NOT NULL,
  site      TEXT,
  model     TEXT,
  value     DOUBLE PRECISION,
  PRIMARY KEY (time, device_id)
) WITH (
  timescaledb.hypertable,
  timescaledb.chunk_interval = '7 days'
);
```

--------------------------------

### Build SQLFlow binary with make sqlflow

Source: https://turbolytics.io/docs/sqlflow/introduction/local-development

Clones the SQLFlow repository and builds the binary. The binary lands at bin/sqlflow. To build a released version, check out its tag first; v1.0.6 is the current one. If you don't have a Go toolchain, build the container instead with make sqlflow-image.

```bash
git clone https://github.com/turbolytics/sql-flow.git
cd sql-flow

make sqlflow
make setup-dev
```

--------------------------------

### Postgres sink encode failure error output

Source: https://turbolytics.io/docs/sqlflow/operations/handling-errors

Example error log when the Postgres sink cannot encode a value for a column. This is a user error (`user.sink.encode_failed`) and exits 10 on the first attempt. The example shows a BIGINT value sent to a smallint column, with SQLSTATE 57014.

```text
ERROR  error flushing sink  {"error": "[user.sink.encode_failed] postgres sink: encode a value for its column (SQLSTATE 57014): ERROR: COPY from stdin failed: unable to encode 1099511627776 into binary format for int2 (OID 21): 1099511627776 is greater than maximum value for int2 (SQLSTATE 57014)", "error.code": "user.sink.encode_failed", "error.class": "user"}
```

--------------------------------

### Publish test data to Kafka

Source: https://turbolytics.io/docs/sqlflow/tutorials/motherduck-sink

Runs the sample-data script to send one million mock user interaction events into Kafka for SQLFlow to ingest.

```bash
uv run python3 cmd/publish-test-data.py --num-messages=1000000 --topic="input-user-clicks-motherduck"
```

--------------------------------

### SQLFlow startup log output

Source: https://turbolytics.io/docs/sqlflow/tutorials/tigerdata-enrich

Expected log output showing the command steps and Kafka consumer initialization. The pipeline is idle, waiting for messages.

```text
INFO  Executing command step  {"name": "load the postgres extension"}
INFO  Executing command step  {"name": "attach the service"}
INFO  initializing kafka source  {"topics": "[readings]", "group.id": "tigerdata-enrich", "auto.offset.reset": "earliest"}
INFO  consumer loop starting
```

--------------------------------

### DLQ error message payload

Source: https://turbolytics.io/docs/sqlflow/operations/handling-errors

Example structured error message published to the DLQ. Includes the failing payload, the phase, and the coded error.

```json
{
  "error": "[user.data.malformed] invalid json: \"!invalid!\"",
  "message": "!invalid!",
  "phase": "handler.write",
  "timestamp": "2026-09-06T18:28:31.015Z"
}
```

--------------------------------

### Print SQLFlow configuration reference

Source: https://turbolytics.io/docs/sqlflow/introduction/configuration

Prints every configuration key with its type and a comment, generated from the schema the validator uses.

```bash
./bin/sqlflow config example
```

--------------------------------

### Transform Stream and Write to Stdout: SQL Query

Source: https://turbolytics.io/docs/sqlflow/tutorials/bluesky-firehose

SQL transformation that extracts the timestamp and post kind from the raw event. Used in the configuration file for the stdout example.

```sql
SELECT
  strftime(to_timestamp(time_us / 1000000), '%Y-%m-%dT%H:%M:%S') as time,
  kind
FROM batch
```

--------------------------------

### Run SQLFlow Bluesky raw consumer pipeline

Source: https://turbolytics.io/docs/sqlflow/tutorials/bluesky-firehose

Run the SQLFlow pipeline in Docker, using the Bluesky raw consumer configuration file. This processes the firehose and writes to Kafka. Requires SQLFLOW_KAFKA_BROKERS and the local config directory mounted.

```bash
docker run -e SQLFLOW_KAFKA_BROKERS=host.docker.internal:29092 \
  -v $(pwd)/dev/config/examples:/examples \
  turbolytics/sql-flow:v2026.09.14.1 \
  run /examples/bluesky/bluesky.kafka.raw.yml --max-msgs=1000
```

--------------------------------

### Create posts_per_minute_by_lang table

Source: https://turbolytics.io/docs/sqlflow/tutorials/tumbling-window-postgres

Creates the 'posts_per_minute_by_lang' table in the Postgres container. The primary key (bucket, lang) is used by the sink for upserts; at startup the sink checks that a unique index or constraint covers exactly these columns and exits with code 10 if none exists.

```bash
docker exec sqlflow-postgres psql -U postgres -c "
CREATE TABLE posts_per_minute_by_lang (
  bucket TIMESTAMPTZ NOT NULL,
  lang   TEXT        NOT NULL,
  posts  INTEGER     NOT NULL,
  PRIMARY KEY (bucket, lang)
);"
```

--------------------------------

### Drain deadline exceeded error log

Source: https://turbolytics.io/docs/sqlflow/operations/running-in-production

Log message when the drain deadline passes, showing the batch could not be written. The batch is never committed and will be replayed on next start.

```text
ERROR  error draining the final batch  {"error": "[system.lifecycle.drain_incomplete] drain deadline 2s reached with 50 messages buffered: [system.sink.unreachable] clickhouse sink: ..."}
```

--------------------------------

### Publish a Message to Kafka

Source: https://turbolytics.io/docs/sqlflow/tutorials/postgres-join

Publishes a clickstream event to the Kafka topic SQLFlow reads from. The producer command starts an interactive console; the message is a JSON object with action, timestamp, and user_id.

```bash
docker exec -it kafka1 kafka-console-producer --bootstrap-server=kafka1:9092 --topic=input-user-clicks-demo
```

```text
>{"action": "button_click", "timestamp": "2025-01-18T12:34:56Z", "user_id": 1}
>
```

--------------------------------

### Batch table schema for inferred handlers

Source: https://turbolytics.io/docs/sqlflow/introduction/configuration

Columns of the batch table produced from the example JSON message. The last three are provenance columns added by SQLFlow, only present under a Kafka source.

```text
user_id          BIGINT
ts               VARCHAR
props            STRUCT(city VARCHAR)
tags             VARCHAR[]
kafka_topic      VARCHAR
kafka_partition  INTEGER
kafka_offset     BIGINT
```

--------------------------------

### Create ClickHouse table

Source: https://turbolytics.io/docs/sqlflow/tutorials/clickhouse-sink

SQL statements executed by the compose file on startup to create the test database and the user_actions table in ClickHouse.

```sql
CREATE DATABASE IF NOT EXISTS test;

CREATE TABLE IF NOT EXISTS test.user_actions (
    timestamp DateTime,
    user_id UInt64,
    action String,
    browser String
) ENGINE = MergeTree()
ORDER BY (timestamp, user_id);
```

--------------------------------

### Handler configuration for InferredMemBatch with SQL aggregation

Source: https://turbolytics.io/docs/sqlflow/testing/unit-testing-sql

Defines a handler of type 'handlers.InferredMemBatch' with a SQL query that aggregates city counts from the `batch` table. This is the `basic.agg.mem.yml` config used by all examples on the page.

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

```

--------------------------------

### Run SQLFlow to stream Kafka to Iceberg

Source: https://turbolytics.io/docs/sqlflow/tutorials/iceberg-sink

Runs SQLFlow in Docker, reading from Kafka and writing to the local Iceberg table. Requires the PYICEBERG_HOME and warehouse mounts; missing either causes the write to fail at startup.

```bash
docker run \
  -e SQLFLOW_KAFKA_BROKERS=host.docker.internal:29092 \
  -e PYICEBERG_HOME=/tmp/iceberg/ \
  -v $(pwd)/dev/config/iceberg/.pyiceberg.yaml:/tmp/iceberg/.pyiceberg.yaml \
  -v /tmp/sqlflow/warehouse:/tmp/sqlflow/warehouse \
  -v $(pwd)/dev/config/examples:/examples \
  turbolytics/sql-flow:v1.0.6 \
  run /examples/kafka.mem.iceberg.yml --max-msgs=5000
```

--------------------------------

### Publish 1M events to Kafka with uv and publish-test-data.py

Source: https://turbolytics.io/docs/sqlflow/tutorials/ducklake-sink

Publishes 1,000,000 test events to the 'input-user-clicks-ducklake' topic. Requires Python dependencies locked with uv; running outside that environment fails with ModuleNotFoundError: No module named 'click'.

```bash
uv sync
uv run python3 cmd/publish-test-data.py --num-messages=1000000 --topic="input-user-clicks-ducklake"
```

--------------------------------

### Run dev invoke under Docker

Source: https://turbolytics.io/docs/sqlflow/testing/unit-testing-sql

Runs the same `dev invoke` command inside a Docker container, mounting the local `dev` directory to `/tmp/conf`. Uses the `turbolytics/sql-flow:v1.0.6` image and points to the example config and fixture.

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

```

--------------------------------

### Pyiceberg configuration file (.pyiceberg.yaml)

Source: https://turbolytics.io/docs/sqlflow/tutorials/iceberg-sink

Defines the Iceberg catalog 'sqlflow_test' with a SQLite URI and warehouse path. This file is read by SQLFlow to resolve the catalog referenced in the pipeline configuration.

```yaml
catalog:
  sqlflow_test:
    uri: sqlite:////tmp/sqlflow/warehouse/catalog.db
    warehouse: file:////tmp/sqlflow/warehouse

```