Add keep strategy for conditional column passthrough (#77)

AGENTS.md

@@ -259,7 +259,7 @@ Follow these steps in order. Do not skip any step.
 
 - **`table_options` is intentionally removed**: The `[table_options]` key was deprecated and now deliberately fails with a targeted error pointing to `[rules]` and `[column_cases]`. Do not re-introduce it.
 
-- **`null` and `redact` reject `domain`**: These strategies produce constant outputs regardless of input, so deterministic domain mapping is meaningless. The validator rejects `domain` with these strategies.
+- **`null` and `redact` reject `domain`**: These strategies produce constant outputs regardless of input, so deterministic domain mapping is meaningless. The validator rejects `domain` with these strategies. The same applies to **`blank`**, **`keep`**, **`empty_array`**, and **`empty_object`**. **`keep`** is additionally restricted to `[column_cases]` only.
 
 - **COPY NULL representation**: In COPY format, `\N` (backslash-N) means NULL. This is different from the string `"NULL"` used in INSERT VALUES format. Handle both correctly.
 

README.md

@@ -34,7 +34,7 @@
 
 ## Why Dumpling?
 
-- **Rich built-in strategies** — from fast clears (`null`, `redact`, `blank`, `empty_array` / `empty_object`) and bounded fakes (`int_range`, `decimal`, `string`) to realistic stand-ins (`email`, `name`, `payment_card`, `faker`, date/time fuzz), with optional **`domain`** so the same source value stays consistent across tables.
+- **Rich built-in strategies** — from fast clears (`null`, `redact`, `blank`, `empty_array` / `empty_object`) and bounded fakes (`int_range`, `decimal`, `string`) to realistic stand-ins (`email`, `name`, `payment_card`, `faker`, date/time fuzz), plus conditional **`keep`** under `column_cases` when some rows must retain the original value, with optional **`domain`** so the same source value stays consistent across tables.
 - **JSON inside columns** — target paths inside `json` / `jsonb` text with the same dot or `__` syntax you use elsewhere; pair with row filters on nested fields.
 - **Row-level control** — **`retain`** and **`delete`** predicates (including nested JSON paths) drop or keep whole rows before transforms run.
 - **Offline by design** — works on dump files only; nothing connects to your database.
@@ -133,7 +133,7 @@ Column rules live under `[rules."schema.table"]` (or `[rules."table"]`) as inlin
 
 #### Choosing a strategy (cheaper vs more realistic)
 
-Prefer **lightweight** strategies when nothing downstream requires lifelike values: **`null`**, **`redact`**, **`blank`**, **`empty_array`**, **`empty_object`**, **`string`**, **`int_range`**, and **`decimal`** are cheap to generate (simple constants, random digits/alnum, or bounded numeric shapes). Use **`blank`** for NOT NULL text where you must clear content without SQL NULL; use **`empty_array`** / **`empty_object`** on JSON path rules (or text columns holding JSON) when the document must keep `[]` / `{}` instead of `null` or `""`.
+Prefer **lightweight** strategies when nothing downstream requires lifelike values: **`null`**, **`redact`**, **`blank`**, **`empty_array`**, **`empty_object`**, **`string`**, **`int_range`**, and **`decimal`** are cheap to generate (simple constants, random digits/alnum, or bounded numeric shapes). Use **`blank`** for NOT NULL text where you must clear content without SQL NULL; use **`empty_array`** / **`empty_object`** on JSON path rules (or text columns holding JSON) when the document must keep `[]` / `{}` instead of `null` or `""`. Use **`keep`** only as a **`column_cases`** override when a default rule scrubs the column but some rows must retain the original value (for example staff allowlist emails).
 
 Reach for **richer** strategies when realism matters for restores, demos, or tests that exercise parsers and validators: **`email`**, **`name`**, **`first_name`**, **`last_name`**, **`phone`**, **`faker`**, **`uuid`**, **`hash`**, **`payment_card`**, and the **`date_fuzz` / `time_fuzz` / `datetime_fuzz`** family do more work (formatting, parsing, digest, or upstream generators). If a cheap strategy would break **CHECK constraints**, **NOT NULL**, **foreign-key shape**, or **import tooling** that validates formats, switch to a strategy that emits compatible values—or keep **`domain`** on the heavier strategy so referential consistency is preserved where you need it.
 
@@ -152,6 +152,14 @@ Reach for **richer** strategies when realism matters for restores, demos, or tes
 - **Behavior:** replace with an **empty string** (`''` in SQL when quoted). If the source cell is SQL **`NULL`**, the cell stays **`NULL`** (same as `null` / `redact` semantics for missing values).
 - **Options:** none. (`domain` is rejected.) **`as_string`** is ignored; output is always the empty string literal when non-NULL.
 
+#### `keep`
+
+- **Behavior:** leave the cell **unchanged** (exact original INSERT/COPY bytes, including SQL `NULL` / COPY `\N`).
+- **Where allowed:** only under **`[column_cases]`**. A default `[rules]` entry of `keep` is rejected so accidental whole-column no-ops stay hard to configure—pair a scrubbing default in `[rules]` with a matching `keep` case for exceptions.
+- **Coverage:** selecting `keep` via `column_cases` still counts the column as covered for `--strict-coverage` / `[sensitive_columns]` (explicit decision to retain).
+- **Reporting / `--check`:** kept cells are not counted as changed.
+- **Options:** none. (`domain`, `as_string`, and other strategy options are rejected.)
+
 #### `empty_array` / `empty_object`
 
 - **Behavior:** replace with the JSON tokens **`[]`** and **`{}`** as **unquoted** SQL/COPY tokens (so they parse as JSON when the column holds JSON). If the source cell is SQL **`NULL`**, the cell stays **`NULL`**.
@@ -243,6 +251,20 @@ when.any = [{ column = "country", op = "in", values = ["DE","FR","GB"] }]
 strategy = { strategy = "hash", salt = "eu-salt", as_string = true }
 ```
 
+To **retain** some values while scrubbing the rest, put the scrubbing strategy in `[rules]` and add a first-match `keep` case:
+
+```toml
+[rules."public.reskinned_inventory_user"]
+email = { strategy = "email", domain = "wms_user_email", unique_within_domain = true }
+
+[[column_cases."public.reskinned_inventory_user".email]]
+when.any = [
+  { column = "email", op = "ilike", value = "%@wearecrew.com" },
+  { column = "email", op = "ilike", value = "%@reskinned.clothing" },
+]
+strategy = { strategy = "keep" }
+```
+
 - `when.any` is OR, `when.all` is AND; you can use either or both. If both are empty, the case matches unconditionally.
 - First-match-wins per column; there is no merge or fallthrough.
 - Row filtering (`row_filters`) is evaluated before cases; deleted rows are not transformed.

src/settings.rs

@@ -78,7 +78,7 @@ pub fn merge_keep_original(cli: bool, cfg: Option<bool>) -> bool {
 
 #[derive(Debug, Clone, Deserialize, Serialize)]
 pub struct AnonymizerSpec {
-    /// Strategy name: null|redact|blank|empty_array|empty_object|uuid|hash|… (see README)
+    /// Strategy name: null|redact|blank|keep|empty_array|empty_object|uuid|hash|… (see README)
     pub strategy: String,
     /// if strategy=hash: optional per-column salt override; otherwise ignored
     pub salt: Option<String>,
@@ -607,6 +607,7 @@ const KNOWN_STRATEGIES: &[&str] = &[
     "null",
     "redact",
     "blank",
+    "keep",
     "empty_array",
     "empty_object",
     "uuid",
@@ -730,6 +731,15 @@ fn validate_anonymizer_spec(spec: &AnonymizerSpec, path: &str) -> anyhow::Result
             KNOWN_STRATEGIES.join(", ")
         );
     }
+    // `keep` is only meaningful as a conditional override; a default keep rule would be a
+    // silent no-op policy. Require it under [column_cases] with a scrubbing default in [rules].
+    if strategy == "keep" && path.starts_with("rules.") {
+        anyhow::bail!(
+            "{}: strategy 'keep' is only allowed under [column_cases] \
+             (set a scrubbing default in [rules] and use keep for matching exceptions)",
+            path
+        );
+    }
 
     let mut unsupported: Vec<&str> = Vec::new();
     let domain = spec.domain.as_deref().map(str::trim);
@@ -742,14 +752,17 @@ fn validate_anonymizer_spec(spec: &AnonymizerSpec, path: &str) -> anyhow::Result
     if domain.is_some()
         && matches!(
             strategy,
-            "null" | "redact" | "blank" | "empty_array" | "empty_object"
+            "null" | "redact" | "blank" | "keep" | "empty_array" | "empty_object"
         )
     {
         unsupported.push("domain");
         if spec.unique_within_domain.is_some() {
             unsupported.push("unique_within_domain");
         }
     }
+    if spec.as_string.is_some() && strategy == "keep" {
+        unsupported.push("as_string");
+    }
     if spec.salt.is_some() && strategy != "hash" {
         unsupported.push("salt");
     }
@@ -1506,6 +1519,65 @@ notes = { strategy = "blank", domain = "x" }
         let _ = fs::remove_file(path);
     }
 
+    #[test]
+    fn keep_strategy_rejected_under_rules() {
+        let path = write_temp_config(
+            r#"
+[rules."public.users"]
+email = { strategy = "keep" }
+"#,
+        );
+        let err =
+            load_config(Some(&path), false).expect_err("expected semantic validation failure");
+        let msg = format!("{:#}", err);
+        assert!(msg.contains("rules.\"public.users\".email"));
+        assert!(msg.contains("only allowed under [column_cases]"));
+        let _ = fs::remove_file(path);
+    }
+
+    #[test]
+    fn keep_strategy_allowed_under_column_cases() {
+        let path = write_temp_config(
+            r#"
+[rules."public.users"]
+email = { strategy = "email", domain = "user_email" }
+
+[[column_cases."public.users".email]]
+when.any = [{ column = "email", op = "ilike", value = "%@example.com" }]
+strategy = { strategy = "keep" }
+"#,
+        );
+        let cfg = load_config(Some(&path), false).expect("keep under column_cases must load");
+        let cases = cfg
+            .column_cases
+            .get("public.users")
+            .and_then(|m| m.get("email"))
+            .expect("email cases");
+        assert_eq!(cases.len(), 1);
+        assert_eq!(cases[0].strategy.strategy, "keep");
+        let _ = fs::remove_file(path);
+    }
+
+    #[test]
+    fn keep_strategy_rejects_domain() {
+        let path = write_temp_config(
+            r#"
+[rules."public.users"]
+email = { strategy = "email" }
+
+[[column_cases."public.users".email]]
+when.any = [{ column = "email", op = "ilike", value = "%@staff.example" }]
+strategy = { strategy = "keep", domain = "noop" }
+"#,
+        );
+        let err =
+            load_config(Some(&path), false).expect_err("expected semantic validation failure");
+        let msg = format!("{:#}", err);
+        assert!(msg.contains("column_cases.\"public.users\".email[0].strategy"));
+        assert!(msg.contains("domain"));
+        let _ = fs::remove_file(path);
+    }
+
     #[test]
     fn malformed_strategy_parameters_fail_validation() {
         let path = write_temp_config(

src/sql.rs

@@ -146,7 +146,7 @@ impl SqlStreamProcessor {
                                 .with_context(|| {
                                     format!(
                                         "failed processing INSERT statement starting with: {}",
-                                        &insert_buf.lines().next().unwrap_or("").trim()
+                                        insert_buf.lines().next().unwrap_or("").trim()
                                     )
                                 })?;
                             if !transformed.is_empty() {
@@ -204,7 +204,7 @@ impl SqlStreamProcessor {
                                 .with_context(|| {
                                     format!(
                                         "failed processing INSERT statement starting with: {}",
-                                        &insert_buf.lines().next().unwrap_or("").trim()
+                                        insert_buf.lines().next().unwrap_or("").trim()
                                     )
                                 })?;
                         if !transformed.is_empty() {
@@ -530,6 +530,11 @@ impl SqlStreamProcessor {
         let col_len = self.lookup_column_max_length(schema, table, col);
 
         if let Some(spec) = selected {
+            // Exact passthrough: leave INSERT/COPY cell bytes unchanged (including NULL / \N).
+            // Coverage still counts via the column_cases / rules entry that selected `keep`.
+            if spec.strategy == "keep" {
+                return Ok(None);
+            }
             let repl = apply_anonymizer(&self.anonymizers, &spec, cell_original, col_len);
             return Ok(Some((repl, vec![spec])));
         }
@@ -2899,6 +2904,152 @@ INSERT INTO public.users (id, email, country, is_admin) VALUES
         assert!(s.contains("INSERT INTO public.users"));
     }
 
+    #[test]
+    fn column_cases_keep_preserves_staff_emails_insert_and_copy() {
+        // Default scrub; keep staff domains via column_cases (issue #77).
+        let mut rules: HashMap<String, HashMap<String, AnonymizerSpec>> = HashMap::new();
+        let mut base_cols: HashMap<String, AnonymizerSpec> = HashMap::new();
+        base_cols.insert(
+            "email".to_string(),
+            AnonymizerSpec {
+                strategy: "redact".to_string(),
+                salt: None,
+                min: None,
+                max: None,
+                scale: None,
+                length: None,
+                min_days: None,
+                max_days: None,
+                min_seconds: None,
+                max_seconds: None,
+                domain: None,
+                unique_within_domain: None,
+                as_string: Some(true),
+                locale: None,
+                faker: None,
+                format: None,
+            },
+        );
+        rules.insert("public.users".to_string(), base_cols);
+
+        let mut column_cases: HashMap<String, HashMap<String, Vec<ColumnCase>>> = HashMap::new();
+        let email_cases: Vec<ColumnCase> = vec![ColumnCase {
+            when: When {
+                any: vec![
+                    crate::settings::Predicate {
+                        column: "email".into(),
+                        op: "ilike".into(),
+                        value: Some(serde_json::json!("%@wearecrew.com")),
+                        values: None,
+                        case_insensitive: None,
+                    },
+                    crate::settings::Predicate {
+                        column: "email".into(),
+                        op: "ilike".into(),
+                        value: Some(serde_json::json!("%@reskinned.clothing")),
+                        values: None,
+                        case_insensitive: None,
+                    },
+                ],
+                all: vec![],
+            },
+            strategy: AnonymizerSpec {
+                strategy: "keep".into(),
+                salt: None,
+                min: None,
+                max: None,
+                scale: None,
+                length: None,
+                min_days: None,
+                max_days: None,
+                min_seconds: None,
+                max_seconds: None,
+                domain: None,
+                unique_within_domain: None,
+                as_string: None,
+                locale: None,
+                faker: None,
+                format: None,
+            },
+        }];
+        let mut per_col: HashMap<String, Vec<ColumnCase>> = HashMap::new();
+        per_col.insert("email".into(), email_cases);
+        column_cases.insert("public.users".into(), per_col);
+
+        let mut sensitive_columns = HashMap::new();
+        sensitive_columns.insert(
+            "public.users".to_string(),
+            std::collections::HashSet::from(["email".to_string()]),
+        );
+
+        let cfg = ResolvedConfig {
+            salt: None,
+            rules,
+            row_filters: HashMap::new(),
+            column_cases,
+            sensitive_columns,
+            output_scan: crate::settings::OutputScanConfig::default(),
+            pg_restore: crate::settings::PgRestoreConfig::default(),
+            keep_original: None,
+            source_path: None,
+        };
+        let reg = AnonymizerRegistry::from_config(&cfg);
+        let mut proc = SqlStreamProcessor::new(reg, cfg, None, DumpFormat::Postgres);
+        let input = r#"
+CREATE TABLE public.users (id int, email text);
+INSERT INTO public.users (id, email) VALUES
+  (1, 'alice@gmail.com'),
+  (2, 'bob@wearecrew.com'),
+  (3, 'carol@reskinned.clothing'),
+  (4, NULL);
+COPY public.users (id, email) FROM stdin;
+5	dave@gmail.com
+6	erin@wearecrew.com
+\.
+"#;
+        let mut reader = std::io::BufReader::new(input.as_bytes());
+        let mut out = Vec::new();
+        proc.process(&mut reader, &mut out).unwrap();
+        let s = String::from_utf8(out).unwrap();
+
+        assert!(
+            !s.contains("alice@gmail.com"),
+            "non-staff INSERT email must be scrubbed"
+        );
+        assert!(
+            !s.contains("dave@gmail.com"),
+            "non-staff COPY email must be scrubbed"
+        );
+        assert!(
+            s.contains("'bob@wearecrew.com'"),
+            "staff INSERT email must be kept: {s}"
+        );
+        assert!(
+            s.contains("'carol@reskinned.clothing'"),
+            "staff INSERT email must be kept: {s}"
+        );
+        assert!(
+            s.contains("erin@wearecrew.com"),
+            "staff COPY email must be kept: {s}"
+        );
+        assert!(
+            s.contains("'REDACTED'"),
+            "default redact must still scrub non-matching rows: {s}"
+        );
+
+        let coverage = proc.sensitive_coverage_summary();
+        assert!(
+            coverage.covered.iter().any(|c| c == "public.users.email"),
+            "keep cases must count as strict-coverage: {:?}",
+            coverage.covered
+        );
+        assert!(
+            coverage.uncovered.is_empty(),
+            "email must not be uncovered: {:?}",
+            coverage.uncovered
+        );
+    }
+
     #[test]
     fn deterministic_domain_mapping_is_consistent_across_tables() {
         let mut rules: HashMap<String, HashMap<String, AnonymizerSpec>> = HashMap::new();