Make catalog decoding consistent across DDL & restarts

.github/workflows/ci.yml

@@ -30,6 +30,18 @@ jobs:
         pg-major: [16, 17, 18]
 
     steps:
+      # ubuntu-latest ships ~14GB free on /; `cargo --all-targets` links
+      # every integration test as its own static binary and can exhaust it
+      # mid-link. ENOSPC surfaces as an lld SIGBUS + a spurious "report to
+      # LLVM" banner. Fast `rm -rf` of the big preinstalled dirs
+      # (android/dotnet/haskell, ~21GB) buys enough room; skip large-packages
+      # (slow apt remove, minutes) and swap-storage (frees /mnt, not /).
+      - name: Free disk space
+        uses: jlumbroso/free-disk-space@v1.3.1
+        with:
+          large-packages: false
+          swap-storage: false
+
       - name: Checkout walshadow
         uses: actions/checkout@v7
         with:
@@ -217,6 +229,12 @@ jobs:
     timeout-minutes: 60
 
     steps:
+      - name: Free disk space
+        uses: jlumbroso/free-disk-space@v1.3.1
+        with:
+          large-packages: false
+          swap-storage: false
+
       - name: Checkout walshadow
         uses: actions/checkout@v7
         with:

architecture/decoder.dot

@@ -27,13 +27,14 @@ digraph decoder {
     label="BufferingDecoderSink::on_record (xact_buffer.rs)"; style="rounded,filled"; color="#4c4641"; fillcolor="#34302c"; fontcolor="#ECE1D7";
     sink     [label="on_record\nSMGR CREATE?\nHeap TRUNCATE?\nother heap op?", fillcolor="#4D4128"];
     marker   [label="note_smgr_create\nmain-fork generation marker\nproves stash completeness", fillcolor="#4D4128"];
-    trunc    [label="handle_truncate\nparse_xl_heap_truncate\nwait_for_replay(source_lsn)\nfor relid in relids:\n  relation_by_oid → fan out\n  HeapOp::Truncate\n(kind in {'r','p'} only)", fillcolor="#4D4128"];
-    relat    [label="ShadowCatalog::relation_at(rfn, lsn)\nLRU + libpq fetch", fillcolor="#4D4D28"];
-    stash    [label="stash_invisible\nSpillEntry::Raw\nmain data + block data/images\ncommit: relation_at(rfn, commit_lsn)\ntoast decode | ordinary skip | discard", fillcolor="#4D4128"];
+    trunc    [label="handle_truncate\nparse_xl_heap_truncate\nfor relid in relids:\n  descriptor_by_oid_at_spanned\n  → fan out HeapOp::Truncate\n(kind in {'r','p'} only)", fillcolor="#4D4128"];
+    relat    [label="DescriptorLog::\ndescriptor_at_spanned(rfn, lsn)\nwait-free interval lookup", fillcolor="#4D4D28"];
+    stash    [label="raw stash\nSpillEntry::Raw\nmain data + block data/images + xid\ncommit: resolve_stash at next_lsn\ntoast | ordinary decode | discard\nAmbiguous = fatal", fillcolor="#4D4128"];
     sink -> marker [label="SMGR CREATE", style=dashed];
     sink -> trunc [label="0x30 TRUNCATE", style=dashed];
     sink -> relat [label="other heap ops"];
-    relat -> stash [label="NotFoundByFilenode\nor known candidate", style=dashed];
+    sink -> stash [label="dirty family\n(defer_catalog_decode)", style=dashed];
+    relat -> stash [label="NotCovered / Dropped /\nAmbiguous, or marker\ncandidate", style=dashed];
   }
   record -> sink;
 

architecture/emitter.dot

@@ -1,13 +1,14 @@
 // walshadow — CH emitter component view
 // Parallel decode+insert pipeline: reorder coordinator (commit order,
-// DDL/TRUNCATE barrier) → decode pool ×M → InsertBatcher (per-table
+// side-effect-free transaction plan then execute, DDL/TRUNCATE barrier)
+// → decode pool ×M → InsertBatcher (per-table
 // TableEncoder, budget/deadline seal) → inserter pool ×N (one complete
 // INSERT per sealed batch) → ack collector (contiguous-done watermark).
 // Zoomed-in view of cluster_ch + cluster_ddl from internals.dot.
 //
 // regeneration spec:
-// sources of truth: plans/emitter.md · src/emit/pipeline/{reorder,decode,batcher,inserter,ack,tail}.rs · src/emit/{ch_emitter,ch_ddl}.rs · src/catalog/type_bridge.rs
-// subsumes: plans/emitter.md § "Stage walk" + "Barrier fence" + "Ack-LSN tracking" + "DdlApplicator"
+// sources of truth: plans/emitter.md · src/emit/pipeline/{reorder,planner,plan_spool,decode,batcher,inserter,ack,tail}.rs · src/emit/{ch_emitter,ch_ddl}.rs · src/catalog/type_bridge.rs
+// subsumes: plans/emitter.md § "Stage walk" + "Transaction planner" + "Barrier fence" + "Ack-LSN tracking" + "DdlApplicator"
 // quality bar:
 //   - decode ×M and inserter ×N read as pools (stacked node or ×M/×N label), not single tasks
 //   - barrier fence visually orders DDL strictly after earlier data durable (placed → FlushAll → durable)
@@ -27,15 +28,17 @@ digraph emitter {
     style="rounded,filled"; color="#4c4641"; fillcolor="#34302c"; fontcolor="#ECE1D7";
 
     reorder [label="transaction reorder\nwalk rows and changes in WAL order\nempty commits and aborts still advance progress\nschema and truncate changes wait for earlier rows", fillcolor="#4D4128"];
+    plan    [label="transaction planner (planner.rs + plan_spool.rs)\nplan first, execute after: detoast + route +\nraw decode, side-effect-free; frozen route view\n(whole-xact config granularity)\nplan ≤1 MiB resident else .plan spool\n[len|crc32c|body]*, seal frame; error = abandon,\nnothing emitted", fillcolor="#4D4128"];
     cat     [label="ShadowCatalog::subscribe\nSchemaEvent\nAdded / Changed / Dropped\n(unbounded mpsc, rides xact buffer\n as ordered_events)", fillcolor="#4D4D28", shape=parallelogram];
+    reorder -> plan [label="drain walk\n(plan → seal → execute)"];
   }
 
   // ═════ decode pool ═════
   subgraph cluster_decode {
     label="decode pool ×M  (emit/pipeline/decode.rs)";
     style="rounded,filled"; color="#4c4641"; fillcolor="#34302c"; fontcolor="#ECE1D7";
 
-    decode [label="decode worker ×M\ndetoast: ChunkMap, then store as-of fetch\nresolve_at_pooled; mapping\noracle PgPending + sampled validate\nchunk 1024 rows / 4 MiB", fillcolor="#4D4128"];
+    decode [label="decode worker ×M\nplanned envelopes: descriptor + route attached,\nvalues resolved at planning\noracle PgPending resolve\nchunk 1024 rows / 4 MiB", fillcolor="#4D4128"];
   }
 
   // ═════ batcher hub ═════
@@ -90,7 +93,7 @@ digraph emitter {
   ack [label="ack collector\nwait for every earlier commit\npublish latest durable position\nfeed manifest + source feedback", fillcolor="#4D3A28", shape=note];
 
   // ═════ row path ═════
-  reorder -> decode [color="#BF8C5F", penwidth=2, lhead=cluster_decode, label="DecodeJob\n(mpmc, bound 4M)"];
+  plan -> decode [color="#BF8C5F", penwidth=2, lhead=cluster_decode, label="execute_plan:\nDecodeJob\n(mpmc, bound 4M)"];
   decode  -> enc    [color="#BF8C5F", penwidth=2, lhead=cluster_batch, label="BatcherMsg::Rows\n(FIFO mpsc 256 —\nshared with FlushAll)"];
   seal    -> ins    [color="#BF8C5F", penwidth=2, lhead=cluster_insert, label="InsertBatch\n(mpmc)"];
   ins     -> chrows [color="#BF8C5F", penwidth=2, label="one complete INSERT\nper sealed batch"];