feat: stage cold appends before the claim, and stream a tiered INSERT's cold rows

CLAUDE.md

@@ -1,3 +1,13 @@
+# HARD RULES: Design Tenets
+
+Every change keeps all five. A change that cannot is raised before it is made, never worked around.
+
+- Transparent to the user: a ColdFront table is used exactly as a native PostgreSQL table. For every supported feature the SQL, the result, the errors and the transaction semantics are what PostgreSQL gives; no new syntax, no extra step, no ColdFront function in the user's statement. A feature outside that set is refused with a clear error, never approximated.
+- Async where possible: a write does its work inside the statement and takes the serializer (the bakery claim, or the node's advisory lock) at commit, for the catalog commit alone. Claim-first is the exception, for a write whose position deletes name data files a concurrent commit could rewrite.
+- Streaming where possible: rows flow from their source to the cold tier in one pass through the DuckDB pipeline, never materialized or rendered row by row. A row-at-a-time path is the fallback for the cases where streaming is proven unsafe, and the fallback must be linear in the row count.
+- One isolation level across both tiers: the transaction isolation PostgreSQL gives a session applies to its cold-tier reads and writes as to its hot-tier ones. The two halves of a statement see one snapshot, and nothing on the cold side sees what the hot side of the same transaction cannot.
+- Interoperable Iceberg: a ColdFront table is a standard Apache Iceberg table that any other Iceberg tool can read, write and compact, and whose writes by those tools ColdFront reads. Nothing ColdFront writes needs a private layout, property or metadata field, and a commit by another tool is a normal event, never a corruption.
+
 # Development Preferences
 
 ## Process

ci/journey.sh

@@ -561,7 +561,7 @@ EOSQL
 # ───────────────────────────────────────────────────────────────────────────
 # Decoupled concurrency / no-409 — parallel cold writers to ONE iceberg-only
 # table must all land. Vanilla serializes them with the local advisory-lock
-# bakery (_exec_iceberg_with_claim, v_armed=false → pg_advisory_xact_lock);
+# bakery (_exec_iceberg_with_claim; _bakery_armed() false → pg_advisory_xact_lock);
 # without it, concurrent Iceberg commits race Lakekeeper's assert-ref-snapshot
 # precondition and 409 (CatalogCommitConflict), which aborts the losing
 # transaction. (Standing rule: multi-writer no-409 probe in vanilla. The bakery
@@ -2399,14 +2399,15 @@ SQL
 # Story 6g: the compactor claims the reference cold writes claim, on a table
 # create_iceberg_table() made (TC-183) and on one adoption registered (TC-184),
 # as on the archiver's. Every registration stores one spelling, each part
-# quoted, and it is the one the compactor builds, so a cold write holding the
-# table's claim makes the compactor's claim wait. Each step then reads the table
-# under its claim, so it works from the write it waited for: expiry (TC-183),
-# compaction (TC-184), and orphan cleanup (TC-185), which with no age window
-# would otherwise delete that write's files.
+# quoted, and it is the one the compactor builds. An append's claim is taken at
+# its COMMIT, so a transaction with a staged cold write holds nothing the
+# compactor waits on: expiry (TC-183), compaction (TC-184) and orphan cleanup
+# (TC-185) run beside the open write, and the write still lands on the result,
+# its staged file kept by the --orphan-age window that protects every
+# in-flight write.
 # ───────────────────────────────────────────────────────────────────────────
 story_compactor_claim_matches_writes() {
-    step "6g. The compactor waits for a cold write's claim on created and adopted tables"
+    step "6g. The compactor runs beside an open cold write on created and adopted tables"
     require_compactor || return
     local col='[{"name":"id","type":"bigint"}]' t
     for t in cf_cw cf_ad cf_or; do
@@ -2417,22 +2418,22 @@ story_compactor_claim_matches_writes() {
     assert_eq "TC-183: the archiver, create_iceberg_table() and adoption register one spelling" \
         '"ice"."public"."cf_ad" "ice"."public"."cf_cw" "ice"."public"."events"' \
         "$(q "$HOST" "SELECT string_agg(iceberg_table, ' ' ORDER BY relname) FROM coldfront.tiered_views WHERE schema_name = 'public' AND relname IN ('cf_ad', 'cf_cw', 'events');")"
-    compactor_waits_for_write TC-183 cf_cw 2 'expired [1-9][0-9]* snapshot' \
+    compactor_runs_beside_write TC-183 cf_cw 2 'expired [1-9][0-9]* snapshot' \
         --expire-snapshots --expire-older-than 0s --expire-retain-last 1 --expire-keep-files
-    compactor_waits_for_write TC-184 cf_ad 5 'compacted: [0-9]+ files'
-    compactor_waits_for_write TC-185 cf_or 1 'deleted [0-9]+ orphan' --orphans --orphan-age 0s
+    compactor_runs_beside_write TC-184 cf_ad 5 'compacted: [0-9]+ files'
+    compactor_runs_beside_write TC-185 cf_or 1 'deleted 0 orphan' --orphans
     for t in cf_cw cf_ad cf_or; do
         q "$HOST" "SELECT coldfront.drop_iceberg_table('public','$t', true);" >/dev/null 2>&1
     done
 }
 
-# compactor_waits_for_write <TC> <table> <writes> <ERE> [flags...]: after <writes>
-# single-row cold writes to public.<table>, one more holds the table's claim for
-# 8 s while the compactor runs with <flags>. The compactor's claim waits on the
-# lock the write holds; once the write commits, the compactor succeeds, its
-# output matches <ERE>, and every write is still there.
-compactor_waits_for_write() {
-    local tc=$1 t=$2 n=$3 pat=$4 w c ec i waited=0
+# compactor_runs_beside_write <TC> <table> <writes> <ERE> [flags...]: after
+# <writes> single-row cold writes to public.<table>, one more stays open for 8 s
+# with its append staged and its claim due at COMMIT, while the compactor runs
+# with <flags>. The compactor finishes while the write is still open, its output
+# matches <ERE>, and every write is still there once the write commits.
+compactor_runs_beside_write() {
+    local tc=$1 t=$2 n=$3 pat=$4 w c ec i open=0
     shift 4
     for i in $(seq 1 "$n"); do echo "INSERT INTO $t VALUES ($i);"; done | qf "$HOST" >/dev/null 2>&1
     qf "$HOST" >/dev/null 2>&1 <<SQL &
@@ -2449,16 +2450,11 @@ SQL
     done
     compactor --table "$t" "$@" >"$TMPD/cf.$tc" 2>&1 &
     c=$!
-    for i in $(seq 1 24); do
-        if [ "$(q "$HOST" "SELECT count(*) FROM pg_stat_activity WHERE wait_event_type = 'Lock' AND query LIKE '%coldfront._claim_iceberg_external%' AND pid <> pg_backend_pid();")" = 1 ]; then
-            waited=1; break
-        fi
-        sleep 0.25
-    done
     wait "$c"; ec=$?
+    open=$(q "$HOST" "SELECT count(*) FROM pg_stat_activity WHERE application_name = 'cf_${tc}_writer' AND wait_event = 'PgSleep';")
     wait "$w"
-    assert_eq "$tc: the compactor's claim waits for the cold write's" 1 "$waited"
-    assert_eq "$tc: the compactor finishes once the write commits" 0 "$ec"
+    assert_eq "$tc: the compactor finished while the cold write was still open" 1 "$open"
+    assert_eq "$tc: the compactor succeeded" 0 "$ec"
     [ "$ec" = 0 ] || tail -5 "$TMPD/cf.$tc"
     assert_eq "$tc: the compactor did its work" 1 "$(grep -c -E "$pat" "$TMPD/cf.$tc")"
     assert_eq "$tc: every write is still there" "$((n + 1))" "$(q "$HOST" "SELECT count(*) FROM $t;")"
@@ -3503,7 +3499,7 @@ story_mesh() {
     assert_eq "write from peer $p1 visible on db1" "1" "$(q "$HOST" "SELECT count(*) FROM iceonly WHERE status='from_peer';")"
 
     # R-A bakery under multi-node contention: concurrent cold writers on db1 AND
-    # a peer to the SAME Iceberg table must both land. Here v_armed is true
+    # a peer to the SAME Iceberg table must both land. Here _bakery_armed() is true
     # (snowflake.node + loopback_dsn set), so this exercises the Ricart-Agrawala
     # claim protocol across nodes — not the local advisory lock — to avoid 409.
     rm -f $TMPD/ra.* 2>/dev/null
@@ -3883,7 +3879,8 @@ story_mesh_claim_failures() {
     ref_t=$(q "$HOST" "SELECT iceberg_table FROM coldfront.tiered_views WHERE relname = 'cf_ct';")
 
     # TC-174: a claim times out while another session holds the table's claim
-    # key. The session holds no advisory lock afterwards, and its next write
+    # key. An append's claim is taken at its COMMIT, which is where the timeout
+    # fires. The session holds no advisory lock afterwards, and its next write
     # clears what the timed-out attempt and an orphan planted on another table
     # left on the node.
     q "$HOST" "INSERT INTO coldfront.claims (iceberg_table, ticket) VALUES ('${ref_t}_tc174', snowflake.nextval());" >/dev/null
@@ -3897,7 +3894,9 @@ SQL
     sleep 1
     out=$(sess "$HOST" 2>&1 <<SQL
 SET statement_timeout = '1s';
+BEGIN;
 INSERT INTO cf_ct VALUES (1741);
+COMMIT;
 RESET statement_timeout;
 SELECT 'locks=' || count(*) FROM pg_locks WHERE locktype = 'advisory' AND pid = pg_backend_pid();
 SELECT pg_sleep(4);
@@ -3911,7 +3910,9 @@ SQL
     assert_eq "TC-174 no claim is left on the node for either table" "0" "$(node_claims "$ref_t")"
 
     # TC-175: one transaction writes cf_ct twice; a peer's write to cf_ct arrives
-    # between the two and stays behind the transaction's claim until COMMIT.
+    # between the two and lands at once. An append's claim is taken at its
+    # COMMIT, so an open transaction holds nothing a peer waits on, and all
+    # three writes land.
     sess "$HOST" >"$TMPD/cf.175a" 2>&1 <<SQL &
 BEGIN;
 INSERT INTO cf_ct VALUES (1751);
@@ -3927,12 +3928,13 @@ SQL
     p_done=$(date +%s.%N)
     wait "$a" 2>/dev/null
     a_commit=$(sed -n 's/^a_commit=//p' "$TMPD/cf.175a")
-    assert_eq "TC-175 a second write to the same table keeps the transaction's claim" "1" "$(later "$p_done" "$a_commit")"
+    assert_eq "TC-175 a peer's write lands while the transaction is still open" "0" "$(later "$p_done" "$a_commit")"
     assert_eq "TC-175 no write errored" "0" "$(cat "$TMPD"/cf.175? | grep -c ERROR)"
     assert_eq "TC-175 all three writes landed" "3" "$(q "$HOST" "SELECT count(*) FROM cf_ct WHERE id BETWEEN 1751 AND 1753;")"
 
     # TC-176: one transaction writes cf_cu, then cf_ct; a peer's write to cf_cu
-    # arrives between the two and stays behind the cf_cu claim until COMMIT.
+    # arrives between the two and lands at once, as in TC-175, and every write
+    # on both tables lands.
     sess "$HOST" >"$TMPD/cf.176a" 2>&1 <<SQL &
 BEGIN;
 INSERT INTO cf_cu VALUES (1761);
@@ -3948,7 +3950,7 @@ SQL
     p_done=$(date +%s.%N)
     wait "$a" 2>/dev/null
     a_commit=$(sed -n 's/^a_commit=//p' "$TMPD/cf.176a")
-    assert_eq "TC-176 a write to a second table keeps the transaction's claim on the first" "1" "$(later "$p_done" "$a_commit")"
+    assert_eq "TC-176 a peer's write to the first table lands while the transaction is still open" "0" "$(later "$p_done" "$a_commit")"
     assert_eq "TC-176 no write errored" "0" "$(cat "$TMPD"/cf.176? | grep -c ERROR)"
     assert_eq "TC-176 all three writes landed" "3" \
         "$(q "$HOST" "SELECT (SELECT count(*) FROM cf_cu WHERE id IN (1761, 1763)) + (SELECT count(*) FROM cf_ct WHERE id = 1762);")"
@@ -4205,7 +4207,7 @@ story_standby_reads() {
     fi
     # The cold sink's step function is reached through the INSERT rewrite, so call
     # it directly: the guard is its first statement, ahead of every lookup.
-    local c; c=$(q_may "$STANDBY" "SELECT coldfront._cold_sink_step(NULL, '${vn%%.*}', '${vn##*.}', '{}'::jsonb);")
+    local c; c=$(q_may "$STANDBY" "SELECT coldfront._cold_sink_step(NULL, '${vn%%.*}', '${vn##*.}', '', 0);")
     assert_err "cold sink on standby → coldfront refuses before the claim" \
         "cannot execute a cold (Iceberg) write on a read-only standby" "$c"
 

docker/entrypoint.sh

@@ -62,10 +62,11 @@ shared_preload_libraries = '${PRELOAD}'
 # postgres_scanner + any non-bundled deps auto-install; it does NOT clobber the
 # pre-placed iceberg/avro (DuckDB skips install for extensions already present).
 # allow_unsigned ON so the locally-built (unsigned) extension loads; autoload ON
-# so ATTACH (TYPE ICEBERG, ...) lazily LOADs it; iceberg_async_parquet ON so the
-# mesh bakery uploads parquet in the background and serializes only the commit
-# POST (safe because the patch refreshes parent_snapshot_id at commit). Vanilla
-# ignores the flag (advisory lock, claim-first).
+# so ATTACH (TYPE ICEBERG, ...) lazily LOADs it; iceberg_async_parquet ON so
+# every cold write uploads its parquet before it takes the serializer (the
+# bakery claim on a mesh node, the advisory lock on a vanilla one) and only the
+# commit POST is serialized (safe because the patch refreshes
+# parent_snapshot_id at commit).
 #
 # iceberg_bakery_patch = on ASSERTS that the duckdb-iceberg in THIS image carries
 # the bakery-aware-commit-refresh patch (it does — the Dockerfile git-applies it).

docs/architecture.md

@@ -343,33 +343,38 @@ ColdFront coordinates concurrent writes across the cluster as follows:
   iceberg-only `INSERT`/`UPDATE`/`DELETE`/`MERGE` wraps in
   `coldfront._exec_iceberg_with_claim`, which holds a globally-ordered
   Snowflake ticket via the Spock-replicated `coldfront.claims` table and waits
-  for its turn before issuing the iceberg commit. There are no 409s and no
-  app-level retry. See the
+  for its turn before issuing the iceberg commit. The bakery produces no 409s
+  and needs no retry in the application. See the
   [Concurrency](architecture_decoupled.md#concurrency-horizontal-scaling-the-bakery-protocol)
   section of the Decoupled Mode page for the full design and benchmarks.
 
 ### Cold-Write Strategy: Stock vs Patched duckdb-iceberg
 
-Most cold writes run through `_exec_iceberg_with_claim`; the tiered `INSERT`'s
-cold sink and the cross-tier move take the claim through `_take_iceberg_claim`
-themselves. What differs is *when* the bakery ticket is held. The async
-ordering is used only on a mesh node with the bakery enabled, and only when
-**both** `coldfront.iceberg_async_parquet` (default `off`) and the build marker
-`coldfront.iceberg_bakery_patch` (asserting the loaded duckdb-iceberg includes
-the patch) are on (`coldfront._iceberg_async_active()`); a vanilla node always
-takes its lock before uploading. The following table shows the two orderings:
+Every cold write runs through `_exec_iceberg_with_claim`. What differs is
+*when* the serializer (the bakery ticket on a mesh node, the advisory lock on a
+vanilla node) is held. The async ordering is used on every topology, for
+appends, when **both** `coldfront.iceberg_async_parquet` (default `off`) and
+the build marker `coldfront.iceberg_bakery_patch` (asserting the loaded
+duckdb-iceberg includes the patch) are on (`coldfront._iceberg_async_active()`).
+The following table shows the two orderings:
 
-| Ordering | duckdb-iceberg | Behavior |
+| Ordering | Applies to | Behavior |
 |---|---|---|
-| stock (default) | stock upstream | The writer takes the bakery ticket first, *then* uploads parquet and commits inside the ticket. This ordering is correct on an unpatched binary, but the whole parquet upload happens under the lock, so concurrent writers serialize on upload and commit. |
-| async (both GUCs `on`) | patched (`iceberg-bakery-aware-commit-refresh-v15.patch`) | The writer uploads parquet in the background *first*, then takes the ticket only for the Lakekeeper commit. Concurrent writers' uploads overlap; only the short commit POST is serialized. |
-
-The code path and the application-visible behavior are identical, so the GUCs
-are purely a performance knob. The patch relocates parent-snapshot stamping
-from upload time into PG's pre-commit phase (inside the bakery ticket, against
-a freshly-fetched table), so overlapping uploads cannot commit a stale parent.
-Async requested without the build marker downgrades safely to the stock
-ordering, noted once per session with a server LOG line - never a silent 409.
+| stock (default) | every write on stock upstream duckdb-iceberg; a `DELETE`, `UPDATE` or `MERGE` (the cross-tier move and `vector_train`'s assignment included) on every build | The writer takes the serializer first, *then* uploads parquet and commits inside it. On an unpatched binary this is what keeps a second writer from capturing a stale parent. A write that references rows the table holds needs it on every build: its position deletes name data files by path, and a compaction or another such write committing between its scan and its commit would rewrite or remove those files, after which the deletes apply to nothing and the rows come back. |
+| async (both GUCs `on`) | appends (`INSERT`, `COPY`, the archiver's exports, the tiered `INSERT`'s cold sink) on the patched build (`iceberg-bakery-aware-commit-refresh-v15.patch`) | The statement uploads its parquet and queues the claim; the extension's transaction callback takes it at PRE_COMMIT, just before pg_duckdb's own PRE_COMMIT callback commits the Iceberg transaction, and releases it at COMMIT. Uploads overlap on one node and across nodes, an open transaction blocks no other writer, and only the short commit POST is serialized. A new data file conflicts with nothing committed in between. An append to a clustered table is checked at commit against the live centroid generation and fails with `serialization_failure` if a retrain replaced it (see [architecture_vectors.md](architecture_vectors.md)). |
+
+For every write but a clustered table's append, the code path and the
+application-visible behavior are identical, so the GUCs are a performance knob;
+a clustered append in the async ordering can fail with `serialization_failure`
+when a retrain overlaps it, which the claim-first ordering, waiting for the
+retrain instead, cannot. The patch refreshes the table from the catalog
+in PG's pre-commit phase (inside the serializer) and re-derives every pending
+snapshot of the transaction from that head, so overlapping uploads and several
+batches alike cannot commit a stale parent. A staged file is referenced by no
+snapshot until its transaction commits, which is what the compactor's
+`--orphan-age` window is for. Async requested without the build marker
+downgrades safely to the stock ordering, noted once per session with a server
+LOG line - never a silent 409.
 The Docker image ships the patched binary and sets both GUCs on
 (`docker/entrypoint.sh`); bare-metal users on a stock binary leave both `off`
 and lose only the upload overlap. See
@@ -649,14 +654,16 @@ pg_duckdb has a fully-native, in-process Postgres-table reader for analytics on
 PG heap data, but that machinery is **not reachable** from the write path into
 an attached Iceberg catalog.
 
-As the workarounds in use today, a tiered `INSERT` renders its cold rows in the
-backend (`coldfront._cold_sink`) and writes them to Iceberg in batched
-`duckdb.raw_query` `INSERT`s, and a decoupled `INSERT … SELECT` from a
-PostgreSQL table reads it through the DuckDB `postgres` extension's
-`pglocal.<schema>.<table>` ATTACH, pipelining rows over libpq (loopback) →
-DuckDB executor → Iceberg writer → S3. Neither is the in-process reader:
-the first pays for the per-row rendering, the second for the libpq round-trip
-per row batch.
+As the workarounds in use today, a tiered `INSERT` streams its cold rows
+through `postgres_query('pgstream', …)` and a decoupled `INSERT … SELECT`
+from a PostgreSQL table reads it through the `pglocal.<schema>.<table>`
+ATTACH, both of the DuckDB `postgres` extension, pipelining rows over libpq
+(loopback) → DuckDB executor → Iceberg writer → S3; a tiered `INSERT` whose
+source cannot be read from a second session renders its cold rows in the
+backend (`coldfront._cold_sink`) and writes them in batched `duckdb.raw_query`
+`INSERT`s. None is the in-process reader: the streams pay for the libpq
+round-trip per row batch and for reading the source once per tier, the sink
+for the per-row rendering.
 
 The desired end-state is a way to drive the native in-process reader straight
 into the Iceberg writer - e.g. a `COPY` form: