Configure on CH meta names, ORDER BY, PRIMARY KEY

README.md

@@ -230,6 +230,33 @@ one CH name), and the whole array pins nothing — scope still comes from
 `replicate` / `auto_create`. `config_column` rows take `match` the same way,
 over the whole `(namespace, relname, attname)` key.
 
+Destination shape — walshadow appends `_lsn`, `_xid`, `_commit_ts`,
+`_is_deleted` to every table and keys on replica identity. Rename or drop the
+appended columns, and pin the sort key:
+
+```toml
+[system_columns]                    # cluster-wide default
+lsn = "_peerdb_version"
+commit_ts = "_peerdb_synced_at"
+is_deleted = "_peerdb_is_deleted"   # false drops the column, and DELETE rows
+
+[table.public.events]
+order_by = ["tenant_id", "id"]      # else replica identity
+primary_key = ["tenant_id"]         # CH index prefix, must prefix order_by
+lsn = "_version"                    # same four keys, this relation only
+
+[table.app."events_*"]
+match = "glob"
+is_deleted = false
+order_by = ["tenant_id", "id"]
+```
+
+`[system_columns]` is boot-only and cluster-wide; a `[table.*]` block or a
+`config_table` row renames per relation, and takes `match` like any other rule
+— walshadow never renames or rekeys a table CH already holds, so shape has to
+land before the first CREATE. See
+[destination tables](docs/destination-tables.md)
+
 
 ## Building from source
 

docs/configuration.md

@@ -61,8 +61,9 @@ including tables created later. `init` writes `[source]`, `[ch]`, and chosen
 Pass file with `--ch-config`. Loader also merges sibling directory formed by
 replacing `.toml` with `.d`, for example `ch-config.d/*.toml`
 
-Unknown keys, invalid values, and incompatible mapping fields fail validation
-instead of falling back silently
+Invalid values and incompatible mapping fields fail validation instead of
+falling back silently. Loader ignores unknown keys, so check spelling here when
+a setting has no effect
 
 ## Precedence
 
@@ -116,6 +117,8 @@ restored
 Apply live:
 
 - table and column rules
+- per-table metadata column names, `order_by`, and `primary_key`, applied when
+  walshadow creates a table, see [Query destination data](destination-tables.md)
 - namespace destinations and drop policy
 - pause state
 - batch sizes, flush timeout, compression, and retry count
@@ -125,6 +128,7 @@ Require restart:
 
 - `replicate_all`
 - runtime-config schema
+- cluster-wide `[system_columns]` names
 - soft-delete and TOAST modes
 - worker-pool sizes and memory limits
 - backup and shadow bootstrap choices
@@ -161,6 +165,10 @@ VALUES
     ('public', 'orders', true, 'copy');
 ```
 
+`config_table` also carries destination shape: `order_by` and `primary_key` as
+`text[]`, and `lsn`, `xid`, `commit_ts`, `is_deleted` for metadata column names.
+See [Query destination data](destination-tables.md)
+
 walshadow reads these tables but never writes them. Keep archive credentials
 and bootstrap configuration in TOML, not source-side tables
 

docs/destination-tables.md

@@ -21,6 +21,83 @@ Source columns are followed by four metadata columns:
 | `_commit_ts` | `DateTime64(6, 'UTC')` | source commit time |
 | `_is_deleted` | `Bool` | delete marker |
 
+Rename these columns, drop the delete marker, or pin the sort key with
+settings below
+
+## Choose sort key
+
+Set `order_by` to sort a destination table on chosen columns instead of source
+row key. Name ClickHouse column names, after any rename:
+
+```toml
+[table.public.events]
+order_by = ["tenant_id", "id"]
+primary_key = ["tenant_id"]
+
+[table.app."events_*"]
+match = "glob"
+order_by = ["tenant_id", "id"]
+```
+
+ClickHouse `PRIMARY KEY` chooses which sort-key prefix its sparse index covers
+and enforces no uniqueness. `primary_key` must be a prefix of `order_by`.
+walshadow ignores an invalid `primary_key`, logs a warning, and indexes whole
+sort key. It also ignores an `order_by` naming a missing or `Nullable` column,
+because ClickHouse rejects nullable sort keys, and falls back to source row key
+
+Both settings apply when walshadow creates a table. walshadow never rekeys a
+table ClickHouse already holds, so choose shape before first delivery, or run
+`ALTER TABLE` in ClickHouse. With `replicate_all = true` a table can reach
+ClickHouse before an exact source-side row arrives: keep custom shape in config
+file, or in a pattern rule which matches before creation
+
+Source-side rows carry same settings as `text[]`:
+
+```sql
+UPDATE walshadow.config_table
+SET order_by = ARRAY['tenant_id', 'id'], primary_key = ARRAY['tenant_id']
+WHERE namespace = 'public' AND relname = 'events';
+```
+
+## Rename metadata columns
+
+`[system_columns]` renames appended columns for every table. walshadow reads it
+at startup only:
+
+```toml
+[system_columns]
+lsn = "_peerdb_version"
+commit_ts = "_peerdb_synced_at"
+is_deleted = "_peerdb_is_deleted"
+```
+
+Set same keys in a `[table.*]` block, or in a `config_table` row, to rename for
+matching relations. Omitted keys inherit cluster-wide names. Names must be
+unique and non-empty: config file fails validation, and a source-side row is
+rejected with a warning, leaving cluster-wide names in place. TOAST mirror
+tables keep fixed names
+
+walshadow uses configured names in `CREATE TABLE` and `INSERT` statements, and
+never renames a column in an existing ClickHouse table. Renaming for an existing
+destination also needs `ALTER TABLE ... RENAME COLUMN` in ClickHouse
+
+## Drop the delete marker
+
+Set `is_deleted = false` for an append-only destination. This drops the marker
+column and discards source `DELETE` rows, counting them in
+`walshadow_emitter_deletes_discarded_total`:
+
+```toml
+[system_columns]
+is_deleted = false               # cluster-wide
+
+[table.app."events_*"]
+match = "glob"
+is_deleted = false               # this pattern only
+```
+
+Source-side rows use an empty string, `is_deleted = ''`, for same result
+
 ## Read current state
 
 Use `FINAL` when query must resolve outstanding row versions immediately

plans/GLOSSARY.md

@@ -561,9 +561,13 @@ diverging `shadow_apply` vs `dispatched` signals shadow lag
 `XLOG_XACT_ASSIGNMENT`; a hint only, authoritative subxact list arrives
 inline on commit/abort record ([xact.md](xact.md))
 
-**synthetic columns** — four trailing columns on every dest table:
-`_lsn` UInt64, `_xid` UInt32, `_commit_ts` DateTime64(6,'UTC'),
-`_is_deleted` Bool ([emitter.md](emitter.md))
+**synthetic columns** — trailing columns on every dest table, default
+names `_lsn` UInt64, `_xid` UInt32, `_commit_ts` DateTime64(6,'UTC'),
+`_is_deleted` Bool. `[system_columns]` renames them cluster-wide and can
+drop the delete marker; a `[table.*]` block or `config_table` row (literal
+or `match` pattern) overrides per relation
+([emitter.md](emitter.md),
+[destination tables guide](../docs/destination-tables.md))
 
 **tail** — reusable batcher + inserter pool + ack collector unit; WAL
 pipeline and bootstrap drain feed the identical tail, `tail.finish`

plans/emitter.md

@@ -386,8 +386,17 @@ CH applies its own zero-init
 ### Synthetic columns
 
 Destination metadata contract lives in
-[`docs/destination-tables.md`](../docs/destination-tables.md). All four values
-remain non-nullable and append after mapped columns in `TableEncoder::new`
+[`docs/destination-tables.md`](../docs/destination-tables.md). Values stay
+non-nullable and append after mapped columns in `TableEncoder::new`. Names come
+from the relation's resolved `SystemColumns`, so every site that renders or
+encodes a metadata column reads them per relation instead of a constant
+
+The delete marker is optional (`is_deleted = false`). Without it a DELETE would
+land as a phantom insert of the old image, so those rows are discarded where the
+placed count is taken (`decode_and_route`, and the object-store gap-replay
+sink), counted in `walshadow_emitter_deletes_discarded_total`. Dropping them
+anywhere later would short the ack collector's per-seq reconcile and pin the
+watermark
 
 `_lsn` is dedup key because emitter ack lags actual CH durability by up
 to one flush window. On restart the manifest floor rewinds to
@@ -449,8 +458,10 @@ table:
 | `Dropped { rel_name }` | gated on the namespace's `DropTableStrategy` (`drop_strategy_for`, else global): `Retain` (default) skips silently, `Warn` skips at WARN, `Drop` runs `DROP TABLE IF EXISTS` |
 
 `render_create_table` builds CREATE off descriptor: attributes through
-`type_bridge::map`, PK columns first in `ORDER BY` (else `_lsn`
-fallback), engine pinned to `ReplacingMergeTree(_lsn)`. Synthetic
+`type_bridge::map`, then the sort key — the operator `order_by`
+([`docs/destination-tables.md`](../docs/destination-tables.md)) when it names
+non-nullable destination columns, else PK columns first (else `_lsn` fallback) —
+engine pinned to `ReplacingMergeTree(_lsn)`. Synthetic
 columns appended after mapped columns, same shape as `TablePlan::build`.
 `render_create_table_from_mapping` builds off the mapping instead (its
 columns are the emitter's INSERT contract), resolving `ORDER BY` key

sql/runtime_config_install.sql

@@ -52,6 +52,14 @@ CREATE TABLE IF NOT EXISTS :"walshadow_schema".config_table (
     initial_load    text,            -- one-time backfill mode for pre-opt-in
                                      -- rows: 'none' | 'copy' | 'base_backup'
                                      -- | 'object_store'; NULL means omitted
+    order_by        text[],          -- ClickHouse ORDER BY columns; NULL
+                                     -- inherits, empty array derives
+    primary_key     text[],          -- ClickHouse PRIMARY KEY; must prefix
+                                     -- order_by
+    lsn             text,            -- per-relation names for the columns
+    xid             text,            -- walshadow appends; NULL inherits
+    commit_ts       text,            -- [system_columns]
+    is_deleted      text,            -- '' drops the marker (and DELETE rows)
     PRIMARY KEY (namespace, relname)
 );
 
@@ -72,7 +80,13 @@ ALTER TABLE :"walshadow_schema".config_table ADD COLUMN IF NOT EXISTS replicate
 ALTER TABLE :"walshadow_schema".config_table ADD COLUMN IF NOT EXISTS initial_load    text;
 ALTER TABLE :"walshadow_schema".config_table ADD COLUMN IF NOT EXISTS target_database text;
 ALTER TABLE :"walshadow_schema".config_table ADD COLUMN IF NOT EXISTS target_table    text;
+ALTER TABLE :"walshadow_schema".config_table ADD COLUMN IF NOT EXISTS order_by        text[];
+ALTER TABLE :"walshadow_schema".config_table ADD COLUMN IF NOT EXISTS primary_key     text[];
 ALTER TABLE :"walshadow_schema".config_table ADD COLUMN IF NOT EXISTS match           text;
+ALTER TABLE :"walshadow_schema".config_table ADD COLUMN IF NOT EXISTS lsn             text;
+ALTER TABLE :"walshadow_schema".config_table ADD COLUMN IF NOT EXISTS xid             text;
+ALTER TABLE :"walshadow_schema".config_table ADD COLUMN IF NOT EXISTS commit_ts       text;
+ALTER TABLE :"walshadow_schema".config_table ADD COLUMN IF NOT EXISTS is_deleted      text;
 ALTER TABLE :"walshadow_schema".config_column ADD COLUMN IF NOT EXISTS match           text;
 
 -- REPLICA IDENTITY FULL logs the complete old-row image on UPDATE/DELETE, so a

src/backfill/backfill_staging.rs

@@ -127,6 +127,9 @@ pub struct StagingSession {
     client: BoxedAsyncClient,
     /// Kept whole for reconnect; shared with the pass that opened the session
     conn: Arc<EmitterConfig>,
+    /// Per-relation destination rules, for the promote's `_lsn` predicate when
+    /// a `[table.*]` block or `config_table` row renamed that column
+    rules: Option<Arc<crate::table_rules::TableRules>>,
     retry: RetryConfig,
     timeout: Duration,
 }
@@ -141,9 +144,27 @@ impl StagingSession {
             retry: emitter.retry.clone(),
             timeout: emitter.insert_timeout,
             conn: emitter,
+            rules: None,
         })
     }
 
+    pub fn with_rules(mut self, rules: Option<Arc<crate::table_rules::TableRules>>) -> Self {
+        self.rules = rules;
+        self
+    }
+
+    /// LSN column of one relation's destination
+    fn lsn_column(&self, rel: &RelName) -> String {
+        match &self.rules {
+            Some(rules) => rules
+                .settings(rel)
+                .system_columns(&self.conn.system_columns)
+                .lsn
+                .clone(),
+            None => self.conn.system_columns.lsn.clone(),
+        }
+    }
+
     async fn attempt_write(&mut self, sql: &str) -> Result<(), EmitterError> {
         exec_drain(&mut self.client, sql, self.timeout).await
     }
@@ -295,8 +316,9 @@ impl StagingSession {
             );
         }
         let list = cols.join(", ");
+        let lsn = quote_ident(&self.lsn_column(&rel.rel));
         self.exec_retry(&format!(
-            "INSERT INTO {} ({list}) SELECT {list} FROM {} WHERE `_lsn` > {}",
+            "INSERT INTO {} ({list}) SELECT {list} FROM {} WHERE {lsn} > {}",
             rel.real_sql(),
             rel.staging_sql(),
             rel.s_lsn

src/backfill/backup_backfill.rs

@@ -350,7 +350,7 @@ async fn walk_and_ship(
         ctx.stats.clone(),
         resolver.clone(),
         DeferredSpool::new(toast_spool_path, DEFERRED_SPOOL_MEM_MAX),
-        ctx.emitter.soft_delete,
+        ctx.emitter.row_policy(),
         ctx.config_rx.as_ref().map(|rx| rx.borrow().clone()),
     ));
 
@@ -901,7 +901,7 @@ async fn replay_gap(
         mapping: ctx.mapping.snapshot().await,
         stats: ctx.stats.clone(),
         budget: ctx.budget.clone(),
-        soft_delete: ctx.emitter.soft_delete,
+        row_policy: ctx.emitter.row_policy(),
         config: ctx.config_rx.as_ref().map(|rx| rx.borrow().clone()),
         batch_rows: ctx.emitter.drain_batch_rows,
         batch_bytes: ctx.emitter.drain_batch_bytes,
@@ -948,7 +948,7 @@ struct ReplaySink {
     stats: Arc<EmitterStats>,
     budget: Option<crate::budget::MemoryBudget>,
     /// Boot-only delete-retention policy, frozen into route snapshots
-    soft_delete: bool,
+    row_policy: crate::emit::route::RowPolicy,
     /// Config snapshot for route freezes: gap replay re-seeds from current
     /// config, not history (route history has no WAL position)
     config: Option<Arc<ResolvedConfig>>,
@@ -1069,6 +1069,19 @@ impl ReplaySink {
                         self.commits_past_s += 1;
                         continue;
                     }
+                    let policy = self
+                        .row_policy
+                        .for_rel(self.config.as_deref(), &rel.rel_name);
+                    // Append-only destination (no delete marker): see
+                    // `decode_and_route`
+                    if policy.system.is_deleted.is_none()
+                        && matches!(heap.decoded.op, crate::decode::heap_decoder::HeapOp::Delete)
+                    {
+                        self.stats
+                            .deletes_discarded
+                            .fetch_add(1, std::sync::atomic::Ordering::Relaxed);
+                        continue;
+                    }
                     let rel = rel.clone();
                     let value_permit = detoast_heap(&mut heap, spool, &ref_maps, &self.resolver)
                         .await
@@ -1084,8 +1097,7 @@ impl ReplaySink {
                         .config
                         .as_ref()
                         .map_or_else(Arc::default, |rc| rc.column_rules.clone());
-                    let route =
-                        crate::emit::route::RouteSnapshot::freeze(mapping, rules, self.soft_delete);
+                    let route = crate::emit::route::RouteSnapshot::freeze(mapping, rules, policy);
                     let seq = if let Some((seq, rows)) = &mut self.open {
                         *rows += 1;
                         *seq

src/backfill/copy_backfill.rs

@@ -506,6 +506,12 @@ impl CopyBackfiller {
         dest.clone()
     }
 
+    /// Live per-relation rules, for destination names the staging promote
+    /// has to match. `None` without a resolver (tests): the boot set stands
+    fn table_rules(&self) -> Option<Arc<crate::table_rules::TableRules>> {
+        self.config_rx.as_ref().map(|rx| rx.borrow().rules.clone())
+    }
+
     fn refresh_gauges(&self, ledger: &Ledger) {
         self.pending
             .store(ledger.pending_count(), Ordering::Relaxed);
@@ -742,7 +748,10 @@ impl CopyBackfiller {
         if plan.rels.is_empty() {
             return;
         }
-        let mut sess = match StagingSession::connect(self.dest_emitter()).await {
+        let mut sess = match StagingSession::connect(self.dest_emitter())
+            .await
+            .map(|s| s.with_rules(self.table_rules()))
+        {
             Ok(s) => s,
             Err(e) => {
                 tracing::error!(
@@ -870,7 +879,9 @@ impl CopyBackfiller {
             table: target.table,
             s_lsn: rec.s_lsn.get(),
         };
-        let mut sess = StagingSession::connect(self.dest_emitter()).await?;
+        let mut sess = StagingSession::connect(self.dest_emitter())
+            .await?
+            .with_rules(self.table_rules());
         match sess.table_uuid(&rel.database, &rel.staging_table()).await? {
             None => {
                 self.mark_done_entry(name).await;
@@ -1058,7 +1069,7 @@ impl CopyBackfiller {
                 self.spill_dir.join("copy_deferred.bin"),
                 crate::backfill::spool::DEFERRED_SPOOL_MEM_MAX,
             ),
-            self.emitter.soft_delete,
+            self.emitter.row_policy(),
             self.config_rx.as_ref().map(|rx| rx.borrow().clone()),
         ));