_is_deleted

Author: serprex

clickhouse-c-rs/clickhouse-c

@@ -9,6 +9,7 @@ database = "demo"
 user = "default"
 password = ""
 compression = "lz4"
+# soft_delete = true
 
 [table."demo.users"]
 target = "demo.users"

docker/ch-config.toml

@@ -1,6 +1,6 @@
 -- walshadow demo destination table. Synthetic columns (_lsn, _xid,
--- _op, _commit_ts) match the emitter's TablePlan; their types are
--- fixed by walshadow::ch_emitter::TablePlan::build.
+-- _commit_ts, _is_deleted) match the emitter's TablePlan; their types
+-- are fixed by walshadow::ch_emitter::TablePlan::build.
 
 CREATE DATABASE IF NOT EXISTS demo;
 
@@ -10,8 +10,8 @@ CREATE TABLE IF NOT EXISTS demo.users (
     email       String,
     _lsn        UInt64,
     _xid        UInt32,
-    _op         Enum8('insert' = 1, 'update' = 2, 'delete' = 3),
-    _commit_ts  DateTime64(6, 'UTC')
+    _commit_ts  DateTime64(6, 'UTC'),
+    _is_deleted Bool
 )
-ENGINE = ReplacingMergeTree(_lsn)
+ENGINE = ReplacingMergeTree(_lsn, _is_deleted)
 ORDER BY id;

docker/init/clickhouse/01-schema.sql

@@ -2,9 +2,9 @@
 # Demo-only: pre-create the four pgbench destination tables on CH so the
 # walshadow emitter's pinned mappings (ch-config.demo.toml) have targets
 # to INSERT into. No-op unless WALSHADOW_DEMO_PGBENCH is set. Column
-# order + synthetic _lsn/_xid/_op/_commit_ts trailer mirror the emitter's
-# TablePlan; engine is ReplacingMergeTree(_lsn) so a row's newest LSN
-# wins on FINAL. Shapes match tests/pgbench_acceptance.rs.
+# order + synthetic _lsn/_xid/_commit_ts/_is_deleted trailer mirror the
+# emitter's TablePlan; engine ReplacingMergeTree(_lsn, _is_deleted) drops
+# deletes on FINAL. Shapes match tests/pgbench_acceptance.rs.
 
 set -euo pipefail
 
@@ -16,31 +16,27 @@ CREATE DATABASE IF NOT EXISTS demo;
 CREATE TABLE IF NOT EXISTS demo.pgbench_accounts (
     aid Int32, bid Int32, abalance Int32, filler String,
     _lsn UInt64, _xid UInt32,
-    _op Enum8('insert' = 1, 'update' = 2, 'delete' = 3),
-    _commit_ts DateTime64(6, 'UTC')
-) ENGINE = ReplacingMergeTree(_lsn) ORDER BY aid;
+    _commit_ts DateTime64(6, 'UTC'), _is_deleted Bool
+) ENGINE = ReplacingMergeTree(_lsn, _is_deleted) ORDER BY aid;
 
 CREATE TABLE IF NOT EXISTS demo.pgbench_branches (
     bid Int32, bbalance Int32, filler Nullable(String),
     _lsn UInt64, _xid UInt32,
-    _op Enum8('insert' = 1, 'update' = 2, 'delete' = 3),
-    _commit_ts DateTime64(6, 'UTC')
-) ENGINE = ReplacingMergeTree(_lsn) ORDER BY bid;
+    _commit_ts DateTime64(6, 'UTC'), _is_deleted Bool
+) ENGINE = ReplacingMergeTree(_lsn, _is_deleted) ORDER BY bid;
 
 CREATE TABLE IF NOT EXISTS demo.pgbench_tellers (
     tid Int32, bid Int32, tbalance Int32, filler Nullable(String),
     _lsn UInt64, _xid UInt32,
-    _op Enum8('insert' = 1, 'update' = 2, 'delete' = 3),
-    _commit_ts DateTime64(6, 'UTC')
-) ENGINE = ReplacingMergeTree(_lsn) ORDER BY tid;
+    _commit_ts DateTime64(6, 'UTC'), _is_deleted Bool
+) ENGINE = ReplacingMergeTree(_lsn, _is_deleted) ORDER BY tid;
 
 CREATE TABLE IF NOT EXISTS demo.pgbench_history (
     tid Int32, bid Int32, aid Int32, delta Int32,
     mtime DateTime64(6), filler Nullable(String),
     _lsn UInt64, _xid UInt32,
-    _op Enum8('insert' = 1, 'update' = 2, 'delete' = 3),
-    _commit_ts DateTime64(6, 'UTC')
-) ENGINE = ReplacingMergeTree(_lsn) ORDER BY (tid, mtime, aid);
+    _commit_ts DateTime64(6, 'UTC'), _is_deleted Bool
+) ENGINE = ReplacingMergeTree(_lsn, _is_deleted) ORDER BY (tid, mtime, aid);
 "
 
 echo "walshadow-demo: pgbench destination tables created on ClickHouse"

docker/init/clickhouse/02-pgbench.sh

@@ -162,8 +162,8 @@ non-nullable by construction, encoded in `TableEncoder::new`:
 |---|---|---|
 | `_lsn` | `UInt64` | source commit-record LSN. `ReplacingMergeTree(_lsn)` keys dedup on this so restart-and-replay window collapses re-emitted rows to latest LSN per PK |
 | `_xid` | `UInt32` | source xid. Lets analytic queries group all rows from one xact, recover xact boundary CH lost when emitter serialised across tables |
-| `_op` | `Enum8('insert'=1,'update'=2,'delete'=3)` | row-op classification. CH-side `WHERE _op != 3` is the cheap "live rows" filter; HOT_UPDATE collapses to UPDATE (code 2), PG-internal distinction doesn't reach CH |
 | `_commit_ts` | `DateTime64(6, 'UTC')` | xact commit timestamp, shifted from PG's 2000-01-01 epoch to Unix via `DATETIME64_PG_EPOCH_US` |
+| `_is_deleted` | `Bool` | 1 on delete, else 0. `Bool` is `UInt8` underneath (1 wire byte), so it satisfies `ReplacingMergeTree`'s `is_deleted` UInt8 requirement. `ReplacingMergeTree(_lsn, _is_deleted)` second arg collapses deletes on FINAL; `WHERE _is_deleted = 0` is the cheap "live rows" filter. `soft_delete` keeps it out of the engine args to retain tombstones |
 
 `_lsn` is dedup key because emitter ack lags actual CH durability by up
 to one flush window. On restart cursor's `emitter_ack_lsn` rewinds to

plans/emitter.md

@@ -1396,7 +1396,7 @@ async fn populate_metrics(
 ///    exists and is non-empty (true once this returns).
 ///
 /// `ch_config` `Some`: bootstrap rows route through the shared insert tail
-/// (synthetic INSERT `_op = 1`, `_lsn = start_lsn`, `_commit_ts = 0`).
+/// (synthetic INSERT `_lsn = start_lsn`, `_commit_ts = 0`, `_is_deleted = 0`).
 /// `wait_through(K)` proves every bootstrap seq durable on CH before
 /// teardown, so the WAL pump resumes against a fully-shipped baseline.
 /// `None`: rows drain to a metrics-only observer via `drain_backfill`.

src/bin/stream.rs

@@ -75,6 +75,9 @@ pub struct DdlConfig {
     /// Per-namespace overrides, fallback to the global fields above when
     /// a namespace has none
     pub namespaces: HashMap<String, NamespaceMapping>,
+    /// Keep `_is_deleted` out of `ReplacingMergeTree`'s args so deletes
+    /// stay queryable; mirrors [`EmitterConfig::soft_delete`]
+    pub soft_delete: bool,
 }
 
 impl DdlConfig {
@@ -92,6 +95,7 @@ impl DdlConfig {
             auto_create_namespaces,
             target_database: cfg.database.clone(),
             namespaces: cfg.namespaces.clone(),
+            soft_delete: cfg.soft_delete,
         }
     }
 
@@ -205,7 +209,7 @@ impl DdlApplicator {
             .config
             .target_database_for(&desc.namespace_name)
             .to_owned();
-        let sql = match render_create_table(desc, &target_db)? {
+        let sql = match render_create_table(desc, &target_db, self.config.soft_delete)? {
             Some(s) => s,
             None => {
                 self.stats.skipped += 1;
@@ -502,6 +506,7 @@ pub fn render_add_column(target: &str, name: &str, resolved: &ResolvedColumn) ->
 pub fn render_create_table(
     desc: &RelDescriptor,
     target_database: &str,
+    soft_delete: bool,
 ) -> Result<Option<String>, EmitterError> {
     let target = format!(
         "{}.{}",
@@ -539,8 +544,15 @@ pub fn render_create_table(
     // Synthetic columns mirror `TablePlan::build`
     col_defs.push("`_lsn` UInt64".into());
     col_defs.push("`_xid` UInt32".into());
-    col_defs.push("`_op` Enum8('insert' = 1, 'update' = 2, 'delete' = 3)".into());
     col_defs.push("`_commit_ts` DateTime64(6, 'UTC')".into());
+    col_defs.push("`_is_deleted` Bool".into());
+
+    // soft_delete keeps `_is_deleted` out of the engine args
+    let engine_args = if soft_delete {
+        "`_lsn`"
+    } else {
+        "`_lsn`, `_is_deleted`"
+    };
 
     // ORDER BY: prefer PK columns, else `_lsn`
     let order_by = if pk_attnums.is_empty() {
@@ -563,7 +575,7 @@ pub fn render_create_table(
     };
 
     let sql = format!(
-        "CREATE TABLE IF NOT EXISTS {target} (\n  {}\n) ENGINE = ReplacingMergeTree(`_lsn`)\nORDER BY {order_by}",
+        "CREATE TABLE IF NOT EXISTS {target} (\n  {}\n) ENGINE = ReplacingMergeTree({engine_args})\nORDER BY {order_by}",
         col_defs.join(",\n  ")
     );
     Ok(Some(sql))
@@ -638,6 +650,7 @@ mod tests {
             auto_create_namespaces: HashSet::new(),
             target_database: "default".into(),
             namespaces,
+            soft_delete: false,
         };
         assert_eq!(cfg.target_database_for("analytics"), "warehouse");
         assert_eq!(cfg.target_database_for("logs"), "default");
@@ -658,6 +671,7 @@ mod tests {
             auto_create_namespaces: HashSet::new(),
             target_database: "default".into(),
             namespaces: HashMap::new(),
+            soft_delete: false,
         };
         assert_eq!(cfg.drop_table_strategy, DropTableStrategy::Retain);
         let cfg = cfg.with_drop_strategy(DropTableStrategy::Drop);
@@ -741,20 +755,39 @@ mod tests {
             ],
             Some(vec![1]),
         );
-        let sql = render_create_table(&d, "default").unwrap().unwrap();
+        let sql = render_create_table(&d, "default", false).unwrap().unwrap();
         assert!(sql.contains("CREATE TABLE IF NOT EXISTS `default`.`orders`"));
         assert!(sql.contains("`id` Int32"));
         assert!(sql.contains("`body` Nullable(String)"));
         assert!(sql.contains("`_lsn` UInt64"));
-        assert!(sql.contains("`_op` Enum8"));
+        assert!(!sql.contains("`_op`"));
+        assert!(sql.contains("`_is_deleted` Bool"));
+        assert!(sql.contains("ENGINE = ReplacingMergeTree(`_lsn`, `_is_deleted`)"));
+        assert!(sql.ends_with("ORDER BY (`id`)"));
+    }
+
+    #[test]
+    fn soft_delete_keeps_is_deleted_out_of_engine_args() {
+        let d = desc(
+            "orders",
+            vec![
+                att(1, "id", INT4OID, true, None),
+                att(2, "body", TEXTOID, false, None),
+            ],
+            Some(vec![1]),
+        );
+        let sql = render_create_table(&d, "default", true).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`)"));
+        assert!(!sql.contains("ReplacingMergeTree(`_lsn`, `_is_deleted`)"));
         assert!(sql.ends_with("ORDER BY (`id`)"));
     }
 
     #[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").unwrap().unwrap();
+        let sql = render_create_table(&d, "default", false).unwrap().unwrap();
         assert!(sql.ends_with("ORDER BY (`_lsn`)"));
     }
 
@@ -763,7 +796,7 @@ mod tests {
         let mut a = att(1, "ship_at", TIMESTAMPTZOID, false, None);
         a.typmod = 3;
         let d = desc("t", vec![a], None);
-        let sql = render_create_table(&d, "db").unwrap().unwrap();
+        let sql = render_create_table(&d, "db", false).unwrap().unwrap();
         assert!(
             sql.contains("`ship_at` Nullable(DateTime64(3, 'UTC'))"),
             "{sql}"
@@ -792,7 +825,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").unwrap();
+        let sql = render_create_table(&d, "db", false).unwrap();
         assert!(sql.is_some(), "fallback path keeps the CREATE renderable");
     }