diff --git a/CHANGELOG.md b/CHANGELOG.md index d91a7a72..f2ae0ba6 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -20,6 +20,11 @@ its batch table. A batch that re-initialised during either failed, and the process exited. The handler skips a refused checkpoint, and the next batch reclaims what it left. +- The reference-table check at startup parsed memory DuckDB had already + freed. It read the handler SQL's AST as a string that pointed into the + query result, and released the result before parsing it. When DuckDB + reused that memory, the check warned `parse serialized sql: invalid + character` or counted another query's tables. It copies the AST first. ### Added @@ -29,6 +34,33 @@ window's sink write or close, however long, never fails the consume loop. The watermark manager's conformance subject proves it with the structured handler. +- A `postgres` sink. `type: postgres` with `dsn`, `table`, `mode: upsert | + append` and, for upsert, `key`. Each batch is a `COPY` into a session + staging table and a server-side `INSERT ... ON CONFLICT`, in its own + transaction, so a flush costs the batch rather than the table. Two rows + with one key in a batch: the last one wins. A column the batch omits takes + its default on insert and keeps its value on update. The probe checks the + table exists and that a unique index or constraint covers exactly the key; + a partial index does not count. Failures classify from SQLSTATE: a refused + or lost connection exits 12 and retries, and a refused value, a missing + column or a constraint violation exits 10. It writes over pgx and touches + no DuckDB connection. +- `sqlflow validate` refuses `late_rows: reemit` with a postgres sink in + upsert mode, warns on it in append mode, and warns on any `sqlcommand` sink + whose SQL carries `ON CONFLICT` while a command attaches a Postgres. The + DuckDB postgres extension runs that upsert by copying every row's key from + the target table into DuckDB on every flush. +- The invariant `sink.flush.idempotent_on_key`: delivering the same batch + twice leaves a keyed sink's destination holding it once. The postgres sink + proves it; every other sink is exempt with a proof that it names no key. +- The image sets `MALLOC_ARENA_MAX=2`. In a loop of 3,600 upserts through the + DuckDB postgres extension it cut native memory growth from 55 KB to 20 KB a + flush. + +### Changed + +- `bluesky.postgres.windowed.yml` and `kafka.postgres.sink.yml` write through + the `postgres` sink, and neither attaches Postgres through DuckDB. ## v2026.09.14 diff --git a/Dockerfile b/Dockerfile index a489b4ae..ec67c9f8 100644 --- a/Dockerfile +++ b/Dockerfile @@ -55,6 +55,13 @@ COPY --from=builder /out/duckdb/libduckdb.so /usr/local/lib/libduckdb.so ENV SQLFLOW_DUCKDB_LIB=/usr/local/lib/libduckdb.so +# glibc opens a malloc arena per thread that allocates, up to eight per core, +# and each arena keeps its own free pages. A cgo process on a small container +# pays for that: the Bluesky demo's Postgres upsert grew native memory 55 KB +# a flush with the default and 20 KB with two arenas, over 3,600 flushes on +# DuckDB v1.5.2 (#290). Two is the usual setting for a Go and cgo container. +ENV MALLOC_ARENA_MAX=2 + WORKDIR /app ENTRYPOINT ["/usr/local/bin/sqlflow"] diff --git a/README.md b/README.md index dfb4a748..b88fe2bf 100644 --- a/README.md +++ b/README.md @@ -661,7 +661,7 @@ sink: ### Sink retries -ClickHouse and Iceberg flushes retry when the destination is not answering. +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. diff --git a/dev/bench/bluesky/demo-10x-noop-sink.yml b/dev/bench/bluesky/demo-10x-noop-sink.yml new file mode 100644 index 00000000..52e529a1 --- /dev/null +++ b/dev/bench/bluesky/demo-10x-noop-sink.yml @@ -0,0 +1,98 @@ +# demo-10x.yml with the window's sink set to noop. +# +# Paced ten times faster than live: replay the capture with +# `go run ./dev/bench/replay -speed 10`. The stream clock then advances ten +# seconds per second, so every timer that paces work against wall clock +# shrinks by the same factor, and the ratios between messages, batches, +# polls and flushes stay what they are live: +# +# poll_interval_seconds 10 -> 1 +# flush_interval_seconds 30 (the default) -> 3 +# +# grace_seconds is stream time and stays. idle_close_seconds only fires when +# the stream stops, and a replay does not stop. Run it with +# dev/bench/replay-soak.sh; bluesky/postgres.sql creates the target table. +commands: + - name: declare the post schema + sql: | + CREATE TABLE IF NOT EXISTS posts ( + time_us BIGINT, + commit STRUCT( + operation TEXT, + record STRUCT(langs TEXT[]) + ) + ); + +tables: + sql: + - name: posts_per_minute_by_lang + # No UNIQUE INDEX, on purpose. DuckDB never frees rows deleted from an + # indexed table, and the engine deletes every closed window, so an index + # grows memory without bound. Each batch appends its own counts and + # emit_sql sums them per minute and language. See #268. + # + # bucket is TIMESTAMPTZ, an instant, so the write into a Postgres + # TIMESTAMPTZ column needs no interpretation and the watermark compares + # instants. + sql: | + CREATE TABLE IF NOT EXISTS posts_per_minute_by_lang ( + bucket TIMESTAMPTZ, + lang TEXT, + posts INTEGER + ); + + # The engine closes the window against a watermark in event time: the + # newest bucket the data has reached, less the grace. The grace outlives + # the batch wait on purpose. A batch flushes when it fills or after + # flush_interval_seconds, so an event from the last seconds of a window + # can reach DuckDB up to that long after the window ended. A grace + # shorter than the batch wait publishes the window before those events + # land. After idle_close_seconds with nothing arriving, every open + # bucket closes. + # + # late_rows: drop, because the sink replaces. A row for a bucket that + # already closed is discarded and counted in window_late_rows_total. + # reemit would run emit_sql over the late rows alone, because the + # bucket's other rows were deleted when it closed, and the upsert would + # replace the bucket's count with theirs. validate refuses that pairing. + window: + time_column: bucket + size_seconds: 60 + grace_seconds: 60 + idle_close_seconds: 60 + late_rows: drop + poll_interval_seconds: 1 + emit_sql: | + SELECT bucket, lang, sum(posts)::INTEGER AS posts + FROM closed + GROUP BY ALL + sink: + # Nothing leaves the process: the control that separates the + # engine's own memory from the write's. + type: noop + +pipeline: + name: bluesky-posts-per-minute-by-lang + batch_size: 500 + flush_interval_seconds: 3 + + source: + type: websocket + websocket: + uri: "{{ SQLFLOW_JETSTREAM_URI|default('wss://jetstream2.us-east.bsky.network/subscribe?wantedCollections=app.bsky.feed.post') }}" + + handler: + type: handlers.StructuredBatch + table: posts + sql: | + INSERT INTO posts_per_minute_by_lang + SELECT + time_bucket(INTERVAL '1 minute', to_timestamp(time_us / 1000000)) AS bucket, + coalesce(commit.record.langs[1], 'unknown') AS lang, + count(*) AS posts + FROM posts + WHERE commit.operation = 'create' + GROUP BY bucket, lang + + sink: + type: noop diff --git a/dev/bench/bluesky/demo-10x-sqlcommand.yml b/dev/bench/bluesky/demo-10x-sqlcommand.yml new file mode 100644 index 00000000..b117dd12 --- /dev/null +++ b/dev/bench/bluesky/demo-10x-sqlcommand.yml @@ -0,0 +1,122 @@ +# demo-10x.yml with the window's sink written through the DuckDB postgres +# extension instead of the postgres sink. validate warns on it, by design. +# +# Paced ten times faster than live: replay the capture with +# `go run ./dev/bench/replay -speed 10`. The stream clock then advances ten +# seconds per second, so every timer that paces work against wall clock +# shrinks by the same factor, and the ratios between messages, batches, +# polls and flushes stay what they are live: +# +# poll_interval_seconds 10 -> 1 +# flush_interval_seconds 30 (the default) -> 3 +# +# grace_seconds is stream time and stays. idle_close_seconds only fires when +# the stream stops, and a replay does not stop. Run it with +# dev/bench/replay-soak.sh; bluesky/postgres.sql creates the target table. +commands: + - name: load postgres extension + sql: | + INSTALL postgres; + LOAD postgres; + + - name: attach postgres + sql: | + ATTACH '{{ SQLFLOW_POSTGRES_URI }}' AS pg (TYPE POSTGRES); + + - name: declare the post schema + sql: | + CREATE TABLE IF NOT EXISTS posts ( + time_us BIGINT, + commit STRUCT( + operation TEXT, + record STRUCT(langs TEXT[]) + ) + ); + +tables: + sql: + - name: posts_per_minute_by_lang + # No UNIQUE INDEX, on purpose. DuckDB never frees rows deleted from an + # indexed table, and the engine deletes every closed window, so an index + # grows memory without bound. Each batch appends its own counts and + # emit_sql sums them per minute and language. See #268. + # + # bucket is TIMESTAMPTZ, an instant, so the write into a Postgres + # TIMESTAMPTZ column needs no interpretation and the watermark compares + # instants. + sql: | + CREATE TABLE IF NOT EXISTS posts_per_minute_by_lang ( + bucket TIMESTAMPTZ, + lang TEXT, + posts INTEGER + ); + + # The engine closes the window against a watermark in event time: the + # newest bucket the data has reached, less the grace. The grace outlives + # the batch wait on purpose. A batch flushes when it fills or after + # flush_interval_seconds, so an event from the last seconds of a window + # can reach DuckDB up to that long after the window ended. A grace + # shorter than the batch wait publishes the window before those events + # land. After idle_close_seconds with nothing arriving, every open + # bucket closes. + # + # late_rows: drop, because the sink replaces. A row for a bucket that + # already closed is discarded and counted in window_late_rows_total. + # reemit would run emit_sql over the late rows alone, because the + # bucket's other rows were deleted when it closed, and the upsert would + # replace the bucket's count with theirs. validate refuses that pairing. + window: + time_column: bucket + size_seconds: 60 + grace_seconds: 60 + idle_close_seconds: 60 + late_rows: drop + poll_interval_seconds: 1 + emit_sql: | + SELECT bucket, lang, sum(posts)::INTEGER AS posts + FROM closed + GROUP BY ALL + sink: + # The write this branch replaces: the upsert the Render demo ran, + # through the DuckDB postgres extension, which copies every row's + # key from the target table on each flush. Kept to measure against + # demo-10x.yml. + # + # updated_at is named and set here because the extension sends a + # column the upsert omits as NULL, and bluesky/postgres.sql + # declares it NOT NULL DEFAULT now(). Omitted, the first flush + # fails with 23502. demo-10x.yml omits it and Postgres applies the + # default. + type: sqlcommand + sqlcommand: + sql: | + INSERT INTO pg.posts_per_minute_by_lang (bucket, lang, posts, updated_at) + SELECT bucket, lang, posts, now() FROM sqlflow_sink_batch + ON CONFLICT (bucket, lang) DO UPDATE + SET posts = EXCLUDED.posts, updated_at = EXCLUDED.updated_at + +pipeline: + name: bluesky-posts-per-minute-by-lang + batch_size: 500 + flush_interval_seconds: 3 + + source: + type: websocket + websocket: + uri: "{{ SQLFLOW_JETSTREAM_URI|default('wss://jetstream2.us-east.bsky.network/subscribe?wantedCollections=app.bsky.feed.post') }}" + + handler: + type: handlers.StructuredBatch + table: posts + sql: | + INSERT INTO posts_per_minute_by_lang + SELECT + time_bucket(INTERVAL '1 minute', to_timestamp(time_us / 1000000)) AS bucket, + coalesce(commit.record.langs[1], 'unknown') AS lang, + count(*) AS posts + FROM posts + WHERE commit.operation = 'create' + GROUP BY bucket, lang + + sink: + type: noop diff --git a/dev/bench/bluesky/demo-10x.yml b/dev/bench/bluesky/demo-10x.yml new file mode 100644 index 00000000..bd5ee91f --- /dev/null +++ b/dev/bench/bluesky/demo-10x.yml @@ -0,0 +1,109 @@ +# The Bluesky demo on the engine's window and the keyed postgres sink: the +# pipeline dev/config/examples/bluesky/bluesky.postgres.windowed.yml teaches. +# +# Paced ten times faster than live: replay the capture with +# `go run ./dev/bench/replay -speed 10`. The stream clock then advances ten +# seconds per second, so every timer that paces work against wall clock +# shrinks by the same factor, and the ratios between messages, batches, +# polls and flushes stay what they are live: +# +# poll_interval_seconds 10 -> 1 +# flush_interval_seconds 30 (the default) -> 3 +# +# grace_seconds is stream time and stays. idle_close_seconds only fires when +# the stream stops, and a replay does not stop. Run it with +# dev/bench/replay-soak.sh; bluesky/postgres.sql creates the target table. +commands: + - name: declare the post schema + sql: | + CREATE TABLE IF NOT EXISTS posts ( + time_us BIGINT, + commit STRUCT( + operation TEXT, + record STRUCT(langs TEXT[]) + ) + ); + +tables: + sql: + - name: posts_per_minute_by_lang + # No UNIQUE INDEX, on purpose. DuckDB never frees rows deleted from an + # indexed table, and the engine deletes every closed window, so an index + # grows memory without bound. Each batch appends its own counts and + # emit_sql sums them per minute and language. See #268. + # + # bucket is TIMESTAMPTZ, an instant, so the write into a Postgres + # TIMESTAMPTZ column needs no interpretation and the watermark compares + # instants. + sql: | + CREATE TABLE IF NOT EXISTS posts_per_minute_by_lang ( + bucket TIMESTAMPTZ, + lang TEXT, + posts INTEGER + ); + + # The engine closes the window against a watermark in event time: the + # newest bucket the data has reached, less the grace. The grace outlives + # the batch wait on purpose. A batch flushes when it fills or after + # flush_interval_seconds, so an event from the last seconds of a window + # can reach DuckDB up to that long after the window ended. A grace + # shorter than the batch wait publishes the window before those events + # land. After idle_close_seconds with nothing arriving, every open + # bucket closes. + # + # late_rows: drop, because the sink replaces. A row for a bucket that + # already closed is discarded and counted in window_late_rows_total. + # reemit would run emit_sql over the late rows alone, because the + # bucket's other rows were deleted when it closed, and the upsert would + # replace the bucket's count with theirs. validate refuses that pairing. + window: + time_column: bucket + size_seconds: 60 + grace_seconds: 60 + idle_close_seconds: 60 + late_rows: drop + poll_interval_seconds: 1 + emit_sql: | + SELECT bucket, lang, sum(posts)::INTEGER AS posts + FROM closed + GROUP BY ALL + sink: + # A keyed sink replaces the row a (bucket, lang) already identifies. + # The engine deletes a window only after the sink accepts it, so a + # crash between the two republishes that window on the next poll, + # and the republish lands harmlessly. Each flush costs the batch: a + # COPY into a staging table and a server-side merge, in one + # transaction. The target needs PRIMARY KEY (bucket, lang) or a + # unique index on the pair; the sink checks at startup. + type: postgres + postgres: + dsn: "{{ SQLFLOW_POSTGRES_URI }}" + table: posts_per_minute_by_lang + mode: upsert + key: [bucket, lang] + +pipeline: + name: bluesky-posts-per-minute-by-lang + batch_size: 500 + flush_interval_seconds: 3 + + source: + type: websocket + websocket: + uri: "{{ SQLFLOW_JETSTREAM_URI|default('wss://jetstream2.us-east.bsky.network/subscribe?wantedCollections=app.bsky.feed.post') }}" + + handler: + type: handlers.StructuredBatch + table: posts + sql: | + INSERT INTO posts_per_minute_by_lang + SELECT + time_bucket(INTERVAL '1 minute', to_timestamp(time_us / 1000000)) AS bucket, + coalesce(commit.record.langs[1], 'unknown') AS lang, + count(*) AS posts + FROM posts + WHERE commit.operation = 'create' + GROUP BY bucket, lang + + sink: + type: noop diff --git a/dev/bench/bluesky/postgres.sql b/dev/bench/bluesky/postgres.sql new file mode 100644 index 00000000..a6af2829 --- /dev/null +++ b/dev/bench/bluesky/postgres.sql @@ -0,0 +1,15 @@ +-- One row per minute per language. The primary key leads with bucket, so a +-- time-range scan uses it and no second index is needed. +-- +-- The pipeline's upsert depends on this being a primary key. The DuckDB +-- postgres extension honours ON CONFLICT against a primary key and fails with a +-- binder error against a unique index alone. +CREATE TABLE IF NOT EXISTS posts_per_minute_by_lang ( + bucket TIMESTAMPTZ NOT NULL, + lang TEXT NOT NULL, + posts INTEGER NOT NULL, + -- Wall-clock time of the last write. bucket is event time. A reader needs + -- both to tell "the stream is behind" from "the stream stopped". + updated_at TIMESTAMPTZ NOT NULL DEFAULT now(), + PRIMARY KEY (bucket, lang) +); diff --git a/dev/bench/leakloop.Dockerfile b/dev/bench/leakloop.Dockerfile new file mode 100644 index 00000000..e86a8ee5 --- /dev/null +++ b/dev/bench/leakloop.Dockerfile @@ -0,0 +1,20 @@ +# Linux runner for the component leak loops: Go and libduckdb at the version +# in DUCKDB_VERSION. The loops read RssAnon, which only Linux reports exactly. +# +# docker build -f dev/bench/leakloop.Dockerfile -t sqlflow-leakloop . +# +# dev/bench/leakloops.sh builds it if it is missing. +FROM golang:1.25-bookworm + +ENV GOTOOLCHAIN=auto +ENV GOFLAGS=-buildvcs=false + +RUN apt-get update \ + && apt-get install -y --no-install-recommends ca-certificates curl unzip \ + && rm -rf /var/lib/apt/lists/* + +COPY DUCKDB_VERSION /tmp/duckdb/DUCKDB_VERSION +COPY scripts/install-libduckdb.sh /tmp/duckdb/scripts/install-libduckdb.sh +RUN /tmp/duckdb/scripts/install-libduckdb.sh /usr/local/lib + +ENV SQLFLOW_DUCKDB_LIB=/usr/local/lib/libduckdb.so diff --git a/dev/bench/leakloops.sh b/dev/bench/leakloops.sh new file mode 100755 index 00000000..ff9741b9 --- /dev/null +++ b/dev/bench/leakloops.sh @@ -0,0 +1,58 @@ +#!/usr/bin/env bash +# Run component leak loops on Linux, each test in its own process. +# +# dev/bench/leakloops.sh ... +# +# dev/bench/leakloops.sh leak-out 'TestSinkSQLCommand__Postgres' ./internal/sinks +# +# The loops build only with -tags leakloop: they report a rate and assert +# nothing, so they are not part of the test suite. One process per test +# because a loop inherits the allocator state of every loop that ran before it +# in the same binary: the third StructuredBatch loop started 78 MiB above the +# first. Writes /.txt. +# +# Environment, passed to every loop: +# SQLFLOW_LEAK_SCALE multiplies every loop's event count (default 1) +# SQLFLOW_LEAK_JETSTREAM host path to a capture from record.py +# SQLFLOW_LEAK_POSTGRES connection string as the container sees it +# SQLFLOW_LEAK_MALLOC_TRIM malloc_trim(0) before every sample when set +# MALLOC_ARENA_MAX glibc arena limit +# BENCH_NETWORK docker network (default dev_default) +set -euo pipefail + +out=${1:?out-dir}; pattern=${2:?test regexp}; shift 2 +[ $# -gt 0 ] || { echo "name at least one package" >&2; exit 2; } +root=$(cd "$(dirname "$0")/../.." && pwd) +image=sqlflow-leakloop +mkdir -p "$out"; out=$(cd "$out" && pwd) + +if ! docker image inspect "$image" >/dev/null 2>&1; then + docker build -q -f "$root/dev/bench/leakloop.Dockerfile" -t "$image" "$root" >/dev/null +fi + +args=(--rm --network "${BENCH_NETWORK:-dev_default}" + -v "$root":/src -v "$out":/out + -v sqlflow-leakloop-gomod:/go/pkg/mod -v sqlflow-leakloop-gocache:/root/.cache/go-build) +for v in SQLFLOW_LEAK_SCALE SQLFLOW_LEAK_POSTGRES SQLFLOW_LEAK_MALLOC_TRIM MALLOC_ARENA_MAX; do + [ -n "${!v:-}" ] && args+=(-e "$v=${!v}") +done +if [ -n "${SQLFLOW_LEAK_JETSTREAM:-}" ]; then + # Mounted under its own name: Posts reads the .gz suffix to know it is gzipped. + capture=$(basename "$SQLFLOW_LEAK_JETSTREAM") + args+=(-v "$(cd "$(dirname "$SQLFLOW_LEAK_JETSTREAM")" && pwd)/$capture":"/capture/$capture":ro + -e "SQLFLOW_LEAK_JETSTREAM=/capture/$capture") +fi + +docker run "${args[@]}" "$image" bash -c ' + set -uo pipefail + pattern=$1; shift + cd /src + for pkg in "$@"; do + name=$(basename "$pkg") + go test -tags leakloop -c -o "/tmp/$name.test" "$pkg" || exit 1 + for t in $(cd "/src/$pkg" && "/tmp/$name.test" -test.list "$pattern"); do + (cd "/src/$pkg" && "/tmp/$name.test" -test.run "^$t\$" -test.v -test.count=1 -test.timeout 120m) > "/out/$t.txt" 2>&1 + printf "%-60s %s\n" "$t" "$(grep -E "^\s+per [a-z]+ after warm-up" "/out/$t.txt" | sed "s/^ *//" || grep -E "^--- " "/out/$t.txt")" + done + done +' _ "$pattern" "$@" diff --git a/dev/bench/record.py b/dev/bench/record.py new file mode 100644 index 00000000..9dc2a34b --- /dev/null +++ b/dev/bench/record.py @@ -0,0 +1,37 @@ +# Record real Jetstream post events to gzipped NDJSON. +# +# uv run --with websockets python dev/bench/record.py +# +# Jetstream replays history from a cursor as fast as the client reads, about +# 5,000 posts a second from here against roughly 30 to 50 live, so two million +# real posts take about seven minutes rather than a day. The capture feeds +# dev/bench/replay and SQLFLOW_LEAK_JETSTREAM. Posts are public user content: +# keep captures out of the repository. +import asyncio, gzip, json, sys, time +import websockets + +hours_back, target, out = float(sys.argv[1]), int(sys.argv[2]), sys.argv[3] +BASE = "wss://jetstream2.us-east.bsky.network/subscribe?wantedCollections=app.bsky.feed.post" + +async def main(): + cursor = int((time.time() - hours_back * 3600) * 1_000_000) + n, start, last_report = 0, time.time(), time.time() + with gzip.open(out, "wt", compresslevel=3) as f: + while n < target: + try: + async with websockets.connect(f"{BASE}&cursor={cursor}", max_size=2**22) as ws: + async for raw in ws: + f.write(raw if raw.endswith("\n") else raw + "\n") + n += 1 + cursor = json.loads(raw)["time_us"] + if time.time() - last_report > 30: + last_report = time.time() + print(f"{n} events, {n/(time.time()-start):.0f}/s, stream at {time.strftime('%H:%M', time.gmtime(cursor/1e6))} UTC", flush=True) + if n >= target: + break + except Exception as e: + print(f"reconnect after {n} events: {e!r}", flush=True) + await asyncio.sleep(2) + print(f"done: {n} events in {time.time()-start:.0f}s -> {out}", flush=True) + +asyncio.run(main()) diff --git a/dev/bench/replay-soak.sh b/dev/bench/replay-soak.sh new file mode 100644 index 00000000..1f4eed58 --- /dev/null +++ b/dev/bench/replay-soak.sh @@ -0,0 +1,81 @@ +#!/usr/bin/env bash +# Run a pipeline container under Render's limits and decompose its memory +# once a minute. +# +# dev/bench/replay-soak.sh