From ef5ced393016ecb79b44520089955b225608deb7 Mon Sep 17 00:00:00 2001 From: serprex <159546+serprex@users.noreply.github.com> Date: Wed, 26 Aug 2026 16:44:53 +0000 Subject: [PATCH] Add name matching capabilities table/column settings can be scoped over regex or glob match --- Cargo.lock | 25 ++ Cargo.toml | 2 + README.md | 50 ++- plans/config.md | 83 ++++- plans/emitter.md | 23 ++ sql/runtime_config_install.sql | 4 + src/backfill/backup_backfill.rs | 14 +- src/bin/stream.rs | 62 +++- src/catalog/desc_log.rs | 16 + src/column_rules.rs | 321 +++++++++++++++++++ src/config.rs | 500 ++++++++++++++++++++++------- src/emit/ch_ddl.rs | 322 +++++++++++++++---- src/emit/ch_emitter.rs | 406 ++++++++++++++++++------ src/emit/pipeline/batcher.rs | 16 +- src/emit/pipeline/bootstrap.rs | 9 +- src/emit/pipeline/plan_spool.rs | 2 +- src/emit/pipeline/planner.rs | 2 +- src/emit/pipeline/reorder.rs | 29 +- src/emit/route.rs | 22 +- src/lib.rs | 2 + src/mapping.rs | 200 ++++++++++-- src/runtime_config.rs | 106 ++++++- src/table_rules.rs | 536 ++++++++++++++++++++++++++++++++ tests/runtime_config_e2e.rs | 94 ++++++ 24 files changed, 2487 insertions(+), 359 deletions(-) create mode 100644 src/column_rules.rs create mode 100644 src/table_rules.rs diff --git a/Cargo.lock b/Cargo.lock index d3e6a7bc..e834b9dc 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -251,6 +251,16 @@ dependencies = [ "alloc-stdlib", ] +[[package]] +name = "bstr" +version = "1.13.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6bb31b46c14244e20ee9984b11bf5c992b91fb6939fea616e3512c8baecdbe5f" +dependencies = [ + "memchr", + "serde_core", +] + [[package]] name = "bumpalo" version = "3.20.3" @@ -838,6 +848,19 @@ version = "0.3.4" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "e4eba85ea1d0a966a983acd07deee566e67395d2d96b6fb39e62b5a833f1eb0b" +[[package]] +name = "globset" +version = "0.4.20" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "07c34a9410465b45bd9787443bc7370f37735bad04b0f0cd57ff1a3186c98988" +dependencies = [ + "aho-corasick", + "bstr", + "log", + "regex-automata", + "regex-syntax", +] + [[package]] name = "gloo-timers" version = "0.3.0" @@ -2914,6 +2937,7 @@ dependencies = [ "crc32c", "fallible-iterator", "futures", + "globset", "libc", "lz4", "mimalloc", @@ -2922,6 +2946,7 @@ dependencies = [ "opentelemetry_sdk", "pglz", "postgres-protocol", + "regex-automata", "serde", "serde_json", "smallvec", diff --git a/Cargo.toml b/Cargo.toml index 22dd8f6e..3d461271 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -56,6 +56,8 @@ libc = "0.2" lz4 = "1" pglz = "0.1" postgres-protocol = "0.6" +globset = "0.4" +regex-automata = "0.4" serde = { version = "1", features = ["derive"] } serde_json = "1" smallvec = "1" diff --git a/README.md b/README.md index c795d13e..80353f65 100644 --- a/README.md +++ b/README.md @@ -140,28 +140,72 @@ host = "localhost" database = "analytics" # Opt a table out (wins over replicate_all): -[table."public.audit_log"] +[table.public.audit_log] replicate = false # Or list explicitly — replicate only what you name: [stream] replicate_all = false -[table."public.users"] +[table.public.users] replicate = true initial_load = "none" -target = "users" +target_table = "users" columns = [ { attnum = 1, target = "id", type = "UInt64" }, { attnum = 2, target = "name", type = "String" }, ] ``` +Table blocks take two key levels, `[table..]`; a name +carrying a dot or other TOML-special character quotes per key rules, e.g. +`[table.public."odd.name"]`. + `replicate_all` skips system schemas (`pg_*`, `information_schema`, the `[runtime_config]` schema). `attnum` values match `pg_attribute.attnum` (1-based) on the source relation; `type` is the CH destination type walshadow advertises in the INSERT block. SIGHUP reloads mappings atomically; connection params stay boot-only. +Name a set of tables instead of one, with `match`: + +```toml +[table.app."events_*"] # each name part an anchored pattern +match = "glob" # exact (default) | glob | regex +replicate = true +initial_load = "copy" + +[table.app."*_audit"] # a guardrail: excludes even what +match = "glob" # a wider opt-in swept in +replicate = false +``` + +Everything a `config_table` row carries — destination, `replicate`, +`initial_load` — also takes `match = "glob"` / `"regex"`, which is how you +scope tables that do not exist yet: under `auto_create` / `replicate_all` a +relation reaches CH the first time it is seen, before a row naming it could +arrive. See [plans/config.md](plans/config.md). + +Columns take the same treatment. A `columns` entry keys on `attnum` or on +`name`, never both, and one array holds one kind: + +```toml +[table.app."*"] +match = "glob" +replicate = true +columns = [ + { name = "*_at", match = "glob", type = "DateTime64(6, 'UTC')" }, + { name = "legacy_id", target = "id" }, # rename, keep the bridge type +] +``` + +An `attnum` entry pins the projection outright, so it needs `target` + `type` +and only fits a block naming one relation. A `name` entry states the CH name +or type of a column the descriptor yields anyway: `target` and `type` are each +optional, a `match` entry may not set `target` (several columns cannot share +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. + ## Building from source diff --git a/plans/config.md b/plans/config.md index 29355ab5..d389a358 100644 --- a/plans/config.md +++ b/plans/config.md @@ -29,12 +29,8 @@ untouched. - `tables` — per-relation destination mapping, keyed `"."` - `namespaces` — per-namespace defaults (`auto_create`, `target_database`, `drop_table_strategy`) -- `columns` — per-column CH-type override from the `config_column` overlay, - keyed `"."` → source attname → CH type expression, - WAL-tracked. Consumed by `TablePlan::build` (the batcher's plan cache), - which resolves attname→attnum against the descriptor at hand and swaps the - column's encode type when the override is wire-compatible (see §column - overrides below) +- `column_rules` — per-column CH names and types from TOML and + `config_column`, applied when building mappings, DDL, and encoder plans - `drop_table_strategy` — global DROP fallback; per-namespace overrides it - `row_budget`, `byte_budget`, `flush_timeout` — emitter batch-seal triggers, read live by the batcher per seal decision (ticker re-armed on change) @@ -82,7 +78,8 @@ left as it stands, since its retention is what a rollback to that server reads. `target_database`, `soft_delete`, `[stream] replicate_all`, and the `[runtime_config] schema` name thread into the DDL applicator at construction -and carry across refreshes unchanged. `replicate_all` (default `true`) +and carry across refreshes unchanged. Refreshed snapshots provide per-table +destinations and scope. `replicate_all` (default `true`) auto-creates and replicates every user table whose namespace is not a system schema (`pg_*`, `information_schema`, the runtime-config schema); an explicit `auto_create` namespace or a `replicate = false` opt-out still wins. @@ -138,6 +135,53 @@ whether the daemon owns a shadow lifecycle at all, which is a per-invocation recovery decision like `--start-lsn` and `--ignore-cursor`, and CLI-only keeps a stale config file from stomping it. +## Name patterns + +Set `match` on a table or column entry to choose how names are read: + +- `exact` matches a literal name and is default +- `glob` supports `*`, `?`, character classes such as `[a-z]`, and choices + such as `{one,two}` +- `regex` supports regular expressions without backreferences + +Glob and regex patterns match whole names. For example, `events_*` matches +`events_2026`, but `events` does not match `my_events`. + +```toml +[table.app."events_*"] +match = "glob" +replicate = true +initial_load = "copy" +target_database = "warehouse" +``` + +```sql +INSERT INTO walshadow.config_table (namespace, relname, match, replicate) +VALUES ('app', 'events_*', 'glob', true); +``` + +Use glob for common prefix and suffix matches. Use regex when glob cannot +express a pattern, for example `v[0-9]+_.*`. + +Matching patterns apply from least specific to most specific. Longer combined +namespace and table patterns are more specific. An exact entry applies last. +Runtime config wins when it repeats a TOML pattern. Any matching pattern with +`replicate = false` blocks pattern-based opt-ins, but an exact entry can +override it. + +`config_column.match` applies to namespace, table, and column names together. +Invalid patterns and unknown match modes are rejected and logged. Other rules +remain active. + +Pattern table rules also control scope: + +- `replicate = false` prevents automatic creation and removes matching routes +- `replicate = true` enables automatic creation and requested initial loads + for matching tables, including tables created later +- existing routes keep pinned column mappings +- patterns never include `pg_*`, `information_schema`, or runtime config + schemas + ## `.config_*` tables DBA runs [`sql/runtime_config_install.sql`](../sql/runtime_config_install.sql) @@ -150,12 +194,14 @@ daemon — preserving walshadow's read-only-source posture. Four tables: `drop_table_strategy` - `config_namespace` — key `namespace`: `target_database`, `auto_create`, `drop_table_strategy` -- `config_table` — key `(namespace, relname)`: `target_database`, - `target_table` (each NULL = derived: namespace default / source relname), - `replicate`, `initial_load` (`none`, `copy`, `base_backup`, `object_store`). +- `config_table` — key `(namespace, relname)`: `match` (`exact`, `glob`, or + `regex`), `target_database`, `target_table` + (each NULL = derived: namespace default / source relname), `replicate`, + `initial_load` (`none`, `copy`, `base_backup`, `object_store`). Name key, not relfilenode, rfn is unknown at row-insert time for forward-declared tables -- `config_column` — key `(namespace, relname, attname)`: `target_type` +- `config_column` — key `(namespace, relname, attname)`: `match` (`exact`, + `glob`, or `regex`), `target_type` Every column is nullable and NULL means "daemon default / TOML applies", so the schema grows additively: a newer daemon reading an older install still works. @@ -277,6 +323,16 @@ destination unchanged. ## Column overrides +Column rules use same precedence as name patterns: least specific patterns +first, followed by an exact rule. When TOML and `config_column` define same +rule, `config_column` wins. + +- TOML name rules may set ClickHouse column names and types. New tables and + columns use these values in mappings and DDL. A custom type removes any + generated default. Nullable columns are omitted from `ORDER BY`. +- `config_column` may change encoder type for an existing column. It cannot + rename a ClickHouse column or alter its type. + `config_column.target_type` reaches the emitted projection in two stages, because the two failure classes surface at different points: @@ -306,9 +362,8 @@ receiver, so backfilled rows encode under the same overrides as WAL-driven rows. The greenfield bootstrap tail stays TOML-only (no resolver exists yet at that phase). -The override changes the projection only — CH-side DDL (`CREATE TABLE` / -`ADD COLUMN`) still renders bridge-derived types; retyping an existing CH -column stays an operator migration. +Runtime overrides change encoder projection only. Retyping an existing +ClickHouse column remains an operator migration. ## Subscribers diff --git a/plans/emitter.md b/plans/emitter.md index c58e0e29..6e63cb74 100644 --- a/plans/emitter.md +++ b/plans/emitter.md @@ -436,6 +436,29 @@ columns = [ ] ``` +A `columns` entry uses either `attnum` or `name`. Do not mix both forms in one +array. + +- `attnum` pins a projection. `target` and `type` are required. +- `name` changes a catalog-derived column. `target` and `type` are optional. + Source name and derived type remain when omitted. Set `match` to `glob` or + `regex` to cover current and future columns. Pattern entries cannot set + `target`, because several columns cannot share one ClickHouse name. + +```toml +[table.app."*"] +match = "glob" +replicate = true +columns = [ + { name = "*_at", match = "glob", type = "DateTime64(6, 'UTC')" }, + { name = "legacy_id", target = "id" }, +] +``` + +Name entries do not select tables for replication. Use `replicate` or +`auto_create` for scope. Column rules apply to mappings, `CREATE TABLE`, +`ADD COLUMN`, and encoder plans. + `MappingHandle = Arc>>` is the live handle the planner's route view resolves from. Handle is cloneable; daemon's SIGHUP task swaps whole inner `HashMap`. Routes diff --git a/sql/runtime_config_install.sql b/sql/runtime_config_install.sql index 17a07fe3..4706e8fe 100644 --- a/sql/runtime_config_install.sql +++ b/sql/runtime_config_install.sql @@ -42,6 +42,7 @@ CREATE TABLE IF NOT EXISTS :"walshadow_schema".config_namespace ( CREATE TABLE IF NOT EXISTS :"walshadow_schema".config_table ( namespace text NOT NULL, relname text NOT NULL, + match text, target_database text, -- ClickHouse database; NULL derives from -- config_namespace.target_database / TOML target_table text, -- ClickHouse table; NULL derives from relname @@ -59,6 +60,7 @@ CREATE TABLE IF NOT EXISTS :"walshadow_schema".config_column ( namespace text NOT NULL, relname text NOT NULL, attname text NOT NULL, + match text, target_type text, -- ClickHouse type expression PRIMARY KEY (namespace, relname, attname) ); @@ -70,6 +72,8 @@ 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 match 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 -- DELETE always carries the key columns the decoder reads (namespace/relname/ diff --git a/src/backfill/backup_backfill.rs b/src/backfill/backup_backfill.rs index 78d2abe9..b85a70af 100644 --- a/src/backfill/backup_backfill.rs +++ b/src/backfill/backup_backfill.rs @@ -1080,18 +1080,12 @@ impl ReplaySink { continue; }; let mapping = Arc::new(mapping.clone()); - let overrides = self + let rules = self .config .as_ref() - .and_then(|rc| rc.columns.get(&rel.rel_name)) - .cloned() - .map(Arc::new) - .unwrap_or_default(); - let route = crate::emit::route::RouteSnapshot::freeze( - mapping, - overrides, - self.soft_delete, - ); + .map_or_else(Arc::default, |rc| rc.column_rules.clone()); + let route = + crate::emit::route::RouteSnapshot::freeze(mapping, rules, self.soft_delete); let seq = if let Some((seq, rows)) = &mut self.open { *rows += 1; *seq diff --git a/src/bin/stream.rs b/src/bin/stream.rs index fbf21b8a..8985e710 100644 --- a/src/bin/stream.rs +++ b/src/bin/stream.rs @@ -1526,6 +1526,12 @@ async fn run_session( .values() .filter_map(|r| r.initial_load.as_ref()), ) + .chain( + emitter_cfg + .table_entries + .iter() + .filter_map(|(_, _, rule)| rule.initial_load.as_ref()), + ) .any(|mode| InitialLoadMode::parse(mode).is_some_and(|m| m != InitialLoadMode::None)); // One validated resident-payload pool for the pipeline and every // concurrent backup pass @@ -1558,7 +1564,7 @@ async fn run_session( // `initial_load` row: COPY covers commits before it, WAL the rest; // the ledger resumes/no-ops rows seen on an earlier boot. for (rel, row) in &seeded_table_rows { - if row.replicate.is_some() { + if row.replicate.is_some() && !row.is_pattern() { walshadow::opt_in::apply_table_opt_in( &resolver, &mut applicator, @@ -1587,9 +1593,31 @@ async fn run_session( .with_context(|| format!("config opt-in for {rel}"))?; } } + let pattern_scoped: Vec<(RelName, walshadow::runtime_config::TableRow)> = { + let snap = config_rx.borrow(); + let config_schema = emitter_cfg.runtime_config_schema.as_deref(); + snap.rules.pattern_scoped( + || desc_log.user_rel_names_at(raw_start.get(), config_schema), + |rel| snap.tables.contains_key(rel), + ) + }; + for (rel, row) in &pattern_scoped { + walshadow::opt_in::apply_table_opt_in( + &resolver, + &mut applicator, + &catalog, + backfiller_effects.as_ref(), + rel, + row, + raw_start.get(), + ) + .await + .with_context(|| format!("pattern opt-in for {rel}"))?; + } let sql_scoped_tables: HashSet = seeded_table_rows .iter() - .filter(|(_, row)| row.replicate.is_some()) + .filter(|(_, row)| row.replicate.is_some() && !row.is_pattern()) + .chain(pattern_scoped.iter()) .map(|(rel, _)| rel.clone()) .collect(); let active_tables: HashSet = config_rx.borrow().tables.keys().cloned().collect(); @@ -2817,13 +2845,16 @@ 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(), + match_kind: row.try_get("match").ok().flatten(), }, ); } for row in client .query( - &format!("SELECT namespace, relname, attname, target_type FROM {s}.config_column"), + &format!( + "SELECT namespace, relname, attname, match, target_type FROM {s}.config_column" + ), &[], ) .await @@ -2835,7 +2866,8 @@ async fn seed_runtime_config( overlay.columns.insert( (RelName::new(&namespace, &relname), attname), ColumnRow { - target_type: row.get("target_type"), + target_type: row.try_get("target_type").ok().flatten(), + match_kind: row.try_get("match").ok().flatten(), }, ); } @@ -4327,13 +4359,21 @@ async fn bootstrap_build_mapping( cli_source_base(args), mapping.clone(), ); - let ddl_cfg = walshadow::ch_ddl::DdlConfig::from_resolved( - &config_rx.borrow(), - emitter_cfg.database.clone(), - emitter_cfg.soft_delete, - emitter_cfg.replicate_all, - emitter_cfg.runtime_config_schema.clone(), - ); + let (ddl_cfg, merged_tables) = { + let snap = config_rx.borrow(); + ( + walshadow::ch_ddl::DdlConfig::from_resolved( + &snap, + emitter_cfg.database.clone(), + emitter_cfg.soft_delete, + emitter_cfg.replicate_all, + emitter_cfg.runtime_config_schema.clone(), + ), + Arc::new(snap.tables.clone()), + ) + }; + // Publish rule-adjusted targets before creating tables + mapping.publish(merged_tables).await; let mut applicator = walshadow::ch_ddl::DdlApplicator::new(emitter_cfg, ddl_cfg, mapping.clone(), config_rx) .await diff --git a/src/catalog/desc_log.rs b/src/catalog/desc_log.rs index e4efccac..b3c88f59 100644 --- a/src/catalog/desc_log.rs +++ b/src/catalog/desc_log.rs @@ -684,6 +684,22 @@ impl DescriptorLog { .collect() } + /// Names of user relations `Present` at `lsn`, TOAST and system schemas + /// dropped — the catalog a name pattern resolves against + pub fn user_rel_names_at(&self, lsn: u64, config_schema: Option<&str>) -> Vec { + self.active_present_at(lsn) + .into_iter() + .filter(|d| { + d.kind != 't' + && !crate::emit::ch_ddl::is_system_namespace( + &d.rel_name.namespace, + config_schema, + ) + }) + .map(|d| d.rel_name.clone()) + .collect() + } + /// One-time baseline on an empty log: writes the ckpt (meta + /// seed batch) so `covered_through` is durable before any tail append. /// Boundaries at or below `covered_through` are baked into the seed — diff --git a/src/column_rules.rs b/src/column_rules.rs new file mode 100644 index 00000000..3a4a8e1b --- /dev/null +++ b/src/column_rules.rs @@ -0,0 +1,321 @@ +//! Resolves column settings from TOML and runtime config + +use ahash::HashMap; + +use crate::schema::RelName; +use crate::table_rules::{MatchKind, NamePattern, RelMatcher, set_if}; + +#[derive(Debug, Clone, Default, PartialEq, Eq)] +pub struct ColumnRule { + pub target_name: Option, + pub target_type: Option, +} + +impl ColumnRule { + pub fn overlay(&mut self, other: &Self) { + set_if(&mut self.target_name, &other.target_name); + set_if(&mut self.target_type, &other.target_type); + } + + pub fn is_empty(&self) -> bool { + self.target_name.is_none() && self.target_type.is_none() + } +} + +#[derive(Debug, Clone)] +pub struct ColumnEntry { + pub rel: RelName, + pub rel_kind: MatchKind, + pub attname: String, + pub att_kind: MatchKind, + pub rule: ColumnRule, +} + +#[derive(Debug, Clone)] +struct ColumnMatcher { + rel: RelMatcher, + att_kind: MatchKind, + attname: NamePattern, + attname_src: String, +} + +impl ColumnMatcher { + fn compile( + rel: &RelName, + rel_kind: MatchKind, + attname: &str, + att_kind: MatchKind, + ) -> Result { + Ok(Self { + rel: RelMatcher::compile(rel, rel_kind)?, + att_kind, + attname: NamePattern::compile(att_kind, attname)?, + attname_src: attname.to_owned(), + }) + } + + fn matches(&self, rel: &RelName, attname: &str) -> bool { + self.rel.matches(rel) && self.attname.is_match(attname) + } + + /// Broadest first, literals last + fn rank(&self) -> (bool, usize, &str, &str, &str) { + let (_, rel_width, namespace, name) = self.rel.rank(); + ( + !self.rel.is_pattern() && self.att_kind == MatchKind::Exact, + rel_width + self.attname_src.len(), + namespace, + name, + &self.attname_src, + ) + } +} + +#[derive(Debug, Clone, Default)] +pub struct ColumnRules { + /// Ranked broadest to narrowest + rules: Vec<(ColumnMatcher, ColumnRule)>, + /// Last valid type for each runtime config row + accepted: HashMap>, +} + +impl ColumnRules { + pub fn settings(&self, rel: &RelName, attname: &str) -> ColumnRule { + let mut merged = ColumnRule::default(); + for (matcher, rule) in &self.rules { + if matcher.matches(rel, attname) { + merged.overlay(rule); + } + } + merged + } + + pub fn accepted_type(&self, rel: &RelName, attname: &str) -> Option<&str> { + self.accepted + .get(rel) + .and_then(|m| m.get(attname)) + .map(String::as_str) + } + + pub fn is_empty(&self) -> bool { + self.rules.is_empty() + } +} + +#[derive(Debug, Default)] +pub struct ColumnRulesBuilder { + rules: Vec<(usize, ColumnMatcher, ColumnRule)>, + accepted: HashMap>, + layer: usize, + rejections: u64, +} + +impl ColumnRulesBuilder { + pub fn new() -> Self { + Self::default() + } + + pub fn next_layer(&mut self) { + self.layer += 1; + } + + pub fn add( + &mut self, + rel: &RelName, + rel_kind: MatchKind, + attname: &str, + att_kind: MatchKind, + rule: ColumnRule, + ) { + match ColumnMatcher::compile(rel, rel_kind, attname, att_kind) { + Ok(matcher) => self.rules.push((self.layer, matcher, rule)), + Err(e) => { + tracing::warn!(target: "walshadow::config", qname = %rel, attname = %attname, error = %e, "column entry rejected"); + self.rejections += 1; + } + } + } + + pub fn record_accepted(&mut self, rel: &RelName, attname: &str, target_type: &str) { + self.accepted + .entry(rel.clone()) + .or_default() + .insert(attname.to_owned(), target_type.to_owned()); + } + + pub fn bump_rejections(&mut self) { + self.rejections += 1; + } + + pub fn finish(mut self) -> (ColumnRules, u64) { + self.rules + .sort_by(|(la, ma, _), (lb, mb, _)| ma.rank().cmp(&mb.rank()).then(la.cmp(lb))); + ( + ColumnRules { + rules: self.rules.into_iter().map(|(_, m, r)| (m, r)).collect(), + accepted: self.accepted, + }, + self.rejections, + ) + } +} + +#[cfg(test)] +mod tests { + use super::*; + + fn rel(ns: &str, name: &str) -> RelName { + RelName::new(ns, name) + } + + fn ty(t: &str) -> ColumnRule { + ColumnRule { + target_type: Some(t.into()), + ..ColumnRule::default() + } + } + + #[test] + fn exact_entry_types_only_its_column() { + let mut b = ColumnRulesBuilder::new(); + b.add( + &rel("public", "events"), + MatchKind::Exact, + "amount", + MatchKind::Exact, + ty("Decimal(38, 9)"), + ); + let (rules, rejected) = b.finish(); + assert_eq!(rejected, 0); + assert_eq!( + rules + .settings(&rel("public", "events"), "amount") + .target_type, + Some("Decimal(38, 9)".into()) + ); + assert!( + rules + .settings(&rel("public", "events"), "net_amount") + .target_type + .is_none(), + "anchored: substring must not match" + ); + assert!( + rules + .settings(&rel("public", "other"), "amount") + .target_type + .is_none() + ); + } + + #[test] + fn glob_attname_spans_relations_a_pattern_block_names() { + let mut b = ColumnRulesBuilder::new(); + b.add( + &rel("app", "*"), + MatchKind::Glob, + "*_at", + MatchKind::Glob, + ty("DateTime64(6, 'UTC')"), + ); + let (rules, _) = b.finish(); + for (r, c) in [("events", "created_at"), ("orders", "shipped_at")] { + assert_eq!( + rules.settings(&rel("app", r), c).target_type, + Some("DateTime64(6, 'UTC')".into()), + "{r}.{c}" + ); + } + assert!( + rules + .settings(&rel("app", "events"), "at_rest") + .target_type + .is_none() + ); + assert!( + rules + .settings(&rel("other", "events"), "created_at") + .target_type + .is_none() + ); + } + + #[test] + fn narrower_entry_and_exact_entry_win() { + let mut b = ColumnRulesBuilder::new(); + b.add( + &rel("*", "*"), + MatchKind::Glob, + "*", + MatchKind::Glob, + ColumnRule { + target_name: Some("broad".into()), + target_type: Some("String".into()), + }, + ); + b.add( + &rel("app", "events"), + MatchKind::Exact, + "amt_*", + MatchKind::Glob, + ty("Decimal(38, 9)"), + ); + let (rules, _) = b.finish(); + let s = rules.settings(&rel("app", "events"), "amt_net"); + assert_eq!(s.target_type, Some("Decimal(38, 9)".into())); + assert_eq!( + s.target_name, + Some("broad".into()), + "broad entry still contributes fields the narrow one omits" + ); + + let mut b = ColumnRulesBuilder::new(); + b.add( + &rel("app", "events"), + MatchKind::Exact, + "amt_*", + MatchKind::Glob, + ty("Decimal(38, 9)"), + ); + b.next_layer(); + b.add( + &rel("app", "events"), + MatchKind::Exact, + "amt_net", + MatchKind::Exact, + ty("Int128"), + ); + let (rules, _) = b.finish(); + assert_eq!( + rules.settings(&rel("app", "events"), "amt_net").target_type, + Some("Int128".into()) + ); + } + + #[test] + fn unparseable_pattern_rejected() { + let mut b = ColumnRulesBuilder::new(); + b.add( + &rel("app", "events"), + MatchKind::Exact, + "am(t", + MatchKind::Regex, + ty("String"), + ); + let (rules, rejected) = b.finish(); + assert_eq!(rejected, 1); + assert!(rules.is_empty()); + } + + #[test] + fn accepted_type_survives_for_retention() { + let mut b = ColumnRulesBuilder::new(); + b.record_accepted(&rel("app", "events"), "amount", "Int128"); + let (rules, _) = b.finish(); + assert_eq!( + rules.accepted_type(&rel("app", "events"), "amount"), + Some("Int128") + ); + assert_eq!(rules.accepted_type(&rel("app", "events"), "other"), None); + } +} diff --git a/src/config.rs b/src/config.rs index 4b9c133b..76e78e75 100644 --- a/src/config.rs +++ b/src/config.rs @@ -32,6 +32,7 @@ use walrus::pg::replication::conn::PgConfig; use walrus::pg::replication::tls::{SslMode, TlsParams}; use crate::ch::{CompressionChoice, EmitterError}; +use crate::column_rules::{ColumnRule, ColumnRules, ColumnRulesBuilder}; use crate::emit::ch_emitter::EmitterConfig; use crate::mapping::{ DropTableStrategy, MappingHandle, NamespaceMapping, TableMapping, TableTarget, @@ -39,7 +40,8 @@ use crate::mapping::{ }; use crate::runtime_config::{ConfigEvent, ConfigOverlay, TableRow}; use crate::schema::{RelDescriptor, RelName, SchemaDiff}; -use ahash::{HashMap, HashMapExt, HashSet}; +use crate::table_rules::{MatchKind, TableRules, TableRulesBuilder}; +use ahash::{HashMap, HashSet}; #[derive(Clone, PartialEq, Eq)] pub struct SourceConn { @@ -166,13 +168,8 @@ pub struct ResolvedConfig { pub tables: HashMap, /// Per-namespace defaults keyed on PG schema name pub namespaces: HashMap, - /// Per-column CH-type override from the `config_column` overlay, keyed - /// rel → source attname → CH type expression. - /// Type strings are parse-validated at merge (Regime A: malformed - /// rejected, prior value kept). Consumed by `TablePlan::build`, which - /// resolves attname→attnum against the descriptor at hand and swaps the - /// column's encode type when the override is wire-compatible. - pub columns: HashMap>, + pub column_rules: Arc, + pub rules: Arc, /// Global DROP TABLE strategy fallback (`retain` / `drop` / `warn`); /// per-namespace `NamespaceMapping::drop_table_strategy` overrides it pub drop_table_strategy: String, @@ -242,7 +239,7 @@ impl Default for ResolvedConfig { &ConfigOverlay::default(), &CliOverrides::default(), &OptInState::default(), - &HashMap::new(), + &ColumnRules::default(), ) .0 } @@ -347,7 +344,7 @@ impl ConfigResolver { ) -> (Arc, watch::Receiver>) { let overlay = ConfigOverlay::default(); let opt_in = OptInState::default(); - let (initial, _) = Self::resolve(base, &overlay, &cli, &opt_in, &HashMap::new()); + let (initial, _) = Self::resolve(base, &overlay, &cli, &opt_in, &ColumnRules::default()); let (tx, rx) = watch::channel(Arc::new(initial)); let this = Arc::new(Self { toml_path, @@ -425,11 +422,18 @@ impl ConfigResolver { ) { let mut inner = self.inner.lock().await; let rel = desc.rel_name.clone(); + // Keep routing target aligned with DDL target + let settings = self.tx.borrow().rules.settings(&rel); let target = TableTarget { - database: db_override.unwrap_or_else(|| Self::target_db_for(&inner, &rel.namespace)), - table: table_override.unwrap_or_else(|| rel.name.to_string()), + database: db_override + .or(settings.target_database) + .unwrap_or_else(|| Self::target_db_for(&inner, &rel.namespace)), + table: table_override + .or(settings.target_table) + .unwrap_or_else(|| rel.name.to_string()), }; - let columns = derive_columns_for_mapping(desc); + let column_rules = self.tx.borrow().column_rules.clone(); + let columns = derive_columns_for_mapping(desc, &column_rules); inner .opt_in .mappings @@ -479,10 +483,11 @@ impl ConfigResolver { row } - /// Whether the operator opted this rel out (`replicate=false`). Read by - /// `DdlApplicator::apply_added` so an excluded rel skips auto-create. pub async fn is_excluded(&self, rel: &RelName) -> bool { - self.inner.lock().await.opt_in.excluded.contains(rel) + if self.inner.lock().await.opt_in.excluded.contains(rel) { + return true; + } + self.tx.borrow().rules.settings(rel).replicate == Some(false) } /// Record an applicator-derived mapping (`auto_create` CREATE TABLE) so @@ -528,13 +533,14 @@ impl ConfigResolver { if inner.opt_in.excluded.contains(rel) { return; } + let rules = self.tx.borrow().column_rules.clone(); if let Some(m) = inner.opt_in.mappings.get_mut(rel) { - fold_diff_into_mapping(m, new, diff); + fold_diff_into_mapping(m, new, diff, &rules); } else if let Some(m) = inner.opt_in.derived.get_mut(rel) { - fold_diff_into_mapping(m, new, diff); + fold_diff_into_mapping(m, new, diff, &rules); } else if let Some(base) = inner.base.tables.get(rel) { let mut m = base.clone(); - fold_diff_into_mapping(&mut m, new, diff); + fold_diff_into_mapping(&mut m, new, diff, &rules); inner.opt_in.derived.insert(rel.clone(), m); } else { return; @@ -569,7 +575,7 @@ impl ConfigResolver { &inner.overlay, &self.cli, &inner.opt_in, - &prev.columns, + &prev.column_rules, ); self.rejections.store(rejections, Ordering::Relaxed); self.mapping @@ -589,21 +595,74 @@ impl ConfigResolver { /// overrides on top. Returns the resolved config and the count of overlay /// values rejected as malformed (kept at the pre-overlay value, logged at /// WARN — Regime A: a bad row never crashes or freezes the pump). - /// `prev_columns` is the last published snapshot's column overrides: a - /// `target_type` that fails to parse falls back to its entry there, so a - /// malformed update can't revert an already-accepted encode type. fn resolve( base: &EmitterConfig, overlay: &ConfigOverlay, cli: &CliOverrides, opt_in: &OptInState, - prev_columns: &HashMap>, + prev_columns: &ColumnRules, ) -> (ResolvedConfig, u64) { let mut rejections = 0u64; + // Runtime config overrides TOML at equal specificity + let mut rules = TableRulesBuilder::new(); + for (rel, kind, rule) in &base.table_entries { + rules.add(rel, *kind, rule.clone()); + } + rules.next_layer(); + for (rel, row) in &overlay.tables { + rules.add_row(rel, row); + } + let (rules, rule_rejections) = rules.finish(); + rejections += rule_rejections; + let rules = Arc::new(rules); + + let mut column_rules = ColumnRulesBuilder::new(); + for e in &base.column_entries { + column_rules.add(&e.rel, e.rel_kind, &e.attname, e.att_kind, e.rule.clone()); + } + column_rules.next_layer(); + for ((rel, attname), row) in &overlay.columns { + let kind = match MatchKind::parse(row.match_kind.as_deref().unwrap_or_default()) { + Ok(k) => k, + Err(e) => { + column_rules.bump_rejections(); + tracing::warn!(target: "walshadow::config", qname = %rel, attname = %attname, error = %e, "config_column.match rejected"); + continue; + } + }; + let Some(ty) = &row.target_type else { + continue; + }; + // Validate syntax now, validate wire compatibility with descriptor + let accepted = if TypeAst::parse(ty, Allocator::stdlib()).is_ok() { + Some(ty.as_str()) + } else { + column_rules.bump_rejections(); + let prior = prev_columns.accepted_type(rel, attname); + tracing::warn!(target: "walshadow::config", qname = %rel, attname = %attname, value = %ty, kept_prior = prior.is_some(), "config_column.target_type rejected: unparseable CH type"); + prior + }; + if let Some(ty) = accepted { + column_rules.record_accepted(rel, attname, ty); + column_rules.add( + rel, + kind, + attname, + kind, + ColumnRule { + target_type: Some(ty.to_owned()), + ..ColumnRule::default() + }, + ); + } + } + let (column_rules, column_rejections) = column_rules.finish(); + rejections += column_rejections; let mut rc = ResolvedConfig { tables: base.tables.clone(), namespaces: base.namespaces.clone(), - columns: HashMap::new(), + column_rules: Arc::new(column_rules), + rules: rules.clone(), drop_table_strategy: base.drop_table_strategy.clone(), row_budget: base.row_budget, byte_budget: base.byte_budget, @@ -710,65 +769,32 @@ impl ConfigResolver { } } - for (rel, row) in &overlay.tables { - // `target_database`/`target_table` override the destination of a - // table already mapped by TOML or opted in above (both carry the - // column projection). A `config_table` row that only sets a target - // for an unmapped table can't be routed without a projection — - // `replicate=true` is the way to bring such a table into scope. - // NULL = that part unchanged. - if row.target_database.is_some() || row.target_table.is_some() { - match rc.tables.get_mut(rel) { - Some(m) => { - if let Some(db) = &row.target_database { - m.target.database = db.clone(); - } - if let Some(t) = &row.target_table { - m.target.table = t.clone(); - } - } - None => tracing::warn!( - target: "walshadow::config", - qname = %rel, - "config_table target ignored: no mapping (set replicate=true to opt-in)", - ), - } + for (rel, m) in rc.tables.iter_mut() { + let settings = rules.settings(rel); + if let Some(db) = settings.target_database { + m.target.database = db; + } + if let Some(t) = settings.target_table { + m.target.table = t; } } - - for ((rel, attname), row) in &overlay.columns { - if let Some(ty) = &row.target_type { - // Parse-validate here so a malformed type never reaches a - // TablePlan build (whose error would poison the batcher). - // Wire-shape compatibility needs the descriptor, so that - // check (with fallback) runs at plan build instead. - if TypeAst::parse(ty, Allocator::stdlib()).is_ok() { - rc.columns - .entry(rel.clone()) - .or_default() - .insert(attname.clone(), ty.clone()); - } else { - rejections += 1; - // Bad update keeps last accepted override (prev snapshot); - // overlay mirrors PG rows so retention can't live there - let prior = prev_columns.get(rel).and_then(|m| m.get(attname)); - if let Some(prior) = prior { - rc.columns - .entry(rel.clone()) - .or_default() - .insert(attname.clone(), prior.clone()); - } - tracing::warn!(target: "walshadow::config", qname = %rel, attname = %attname, value = %ty, kept_prior = prior.is_some(), "config_column.target_type rejected: unparseable CH type"); - } + for (rel, row) in &overlay.tables { + let names_target = row.target_database.is_some() || row.target_table.is_some(); + if names_target && !row.is_pattern() && !rc.tables.contains_key(rel) { + tracing::warn!( + target: "walshadow::config", + qname = %rel, + "config_table target ignored: no mapping (set replicate=true to opt-in)", + ); } } - // Opt-out (last, so exclusion wins over any TOML/overlay mapping): - // a `replicate=false` rel leaves the routing map, so route planning - // resolves None and its rows discard mid-stream. + // Apply exclusions after mappings for rel in &opt_in.excluded { rc.tables.remove(rel); } + rc.tables + .retain(|rel, _| rules.settings(rel).replicate != Some(false)); // Layer 1: CLI (top). Survives SIGHUP + stale overlay rows. if let Some(v) = &cli.drop_table_strategy { @@ -805,6 +831,7 @@ impl ConfigResolver { mod tests { use super::*; use crate::runtime_config::{GlobalRow, NamespaceRow, TableRow}; + use ahash::HashMapExt; fn base_with(drop_strategy: &str) -> EmitterConfig { EmitterConfig::from_toml_str(&format!( @@ -833,7 +860,7 @@ mod tests { &overlay, &CliOverrides::default(), &OptInState::default(), - &HashMap::new(), + &ColumnRules::default(), ); assert_eq!(r.drop_table_strategy, "drop"); // CLI beats overlay. @@ -846,7 +873,7 @@ mod tests { &overlay, &cli, &OptInState::default(), - &HashMap::new(), + &ColumnRules::default(), ); assert_eq!(r.drop_table_strategy, "warn"); } @@ -859,7 +886,7 @@ mod tests { &ConfigOverlay::default(), &CliOverrides::default(), &OptInState::default(), - &HashMap::new(), + &ColumnRules::default(), ); assert_eq!(r.drop_table_strategy, "drop"); } @@ -884,7 +911,7 @@ mod tests { &overlay, &CliOverrides::default(), &OptInState::default(), - &HashMap::new(), + &ColumnRules::default(), ); assert_eq!(rej, 0); assert_eq!(r.row_budget, 1000); @@ -910,7 +937,7 @@ mod tests { &overlay, &CliOverrides::default(), &OptInState::default(), - &HashMap::new(), + &ColumnRules::default(), ); assert_eq!(rej, 3); // Prior (TOML/base) values survive each rejection. @@ -950,7 +977,7 @@ mod tests { &overlay, &CliOverrides::default(), &OptInState::default(), - &HashMap::new(), + &ColumnRules::default(), ); let ns = r.namespaces.get("public").unwrap(); assert!(ns.auto_create); @@ -964,6 +991,178 @@ mod tests { ); } + #[test] + fn pattern_opt_out_drops_a_toml_mapped_relation() { + let base = EmitterConfig::from_toml_str( + "[ch]\n\ + [table.app.tmp_scratch]\n\ + columns = [{ attnum = 1, target = \"id\", type = \"UInt64\" }]\n", + ) + .unwrap(); + let mut overlay = ConfigOverlay::default(); + overlay.tables.insert( + RelName::new("app", "tmp_*"), + TableRow { + match_kind: Some("glob".into()), + replicate: Some(false), + ..Default::default() + }, + ); + let (r, _) = ConfigResolver::resolve( + &base, + &overlay, + &CliOverrides::default(), + &OptInState::default(), + &ColumnRules::default(), + ); + assert!( + !r.tables.contains_key(&RelName::new("app", "tmp_scratch")), + "an excluding pattern takes the mapping out of the routing map" + ); + } + + #[test] + fn overlay_pattern_scope_expands_over_present_relations() { + let base = EmitterConfig::from_toml_str("[ch]\n").unwrap(); + let mut overlay = ConfigOverlay::default(); + overlay.tables.insert( + RelName::new("app", "*_audit"), + TableRow { + match_kind: Some("glob".into()), + replicate: Some(false), + ..Default::default() + }, + ); + overlay.tables.insert( + RelName::new("app", "events_*"), + TableRow { + match_kind: Some("glob".into()), + replicate: Some(true), + initial_load: Some("copy".into()), + ..Default::default() + }, + ); + let (r, _) = ConfigResolver::resolve( + &base, + &overlay, + &CliOverrides::default(), + &OptInState::default(), + &ColumnRules::default(), + ); + let present = [ + RelName::new("app", "events_1"), + RelName::new("app", "events_audit"), + RelName::new("app", "orders"), + ]; + let scoped = r.rules.pattern_scoped(|| present.to_vec(), |_| false); + assert_eq!(scoped.len(), 2, "opt-in and opt-out both dispatch"); + let opted: Vec<_> = scoped + .iter() + .filter(|(_, row)| row.replicate == Some(true)) + .map(|(rel, _)| rel.name.to_string()) + .collect(); + assert_eq!(opted, ["events_1"]); + assert_eq!( + r.rules.settings(&present[1]).replicate, + Some(false), + "an excluding pattern is a guardrail: it beats a matching opt-in" + ); + } + + #[test] + fn overlay_pattern_retargets_a_mapped_relation() { + let base = EmitterConfig::from_toml_str( + "[ch]\n\ + [table.app.events_1]\n\ + columns = [{ attnum = 1, target = \"id\", type = \"UInt64\" }]\n", + ) + .unwrap(); + let mut overlay = ConfigOverlay::default(); + overlay.tables.insert( + RelName::new("app", "events_*"), + TableRow { + match_kind: Some("glob".into()), + target_database: Some("warehouse".into()), + ..Default::default() + }, + ); + let (r, _) = ConfigResolver::resolve( + &base, + &overlay, + &CliOverrides::default(), + &OptInState::default(), + &ColumnRules::default(), + ); + let t = r.tables.get(&RelName::new("app", "events_1")).unwrap(); + assert_eq!(t.target.database, "warehouse"); + assert_eq!(t.columns.len(), 1, "TOML projection preserved"); + } + + #[test] + fn overlay_bad_pattern_rejected_and_counted() { + let base = EmitterConfig::from_toml_str("[ch]\n").unwrap(); + let mut overlay = ConfigOverlay::default(); + overlay.tables.insert( + RelName::new("app", "ev(nt"), + TableRow { + match_kind: Some("regex".into()), + replicate: Some(true), + ..Default::default() + }, + ); + overlay.tables.insert( + RelName::new("app", "orders"), + TableRow { + match_kind: Some("like".into()), + ..Default::default() + }, + ); + let (r, rejections) = ConfigResolver::resolve( + &base, + &overlay, + &CliOverrides::default(), + &OptInState::default(), + &ColumnRules::default(), + ); + assert_eq!(rejections, 2, "unparseable regex + unknown match kind"); + assert!(!r.rules.has_patterns()); + } + + #[test] + fn overlay_row_overrides_toml_entry() { + let base = EmitterConfig::from_toml_str( + "[ch]\n\ + [table.app.\"events_*\"]\n\ + match = \"glob\"\n\ + replicate = true\n\ + target_database = \"toml_db\"\n", + ) + .unwrap(); + let mut overlay = ConfigOverlay::default(); + overlay.tables.insert( + RelName::new("app", "events_*"), + TableRow { + match_kind: Some("glob".into()), + target_database: Some("sql_db".into()), + ..Default::default() + }, + ); + let (r, _) = ConfigResolver::resolve( + &base, + &overlay, + &CliOverrides::default(), + &OptInState::default(), + &ColumnRules::default(), + ); + let rel = RelName::new("app", "events_2026"); + 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); + assert_eq!(ddl.declared_scope(&rel), Some(true)); + } + fn auto_create_set(base: &EmitterConfig, overlay: &ConfigOverlay) -> ahash::HashSet { use crate::emit::ch_ddl::DdlConfig; let (r, _) = ConfigResolver::resolve( @@ -971,7 +1170,7 @@ mod tests { overlay, &CliOverrides::default(), &OptInState::default(), - &HashMap::new(), + &ColumnRules::default(), ); DdlConfig::from_resolved(&r, "db".into(), false, false, None).auto_create_namespaces } @@ -1104,7 +1303,7 @@ mod tests { &ConfigOverlay::default(), &CliOverrides::default(), &opt_in, - &HashMap::new(), + &ColumnRules::default(), ); assert!( r.tables.contains_key(&RelName::new("public", "events")), @@ -1307,12 +1506,14 @@ mod tests { (RelName::new("public", "t"), "amount".into()), ColumnRow { target_type: Some("Int128".into()), + ..ColumnRow::default() }, ); overlay.columns.insert( (RelName::new("public", "t"), "bad".into()), ColumnRow { target_type: Some("NotAType(".into()), + ..ColumnRow::default() }, ); let (r, rej) = ConfigResolver::resolve( @@ -1320,15 +1521,88 @@ mod tests { &overlay, &CliOverrides::default(), &OptInState::default(), - &HashMap::new(), + &ColumnRules::default(), ); assert_eq!(rej, 1, "unparseable type rejected"); - let t = r - .columns - .get(&RelName::new("public", "t")) - .expect("table entry"); - assert_eq!(t.get("amount").map(String::as_str), Some("Int128")); - assert!(!t.contains_key("bad")); + let rel = RelName::new("public", "t"); + assert_eq!( + r.column_rules.settings(&rel, "amount").target_type, + Some("Int128".into()) + ); + assert!(r.column_rules.settings(&rel, "bad").target_type.is_none()); + } + + #[test] + fn column_rules_layer_toml_under_a_pattern_overlay_row() { + use crate::runtime_config::ColumnRow; + let base = EmitterConfig::from_toml_str( + "[ch]\n\ + [table.app.events]\n\ + replicate = true\n\ + columns = [{ name = \"amount\", target = \"amt\", type = \"Decimal(38, 2)\" }]\n", + ) + .unwrap(); + let mut overlay = ConfigOverlay::default(); + overlay.columns.insert( + (RelName::new("app", "*"), "amount".into()), + ColumnRow { + target_type: Some("Int128".into()), + match_kind: Some("glob".into()), + }, + ); + let (r, rej) = ConfigResolver::resolve( + &base, + &overlay, + &CliOverrides::default(), + &OptInState::default(), + &ColumnRules::default(), + ); + assert_eq!(rej, 0); + let s = r + .column_rules + .settings(&RelName::new("app", "events"), "amount"); + assert_eq!( + s.target_name.as_deref(), + Some("amt"), + "TOML names the CH column: the overlay row states no name" + ); + assert_eq!( + s.target_type.as_deref(), + Some("Decimal(38, 2)"), + "a literal entry outranks a pattern however late its layer" + ); + assert_eq!( + r.column_rules + .settings(&RelName::new("app", "orders"), "amount") + .target_type + .as_deref(), + Some("Int128") + ); + assert!( + r.column_rules + .settings(&RelName::new("other", "events"), "amount") + .target_type + .is_none() + ); + overlay.columns.insert( + (RelName::new("app", "events"), "amount".into()), + ColumnRow { + target_type: Some("Int128".into()), + match_kind: None, + }, + ); + let (r, _) = ConfigResolver::resolve( + &base, + &overlay, + &CliOverrides::default(), + &OptInState::default(), + &ColumnRules::default(), + ); + let s = r + .column_rules + .settings(&RelName::new("app", "events"), "amount"); + assert_eq!(s.target_type.as_deref(), Some("Int128")); + assert_eq!(s.target_name.as_deref(), Some("amt"), "name still TOML's"); } #[test] @@ -1341,6 +1615,7 @@ mod tests { key.clone(), ColumnRow { target_type: Some("Decimal(38, 2)".into()), + ..ColumnRow::default() }, ); let (first, rej) = ConfigResolver::resolve( @@ -1348,7 +1623,7 @@ mod tests { &overlay, &CliOverrides::default(), &OptInState::default(), - &HashMap::new(), + &ColumnRules::default(), ); assert_eq!(rej, 0); // Malformed update replaces the overlay row wholesale; merge keeps @@ -1357,6 +1632,7 @@ mod tests { key, ColumnRow { target_type: Some("NotAType(".into()), + ..ColumnRow::default() }, ); let (second, rej) = ConfigResolver::resolve( @@ -1364,14 +1640,13 @@ mod tests { &overlay, &CliOverrides::default(), &OptInState::default(), - &first.columns, + &first.column_rules, ); assert_eq!(rej, 1); let amount = |r: &ResolvedConfig| { - r.columns - .get(&RelName::new("public", "t")) - .and_then(|t| t.get("amount")) - .cloned() + r.column_rules + .settings(&RelName::new("public", "t"), "amount") + .target_type }; assert_eq!(amount(&second).as_deref(), Some("Decimal(38, 2)")); // Retention carries forward while the bad row stays in the overlay @@ -1380,7 +1655,7 @@ mod tests { &overlay, &CliOverrides::default(), &OptInState::default(), - &second.columns, + &second.column_rules, ); assert_eq!(rej, 1); assert_eq!(amount(&third).as_deref(), Some("Decimal(38, 2)")); @@ -1403,19 +1678,28 @@ mod tests { attname: "amount".into(), row: ColumnRow { target_type: Some(ty.into()), + ..ColumnRow::default() }, }; resolver.apply_config_event(upsert("Decimal(38, 2)")).await; assert!(rx.changed().await.is_ok()); assert_eq!( - rx.borrow_and_update().columns[&RelName::new("public", "t")]["amount"], - "Decimal(38, 2)" + rx.borrow_and_update() + .column_rules + .settings(&RelName::new("public", "t"), "amount") + .target_type + .as_deref(), + Some("Decimal(38, 2)") ); resolver.apply_config_event(upsert("NotAType(")).await; assert!(rx.changed().await.is_ok()); assert_eq!( - rx.borrow_and_update().columns[&RelName::new("public", "t")]["amount"], - "Decimal(38, 2)", + rx.borrow_and_update() + .column_rules + .settings(&RelName::new("public", "t"), "amount") + .target_type + .as_deref(), + Some("Decimal(38, 2)"), "malformed update keeps last accepted override" ); assert_eq!(resolver.rejections(), 1); @@ -1427,11 +1711,7 @@ mod tests { }) .await; assert!(rx.changed().await.is_ok()); - assert!( - !rx.borrow_and_update() - .columns - .contains_key(&RelName::new("public", "t")) - ); + assert!(rx.borrow_and_update().column_rules.is_empty()); assert_eq!(resolver.rejections(), 0, "gauge clears with the bad row"); } @@ -1556,7 +1836,7 @@ mod tests { &ConfigOverlay::default(), &CliOverrides::default(), &OptInState::default(), - &HashMap::new(), + &ColumnRules::default(), ); assert_eq!(r.source.host, "pg-b"); assert_eq!(r.source.port, 5433); @@ -1576,7 +1856,7 @@ mod tests { &ConfigOverlay::default(), &CliOverrides::default(), &OptInState::default(), - &HashMap::new(), + &ColumnRules::default(), ); assert_eq!(r.source.slot.as_deref(), Some("target_phys")); @@ -1589,7 +1869,7 @@ mod tests { &ConfigOverlay::default(), &cli, &OptInState::default(), - &HashMap::new(), + &ColumnRules::default(), ); assert_eq!(r.source.slot.as_deref(), Some("pinned")); } diff --git a/src/emit/ch_ddl.rs b/src/emit/ch_ddl.rs index dfedacb7..dcb594f9 100644 --- a/src/emit/ch_ddl.rs +++ b/src/emit/ch_ddl.rs @@ -31,13 +31,16 @@ use crate::ch::{ EmitterError, backoff_step, connect_client, exec_drain, is_retryable, quote_ident, reconnect_if_idle, }; +use crate::column_rules::ColumnRules; use crate::config::{ConfigResolver, ResolvedConfig}; use crate::emit::ch_emitter::{EmitterConfig, RetryConfig}; use crate::mapping::{ ColumnMapping, DropTableStrategy, MappingHandle, MappingSnapshot, NamespaceMapping, - TableMapping, TableTarget, derive_columns_for_mapping, fold_diff_into_mapping, + 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 ahash::{HashMap, HashSet, HashSetExt}; /// Knobs that don't ride the INSERT pump. [`DdlApplicator`] rebuilds them @@ -62,6 +65,8 @@ pub struct DdlConfig { /// Keep `_is_deleted` out of `ReplacingMergeTree`'s args so deletes /// stay queryable; mirrors [`EmitterConfig::soft_delete`] pub soft_delete: bool, + pub rules: Arc, + pub column_rules: Arc, } impl DdlConfig { @@ -91,6 +96,8 @@ impl DdlConfig { target_database, namespaces: resolved.namespaces.clone(), soft_delete, + rules: resolved.rules.clone(), + column_rules: resolved.column_rules.clone(), } } @@ -113,9 +120,42 @@ impl DdlConfig { self.drop_table_strategy = s; self } + + fn create_target(&self, rel: &RelName) -> TableTarget { + let settings = self.rules.settings(rel); + TableTarget { + database: settings + .target_database + .unwrap_or_else(|| self.target_database_for(&rel.namespace).to_owned()), + table: settings + .target_table + .unwrap_or_else(|| rel.name.to_string()), + } + } + + /// Resolve explicit scope without opting in system relations + pub fn declared_scope(&self, rel: &RelName) -> Option { + match self.rules.settings(rel).replicate { + Some(true) + if is_system_namespace(&rel.namespace, self.runtime_config_schema.as_deref()) => + { + None + } + other => other, + } + } + + fn auto_creates(&self, rel: &RelName) -> bool { + if let Some(replicate) = self.declared_scope(rel) { + return replicate; + } + self.auto_create_namespaces.contains(&*rel.namespace) + || (self.replicate_all + && !is_system_namespace(&rel.namespace, self.runtime_config_schema.as_deref())) + } } -fn is_system_namespace(ns: &str, runtime_config_schema: Option<&str>) -> bool { +pub(crate) fn is_system_namespace(ns: &str, runtime_config_schema: Option<&str>) -> bool { ns == "pg_catalog" || ns == "information_schema" || ns == "pg_toast" @@ -270,31 +310,29 @@ impl DdlApplicator { self.stats.skipped += 1; return Ok(()); } - let ns = &*desc.rel_name.namespace; - let auto = self.config.auto_create_namespaces.contains(ns) - || (self.config.replicate_all - && !is_system_namespace(ns, self.config.runtime_config_schema.as_deref())); - if !auto { + if !self.config.auto_creates(&desc.rel_name) { self.stats.skipped += 1; return Ok(()); } // Drives both CREATE TABLE and the row-routing mapping below so - // rows and DDL land in the same database - let target_db = self - .config - .target_database_for(&desc.rel_name.namespace) - .to_owned(); - let Some(sql) = render_create_table(desc, &target_db, self.config.soft_delete)? else { + // 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, + )? + else { self.stats.skipped += 1; return Ok(()); }; - self.ensure_database(&target_db).await?; + self.ensure_database(&target.database).await?; self.execute(&sql).await?; self.stats.creates_applied += 1; // Auto-derive a TableMapping so the emitter ships rows against // the new CH table without TOML edits - let target = TableTarget::new(&target_db, &desc.rel_name.name); - let columns = derive_columns_for_mapping(desc); + let columns = derive_columns_for_mapping(desc, &self.config.column_rules); let mapping = TableMapping { target, columns }; self.register_mapping(&desc.rel_name, mapping).await; Ok(()) @@ -309,11 +347,14 @@ 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_db = self - .config - .target_database_for(&desc.rel_name.namespace) - .to_owned(); - let Some(sql) = render_create_table(desc, &target_db, self.config.soft_delete)? else { + 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, + )? + else { tracing::warn!( target: "walshadow::ch_ddl", qname = %desc.rel_name, @@ -322,7 +363,7 @@ impl DdlApplicator { self.stats.skipped += 1; return Ok(false); }; - self.ensure_database(&target_db).await?; + self.ensure_database(&target.database).await?; self.execute(&sql).await?; self.stats.creates_applied += 1; Ok(true) @@ -373,7 +414,12 @@ impl DdlApplicator { self.stats.skipped += 1; continue; }; - let sql = render_add_column(&target, &att.name, &resolved); + let (name, resolved) = apply_column_rule( + &att.name, + resolved, + self.config.column_rules.settings(&new.rel_name, &att.name), + ); + let sql = render_add_column(&target, &name, &resolved); self.execute(&sql).await?; self.stats.alters_applied += 1; } @@ -537,7 +583,7 @@ impl DdlApplicator { if let Some(r) = &self.resolver { r.apply_schema_diff(new, diff).await; } else { - mutate_mapping_for_diff(&self.mapping, new, diff).await; + mutate_mapping_for_diff(&self.mapping, new, diff, &self.config.column_rules).await; } } @@ -604,20 +650,22 @@ fn predict_route_effect( ) -> Result)>, EmitterError> { match event { SchemaEvent::Added { desc } => { + // `replicate_all` is not predicted: it maps on first sight, which + // the executor's own apply covers if mapping.contains_key(&desc.rel_name) || excluded - || !cfg + || !(cfg .auto_create_namespaces .contains(&*desc.rel_name.namespace) + || cfg.declared_scope(&desc.rel_name) == Some(true)) { return Ok(None); } - let target_db = cfg.target_database_for(&desc.rel_name.namespace).to_owned(); - if render_create_table(desc, &target_db, cfg.soft_delete)?.is_none() { + let target = cfg.create_target(&desc.rel_name); + if render_create_table(desc, &target, cfg.soft_delete, &cfg.column_rules)?.is_none() { return Ok(None); } - let target = TableTarget::new(&target_db, &desc.rel_name.name); - let columns = derive_columns_for_mapping(desc); + let columns = derive_columns_for_mapping(desc, &cfg.column_rules); Ok(Some(( desc.rel_name.clone(), Some(TableMapping { target, columns }), @@ -627,7 +675,7 @@ fn predict_route_effect( let Some(mut m) = mapping.get(&new.rel_name).cloned() else { return Ok(None); }; - fold_diff_into_mapping(&mut m, new, diff); + fold_diff_into_mapping(&mut m, new, diff, &cfg.column_rules); Ok(Some((new.rel_name.clone(), Some(m)))) } SchemaEvent::Dropped { rel_name, .. } => { @@ -662,11 +710,16 @@ async fn mapping_columns_at( /// whose `target_name` still equals the OLD source name; an operator-pinned /// different name is left alone (CH runs no ALTER for it either, see /// `apply_changed`) -async fn mutate_mapping_for_diff(mapping: &MappingHandle, new: &RelDescriptor, diff: &SchemaDiff) { +async fn mutate_mapping_for_diff( + mapping: &MappingHandle, + new: &RelDescriptor, + diff: &SchemaDiff, + rules: &ColumnRules, +) { mapping .mutate(|m| { if let Some(target_mapping) = Arc::make_mut(m).get_mut(&new.rel_name) { - fold_diff_into_mapping(target_mapping, new, diff); + fold_diff_into_mapping(target_mapping, new, diff, rules); } }) .await; @@ -723,12 +776,13 @@ fn render_create_sql( /// when a column's type can't be bridged; caller logs + skips. pub fn render_create_table( desc: &RelDescriptor, - target_database: &str, + target: &TableTarget, soft_delete: bool, + rules: &ColumnRules, ) -> Result, EmitterError> { - let target = TableTarget::new(target_database, &desc.rel_name.name).sql(); + let target = target.sql(); let pk_attnums = replident_key_attnums(desc); - let mut col_defs: Vec = Vec::with_capacity(desc.attributes.len() + 4); + let mut cols = Vec::with_capacity(desc.attributes.len()); for att in &desc.attributes { if att.dropped { continue; @@ -739,20 +793,29 @@ pub fn render_create_table( // TOML override and re-triggers via Added on next refetch return Ok(None); }; - let mut def = format!("{} {}", quote_ident(&att.name), resolved.ch_type); - if let Some(d) = resolved.default_sql { - def.push_str(" DEFAULT "); - def.push_str(&d); - } - col_defs.push(def); + let (name, resolved) = apply_column_rule( + &att.name, + resolved, + rules.settings(&desc.rel_name, &att.name), + ); + cols.push((att.attnum, quote_ident(&name), resolved)); } + let col_defs: Vec = cols + .iter() + .map(|(_, name, r)| { + r.default_sql.as_ref().map_or_else( + || format!("{name} {}", r.ch_type), + |d| format!("{name} {} DEFAULT {d}", r.ch_type), + ) + }) + .collect(); + // ClickHouse rejects Nullable columns in ORDER BY let key_names: Vec = pk_attnums .iter() .filter_map(|a| { - desc.attributes - .iter() - .find(|att| att.attnum == *a && !att.dropped) - .map(|att| quote_ident(&att.name)) + cols.iter() + .find(|(attnum, _, r)| attnum == a && !r.ch_type.starts_with("Nullable(")) + .map(|(_, name, _)| name.clone()) }) .collect(); Ok(Some(render_create_sql( @@ -796,8 +859,12 @@ pub fn render_create_table_from_mapping( mod tests { use super::*; use crate::mapping::{ColumnMapping, TableMapping}; + fn dest(database: &str, desc: &RelDescriptor) -> TableTarget { + TableTarget::new(database, &desc.rel_name.name) + } use crate::schema::{INT4OID, TEXTOID, TIMESTAMPTZOID}; use crate::schema::{RelAttr, RelDescriptor, ReplIdent, SchemaDiff}; + use crate::table_rules::MatchKind; #[test] fn system_namespaces_excluded_from_replicate_all() { @@ -840,6 +907,8 @@ mod tests { target_database: "default".into(), namespaces, soft_delete: false, + rules: Arc::default(), + column_rules: Arc::default(), }; assert_eq!(cfg.target_database_for("analytics"), "warehouse"); assert_eq!(cfg.target_database_for("logs"), "default"); @@ -863,6 +932,8 @@ mod tests { target_database: "default".into(), namespaces: HashMap::new(), soft_delete: false, + rules: Arc::default(), + column_rules: Arc::default(), }; assert_eq!(cfg.drop_table_strategy, DropTableStrategy::Retain); let cfg = cfg.with_drop_strategy(DropTableStrategy::Drop); @@ -909,6 +980,68 @@ mod tests { } } + #[test] + fn pattern_entry_decides_scope_and_destination() { + use crate::table_rules::{TableRule, TableRulesBuilder}; + let mut b = TableRulesBuilder::new(); + b.add( + &RelName::new("app", "events_*"), + MatchKind::Glob, + TableRule { + replicate: Some(true), + target_database: Some("warehouse".into()), + ..TableRule::default() + }, + ); + b.add( + &RelName::new("app", "*_audit"), + MatchKind::Glob, + TableRule { + replicate: Some(false), + ..TableRule::default() + }, + ); + b.add( + &RelName::new("*", "*"), + MatchKind::Glob, + TableRule { + replicate: Some(true), + ..TableRule::default() + }, + ); + let (rules, rejected) = b.finish(); + assert_eq!(rejected, 0); + let cfg = DdlConfig { + drop_table_strategy: DropTableStrategy::Retain, + auto_create_namespaces: HashSet::new(), + replicate_all: false, + runtime_config_schema: Some("walshadow".into()), + target_database: "default".into(), + namespaces: ahash::HashMap::default(), + soft_delete: false, + 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), + TableTarget::new("warehouse", "events_1") + ); + assert_eq!( + cfg.declared_scope(&RelName::new("app", "events_audit")), + Some(false), + "guardrail beats the matching opt-in" + ); + 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"))); + assert_eq!( + cfg.create_target(&RelName::new("other", "t")), + TableTarget::new("default", "t") + ); + } + #[test] fn render_add_column_emits_idempotent_alter_with_default() { let resolved = ResolvedColumn { @@ -935,6 +1068,65 @@ mod tests { ); } + #[test] + fn render_create_table_states_the_rule_name_and_type() { + let d = desc( + "orders", + vec![ + att(1, "id", INT4OID, true, None), + att(2, "net_amount", INT4OID, false, None), + ], + Some(vec![1]), + ); + let mut b = crate::column_rules::ColumnRulesBuilder::new(); + b.add( + &RelName::new("public", "*"), + MatchKind::Glob, + "*_amount", + MatchKind::Glob, + crate::column_rules::ColumnRule { + target_name: None, + target_type: Some("Decimal(38, 9)".into()), + }, + ); + b.add( + &RelName::new("public", "orders"), + MatchKind::Exact, + "id", + MatchKind::Exact, + crate::column_rules::ColumnRule { + target_name: Some("order_id".into()), + target_type: None, + }, + ); + let sql = render_create_table(&d, &dest("db", &d), false, &b.finish().0) + .unwrap() + .expect("renderable"); + assert!(sql.contains("`order_id` Int32"), "{sql}"); + assert!(sql.contains("`net_amount` Decimal(38, 9)"), "{sql}"); + assert!(sql.ends_with("ORDER BY (`order_id`)"), "{sql}"); + } + + #[test] + fn render_create_table_drops_a_key_a_rule_made_nullable() { + let d = desc("t", vec![att(1, "id", INT4OID, true, None)], Some(vec![1])); + let mut b = crate::column_rules::ColumnRulesBuilder::new(); + b.add( + &RelName::new("public", "t"), + MatchKind::Exact, + "id", + MatchKind::Exact, + crate::column_rules::ColumnRule { + target_name: None, + target_type: Some("Nullable(Int32)".into()), + }, + ); + let sql = render_create_table(&d, &dest("db", &d), false, &b.finish().0) + .unwrap() + .expect("renderable"); + assert!(sql.ends_with("ORDER BY (`_lsn`)"), "{sql}"); + } + #[test] fn render_create_table_uses_pk_for_order_by() { let d = desc( @@ -945,7 +1137,9 @@ mod tests { ], Some(vec![1]), ); - let sql = render_create_table(&d, "default", false).unwrap().unwrap(); + let sql = render_create_table(&d, &dest("default", &d), 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)")); @@ -971,7 +1165,9 @@ mod tests { d.replident = ReplIdent::Full { pk_attnums: Some(vec![1]), }; - let sql = render_create_table(&d, "default", false).unwrap().unwrap(); + let sql = render_create_table(&d, &dest("default", &d), false, &ColumnRules::default()) + .unwrap() + .unwrap(); assert!(sql.ends_with("ORDER BY (`id`)"), "{sql}"); assert!(!sql.contains("ORDER BY _lsn"), "{sql}"); } @@ -988,7 +1184,9 @@ mod tests { ], Some(vec![2, 1]), ); - let sql = render_create_table(&d, "default", false).unwrap().unwrap(); + let sql = render_create_table(&d, &dest("default", &d), false, &ColumnRules::default()) + .unwrap() + .unwrap(); assert!(sql.contains("`a` Int32"), "{sql}"); assert!(sql.contains("`b` Int32"), "{sql}"); assert!(!sql.contains("`a` Nullable"), "{sql}"); @@ -1007,7 +1205,9 @@ mod tests { ], Some(vec![1]), ); - let sql = render_create_table(&d, "default", true).unwrap().unwrap(); + let sql = render_create_table(&d, &dest("default", &d), 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`)")); @@ -1018,7 +1218,9 @@ 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, "default", false).unwrap().unwrap(); + let sql = render_create_table(&d, &dest("default", &d), false, &ColumnRules::default()) + .unwrap() + .unwrap(); assert!(sql.ends_with("ORDER BY (`_lsn`)")); } @@ -1038,7 +1240,9 @@ mod tests { index_oid: 16500, key_attnums: vec![2, 1], }; - let sql = render_create_table(&d, "default", false).unwrap().unwrap(); + let sql = render_create_table(&d, &dest("default", &d), false, &ColumnRules::default()) + .unwrap() + .unwrap(); assert!(!sql.contains("`key` Nullable"), "{sql}"); assert!(!sql.contains("`tenant` Nullable"), "{sql}"); assert!(sql.contains("`body` Nullable(String)"), "{sql}"); @@ -1053,7 +1257,9 @@ mod tests { vec![att(2, "body", TEXTOID, false, None)], Some(vec![1]), ); - let sql = render_create_table(&d, "default", false).unwrap().unwrap(); + let sql = render_create_table(&d, &dest("default", &d), false, &ColumnRules::default()) + .unwrap() + .unwrap(); assert!(sql.ends_with("ORDER BY (`_lsn`)"), "{sql}"); } @@ -1062,7 +1268,9 @@ 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, "db", false).unwrap().unwrap(); + let sql = render_create_table(&d, &dest("db", &d), false, &ColumnRules::default()) + .unwrap() + .unwrap(); assert!( sql.contains("`ship_at` Nullable(DateTime64(3, 'UTC'))"), "{sql}" @@ -1143,7 +1351,7 @@ 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, "db", false).unwrap(); + let sql = render_create_table(&d, &dest("db", &d), false, &ColumnRules::default()).unwrap(); assert!(sql.is_some(), "fallback path keeps the CREATE renderable"); } @@ -1218,6 +1426,8 @@ mod tests { target_database: "default".into(), namespaces: ahash::HashMap::default(), soft_delete: false, + rules: Arc::default(), + column_rules: Arc::default(), }; let dropped = SchemaEvent::Dropped { oid: 16400, @@ -1255,7 +1465,7 @@ mod tests { renamed_columns: vec![], type_changes: vec![], }; - mutate_mapping_for_diff(&handle, &new, &diff).await; + mutate_mapping_for_diff(&handle, &new, &diff, &ColumnRules::default()).await; let folded = handle .with(|m| { m.get(&RelName::new("public", "orders")) @@ -1269,6 +1479,6 @@ mod tests { // Unmapped relation: early return let ghost = desc("ghost", vec![att(1, "id", INT4OID, true, None)], None); - mutate_mapping_for_diff(&handle, &ghost, &diff).await; + mutate_mapping_for_diff(&handle, &ghost, &diff, &ColumnRules::default()).await; } } diff --git a/src/emit/ch_emitter.rs b/src/emit/ch_emitter.rs index 40ff266b..fa0ae463 100644 --- a/src/emit/ch_emitter.rs +++ b/src/emit/ch_emitter.rs @@ -30,6 +30,7 @@ use clickhouse_c::{Allocator, ColumnBuilder, Kind, TypeAst}; #[cfg(test)] use crate::ch::is_retryable; use crate::ch::{CompressionChoice, ConnectionConfig, EmitterError, quote_ident}; +use crate::column_rules::{ColumnEntry, ColumnRule, ColumnRules}; #[cfg(test)] use crate::decode::decoder_sink::DecoderSinkError; use crate::decode::heap_decoder::{ColumnValue, CommittedTuple, HeapOp}; @@ -41,6 +42,7 @@ use crate::schema::{RelDescriptor, RelName}; use crate::source::queueing_record_sink::{ DEFAULT_QUEUEING_BATCH_SIZE, DEFAULT_QUEUEING_RECORD_SINK_CAPACITY, }; +use crate::table_rules::{MatchKind, TableRule}; use ahash::{HashMap, HashMapExt}; /// Microseconds between PG `TimestampTz` epoch (2000-01-01 UTC) and Unix @@ -113,6 +115,10 @@ pub struct EmitterConfig { /// Per-table initial-load mode from TOML `[table.*]` blocks. Applies at /// boot for pinned mappings; SQL opt-ins carry their own mode. pub table_initial_loads: HashMap, + /// Table rules in declaration order + pub table_entries: Vec<(RelName, MatchKind, TableRule)>, + /// Column rules in declaration order + pub column_entries: Vec, pub table_opt_ins: HashMap, /// `[stream] paused`: pump idles (stops consuming source WAL) when true. /// Live via reload. @@ -305,6 +311,8 @@ impl Default for EmitterConfig { flush_timeout: Duration::from_millis(DEFAULT_FLUSH_TIMEOUT_MS), tables: HashMap::new(), table_initial_loads: HashMap::new(), + table_entries: Vec::new(), + column_entries: Vec::new(), table_opt_ins: HashMap::new(), paused: false, replicate_all: true, @@ -699,97 +707,178 @@ impl EmitterConfig { })?; let replicate = t.get("replicate").and_then(Value::as_bool); let rel = RelName::new(ns, name); - let Some(cols_v) = t.get("columns").and_then(Value::as_array) else { + let ctx = format!("table.{ns}.{name}"); + let kind = match t.get("match") { + None => MatchKind::Exact, + Some(v) => { + let s = v.as_str().ok_or_else(|| { + EmitterError::Config(format!("{ctx}.match: expected a string")) + })?; + MatchKind::parse(s) + .map_err(|e| EmitterError::Config(format!("{ctx}.match: {e}")))? + } + }; + let rule = TableRule { + target_database: t + .get("target_database") + .and_then(Value::as_str) + .map(String::from), + target_table: t + .get("target_table") + .and_then(Value::as_str) + .map(String::from), + replicate, + initial_load: t + .get("initial_load") + .and_then(Value::as_str) + .map(String::from), + }; + out.table_entries.push((rel.clone(), kind, rule.clone())); + let cols_v = match t.get("columns") { + None => &[], + Some(v) => v.as_array().map(Vec::as_slice).ok_or_else(|| { + EmitterError::Config(format!("{ctx}.columns: expected an array")) + })?, + }; + let mut pinned = Vec::new(); + let mut named = Vec::new(); + for (i, c) in cols_v.iter().enumerate() { + let ctx = format!("{ctx}.columns[{i}]"); + let ct = c.as_table().ok_or_else(|| { + EmitterError::Config(format!("{ctx}: expected a table")) + })?; + let str_field = |key: &str| -> Result, EmitterError> { + match ct.get(key) { + None => Ok(None), + Some(v) => { + v.as_str().map(|s| Some(s.to_string())).ok_or_else(|| { + EmitterError::Config(format!( + "{ctx}.{key}: expected a string" + )) + }) + } + } + }; + let target = str_field("target")?; + let target_type = str_field("type")?; + let att_kind = match str_field("match")? { + None => MatchKind::Exact, + Some(s) => MatchKind::parse(&s) + .map_err(|e| EmitterError::Config(format!("{ctx}.match: {e}")))?, + }; + match (ct.get("attnum"), str_field("name")?) { + (Some(_), Some(_)) => { + return Err(EmitterError::Config(format!( + "{ctx}: attnum and name are alternatives, not both" + ))); + } + (None, None) => { + return Err(EmitterError::Config(format!( + "{ctx}: missing attnum or name" + ))); + } + (Some(a), None) => { + if ct.contains_key("match") { + return Err(EmitterError::Config(format!( + "{ctx}.match: an attnum entry names one column already" + ))); + } + let src_attnum = a.as_integer().ok_or_else(|| { + EmitterError::Config(format!( + "{ctx}.attnum: expected an integer" + )) + })?; + pinned.push(ColumnMapping { + src_attnum: i16::try_from(src_attnum).map_err(|_| { + EmitterError::Config(format!( + "{ctx}.attnum {src_attnum} out of i16 range" + )) + })?, + target_name: target.ok_or_else(|| { + EmitterError::Config(format!("{ctx}: missing target")) + })?, + target_type: target_type.ok_or_else(|| { + EmitterError::Config(format!("{ctx}: missing type")) + })?, + }); + } + (None, Some(attname)) => { + if target.is_some() && att_kind != MatchKind::Exact { + return Err(EmitterError::Config(format!( + "{ctx}.target: a `match = \"{}\"` entry can name \ + several columns, which cannot share one target", + att_kind.as_str() + ))); + } + if target.is_none() && target_type.is_none() { + return Err(EmitterError::Config(format!( + "{ctx}: sets neither target nor type" + ))); + } + named.push(ColumnEntry { + rel: rel.clone(), + rel_kind: kind, + attname, + att_kind, + rule: ColumnRule { + target_name: target, + target_type, + }, + }); + } + } + } + if !pinned.is_empty() && !named.is_empty() { + return Err(EmitterError::Config(format!( + "{ctx}.columns: attnum entries pin the whole projection, so a \ + name entry beside them would never apply" + ))); + } + out.column_entries.append(&mut named); + if kind != MatchKind::Exact { + if !pinned.is_empty() { + return Err(EmitterError::Config(format!( + "{ctx}.columns: a `match = \"{}\"` entry cannot pin attnums; \ + key the entries on `name` instead", + kind.as_str() + ))); + } + continue; + } + if pinned.is_empty() { out.table_opt_ins.insert( rel, TableRow { - target_database: t - .get("target_database") - .and_then(Value::as_str) - .map(String::from), - target_table: t - .get("target_table") - .and_then(Value::as_str) - .map(String::from), + target_database: rule.target_database, + target_table: rule.target_table, replicate, - initial_load: t - .get("initial_load") - .and_then(Value::as_str) - .map(String::from), + initial_load: rule.initial_load, + ..TableRow::default() }, ); continue; - }; + } if replicate == Some(false) { continue; } - let database = t - .get("target_database") - .and_then(Value::as_str) + let database = rule + .target_database .or_else(|| { out.namespaces .get(ns.as_str()) - .and_then(|n| n.target_database.as_deref()) + .and_then(|n| n.target_database.clone()) }) - .unwrap_or(&out.database) - .to_string(); - let table = t - .get("target_table") - .and_then(Value::as_str) - .unwrap_or(name) - .to_string(); - let mut columns = Vec::with_capacity(cols_v.len()); - for (i, c) in cols_v.iter().enumerate() { - let ct = c.as_table().ok_or_else(|| { - EmitterError::Config(format!( - "table.{ns}.{name}.columns[{i}]: expected a table" - )) - })?; - let src_attnum = - ct.get("attnum") - .and_then(Value::as_integer) - .ok_or_else(|| { - EmitterError::Config(format!( - "table.{ns}.{name}.columns[{i}]: missing attnum" - )) - })?; - let target_name = ct - .get("target") - .and_then(Value::as_str) - .ok_or_else(|| { - EmitterError::Config(format!( - "table.{ns}.{name}.columns[{i}]: missing target" - )) - })? - .to_string(); - let target_type = ct - .get("type") - .and_then(Value::as_str) - .ok_or_else(|| { - EmitterError::Config(format!( - "table.{ns}.{name}.columns[{i}]: missing type" - )) - })? - .to_string(); - columns.push(ColumnMapping { - src_attnum: i16::try_from(src_attnum).map_err(|_| { - EmitterError::Config(format!( - "table.{ns}.{name}.columns[{i}].attnum {src_attnum} out of i16 range" - )) - })?, - target_name, - target_type, - }); - } + .unwrap_or_else(|| out.database.clone()); + let table = rule.target_table.unwrap_or_else(|| name.to_string()); out.tables.insert( rel.clone(), TableMapping { target: TableTarget { database, table }, - columns, + columns: pinned, }, ); - if let Some(mode) = t.get("initial_load").and_then(Value::as_str) { - out.table_initial_loads.insert(rel, mode.to_string()); + if let Some(mode) = rule.initial_load { + out.table_initial_loads.insert(rel, mode); } } } @@ -851,19 +940,12 @@ pub(crate) struct DecimalWire { } impl TablePlan { - /// Synthetic columns always non-nullable (emitter always populates). - /// - /// `column_overrides` is the `config_column` overlay slice for this rel - /// (source attname → CH type). Resolved here because this is where the - /// descriptor meets the mapping: attname→attnum comes from `rel`, and an - /// inadmissible override (see [`override_wire`]) falls back to the - /// mapping's type with a WARN — a config row must degrade, never poison - /// the batcher (Regime A). + /// Synthetic columns are always non-nullable pub(crate) fn build( alloc: Allocator, rel: &RelDescriptor, mapping: &TableMapping, - column_overrides: Option<&HashMap>, + column_rules: &ColumnRules, ) -> Result { let mut columns = Vec::with_capacity(mapping.columns.len()); let mut col_sql = Vec::with_capacity(mapping.columns.len() + 4); @@ -882,13 +964,13 @@ impl TablePlan { ast, decimal, }; - if let Some(ov) = column_overrides - && let Some(ty) = rel - .attributes - .iter() - .find(|a| a.attnum == c.src_attnum && !a.dropped) - .and_then(|a| ov.get(&a.name)) + if let Some(ty) = rel + .attributes + .iter() + .find(|a| a.attnum == c.src_attnum && !a.dropped) + .and_then(|a| column_rules.settings(&rel.rel_name, &a.name).target_type) { + let ty = &ty; match TypeAst::parse(ty, alloc) { Ok(oast) => { if let Some(decimal) = override_wire(&plan.ast, &oast) { @@ -1893,6 +1975,23 @@ mod tests { } } + fn col_rules(entries: &[(&str, &str)]) -> ColumnRules { + let mut b = crate::column_rules::ColumnRulesBuilder::new(); + for (attname, target_type) in entries { + b.add( + &RelName::new("public", "foo"), + MatchKind::Exact, + attname, + MatchKind::Exact, + ColumnRule { + target_type: Some((*target_type).into()), + ..ColumnRule::default() + }, + ); + } + b.finish().0 + } + fn mk_rel() -> RelDescriptor { use crate::schema::{RelAttr, ReplIdent}; use walrus::pg::walparser::RelFileNode; @@ -2228,7 +2327,7 @@ mod tests { let alloc = Allocator::stdlib(); let rel = mk_rel(); let m = mk_mapping(); - let plan = TablePlan::build(alloc, &rel, &m, None).expect("plan builds"); + let plan = TablePlan::build(alloc, &rel, &m, &ColumnRules::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`")); @@ -2246,8 +2345,8 @@ mod tests { let mut m = mk_mapping(); // numeric-shaped default: the plan drill `numeric(38,0)` → `Int128` m.columns[0].target_type = "Decimal(38, 0)".into(); - let overrides = HashMap::from_iter([("id".to_string(), "Int128".to_string())]); - let plan = TablePlan::build(alloc, &rel, &m, Some(&overrides)).unwrap(); + let rules = col_rules(&[("id", "Int128")]); + let plan = TablePlan::build(alloc, &rel, &m, &rules).unwrap(); assert_eq!(plan.columns[0].type_repr, "Int128"); // scale-0 decimal wire keeps the numeric text→scaled encode path assert_eq!( @@ -2267,8 +2366,8 @@ mod tests { let mut m = mk_mapping(); // Operator-renamed CH column: override still keys on source attname m.columns[1].target_name = "label".into(); - let overrides = HashMap::from_iter([("name".to_string(), "String".to_string())]); - let plan = TablePlan::build(alloc, &rel, &m, Some(&overrides)).unwrap(); + let rules = col_rules(&[("name", "String")]); + let plan = TablePlan::build(alloc, &rel, &m, &rules).unwrap(); assert_eq!(plan.columns[1].name, "label"); assert_eq!(plan.columns[1].type_repr, "String"); } @@ -2280,8 +2379,8 @@ mod tests { let m = mk_mapping(); // encode_value writes int4 as 4 LE bytes; no textualization exists, // so Int32 → String must fall back rather than poison the batcher - let overrides = HashMap::from_iter([("id".to_string(), "String".to_string())]); - let plan = TablePlan::build(alloc, &rel, &m, Some(&overrides)).unwrap(); + let rules = col_rules(&[("id", "String")]); + let plan = TablePlan::build(alloc, &rel, &m, &rules).unwrap(); assert_eq!(plan.columns[0].type_repr, "Int32"); } @@ -2320,7 +2419,7 @@ mod tests { let alloc = Allocator::stdlib(); let rel = mk_rel(); let m = mk_mapping(); - let plan = TablePlan::build(alloc, &rel, &m, None).expect("plan builds"); + let plan = TablePlan::build(alloc, &rel, &m, &ColumnRules::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) @@ -2368,7 +2467,7 @@ 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, None).expect("plan builds"); + let plan = TablePlan::build(alloc, &rel, &m, &ColumnRules::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(); @@ -2388,7 +2487,7 @@ mod tests { let alloc = Allocator::stdlib(); let rel = mk_rel(); let m = mk_mapping(); - let plan = TablePlan::build(alloc, &rel, &m, None).expect("plan builds"); + let plan = TablePlan::build(alloc, &rel, &m, &ColumnRules::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] { @@ -2402,7 +2501,7 @@ mod tests { let alloc = Allocator::stdlib(); let rel = mk_rel(); let m = mk_mapping(); - let plan = TablePlan::build(alloc, &rel, &m, None).unwrap(); + let plan = TablePlan::build(alloc, &rel, &m, &ColumnRules::default()).unwrap(); let mut enc = TableEncoder::new(plan).unwrap(); enc.append_row(&committed(7, Some("seven")), &m, OP_INSERT) .unwrap(); @@ -2541,6 +2640,119 @@ mod tests { assert!(!c.table_initial_loads.contains_key(&rel)); } + #[test] + fn config_table_pattern_entry_keys_on_pattern() { + let c = EmitterConfig::from_toml_str( + "[ch]\n\ + [table.public.\"events_*\"]\n\ + match = \"glob\"\n\ + replicate = true\n\ + initial_load = \"copy\"\n", + ) + .unwrap(); + let (rel, kind, rule) = &c.table_entries[0]; + assert_eq!(*rel, RelName::new("public", "events_*")); + assert_eq!(*kind, MatchKind::Glob); + assert_eq!(rule.replicate, Some(true)); + assert_eq!(rule.initial_load.as_deref(), Some("copy")); + assert!(c.table_opt_ins.is_empty()); + } + + #[test] + fn config_table_literal_entry_lands_in_entries_and_opt_ins() { + let c = EmitterConfig::from_toml_str( + "[ch]\n\ + [table.public.events]\n\ + replicate = true\n\ + target_table = \"ev\"\n", + ) + .unwrap(); + let rel = RelName::new("public", "events"); + let (_, kind, rule) = &c.table_entries[0]; + assert_eq!(*kind, MatchKind::Exact); + assert_eq!(rule.target_table.as_deref(), Some("ev")); + assert!(c.table_opt_ins.contains_key(&rel)); + } + + #[test] + fn config_table_rejects_bad_match_and_pattern_columns() { + assert!( + EmitterConfig::from_toml_str("[ch]\n[table.public.t]\nmatch = \"like\"\n").is_err(), + "a typo must not read as a literal name" + ); + assert!( + EmitterConfig::from_toml_str( + "[ch]\n[table.public.\"t.*\"]\nmatch = \"regex\"\n\ + columns = [{ attnum = 1, target = \"id\", type = \"UInt64\" }]\n" + ) + .is_err() + ); + } + + #[test] + fn config_column_name_entries_state_rules_not_a_projection() { + let c = EmitterConfig::from_toml_str( + "[ch]\n\ + [table.app.events]\n\ + replicate = true\n\ + columns = [\n \ + { name = \"legacy_id\", target = \"id\", type = \"UInt64\" },\n \ + { name = \"*_at\", match = \"glob\", type = \"DateTime64(6, 'UTC')\" },\n\ + ]\n", + ) + .unwrap(); + let rel = RelName::new("app", "events"); + assert!(!c.tables.contains_key(&rel)); + assert!(c.table_opt_ins.contains_key(&rel)); + assert_eq!(c.column_entries.len(), 2); + let first = &c.column_entries[0]; + assert_eq!(first.rel, rel); + assert_eq!(first.rel_kind, MatchKind::Exact); + assert_eq!(first.attname, "legacy_id"); + assert_eq!(first.att_kind, MatchKind::Exact); + assert_eq!(first.rule.target_name.as_deref(), Some("id")); + assert_eq!(first.rule.target_type.as_deref(), Some("UInt64")); + let second = &c.column_entries[1]; + assert_eq!(second.att_kind, MatchKind::Glob); + assert!(second.rule.target_name.is_none()); + } + + #[test] + fn config_column_name_entries_ride_a_pattern_block() { + let c = EmitterConfig::from_toml_str( + "[ch]\n\ + [table.app.\"*\"]\n\ + match = \"glob\"\n\ + replicate = true\n\ + columns = [{ name = \"*_at\", match = \"glob\", type = \"DateTime64(6, 'UTC')\" }]\n", + ) + .unwrap(); + let e = &c.column_entries[0]; + assert_eq!(e.rel, RelName::new("app", "*")); + assert_eq!(e.rel_kind, MatchKind::Glob); + assert!(c.table_opt_ins.is_empty(), "a pattern queues no opt-in"); + } + + #[test] + fn config_column_entry_shapes_are_exclusive() { + let bad = [ + "columns = [{ attnum = 1, name = \"id\", type = \"UInt64\" }]", + "columns = [{ target = \"id\", type = \"UInt64\" }]", + "columns = [{ attnum = 1, match = \"glob\", target = \"id\", type = \"UInt64\" }]", + "columns = [{ name = \"*_at\", match = \"glob\", target = \"ts\" }]", + "columns = [{ name = \"id\" }]", + "columns = [{ attnum = 1, target = \"id\", type = \"UInt64\" }, \ + { name = \"ts\", type = \"DateTime\" }]", + ]; + for body in bad { + assert!( + EmitterConfig::from_toml_str(&format!("[ch]\n[table.app.events]\n{body}\n")) + .is_err(), + "{body}" + ); + } + } + #[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 5ca147eb..992cc8e6 100644 --- a/src/emit/pipeline/batcher.rs +++ b/src/emit/pipeline/batcher.rs @@ -308,7 +308,7 @@ async fn handle_row( ctx.alloc, &row.rel, &row.route.mapping, - Some(&row.route.column_overrides), + &row.route.column_rules, ) .map_err(|e| e.to_string())?; let meta = Arc::new(BatchMeta::from_plan(&plan, e.key().clone(), ctx.epoch)); @@ -640,7 +640,17 @@ mod tests { None, ); // Int32 → UInt32 is wire-compatible (same fixed width), admissible - let overrides = HashMap::from_iter([(String::from("id"), String::from("UInt32"))]); + let mut rules = crate::column_rules::ColumnRulesBuilder::new(); + rules.add( + &RelName::new("public", "t"), + crate::table_rules::MatchKind::Exact, + "id", + crate::table_rules::MatchKind::Exact, + crate::column_rules::ColumnRule { + target_type: Some("UInt32".into()), + ..Default::default() + }, + ); let mut r = row(0, 1); r.route = RouteSnapshot::freeze( Arc::new(TableMapping { @@ -651,7 +661,7 @@ mod tests { target_type: "Int32".into(), }], }), - Arc::new(overrides), + Arc::new(rules.finish().0), false, ); msg_tx.send(BatcherMsg::Row(r)).await.expect("send row"); diff --git a/src/emit/pipeline/bootstrap.rs b/src/emit/pipeline/bootstrap.rs index d1976703..c53f9734 100644 --- a/src/emit/pipeline/bootstrap.rs +++ b/src/emit/pipeline/bootstrap.rs @@ -55,15 +55,12 @@ pub async fn drain( .await .iter() .map(|(name, mapping)| { - let overrides = config + let rules = config .as_ref() - .and_then(|rc| rc.columns.get(name)) - .cloned() - .map(Arc::new) - .unwrap_or_default(); + .map_or_else(Arc::default, |rc| rc.column_rules.clone()); ( name.clone(), - RouteSnapshot::freeze(Arc::new(mapping.clone()), overrides, soft_delete), + RouteSnapshot::freeze(Arc::new(mapping.clone()), rules, soft_delete), ) }) .collect(); diff --git a/src/emit/pipeline/plan_spool.rs b/src/emit/pipeline/plan_spool.rs index d4af90cd..ee53f943 100644 --- a/src/emit/pipeline/plan_spool.rs +++ b/src/emit/pipeline/plan_spool.rs @@ -590,7 +590,7 @@ mod tests { target: TableTarget::new("db", "t"), columns: Vec::new(), }), - Arc::new(HashMap::new()), + Arc::default(), false, ) } diff --git a/src/emit/pipeline/planner.rs b/src/emit/pipeline/planner.rs index 95f5d885..5b6388af 100644 --- a/src/emit/pipeline/planner.rs +++ b/src/emit/pipeline/planner.rs @@ -326,7 +326,7 @@ mod tests { target: TableTarget::new("db", "t"), columns: Vec::new(), }), - Arc::new(HashMap::new()), + Arc::default(), false, ) } diff --git a/src/emit/pipeline/reorder.rs b/src/emit/pipeline/reorder.rs index 02394b9c..76590f84 100644 --- a/src/emit/pipeline/reorder.rs +++ b/src/emit/pipeline/reorder.rs @@ -239,9 +239,22 @@ impl ReorderSink { let desired: Vec<(RelName, TableRow)> = { let rx = self.reload_rx.as_mut().unwrap(); let snap = rx.borrow_and_update(); + let config_schema = self + .applicator + .as_ref() + .and_then(|a| a.config().runtime_config_schema.clone()); + // Keep prior pattern opt-ins in desired set + let scoped = snap.rules.pattern_scoped( + || { + self.log + .user_rel_names_at(commit_lsn, config_schema.as_deref()) + }, + |rel| snap.tables.contains_key(rel) && !self.applied_opt_ins.contains(rel), + ); snap.table_opt_ins .iter() .map(|(rel, row)| (rel.clone(), row.clone())) + .chain(scoped) .collect() }; let desired_in: HashSet = desired @@ -364,7 +377,7 @@ impl ReorderSink { // register / drop the descriptor-derived mapping. `commit_lsn` is the // backfill boundary `S` for an `initial_load` opt-in. match event { - ConfigEvent::TableUpserted { rel, row } => { + ConfigEvent::TableUpserted { rel, row } if !row.is_pattern() => { if let Some(applicator) = self.applicator.as_mut() { crate::backfill::opt_in::apply_table_opt_in( &resolver, @@ -379,7 +392,10 @@ impl ReorderSink { .map_err(|e| SinkError::Other(format!("opt-in: {e}")))?; } } - ConfigEvent::TableRemoved { rel } => { + ConfigEvent::TableRemoved { + rel, + pattern: false, + } => { resolver.exclude_table(rel).await; if let Some(b) = &self.backfiller { b.note_opt_out(rel).await; @@ -998,14 +1014,11 @@ impl PlanRouteView for ReorderRouteView<'_> { None => self.mapping.as_ref().and_then(|m| m.get(rel_name)).cloned(), }; let route = mapped.map(|m| { - let overrides = self + let rules = self .config .as_ref() - .and_then(|rc| rc.columns.get(rel_name)) - .cloned() - .map(Arc::new) - .unwrap_or_default(); - RouteSnapshot::freeze(Arc::new(m), overrides, self.soft_delete) + .map_or_else(Arc::default, |rc| rc.column_rules.clone()); + RouteSnapshot::freeze(Arc::new(m), rules, self.soft_delete) }); let result = if route.is_none() { self.stats diff --git a/src/emit/route.rs b/src/emit/route.rs index b43a3735..d0621eb4 100644 --- a/src/emit/route.rs +++ b/src/emit/route.rs @@ -1,21 +1,11 @@ -//! Route envelopes — frozen routing/encoding state attached to rows. -//! -//! [`RouteSnapshot`] freezes the encoder-plan inputs a relation resolved to -//! over one WAL interval: destination mapping, `config_column` overrides, -//! encoding policy. Rows carry the snapshot to the batcher so a mapping or -//! config change never reinterprets rows already routed. +//! Route snapshots freeze routing and encoding inputs for each WAL interval use std::sync::Arc; +use crate::column_rules::ColumnRules; use crate::decode::heap_decoder::DescribedHeap; use crate::mapping::{TableMapping, TableTarget}; -use ahash::HashMap; -/// `config_column` overlay slice for one relation: source attname → CH type -pub type ColumnOverrides = HashMap; - -/// Encoder-plan inputs beyond mapping + overrides. Alloc-free (no parsed -/// type ASTs) so snapshots can serialize and dedup by content #[derive(Debug)] pub struct RowEncodingSnapshot { pub destination: TableTarget, @@ -28,9 +18,7 @@ pub struct RowEncodingSnapshot { #[derive(Debug)] pub struct RouteSnapshot { pub mapping: Arc, - /// Overlay slice frozen with the route, consumed at batcher plan build. - /// Empty when the overlay is off or names no columns for this relation - pub column_overrides: Arc, + pub column_rules: Arc, pub encoding: Arc, } @@ -38,7 +26,7 @@ impl RouteSnapshot { /// Freeze encoder-plan inputs; destination derives from mapping target pub fn freeze( mapping: Arc, - column_overrides: Arc, + column_rules: Arc, soft_delete: bool, ) -> Arc { let encoding = Arc::new(RowEncodingSnapshot { @@ -47,7 +35,7 @@ impl RouteSnapshot { }); Arc::new(Self { mapping, - column_overrides, + column_rules, encoding, }) } diff --git a/src/lib.rs b/src/lib.rs index 9e74389a..ead12eeb 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -22,6 +22,7 @@ pub mod backfill; pub mod budget; pub mod catalog; pub mod ch; +pub mod column_rules; pub mod config; pub mod decode; pub mod emit; @@ -35,6 +36,7 @@ pub mod record; pub mod runtime_config; pub mod schema; pub mod source; +pub mod table_rules; pub mod toast; pub mod xact; diff --git a/src/mapping.rs b/src/mapping.rs index f8accd1d..2e81bc17 100644 --- a/src/mapping.rs +++ b/src/mapping.rs @@ -2,8 +2,9 @@ use std::sync::Arc; -use crate::catalog::type_bridge; -use crate::schema::{RelDescriptor, RelName, SchemaDiff, replident_key_attnums}; +use crate::catalog::type_bridge::{self, ResolvedColumn}; +use crate::column_rules::{ColumnRule, ColumnRules}; +use crate::schema::{RelAttr, RelDescriptor, RelName, SchemaDiff, replident_key_attnums}; use ahash::HashMap; use tokio::sync::RwLock; @@ -145,28 +146,63 @@ pub fn mapping_handle(tables: HashMap) -> MappingHandle { }) } -pub fn derive_columns_for_mapping(desc: &RelDescriptor) -> Vec { +/// CH name and type a rule states for an attribute, falling back to the +/// bridge. A stated type drops the bridge default, which belonged to the +/// type the rule just replaced +pub fn apply_column_rule( + attname: &str, + resolved: ResolvedColumn, + rule: ColumnRule, +) -> (String, ResolvedColumn) { + ( + rule.target_name.unwrap_or_else(|| attname.to_owned()), + rule.target_type.map_or(resolved, |ch_type| ResolvedColumn { + ch_type, + default_sql: None, + }), + ) +} + +pub fn map_column( + rel: &RelName, + attr: &RelAttr, + pk_member: bool, + rules: &ColumnRules, +) -> Option { + let resolved = type_bridge::map(attr, pk_member).ok()?; + let (target_name, resolved) = + apply_column_rule(&attr.name, resolved, rules.settings(rel, &attr.name)); + Some(ColumnMapping { + src_attnum: attr.attnum, + target_name, + target_type: resolved.ch_type, + }) +} + +pub fn derive_columns_for_mapping(desc: &RelDescriptor, rules: &ColumnRules) -> Vec { let keys = replident_key_attnums(desc); desc.attributes .iter() .filter(|attr| !attr.dropped) - .filter_map(|attr| { - type_bridge::map(attr, keys.contains(&attr.attnum)) - .ok() - .map(|resolved| ColumnMapping { - src_attnum: attr.attnum, - target_name: attr.name.clone(), - target_type: resolved.ch_type, - }) - }) + .filter_map(|attr| map_column(&desc.rel_name, attr, keys.contains(&attr.attnum), rules)) .collect() } -pub fn fold_diff_into_mapping(target: &mut TableMapping, new: &RelDescriptor, diff: &SchemaDiff) { +pub fn fold_diff_into_mapping( + target: &mut TableMapping, + new: &RelDescriptor, + diff: &SchemaDiff, + rules: &ColumnRules, +) { for (attnum, old_name, new_name) in &diff.renamed_columns { + // Preserve configured target name after source rename + let renamed_to = rules + .settings(&new.rel_name, new_name) + .target_name + .unwrap_or_else(|| new_name.clone()); for column in &mut target.columns { if column.src_attnum == *attnum && column.target_name == *old_name { - column.target_name.clone_from(new_name); + column.target_name.clone_from(&renamed_to); } } } @@ -178,12 +214,8 @@ pub fn fold_diff_into_mapping(target: &mut TableMapping, new: &RelDescriptor, di continue; } let key = replident_key_attnums(new).contains(&attr.attnum); - if let Ok(resolved) = type_bridge::map(attr, key) { - target.columns.push(ColumnMapping { - src_attnum: attr.attnum, - target_name: attr.name.clone(), - target_type: resolved.ch_type, - }); + if let Some(column) = map_column(&new.rel_name, attr, key, rules) { + target.columns.push(column); } } } @@ -229,4 +261,132 @@ mod tests { assert!(!planned.contains_key(&rel2), "snapshot predates the swap"); assert!(handle.with(|m| m.contains_key(&rel2)).await); } + + fn attr(attnum: i16, name: &str, type_oid: u32) -> crate::schema::RelAttr { + crate::schema::RelAttr { + attnum, + name: name.into(), + type_oid, + typmod: -1, + not_null: true, + dropped: false, + type_name: String::new(), + type_byval: true, + type_len: 4, + type_align: 'i', + type_storage: 'p', + missing_text: None, + } + } + + fn events_desc(attrs: Vec) -> RelDescriptor { + RelDescriptor { + rfn: walrus::pg::walparser::RelFileNode { + spc_node: 1663, + db_node: 5, + rel_node: 16385, + }, + oid: 16385, + toast_oid: 0, + namespace_oid: 2200, + rel_name: RelName::new("app", "events"), + kind: 'r', + persistence: 'p', + replident: crate::schema::ReplIdent::Default { pk_attnums: None }, + attributes: attrs, + } + } + + fn amount_rules() -> ColumnRules { + let mut b = crate::column_rules::ColumnRulesBuilder::new(); + b.add( + &RelName::new("app", "events"), + crate::table_rules::MatchKind::Exact, + "legacy_id", + crate::table_rules::MatchKind::Exact, + crate::column_rules::ColumnRule { + target_name: Some("id".into()), + target_type: None, + }, + ); + b.add( + &RelName::new("app", "*"), + crate::table_rules::MatchKind::Glob, + "*_amount", + crate::table_rules::MatchKind::Glob, + crate::column_rules::ColumnRule { + target_name: None, + target_type: Some("Decimal(38, 9)".into()), + }, + ); + b.finish().0 + } + + #[test] + fn derived_columns_take_rule_name_and_type() { + let desc = events_desc(vec![ + attr(1, "legacy_id", crate::schema::INT4OID), + attr(2, "net_amount", crate::schema::NUMERICOID), + attr(3, "note", crate::schema::TEXTOID), + ]); + let columns = derive_columns_for_mapping(&desc, &amount_rules()); + assert_eq!(columns[0].src_attnum, 1); + assert_eq!(columns[0].target_name, "id", "rule renames"); + assert_eq!(columns[1].target_type, "Decimal(38, 9)", "glob retypes"); + assert_eq!(columns[2].target_name, "note"); + assert_eq!(columns[2].target_type, "String", "unmatched keeps bridge"); + } + + #[test] + fn folded_column_takes_the_rule_its_name_matches() { + let rules = amount_rules(); + let old = events_desc(vec![attr(1, "legacy_id", crate::schema::INT4OID)]); + let mut mapping = TableMapping { + target: TableTarget::new("db", "events"), + columns: derive_columns_for_mapping(&old, &rules), + }; + let added = attr(2, "gross_amount", crate::schema::NUMERICOID); + let new = events_desc(vec![ + attr(1, "legacy_id", crate::schema::INT4OID), + added.clone(), + ]); + fold_diff_into_mapping( + &mut mapping, + &new, + &SchemaDiff { + added_columns: vec![added], + dropped_columns: vec![], + renamed_columns: vec![], + type_changes: vec![], + }, + &rules, + ); + assert_eq!(mapping.columns[1].target_type, "Decimal(38, 9)"); + } + + #[test] + fn rename_lands_on_the_name_the_rule_states() { + let rules = amount_rules(); + let mut mapping = TableMapping { + target: TableTarget::new("db", "events"), + columns: vec![ColumnMapping { + src_attnum: 1, + target_name: "id_v1".into(), + target_type: "Int32".into(), + }], + }; + let new = events_desc(vec![attr(1, "legacy_id", crate::schema::INT4OID)]); + fold_diff_into_mapping( + &mut mapping, + &new, + &SchemaDiff { + added_columns: vec![], + dropped_columns: vec![], + renamed_columns: vec![(1, "id_v1".into(), "legacy_id".into())], + type_changes: vec![], + }, + &rules, + ); + assert_eq!(mapping.columns[0].target_name, "id"); + } } diff --git a/src/runtime_config.rs b/src/runtime_config.rs index b0ac44de..aa8ce461 100644 --- a/src/runtime_config.rs +++ b/src/runtime_config.rs @@ -21,6 +21,7 @@ use crate::decode::heap_decoder::{ColumnValue, DecodedHeap, HeapOp}; use crate::schema::{RelDescriptor, RelName}; +use crate::table_rules::MatchKind; use ahash::HashMap; pub const CONFIG_GLOBAL: &str = "config_global"; @@ -82,6 +83,16 @@ 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, + pub match_kind: Option, +} + +impl TableRow { + pub fn is_pattern(&self) -> bool { + !matches!( + MatchKind::parse(self.match_kind.as_deref().unwrap_or_default()), + Ok(MatchKind::Exact) + ) + } } /// Parsed `config_table.initial_load` mode. @@ -125,6 +136,7 @@ impl InitialLoadMode { #[derive(Debug, Clone, Default, PartialEq)] pub struct ColumnRow { pub target_type: Option, + pub match_kind: Option, } /// One applied config change, interpreted from a config-table heap write and @@ -148,6 +160,7 @@ pub enum ConfigEvent { }, TableRemoved { rel: RelName, + pattern: bool, }, ColumnUpserted { rel: RelName, @@ -197,7 +210,7 @@ impl ConfigOverlay { ConfigEvent::TableUpserted { rel, row } => { self.tables.insert(rel, row); } - ConfigEvent::TableRemoved { rel } => { + ConfigEvent::TableRemoved { rel, .. } => { self.tables.remove(&rel); } ConfigEvent::ColumnUpserted { rel, attname, row } => { @@ -322,7 +335,12 @@ pub fn interpret( &field_string(rel, &cols, "relname")?, ); if removed { - return Some(ConfigEvent::TableRemoved { rel: key }); + let pattern = TableRow { + match_kind: field_string(rel, &cols, "match"), + ..TableRow::default() + } + .is_pattern(); + return Some(ConfigEvent::TableRemoved { rel: key, pattern }); } Some(ConfigEvent::TableUpserted { rel: key, @@ -331,6 +349,7 @@ 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"), + match_kind: field_string(rel, &cols, "match"), }, }) } @@ -348,6 +367,7 @@ pub fn interpret( attname, row: ColumnRow { target_type: field_string(rel, &cols, "target_type"), + match_kind: field_string(rel, &cols, "match"), }, }) } @@ -628,6 +648,88 @@ mod tests { } } + #[test] + fn table_upsert_reads_match_kind() { + let r = rel( + CONFIG_TABLE, + vec![ + attr(1, "namespace", 25), + attr(2, "relname", 25), + attr(3, "match", 25), + attr(4, "replicate", 16), + ], + ); + let new = vec![ + Some(ColumnValue::Text("app".into())), + Some(ColumnValue::Text("events_.*".into())), + Some(ColumnValue::Text("regex".into())), + Some(ColumnValue::Bool(true)), + ]; + match interpret( + ConfigTableKind::Table, + &heap(HeapOp::Insert, Some(new.clone()), None), + &r, + ) + .unwrap() + { + ConfigEvent::TableUpserted { rel, row } => { + assert_eq!(rel, RelName::new("app", "events_.*")); + assert!(row.is_pattern()); + assert_eq!(row.replicate, Some(true)); + } + other => panic!("expected TableUpserted, got {other:?}"), + } + match interpret( + ConfigTableKind::Table, + &heap(HeapOp::Delete, None, Some(new)), + &r, + ) + .unwrap() + { + ConfigEvent::TableRemoved { rel, pattern } => { + assert_eq!(rel, RelName::new("app", "events_.*")); + assert!(pattern); + } + other => panic!("expected TableRemoved, got {other:?}"), + } + } + + #[test] + fn column_upsert_reads_match_kind() { + let r = rel( + CONFIG_COLUMN, + vec![ + attr(1, "namespace", 25), + attr(2, "relname", 25), + attr(3, "attname", 25), + attr(4, "match", 25), + attr(5, "target_type", 25), + ], + ); + let new = vec![ + Some(ColumnValue::Text("app".into())), + Some(ColumnValue::Text("*".into())), + Some(ColumnValue::Text("*_amount".into())), + Some(ColumnValue::Text("glob".into())), + Some(ColumnValue::Text("Decimal(38, 9)".into())), + ]; + match interpret( + ConfigTableKind::Column, + &heap(HeapOp::Insert, Some(new), None), + &r, + ) + .unwrap() + { + ConfigEvent::ColumnUpserted { rel, attname, row } => { + assert_eq!(rel, RelName::new("app", "*")); + assert_eq!(attname, "*_amount"); + assert_eq!(row.match_kind.as_deref(), Some("glob")); + assert_eq!(row.target_type.as_deref(), Some("Decimal(38, 9)")); + } + other => panic!("expected ColumnUpserted, got {other:?}"), + } + } + #[test] fn initial_load_parse_accepts_explicit_none() { assert_eq!(InitialLoadMode::parse("none"), Some(InitialLoadMode::None)); diff --git a/src/table_rules.rs b/src/table_rules.rs new file mode 100644 index 00000000..dd3c954b --- /dev/null +++ b/src/table_rules.rs @@ -0,0 +1,536 @@ +//! Resolves table settings from TOML and runtime config + +use globset::{Glob, GlobMatcher}; +use regex_automata::meta::Regex; + +use crate::runtime_config::TableRow; +use crate::schema::RelName; + +/// Overwrite only where the narrower layer states a value +pub(crate) fn set_if(dst: &mut Option, src: &Option) { + if src.is_some() { + dst.clone_from(src); + } +} + +/// Pattern syntax used by config entry +#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)] +pub enum MatchKind { + #[default] + Exact, + Glob, + Regex, +} + +impl MatchKind { + pub fn parse(s: &str) -> Result { + match s.trim().to_ascii_lowercase().as_str() { + "" | "exact" => Ok(Self::Exact), + "glob" => Ok(Self::Glob), + "regex" => Ok(Self::Regex), + other => Err(format!( + "unknown match kind `{other}` (expected exact / glob / regex)" + )), + } + } + + pub fn as_str(self) -> &'static str { + match self { + Self::Exact => "exact", + Self::Glob => "glob", + Self::Regex => "regex", + } + } +} + +#[derive(Debug, Clone)] +pub(crate) enum NamePattern { + Literal(String), + Glob(GlobMatcher), + Regex(Regex), +} + +impl NamePattern { + pub(crate) fn compile(kind: MatchKind, pattern: &str) -> Result { + match kind { + MatchKind::Exact => Ok(Self::Literal(pattern.to_owned())), + MatchKind::Glob => Glob::new(pattern) + .map(|g| Self::Glob(g.compile_matcher())) + .map_err(|e| format!("invalid glob `{pattern}`: {e}")), + MatchKind::Regex => Regex::new(&format!("^(?:{pattern})$")) + .map(Self::Regex) + .map_err(|e| format!("invalid regex `{pattern}`: {e}")), + } + } + + pub(crate) fn is_match(&self, name: &str) -> bool { + match self { + Self::Literal(s) => s == name, + Self::Glob(g) => g.is_match(name), + Self::Regex(r) => r.is_match(name), + } + } +} + +#[derive(Debug, Clone, Default, PartialEq, Eq)] +pub struct TableRule { + pub target_database: Option, + pub target_table: Option, + pub replicate: Option, + pub initial_load: Option, +} + +impl TableRule { + pub fn from_row(row: &TableRow) -> Self { + Self { + target_database: row.target_database.clone(), + target_table: row.target_table.clone(), + replicate: row.replicate, + initial_load: row.initial_load.clone(), + } + } + + pub fn overlay(&mut self, other: &Self) { + 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); + } +} + +#[derive(Debug, Clone)] +pub struct RelMatcher { + src: RelName, + kind: MatchKind, + namespace: NamePattern, + name: NamePattern, +} + +impl RelMatcher { + pub fn compile(src: &RelName, kind: MatchKind) -> Result { + Ok(Self { + src: src.clone(), + kind, + namespace: NamePattern::compile(kind, &src.namespace)?, + name: NamePattern::compile(kind, &src.name)?, + }) + } + + pub fn matches(&self, rel: &RelName) -> bool { + self.namespace.is_match(&rel.namespace) && self.name.is_match(&rel.name) + } + + pub(crate) fn is_pattern(&self) -> bool { + self.kind != MatchKind::Exact + } + + pub(crate) fn width(&self) -> usize { + self.src.namespace.len() + self.src.name.len() + } + + /// Broadest first, literals last + pub(crate) fn rank(&self) -> (bool, usize, &str, &str) { + ( + !self.is_pattern(), + self.width(), + &self.src.namespace, + &self.src.name, + ) + } +} + +#[derive(Debug, Clone, Default)] +pub struct TableRules { + /// Ranked broadest to narrowest + rules: Vec<(RelMatcher, TableRule)>, + has_patterns: bool, +} + +impl TableRules { + /// Merged rule plus whether a pattern entry contributed + fn merge(&self, rel: &RelName) -> (TableRule, bool) { + let mut merged = TableRule::default(); + let mut from_pattern = false; + let mut barred = false; + let mut literal_replicate = None; + for (matcher, rule) in &self.rules { + if !matcher.matches(rel) { + continue; + } + merged.overlay(rule); + if matcher.is_pattern() { + from_pattern = true; + barred |= rule.replicate == Some(false); + } else { + set_if(&mut literal_replicate, &rule.replicate); + } + } + // Any matching exclusion wins over pattern opt-ins, literal entry aside + if barred { + merged.replicate = literal_replicate.or(Some(false)); + } + (merged, from_pattern) + } + + pub fn settings(&self, rel: &RelName) -> TableRule { + self.merge(rel).0 + } + + fn pattern_scope(&self, rel: &RelName) -> Option { + let (rule, from_pattern) = self.merge(rel); + (from_pattern && rule.replicate.is_some()).then(|| TableRow { + target_database: rule.target_database, + target_table: rule.target_table, + replicate: rule.replicate, + initial_load: rule.initial_load, + ..TableRow::default() + }) + } + + /// Scope intent a pattern entry states over relations the catalog holds. + /// `present` runs only when a pattern is in force, since listing the + /// catalog costs more than the lookup + pub fn pattern_scoped( + &self, + present: impl FnOnce() -> Vec, + mapped: impl Fn(&RelName) -> bool, + ) -> Vec<(RelName, TableRow)> { + if !self.has_patterns { + return Vec::new(); + } + present() + .into_iter() + .filter_map(|rel| self.pattern_scope(&rel).map(|row| (rel, row))) + .filter(|(rel, row)| row.replicate != Some(true) || !mapped(rel)) + .collect() + } + + pub fn has_patterns(&self) -> bool { + self.has_patterns + } +} + +#[derive(Debug, Default)] +pub struct TableRulesBuilder { + rules: Vec<(usize, RelMatcher, TableRule)>, + layer: usize, + rejections: u64, +} + +impl TableRulesBuilder { + pub fn new() -> Self { + Self::default() + } + + pub fn next_layer(&mut self) { + self.layer += 1; + } + + pub fn add(&mut self, key: &RelName, kind: MatchKind, rule: TableRule) { + match RelMatcher::compile(key, kind) { + Ok(matcher) => self.rules.push((self.layer, matcher, rule)), + Err(e) => { + tracing::warn!(target: "walshadow::config", qname = %key, error = %e, "table entry rejected"); + self.rejections += 1; + } + } + } + + pub fn add_row(&mut self, key: &RelName, row: &TableRow) { + let kind = match MatchKind::parse(row.match_kind.as_deref().unwrap_or_default()) { + Ok(k) => k, + Err(e) => { + tracing::warn!(target: "walshadow::config", qname = %key, error = %e, "config_table.match rejected"); + self.rejections += 1; + return; + } + }; + self.add(key, kind, TableRule::from_row(row)); + } + + pub fn finish(mut self) -> (TableRules, u64) { + self.rules + .sort_by(|(la, ma, _), (lb, mb, _)| ma.rank().cmp(&mb.rank()).then(la.cmp(lb))); + let rules: Vec<_> = self.rules.into_iter().map(|(_, m, r)| (m, r)).collect(); + ( + TableRules { + has_patterns: rules.iter().any(|(m, _)| m.is_pattern()), + rules, + }, + self.rejections, + ) + } +} + +#[cfg(test)] +mod tests { + use super::*; + + fn rel(ns: &str, name: &str) -> RelName { + RelName::new(ns, name) + } + + fn target(table: &str) -> TableRule { + TableRule { + target_table: Some(table.into()), + ..TableRule::default() + } + } + + #[test] + fn exact_entry_retargets_only_its_relation() { + let mut b = TableRulesBuilder::new(); + b.add(&rel("public", "events"), MatchKind::Exact, target("ev")); + let (rules, rejected) = b.finish(); + assert_eq!(rejected, 0); + assert_eq!( + rules.settings(&rel("public", "events")).target_table, + Some("ev".into()) + ); + assert!( + rules + .settings(&rel("public", "other")) + .target_table + .is_none() + ); + } + + #[test] + fn regex_entry_applies_to_unknown_relations() { + let mut b = TableRulesBuilder::new(); + b.add( + &rel("public", "events_.*"), + MatchKind::Regex, + TableRule { + replicate: Some(true), + initial_load: Some("copy".into()), + ..TableRule::default() + }, + ); + let (rules, _) = b.finish(); + let hit = rel("public", "events_2026"); + let s = rules.settings(&hit); + assert_eq!(s.replicate, Some(true)); + assert_eq!(s.initial_load.as_deref(), Some("copy")); + assert!(rules.pattern_scope(&hit).is_some()); + let miss = rel("public", "my_events_2026"); + assert_eq!( + rules.settings(&miss).replicate, + None, + "anchored: substring must not match" + ); + assert!(rules.pattern_scope(&miss).is_none()); + } + + #[test] + fn narrower_pattern_and_exact_entry_win() { + let mut b = TableRulesBuilder::new(); + b.add( + &rel("*", "*"), + MatchKind::Glob, + TableRule { + target_database: Some("broad".into()), + target_table: Some("broad".into()), + ..TableRule::default() + }, + ); + b.add( + &rel("public", "events_*"), + MatchKind::Glob, + target("narrow"), + ); + let (rules, _) = b.finish(); + let s = rules.settings(&rel("public", "events_1")); + assert_eq!( + s.target_table, + Some("narrow".into()), + "wider pattern applies first" + ); + assert_eq!( + s.target_database, + Some("broad".into()), + "broad pattern still contributes fields the narrow one omits" + ); + + let mut b = TableRulesBuilder::new(); + b.add(&rel("*", "*"), MatchKind::Glob, target("broad")); + b.next_layer(); + b.add( + &rel("public", "events_1"), + MatchKind::Exact, + target("exact"), + ); + let (rules, _) = b.finish(); + assert_eq!( + rules.settings(&rel("public", "events_1")).target_table, + Some("exact".into()) + ); + } + + #[test] + fn unparseable_pattern_rejected() { + let mut b = TableRulesBuilder::new(); + b.add( + &rel("public", "ev(nt"), + MatchKind::Regex, + TableRule::default(), + ); + b.add( + &rel("public", "ev[nt"), + MatchKind::Glob, + TableRule::default(), + ); + let (rules, rejected) = b.finish(); + assert_eq!(rejected, 2); + assert!(!rules.has_patterns()); + } + + #[test] + fn pattern_scope_lists_matching_present_relations() { + let mut b = TableRulesBuilder::new(); + b.add( + &rel("public", "events_*"), + MatchKind::Glob, + TableRule { + replicate: Some(true), + initial_load: Some("copy".into()), + ..TableRule::default() + }, + ); + let (rules, _) = b.finish(); + let present = [ + rel("public", "events_1"), + rel("public", "orders"), + rel("other", "events_2"), + ]; + let scoped = rules.pattern_scoped(|| present.to_vec(), |_| false); + assert_eq!(scoped.len(), 1); + assert_eq!(scoped[0].0, rel("public", "events_1")); + assert_eq!(scoped[0].1.replicate, Some(true)); + assert_eq!(scoped[0].1.initial_load.as_deref(), Some("copy")); + assert!( + rules + .pattern_scoped(|| present.to_vec(), |_| true) + .is_empty(), + "an already-mapped relation keeps its pinned projection" + ); + assert!( + rules.pattern_scope(&rel("public", "orders")).is_none(), + "no scope intent without a match" + ); + } + + #[test] + fn excluding_pattern_beats_matching_opt_in() { + let mut b = TableRulesBuilder::new(); + b.add( + &rel("app", "events_*"), + MatchKind::Glob, + TableRule { + replicate: Some(true), + ..TableRule::default() + }, + ); + b.add( + &rel("app", "*_audit"), + MatchKind::Glob, + TableRule { + replicate: Some(false), + ..TableRule::default() + }, + ); + let (rules, _) = b.finish(); + assert_eq!( + rules.settings(&rel("app", "events_1")).replicate, + Some(true) + ); + assert_eq!( + rules.settings(&rel("app", "events_audit")).replicate, + Some(false), + "guardrail regardless of which pattern is wider" + ); + let mut b = TableRulesBuilder::new(); + b.add( + &rel("app", "*_audit"), + MatchKind::Glob, + TableRule { + replicate: Some(false), + ..TableRule::default() + }, + ); + b.next_layer(); + b.add( + &rel("app", "events_audit"), + MatchKind::Exact, + TableRule { + replicate: Some(true), + ..TableRule::default() + }, + ); + let (rules, _) = b.finish(); + assert_eq!( + rules.settings(&rel("app", "events_audit")).replicate, + Some(true) + ); + } + + #[test] + fn glob_entry_reads_wildcards_not_regex() { + let mut b = TableRulesBuilder::new(); + b.add( + &rel("app", "events_*"), + MatchKind::Glob, + TableRule { + replicate: Some(true), + ..TableRule::default() + }, + ); + let (rules, rejected) = b.finish(); + assert_eq!(rejected, 0); + for name in ["events_2026", "events_"] { + assert_eq!( + rules.settings(&rel("app", name)).replicate, + Some(true), + "glob `events_*` must match {name}" + ); + } + for name in ["events", "other_events_1"] { + assert_eq!( + rules.settings(&rel("app", name)).replicate, + None, + "glob `events_*` must not match {name}" + ); + } + } + + #[test] + fn glob_takes_dots_and_regex_metacharacters_literally() { + let mut b = TableRulesBuilder::new(); + b.add( + &rel("app", "v1.*"), + MatchKind::Glob, + TableRule { + replicate: Some(true), + ..TableRule::default() + }, + ); + let (rules, _) = b.finish(); + assert_eq!( + rules.settings(&rel("app", "v1.events")).replicate, + Some(true) + ); + assert_eq!( + rules.settings(&rel("app", "v1x")).replicate, + None, + "the dot is a dot, not any-character" + ); + } + + #[test] + fn match_kind_parse() { + assert_eq!(MatchKind::parse("").unwrap(), MatchKind::Exact); + assert_eq!(MatchKind::parse(" Regex ").unwrap(), MatchKind::Regex); + assert_eq!(MatchKind::parse("glob").unwrap(), MatchKind::Glob); + assert!(MatchKind::parse("like").is_err()); + } +} diff --git a/tests/runtime_config_e2e.rs b/tests/runtime_config_e2e.rs index 3d1ed25a..632ab0b5 100644 --- a/tests/runtime_config_e2e.rs +++ b/tests/runtime_config_e2e.rs @@ -61,6 +61,9 @@ //! * Route snapshots attach at planning: a transaction planned before //! the opt-in never re-routes, one planned after routes whole. //! +//! 9. `pattern_row_scopes_tables_by_glob` +//! * Glob rules include matching tables and exclude guarded tables +//! //! 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). @@ -946,3 +949,94 @@ async fn pre_opt_in_xact_discards_post_opt_in_routes() { .expect("ch v"); assert_eq!(v, "post-opt-in"); } + +/// Drill 9: glob rules scope tables created later +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn pattern_row_scopes_tables_by_glob() { + 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_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-glob-scope", + 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, match, replicate) \ + VALUES ('app', 'events_*', 'glob', true), ('app', '*_audit', 'glob', false)" + .into(), + "CREATE TABLE app.events_2026 (id bigint PRIMARY KEY, body text)".into(), + "CREATE TABLE app.events_audit (id bigint PRIMARY KEY, body text)".into(), + "CREATE TABLE app.orders (id bigint PRIMARY KEY, body text)".into(), + "INSERT INTO app.events_2026 (id, body) VALUES (1, 'in-scope')".into(), + "INSERT INTO app.events_audit (id, body) VALUES (1, 'barred')".into(), + "INSERT INTO app.orders (id, body) VALUES (1, 'unscoped')".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 body = ch + .query( + "SELECT argMax(body, _lsn) FROM walshadow_test.events_2026 \ + WHERE _is_deleted = 0 AND id = 1", + ) + .expect("ch body"); + assert_eq!(body, "in-scope", "the opt-in pattern creates and routes"); + + let others = ch + .query( + "SELECT count() FROM system.tables WHERE database = 'walshadow_test' \ + AND name IN ('events_audit', 'orders')", + ) + .expect("ch table existence"); + assert_eq!( + others, "0", + "excluded / unmatched relations must not create" + ); +}