fix(core): normalize merge option columns for scd2 runtime

Author: a.metwalli

crates/floe-core/src/config/validate.rs

@@ -312,6 +312,30 @@ fn validate_merge_options(
         .iter()
         .map(|column| column.name.as_str())
         .collect::<HashSet<_>>();
+    let normalize_strategy = if entity
+        .schema
+        .normalize_columns
+        .as_ref()
+        .and_then(|normalize| normalize.enabled)
+        .unwrap_or(false)
+    {
+        entity
+            .schema
+            .normalize_columns
+            .as_ref()
+            .and_then(|normalize| normalize.strategy.as_deref())
+            .or(Some("snake_case"))
+    } else {
+        None
+    };
+    let resolved_output_columns = crate::checks::normalize::resolve_output_columns(
+        &entity.schema.columns,
+        normalize_strategy,
+    );
+    let resolved_output_column_names = resolved_output_columns
+        .iter()
+        .map(|column| column.name.as_str())
+        .collect::<HashSet<_>>();
     let primary_key_columns = entity
         .schema
         .primary_key
@@ -394,7 +418,7 @@ fn validate_merge_options(
                     entity.name, field
                 ))));
             }
-            if schema_columns.contains(value) {
+            if resolved_output_column_names.contains(value) {
                 return Err(Box::new(ConfigError(format!(
                     "entity.name={} sink.accepted.merge.scd2.{}={} collides with schema column name",
                     entity.name, field, value

crates/floe-core/src/io/write/strategy/merge/scd1.rs

@@ -105,7 +105,7 @@ impl MergeBackend for DeltaMergeBackend {
             .map(|name| name.to_string())
             .collect::<Vec<_>>();
         let merge_key_set = merge_key.iter().map(String::as_str).collect::<HashSet<_>>();
-        let ignore_columns = shared::resolve_merge_ignore_columns(ctx.entity);
+        let ignore_columns = shared::resolve_merge_ignore_columns(ctx.entity)?;
         let update_columns = source_columns
             .iter()
             .filter(|name| {

crates/floe-core/src/io/write/strategy/merge/scd2.rs

@@ -54,9 +54,9 @@ impl MergeBackend for DeltaMergeBackend {
         let mut perf = shared::DeltaMergePerfBreakdown::default();
         let merge_key = shared::resolve_merge_key(ctx.entity)?;
         let merge_key_set = merge_key.iter().map(String::as_str).collect::<HashSet<_>>();
-        let ignore_columns = shared::resolve_merge_ignore_columns(ctx.entity);
+        let ignore_columns = shared::resolve_merge_ignore_columns(ctx.entity)?;
         let compare_columns =
-            shared::resolve_merge_compare_columns(ctx.entity).unwrap_or_else(|| {
+            shared::resolve_merge_compare_columns(ctx.entity)?.unwrap_or_else(|| {
                 source_df
                     .get_column_names()
                     .iter()

crates/floe-core/src/io/write/strategy/merge/shared.rs

@@ -5,7 +5,7 @@ use deltalake::protocol::SaveMode;
 use deltalake::table::builder::DeltaTableBuilder;
 use deltalake::{datafusion::prelude::SessionContext, DeltaTable};
 use polars::prelude::DataFrame;
-use std::collections::HashSet;
+use std::collections::{HashMap, HashSet};
 
 use crate::errors::RunError;
 use crate::io::format::AcceptedMergeMetrics;
@@ -140,46 +140,80 @@ pub(crate) fn resolve_merge_key(entity: &config::EntityConfig) -> FloeResult<Vec
     Ok(primary_key.clone())
 }
 
-pub(crate) fn resolve_merge_ignore_columns(entity: &config::EntityConfig) -> HashSet<String> {
-    entity
+pub(crate) fn resolve_merge_ignore_columns(
+    entity: &config::EntityConfig,
+) -> FloeResult<HashSet<String>> {
+    let Some(columns) = entity
         .sink
         .accepted
         .merge
         .as_ref()
         .and_then(|merge| merge.ignore_columns.as_ref())
-        .map(|columns| {
-            columns
-                .iter()
-                .map(|column| column.trim())
-                .filter(|column| !column.is_empty())
-                .map(ToOwned::to_owned)
-                .collect::<HashSet<_>>()
+    else {
+        return Ok(HashSet::new());
+    };
+
+    let schema_to_output = schema_to_output_column_name_map(entity)?;
+    let resolved = columns
+        .iter()
+        .map(|column| column.trim())
+        .filter(|column| !column.is_empty())
+        .map(|column| {
+            schema_to_output
+                .get(column)
+                .cloned()
+                .unwrap_or_else(|| column.to_string())
         })
-        .unwrap_or_default()
+        .collect::<HashSet<_>>();
+    Ok(resolved)
 }
 
-pub(crate) fn resolve_merge_compare_columns(entity: &config::EntityConfig) -> Option<Vec<String>> {
-    entity
+pub(crate) fn resolve_merge_compare_columns(
+    entity: &config::EntityConfig,
+) -> FloeResult<Option<Vec<String>>> {
+    let Some(columns) = entity
         .sink
         .accepted
         .merge
         .as_ref()
         .and_then(|merge| merge.compare_columns.as_ref())
-        .map(|columns| {
-            let mut seen = HashSet::new();
-            columns
-                .iter()
-                .map(|column| column.trim())
-                .filter(|column| !column.is_empty())
-                .filter_map(|column| {
-                    if seen.insert(column.to_string()) {
-                        Some(column.to_string())
-                    } else {
-                        None
-                    }
-                })
-                .collect::<Vec<_>>()
+    else {
+        return Ok(None);
+    };
+
+    let schema_to_output = schema_to_output_column_name_map(entity)?;
+    let mut seen = HashSet::new();
+    let resolved = columns
+        .iter()
+        .map(|column| column.trim())
+        .filter(|column| !column.is_empty())
+        .map(|column| {
+            schema_to_output
+                .get(column)
+                .cloned()
+                .unwrap_or_else(|| column.to_string())
         })
+        .filter(|column| seen.insert(column.clone()))
+        .collect::<Vec<_>>();
+    Ok(Some(resolved))
+}
+
+fn schema_to_output_column_name_map(
+    entity: &config::EntityConfig,
+) -> FloeResult<HashMap<String, String>> {
+    let normalize_strategy = crate::checks::normalize::resolve_normalize_strategy(entity)?;
+    let output_columns = crate::checks::normalize::resolve_output_columns(
+        &entity.schema.columns,
+        normalize_strategy.as_deref(),
+    );
+    let mut mapping = HashMap::with_capacity(entity.schema.columns.len());
+    for (schema_column, output_column) in entity.schema.columns.iter().zip(output_columns.iter()) {
+        mapping.insert(
+            schema_column.name.trim().to_string(),
+            output_column.name.clone(),
+        );
+    }
+    Ok(mapping)
 }
 
 pub(crate) fn resolve_scd2_system_columns(entity: &config::EntityConfig) -> Scd2SystemColumns {

crates/floe-core/tests/integration/delta_run.rs

@@ -721,6 +721,89 @@ entities:
     assert_eq!(status.get(0), Some("active"));
 }
 
+#[test]
+fn local_delta_merge_scd2_compare_columns_map_to_normalized_output_names() {
+    let temp_dir = tempfile::TempDir::new().expect("temp dir");
+    let root = temp_dir.path();
+    let input_dir = root.join("in");
+    let accepted_dir = root.join("out/accepted/customer_delta");
+    let report_dir = root.join("report");
+
+    fs::create_dir_all(&input_dir).expect("create input dir");
+    write_csv(&input_dir, "batch1.csv", "id;name;status\n1;Alice;active\n");
+
+    let yaml = format!(
+        r#"version: "0.1"
+report:
+  path: "{report_dir}"
+entities:
+  - name: "customer"
+    source:
+      format: "csv"
+      path: "{input_dir}"
+    sink:
+      write_mode: "merge_scd2"
+      accepted:
+        format: "delta"
+        path: "{accepted_dir}"
+        merge:
+          compare_columns: ["Name"]
+    policy:
+      severity: "warn"
+    schema:
+      primary_key: ["id"]
+      normalize_columns:
+        enabled: true
+        strategy: "lower"
+      columns:
+        - name: "id"
+          type: "string"
+        - name: "Name"
+          type: "string"
+        - name: "status"
+          type: "string"
+"#,
+        report_dir = report_dir.display(),
+        input_dir = input_dir.display(),
+        accepted_dir = accepted_dir.display(),
+    );
+    let config_path = write_config(root, &yaml);
+
+    run(
+        &config_path,
+        RunOptions {
+            run_id: Some("it-delta-merge-scd2-compare-normalized-init".to_string()),
+            entities: Vec::new(),
+            dry_run: false,
+        },
+    )
+    .expect("initial merge_scd2 run");
+
+    fs::remove_file(input_dir.join("batch1.csv")).expect("remove first batch");
+    write_csv(
+        &input_dir,
+        "batch2.csv",
+        "id;name;status\n1;Alice Updated;active\n",
+    );
+
+    let outcome = run(
+        &config_path,
+        RunOptions {
+            run_id: Some("it-delta-merge-scd2-compare-normalized-second".to_string()),
+            entities: Vec::new(),
+            dry_run: false,
+        },
+    )
+    .expect("second merge_scd2 run");
+
+    let report = &outcome.entity_outcomes[0].report;
+    assert_eq!(report.accepted_output.updated_count, Some(1));
+    assert_eq!(report.accepted_output.inserted_count, Some(1));
+
+    let df = read_local_delta_table(&accepted_dir);
+    assert!(df.column("name").is_ok());
+}
+
 #[test]
 fn local_delta_merge_scd2_bootstrap_preserves_configured_nullable_columns() {
     let temp_dir = tempfile::TempDir::new().expect("temp dir");

crates/floe-core/tests/unit/config/config_validation.rs

@@ -1565,6 +1565,45 @@ entities:
     );
 }
 
+#[test]
+fn sink_level_merge_scd2_system_column_names_must_not_collide_with_normalized_business_columns() {
+    let yaml = r#"version: "0.1"
+entities:
+  - name: "customer"
+    source:
+      format: "csv"
+      path: "/tmp/input"
+    sink:
+      write_mode: "merge_scd2"
+      accepted:
+        format: "delta"
+        path: "/tmp/out_delta"
+        merge:
+          scd2:
+            current_flag_column: "order_id"
+    policy:
+      severity: "warn"
+    schema:
+      primary_key: ["customer_id"]
+      normalize_columns:
+        enabled: true
+        strategy: "snake_case"
+      columns:
+        - name: "customer_id"
+          type: "string"
+        - name: "Order ID"
+          type: "string"
+"#;
+    assert_validation_error(
+        yaml,
+        &[
+            "entity.name=customer",
+            "sink.accepted.merge.scd2.current_flag_column=order_id",
+            "collides with schema column name",
+        ],
+    );
+}
+
 #[test]
 fn sink_level_merge_scd2_system_column_names_must_be_unique() {
     let yaml = r#"version: "0.1"