Window tables: a leak loop, and examples that append and sum instead of an indexed upsert

[turbolytics]

What this changes

The Bluesky demo grew from 100 MB to 450 MB in a day on a 512 MB plan. The cause is the window pattern every tumbling example and the README taught: a UNIQUE INDEX on the window table with an ON CONFLICT upsert. DuckDB never frees rows deleted from an indexed table once a checkpoint writes them, and the manager deletes every closed window, so the table grows without bound. That happens in memory and with a state path alike. See #268.

Two commits:

1. A leak loop. internal/managers/window_leak_test.go drives the real StructuredBatch handler and the real tumbling manager over one DuckDB connection and reads DuckDB's own accounting. It runs 1,000 batches in the -short pass, about three seconds a test, and SQLFLOW_LEAK_BATCHES scales it to millions of messages.
2. Examples and README that no longer leak. Each window table drops its index. The handler appends a batch's counts, and the collect query sums them with GROUP BY ALL, casting the sum back to the column type.

Changed configs: tumbling.window.yml, kafka.stateful.window.yml, bluesky/bluesky.kafka.windowed.yml, bluesky/bluesky.postgres.windowed.yml, and .claude/skills/slow-soak/slow.yml. The README's Tumbling windows section now shows the handler SQL and warns against the index.

What breaks if this is wrong: a window publishes split or partial counts, or a sink upsert sees a duplicate key. The A/B below shows neither.

Verification

The leak loop, 500,000 messages per test:

Window table	Row groups	Rows held	Live rows	DuckDB table memory
Unique index, upsert, in memory	2 → 85	20,400	120	1 → 63 MiB
Unique index, upsert, file-backed	2 → 85	20,400	120	1 → 64 MiB
No index, plain insert, collect sums	1 → 1	960	480	0 → 0 MiB

The no-index table held one row group through 15 million messages and 30,000 polls.

The examples, old SQL from origin/main against the branch:

• SQL A/B, all five configs. The same 400 batches with late events and an idle close published identical rows in identical column types. The Postgres example's sink upserts produced identical keyed tables. The harness fails at batch 9 on a config missing the sum.
• Engine A/B, tumbling.window.yml. Old and new configs ran against Kafka on the same 3,000 events and published the same 27 rows, one per key, 2,433 total.
• sqlflow validate: all five valid.

The template checks:

• go test -short -race ./...: pass, no data races.
• uv run --locked pytest tests/tooling -q: 205 passed.
• make coverage-page && git status --short docs/coverage: clean.
• uv run --locked pytest tests/release -q -k "state_durability or config_validation_accepts": 2 passed. The durability test runs the changed kafka.stateful.window.yml through a restart. Run against turbolytics/sql-flow:v1.2.0, because the engine is unchanged here and the suite mounts dev/config from the checkout.
• make soak: not run. No engine code path changes.

Notes for the reviewer

• GROUP BY ALL, not GROUP BY bucket. Two examples output a reformatted bucket AS bucket. DuckDB resolves GROUP BY bucket to the table's column and splits one output window into several rows. Measured on DuckDB 1.5.2.
• benchmark.stateful.mem.yml keeps its index. Its manager deletes nothing, and an index with upserts and no deletes held one row group over 40,000 batches. Removing the index there would grow the table forever and invalidate the durable-state cost numbers.
• Resident memory is logged, not asserted. On macOS it is peak RSS and only rises. The Linux memory soak stays the gate for native growth.
• DuckDB has an experimental fix, not exposed by sqlflow. vacuum_rebuild_indexes lets a checkpoint compact an indexed table and rebuild its ART index. It is off by default and startup-only, so commands: cannot set it. Passed through sqlflow's ADBC database options, it made the indexed upsert flat through the real handler and manager: one row group, 240 rows for 120 live, over 500,000 messages, in memory and on disk. Exposing it is a separate engine change.
• Found along the way: StructuredBatch with a state path dies after its first batch: CHECKPOINT inside the state transaction #276, and an unfiled crash where a state path plus a window sink writing to an attached database aborts on the first closed window.
• The demo's config fix is pipeline: drop the window table's unique index, which leaked every deleted row sql-flow-bluesky-demo#3.

Related: #268, #247, #276.