diff --git a/README.md b/README.md index 7dad4888..62504ac0 100644 --- a/README.md +++ b/README.md @@ -230,6 +230,33 @@ one CH name), and the whole array pins nothing — scope still comes from `replicate` / `auto_create`. `config_column` rows take `match` the same way, over the whole `(namespace, relname, attname)` key. +Destination shape — walshadow appends `_lsn`, `_xid`, `_commit_ts`, +`_is_deleted` to every table and keys on replica identity. Rename or drop the +appended columns, and pin the sort key: + +```toml +[system_columns] # cluster-wide default +lsn = "_peerdb_version" +commit_ts = "_peerdb_synced_at" +is_deleted = "_peerdb_is_deleted" # false drops the column, and DELETE rows + +[table.public.events] +order_by = ["tenant_id", "id"] # else replica identity +primary_key = ["tenant_id"] # CH index prefix, must prefix order_by +lsn = "_version" # same four keys, this relation only + +[table.app."events_*"] +match = "glob" +is_deleted = false +order_by = ["tenant_id", "id"] +``` + +`[system_columns]` is boot-only and cluster-wide; a `[table.*]` block or a +`config_table` row renames per relation, and takes `match` like any other rule +— walshadow never renames or rekeys a table CH already holds, so shape has to +land before the first CREATE. See +[destination tables](docs/destination-tables.md) + ## Building from source diff --git a/docs/configuration.md b/docs/configuration.md index e997b048..2cd05cbb 100644 --- a/docs/configuration.md +++ b/docs/configuration.md @@ -61,8 +61,9 @@ including tables created later. `init` writes `[source]`, `[ch]`, and chosen Pass file with `--ch-config`. Loader also merges sibling directory formed by replacing `.toml` with `.d`, for example `ch-config.d/*.toml` -Unknown keys, invalid values, and incompatible mapping fields fail validation -instead of falling back silently +Invalid values and incompatible mapping fields fail validation instead of +falling back silently. Loader ignores unknown keys, so check spelling here when +a setting has no effect ## Precedence @@ -116,6 +117,8 @@ restored Apply live: - table and column rules +- per-table metadata column names, `order_by`, and `primary_key`, applied when + walshadow creates a table, see [Query destination data](destination-tables.md) - namespace destinations and drop policy - pause state - batch sizes, flush timeout, compression, and retry count @@ -125,6 +128,7 @@ Require restart: - `replicate_all` - runtime-config schema +- cluster-wide `[system_columns]` names - soft-delete and TOAST modes - worker-pool sizes and memory limits - backup and shadow bootstrap choices @@ -161,6 +165,10 @@ VALUES ('public', 'orders', true, 'copy'); ``` +`config_table` also carries destination shape: `order_by` and `primary_key` as +`text[]`, and `lsn`, `xid`, `commit_ts`, `is_deleted` for metadata column names. +See [Query destination data](destination-tables.md) + walshadow reads these tables but never writes them. Keep archive credentials and bootstrap configuration in TOML, not source-side tables diff --git a/docs/destination-tables.md b/docs/destination-tables.md index 894cedae..37cf1a02 100644 --- a/docs/destination-tables.md +++ b/docs/destination-tables.md @@ -21,6 +21,83 @@ Source columns are followed by four metadata columns: | `_commit_ts` | `DateTime64(6, 'UTC')` | source commit time | | `_is_deleted` | `Bool` | delete marker | +Rename these columns, drop the delete marker, or pin the sort key with +settings below + +## Choose sort key + +Set `order_by` to sort a destination table on chosen columns instead of source +row key. Name ClickHouse column names, after any rename: + +```toml +[table.public.events] +order_by = ["tenant_id", "id"] +primary_key = ["tenant_id"] + +[table.app."events_*"] +match = "glob" +order_by = ["tenant_id", "id"] +``` + +ClickHouse `PRIMARY KEY` chooses which sort-key prefix its sparse index covers +and enforces no uniqueness. `primary_key` must be a prefix of `order_by`. +walshadow ignores an invalid `primary_key`, logs a warning, and indexes whole +sort key. It also ignores an `order_by` naming a missing or `Nullable` column, +because ClickHouse rejects nullable sort keys, and falls back to source row key + +Both settings apply when walshadow creates a table. walshadow never rekeys a +table ClickHouse already holds, so choose shape before first delivery, or run +`ALTER TABLE` in ClickHouse. With `replicate_all = true` a table can reach +ClickHouse before an exact source-side row arrives: keep custom shape in config +file, or in a pattern rule which matches before creation + +Source-side rows carry same settings as `text[]`: + +```sql +UPDATE walshadow.config_table +SET order_by = ARRAY['tenant_id', 'id'], primary_key = ARRAY['tenant_id'] +WHERE namespace = 'public' AND relname = 'events'; +``` + +## Rename metadata columns + +`[system_columns]` renames appended columns for every table. walshadow reads it +at startup only: + +```toml +[system_columns] +lsn = "_peerdb_version" +commit_ts = "_peerdb_synced_at" +is_deleted = "_peerdb_is_deleted" +``` + +Set same keys in a `[table.*]` block, or in a `config_table` row, to rename for +matching relations. Omitted keys inherit cluster-wide names. Names must be +unique and non-empty: config file fails validation, and a source-side row is +rejected with a warning, leaving cluster-wide names in place. TOAST mirror +tables keep fixed names + +walshadow uses configured names in `CREATE TABLE` and `INSERT` statements, and +never renames a column in an existing ClickHouse table. Renaming for an existing +destination also needs `ALTER TABLE ... RENAME COLUMN` in ClickHouse + +## Drop the delete marker + +Set `is_deleted = false` for an append-only destination. This drops the marker +column and discards source `DELETE` rows, counting them in +`walshadow_emitter_deletes_discarded_total`: + +```toml +[system_columns] +is_deleted = false # cluster-wide + +[table.app."events_*"] +match = "glob" +is_deleted = false # this pattern only +``` + +Source-side rows use an empty string, `is_deleted = ''`, for same result + ## Read current state Use `FINAL` when query must resolve outstanding row versions immediately diff --git a/plans/GLOSSARY.md b/plans/GLOSSARY.md index 68427b17..c576b85b 100644 --- a/plans/GLOSSARY.md +++ b/plans/GLOSSARY.md @@ -561,9 +561,13 @@ diverging `shadow_apply` vs `dispatched` signals shadow lag `XLOG_XACT_ASSIGNMENT`; a hint only, authoritative subxact list arrives inline on commit/abort record ([xact.md](xact.md)) -**synthetic columns** — four trailing columns on every dest table: -`_lsn` UInt64, `_xid` UInt32, `_commit_ts` DateTime64(6,'UTC'), -`_is_deleted` Bool ([emitter.md](emitter.md)) +**synthetic columns** — trailing columns on every dest table, default +names `_lsn` UInt64, `_xid` UInt32, `_commit_ts` DateTime64(6,'UTC'), +`_is_deleted` Bool. `[system_columns]` renames them cluster-wide and can +drop the delete marker; a `[table.*]` block or `config_table` row (literal +or `match` pattern) overrides per relation +([emitter.md](emitter.md), +[destination tables guide](../docs/destination-tables.md)) **tail** — reusable batcher + inserter pool + ack collector unit; WAL pipeline and bootstrap drain feed the identical tail, `tail.finish` diff --git a/plans/emitter.md b/plans/emitter.md index 4704466c..73fb785f 100644 --- a/plans/emitter.md +++ b/plans/emitter.md @@ -386,8 +386,17 @@ CH applies its own zero-init ### Synthetic columns Destination metadata contract lives in -[`docs/destination-tables.md`](../docs/destination-tables.md). All four values -remain non-nullable and append after mapped columns in `TableEncoder::new` +[`docs/destination-tables.md`](../docs/destination-tables.md). Values stay +non-nullable and append after mapped columns in `TableEncoder::new`. Names come +from the relation's resolved `SystemColumns`, so every site that renders or +encodes a metadata column reads them per relation instead of a constant + +The delete marker is optional (`is_deleted = false`). Without it a DELETE would +land as a phantom insert of the old image, so those rows are discarded where the +placed count is taken (`decode_and_route`, and the object-store gap-replay +sink), counted in `walshadow_emitter_deletes_discarded_total`. Dropping them +anywhere later would short the ack collector's per-seq reconcile and pin the +watermark `_lsn` is dedup key because emitter ack lags actual CH durability by up to one flush window. On restart the manifest floor rewinds to @@ -449,8 +458,10 @@ table: | `Dropped { rel_name }` | gated on the namespace's `DropTableStrategy` (`drop_strategy_for`, else global): `Retain` (default) skips silently, `Warn` skips at WARN, `Drop` runs `DROP TABLE IF EXISTS` | `render_create_table` builds CREATE off descriptor: attributes through -`type_bridge::map`, PK columns first in `ORDER BY` (else `_lsn` -fallback), engine pinned to `ReplacingMergeTree(_lsn)`. Synthetic +`type_bridge::map`, then the sort key — the operator `order_by` +([`docs/destination-tables.md`](../docs/destination-tables.md)) when it names +non-nullable destination columns, else PK columns first (else `_lsn` fallback) — +engine pinned to `ReplacingMergeTree(_lsn)`. Synthetic columns appended after mapped columns, same shape as `TablePlan::build`. `render_create_table_from_mapping` builds off the mapping instead (its columns are the emitter's INSERT contract), resolving `ORDER BY` key diff --git a/sql/runtime_config_install.sql b/sql/runtime_config_install.sql index 4706e8fe..8696fcc9 100644 --- a/sql/runtime_config_install.sql +++ b/sql/runtime_config_install.sql @@ -52,6 +52,14 @@ CREATE TABLE IF NOT EXISTS :"walshadow_schema".config_table ( initial_load text, -- one-time backfill mode for pre-opt-in -- rows: 'none' | 'copy' | 'base_backup' -- | 'object_store'; NULL means omitted + order_by text[], -- ClickHouse ORDER BY columns; NULL + -- inherits, empty array derives + primary_key text[], -- ClickHouse PRIMARY KEY; must prefix + -- order_by + lsn text, -- per-relation names for the columns + xid text, -- walshadow appends; NULL inherits + commit_ts text, -- [system_columns] + is_deleted text, -- '' drops the marker (and DELETE rows) PRIMARY KEY (namespace, relname) ); @@ -72,7 +80,13 @@ ALTER TABLE :"walshadow_schema".config_table ADD COLUMN IF NOT EXISTS replicate ALTER TABLE :"walshadow_schema".config_table ADD COLUMN IF NOT EXISTS initial_load text; ALTER TABLE :"walshadow_schema".config_table ADD COLUMN IF NOT EXISTS target_database text; ALTER TABLE :"walshadow_schema".config_table ADD COLUMN IF NOT EXISTS target_table text; +ALTER TABLE :"walshadow_schema".config_table ADD COLUMN IF NOT EXISTS order_by text[]; +ALTER TABLE :"walshadow_schema".config_table ADD COLUMN IF NOT EXISTS primary_key text[]; ALTER TABLE :"walshadow_schema".config_table ADD COLUMN IF NOT EXISTS match text; +ALTER TABLE :"walshadow_schema".config_table ADD COLUMN IF NOT EXISTS lsn text; +ALTER TABLE :"walshadow_schema".config_table ADD COLUMN IF NOT EXISTS xid text; +ALTER TABLE :"walshadow_schema".config_table ADD COLUMN IF NOT EXISTS commit_ts text; +ALTER TABLE :"walshadow_schema".config_table ADD COLUMN IF NOT EXISTS is_deleted text; ALTER TABLE :"walshadow_schema".config_column ADD COLUMN IF NOT EXISTS match text; -- REPLICA IDENTITY FULL logs the complete old-row image on UPDATE/DELETE, so a diff --git a/src/backfill/backfill_staging.rs b/src/backfill/backfill_staging.rs index 3633da80..87b62c5c 100644 --- a/src/backfill/backfill_staging.rs +++ b/src/backfill/backfill_staging.rs @@ -127,6 +127,9 @@ pub struct StagingSession { client: BoxedAsyncClient, /// Kept whole for reconnect; shared with the pass that opened the session conn: Arc, + /// Per-relation destination rules, for the promote's `_lsn` predicate when + /// a `[table.*]` block or `config_table` row renamed that column + rules: Option>, retry: RetryConfig, timeout: Duration, } @@ -141,9 +144,27 @@ impl StagingSession { retry: emitter.retry.clone(), timeout: emitter.insert_timeout, conn: emitter, + rules: None, }) } + pub fn with_rules(mut self, rules: Option>) -> Self { + self.rules = rules; + self + } + + /// LSN column of one relation's destination + fn lsn_column(&self, rel: &RelName) -> String { + match &self.rules { + Some(rules) => rules + .settings(rel) + .system_columns(&self.conn.system_columns) + .lsn + .clone(), + None => self.conn.system_columns.lsn.clone(), + } + } + async fn attempt_write(&mut self, sql: &str) -> Result<(), EmitterError> { exec_drain(&mut self.client, sql, self.timeout).await } @@ -295,8 +316,9 @@ impl StagingSession { ); } let list = cols.join(", "); + let lsn = quote_ident(&self.lsn_column(&rel.rel)); self.exec_retry(&format!( - "INSERT INTO {} ({list}) SELECT {list} FROM {} WHERE `_lsn` > {}", + "INSERT INTO {} ({list}) SELECT {list} FROM {} WHERE {lsn} > {}", rel.real_sql(), rel.staging_sql(), rel.s_lsn diff --git a/src/backfill/backup_backfill.rs b/src/backfill/backup_backfill.rs index b85a70af..bd1995b7 100644 --- a/src/backfill/backup_backfill.rs +++ b/src/backfill/backup_backfill.rs @@ -350,7 +350,7 @@ async fn walk_and_ship( ctx.stats.clone(), resolver.clone(), DeferredSpool::new(toast_spool_path, DEFERRED_SPOOL_MEM_MAX), - ctx.emitter.soft_delete, + ctx.emitter.row_policy(), ctx.config_rx.as_ref().map(|rx| rx.borrow().clone()), )); @@ -901,7 +901,7 @@ async fn replay_gap( mapping: ctx.mapping.snapshot().await, stats: ctx.stats.clone(), budget: ctx.budget.clone(), - soft_delete: ctx.emitter.soft_delete, + row_policy: ctx.emitter.row_policy(), config: ctx.config_rx.as_ref().map(|rx| rx.borrow().clone()), batch_rows: ctx.emitter.drain_batch_rows, batch_bytes: ctx.emitter.drain_batch_bytes, @@ -948,7 +948,7 @@ struct ReplaySink { stats: Arc, budget: Option, /// Boot-only delete-retention policy, frozen into route snapshots - soft_delete: bool, + row_policy: crate::emit::route::RowPolicy, /// Config snapshot for route freezes: gap replay re-seeds from current /// config, not history (route history has no WAL position) config: Option>, @@ -1069,6 +1069,19 @@ impl ReplaySink { self.commits_past_s += 1; continue; } + let policy = self + .row_policy + .for_rel(self.config.as_deref(), &rel.rel_name); + // Append-only destination (no delete marker): see + // `decode_and_route` + if policy.system.is_deleted.is_none() + && matches!(heap.decoded.op, crate::decode::heap_decoder::HeapOp::Delete) + { + self.stats + .deletes_discarded + .fetch_add(1, std::sync::atomic::Ordering::Relaxed); + continue; + } let rel = rel.clone(); let value_permit = detoast_heap(&mut heap, spool, &ref_maps, &self.resolver) .await @@ -1084,8 +1097,7 @@ impl ReplaySink { .config .as_ref() .map_or_else(Arc::default, |rc| rc.column_rules.clone()); - let route = - crate::emit::route::RouteSnapshot::freeze(mapping, rules, self.soft_delete); + let route = crate::emit::route::RouteSnapshot::freeze(mapping, rules, policy); let seq = if let Some((seq, rows)) = &mut self.open { *rows += 1; *seq diff --git a/src/backfill/copy_backfill.rs b/src/backfill/copy_backfill.rs index b0a1769a..5bd0fb17 100644 --- a/src/backfill/copy_backfill.rs +++ b/src/backfill/copy_backfill.rs @@ -506,6 +506,12 @@ impl CopyBackfiller { dest.clone() } + /// Live per-relation rules, for destination names the staging promote + /// has to match. `None` without a resolver (tests): the boot set stands + fn table_rules(&self) -> Option> { + self.config_rx.as_ref().map(|rx| rx.borrow().rules.clone()) + } + fn refresh_gauges(&self, ledger: &Ledger) { self.pending .store(ledger.pending_count(), Ordering::Relaxed); @@ -742,7 +748,10 @@ impl CopyBackfiller { if plan.rels.is_empty() { return; } - let mut sess = match StagingSession::connect(self.dest_emitter()).await { + let mut sess = match StagingSession::connect(self.dest_emitter()) + .await + .map(|s| s.with_rules(self.table_rules())) + { Ok(s) => s, Err(e) => { tracing::error!( @@ -870,7 +879,9 @@ impl CopyBackfiller { table: target.table, s_lsn: rec.s_lsn.get(), }; - let mut sess = StagingSession::connect(self.dest_emitter()).await?; + let mut sess = StagingSession::connect(self.dest_emitter()) + .await? + .with_rules(self.table_rules()); match sess.table_uuid(&rel.database, &rel.staging_table()).await? { None => { self.mark_done_entry(name).await; @@ -1058,7 +1069,7 @@ impl CopyBackfiller { self.spill_dir.join("copy_deferred.bin"), crate::backfill::spool::DEFERRED_SPOOL_MEM_MAX, ), - self.emitter.soft_delete, + self.emitter.row_policy(), self.config_rx.as_ref().map(|rx| rx.borrow().clone()), )); diff --git a/src/bin/stream.rs b/src/bin/stream.rs index 66827ec1..996f89d1 100644 --- a/src/bin/stream.rs +++ b/src/bin/stream.rs @@ -1568,6 +1568,7 @@ async fn run_session( &config_rx.borrow(), emitter_cfg.database.clone(), emitter_cfg.soft_delete, + emitter_cfg.system_columns.clone(), emitter_cfg.replicate_all, emitter_cfg.runtime_config_schema.clone(), ); @@ -2897,6 +2898,14 @@ async fn seed_runtime_config( target_table: row.try_get("target_table").ok().flatten(), replicate: row.try_get("replicate").ok().flatten(), initial_load: row.try_get("initial_load").ok().flatten(), + order_by: row.try_get("order_by").ok().flatten(), + primary_key: row.try_get("primary_key").ok().flatten(), + system: walshadow::mapping::SystemColumnNames { + lsn: row.try_get("lsn").ok().flatten(), + xid: row.try_get("xid").ok().flatten(), + commit_ts: row.try_get("commit_ts").ok().flatten(), + is_deleted: row.try_get("is_deleted").ok().flatten(), + }, match_kind: row.try_get("match").ok().flatten(), }, ); @@ -3479,6 +3488,9 @@ async fn populate_metrics( emitter_unsupported_relations: emitter_stats .map(|s| s.unsupported_relations.load(Ordering::Relaxed)) .unwrap_or(0), + emitter_deletes_discarded: emitter_stats + .map(|s| s.deletes_discarded.load(Ordering::Relaxed)) + .unwrap_or(0), oracle_resolved_total: oracle_stats .map(|s| s.resolved.load(Ordering::Relaxed)) .unwrap_or(0), @@ -4262,10 +4274,10 @@ async fn run_bootstrap( let ch_target = match ch_config { Some(emitter_cfg) => { - let mapping = bootstrap_build_mapping(&emitter_cfg, &drain_catalog, args) + let (mapping, resolved) = bootstrap_build_mapping(&emitter_cfg, &drain_catalog, args) .await .context("bootstrap: build mapping")?; - Some((emitter_cfg, mapping)) + Some((emitter_cfg, mapping, resolved)) } None => None, }; @@ -4277,7 +4289,7 @@ async fn run_bootstrap( let cfg = BootstrapConfig::new(shadow_data_dir.clone()); let (rx, pump) = spawn_greenfield_bootstrap(cfg, source, catalog_map, store_toast); - let (shipped, outcome) = if let Some((emitter_cfg, mapping)) = ch_target { + let (shipped, outcome) = if let Some((emitter_cfg, mapping, resolved)) = ch_target { // Route bootstrap rows through the shared insert tail. Bootstrap // is the easy case: every row op=Insert at _lsn = start_lsn, no // aborts / TRUNCATE / DDL. Keep operator's flush_timeout; tail @@ -4320,9 +4332,11 @@ async fn run_bootstrap( deferred_path, walshadow::spool::DEFERRED_SPOOL_MEM_MAX, ), - emitter_cfg.soft_delete, - // Static mapping above: no overlay during greenfield bootstrap - None, + emitter_cfg.row_policy(), + // No source-PG overlay during greenfield bootstrap, but the same + // snapshot the CREATEs above rendered from: per-relation system + // column names have to match what CH now holds + Some(resolved), )); let (drain_res, pump_res) = tokio::join!(drain, pump); let drain_outcome = drain_res @@ -4404,11 +4418,13 @@ async fn run_bootstrap( /// Routing map for the bootstrap drain: explicit `[table.*]` seeded up front, /// then every seeded relation run through the DDL applicator's `Added` path so /// `auto_create` namespaces get their CH table created and mapping registered. +/// Returns the snapshot those CREATEs rendered from, so the drain freezes +/// routes against the same per-relation rules. async fn bootstrap_build_mapping( emitter_cfg: &EmitterConfig, catalog: &walshadow::backup_page_walk::CatalogMap, args: &Args, -) -> Result { +) -> Result<(MappingHandle, Arc)> { let mapping = walshadow::mapping::mapping_handle(emitter_cfg.tables.clone()); let cli_overrides = CliOverrides { drop_table_strategy: args.drop_table_strategy, @@ -4424,17 +4440,19 @@ async fn bootstrap_build_mapping( cli_base(args), mapping.clone(), ); - let (ddl_cfg, merged_tables) = { + let (ddl_cfg, merged_tables, resolved) = { let snap = config_rx.borrow(); ( walshadow::ch_ddl::DdlConfig::from_resolved( &snap, emitter_cfg.database.clone(), emitter_cfg.soft_delete, + emitter_cfg.system_columns.clone(), emitter_cfg.replicate_all, emitter_cfg.runtime_config_schema.clone(), ), Arc::new(snap.tables.clone()), + snap.clone(), ) }; // Publish rule-adjusted targets before creating tables @@ -4449,7 +4467,7 @@ async fn bootstrap_build_mapping( .await .with_context(|| format!("bootstrap: ensure CH table {}", desc.rel_name))?; } - Ok(mapping) + Ok((mapping, resolved)) } /// Mark bootstrap before extraction, clear only after backup and required diff --git a/src/config.rs b/src/config.rs index db70908e..263371c3 100644 --- a/src/config.rs +++ b/src/config.rs @@ -1133,11 +1133,113 @@ mod tests { let s = r.rules.settings(&rel); assert_eq!(s.target_database.as_deref(), Some("sql_db")); assert_eq!(s.replicate, Some(true), "TOML scope still applies"); - let ddl = - crate::emit::ch_ddl::DdlConfig::from_resolved(&r, "db".into(), false, false, None); + let ddl = crate::emit::ch_ddl::DdlConfig::from_resolved( + &r, + "db".into(), + false, + Arc::default(), + false, + None, + ); assert_eq!(ddl.declared_scope(&rel), Some(true)); } + #[test] + fn overlay_key_lists_override_toml() { + let base = EmitterConfig::from_toml_str( + "[ch]\n\ + [table.public.events]\n\ + order_by = [\"id\"]\n\ + primary_key = [\"id\"]\n\ + [table.public.other]\n\ + order_by = [\"a\"]\n", + ) + .unwrap(); + let mut overlay = ConfigOverlay::default(); + overlay.tables.insert( + RelName::new("public", "events"), + TableRow { + order_by: Some(vec!["tenant".into(), "id".into()]), + primary_key: Some(vec!["tenant".into()]), + ..Default::default() + }, + ); + let (r, _) = ConfigResolver::resolve( + &base, + &overlay, + &CliOverrides::default(), + &OptInState::default(), + &ColumnRules::default(), + ); + let events = RelName::new("public", "events"); + let s = r.rules.settings(&events); + assert_eq!(s.order_by.unwrap(), ["tenant", "id"]); + assert_eq!(s.primary_key.unwrap(), ["tenant"]); + let other = r.rules.settings(&RelName::new("public", "other")); + assert_eq!( + other.order_by.unwrap(), + ["a"], + "untouched relation keeps its TOML list" + ); + // A row reaches the DDL applicator through the resolved snapshot + let ddl = crate::emit::ch_ddl::DdlConfig::from_resolved( + &r, + "db".into(), + false, + Arc::default(), + false, + None, + ); + let settings = ddl.rules.settings(&events); + let shape = ddl.create_shape(&settings); + assert_eq!(shape.order_by, ["tenant", "id"]); + assert_eq!(shape.primary_key, ["tenant"]); + } + + /// A pattern row shapes relations no row can name: under `replicate_all` + /// the CREATE fires the first time a table is seen, before a literal row + /// for it could arrive + #[test] + fn overlay_pattern_row_renames_system_columns_of_unknown_relation() { + let base = EmitterConfig::from_toml_str("[ch]\n").unwrap(); + let mut overlay = ConfigOverlay::default(); + overlay.tables.insert( + RelName::new("app", "events_*"), + TableRow { + match_kind: Some("glob".into()), + system: crate::mapping::SystemColumnNames { + lsn: Some("_peerdb_version".into()), + is_deleted: Some(String::new()), + ..Default::default() + }, + ..Default::default() + }, + ); + let (r, rejections) = ConfigResolver::resolve( + &base, + &overlay, + &CliOverrides::default(), + &OptInState::default(), + &ColumnRules::default(), + ); + assert_eq!(rejections, 0); + let policy = crate::emit::route::RowPolicy::default(); + // No mapping, no descriptor, no CREATE yet: the shape is still known + let sys = policy + .for_rel(Some(&r), &RelName::new("app", "events_2026")) + .system; + assert_eq!(sys.lsn, "_peerdb_version"); + assert!(sys.is_deleted.is_none()); + assert_eq!(sys.xid, "_xid", "unnamed column inherits"); + assert_eq!( + *policy + .for_rel(Some(&r), &RelName::new("app", "orders")) + .system, + crate::mapping::SystemColumns::default(), + "a relation the pattern misses keeps the cluster-wide names" + ); + } + fn auto_create_set(base: &EmitterConfig, overlay: &ConfigOverlay) -> ahash::HashSet { use crate::emit::ch_ddl::DdlConfig; let (r, _) = ConfigResolver::resolve( @@ -1147,7 +1249,8 @@ mod tests { &OptInState::default(), &ColumnRules::default(), ); - DdlConfig::from_resolved(&r, "db".into(), false, false, None).auto_create_namespaces + DdlConfig::from_resolved(&r, "db".into(), false, Arc::default(), false, None) + .auto_create_namespaces } #[test] diff --git a/src/decode/codecs.rs b/src/decode/codecs.rs index b9169cfc..687b889b 100644 --- a/src/decode/codecs.rs +++ b/src/decode/codecs.rs @@ -6,7 +6,8 @@ //! every other Tier 3 type. Surfaced as //! [`crate::decode::heap_decoder::ColumnValue::PgPending`] carrying raw on-disk //! bytes; resolved at emit time by shadow PG's own `typoutput`, reached over -//! the bridge socket. One source of truth, no codec drift. +//! the bridge socket. One source of truth, no codec drift. Runtime config +//! reads its own `text[]` columns locally, decoding before the bridge exists. //! //! Each decoder takes the varlena *body* (or raw fixed-width bytes for //! `interval`) and produces a tagged value whose `text` matches PG @@ -44,6 +45,53 @@ pub enum CodecError { UnsupportedArrayElement(u32), } +/// Decode on-disk `text[]` body (outer varlena header stripped) for runtime +/// config's key lists. `None` for anything else: nested dims, NULL elements +/// (`dataoffset != 0`), other element types, malformed bytes +pub fn decode_text_array(body: &[u8]) -> Option> { + let word = |at: usize| Some(i32::from_le_bytes(body.get(at..at + 4)?.try_into().ok()?)); + let (ndim, dataoffset, elemtype) = (word(0)?, word(4)?, word(8)? as u32); + if dataoffset != 0 || elemtype != crate::schema::TEXTOID || !(0..=1).contains(&ndim) { + return None; + } + if ndim == 0 { + return Some(Vec::new()); + } + let nitems = usize::try_from(word(12)?).ok()?; + // ARR_DATA_PTR: MAXALIGN(sizeof(ArrayType) + 2 * 4 * ndim) less stripped header + let mut cursor = 20usize; + let mut values = Vec::new(); + for _ in 0..nitems { + cursor = cursor.next_multiple_of(4); + // Elements keep their own varlena header, nominal-aligned (`array_out`) + let first = *body.get(cursor)?; + let (header, total) = if first & 1 != 0 { + (1, (first >> 1) as usize) + } else { + (4, (word(cursor)? as u32 >> 2) as usize) + }; + let text = body.get(cursor + header..cursor.checked_add(total)?)?; + values.push(std::str::from_utf8(text).ok()?.to_owned()); + cursor += total; + } + Some(values) +} + +#[cfg(test)] +pub(crate) fn text_array_body(values: &[&str]) -> Vec { + let mut body = vec![0; 20]; + body[0..4].copy_from_slice(&1i32.to_le_bytes()); + body[8..12].copy_from_slice(&crate::schema::TEXTOID.to_le_bytes()); + body[12..16].copy_from_slice(&(values.len() as i32).to_le_bytes()); + body[16..20].copy_from_slice(&1i32.to_le_bytes()); + for value in values { + body.extend_from_slice(&(((value.len() + 4) as u32) << 2).to_le_bytes()); + body.extend_from_slice(value.as_bytes()); + body.resize(body.len().next_multiple_of(4), 0); + } + body +} + // numeric — varlena arbitrary-precision decimal. On-disk layout per // `src/backend/utils/adt/numeric.c`: // @@ -465,6 +513,24 @@ pub(crate) fn uuid_to_ch_wire(b: &[u8; 16]) -> [u8; 16] { mod tests { use super::*; + #[test] + fn text_array_decodes_identifiers_without_delimiter_parsing() { + let body = text_array_body(&["tenant,id", "tick`"]); + assert_eq!( + decode_text_array(&body).unwrap(), + ["tenant,id".to_owned(), "tick`".to_owned()] + ); + + let mut empty = vec![0; 12]; + empty[8..12].copy_from_slice(&crate::schema::TEXTOID.to_le_bytes()); + assert!(decode_text_array(&empty).unwrap().is_empty()); + + // dataoffset != 0 marks a NULL bitmap + let mut with_null = text_array_body(&["tenant"]); + with_null[4..8].copy_from_slice(&32i32.to_le_bytes()); + assert_eq!(decode_text_array(&with_null), None); + } + #[test] fn uuid_ch_wire_reverses_each_half() { let pg = [ diff --git a/src/emit/ch_ddl.rs b/src/emit/ch_ddl.rs index 70bc2d2c..9fea9fbf 100644 --- a/src/emit/ch_ddl.rs +++ b/src/emit/ch_ddl.rs @@ -36,11 +36,11 @@ use crate::config::{ConfigResolver, ResolvedConfig}; use crate::emit::ch_emitter::{EmitterConfig, RetryConfig}; use crate::mapping::{ ColumnMapping, DropTableStrategy, MappingHandle, MappingSnapshot, NamespaceMapping, - TableMapping, TableTarget, apply_column_rule, derive_columns_for_mapping, + SystemColumns, TableMapping, TableTarget, apply_column_rule, derive_columns_for_mapping, fold_diff_into_mapping, }; use crate::schema::{RelDescriptor, RelName, SchemaDiff, SchemaEvent, replident_key_attnums}; -use crate::table_rules::TableRules; +use crate::table_rules::{TableRule, TableRules}; use ahash::{HashMap, HashSet, HashSetExt}; /// Knobs that don't ride the INSERT pump. [`DdlApplicator`] rebuilds them @@ -62,21 +62,27 @@ pub struct DdlConfig { /// Per-namespace overrides, fallback to the global fields above when /// a namespace has none pub namespaces: HashMap, - /// Keep `_is_deleted` out of `ReplacingMergeTree`'s args so deletes + /// Keep the delete marker out of `ReplacingMergeTree`'s args so deletes /// stay queryable; mirrors [`EmitterConfig::soft_delete`] pub soft_delete: bool, + /// Cluster-wide names of the appended columns, and whether the delete + /// marker exists; mirrors [`EmitterConfig::system_columns`]. A per-relation + /// rule renames over it + pub system: Arc, pub rules: Arc, pub column_rules: Arc, } impl DdlConfig { /// Build from a resolved snapshot. `target_database`, `soft_delete`, - /// `replicate_all` and `runtime_config_schema` are boot-only knobs the - /// resolver does not republish, so callers thread them through unchanged. + /// `system`, `replicate_all` and `runtime_config_schema` are boot-only + /// knobs the resolver does not republish, so callers thread them through + /// unchanged. pub fn from_resolved( resolved: &ResolvedConfig, target_database: String, soft_delete: bool, + system: Arc, replicate_all: bool, runtime_config_schema: Option, ) -> Self { @@ -94,6 +100,7 @@ impl DdlConfig { target_database, namespaces: resolved.namespaces.clone(), soft_delete, + system, rules: resolved.rules.clone(), column_rules: resolved.column_rules.clone(), } @@ -118,18 +125,29 @@ impl DdlConfig { self } - fn create_target(&self, rel: &RelName) -> TableTarget { - let settings = self.rules.settings(rel); + fn create_target(&self, settings: &TableRule, rel: &RelName) -> TableTarget { TableTarget { database: settings .target_database + .clone() .unwrap_or_else(|| self.target_database_for(&rel.namespace).to_owned()), table: settings .target_table + .clone() .unwrap_or_else(|| rel.name.to_string()), } } + /// Destination shape for one relation's `CREATE TABLE` + pub(crate) fn create_shape<'a>(&'a self, settings: &'a TableRule) -> CreateShape<'a> { + CreateShape { + system: settings.system_columns(&self.system), + soft_delete: self.soft_delete, + order_by: settings.order_by.as_deref().unwrap_or_default(), + primary_key: settings.primary_key.as_deref().unwrap_or_default(), + } + } + /// Resolve explicit scope without opting in system relations pub fn declared_scope(&self, rel: &RelName) -> Option { match self.rules.settings(rel).replicate { @@ -152,6 +170,19 @@ impl DdlConfig { } } +/// Destination-shape inputs for one `CREATE TABLE` +pub struct CreateShape<'a> { + pub system: Arc, + /// Keep the delete marker out of `ReplacingMergeTree`'s args + pub soft_delete: bool, + /// Operator `ORDER BY` (destination column names); empty derives the key + /// from replica identity + pub order_by: &'a [String], + /// Operator `PRIMARY KEY`: the sparse-index prefix CH indexes, not a + /// uniqueness constraint. Empty leaves it equal to `ORDER BY` + pub primary_key: &'a [String], +} + pub(crate) fn is_system_namespace(ns: &str, runtime_config_schema: Option<&str>) -> bool { ns == "pg_catalog" || ns == "information_schema" @@ -231,10 +262,11 @@ impl DdlApplicator { &self.config } - /// Fold a republished snapshot into `config` (namespaces + drop - /// strategy). `target_database` + `soft_delete` are boot-only, so they - /// carry over. No-op until the resolver sends a new value; called at - /// each apply so DDL runs against the current config. + /// Fold a republished snapshot into `config` (namespaces, drop strategy, + /// table + column rules). `target_database`, `soft_delete` and the + /// system-column names are boot-only, so they carry over. No-op until the + /// resolver sends a new value; called at each apply so DDL runs against + /// the current config. async fn refresh_config(&mut self) -> Result<(), EmitterError> { if !self.config_rx.has_changed().unwrap_or(false) { return Ok(()); @@ -245,6 +277,7 @@ impl DdlApplicator { &snap, self.config.target_database.clone(), self.config.soft_delete, + self.config.system.clone(), self.config.replicate_all, self.config.runtime_config_schema.clone(), ); @@ -295,7 +328,9 @@ impl DdlApplicator { // no-ops an operator-managed table and re-creates after strategy=drop. if let Some(m) = self.mapping_for(&desc.rel_name).await { self.ensure_database(&m.target.database).await?; - let sql = render_create_table_from_mapping(desc, &m, self.config.soft_delete); + let settings = self.config.rules.settings(&desc.rel_name); + let sql = + render_create_table_from_mapping(desc, &m, &self.config.create_shape(&settings)); self.execute(&sql).await?; self.stats.creates_applied += 1; return Ok(()); @@ -312,13 +347,10 @@ impl DdlApplicator { } // Drives both CREATE TABLE and the row-routing mapping below so // rows and DDL land in the same place - let target = self.config.create_target(&desc.rel_name); - let Some(sql) = render_create_table( - desc, - &target, - self.config.soft_delete, - &self.config.column_rules, - )? + let settings = self.config.rules.settings(&desc.rel_name); + let target = self.config.create_target(&settings, &desc.rel_name); + let shape = self.config.create_shape(&settings); + let Some(sql) = render_create_table(desc, &target, &shape, &self.config.column_rules)? else { self.stats.skipped += 1; return Ok(()); @@ -343,13 +375,10 @@ impl DdlApplicator { /// Idempotent: `IF NOT EXISTS` no-ops a re-create. pub async fn ensure_ch_table(&mut self, desc: &RelDescriptor) -> Result { self.refresh_config().await?; - let target = self.config.create_target(&desc.rel_name); - let Some(sql) = render_create_table( - desc, - &target, - self.config.soft_delete, - &self.config.column_rules, - )? + let settings = self.config.rules.settings(&desc.rel_name); + let target = self.config.create_target(&settings, &desc.rel_name); + let shape = self.config.create_shape(&settings); + let Some(sql) = render_create_table(desc, &target, &shape, &self.config.column_rules)? else { tracing::warn!( target: "walshadow::ch_ddl", @@ -568,6 +597,7 @@ impl DdlApplicator { rc, self.config.target_database.clone(), self.config.soft_delete, + self.config.system.clone(), self.config.replicate_all, self.config.runtime_config_schema.clone(), ) @@ -657,8 +687,10 @@ fn predict_route_effect( { return Ok(None); } - let target = cfg.create_target(&desc.rel_name); - if render_create_table(desc, &target, cfg.soft_delete, &cfg.column_rules)?.is_none() { + let settings = cfg.rules.settings(&desc.rel_name); + let target = cfg.create_target(&settings, &desc.rel_name); + let shape = cfg.create_shape(&settings); + if render_create_table(desc, &target, &shape, &cfg.column_rules)?.is_none() { return Ok(None); } let columns = derive_columns_for_mapping(desc, &cfg.column_rules); @@ -740,40 +772,117 @@ pub fn render_add_column(target: &str, name: &str, resolved: &ResolvedColumn) -> } /// Shared CREATE tail: synthetic columns (mirror `TablePlan::build`), -/// engine, `ORDER BY` key names (else `_lsn`) +/// engine, `ORDER BY` key names (else the LSN column) fn render_create_sql( target: &str, mut col_defs: Vec, key_names: Vec, - soft_delete: bool, + shape: &CreateShape<'_>, ) -> String { - col_defs.push("`_lsn` UInt64".into()); - col_defs.push("`_xid` UInt32".into()); - col_defs.push("`_commit_ts` DateTime64(6, 'UTC')".into()); - col_defs.push("`_is_deleted` Bool".into()); - // soft_delete keeps `_is_deleted` out of the engine args - let engine_args = if soft_delete { - "`_lsn`" - } else { - "`_lsn`, `_is_deleted`" + let sys = &shape.system; + let lsn = quote_ident(&sys.lsn); + col_defs.push(format!("{lsn} UInt64")); + col_defs.push(format!("{} UInt32", quote_ident(&sys.xid))); + col_defs.push(format!( + "{} DateTime64(6, 'UTC')", + quote_ident(&sys.commit_ts) + )); + // soft_delete keeps the delete marker out of the engine args; without the + // marker column there is nothing to pass either + let engine_args = match &sys.is_deleted { + Some(name) => { + let marker = quote_ident(name); + col_defs.push(format!("{marker} Bool")); + if shape.soft_delete { + lsn.clone() + } else { + format!("{lsn}, {marker}") + } + } + None => lsn.clone(), }; - let order_by = if key_names.is_empty() { - "(`_lsn`)".to_string() + let keys = if key_names.is_empty() { + vec![lsn] } else { - format!("({})", key_names.join(", ")) + key_names }; + // CH indexes the PRIMARY KEY prefix of the sorting key; a non-prefix is a + // CREATE-time error there, so drop it and let the key default to ORDER BY + let primary_key = match shape.primary_key { + [] => String::new(), + pk if pk + .iter() + .map(|n| quote_ident(n)) + .eq(keys.iter().take(pk.len()).cloned()) => + { + format!("\nPRIMARY KEY ({})", keys[..pk.len()].join(", ")) + } + pk => { + tracing::warn!( + target: "walshadow::ch_ddl", + table = %target, + primary_key = ?pk, + "primary_key ignored: not a prefix of ORDER BY", + ); + String::new() + } + }; + let order_by = keys.join(", "); format!( - "CREATE TABLE IF NOT EXISTS {target} (\n {}\n) ENGINE = ReplacingMergeTree({engine_args})\nORDER BY {order_by}", + "CREATE TABLE IF NOT EXISTS {target} (\n {}\n) ENGINE = ReplacingMergeTree({engine_args})\nORDER BY ({order_by}){primary_key}", col_defs.join(",\n ") ) } +/// Operator `ORDER BY` names → quoted key list. `orderable` maps every +/// destination column name a sort key may use to whether CH would reject it +/// (`Nullable` sort keys are illegal). An override naming an absent or +/// nullable column is dropped whole with a WARN, so one bad config row +/// degrades to the replica-identity key instead of failing every CREATE. +fn resolve_order_by( + target: &str, + order_by: &[String], + orderable: &HashMap<&str, bool>, + derived: Vec, +) -> Vec { + if order_by.is_empty() { + return derived; + } + let mut keys = Vec::with_capacity(order_by.len()); + for name in order_by { + let Some(false) = orderable.get(name.as_str()) else { + tracing::warn!( + target: "walshadow::ch_ddl", + table = %target, + column = %name, + reason = if orderable.contains_key(name.as_str()) { "nullable" } else { "absent" }, + "order_by ignored; keying on replica identity", + ); + return derived; + }; + keys.push(quote_ident(name)); + } + keys +} + +/// System columns are always non-nullable, so any of them may sort +fn orderable_system<'a>(sys: &'a SystemColumns, orderable: &mut HashMap<&'a str, bool>) { + for name in sys.names() { + orderable.insert(name, false); + } +} + +/// ClickHouse rejects Nullable columns in a sorting key +fn is_nullable(ch_type: &str) -> bool { + ch_type.starts_with("Nullable(") +} + /// `CREATE TABLE IF NOT EXISTS` for an autodiscovered relation. `None` /// when a column's type can't be bridged; caller logs + skips. pub fn render_create_table( desc: &RelDescriptor, target: &TableTarget, - soft_delete: bool, + shape: &CreateShape<'_>, rules: &ColumnRules, ) -> Result, EmitterError> { let target = target.sql(); @@ -794,32 +903,34 @@ pub fn render_create_table( resolved, rules.settings(&desc.rel_name, &att.name), ); - cols.push((att.attnum, quote_ident(&name), resolved)); + cols.push((att.attnum, name, resolved)); } let col_defs: Vec = cols .iter() .map(|(_, name, r)| { + let name = quote_ident(name); r.default_sql.as_ref().map_or_else( || format!("{name} {}", r.ch_type), |d| format!("{name} {} DEFAULT {d}", r.ch_type), ) }) .collect(); + let mut orderable: HashMap<&str, bool> = HashMap::default(); + orderable_system(&shape.system, &mut orderable); + for (_, name, r) in &cols { + orderable.insert(name, is_nullable(&r.ch_type)); + } // ClickHouse rejects Nullable columns in ORDER BY - let key_names: Vec = pk_attnums + let derived: Vec = pk_attnums .iter() .filter_map(|a| { cols.iter() - .find(|(attnum, _, r)| attnum == a && !r.ch_type.starts_with("Nullable(")) - .map(|(_, name, _)| name.clone()) + .find(|(attnum, _, r)| attnum == a && !is_nullable(&r.ch_type)) + .map(|(_, name, _)| quote_ident(name)) }) .collect(); - Ok(Some(render_create_sql( - &target, - col_defs, - key_names, - soft_delete, - ))) + let key_names = resolve_order_by(&target, shape.order_by, &orderable, derived); + Ok(Some(render_create_sql(&target, col_defs, key_names, shape))) } /// CH `UNKNOWN_DATABASE` @@ -860,39 +971,61 @@ pub async fn ensure_boot_database(cfg: &EmitterConfig) -> Result, ) -> String { let col_defs: Vec = mapping .columns .iter() .map(|c| format!("{} {}", quote_ident(&c.target_name), c.target_type)) .collect(); - let key_names: Vec = replident_key_attnums(desc) + let mut orderable: HashMap<&str, bool> = HashMap::default(); + orderable_system(&shape.system, &mut orderable); + for c in &mapping.columns { + orderable.insert(&c.target_name, is_nullable(&c.target_type)); + } + let derived: Vec = replident_key_attnums(desc) .iter() .filter_map(|a| { mapping .columns .iter() - .find(|c| c.src_attnum == *a && !c.target_type.starts_with("Nullable")) + .find(|c| c.src_attnum == *a && !is_nullable(&c.target_type)) .map(|c| quote_ident(&c.target_name)) }) .collect(); - render_create_sql(&mapping.target.sql(), col_defs, key_names, soft_delete) + let target = mapping.target.sql(); + let key_names = resolve_order_by(&target, shape.order_by, &orderable, derived); + render_create_sql(&target, col_defs, key_names, shape) } #[cfg(test)] mod tests { use super::*; use crate::mapping::{ColumnMapping, TableMapping}; + use std::sync::LazyLock; + + static SYS: LazyLock> = LazyLock::new(Arc::default); + fn dest(database: &str, desc: &RelDescriptor) -> TableTarget { TableTarget::new(database, &desc.rel_name.name) } + + /// Default system columns, no operator key override + fn shape(soft_delete: bool) -> CreateShape<'static> { + CreateShape { + system: SYS.clone(), + soft_delete, + order_by: &[], + primary_key: &[], + } + } use crate::schema::{INT4OID, TEXTOID, TIMESTAMPTZOID}; use crate::schema::{RelAttr, RelDescriptor, ReplIdent, SchemaDiff}; use crate::table_rules::MatchKind; @@ -938,6 +1071,7 @@ mod tests { target_database: "default".into(), namespaces, soft_delete: false, + system: Arc::default(), rules: Arc::default(), column_rules: Arc::default(), }; @@ -963,6 +1097,7 @@ mod tests { target_database: "default".into(), namespaces: HashMap::new(), soft_delete: false, + system: Arc::default(), rules: Arc::default(), column_rules: Arc::default(), }; @@ -1021,6 +1156,11 @@ mod tests { TableRule { replicate: Some(true), target_database: Some("warehouse".into()), + system: crate::mapping::SystemColumnNames { + lsn: Some("_peerdb_version".into()), + is_deleted: Some(String::new()), + ..Default::default() + }, ..TableRule::default() }, ); @@ -1050,13 +1190,14 @@ mod tests { target_database: "default".into(), namespaces: ahash::HashMap::default(), soft_delete: false, + system: Arc::default(), rules: Arc::new(rules), column_rules: Arc::default(), }; let events = RelName::new("app", "events_1"); assert!(cfg.auto_creates(&events)); assert_eq!( - cfg.create_target(&events), + cfg.create_target(&cfg.rules.settings(&events), &events), TableTarget::new("warehouse", "events_1") ); assert_eq!( @@ -1067,10 +1208,22 @@ mod tests { assert!(!cfg.auto_creates(&RelName::new("app", "events_audit"))); assert_eq!(cfg.declared_scope(&RelName::new("walshadow", "x")), None); assert!(!cfg.auto_creates(&RelName::new("pg_catalog", "pg_class"))); + let settings = cfg.rules.settings(&events); + let shape = cfg.create_shape(&settings); + assert_eq!(shape.system.lsn, "_peerdb_version"); + assert!(shape.system.is_deleted.is_none(), "marker dropped"); + assert_eq!(shape.system.xid, "_xid", "unnamed column inherits"); + let other = RelName::new("other", "t"); assert_eq!( - cfg.create_target(&RelName::new("other", "t")), + cfg.create_target(&cfg.rules.settings(&other), &other), TableTarget::new("default", "t") ); + let settings = cfg.rules.settings(&other); + assert_eq!( + cfg.create_shape(&settings).system.lsn, + "_lsn", + "a relation no entry renames keeps the cluster-wide names" + ); } #[test] @@ -1130,7 +1283,7 @@ mod tests { target_type: None, }, ); - let sql = render_create_table(&d, &dest("db", &d), false, &b.finish().0) + let sql = render_create_table(&d, &dest("db", &d), &shape(false), &b.finish().0) .unwrap() .expect("renderable"); assert!(sql.contains("`order_id` Int32"), "{sql}"); @@ -1152,7 +1305,7 @@ mod tests { target_type: Some("Nullable(Int32)".into()), }, ); - let sql = render_create_table(&d, &dest("db", &d), false, &b.finish().0) + let sql = render_create_table(&d, &dest("db", &d), &shape(false), &b.finish().0) .unwrap() .expect("renderable"); assert!(sql.ends_with("ORDER BY (`_lsn`)"), "{sql}"); @@ -1168,9 +1321,14 @@ mod tests { ], Some(vec![1]), ); - let sql = render_create_table(&d, &dest("default", &d), false, &ColumnRules::default()) - .unwrap() - .unwrap(); + let sql = render_create_table( + &d, + &dest("default", &d), + &shape(false), + &ColumnRules::default(), + ) + .unwrap() + .unwrap(); assert!(sql.contains("CREATE TABLE IF NOT EXISTS `default`.`orders`")); assert!(sql.contains("`id` Int32")); assert!(sql.contains("`body` Nullable(String)")); @@ -1196,9 +1354,14 @@ mod tests { d.replident = ReplIdent::Full { pk_attnums: Some(vec![1]), }; - let sql = render_create_table(&d, &dest("default", &d), false, &ColumnRules::default()) - .unwrap() - .unwrap(); + let sql = render_create_table( + &d, + &dest("default", &d), + &shape(false), + &ColumnRules::default(), + ) + .unwrap() + .unwrap(); assert!(sql.ends_with("ORDER BY (`id`)"), "{sql}"); assert!(!sql.contains("ORDER BY _lsn"), "{sql}"); } @@ -1215,9 +1378,14 @@ mod tests { ], Some(vec![2, 1]), ); - let sql = render_create_table(&d, &dest("default", &d), false, &ColumnRules::default()) - .unwrap() - .unwrap(); + let sql = render_create_table( + &d, + &dest("default", &d), + &shape(false), + &ColumnRules::default(), + ) + .unwrap() + .unwrap(); assert!(sql.contains("`a` Int32"), "{sql}"); assert!(sql.contains("`b` Int32"), "{sql}"); assert!(!sql.contains("`a` Nullable"), "{sql}"); @@ -1236,9 +1404,14 @@ mod tests { ], Some(vec![1]), ); - let sql = render_create_table(&d, &dest("default", &d), true, &ColumnRules::default()) - .unwrap() - .unwrap(); + let sql = render_create_table( + &d, + &dest("default", &d), + &shape(true), + &ColumnRules::default(), + ) + .unwrap() + .unwrap(); // Column always present; soft_delete only drops it from the engine assert!(sql.contains("`_is_deleted` Bool")); assert!(sql.contains("ENGINE = ReplacingMergeTree(`_lsn`)")); @@ -1249,9 +1422,14 @@ mod tests { #[test] fn render_create_table_falls_back_to_lsn_when_no_pk() { let d = desc("events", vec![att(1, "body", TEXTOID, false, None)], None); - let sql = render_create_table(&d, &dest("default", &d), false, &ColumnRules::default()) - .unwrap() - .unwrap(); + let sql = render_create_table( + &d, + &dest("default", &d), + &shape(false), + &ColumnRules::default(), + ) + .unwrap() + .unwrap(); assert!(sql.ends_with("ORDER BY (`_lsn`)")); } @@ -1271,9 +1449,14 @@ mod tests { index_oid: 16500, key_attnums: vec![2, 1], }; - let sql = render_create_table(&d, &dest("default", &d), false, &ColumnRules::default()) - .unwrap() - .unwrap(); + let sql = render_create_table( + &d, + &dest("default", &d), + &shape(false), + &ColumnRules::default(), + ) + .unwrap() + .unwrap(); assert!(!sql.contains("`key` Nullable"), "{sql}"); assert!(!sql.contains("`tenant` Nullable"), "{sql}"); assert!(sql.contains("`body` Nullable(String)"), "{sql}"); @@ -1288,9 +1471,14 @@ mod tests { vec![att(2, "body", TEXTOID, false, None)], Some(vec![1]), ); - let sql = render_create_table(&d, &dest("default", &d), false, &ColumnRules::default()) - .unwrap() - .unwrap(); + let sql = render_create_table( + &d, + &dest("default", &d), + &shape(false), + &ColumnRules::default(), + ) + .unwrap() + .unwrap(); assert!(sql.ends_with("ORDER BY (`_lsn`)"), "{sql}"); } @@ -1299,7 +1487,7 @@ mod tests { let mut a = att(1, "ship_at", TIMESTAMPTZOID, false, None); a.typmod = 3; let d = desc("t", vec![a], None); - let sql = render_create_table(&d, &dest("db", &d), false, &ColumnRules::default()) + let sql = render_create_table(&d, &dest("db", &d), &shape(false), &ColumnRules::default()) .unwrap() .unwrap(); assert!( @@ -1333,7 +1521,7 @@ mod tests { }, ], }; - let sql = render_create_table_from_mapping(&d, &m, false); + let sql = render_create_table_from_mapping(&d, &m, &shape(false)); assert!( sql.contains("CREATE TABLE IF NOT EXISTS `warehouse`.`orders_pinned`"), "{sql}" @@ -1356,10 +1544,198 @@ mod tests { target_type: "Nullable(Int32)".into(), }], }; - let sql = render_create_table_from_mapping(&d, &m, false); + let sql = render_create_table_from_mapping(&d, &m, &shape(false)); + assert!(sql.ends_with("ORDER BY (`_lsn`)"), "{sql}"); + } + + #[test] + fn render_create_table_renames_system_columns() { + let sys = SystemColumns { + lsn: "_peerdb_version".into(), + xid: "_xid".into(), + commit_ts: "_peerdb_synced_at".into(), + is_deleted: Some("_peerdb_is_deleted".into()), + }; + let d = desc("t", vec![att(1, "id", INT4OID, true, None)], Some(vec![1])); + let sql = render_create_table( + &d, + &dest("default", &d), + &CreateShape { + system: sys.clone().into(), + soft_delete: false, + order_by: &[], + primary_key: &[], + }, + &ColumnRules::default(), + ) + .unwrap() + .unwrap(); + assert!(sql.contains("`_peerdb_version` UInt64"), "{sql}"); + assert!( + sql.contains("`_peerdb_synced_at` DateTime64(6, 'UTC')"), + "{sql}" + ); + assert!(sql.contains("`_peerdb_is_deleted` Bool"), "{sql}"); + assert!( + sql.contains("ReplacingMergeTree(`_peerdb_version`, `_peerdb_is_deleted`)"), + "{sql}" + ); + assert!(!sql.contains("`_lsn`"), "{sql}"); + } + + #[test] + fn render_create_table_without_delete_marker() { + // No marker column, so nothing to hand ReplacingMergeTree as its + // deletion arg; keyless tables still sort on the LSN column + let sys = SystemColumns { + is_deleted: None, + ..SystemColumns::default() + }; + let d = desc("t", vec![att(1, "id", INT4OID, true, None)], None); + let sql = render_create_table( + &d, + &dest("default", &d), + &CreateShape { + system: sys.clone().into(), + soft_delete: false, + order_by: &[], + primary_key: &[], + }, + &ColumnRules::default(), + ) + .unwrap() + .unwrap(); + assert!(!sql.contains("_is_deleted"), "{sql}"); + assert!(sql.contains("ENGINE = ReplacingMergeTree(`_lsn`)"), "{sql}"); assert!(sql.ends_with("ORDER BY (`_lsn`)"), "{sql}"); } + #[test] + fn render_create_table_takes_operator_order_by_over_pk() { + let d = desc( + "t", + vec![ + att(1, "id", INT4OID, true, None), + att(2, "tenant", INT4OID, true, None), + ], + Some(vec![1]), + ); + let order_by = vec!["tenant".to_string(), "id".to_string()]; + let sql = render_create_table( + &d, + &dest("default", &d), + &CreateShape { + system: SYS.clone(), + soft_delete: false, + order_by: &order_by, + primary_key: &[], + }, + &ColumnRules::default(), + ) + .unwrap() + .unwrap(); + assert!(sql.ends_with("ORDER BY (`tenant`, `id`)"), "{sql}"); + } + + #[test] + fn render_create_table_order_by_ignored_when_column_absent_or_nullable() { + let d = desc( + "t", + vec![ + att(1, "id", INT4OID, true, None), + att(2, "body", TEXTOID, false, None), + ], + Some(vec![1]), + ); + for keys in [vec!["nope".to_string()], vec!["body".to_string()]] { + let sql = render_create_table( + &d, + &dest("default", &d), + &CreateShape { + system: SYS.clone(), + soft_delete: false, + order_by: &keys, + primary_key: &[], + }, + &ColumnRules::default(), + ) + .unwrap() + .unwrap(); + assert!(sql.ends_with("ORDER BY (`id`)"), "{keys:?}: {sql}"); + } + } + + #[test] + fn render_create_table_primary_key_prefix_renders_non_prefix_drops() { + let d = desc( + "t", + vec![ + att(1, "id", INT4OID, true, None), + att(2, "tenant", INT4OID, true, None), + ], + Some(vec![1]), + ); + let order_by = vec!["tenant".to_string(), "id".to_string()]; + let mk = |primary_key: &[String]| { + render_create_table( + &d, + &dest("default", &d), + &CreateShape { + system: SYS.clone(), + soft_delete: false, + order_by: &order_by, + primary_key, + }, + &ColumnRules::default(), + ) + .unwrap() + .unwrap() + }; + let sql = mk(&["tenant".to_string()]); + assert!( + sql.ends_with("ORDER BY (`tenant`, `id`)\nPRIMARY KEY (`tenant`)"), + "{sql}" + ); + // `id` is not a prefix of (tenant, id): CH would reject the CREATE + let sql = mk(&["id".to_string()]); + assert!(sql.ends_with("ORDER BY (`tenant`, `id`)"), "{sql}"); + } + + #[test] + fn render_create_table_from_mapping_takes_operator_order_by() { + let d = desc("t", vec![att(1, "id", INT4OID, true, None)], Some(vec![1])); + let m = TableMapping { + target: TableTarget::new("db", "t"), + columns: vec![ + ColumnMapping { + src_attnum: 1, + target_name: "order_id".into(), + target_type: "Int32".into(), + }, + ColumnMapping { + src_attnum: 2, + target_name: "tenant".into(), + target_type: "Int32".into(), + }, + ], + }; + let order_by = vec!["tenant".to_string(), "order_id".to_string()]; + let sql = render_create_table_from_mapping( + &d, + &m, + &CreateShape { + system: SYS.clone(), + soft_delete: false, + order_by: &order_by, + primary_key: &["tenant".to_string()], + }, + ); + assert!( + sql.ends_with("ORDER BY (`tenant`, `order_id`)\nPRIMARY KEY (`tenant`)"), + "{sql}" + ); + } + #[test] fn drop_table_strategy_parses() { assert_eq!( @@ -1382,7 +1758,8 @@ mod tests { // type_bridge falls back to String for unknown OIDs today, so // this never hits None; revisit if the bridge grows strictness let d = desc("t", vec![att(1, "id", 99999, true, None)], None); - let sql = render_create_table(&d, &dest("db", &d), false, &ColumnRules::default()).unwrap(); + let sql = render_create_table(&d, &dest("db", &d), &shape(false), &ColumnRules::default()) + .unwrap(); assert!(sql.is_some(), "fallback path keeps the CREATE renderable"); } @@ -1457,6 +1834,7 @@ mod tests { target_database: "default".into(), namespaces: ahash::HashMap::default(), soft_delete: false, + system: Arc::default(), rules: Arc::default(), column_rules: Arc::default(), }; diff --git a/src/emit/ch_emitter.rs b/src/emit/ch_emitter.rs index cddd78cd..92faacc8 100644 --- a/src/emit/ch_emitter.rs +++ b/src/emit/ch_emitter.rs @@ -3,8 +3,9 @@ //! //! Synthetic columns `_lsn UInt64`, `_xid UInt32`, `_commit_ts //! DateTime64(6, 'UTC')`, `_is_deleted Bool` append after every mapped -//! column. `_is_deleted` (1 on delete) wires `ReplacingMergeTree`'s -//! deletion arg unless `EmitterConfig::soft_delete` keeps it queryable. +//! column; [`SystemColumns`] renames them and can drop the delete marker. +//! `_is_deleted` (1 on delete) wires `ReplacingMergeTree`'s deletion arg +//! unless `EmitterConfig::soft_delete` keeps it queryable. //! PG `TimestampTz` epoch is 2000-01-01; shift to Unix epoch //! (`DATETIME64_PG_EPOCH_US`) to match CH `DateTime64(6)`. //! @@ -37,8 +38,8 @@ use crate::column_rules::{ColumnEntry, ColumnRule, ColumnRules}; use crate::decode::decoder_sink::DecoderSinkError; use crate::decode::heap_decoder::{ColumnValue, CommittedTuple, HeapOp}; use crate::mapping::{ - ColumnMapping, DropTableStrategy, NamespaceMapping, TableMapping, TableTarget, ToastConfig, - ToastMode, + ColumnMapping, DropTableStrategy, NamespaceMapping, SystemColumnNames, SystemColumns, + TableMapping, TableTarget, ToastConfig, ToastMode, }; use crate::runtime_config::{InitialLoadMode, TableRow}; use crate::schema::{RelDescriptor, RelName}; @@ -56,7 +57,8 @@ pub(crate) const DATETIME64_PG_EPOCH_US: i64 = walrus::pg::replication::PG_EPOCH /// Days between PG `date` epoch (2000-01-01) and the Unix epoch. pub(crate) const DATE32_PG_EPOCH_DAYS: i32 = (DATETIME64_PG_EPOCH_US / 1_000_000 / 86_400) as i32; -/// Heap op codes for [`TableEncoder::append_row`]; `OP_DELETE` sets `_is_deleted` +/// Heap op codes for [`TableEncoder::append_row`]; `OP_DELETE` sets the +/// delete marker pub(crate) const OP_INSERT: i8 = 1; pub(crate) const OP_UPDATE: i8 = 2; pub(crate) const OP_DELETE: i8 = 3; @@ -148,10 +150,14 @@ pub struct EmitterConfig { /// full `insert_timeout` on a dead socket. Guards the start of a run, /// when connections sat idle since the previous one. pub idle_reconnect: Duration, - /// Keep `_is_deleted` out of `ReplacingMergeTree`'s args so delete + /// Keep the delete marker out of `ReplacingMergeTree`'s args so delete /// tombstones stay queryable instead of collapsing on FINAL. Column - /// always emitted; off by default + /// still emitted; off by default pub soft_delete: bool, + /// `[system_columns]`: cluster-wide names of the columns walshadow appends + /// per row, and whether the delete marker exists. Boot-only; per-relation + /// renames layer over it via `table_entries` / `config_table` + pub system_columns: Arc, /// Where externally-TOASTed chunks live + miss policy. `[toast]` block; /// default disabled (NULL/default-fill unrecoverable values) pub toast: ToastConfig, @@ -322,6 +328,7 @@ impl Default for EmitterConfig { insert_timeout: Duration::from_secs(DEFAULT_INSERT_TIMEOUT_SECS), idle_reconnect: Duration::from_secs(DEFAULT_IDLE_RECONNECT_SECS), soft_delete: false, + system_columns: Arc::default(), toast: ToastConfig::default(), decode_chunk_rows: DEFAULT_DECODE_CHUNK_ROWS, drain_batch_rows: DEFAULT_DRAIN_BATCH_ROWS, @@ -527,6 +534,8 @@ struct ConfigDocument { #[serde(default)] stream: StreamPatch, #[serde(default)] + system_columns: SystemColumns, + #[serde(default)] source: crate::config::SourceConn, backup: Option, #[serde(default)] @@ -606,6 +615,14 @@ struct TablePatch { target_table: Option, #[serde(default, deserialize_with = "crate::toml_de::de_from_str")] initial_load: Option, + order_by: Option>, + primary_key: Option>, + /// Same four keys as `[system_columns]`, for this entry's relations alone + lsn: Option, + xid: Option, + commit_ts: Option, + #[serde(default, deserialize_with = "crate::mapping::de_marker_override")] + is_deleted: Option, #[serde( rename = "match", default, @@ -632,6 +649,14 @@ struct ColumnPatch { } impl EmitterConfig { + /// Boot-only row-shape knobs frozen into every route + pub fn row_policy(&self) -> crate::emit::route::RowPolicy { + crate::emit::route::RowPolicy { + soft_delete: self.soft_delete, + system: self.system_columns.clone(), + } + } + /// Parse a TOML config of the shape: /// /// ```toml @@ -643,11 +668,21 @@ impl EmitterConfig { /// password = "" /// compression = "lz4" # one of none / lz4 / zstd /// + /// [system_columns] # optional: rename what walshadow appends per row + /// lsn = "_lsn" + /// xid = "_xid" + /// commit_ts = "_commit_ts" + /// is_deleted = "_is_deleted" # false drops the marker (and DELETE rows) + /// /// [table.public.foo] # [table..], quote weird names /// replicate = true /// initial_load = "none" # one of none / copy / base_backup / object_store /// target_database = "default" # optional: namespace override, else [ch] database /// target_table = "foo" # optional: source relname + /// order_by = ["id"] # optional: CH ORDER BY, else replica identity + /// primary_key = ["id"] # optional: index prefix of order_by + /// lsn = "_peerdb_version" # optional: per-relation system column + /// is_deleted = false # renames, same keys as [system_columns] /// columns = [ /// { attnum = 1, target = "id", type = "UInt64" }, /// { attnum = 2, target = "name", type = "Nullable(String)" }, @@ -697,6 +732,10 @@ impl EmitterConfig { .map_or(out.retry.max_backoff, Duration::from_millis); out.drop_table_strategy = ch.drop_table_strategy.unwrap_or(out.drop_table_strategy); out.soft_delete = ch.soft_delete.unwrap_or(out.soft_delete); + doc.system_columns + .validate() + .map_err(EmitterError::Config)?; + out.system_columns = Arc::new(doc.system_columns); out.toast.mode = doc.toast.mode.unwrap_or(out.toast.mode); out.resident_payload_max = doc .memory @@ -734,11 +773,21 @@ impl EmitterConfig { let ctx = format!("table.{ns}.{name}"); let kind = t.match_kind.unwrap_or(MatchKind::Exact); let replicate = t.replicate; + let system = SystemColumnNames { + lsn: t.lsn, + xid: t.xid, + commit_ts: t.commit_ts, + is_deleted: t.is_deleted, + }; + system.validate(&ctx).map_err(EmitterError::Config)?; let rule = TableRule { + system, target_database: t.target_database, target_table: t.target_table, replicate, initial_load: t.initial_load.map(|m| m.as_str().to_string()), + order_by: t.order_by, + primary_key: t.primary_key, }; out.table_entries.push((rel.clone(), kind, rule.clone())); let mut pinned = Vec::new(); @@ -863,8 +912,9 @@ pub(crate) struct TablePlan { pub(crate) synth_lsn: ColumnPlan, pub(crate) synth_xid: ColumnPlan, pub(crate) synth_commit_ts: ColumnPlan, - /// `_is_deleted Bool` (1 on delete, else 0), always appended last - pub(crate) synth_is_deleted: ColumnPlan, + /// Delete marker `Bool` (1 on delete, else 0), appended last. `None` when + /// `[system_columns] is_deleted = false` drops it + pub(crate) synth_is_deleted: Option, /// Pre-formatted so on-tuple paths don't reassemble per row pub(crate) insert_sql: String, } @@ -916,6 +966,7 @@ impl TablePlan { rel: &RelDescriptor, mapping: &TableMapping, column_rules: &ColumnRules, + system: &SystemColumns, ) -> Result { let mut columns = Vec::with_capacity(mapping.columns.len()); let mut col_sql = Vec::with_capacity(mapping.columns.len() + 4); @@ -983,14 +1034,18 @@ impl TablePlan { decimal: None, }) }; - let synth_lsn = mk("_lsn", "UInt64")?; - let synth_xid = mk("_xid", "UInt32")?; - let synth_commit_ts = mk("_commit_ts", "DateTime64(6, 'UTC')")?; - let synth_is_deleted = mk("_is_deleted", "Bool")?; + let synth_lsn = mk(&system.lsn, "UInt64")?; + let synth_xid = mk(&system.xid, "UInt32")?; + let synth_commit_ts = mk(&system.commit_ts, "DateTime64(6, 'UTC')")?; + let synth_is_deleted = system + .is_deleted + .as_deref() + .map(|name| mk(name, "Bool")) + .transpose()?; col_sql.push(quote_ident(&synth_lsn.name)); col_sql.push(quote_ident(&synth_xid.name)); col_sql.push(quote_ident(&synth_commit_ts.name)); - col_sql.push(quote_ident(&synth_is_deleted.name)); + col_sql.extend(synth_is_deleted.iter().map(|c| quote_ident(&c.name))); let insert_sql = format!( "INSERT INTO {} ({}) FORMAT Native", mapping.target.sql(), @@ -1012,7 +1067,7 @@ pub(crate) struct TableEncoder { pub(crate) plan: TablePlan, pub(crate) rows: usize, pub(crate) approx_bytes: usize, - /// Mirrors `plan.columns + 4 synth` + /// Mirrors `plan.columns` plus the synthetic columns the plan carries pub(crate) buffers: Vec, } @@ -1192,7 +1247,7 @@ impl ColumnBuf { } } -/// Fresh per-column buffers matching `plan` (mapped + four synthetic). +/// Fresh per-column buffers matching `plan` (mapped + synthetic). /// Shared by [`TableEncoder::new`] and [`TableEncoder::take_block`] so /// synthetic-column widths live in one place. pub(crate) fn fresh_buffers(plan: &TablePlan) -> Result, EmitterError> { @@ -1203,19 +1258,21 @@ pub(crate) fn fresh_buffers(plan: &TablePlan) -> Result, EmitterE buffers.push(ColumnBuf::Fixed { width: 8, bytes: Vec::new(), - }); // _lsn UInt64 + }); // lsn UInt64 buffers.push(ColumnBuf::Fixed { width: 4, bytes: Vec::new(), - }); // _xid UInt32 + }); // xid UInt32 buffers.push(ColumnBuf::Fixed { width: 8, bytes: Vec::new(), - }); // _commit_ts DateTime64(6) - buffers.push(ColumnBuf::Fixed { - width: 1, - bytes: Vec::new(), - }); // _is_deleted Bool (1 wire byte, same as UInt8) + }); // commit_ts DateTime64(6) + if plan.synth_is_deleted.is_some() { + buffers.push(ColumnBuf::Fixed { + width: 1, + bytes: Vec::new(), + }); // delete marker Bool (1 wire byte, same as UInt8) + } Ok(buffers) } @@ -1277,14 +1334,16 @@ impl TableEncoder { })?, } } - // Synthetic columns: _lsn, _xid, _commit_ts (unix micros), _is_deleted + // Synthetic columns: lsn, xid, commit_ts (unix micros), delete marker let off = mapping.columns.len(); push_fixed(&mut self.buffers[off], &decoded.source_lsn.to_le_bytes())?; push_fixed(&mut self.buffers[off + 1], &decoded.xid.to_le_bytes())?; let unix_us = committed.commit_ts.saturating_add(DATETIME64_PG_EPOCH_US); push_fixed(&mut self.buffers[off + 2], &unix_us.to_le_bytes())?; - let is_deleted: u8 = (op_code == OP_DELETE).into(); - push_fixed(&mut self.buffers[off + 3], &is_deleted.to_le_bytes())?; + if self.plan.synth_is_deleted.is_some() { + let is_deleted: u8 = (op_code == OP_DELETE).into(); + push_fixed(&mut self.buffers[off + 3], &is_deleted.to_le_bytes())?; + } self.rows += 1; self.approx_bytes = self.buffers.iter().map(ColumnBuf::approx_size).sum(); Ok(()) @@ -1686,6 +1745,9 @@ crate::atomic_stats! { pub blocks_sent, pub xacts_committed, pub unsupported_relations, + /// DELETE rows dropped because `[system_columns] is_deleted = false` + /// leaves them nowhere to land + pub deletes_discarded, /// `retries_attempted` counts one per failing operation, not per /// attempt (one op needing 3 retries adds 3) pub reconnects, @@ -2301,7 +2363,14 @@ mod tests { let alloc = Allocator::stdlib(); let rel = mk_rel(); let m = mk_mapping(); - let plan = TablePlan::build(alloc, &rel, &m, &ColumnRules::default()).expect("plan builds"); + let plan = TablePlan::build( + alloc, + &rel, + &m, + &ColumnRules::default(), + &SystemColumns::default(), + ) + .expect("plan builds"); assert!(plan.insert_sql.contains("INSERT INTO `default`.`foo`")); assert!(plan.insert_sql.contains("`id`")); assert!(plan.insert_sql.contains("`name`")); @@ -2320,7 +2389,7 @@ mod tests { // numeric-shaped default: the plan drill `numeric(38,0)` → `Int128` m.columns[0].target_type = "Decimal(38, 0)".into(); let rules = col_rules(&[("id", "Int128")]); - let plan = TablePlan::build(alloc, &rel, &m, &rules).unwrap(); + let plan = TablePlan::build(alloc, &rel, &m, &rules, &SystemColumns::default()).unwrap(); assert_eq!(plan.columns[0].type_repr, "Int128"); // scale-0 decimal wire keeps the numeric text→scaled encode path assert_eq!( @@ -2341,7 +2410,7 @@ mod tests { // Operator-renamed CH column: override still keys on source attname m.columns[1].target_name = "label".into(); let rules = col_rules(&[("name", "String")]); - let plan = TablePlan::build(alloc, &rel, &m, &rules).unwrap(); + let plan = TablePlan::build(alloc, &rel, &m, &rules, &SystemColumns::default()).unwrap(); assert_eq!(plan.columns[1].name, "label"); assert_eq!(plan.columns[1].type_repr, "String"); } @@ -2354,7 +2423,7 @@ mod tests { // encode_value writes int4 as 4 LE bytes; no textualization exists, // so Int32 → String must fall back rather than poison the batcher let rules = col_rules(&[("id", "String")]); - let plan = TablePlan::build(alloc, &rel, &m, &rules).unwrap(); + let plan = TablePlan::build(alloc, &rel, &m, &rules, &SystemColumns::default()).unwrap(); assert_eq!(plan.columns[0].type_repr, "Int32"); } @@ -2393,7 +2462,14 @@ mod tests { let alloc = Allocator::stdlib(); let rel = mk_rel(); let m = mk_mapping(); - let plan = TablePlan::build(alloc, &rel, &m, &ColumnRules::default()).expect("plan builds"); + let plan = TablePlan::build( + alloc, + &rel, + &m, + &ColumnRules::default(), + &SystemColumns::default(), + ) + .expect("plan builds"); assert!(plan.insert_sql.contains("`_is_deleted`")); let mut enc = TableEncoder::new(plan).unwrap(); enc.append_row(&committed(1, Some("a")), &m, OP_INSERT) @@ -2441,7 +2517,14 @@ mod tests { let rel = mk_rel(); let mut m = mk_mapping(); m.columns[1].target_type = "String".into(); - let plan = TablePlan::build(alloc, &rel, &m, &ColumnRules::default()).expect("plan builds"); + let plan = TablePlan::build( + alloc, + &rel, + &m, + &ColumnRules::default(), + &SystemColumns::default(), + ) + .expect("plan builds"); let mut enc = TableEncoder::new(plan).unwrap(); // Delete: non-key column absent from the key-only old image enc.append_row(&committed_delete(3), &m, OP_DELETE).unwrap(); @@ -2461,7 +2544,14 @@ mod tests { let alloc = Allocator::stdlib(); let rel = mk_rel(); let m = mk_mapping(); - let plan = TablePlan::build(alloc, &rel, &m, &ColumnRules::default()).expect("plan builds"); + let plan = TablePlan::build( + alloc, + &rel, + &m, + &ColumnRules::default(), + &SystemColumns::default(), + ) + .expect("plan builds"); let mut enc = TableEncoder::new(plan).unwrap(); enc.append_row(&committed_delete(3), &m, OP_DELETE).unwrap(); match &enc.buffers[1] { @@ -2475,7 +2565,14 @@ mod tests { let alloc = Allocator::stdlib(); let rel = mk_rel(); let m = mk_mapping(); - let plan = TablePlan::build(alloc, &rel, &m, &ColumnRules::default()).unwrap(); + let plan = TablePlan::build( + alloc, + &rel, + &m, + &ColumnRules::default(), + &SystemColumns::default(), + ) + .unwrap(); let mut enc = TableEncoder::new(plan).unwrap(); enc.append_row(&committed(7, Some("seven")), &m, OP_INSERT) .unwrap(); @@ -2767,6 +2864,118 @@ mod tests { } } + #[test] + fn config_system_columns_reach_config() { + let c = EmitterConfig::from_toml_str( + "[ch]\n\ + [system_columns]\n\ + lsn = \"_peerdb_version\"\n\ + is_deleted = false\n", + ) + .unwrap(); + assert_eq!(c.system_columns.lsn, "_peerdb_version"); + assert!(c.system_columns.is_deleted.is_none()); + assert!(!c.row_policy().soft_delete); + // Absent section keeps the defaults + let d = EmitterConfig::from_toml_str("[ch]\n").unwrap(); + assert_eq!(*d.system_columns, SystemColumns::default()); + } + + #[test] + fn table_block_renames_system_columns_for_its_relations() { + let c = EmitterConfig::from_toml_str( + "[ch]\n\ + [system_columns]\n\ + lsn = \"_v\"\n\ + [table.public.events]\n\ + lsn = \"_peerdb_version\"\n\ + is_deleted = false\n\ + [table.app.\"events_*\"]\n\ + match = \"glob\"\n\ + commit_ts = \"_peerdb_synced_at\"\n", + ) + .unwrap(); + assert_eq!(c.system_columns.lsn, "_v", "cluster-wide default stands"); + let entry = |rel: RelName| { + c.table_entries + .iter() + .find(|(r, _, _)| *r == rel) + .expect("entry") + }; + let (_, kind, rule) = entry(RelName::new("public", "events")); + assert_eq!(*kind, MatchKind::Exact); + assert_eq!(rule.system.lsn.as_deref(), Some("_peerdb_version")); + assert_eq!(rule.system.is_deleted.as_deref(), Some("")); + let (_, kind, rule) = entry(RelName::new("app", "events_*")); + assert_eq!(*kind, MatchKind::Glob); + assert_eq!(rule.system.commit_ts.as_deref(), Some("_peerdb_synced_at")); + } + + #[test] + fn table_block_rejects_rename_onto_another_system_column() { + assert!(EmitterConfig::from_toml_str("[ch]\n[table.public.t]\nlsn = \"_xid\"\n").is_err()); + assert!(EmitterConfig::from_toml_str("[ch]\n[table.public.t]\nis_deleted = 7\n").is_err()); + } + + #[test] + fn config_table_key_lists_parse_from_arrays() { + let c = EmitterConfig::from_toml_str( + "[ch]\n\ + [table.public.events]\n\ + order_by = [\"tenant\", \"id\"]\n\ + primary_key = [\"tenant\"]\n", + ) + .unwrap(); + let rel = RelName::new("public", "events"); + let (_, _, rule) = &c.table_entries[0]; + assert_eq!( + rule.order_by.as_deref(), + Some(["tenant".to_string(), "id".to_string()].as_slice()) + ); + assert_eq!( + rule.primary_key.as_deref(), + Some(["tenant".to_string()].as_slice()) + ); + // Key lists also apply to a columns-less opt-in block + assert!(!c.tables.contains_key(&rel)); + // Arrays only: a bare string is a config error, never a split list + for bad in [ + "order_by = 7", + "order_by = \"tenant, id\"", + "primary_key = \"tenant\"", + ] { + assert!( + EmitterConfig::from_toml_str(&format!("[ch]\n[table.public.t]\n{bad}\n")).is_err(), + "{bad}" + ); + } + } + + #[test] + fn plan_renames_synthetic_columns_and_can_drop_the_marker() { + let alloc = Allocator::stdlib(); + let rel = mk_rel(); + let m = mk_mapping(); + let sys = SystemColumns { + lsn: "_v".into(), + xid: "_x".into(), + commit_ts: "_at".into(), + is_deleted: None, + }; + let plan = + TablePlan::build(alloc, &rel, &m, &ColumnRules::default(), &sys).expect("plan builds"); + assert!( + plan.insert_sql + .ends_with("`_v`, `_x`, `_at`) FORMAT Native"), + "{}", + plan.insert_sql + ); + assert!(plan.synth_is_deleted.is_none()); + let enc = TableEncoder::new(plan).expect("encoder"); + // Mapped columns plus three synthetic, no marker buffer + assert_eq!(enc.buffers.len(), m.columns.len() + 3); + } + #[test] fn config_soft_delete_defaults_off_and_parses_on() { assert!(!EmitterConfig::default().soft_delete); diff --git a/src/emit/pipeline/batcher.rs b/src/emit/pipeline/batcher.rs index 992cc8e6..b8a731f7 100644 --- a/src/emit/pipeline/batcher.rs +++ b/src/emit/pipeline/batcher.rs @@ -74,8 +74,8 @@ pub struct ColMeta { pub struct BatchMeta { pub table_key: RelName, pub insert_sql: String, - /// Order matches `InsertBatch::buffers`: mapped columns then the four - /// synthetic (`_lsn`, `_xid`, `_commit_ts`, `_is_deleted`). + /// Order matches `InsertBatch::buffers`: mapped columns then the + /// synthetic ones (lsn, xid, commit_ts, delete marker when configured). pub columns: Vec, pub schema_epoch: u64, } @@ -90,11 +90,14 @@ impl BatchMeta { }); } for synth in [ - &plan.synth_lsn, - &plan.synth_xid, - &plan.synth_commit_ts, - &plan.synth_is_deleted, - ] { + Some(&plan.synth_lsn), + Some(&plan.synth_xid), + Some(&plan.synth_commit_ts), + plan.synth_is_deleted.as_ref(), + ] + .into_iter() + .flatten() + { columns.push(ColMeta { name: synth.name.clone(), type_repr: synth.type_repr.clone(), @@ -309,6 +312,7 @@ async fn handle_row( &row.rel, &row.route.mapping, &row.route.column_rules, + row.route.system_columns(), ) .map_err(|e| e.to_string())?; let meta = Arc::new(BatchMeta::from_plan(&plan, e.key().clone(), ctx.epoch)); @@ -480,7 +484,7 @@ mod tests { }], }), Arc::default(), - false, + Default::default(), ) } @@ -662,7 +666,7 @@ mod tests { }], }), Arc::new(rules.finish().0), - false, + Default::default(), ); msg_tx.send(BatcherMsg::Row(r)).await.expect("send row"); drop(msg_tx); diff --git a/src/emit/pipeline/bootstrap.rs b/src/emit/pipeline/bootstrap.rs index c53f9734..3288c35e 100644 --- a/src/emit/pipeline/bootstrap.rs +++ b/src/emit/pipeline/bootstrap.rs @@ -16,7 +16,7 @@ use crate::decode::heap_decoder::{ColumnValue, ToastPointer}; use crate::emit::ch_emitter::EmitterStats; use crate::emit::pipeline::ack::AckHandle; use crate::emit::pipeline::batcher::{BatcherMsg, RoutedRow}; -use crate::emit::route::RouteSnapshot; +use crate::emit::route::{RouteSnapshot, RowPolicy}; use crate::mapping::{MappingHandle, TableMapping}; use crate::schema::RelDescriptor; use crate::toast::{ @@ -46,7 +46,7 @@ pub async fn drain( stats: Arc, resolver: ToastResolver, mut deferred: DeferredSpool, - soft_delete: bool, + row_policy: RowPolicy, config: Option>, ) -> Result { // Routes frozen once per pass from the caller's config snapshot @@ -58,9 +58,10 @@ pub async fn drain( let rules = config .as_ref() .map_or_else(Arc::default, |rc| rc.column_rules.clone()); + let policy = row_policy.for_rel(config.as_deref(), name); ( name.clone(), - RouteSnapshot::freeze(Arc::new(mapping.clone()), rules, soft_delete), + RouteSnapshot::freeze(Arc::new(mapping.clone()), rules, policy), ) }) .collect(); @@ -565,7 +566,7 @@ mod tests { stats.clone(), ToastResolver::disabled(), mem_spool(), - false, + Default::default(), None, )); @@ -617,7 +618,7 @@ mod tests { stats.clone(), ToastResolver::disabled(), mem_spool(), - false, + Default::default(), None, )); @@ -667,7 +668,7 @@ mod tests { stats.clone(), resolver, mem_spool(), - false, + Default::default(), None, )); @@ -732,7 +733,7 @@ mod tests { ToastResolver::with_store(store, stats.clone()), // Threshold 0: deferred referrer rides a real spool file DeferredSpool::new(spool_tmp.path().join("bootstrap_deferred.bin"), 0), - false, + Default::default(), None, )); @@ -790,7 +791,7 @@ mod tests { stats.clone(), ToastResolver::with_store(store, stats.clone()), mem_spool(), - false, + Default::default(), None, )); // Wait for the referrer to defer, then unmap before walk EOF diff --git a/src/emit/pipeline/decode.rs b/src/emit/pipeline/decode.rs index d960040f..be76ccc7 100644 --- a/src/emit/pipeline/decode.rs +++ b/src/emit/pipeline/decode.rs @@ -17,7 +17,7 @@ use std::sync::atomic::Ordering; use tokio::sync::mpsc; use tokio::task::JoinHandle; -use crate::decode::heap_decoder::CommittedTuple; +use crate::decode::heap_decoder::{CommittedTuple, HeapOp}; use crate::emit::ch_emitter::EmitterStats; use crate::emit::pipeline::Fatal; use crate::emit::pipeline::ack::AckHandle; @@ -109,6 +109,12 @@ pub async fn decode_and_route( let Some(route) = envelope.route else { continue; }; + // No delete-marker column: a DELETE would land as a phantom insert of + // the old image, so drop it (append-only destination) + if route.drops_deletes() && matches!(envelope.described.decoded.op, HeapOp::Delete) { + ctx.stats.deletes_discarded.fetch_add(1, Ordering::Relaxed); + continue; + } let mut heap = envelope.described; let value_permit = detoast_heap(&mut heap, spool, &ref_maps, &ctx.resolver) .await @@ -193,3 +199,146 @@ pub fn spawn_pool( } handles } + +#[cfg(test)] +mod tests { + use super::*; + use crate::decode::heap_decoder::{ColumnValue, DecodedHeap, DecodedTuple, DescribedHeap}; + use crate::emit::route::{RouteSnapshot, RowPolicy}; + use crate::mapping::{ColumnMapping, SystemColumns, TableMapping, TableTarget}; + use crate::schema::{RelAttr, RelDescriptor, RelName, ReplIdent}; + use crate::toast::ToastResolver; + use walrus::pg::walparser::RelFileNode; + + const RFN: RelFileNode = RelFileNode { + spc_node: 1663, + db_node: 5, + rel_node: 16385, + }; + + fn rel() -> Arc { + Arc::new(RelDescriptor { + rfn: RFN, + oid: 16385, + toast_oid: 0, + namespace_oid: 2200, + rel_name: RelName::new("public", "t"), + kind: 'r', + persistence: 'p', + replident: ReplIdent::Default { pk_attnums: None }, + attributes: vec![RelAttr { + attnum: 1, + name: "id".into(), + type_oid: 23, + typmod: -1, + not_null: true, + dropped: false, + type_name: "int4".into(), + type_byval: true, + type_len: 4, + type_align: 'i', + type_storage: 'p', + missing_text: None, + }], + }) + } + + fn heap(op: HeapOp, route: Arc) -> RoutedHeap { + let tuple = Some(DecodedTuple { + columns: vec![Some(ColumnValue::Int4(1))], + partial: false, + }); + let (new, old) = match op { + HeapOp::Delete => (None, tuple), + _ => (tuple, None), + }; + RoutedHeap { + described: DescribedHeap { + decoded: DecodedHeap { + rfn: RFN, + xid: 7, + source_lsn: 0x1000, + op, + new, + old, + }, + descriptor: rel(), + descriptor_valid_from: 0x40, + }, + route: Some(route), + } + } + + fn route(system: SystemColumns) -> Arc { + RouteSnapshot::freeze( + Arc::new(TableMapping { + target: TableTarget::new("default", "t"), + columns: vec![ColumnMapping { + src_attnum: 1, + target_name: "id".into(), + target_type: "Int32".into(), + }], + }), + Arc::default(), + RowPolicy { + soft_delete: false, + system: Arc::new(system), + }, + ) + } + + /// Without a delete-marker column a DELETE has nowhere to land: it must + /// drop here, before the placed count the ack collector reconciles + #[tokio::test] + async fn deletes_drop_when_marker_disabled() { + let (msg_tx, mut msg_rx) = mpsc::channel(8); + let stats = Arc::new(EmitterStats::default()); + let ctx = DecodeCtx { + oracle: None, + msg_tx, + stats: stats.clone(), + resolver: ToastResolver::disabled(), + chunk_rows: 8, + }; + let no_marker = route(SystemColumns { + is_deleted: None, + ..SystemColumns::default() + }); + let routed = decode_and_route( + &ctx, + 0, + 0, + 0x2000, + vec![ + heap(HeapOp::Insert, no_marker.clone()), + heap(HeapOp::Delete, no_marker), + ], + Vec::new(), + None, + ) + .await + .expect("decode"); + assert_eq!(routed, 1, "insert routed, delete dropped"); + assert_eq!(stats.deletes_discarded.load(Ordering::Relaxed), 1); + match msg_rx.recv().await { + Some(BatcherMsg::Rows(chunk)) => assert_eq!(chunk.rows.len(), 1), + other => panic!("expected one row chunk, got {}", other.is_some()), + } + + // Default policy keeps the marker, so the DELETE rides through + let marked = route(SystemColumns::default()); + let routed = decode_and_route( + &ctx, + 1, + 0, + 0x3000, + vec![heap(HeapOp::Delete, marked)], + Vec::new(), + None, + ) + .await + .expect("decode"); + assert_eq!(routed, 1); + assert_eq!(stats.deletes_discarded.load(Ordering::Relaxed), 1); + } +} diff --git a/src/emit/pipeline/mod.rs b/src/emit/pipeline/mod.rs index 34a82c36..6447b524 100644 --- a/src/emit/pipeline/mod.rs +++ b/src/emit/pipeline/mod.rs @@ -269,7 +269,7 @@ impl PipelineConfig { retires, resume_floor, mapping, - emitter.soft_delete, + emitter.row_policy(), ); Ok(( diff --git a/src/emit/pipeline/plan_spool.rs b/src/emit/pipeline/plan_spool.rs index ee53f943..cba959da 100644 --- a/src/emit/pipeline/plan_spool.rs +++ b/src/emit/pipeline/plan_spool.rs @@ -591,7 +591,7 @@ mod tests { columns: Vec::new(), }), Arc::default(), - false, + Default::default(), ) } diff --git a/src/emit/pipeline/planner.rs b/src/emit/pipeline/planner.rs index 5b6388af..bc7e6990 100644 --- a/src/emit/pipeline/planner.rs +++ b/src/emit/pipeline/planner.rs @@ -327,7 +327,7 @@ mod tests { columns: Vec::new(), }), Arc::default(), - false, + Default::default(), ) } diff --git a/src/emit/pipeline/reorder.rs b/src/emit/pipeline/reorder.rs index bd69574d..3a60da46 100644 --- a/src/emit/pipeline/reorder.rs +++ b/src/emit/pipeline/reorder.rs @@ -47,7 +47,7 @@ use crate::emit::pipeline::batcher::BatcherMsg; use crate::emit::pipeline::decode::DecodeJob; use crate::emit::pipeline::plan_spool::{PlanItem, SealedPlan}; use crate::emit::pipeline::planner::{PlanRouteView, Planner, drain_reason}; -use crate::emit::route::{RouteSnapshot, RoutedHeap}; +use crate::emit::route::{RouteSnapshot, RoutedHeap, RowPolicy}; use crate::mapping::{MappingHandle, MappingSnapshot, TableMapping}; use crate::pos::{Floor, Monotone}; use crate::runtime_config::{ConfigEvent, TableRow}; @@ -122,8 +122,8 @@ pub struct ReorderSink { /// Shared routing map, snapshotted into `route_mapping` at route-state /// resets (this coordinator's own event applies are the fenced writers). mapping: MappingHandle, - /// Boot-only delete-retention policy, frozen into route snapshots. - soft_delete: bool, + /// Boot-only row-shape policy, frozen into route snapshots. + row_policy: RowPolicy, /// Byte cap per transaction plan spool file plan_disk_max: u64, /// Plan spool directory (the xact spill dir), cached at spawn so the @@ -162,7 +162,7 @@ impl ReorderSink { retires: RetireLedger, resume_floor: Arc>, mapping: MappingHandle, - soft_delete: bool, + row_policy: RowPolicy, ) -> Self { // subscribe() marks the current value seen, so a `ctl reload` // racing pipeline spawn would stay invisible to has_changed — @@ -204,7 +204,7 @@ impl ReorderSink { applied_opt_ins: HashSet::new(), pending_opt_ins: HashMap::new(), mapping, - soft_delete, + row_policy, route_mapping: None, route_config: None, @@ -851,7 +851,7 @@ impl ReorderSink { let mut view = ReorderRouteView::new( self.route_mapping.clone(), self.route_config.clone(), - self.soft_delete, + self.row_policy.clone(), self.applicator.as_mut(), self.stats.clone(), ); @@ -978,7 +978,7 @@ pub struct ReorderRouteView<'a> { /// Catalog fold above `mapping`; `None` value = locally dropped overlay: HashMap>, memo: HashMap>>, - soft_delete: bool, + row_policy: RowPolicy, applicator: Option<&'a mut DdlApplicator>, stats: Arc, } @@ -987,7 +987,7 @@ impl<'a> ReorderRouteView<'a> { pub fn new( mapping: Option, config: Option>, - soft_delete: bool, + row_policy: RowPolicy, applicator: Option<&'a mut DdlApplicator>, stats: Arc, ) -> Self { @@ -996,7 +996,7 @@ impl<'a> ReorderRouteView<'a> { config, overlay: HashMap::new(), memo: HashMap::new(), - soft_delete, + row_policy, applicator, stats, } @@ -1018,7 +1018,8 @@ impl PlanRouteView for ReorderRouteView<'_> { .config .as_ref() .map_or_else(Arc::default, |rc| rc.column_rules.clone()); - RouteSnapshot::freeze(Arc::new(m), rules, self.soft_delete) + let policy = self.row_policy.for_rel(self.config.as_deref(), rel_name); + RouteSnapshot::freeze(Arc::new(m), rules, policy) }); let result = if route.is_none() { self.stats @@ -1150,8 +1151,13 @@ mod tests { )])); let stats = Arc::new(EmitterStats::default()); - let mut view = - ReorderRouteView::new(Some(handle.snapshot().await), None, false, None, stats); + let mut view = ReorderRouteView::new( + Some(handle.snapshot().await), + None, + RowPolicy::default(), + None, + stats, + ); assert!(view.route_for(&heap_of(&planned)).is_some()); assert!(view.route_for(&heap_of(&added)).is_none()); diff --git a/src/emit/route.rs b/src/emit/route.rs index d0621eb4..a47e31fd 100644 --- a/src/emit/route.rs +++ b/src/emit/route.rs @@ -4,14 +4,41 @@ use std::sync::Arc; use crate::column_rules::ColumnRules; use crate::decode::heap_decoder::DescribedHeap; -use crate::mapping::{TableMapping, TableTarget}; +use crate::mapping::{SystemColumns, TableMapping, TableTarget}; +use crate::schema::RelName; + +/// Row-shape knobs, cloned into every frozen route so execution never reads +/// live config. `soft_delete` is boot-only; the system columns are the +/// cluster-wide `[system_columns]` set until [`Self::for_rel`] layers a +/// per-relation rename over it +#[derive(Debug, Clone, Default)] +pub struct RowPolicy { + /// CH-side delete retention policy (the delete marker stays queryable) + pub soft_delete: bool, + /// Names + presence of the columns walshadow appends per row + pub system: Arc, +} + +impl RowPolicy { + /// Policy for one relation: a `[table.*]` block or `config_table` row can + /// rename its system columns or drop its delete marker. Without a config + /// snapshot, or with no rename for this relation, the cluster-wide set + /// stands + pub fn for_rel(&self, config: Option<&crate::config::ResolvedConfig>, rel: &RelName) -> Self { + let Some(rc) = config else { + return self.clone(); + }; + Self { + system: rc.rules.settings(rel).system_columns(&self.system), + ..self.clone() + } + } +} #[derive(Debug)] pub struct RowEncodingSnapshot { pub destination: TableTarget, - /// CH-side delete retention policy (`_is_deleted` stays queryable); - /// boot-only knob, snapshotted so execution never reads live config - pub soft_delete: bool, + pub policy: RowPolicy, } /// Frozen route for one relation over one WAL interval @@ -27,11 +54,11 @@ impl RouteSnapshot { pub fn freeze( mapping: Arc, column_rules: Arc, - soft_delete: bool, + policy: RowPolicy, ) -> Arc { let encoding = Arc::new(RowEncodingSnapshot { destination: mapping.target.clone(), - soft_delete, + policy, }); Arc::new(Self { mapping, @@ -39,6 +66,15 @@ impl RouteSnapshot { encoding, }) } + + pub fn system_columns(&self) -> &SystemColumns { + &self.encoding.policy.system + } + + /// DELETE carries no marker column to set, so those rows are dropped + pub fn drops_deletes(&self) -> bool { + self.encoding.policy.system.is_deleted.is_none() + } } /// Described heap plus its resolved route. `route = None` means the relation diff --git a/src/mapping.rs b/src/mapping.rs index b57eae35..8f7deab3 100644 --- a/src/mapping.rs +++ b/src/mapping.rs @@ -5,6 +5,7 @@ use std::sync::Arc; use crate::catalog::type_bridge::{self, ResolvedColumn}; use crate::column_rules::{ColumnRule, ColumnRules}; use crate::schema::{RelAttr, RelDescriptor, RelName, SchemaDiff, replident_key_attnums}; +use crate::table_rules::set_if; use ahash::HashMap; use tokio::sync::RwLock; @@ -117,6 +118,191 @@ pub struct ToastConfig { pub mode: ToastMode, } +/// Names of the columns walshadow appends to every replicated CH table, and +/// whether the delete marker exists at all. `[system_columns]` sets the +/// cluster-wide default; a `[table.*]` block or `config_table` row renames per +/// relation ([`SystemColumnNames`]). A rename does not ALTER tables already +/// created, and every INSERT column list rebuilds from these, so an operator +/// renaming after first sync must ALTER the destination themselves. TOAST +/// mirror tables are walshadow-internal and keep their own fixed names. +#[derive(Debug, Clone, PartialEq, Eq, serde::Deserialize)] +#[serde(default)] +pub struct SystemColumns { + pub lsn: String, + pub xid: String, + pub commit_ts: String, + /// `None` leaves the delete marker out of both DDL and INSERT. DELETE rows + /// then have nowhere to land, so they are discarded (counted in + /// `emitter_deletes_discarded`) — for append-only destinations + #[serde(deserialize_with = "de_delete_marker")] + pub is_deleted: Option, +} + +impl Default for SystemColumns { + fn default() -> Self { + Self { + lsn: "_lsn".into(), + xid: "_xid".into(), + commit_ts: "_commit_ts".into(), + is_deleted: Some("_is_deleted".into()), + } + } +} + +/// `is_deleted = false` (or an empty name) drops the delete marker +fn de_delete_marker<'de, D: serde::Deserializer<'de>>(d: D) -> Result, D::Error> { + use serde::Deserialize; + match toml::Value::deserialize(d)? { + toml::Value::Boolean(false) => Ok(None), + toml::Value::Boolean(true) => Ok(SystemColumns::default().is_deleted), + toml::Value::String(name) => Ok((!name.is_empty()).then_some(name)), + v => Err(serde::de::Error::custom(format!( + "expected a string or false, got {}", + v.type_str() + ))), + } +} + +impl SystemColumns { + /// A blank rename or a name colliding with another system column yields a + /// CH table walshadow cannot INSERT into, so both are config errors + pub fn validate(&self) -> Result<(), String> { + for (key, name) in [ + ("lsn", &self.lsn), + ("xid", &self.xid), + ("commit_ts", &self.commit_ts), + ] { + if name.is_empty() { + return Err(format!("system_columns.{key}: name must not be empty")); + } + } + let names = self.names(); + for (i, a) in names.iter().enumerate() { + if names[i + 1..].contains(a) { + return Err(format!("system_columns: `{a}` named twice")); + } + } + Ok(()) + } + + /// Every system column present, in the order the emitter appends them + pub fn names(&self) -> Vec<&str> { + let mut v = vec![ + self.lsn.as_str(), + self.xid.as_str(), + self.commit_ts.as_str(), + ]; + v.extend(self.is_deleted.as_deref()); + v + } + + /// Apply per-relation renames. Total rather than fallible: a rename onto a + /// name another system column holds is skipped, so no merge of layers can + /// render a CH table with two identically named columns. Entries are + /// name-checked where they are parsed, so a skip here means two layers + /// collided + pub fn renamed(&self, names: &SystemColumnNames) -> Self { + let mut out = self.clone(); + for (i, over) in [&names.lsn, &names.xid, &names.commit_ts] + .into_iter() + .enumerate() + { + let Some(name) = over.as_deref().filter(|n| !n.is_empty()) else { + continue; + }; + if out + .names() + .iter() + .enumerate() + .any(|(j, n)| j != i && *n == name) + { + continue; + } + let field = match i { + 0 => &mut out.lsn, + 1 => &mut out.xid, + _ => &mut out.commit_ts, + }; + *field = name.into(); + } + match names.is_deleted.as_deref() { + Some("") => out.is_deleted = None, + Some(name) if !out.names()[..3].contains(&name) => out.is_deleted = Some(name.into()), + _ => {} + } + out + } +} + +/// Per-relation renames layered over [`SystemColumns`]. `None` inherits the +/// cluster-wide name. An empty string inherits too — a blank `text` overlay +/// column must not blank a column name — except on `is_deleted`, where it drops +/// the marker (and with it every DELETE row), like `[system_columns] +/// is_deleted = false` +#[derive(Debug, Clone, Default, PartialEq, Eq)] +pub struct SystemColumnNames { + pub lsn: Option, + pub xid: Option, + pub commit_ts: Option, + pub is_deleted: Option, +} + +impl SystemColumnNames { + pub fn is_empty(&self) -> bool { + self.lsn.is_none() + && self.xid.is_none() + && self.commit_ts.is_none() + && self.is_deleted.is_none() + } + + pub fn overlay(&mut self, other: &Self) { + set_if(&mut self.lsn, &other.lsn); + set_if(&mut self.xid, &other.xid); + set_if(&mut self.commit_ts, &other.commit_ts); + set_if(&mut self.is_deleted, &other.is_deleted); + } + + /// Reject a set naming one column twice, or renaming onto a name another + /// system column holds by default: both yield a CH table walshadow cannot + /// INSERT into. `ctx` names the config entry + pub fn validate(&self, ctx: &str) -> Result<(), String> { + let renamed = SystemColumns::default().renamed(self); + let landed = renamed.names(); + let wanted = [ + self.lsn.as_deref(), + self.xid.as_deref(), + self.commit_ts.as_deref(), + self.is_deleted.as_deref(), + ]; + for (i, want) in wanted.into_iter().enumerate() { + let Some(want) = want.filter(|n| !n.is_empty()) else { + continue; + }; + if landed.get(i) != Some(&want) { + return Err(format!("{ctx}: `{want}` names two system columns")); + } + } + Ok(()) + } +} + +/// Per-relation delete marker: `false` (or a blank name) drops it for this +/// relation alone, `true` restores the cluster-wide name +pub(crate) fn de_marker_override<'de, D: serde::Deserializer<'de>>( + d: D, +) -> Result, D::Error> { + use serde::Deserialize; + match toml::Value::deserialize(d)? { + toml::Value::Boolean(false) => Ok(Some(String::new())), + toml::Value::Boolean(true) => Ok(SystemColumns::default().is_deleted), + toml::Value::String(name) => Ok(Some(name)), + v => Err(serde::de::Error::custom(format!( + "expected a string or false, got {}", + v.type_str() + ))), + } +} + /// Immutable routing-map version. Planners snapshot one per transaction so a /// concurrent republish can't split a transaction across mapping versions pub type MappingSnapshot = Arc>; @@ -243,6 +429,93 @@ fn quote_ident(name: &str) -> String { mod tests { use super::*; + /// Parse a `[system_columns]` body the way `ConfigDocument` does + fn system_columns(body: &str) -> Result { + let sys: SystemColumns = toml::from_str(body).map_err(crate::toml_de::message)?; + sys.validate()?; + Ok(sys) + } + + #[test] + fn system_columns_default_when_section_absent() { + let sys = system_columns("").unwrap(); + assert_eq!(sys, SystemColumns::default()); + assert_eq!(sys.names(), ["_lsn", "_xid", "_commit_ts", "_is_deleted"]); + } + + #[test] + fn system_columns_rename_and_disable_marker() { + let sys = system_columns("lsn = \"_peerdb_version\"\ncommit_ts = \"_peerdb_synced_at\"\n") + .unwrap(); + assert_eq!(sys.lsn, "_peerdb_version"); + assert_eq!(sys.commit_ts, "_peerdb_synced_at"); + assert_eq!(sys.xid, "_xid"); + assert_eq!(sys.is_deleted.as_deref(), Some("_is_deleted")); + + let off = system_columns("is_deleted = false\n").unwrap(); + assert!(off.is_deleted.is_none()); + assert_eq!(off.names(), ["_lsn", "_xid", "_commit_ts"]); + let renamed = system_columns("is_deleted = \"gone\"\n").unwrap(); + assert_eq!(renamed.is_deleted.as_deref(), Some("gone")); + } + + #[test] + fn system_columns_reject_empty_and_colliding_names() { + // Both yield a CH table walshadow cannot INSERT into + assert!(system_columns("lsn = \"\"\n").is_err()); + assert!(system_columns("xid = \"_lsn\"\n").is_err()); + assert!(system_columns("lsn = 7\n").is_err()); + } + + #[test] + fn per_relation_rename_inherits_unnamed_columns() { + let names = SystemColumnNames { + lsn: Some("_peerdb_version".into()), + is_deleted: Some(String::new()), + ..SystemColumnNames::default() + }; + names.validate("table.public.t").unwrap(); + let sys = SystemColumns::default().renamed(&names); + assert_eq!(sys.lsn, "_peerdb_version"); + assert_eq!(sys.xid, "_xid", "unnamed column inherits"); + assert!(sys.is_deleted.is_none(), "blank marker drops it"); + } + + #[test] + fn per_relation_blank_name_inherits() { + let sys = SystemColumns::default().renamed(&SystemColumnNames { + lsn: Some(String::new()), + ..SystemColumnNames::default() + }); + assert_eq!( + sys.lsn, "_lsn", + "a blank overlay column must not blank a name" + ); + } + + #[test] + fn per_relation_rename_onto_another_system_column_rejected() { + let onto = SystemColumnNames { + lsn: Some("_xid".into()), + ..SystemColumnNames::default() + }; + assert!(onto.validate("t").is_err()); + assert!( + SystemColumnNames { + lsn: Some("_v".into()), + xid: Some("_v".into()), + ..SystemColumnNames::default() + } + .validate("t") + .is_err() + ); + // Layers can still collide after their own checks pass; the merge + // keeps the name it holds rather than rendering two columns as one + let sys = SystemColumns::default().renamed(&onto); + assert_eq!(sys.names().len(), 4); + assert_eq!(sys.lsn, "_lsn"); + } + fn one_table() -> (RelName, HashMap) { let rel = RelName::new("public", "t"); let map = HashMap::from_iter([( diff --git a/src/ops/metrics.rs b/src/ops/metrics.rs index bc816e81..82029b15 100644 --- a/src/ops/metrics.rs +++ b/src/ops/metrics.rs @@ -126,6 +126,8 @@ pub struct MetricsSnapshot { pub emitter_blocks_total: u64, pub emitter_xacts_total: u64, pub emitter_unsupported_relations: u64, + /// DELETE rows dropped for want of a delete-marker column + pub emitter_deletes_discarded: u64, /// Forward-declared per-table opt-ins (`config_table.replicate=true`) /// awaiting their `CREATE TABLE`. pub config_pending_decl_rels: u64, @@ -721,6 +723,12 @@ pub fn render(snap: &MetricsSnapshot) -> String { "counter", snap.emitter_unsupported_relations, ), + ( + "walshadow_emitter_deletes_discarded_total", + "DELETE rows dropped because is_deleted = false leaves no marker column.", + "counter", + snap.emitter_deletes_discarded, + ), ( "walshadow_pump_queue_depth", "Records buffered between the WAL pump and the queueing worker.", diff --git a/src/runtime_config.rs b/src/runtime_config.rs index 8bb52b2d..b43b56d0 100644 --- a/src/runtime_config.rs +++ b/src/runtime_config.rs @@ -83,6 +83,14 @@ pub struct TableRow { /// parses at dispatch in [`crate::backfill::opt_in`], validate-late like every /// overlay value); absent / `none` streams from opt-in LSN. pub initial_load: Option, + /// CH `ORDER BY` for this table's `CREATE`. NULL keeps startup TOML value; + /// empty array uses replica identity + pub order_by: Option>, + /// CH `PRIMARY KEY` sparse-index prefix. Ignored unless prefix of `ORDER BY` + pub primary_key: Option>, + /// Per-relation renames of the columns walshadow appends. NULL on a column + /// inherits `[system_columns]`; `is_deleted = ''` drops the marker + pub system: crate::mapping::SystemColumnNames, pub match_kind: Option, } @@ -297,6 +305,19 @@ fn field_string(rel: &RelDescriptor, cols: &[Option], name: &str) - } } +fn field_string_array( + rel: &RelDescriptor, + cols: &[Option], + name: &str, +) -> Option> { + match column(rel, cols, name)? { + ColumnValue::PgPending { type_oid, raw } if *type_oid == crate::schema::TEXTARRAYOID => { + crate::decode::codecs::decode_text_array(raw) + } + _ => None, + } +} + /// Interpret a config-table heap write into a [`ConfigEvent`]. `rel` must /// describe the same relation `decoded` targets. `None` when the write carries /// no usable image or the row key is missing. @@ -356,6 +377,14 @@ pub fn interpret( target_table: field_string(rel, &cols, "target_table"), replicate: field_bool(rel, &cols, "replicate"), initial_load: field_string(rel, &cols, "initial_load"), + order_by: field_string_array(rel, &cols, "order_by"), + primary_key: field_string_array(rel, &cols, "primary_key"), + system: crate::mapping::SystemColumnNames { + lsn: field_string(rel, &cols, "lsn"), + xid: field_string(rel, &cols, "xid"), + commit_ts: field_string(rel, &cols, "commit_ts"), + is_deleted: field_string(rel, &cols, "is_deleted"), + }, match_kind: field_string(rel, &cols, "match"), }, }) @@ -423,6 +452,13 @@ mod tests { } } + fn text_array(values: &[&str]) -> ColumnValue { + ColumnValue::PgPending { + type_oid: crate::schema::TEXTARRAYOID, + raw: crate::decode::codecs::text_array_body(values), + } + } + fn heap( op: HeapOp, new: Option>>, @@ -595,6 +631,8 @@ mod tests { attr(4, "target_table", 25), attr(5, "replicate", 16), attr(6, "initial_load", 25), + attr(7, "order_by", crate::schema::TEXTARRAYOID), + attr(8, "primary_key", crate::schema::TEXTARRAYOID), ], ); let new = vec![ @@ -604,6 +642,8 @@ mod tests { Some(ColumnValue::Text("events".into())), Some(ColumnValue::Bool(true)), Some(ColumnValue::Text("copy".into())), + Some(text_array(&["tenant", "id"])), + Some(text_array(&["tenant"])), ]; match interpret( ConfigTableKind::Table, @@ -618,6 +658,8 @@ mod tests { assert_eq!(row.target_table.as_deref(), Some("events")); assert_eq!(row.replicate, Some(true)); assert_eq!(row.initial_load.as_deref(), Some("copy")); + assert_eq!(row.order_by.unwrap(), ["tenant", "id"]); + assert_eq!(row.primary_key.unwrap(), ["tenant"]); } other => panic!("expected TableUpserted, got {other:?}"), } @@ -664,6 +706,8 @@ mod tests { attr(2, "relname", 25), attr(3, "match", 25), attr(4, "replicate", 16), + attr(5, "lsn", 25), + attr(6, "is_deleted", 25), ], ); let new = vec![ @@ -671,6 +715,8 @@ mod tests { Some(ColumnValue::Text("events_.*".into())), Some(ColumnValue::Text("regex".into())), Some(ColumnValue::Bool(true)), + Some(ColumnValue::Text("_peerdb_version".into())), + Some(ColumnValue::Text(String::new())), ]; match interpret( ConfigTableKind::Table, @@ -683,6 +729,9 @@ mod tests { assert_eq!(rel, RelName::new("app", "events_.*")); assert!(row.is_pattern()); assert_eq!(row.replicate, Some(true)); + assert_eq!(row.system.lsn.as_deref(), Some("_peerdb_version")); + assert_eq!(row.system.is_deleted.as_deref(), Some("")); + assert_eq!(row.system.xid, None, "absent column inherits"); } other => panic!("expected TableUpserted, got {other:?}"), } diff --git a/src/schema.rs b/src/schema.rs index 8798b498..968eac23 100644 --- a/src/schema.rs +++ b/src/schema.rs @@ -24,6 +24,7 @@ pub const CIDROID: u32 = 650; pub const FLOAT4OID: u32 = 700; pub const FLOAT8OID: u32 = 701; pub const INETOID: u32 = 869; +pub const TEXTARRAYOID: u32 = 1009; pub const BPCHAROID: u32 = 1042; pub const VARCHAROID: u32 = 1043; pub const DATEOID: u32 = 1082; diff --git a/src/table_rules.rs b/src/table_rules.rs index bfd9564a..9e9677ab 100644 --- a/src/table_rules.rs +++ b/src/table_rules.rs @@ -1,8 +1,11 @@ //! Resolves table settings from TOML and runtime config +use std::sync::Arc; + use globset::{Glob, GlobMatcher}; use regex_automata::meta::Regex; +use crate::mapping::{SystemColumnNames, SystemColumns}; use crate::runtime_config::TableRow; use crate::schema::RelName; @@ -78,27 +81,52 @@ impl NamePattern { #[derive(Debug, Clone, Default, PartialEq, Eq)] pub struct TableRule { + /// Per-relation renames of the columns walshadow appends, layered over + /// `[system_columns]` at `CREATE TABLE` and at every route freeze + pub system: SystemColumnNames, pub target_database: Option, pub target_table: Option, pub replicate: Option, pub initial_load: Option, + /// CH `ORDER BY` at `CREATE TABLE`, destination column names. An empty + /// list derives the key from replica identity + pub order_by: Option>, + /// CH `PRIMARY KEY`: the sparse-index prefix CH indexes, not a uniqueness + /// constraint. CH requires a prefix of the effective `ORDER BY` + pub primary_key: Option>, } impl TableRule { pub fn from_row(row: &TableRow) -> Self { Self { + system: row.system.clone(), target_database: row.target_database.clone(), target_table: row.target_table.clone(), replicate: row.replicate, initial_load: row.initial_load.clone(), + order_by: row.order_by.clone(), + primary_key: row.primary_key.clone(), } } + /// System columns this rule resolves to, `default` renamed where the rule + /// names something. An untouched rule hands back the cluster-wide set, so + /// a route freeze costs an `Arc` clone rather than a copy + pub fn system_columns(&self, default: &Arc) -> Arc { + if self.system.is_empty() { + return default.clone(); + } + Arc::new(default.renamed(&self.system)) + } + pub fn overlay(&mut self, other: &Self) { + self.system.overlay(&other.system); set_if(&mut self.target_database, &other.target_database); set_if(&mut self.target_table, &other.target_table); set_if(&mut self.replicate, &other.replicate); set_if(&mut self.initial_load, &other.initial_load); + set_if(&mut self.order_by, &other.order_by); + set_if(&mut self.primary_key, &other.primary_key); } } @@ -231,6 +259,11 @@ impl TableRulesBuilder { } pub fn add(&mut self, key: &RelName, kind: MatchKind, rule: TableRule) { + if let Err(e) = rule.system.validate(&format!("table {key}")) { + tracing::warn!(target: "walshadow::config", error = %e, "table entry rejected"); + self.rejections += 1; + return; + } match RelMatcher::compile(key, kind) { Ok(matcher) => self.rules.push((self.layer, matcher, rule)), Err(e) => { diff --git a/src/xact/xact_buffer.rs b/src/xact/xact_buffer.rs index 2dcc15df..a4c4b3f8 100644 --- a/src/xact/xact_buffer.rs +++ b/src/xact/xact_buffer.rs @@ -1588,7 +1588,9 @@ impl MergeSource { enum MergeItem { Heap(Box), - Event(DrainEntry), + /// Boxed: a config row carries every per-relation setting, so the variant + /// dwarfs the heap pointer the merge yields per row + Event(Box), } /// Lazy k-way merge over per-xid sources + event queues, `source_lsn` ASC. @@ -1698,7 +1700,7 @@ impl MergedDrain { match pick { Pick::Event(i) => { let (_lsn, ev) = self.events[i].pop_front().expect("just peeked head"); - return Ok(Some(MergeItem::Event(ev))); + return Ok(Some(MergeItem::Event(Box::new(ev)))); } Pick::Data(i) => { let entry = self.sources[i] @@ -2245,7 +2247,7 @@ impl CommittedDrain { Some(MergeItem::Event(event)) => ordered_events.push(OrderedEvent { heap_idx: heaps.len(), row_idx: m.rows.len(), - event, + event: *event, }), Some(MergeItem::Heap(h)) => { if h.decoded.op == HeapOp::Truncate { diff --git a/tests/bootstrap_pipeline_ch.rs b/tests/bootstrap_pipeline_ch.rs index b4b2b834..e77741d7 100644 --- a/tests/bootstrap_pipeline_ch.rs +++ b/tests/bootstrap_pipeline_ch.rs @@ -181,7 +181,7 @@ async fn bootstrap_tail_fans_out_n2() { std::env::temp_dir().join("ws-bootstrap-ch-unused.bin"), walshadow::spool::DEFERRED_SPOOL_MEM_MAX, ), - false, + Default::default(), None, )); let outcome = drain.await.expect("drain join").expect("drain ok"); diff --git a/tests/common/inproc_harness.rs b/tests/common/inproc_harness.rs index 8bbdd86c..ef4c98f2 100644 --- a/tests/common/inproc_harness.rs +++ b/tests/common/inproc_harness.rs @@ -850,6 +850,7 @@ async fn build_pipeline_inner( &config_rx.borrow(), emitter_cfg.database.clone(), emitter_cfg.soft_delete, + emitter_cfg.system_columns.clone(), emitter_cfg.replicate_all, emitter_cfg.runtime_config_schema.clone(), ); diff --git a/tests/emitter_budget_flush.rs b/tests/emitter_budget_flush.rs index ce194208..cbca3c7f 100644 --- a/tests/emitter_budget_flush.rs +++ b/tests/emitter_budget_flush.rs @@ -162,7 +162,11 @@ async fn budget_trips_seal_complete_inserts() { .expect("spawn tail"); let rel = rel_descriptor(); - let route = walshadow::emit::route::RouteSnapshot::freeze(mapping(), Arc::default(), false); + let route = walshadow::emit::route::RouteSnapshot::freeze( + mapping(), + Arc::default(), + Default::default(), + ); const N: i32 = 5; let commit_lsn = 0xC0FFEE; ack.register(0, commit_lsn); diff --git a/tests/emitter_native_types.rs b/tests/emitter_native_types.rs index 5aef34e5..6ce29830 100644 --- a/tests/emitter_native_types.rs +++ b/tests/emitter_native_types.rs @@ -163,7 +163,11 @@ async fn native_numeric_time_timetz_round_trip() { .send(BatcherMsg::Row(RoutedRow { seq: 0, rel, - route: walshadow::emit::route::RouteSnapshot::freeze(mapping, Arc::default(), false), + route: walshadow::emit::route::RouteSnapshot::freeze( + mapping, + Arc::default(), + Default::default(), + ), committed: tuple, value_permit: None, })) diff --git a/tests/emitter_tls.rs b/tests/emitter_tls.rs index 47b8fdf5..be7d039d 100644 --- a/tests/emitter_tls.rs +++ b/tests/emitter_tls.rs @@ -380,7 +380,11 @@ async fn emitter_tls_round_trip() { .send(BatcherMsg::Row(RoutedRow { seq: 0, rel, - route: walshadow::emit::route::RouteSnapshot::freeze(mapping, Arc::default(), false), + route: walshadow::emit::route::RouteSnapshot::freeze( + mapping, + Arc::default(), + Default::default(), + ), committed: tuple, value_permit: None, })) diff --git a/tests/runtime_config_e2e.rs b/tests/runtime_config_e2e.rs index 58d0da4b..05d6d431 100644 --- a/tests/runtime_config_e2e.rs +++ b/tests/runtime_config_e2e.rs @@ -63,8 +63,22 @@ //! //! 9. `pattern_row_scopes_tables_by_glob` //! * Glob rules include matching tables and exclude guarded tables +//! 10. `opt_in_row_pins_order_by_and_primary_key` +//! * `config_table` row opts a table in and names `order_by` / +//! `primary_key` in the same row. +//! * Expect: the auto-created CH table keys on the operator's columns, +//! not the declared PK order, with the index prefix they asked for +//! (plans/config.md §Destination shape). +//! +//! 10. `pattern_row_shapes_auto_created_tables` +//! * `config_table` row with `match = 'glob'` names system columns and +//! the sort key for `app.events_*` before those tables exist. +//! * Expect: the auto-created CH table carries the renamed LSN column, +//! no delete marker, and the operator sort key, while a relation the +//! pattern misses keeps the cluster-wide names. //! //! Source-side `config_*` install runs the real `sql/runtime_config_install.sql` + //! inside the bootstrap schema dump, so the drills double as install-script //! coverage (psql `\if` default-schema guard included). @@ -1066,3 +1080,185 @@ async fn pattern_row_scopes_tables_by_glob() { "excluded / unmatched relations must not create" ); } + +/// Drill 10: a `config_table` row carries the sort key of the very table its +/// `replicate = true` creates, so the CREATE cannot fall back to the PK. +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn opt_in_row_pins_order_by_and_primary_key() { + if !fx::pg_available() || !fx::pg_basebackup_available() || !fx::clickhouse_available() { + eprintln!("skip: missing initdb / pg_basebackup / clickhouse"); + return; + } + + let slot = fx::Ports::alloc(); + let tmp = tempfile::tempdir().unwrap(); + let schema_sql = format!( + "{INSTALL_SQL}\n\ + CREATE SCHEMA app;\n\ + CREATE TABLE app.keyed (id bigint, tenant bigint, body text, \ + PRIMARY KEY (id, tenant));\n" + ); + let ( + fx::BootstrappedClusters { + source, + shadow, + shadow_filter_dir, + }, + shadow_stream_state, + ) = fx::bootstrap_clusters(&tmp, &schema_sql, slot.source, slot.shadow, slot.walsender).await; + let _src_stop = fx::StopOnDrop { sh: &source }; + let _shd_stop = fx::StopOnDrop { sh: &shadow }; + + let ch_tmp = tempfile::tempdir().unwrap(); + let ch = fx::ChServer::spawn(ch_tmp, slot.ch_tcp, slot.ch_http).expect("spawn ch"); + ch.query("CREATE DATABASE IF NOT EXISTS walshadow_test") + .expect("create db"); + + // replicate_all off so the opt-in row is what creates the CH table: + // auto-create fires on first sight of a relation, before any config row + // for it can arrive, and walshadow never rekeys a table CH already holds + let mut pipeline = fx::build_pipeline_with( + fx::BuildPipelineArgs { + tmp: &tmp, + source: &source, + shadow: &shadow, + shadow_filter_dir: &shadow_filter_dir, + shadow_stream_state, + ch_database: "walshadow_test", + ch_tcp_port: slot.ch_tcp, + mappings: vec![], + app_name: "walshadow-config-order-by", + ddl: Some(overlay_ddl_args()), + }, + |cfg| cfg.replicate_all = false, + ) + .await; + + let driver = fx::spawn_workload( + &source, + vec![ + "INSERT INTO walshadow.config_table \ + (namespace, relname, replicate, order_by, primary_key) \ + VALUES ('app', 'keyed', true, ARRAY['tenant', 'id'], ARRAY['tenant'])" + .into(), + "INSERT INTO app.keyed (id, tenant, body) VALUES (1, 7, 'keyed')".into(), + "SELECT pg_switch_wal()".into(), + ], + ); + + let shipped = fx::pump_segments(&mut pipeline, 1, Duration::from_secs(45)).await; + let _ = driver.join(); + assert!(shipped >= 1, "no segments shipped in 45s"); + + let target = pipeline.stream.dispatched_lsn(); + let observed = shadow + .wait_for_replay(target, Duration::from_secs(30)) + .expect("shadow replay"); + assert!(observed >= target); + pipeline.shutdown().await.expect("pipeline drains clean"); + + let ddl = ch + .query("SHOW CREATE TABLE walshadow_test.keyed") + .expect("show create"); + assert!(ddl.contains("ORDER BY (tenant, id)"), "{ddl}"); + assert!(ddl.contains("PRIMARY KEY (tenant)"), "{ddl}"); + + let n = ch + .query("SELECT count() FROM walshadow_test.keyed FINAL WHERE _is_deleted = 0") + .expect("ch count"); + assert_eq!(n, "1", "post-opt-in insert must reach CH"); +} + +/// Drill 10: a `match = 'glob'` row shapes tables that do not exist yet. +/// `replicate_all` creates a relation the first time it is seen, so a literal +/// `config_table` row can never beat the CREATE — the pattern row, committed +/// before the source `CREATE TABLE`, is the only way to name the system +/// columns of an auto-created table (plans/config.md §Destination shape). +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn pattern_row_shapes_auto_created_tables() { + if !fx::pg_available() || !fx::pg_basebackup_available() || !fx::clickhouse_available() { + eprintln!("skip: missing initdb / pg_basebackup / clickhouse"); + return; + } + + let slot = fx::Ports::alloc(); + let tmp = tempfile::tempdir().unwrap(); + let schema_sql = format!("{INSTALL_SQL}\nCREATE SCHEMA app;\n"); + let ( + fx::BootstrappedClusters { + source, + shadow, + shadow_filter_dir, + }, + shadow_stream_state, + ) = fx::bootstrap_clusters(&tmp, &schema_sql, slot.source, slot.shadow, slot.walsender).await; + let _src_stop = fx::StopOnDrop { sh: &source }; + let _shd_stop = fx::StopOnDrop { sh: &shadow }; + + let ch_tmp = tempfile::tempdir().unwrap(); + let ch = fx::ChServer::spawn(ch_tmp, slot.ch_tcp, slot.ch_http).expect("spawn ch"); + ch.query("CREATE DATABASE IF NOT EXISTS walshadow_test") + .expect("create db"); + + let mut pipeline = fx::build_pipeline(fx::BuildPipelineArgs { + tmp: &tmp, + source: &source, + shadow: &shadow, + shadow_filter_dir: &shadow_filter_dir, + shadow_stream_state, + ch_database: "walshadow_test", + ch_tcp_port: slot.ch_tcp, + mappings: vec![], + app_name: "walshadow-config-glob-shape", + ddl: Some(overlay_ddl_args()), + }) + .await; + + let driver = fx::spawn_workload( + &source, + vec![ + "INSERT INTO walshadow.config_table \ + (namespace, relname, match, lsn, is_deleted, order_by) \ + VALUES ('app', 'events_*', 'glob', '_peerdb_version', '', \ + ARRAY['tenant', 'id'])" + .into(), + "CREATE TABLE app.events_2026 (id bigint, tenant bigint, body text, \ + PRIMARY KEY (id, tenant))" + .into(), + "CREATE TABLE app.orders (id bigint PRIMARY KEY, body text)".into(), + "INSERT INTO app.events_2026 (id, tenant, body) VALUES (1, 7, 'shaped')".into(), + "INSERT INTO app.orders (id, body) VALUES (1, 'unshaped')".into(), + "SELECT pg_switch_wal()".into(), + ], + ); + + let shipped = fx::pump_segments(&mut pipeline, 1, Duration::from_secs(45)).await; + let _ = driver.join(); + assert!(shipped >= 1, "no segments shipped in 45s"); + + let target = pipeline.stream.dispatched_lsn(); + let observed = shadow + .wait_for_replay(target, Duration::from_secs(30)) + .expect("shadow replay"); + assert!(observed >= target); + pipeline.shutdown().await.expect("pipeline drains clean"); + + let ddl = ch + .query("SHOW CREATE TABLE walshadow_test.events_2026") + .expect("show create"); + assert!(ddl.contains("`_peerdb_version` UInt64"), "{ddl}"); + assert!(!ddl.contains("_is_deleted"), "marker dropped: {ddl}"); + assert!(ddl.contains("ORDER BY (tenant, id)"), "{ddl}"); + + // A relation the pattern misses keeps the cluster-wide names + let other = ch + .query("SHOW CREATE TABLE walshadow_test.orders") + .expect("show create"); + assert!(other.contains("`_lsn` UInt64"), "{other}"); + assert!(other.contains("_is_deleted"), "{other}"); + + let body = ch + .query("SELECT argMax(body, _peerdb_version) FROM walshadow_test.events_2026 WHERE id = 1") + .expect("ch body"); + assert_eq!(body, "shaped", "rows INSERT under the renamed columns"); +} diff --git a/tests/system_columns_cdc.rs b/tests/system_columns_cdc.rs new file mode 100644 index 00000000..8ed6d702 --- /dev/null +++ b/tests/system_columns_cdc.rs @@ -0,0 +1,155 @@ +//! Destination-shape config end-to-end: renamed system columns, no delete +//! marker, operator `ORDER BY` + `PRIMARY KEY`. +//! +//! One drill: namespace `auto_create` renders the `CREATE TABLE`, so the +//! rendered shape and the INSERT contract must agree — a mismatch fails on the +//! first row, not in a unit test's string compare. The DELETE has nowhere to +//! land without a marker column, so CH keeps the row source no longer has. + +#![cfg(target_os = "linux")] + +#[path = "common/inproc_harness.rs"] +mod fx; + +use std::sync::Arc; +use std::time::Duration; + +use walshadow::mapping::{NamespaceMapping, SystemColumns}; +use walshadow::schema::RelName; +use walshadow::table_rules::{MatchKind, TableRule}; + +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn renamed_system_columns_and_operator_keys() { + if !fx::pg_available() || !fx::pg_basebackup_available() || !fx::clickhouse_available() { + eprintln!("skip: missing initdb / pg_basebackup / clickhouse"); + return; + } + + let slot = fx::Ports::alloc(); + let tmp = tempfile::tempdir().unwrap(); + let ( + fx::BootstrappedClusters { + source, + shadow, + shadow_filter_dir, + }, + shadow_stream_state, + ) = fx::bootstrap_clusters( + &tmp, + "CREATE SCHEMA sc;\n", + slot.source, + slot.shadow, + slot.walsender, + ) + .await; + let _src_stop = fx::StopOnDrop { sh: &source }; + let _shd_stop = fx::StopOnDrop { sh: &shadow }; + + let ch_tmp = tempfile::tempdir().unwrap(); + let ch = fx::ChServer::spawn(ch_tmp, slot.ch_tcp, slot.ch_http).expect("spawn ch"); + ch.query("CREATE DATABASE IF NOT EXISTS walshadow_test") + .expect("create db"); + + let mut ddl_args = fx::DdlPipelineArgs::default(); + ddl_args.namespaces.insert( + "sc".into(), + NamespaceMapping { + target_database: Some("walshadow_test".into()), + auto_create: true, + drop_table_strategy: None, + }, + ); + + let mut pipeline = fx::build_pipeline_with( + fx::BuildPipelineArgs { + tmp: &tmp, + source: &source, + shadow: &shadow, + shadow_filter_dir: &shadow_filter_dir, + shadow_stream_state, + ch_database: "walshadow_test", + ch_tcp_port: slot.ch_tcp, + mappings: vec![], + app_name: "walshadow-system-columns", + ddl: Some(ddl_args), + }, + |cfg| { + cfg.system_columns = Arc::new(SystemColumns { + lsn: "_peerdb_version".into(), + xid: "_x".into(), + commit_ts: "_peerdb_synced_at".into(), + is_deleted: None, + }); + // Reverse of the declared PK, so the key can only come from config + cfg.table_entries.push(( + RelName::new("sc", "t"), + MatchKind::Exact, + TableRule { + order_by: Some(vec!["tenant".into(), "id".into()]), + primary_key: Some(vec!["tenant".into()]), + ..TableRule::default() + }, + )); + }, + ) + .await; + let stats = pipeline.stats.clone(); + + let driver = fx::spawn_workload( + &source, + vec![ + "CREATE TABLE sc.t (id bigint, tenant bigint, body text, PRIMARY KEY (id, tenant))" + .into(), + "INSERT INTO sc.t (id, tenant, body) VALUES (1, 7, 'a'), (2, 7, 'b')".into(), + "DELETE FROM sc.t WHERE id = 2".into(), + "SELECT pg_switch_wal()".into(), + ], + ); + + let shipped = fx::pump_segments(&mut pipeline, 1, Duration::from_secs(60)).await; + let _ = driver.join(); + assert!(shipped >= 1, "no segments shipped in 60s"); + + let target = pipeline.stream.dispatched_lsn(); + let observed = shadow + .wait_for_replay(target, Duration::from_secs(30)) + .expect("shadow replay"); + assert!(observed >= target); + pipeline.shutdown().await.expect("pipeline drains clean"); + + let ddl = ch + .query("SHOW CREATE TABLE walshadow_test.t") + .expect("show create"); + assert!(ddl.contains("ReplacingMergeTree(_peerdb_version)"), "{ddl}"); + assert!(ddl.contains("ORDER BY (tenant, id)"), "{ddl}"); + assert!(ddl.contains("PRIMARY KEY (tenant)"), "{ddl}"); + + let cols = ch + .query( + "SELECT arrayStringConcat(groupArray(name), ',') FROM \ + (SELECT name FROM system.columns \ + WHERE database = 'walshadow_test' AND table = 't' AND name LIKE '\\_%' \ + ORDER BY position)", + ) + .expect("system.columns"); + assert_eq!(cols, "_peerdb_version,_x,_peerdb_synced_at", "{cols}"); + + // Source deleted id=2; with no marker column that row has no + // representation, so it stays on CH and the drop is counted + assert_eq!( + source.psql_one("SELECT count(*) FROM sc.t").unwrap(), + "1", + "source row deleted" + ); + let ch_count = ch + .query("SELECT count() FROM walshadow_test.t FINAL") + .expect("ch count"); + assert_eq!(ch_count, "2", "delete row dropped, insert retained"); + assert_eq!( + stats + .deletes_discarded + .load(std::sync::atomic::Ordering::Relaxed), + 1, + "one DELETE discarded", + ); +}