Add pg oracle during bootstrap to decode complex types

plans/bootstrap.md

@@ -54,10 +54,10 @@ for rendered diagram. Five clusters top→bottom:
    `tail.finish` seals partial batches and waits all seqs durable
    before handoff. Metrics-only runs (no `--ch-config`) instead drain
    through `drain_backfill` into a counting `TupleObserver`.
-   Bridge-routed tier-3 values (jsonb, arrays, hstore, …) can't be
-   resolved here — the shadow/bridge don't exist until after bootstrap —
-   so they currently land empty; in-tree types (geography, vector) are
-   fine. Fix in [future/greenfield_oracle.md](future/greenfield_oracle.md)
+   Bridge-routed tier-3 values (jsonb, arrays, hstore, enums, …) are
+   resolved here by the bootstrap oracle (see below), since the real
+   shadow/bridge don't exist until after bootstrap; in-tree types
+   (geography, vector) render without it.
 4. **Shadow handoff** — `BootstrapOutcome { start, end }` returned;
    daemon writes `standby.signal` and calls `materialize_conf` to
    replace shadow's config files. Config includes walshadow settings,
@@ -75,6 +75,37 @@ for rendered diagram. Five clusters top→bottom:
 Phases 1-3 run synchronously inside `run_bootstrap`; phases 4-5 hand off
 to daemon's main loop
 
+## Bootstrap oracle — tier-3 resolution
+
+The page-walk yields raw on-disk Datums, so bridge-routed tier-3 types
+(jsonb, arrays, hstore, citext, enums, ranges, domains) carry the source
+`atttypid` and need a PG to run `typoutput` by OID — but the real shadow
+isn't up until phase 4. `run_bootstrap` therefore stands up a throwaway,
+**OID-exact** side PG (`BootstrapOracle`, `src/backfill/bootstrap_oracle.rs`)
+before the drain and points the bridge/oracle at it, resolving tier-3 inline,
+single-pass, for Direct and ObjectStore alike.
+
+- **OID-exact from schema only.** `initdb` → start `postgres -b` (binary-upgrade
+  mode) → `pg_dump --binary-upgrade --schema-only --no-owner --no-privileges`
+  from source, applied in → stop → restart normal with `shared_preload_libraries
+  = 'walshadow'`. `--binary-upgrade` pins every `pg_type`/`pg_enum` OID (and
+  extension member OIDs) to the source's — the pg_upgrade mechanism. Schema-only,
+  so dataset size is irrelevant. `-b` is needed only to apply the dump; serving
+  runs in normal mode. `Drop` stops it and removes the scratch dir.
+- **Drain wiring.** `bootstrap::drain` takes `oracle: Option<Arc<Oracle>>`; a
+  `resolve_row` helper runs on every route path — `render_ext_columns` (in-tree
+  geography/vector, no PG) then `resolve_pending_tuple` (bridge).
+- **Gated.** Provisioned only when the mapped set has a bridge-routed tier-3
+  column (`needs_bridge`); in-tree-only (vector/geography) or scalar schemas skip
+  it.
+- **Fails hard.** If provisioning fails (e.g. an extension `.so` isn't
+  installable on the host — same requirement the shadow already has), bootstrap
+  errors out rather than loading empty columns; `Unsupported` stays fail-closed.
+- **Opt-in backup-backfill** (`initial_load = base_backup`/`object_store`,
+  `backup_backfill.rs`) has the same raw-Datum shape and reuses the **live**
+  oracle via `PassContext.oracle` — steady state, real shadow already up.
+- Coverage validated by `tests/bootstrap_types_e2e.rs`.
+
 ## Restart contract
 
 A **completion marker** (`walshadow.bootstrap_complete`, written into the

plans/future/INDEX.md

@@ -16,7 +16,6 @@ rationale under `plans/` only when code cannot express it
 * [failover.md](failover.md) — beyond the switchover crossing in [../failover.md](../failover.md): unplanned promotion (transaction-state fence, overwrite contrecord), slotless pause windows, timeline-aware archive and base-backup replay
 * [sync_commit_witness.md](sync_commit_witness.md) — walshadow as RPO=0 durability standby
 * [two_phase_commit.md](two_phase_commit.md) — `XLOG_XACT_PREPARE` handling and gxid-keyed buffer
-* [greenfield_oracle.md](greenfield_oracle.md) — resolve bridge-routed tier-3 (jsonb/arrays/hstore) during greenfield bootstrap via a throwaway PG (walshadow module + source extensions) torn down after; OID-matching constraint (restore-from-backup vs resolve-by-name)
 * [oracle_native_blocks.md](oracle_native_blocks.md) — bridge/pgext emits CH-native column bytes instead of `typoutput` text, batching `DECODE` from row-at-a-time to a column-major list of rows (matches CH's columnar block, amortizes the round trip); removes the emitter-side composite re-parse (`ColumnBuf::{Array,Map,Json}`, text parsers, `NodeArena`); does not solve greenfield
 * [ch_bounce_recovery.md](ch_bounce_recovery.md) — deeper re-emit-from-spill on retry-budget exhaustion
 * [pinned_ddl_baseline.md](pinned_ddl_baseline.md) — schema-event outcome must be a function of config + baseline, not cache warmth: CH-existence / persisted-baseline options for cross-restart consistency, drop detection across downtime, opt-in mapping vs republish

plans/future/oracle_native_blocks.md

@@ -66,8 +66,8 @@ column-major batch):
 
 This only changes *where* serialization happens; it still needs a live PG with
 the walshadow module to do the decoding. Greenfield has none until after
-bootstrap — that gap is [greenfield_oracle.md](greenfield_oracle.md), and the
-two are independent.
+bootstrap — that is handled by the bootstrap oracle
+([../bootstrap.md](../bootstrap.md)), independently of this.
 
 ## Alternative considered
 

plans/oracle.md

@@ -60,9 +60,9 @@ for `Text` on each item that rendered
 Both the live decode pool (`emit/pipeline/decode.rs`) and the object-store /
 COPY backfill paths call it, so backfilled rows resolve identically to
 streamed ones. **Greenfield bootstrap is the exception**: it runs before the
-shadow/bridge exist, so bridge-routed types there resolve to nothing (in-tree
-types still work) — the fix is
-[future/greenfield_oracle.md](future/greenfield_oracle.md).
+shadow/bridge exist, so it resolves them against a throwaway OID-exact side PG
+instead of the shadow — see the bootstrap oracle in
+[bootstrap.md](bootstrap.md).
 
 Two alternatives considered (insert + select round-trip;
 `SELECT $1::bytea::<typ>::text`) require reconstructing wire format from

src/backfill/backfill_types.rs

@@ -10,6 +10,7 @@ use crate::catalog::shadow_catalog::ShadowCatalog;
 use crate::config::ResolvedConfig;
 use crate::emit::ch_emitter::{EmitterConfig, EmitterStats};
 use crate::mapping::MappingHandle;
+use crate::ops::oracle::Oracle;
 use crate::schema::RelDescriptor;
 
 #[derive(Debug, Clone)]
@@ -28,6 +29,7 @@ pub struct PassContext {
     pub scratch_dir: PathBuf,
     pub config_rx: Option<watch::Receiver<Arc<ResolvedConfig>>>,
     pub budget: Option<MemoryBudget>,
+    pub oracle: Option<Arc<Oracle>>,
 }
 
 #[derive(Debug, Default, Clone)]

src/backfill/backup_backfill.rs

@@ -352,6 +352,7 @@ async fn walk_and_ship(
         DeferredSpool::new(toast_spool_path, DEFERRED_SPOOL_MEM_MAX),
         ctx.emitter.row_policy(),
         ctx.config_rx.as_ref().map(|rx| rx.borrow().clone()),
+        ctx.oracle.clone(),
     ));
 
     // Success signal before the joins: gate resolves deferred tuples only

src/backfill/bootstrap_oracle.rs

@@ -0,0 +1,171 @@
+use std::path::PathBuf;
+use std::process::Command;
+use std::sync::Arc;
+use std::time::Duration;
+
+use anyhow::{Context, Result};
+
+use crate::backfill::backup_page_walk::CatalogMap;
+use crate::catalog::shadow::{BridgeConf, Shadow, ShadowConfig};
+use crate::ops::oracle::Oracle;
+use crate::schema::{
+    BOOLOID, BPCHAROID, BYTEAOID, CHAROID, CIDROID, DATEOID, FLOAT4OID, FLOAT8OID, INETOID,
+    INT2OID, INT4OID, INT8OID, INTERVALOID, JSONOID, NAMEOID, NUMERICOID, OIDOID, RelAttr, TEXTOID,
+    TIMEOID, TIMESTAMPOID, TIMESTAMPTZOID, TIMETZOID, UUIDOID, VARCHAROID,
+};
+
+const ORACLE_PORT: u16 = 55440;
+
+pub struct BootstrapOracle {
+    shadow: Shadow,
+    oracle: Arc<Oracle>,
+    base_dir: PathBuf,
+}
+
+impl BootstrapOracle {
+    pub async fn provision(
+        base_dir: PathBuf,
+        source_conninfo: String,
+        source_password: Option<String>,
+        bridge_lib_dir: Option<PathBuf>,
+        connect_budget: Duration,
+    ) -> Result<Self> {
+        let data_dir = base_dir.join("pg");
+        let socket_dir = base_dir.join("sock");
+        let bridge_socket = socket_dir.join("walshadow-bridge.sock");
+
+        let (b_data, b_sock, b_bridge, b_base) = (
+            data_dir.clone(),
+            socket_dir.clone(),
+            bridge_socket.clone(),
+            base_dir.clone(),
+        );
+        let shadow = tokio::task::spawn_blocking(move || -> Result<Shadow> {
+            std::fs::remove_dir_all(&b_base).ok();
+            std::fs::create_dir_all(&b_sock)?;
+
+            let cfg_a = oracle_cfg(&b_data, &b_base, &b_sock, None);
+            let a = Shadow::new(cfg_a);
+            a.initdb().context("initdb")?;
+            a.write_base_conf().context("base conf")?;
+            a.start_binary_upgrade().context("start -b")?;
+            let dump = run_pg_dump(&source_conninfo, source_password.as_deref())
+                .context("pg_dump --binary-upgrade")?;
+            a.apply_schema_dump(&dump).context("apply schema")?;
+            a.stop().context("stop -b")?;
+
+            let mut bridge = BridgeConf::in_dir(&b_sock);
+            bridge.socket_path = b_bridge;
+            bridge.library_dir = bridge_lib_dir;
+            let cfg_b = oracle_cfg(&b_data, &b_base, &b_sock, Some(bridge));
+            let b = Shadow::new(cfg_b);
+            b.write_base_conf().context("serve conf")?;
+            b.start().context("start serve")?;
+            Ok(b)
+        })
+        .await
+        .context("bootstrap oracle provision task")?
+        .context("bootstrap oracle provision")?;
+
+        let bridge = crate::ops::bridge::connect_with_budget(&bridge_socket, connect_budget)
+            .await
+            .context("bootstrap oracle bridge connect")?;
+        Ok(Self {
+            shadow,
+            oracle: Arc::new(Oracle::new(Arc::new(bridge))),
+            base_dir,
+        })
+    }
+
+    pub fn oracle(&self) -> Arc<Oracle> {
+        self.oracle.clone()
+    }
+}
+
+impl Drop for BootstrapOracle {
+    fn drop(&mut self) {
+        let _ = self.shadow.stop();
+        let _ = std::fs::remove_dir_all(&self.base_dir);
+    }
+}
+
+fn oracle_cfg(
+    data_dir: &std::path::Path,
+    filter_out_dir: &std::path::Path,
+    socket_dir: &std::path::Path,
+    bridge: Option<BridgeConf>,
+) -> ShadowConfig {
+    let mut cfg = ShadowConfig::new(data_dir.to_path_buf(), filter_out_dir.to_path_buf());
+    cfg.socket_dir = socket_dir.to_path_buf();
+    cfg.port = ORACLE_PORT;
+    cfg.user = "postgres".into();
+    cfg.dbname = "postgres".into();
+    cfg.bridge = bridge;
+    cfg
+}
+
+fn run_pg_dump(conninfo: &str, password: Option<&str>) -> Result<String> {
+    let mut cmd = Command::new("pg_dump");
+    cmd.args([
+        "--binary-upgrade",
+        "--schema-only",
+        "--no-owner",
+        "--no-privileges",
+        "-d",
+        conninfo,
+    ]);
+    if let Some(pw) = password {
+        cmd.env("PGPASSWORD", pw);
+    }
+    let out = cmd.output().context("spawn pg_dump")?;
+    if !out.status.success() {
+        anyhow::bail!("pg_dump failed: {}", String::from_utf8_lossy(&out.stderr));
+    }
+    String::from_utf8(out.stdout).context("pg_dump output not utf8")
+}
+
+pub fn needs_bridge(catalog: &CatalogMap) -> bool {
+    catalog.descriptors().any(|d| {
+        d.attributes
+            .iter()
+            .any(|a| !a.dropped && attr_needs_bridge(a))
+    })
+}
+
+fn attr_needs_bridge(a: &RelAttr) -> bool {
+    !is_native_scalar(a.type_oid)
+        && !matches!(
+            a.type_name.as_str(),
+            "geography" | "geometry" | "vector" | "halfvec"
+        )
+}
+
+fn is_native_scalar(oid: u32) -> bool {
+    matches!(
+        oid,
+        BOOLOID
+            | BYTEAOID
+            | CHAROID
+            | NAMEOID
+            | INT8OID
+            | INT2OID
+            | INT4OID
+            | TEXTOID
+            | OIDOID
+            | JSONOID
+            | CIDROID
+            | FLOAT4OID
+            | FLOAT8OID
+            | INETOID
+            | BPCHAROID
+            | VARCHAROID
+            | DATEOID
+            | TIMEOID
+            | TIMESTAMPOID
+            | TIMESTAMPTZOID
+            | INTERVALOID
+            | TIMETZOID
+            | NUMERICOID
+            | UUIDOID
+    )
+}

src/backfill/copy_backfill.rs

@@ -72,6 +72,7 @@ use crate::decode::heap_decoder::ColumnValue;
 use crate::emit::ch_emitter::{EmitterConfig, EmitterStats};
 use crate::emit::pipeline::{Fatal, bootstrap, tail};
 use crate::mapping::MappingHandle;
+use crate::ops::oracle::Oracle;
 use crate::pg::{current_wal_lsn, quote_ident};
 use crate::pos::{Pos, Snapshot};
 use crate::runtime_config::InitialLoadMode;
@@ -412,6 +413,7 @@ pub struct CopyBackfiller {
     /// Pipeline's resident-payload pool: backup passes run concurrently
     /// with live streaming and draw from the same budget
     budget: Option<crate::budget::MemoryBudget>,
+    oracle: Option<Arc<Oracle>>,
     /// Fixed scratch paths require one cluster backup pass at a time
     backup_pass_lock: Mutex<()>,
     inner: Mutex<Inner>,
@@ -434,6 +436,7 @@ impl CopyBackfiller {
         spill_dir: &Path,
         config_rx: Option<watch::Receiver<Arc<ResolvedConfig>>>,
         budget: Option<crate::budget::MemoryBudget>,
+        oracle: Option<Arc<Oracle>>,
     ) -> Self {
         let ledger = Ledger::load(spill_dir).await;
         let emitter = Arc::new(emitter);
@@ -454,6 +457,7 @@ impl CopyBackfiller {
             spill_dir: spill_dir.to_path_buf(),
             config_rx,
             budget,
+            oracle,
             backup_pass_lock: Mutex::new(()),
             inner: Mutex::new(Inner {
                 ledger,
@@ -726,6 +730,7 @@ impl CopyBackfiller {
             scratch_dir: self.spill_dir.join("backup_backfill"),
             config_rx: self.config_rx.clone(),
             budget: self.budget.clone(),
+            oracle: self.oracle.clone(),
         };
         let outcome = crate::backfill::backup_backfill::run_pass(&ctx, mode, reqs).await?;
         self.publish_staged(&staging, reqs).await;
@@ -1071,6 +1076,7 @@ impl CopyBackfiller {
             ),
             self.emitter.row_policy(),
             self.config_rx.as_ref().map(|rx| rx.borrow().clone()),
+            None,
         ));
 
         let plan = column_plan(desc);

src/backfill/mod.rs

@@ -8,6 +8,7 @@ pub(super) mod backup_sink;
 pub mod backup_source;
 pub mod backup_source_direct;
 pub mod backup_source_object_store;
+pub mod bootstrap_oracle;
 pub mod copy_backfill;
 pub mod opt_in;
 pub mod pg_path;