From 53994525fa919897488829e81884ccbe56154951 Mon Sep 17 00:00:00 2001 From: Harshil Goel Date: Mon, 31 Aug 2026 17:06:54 +0530 Subject: [PATCH] Add pg oracle during bootstrap to decode complex types --- plans/bootstrap.md | 39 +++- plans/future/INDEX.md | 1 - plans/future/greenfield_oracle.md | 57 ------ plans/future/oracle_native_blocks.md | 4 +- plans/oracle.md | 6 +- src/backfill/backfill_types.rs | 2 + src/backfill/backup_backfill.rs | 1 + src/backfill/bootstrap_oracle.rs | 171 ++++++++++++++++ src/backfill/copy_backfill.rs | 6 + src/backfill/mod.rs | 1 + src/bin/stream.rs | 36 ++++ src/catalog/shadow.rs | 69 ++++--- src/emit/pipeline/bootstrap.rs | 24 ++- tests/bootstrap_pipeline_ch.rs | 1 + tests/bootstrap_types_e2e.rs | 296 +++++++++++++++++++++++++++ tests/common/inproc_harness.rs | 1 + 16 files changed, 619 insertions(+), 96 deletions(-) delete mode 100644 plans/future/greenfield_oracle.md create mode 100644 src/backfill/bootstrap_oracle.rs create mode 100644 tests/bootstrap_types_e2e.rs diff --git a/plans/bootstrap.md b/plans/bootstrap.md index bc821ba8..24411390 100644 --- a/plans/bootstrap.md +++ b/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>`; 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 diff --git a/plans/future/INDEX.md b/plans/future/INDEX.md index f5717ce0..f92917e0 100644 --- a/plans/future/INDEX.md +++ b/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 diff --git a/plans/future/greenfield_oracle.md b/plans/future/greenfield_oracle.md deleted file mode 100644 index afc3dc96..00000000 --- a/plans/future/greenfield_oracle.md +++ /dev/null @@ -1,57 +0,0 @@ -# greenfield_oracle — a bridge for tier-3 during bootstrap - -## Problem - -Greenfield bootstrap ([../bootstrap.md](../bootstrap.md)) page-walks a -base/object-store backup and drains rows through `pipeline::bootstrap::drain` -**before the shadow PG and its bridge exist** (bridge is created after -`run_bootstrap` returns). So bridge-routed tier-3 values — jsonb, arrays, -hstore, tsvector, ranges, domains — have no PG to render them and land empty. -In-tree types (geography, vector; see [../oracle.md](../oracle.md)) are fine. - -`BackupSource` is a page reader, not a running PG, so there is nothing to -decode on-disk tier-3 Datums during the drain. - -## Approach - -Stand up a **throwaway Postgres with the walshadow module + the source's -extensions**, point the bridge/oracle at it for the bootstrap drain, then tear -it down once bootstrap completes. `resolve_decoded_heap` already takes an -`Option<&Oracle>`; wire this oracle in for the greenfield drain and the whole -tier-3 path resolves exactly like live. - -Lifecycle: create → (extensions/module ready) → pass `Some(oracle)` to -`bootstrap::drain` → drain → drop the temp PG + its datadir/socket before the -real shadow is materialized for streaming. - -## The OID-matching constraint (the hard part) - -`ws_decode_datum_text` renders a Datum by running the type's `typoutput`, -looked up by **OID**. Built-in tier-3 OIDs are stable across clusters (jsonb -3802, `int4[]` 1007, …) so a fresh `initdb` + `CREATE EXTENSION` handles them. -But **extension type OIDs are assigned at `CREATE EXTENSION` time and differ -per cluster** — a fresh temp PG's `hstore`/`geography`/`vector` OID won't match -the source OID carried in the on-disk bytes, so typoutput lookup misfires. - -Two ways to satisfy it: - -- **Restore the temp PG from the backup** (it then carries the source catalog, - OIDs match) — essentially a short-lived shadow. Reuses the base-backup we - already fetched; heaviest but exact. Overlaps with the Option-A framing in - the earlier analysis. -- **Resolve by type name, not OID** — extend the bridge `DECODE` protocol to - carry the type name; the worker looks up `typoutput` via - `regtype`/`pg_type.typname` in the temp PG (which has the same-named - extensions installed). Lets a plain `initdb` + extensions work regardless of - OID drift. Smaller PG, but a protocol + worker change. - -## Open questions - -- Which extensions to install: derive from the source catalog (types actually - present) vs a fixed set; fail-soft when one isn't available. -- Cost/timing: temp-PG spin-up vs bootstrap duration; only worth it when tier-3 - columns exist in the mapped set. -- Interaction with restart/resume: bootstrap re-runs must recreate/tear down - the temp PG idempotently. -- Does not change the emitter contract; it only makes `Some(oracle)` available - earlier. Orthogonal to [oracle_native_blocks.md](oracle_native_blocks.md). diff --git a/plans/future/oracle_native_blocks.md b/plans/future/oracle_native_blocks.md index 8793598b..f41cdace 100644 --- a/plans/future/oracle_native_blocks.md +++ b/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 diff --git a/plans/oracle.md b/plans/oracle.md index 0168bd11..3b3b925e 100644 --- a/plans/oracle.md +++ b/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::::text`) require reconstructing wire format from diff --git a/src/backfill/backfill_types.rs b/src/backfill/backfill_types.rs index 7baf672b..dd7f2bde 100644 --- a/src/backfill/backfill_types.rs +++ b/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>>, pub budget: Option, + pub oracle: Option>, } #[derive(Debug, Default, Clone)] diff --git a/src/backfill/backup_backfill.rs b/src/backfill/backup_backfill.rs index bd1995b7..e957f9b2 100644 --- a/src/backfill/backup_backfill.rs +++ b/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 diff --git a/src/backfill/bootstrap_oracle.rs b/src/backfill/bootstrap_oracle.rs new file mode 100644 index 00000000..1fb4e6af --- /dev/null +++ b/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, + base_dir: PathBuf, +} + +impl BootstrapOracle { + pub async fn provision( + base_dir: PathBuf, + source_conninfo: String, + source_password: Option, + bridge_lib_dir: Option, + connect_budget: Duration, + ) -> Result { + 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 { + 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 { + 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, +) -> 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 { + 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 + ) +} diff --git a/src/backfill/copy_backfill.rs b/src/backfill/copy_backfill.rs index 5bd0fb17..6031394b 100644 --- a/src/backfill/copy_backfill.rs +++ b/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, + oracle: Option>, /// Fixed scratch paths require one cluster backup pass at a time backup_pass_lock: Mutex<()>, inner: Mutex, @@ -434,6 +436,7 @@ impl CopyBackfiller { spill_dir: &Path, config_rx: Option>>, budget: Option, + oracle: Option>, ) -> 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); diff --git a/src/backfill/mod.rs b/src/backfill/mod.rs index 4b79da5c..890dafca 100644 --- a/src/backfill/mod.rs +++ b/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; diff --git a/src/bin/stream.rs b/src/bin/stream.rs index 996f89d1..fd9b5b0f 100644 --- a/src/bin/stream.rs +++ b/src/bin/stream.rs @@ -47,6 +47,7 @@ use tokio_util::sync::CancellationToken; use walrus::pg::backup::{BACKUP_NAME_PREFIX, format_pg_lsn}; use walrus::pg::replication::base_backup::BaseBackupOpts; use walrus::pg::replication::conn::PgConfig; +use walrus::pg::replication::tls::SslMode; use walshadow::backfill_bootstrap::{ BootstrapConfig, BootstrapOutcome, drain_backfill, seed_in_snapshot, spawn_greenfield_bootstrap, }; @@ -1604,6 +1605,7 @@ async fn run_session( &args.spill_dir, Some(config_rx.clone()), Some(pipeline_budget.clone()), + oracle.clone(), ) .await, )); @@ -4318,6 +4320,39 @@ async fn run_bootstrap( "bootstrap insert tail started", ); + let source_conninfo = format!( + "host={} port={} user={} dbname={} sslmode={}", + src_cfg.host, + src_cfg.port, + src_cfg.user, + src_cfg.database, + if src_cfg.sslmode == SslMode::Disable { + "disable" + } else { + "prefer" + }, + ); + let bootstrap_oracle = + if walshadow::backfill::bootstrap_oracle::needs_bridge(&drain_catalog) { + Some( + walshadow::backfill::bootstrap_oracle::BootstrapOracle::provision( + args.spill_dir.join("bootstrap_oracle"), + source_conninfo, + src_cfg.password.clone(), + args.bridge_lib_dir.clone(), + Duration::from_secs(args.shadow_connect_timeout), + ) + .await + .context( + "bootstrap oracle: greenfield needs it to resolve tier-3 types; \ + refusing to load empty columns", + )?, + ) + } else { + None + }; + let oracle = bootstrap_oracle.as_ref().map(|o| o.oracle()); + let deferred_path = args.spill_dir.join("bootstrap_deferred.bin"); tokio::fs::remove_file(&deferred_path).await.ok(); let drain = tokio::spawn(bootstrap::drain( @@ -4337,6 +4372,7 @@ async fn run_bootstrap( // snapshot the CREATEs above rendered from: per-relation system // column names have to match what CH now holds Some(resolved), + oracle, )); let (drain_res, pump_res) = tokio::join!(drain, pump); let drain_outcome = drain_res diff --git a/src/catalog/shadow.rs b/src/catalog/shadow.rs index 6d0b1821..2d5825cf 100644 --- a/src/catalog/shadow.rs +++ b/src/catalog/shadow.rs @@ -405,34 +405,44 @@ impl Shadow { } pub fn start(&self) -> Result<()> { + self.start_with_opts(None) + } + + pub fn start_binary_upgrade(&self) -> Result<()> { + self.start_with_opts(Some("-b")) + } + + fn start_with_opts(&self, pg_opts: Option<&str>) -> Result<()> { let log = self.config.data_dir.join("startup.log"); - let res = self.run( - "pg_ctl", - [ - "-D", - self.config.data_str(), - "-l", - log.to_str().expect("non-utf8 log path"), - "-w", - "-t", - "86400", - "start", - ], - ); - match res { - Ok(_) => Ok(()), - // pg_ctl only reports "could not start server" - // Include log where Postgres reports required GUC value + let log_str = log.to_str().expect("non-utf8 log path"); + let mut args: Vec<&str> = vec![ + "-D", + self.config.data_str(), + "-l", + log_str, + "-w", + "-t", + "86400", + ]; + if let Some(o) = pg_opts { + args.push("-o"); + args.push(o); + } + args.push("start"); + let res = self.run("pg_ctl", args); + if let Err(ShadowError::Process { + cmd, + status, + stderr, + }) = res + { Err(ShadowError::Process { - cmd, - status, - stderr, - }) => Err(ShadowError::Process { cmd, status, stderr: format!("{stderr}\nstartup.log tail:\n{}", log_tail(&log)), - }), - Err(e) => Err(e), + }) + } else { + res.map(|_| ()) } } @@ -522,12 +532,15 @@ impl Shadow { .stdout(Stdio::piped()) .stderr(Stdio::piped()) .spawn()?; - child - .stdin - .as_mut() - .expect("piped") - .write_all(sql.as_bytes())?; + // Feed stdin from a thread while the parent drains stdout/stderr, or a + // dump large/noisy enough to fill psql's output pipe deadlocks the write. + let mut stdin = child.stdin.take().expect("piped"); + let payload = sql.as_bytes().to_vec(); + let writer = std::thread::spawn(move || { + let _ = stdin.write_all(&payload); + }); let out = child.wait_with_output()?; + let _ = writer.join(); self.check("psql -f -", out).map(|_| ()) } diff --git a/src/emit/pipeline/bootstrap.rs b/src/emit/pipeline/bootstrap.rs index 3288c35e..3c302e77 100644 --- a/src/emit/pipeline/bootstrap.rs +++ b/src/emit/pipeline/bootstrap.rs @@ -18,7 +18,8 @@ use crate::emit::pipeline::ack::AckHandle; use crate::emit::pipeline::batcher::{BatcherMsg, RoutedRow}; use crate::emit::route::{RouteSnapshot, RowPolicy}; use crate::mapping::{MappingHandle, TableMapping}; -use crate::schema::RelDescriptor; +use crate::ops::oracle::{Oracle, render_ext_columns, resolve_pending_tuple}; +use crate::schema::{RelAttr, RelDescriptor}; use crate::toast::{ CHUNK_PUT_BATCH, CHUNK_PUT_BYTES, FetchedValue, ToastResolver, ToastRow, check_value_caps, detoasted_value, finish_value, pointer_extsize, @@ -48,6 +49,7 @@ pub async fn drain( mut deferred: DeferredSpool, row_policy: RowPolicy, config: Option>, + oracle: Option>, ) -> Result { // Routes frozen once per pass from the caller's config snapshot let routes: HashMap<_, _> = mapping_handle @@ -128,11 +130,14 @@ pub async fn drain( } let mut tuple = tuple; let permit = resolve_or_fill_toast(&mut tuple, &rel, &route.mapping, &resolver).await?; + resolve_row(oracle.as_deref(), &rel.attributes, &mut tuple.columns).await; route_row(&msg_tx, seq, rel, route, tuple, permit).await?; bump(&mut open, &mut rows_routed); continue; } + let mut tuple = tuple; + resolve_row(oracle.as_deref(), &rel.attributes, &mut tuple.columns).await; route_row(&msg_tx, seq, rel, route, tuple, None).await?; bump(&mut open, &mut rows_routed); } @@ -174,6 +179,7 @@ pub async fn drain( continue; }; let permit = resolve_or_fill_toast(&mut tuple, &rel, &route.mapping, &resolver).await?; + resolve_row(oracle.as_deref(), &rel.attributes, &mut tuple.columns).await; route_row(&msg_tx, seq, rel, route, tuple, permit).await?; placed += 1; rows_routed += 1; @@ -202,6 +208,17 @@ fn bump(open: &mut Option<(walrus::pg::walparser::RelFileNode, u64, u64)>, rows_ *rows_routed += 1; } +async fn resolve_row( + oracle: Option<&Oracle>, + attrs: &[RelAttr], + columns: &mut [Option], +) { + render_ext_columns(attrs, columns); + if let Some(o) = oracle { + resolve_pending_tuple(o, columns).await; + } +} + async fn route_row( msg_tx: &mpsc::Sender, seq: u64, @@ -568,6 +585,7 @@ mod tests { mem_spool(), Default::default(), None, + None, )); let mut by_seq: HashMap = HashMap::new(); @@ -620,6 +638,7 @@ mod tests { mem_spool(), Default::default(), None, + None, )); let mut seqs: Vec = Vec::new(); @@ -670,6 +689,7 @@ mod tests { mem_spool(), Default::default(), None, + None, )); let mut rows = Vec::new(); @@ -735,6 +755,7 @@ mod tests { DeferredSpool::new(spool_tmp.path().join("bootstrap_deferred.bin"), 0), Default::default(), None, + None, )); let mut rows = Vec::new(); @@ -793,6 +814,7 @@ mod tests { mem_spool(), Default::default(), None, + None, )); // Wait for the referrer to defer, then unmap before walk EOF let deadline = std::time::Instant::now() + std::time::Duration::from_secs(10); diff --git a/tests/bootstrap_pipeline_ch.rs b/tests/bootstrap_pipeline_ch.rs index e77741d7..87d4cc5a 100644 --- a/tests/bootstrap_pipeline_ch.rs +++ b/tests/bootstrap_pipeline_ch.rs @@ -183,6 +183,7 @@ async fn bootstrap_tail_fans_out_n2() { ), Default::default(), None, + None, )); let outcome = drain.await.expect("drain join").expect("drain ok"); assert_eq!(outcome.next_seq, 2, "one seq per rfn"); diff --git a/tests/bootstrap_types_e2e.rs b/tests/bootstrap_types_e2e.rs new file mode 100644 index 00000000..91f1cd16 --- /dev/null +++ b/tests/bootstrap_types_e2e.rs @@ -0,0 +1,296 @@ +//! Data-type coverage for greenfield Direct bootstrap → ClickHouse. +//! +//! One source row spanning every mapped type — scalars, numeric, uuid, +//! temporal, json/jsonb, and the extension / oracle-resolved types +//! (`hstore`, `citext`, enum, arrays, `pgvector`, and `postgis` when +//! available) — bootstraps via `--bootstrap-mode=direct` into an +//! auto-created CH table, then every tier-3 column is asserted populated. +//! +//! Tier-3 columns land empty unless the bootstrap oracle (an OID-pinned +//! throwaway PG built from the source schema via `pg_dump --binary-upgrade`) +//! resolves them during the drain, so this is the end-to-end proof of that +//! path. `hstore`/`citext`/enum/arrays go through the bridge worker; +//! `pgvector`/`postgis` render in-tree. +//! +//! Skipped when `initdb` / `pg_basebackup` / `clickhouse` or the `hstore` / +//! `citext` / `vector` extensions are absent (`postgis` columns are added +//! only when installed). Needs `walshadow.so` on the PG library path +//! (`make -C pgext install`), same as the other bootstrap-CH drills. + +#![cfg(target_os = "linux")] + +#[path = "common/bootstrap_ch_fixture.rs"] +mod fx; + +use std::fs; +use std::net::SocketAddr; +use std::os::unix::process::CommandExt; +use std::path::Path; +use std::process::{Command, Stdio}; +use std::time::Duration; + +use anyhow::{Context, Result}; +use walshadow::shadow::{Shadow, ShadowConfig}; + +fn extension_available(name: &str) -> bool { + let Ok(out) = Command::new("pg_config").arg("--sharedir").output() else { + return false; + }; + if !out.status.success() { + return false; + } + let dir = String::from_utf8_lossy(&out.stdout).trim().to_string(); + Path::new(&dir) + .join(format!("extension/{name}.control")) + .exists() +} + +fn make_source(tmp: &tempfile::TempDir) -> Shadow { + let mut cfg = ShadowConfig::new( + tmp.path().join("source-data"), + tmp.path().join("source-filtered"), + ); + cfg.port = fx::PG_SOURCE_PORT; + cfg.socket_dir = tmp.path().join("source-sock"); + cfg.ctl_timeout = Duration::from_secs(60); + fs::create_dir_all(&cfg.filter_out_dir).unwrap(); + fs::create_dir_all(&cfg.socket_dir).unwrap(); + Shadow::new(cfg) +} + +fn load_types_workload(source: &Shadow, has_postgis: bool) -> Result<()> { + let mut cols = String::from( + "id int PRIMARY KEY, c_bool bool, c_int2 smallint, c_int4 int, c_int8 bigint, \ + c_float4 real, c_float8 double precision, c_num numeric(10,2), c_num_u numeric, \ + c_uuid uuid, c_text text, c_varchar varchar(20), c_date date, c_ts timestamp, \ + c_inet inet, c_json json, c_jsonb jsonb, c_hstore hstore, c_citext citext, \ + c_enum mood, c_int_arr int4[], c_text_arr text[], c_int8_arr bigint[], c_vector vector(3)", + ); + let mut names = String::from( + "id, c_bool, c_int2, c_int4, c_int8, c_float4, c_float8, c_num, c_num_u, c_uuid, \ + c_text, c_varchar, c_date, c_ts, c_inet, c_json, c_jsonb, c_hstore, c_citext, \ + c_enum, c_int_arr, c_text_arr, c_int8_arr, c_vector", + ); + let mut vals = String::from( + "1, true, 32000, 123456, 9000000000, 1.5, 2.5, 1234.56, 3.14159, \ + '11111111-1111-1111-1111-111111111111', 'hello', 'vc', '2024-01-15', \ + '2024-01-15 13:45:30', '192.168.1.1', '{\"a\": 1, \"b\": [2,3]}', \ + '{\"k\": 42, \"arr\": [1,2,3]}', 'x=>1, y=>2', 'CaseInsensitive', 'happy', \ + '{1,2,3}', '{a,b,c}', '{100,200}', '[0.1,0.2,0.3]'", + ); + if has_postgis { + cols.push_str(", c_geog geography(Point,4326), c_geom geometry(Point,4326)"); + names.push_str(", c_geog, c_geom"); + vals.push_str(", 'SRID=4326;POINT(30.5 50.25)', 'SRID=4326;POINT(1 2)'"); + } + + let mut sql = String::from( + "CREATE EXTENSION IF NOT EXISTS hstore;\n\ + CREATE EXTENSION IF NOT EXISTS citext;\n\ + CREATE EXTENSION IF NOT EXISTS vector;\n", + ); + if has_postgis { + sql.push_str("CREATE EXTENSION IF NOT EXISTS postgis;\n"); + } + sql.push_str("CREATE TYPE mood AS ENUM ('sad','ok','happy');\n"); + sql.push_str(&format!("CREATE TABLE public.all_types ({cols});\n")); + sql.push_str("ALTER TABLE public.all_types REPLICA IDENTITY FULL;\n"); + sql.push_str(&format!( + "INSERT INTO public.all_types ({names}) VALUES ({vals});\n" + )); + sql.push_str("CHECKPOINT;\nSELECT pg_switch_wal();\n"); + source + .apply_schema_dump(&sql) + .context("apply source schema") +} + +fn write_autocreate_config(path: &Path, ch_port: u16) -> Result<()> { + let body = format!( + "[ch]\n\ + host = \"127.0.0.1\"\n\ + port = {ch_port}\n\ + database = \"default\"\n\ + compression = \"lz4\"\n\ + \n\ + [table.\"public\".\"all_types\"]\n\ + replicate = true\n\ + initial_load = \"none\"\n" + ); + fs::write(path, body).context("write ch-config") +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn direct_bootstrap_all_types_end_to_end() { + if !fx::pg_available() || !fx::pg_basebackup_available() || !fx::clickhouse_available() { + eprintln!("skip: missing initdb / pg_basebackup / clickhouse"); + return; + } + for ext in ["hstore", "citext", "vector"] { + if !extension_available(ext) { + eprintln!("skip: extension {ext} not installed"); + return; + } + } + let has_postgis = extension_available("postgis"); + + let slot = fx::Ports::alloc(); + let tmp = tempfile::tempdir().unwrap(); + + let source = make_source(&tmp); + source.initdb().expect("initdb source"); + source.write_base_conf().expect("source base conf"); + fx::append_source_conf(&source).expect("append source conf"); + source.start().expect("start source"); + let _src_stop = fx::StopOnDrop { sh: &source }; + load_types_workload(&source, has_postgis).expect("load types workload"); + + let ch_tmp = tempfile::tempdir().unwrap(); + let ch = fx::ChServer::spawn(ch_tmp, slot.ch_tcp, slot.ch_http).expect("spawn ch"); + + let ch_config_path = tmp.path().join("ch-config.toml"); + write_autocreate_config(&ch_config_path, slot.ch_tcp).expect("write ch-config"); + + let bootstrap_shadow_data_dir = tmp.path().join("shadow-data"); + let shadow_sock = tmp.path().join("shadow-sock"); + fs::create_dir_all(&shadow_sock).unwrap(); + let shadow_filter_dir = tmp.path().join("filtered"); + fs::create_dir_all(&shadow_filter_dir).unwrap(); + let spill_dir = tmp.path().join("spill"); + fs::create_dir_all(&spill_dir).unwrap(); + + let bin = env!("CARGO_BIN_EXE_walshadow-stream"); + let stderr_path = tmp.path().join("daemon.stderr.log"); + let stderr_file = fs::File::create(&stderr_path).expect("open daemon stderr log"); + let metrics_addr: SocketAddr = format!("127.0.0.1:{}", slot.metrics).parse().unwrap(); + let child = Command::new(bin) + .args([ + "--host", + source.config().socket_dir.to_str().unwrap(), + "--port", + &fx::PG_SOURCE_PORT.to_string(), + "--user", + "postgres", + "--dbname", + "postgres", + "--sslmode", + "disable", + "--out-dir", + shadow_filter_dir.to_str().unwrap(), + "--shadow-socket-dir", + shadow_sock.to_str().unwrap(), + "--shadow-port", + &fx::PG_SHADOW_PORT.to_string(), + "--shadow-user", + "postgres", + "--shadow-dbname", + "postgres", + "--bridge-lib-dir", + fx::pgext_dir().to_str().unwrap(), + "--spill-dir", + spill_dir.to_str().unwrap(), + "--status-interval", + "1", + "--metrics-bind", + &metrics_addr.to_string(), + "--walsender-bind", + &format!("127.0.0.1:{}", slot.walsender), + "--retention-bytes", + "0", + "--ch-config", + ch_config_path.to_str().unwrap(), + "--bootstrap-mode", + "direct", + "--bootstrap-shadow-data-dir", + bootstrap_shadow_data_dir.to_str().unwrap(), + "--bootstrap-shadow-replay-timeout", + "120", + ]) + .env("RUST_LOG", "warn,walshadow=info") + .stdout(Stdio::null()) + .stderr(Stdio::from(stderr_file)) + .process_group(0) + .spawn() + .expect("spawn walshadow-stream"); + let guard = fx::ChildGuard::new(child); + + let result = (|| -> Result<()> { + fx::wait_for_listen(metrics_addr, Duration::from_secs(30)) + .context("daemon metrics endpoint never came up")?; + + let deadline = std::time::Instant::now() + Duration::from_secs(90); + loop { + let n = ch + .query("SELECT count() FROM default.all_types FINAL WHERE _is_deleted = 0") + .unwrap_or_default(); + if n == "1" { + break; + } + let stderr = fs::read_to_string(&stderr_path).unwrap_or_default(); + if stderr.contains("oracle unavailable") { + anyhow::bail!("bootstrap oracle degraded (tier-3 would be empty)"); + } + if std::time::Instant::now() >= deadline { + anyhow::bail!("bootstrap row never reached CH (got {n:?})"); + } + std::thread::sleep(Duration::from_millis(250)); + } + + let mut checks: Vec<(&str, String)> = vec![ + ("c_bool", "true".into()), + ("c_int4", "123456".into()), + ("c_int8", "9000000000".into()), + ("c_num", "1234.56".into()), + ("c_num_u", "3.14159".into()), + ("c_text", "hello".into()), + ("c_uuid", "11111111-1111-1111-1111-111111111111".into()), + ("c_jsonb.k", "42".into()), + ("c_jsonb.arr[1]", "1".into()), + ("c_hstore['x']", "1".into()), + ("c_citext", "CaseInsensitive".into()), + ("c_enum", "happy".into()), + ("arrayStringConcat(c_int_arr, ',')", "1,2,3".into()), + ("arrayStringConcat(c_text_arr, ',')", "a,b,c".into()), + ("arrayStringConcat(c_int8_arr, ',')", "100,200".into()), + ("length(c_vector)", "3".into()), + ("round(c_vector[1], 2)", "0.1".into()), + ]; + if has_postgis { + checks.push(("c_geog", "POINT(30.5 50.25)".into())); + checks.push(("c_geom", "POINT(1 2)".into())); + } + + let mut fails = Vec::new(); + for (expr, want) in &checks { + let got = ch + .query(&format!( + "SELECT toString({expr}) FROM default.all_types FINAL WHERE id = 1" + )) + .unwrap_or_default(); + if &got != want { + fails.push(format!("{expr}: got {got:?}, want {want:?}")); + } + } + if !fails.is_empty() { + anyhow::bail!("column mismatches:\n {}", fails.join("\n ")); + } + Ok(()) + })(); + + let _ = guard.into_inner().map(|mut c| { + let _ = c.kill(); + let _ = c.wait(); + }); + if bootstrap_shadow_data_dir.join("postmaster.pid").exists() { + let mut shadow_cfg = + ShadowConfig::new(bootstrap_shadow_data_dir.clone(), shadow_filter_dir.clone()); + shadow_cfg.port = fx::PG_SHADOW_PORT; + shadow_cfg.socket_dir = shadow_sock.clone(); + shadow_cfg.ctl_timeout = Duration::from_secs(60); + let _ = Shadow::new(shadow_cfg).stop(); + } + + if let Err(e) = result { + let stderr = fs::read_to_string(&stderr_path).unwrap_or_default(); + panic!("{e:#}\n--- daemon stderr ---\n{stderr}"); + } +} diff --git a/tests/common/inproc_harness.rs b/tests/common/inproc_harness.rs index ef4c98f2..b1aace8a 100644 --- a/tests/common/inproc_harness.rs +++ b/tests/common/inproc_harness.rs @@ -888,6 +888,7 @@ async fn build_pipeline_inner( &spill_dir, Some(config_rx.clone()), None, + oracle.clone(), ) .await, ))