A keyed Postgres sink on pgx, and the leak loops that found the demo's growth

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
 

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"]

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.

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

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