feat: pass ColumnMappingMode to physical_name (#1403)

<!--
Thanks for sending a pull request!  Here are some tips for you:
1. If this is your first time, please read our contributor guidelines:
https://github.com/delta-incubator/delta-kernel-rs/blob/main/CONTRIBUTING.md
2. Run `cargo t --all-features --all-targets` to get started testing,
and run `cargo fmt`.
  3. Ensure you have added or run the appropriate tests for your PR.
4. If the PR is unfinished, add '[WIP]' in your PR title, e.g., '[WIP]
Your PR title ...'.
  5. Be sure to keep the PR description updated to reflect all changes.
-->
Fixes #1120
<!--
PR title formatting:
This project uses conventional commits:
https://www.conventionalcommits.org/

Each PR corresponds to a commit on the `main` branch, with the title of
the PR (typically) being
used for the commit message on main. In order to ensure proper
formatting in the CHANGELOG please
ensure your PR title adheres to the conventional commit specification.

Examples:
- new feature PR: "feat: new API for snapshot.update()"
- bugfix PR: "fix: correctly apply DV in read-table example"
-->

## What changes are proposed in this pull request?
This PR makes StructField::physical_name() aware of ColumnMappingMode by
adding it as a required parameter.

Previously, physical_name() would always return the physical name from
metadata if present, regardless of the column mapping mode. This was
incorrect behavior - when column mapping mode is
None, the logical name should be used even if physical name metadata
exists in the field.

  The key changes are:
1. Updated StructField::physical_name() to take column_mapping_mode:
ColumnMappingMode parameter
  2. When mode is None, always return the logical name
3. When mode is Id or Name, return the physical name from metadata (or
logical name as fallback)
4. Added column_mapping_mode field to StateInfo struct to avoid
threading it through many function calls
5. Updated all call sites throughout the codebase to pass the column
mapping mode
6. Refactored AddRemoveDedupVisitor to accept Arc<StateInfo> instead of
8 individual parameters, following feedback to avoid excessive parameter
passing

<!--
Uncomment this section if there are any changes affecting public APIs:
### This PR affects the following public APIs

If there are breaking changes, please ensure the `breaking-changes`
label gets added by CI, and describe why the changes are needed.

Note that _new_ public APIs are not considered breaking.
-->

---------

Co-authored-by: Nick Lanham <[email protected]>
Co-authored-by: Zach Schuermann <[email protected]>

Author: 3 people

kernel/src/scan/log_replay.rs

@@ -13,8 +13,8 @@ use crate::kernel_predicates::{DefaultKernelPredicateEvaluator, KernelPredicateE
 use crate::log_replay::{ActionsBatch, FileActionDeduplicator, FileActionKey, LogReplayProcessor};
 use crate::scan::Scalar;
 use crate::schema::ToSchema as _;
-use crate::schema::{ColumnNamesAndTypes, DataType, MapType, SchemaRef, StructField, StructType};
-use crate::transforms::{get_transform_expr, parse_partition_values, TransformSpec};
+use crate::schema::{ColumnNamesAndTypes, DataType, MapType, StructField, StructType};
+use crate::transforms::{get_transform_expr, parse_partition_values};
 use crate::utils::require;
 use crate::{DeltaResult, Engine, Error, ExpressionEvaluator};
 
@@ -93,9 +93,7 @@ impl ScanLogReplayProcessor {
 struct AddRemoveDedupVisitor<'seen> {
     deduplicator: FileActionDeduplicator<'seen>,
     selection_vector: Vec<bool>,
-    logical_schema: SchemaRef,
-    physical_schema: SchemaRef,
-    transform_spec: Option<Arc<TransformSpec>>,
+    state_info: Arc<StateInfo>,
     partition_filter: Option<PredicateRef>,
     row_transform_exprs: Vec<Option<ExpressionRef>>,
 }
@@ -113,9 +111,7 @@ impl AddRemoveDedupVisitor<'_> {
     fn new(
         seen: &mut HashSet<FileActionKey>,
         selection_vector: Vec<bool>,
-        logical_schema: SchemaRef,
-        physical_schema: SchemaRef,
-        transform_spec: Option<Arc<TransformSpec>>,
+        state_info: Arc<StateInfo>,
         partition_filter: Option<PredicateRef>,
         is_log_batch: bool,
     ) -> AddRemoveDedupVisitor<'_> {
@@ -129,9 +125,7 @@ impl AddRemoveDedupVisitor<'_> {
                 Self::REMOVE_DV_START_INDEX,
             ),
             selection_vector,
-            logical_schema,
-            physical_schema,
-            transform_spec,
+            state_info,
             partition_filter,
             row_transform_exprs: Vec::new(),
         }
@@ -179,12 +173,16 @@ impl AddRemoveDedupVisitor<'_> {
         // WARNING: It's not safe to partition-prune removes (just like it's not safe to data skip
         // removes), because they are needed to suppress earlier incompatible adds we might
         // encounter if the table's schema was replaced after the most recent checkpoint.
-        let partition_values = match &self.transform_spec {
+        let partition_values = match &self.state_info.transform_spec {
             Some(transform) if is_add => {
                 let partition_values =
                     getters[Self::ADD_PARTITION_VALUES_INDEX].get(i, "add.partitionValues")?;
-                let partition_values =
-                    parse_partition_values(&self.logical_schema, transform, &partition_values)?;
+                let partition_values = parse_partition_values(
+                    &self.state_info.logical_schema,
+                    transform,
+                    &partition_values,
+                    self.state_info.column_mapping_mode,
+                )?;
                 if self.is_file_partition_pruned(&partition_values) {
                     return Ok(false);
                 }
@@ -200,13 +198,14 @@ impl AddRemoveDedupVisitor<'_> {
         let base_row_id: Option<i64> =
             getters[Self::BASE_ROW_ID_INDEX].get_opt(i, "add.baseRowId")?;
         let transform = self
+            .state_info
             .transform_spec
             .as_ref()
             .map(|transform| {
                 get_transform_expr(
                     transform,
                     partition_values,
-                    &self.physical_schema,
+                    &self.state_info.physical_schema,
                     base_row_id,
                 )
             })
@@ -351,9 +350,7 @@ impl LogReplayProcessor for ScanLogReplayProcessor {
         let mut visitor = AddRemoveDedupVisitor::new(
             &mut self.seen_file_keys,
             selection_vector,
-            self.state_info.logical_schema.clone(),
-            self.state_info.physical_schema.clone(),
-            self.state_info.transform_spec.clone(),
+            self.state_info.clone(),
             self.partition_filter.clone(),
             is_log_batch,
         );
@@ -407,6 +404,7 @@ mod tests {
     };
     use crate::scan::PhysicalPredicate;
     use crate::schema::MetadataColumnSpec;
+    use crate::table_features::ColumnMappingMode;
     use crate::Expression as Expr;
     use crate::{
         engine::sync::SyncEngine,
@@ -471,6 +469,7 @@ mod tests {
             physical_schema: logical_schema.clone(),
             physical_predicate: PhysicalPredicate::None,
             transform_spec: None,
+            column_mapping_mode: ColumnMappingMode::None,
         });
         let iter = scan_action_iter(
             &SyncEngine::new(),

kernel/src/scan/mod.rs

@@ -27,6 +27,7 @@ use crate::schema::{
     ArrayType, DataType, MapType, PrimitiveType, Schema, SchemaRef, SchemaTransform, StructField,
     ToSchema as _,
 };
+use crate::table_features::ColumnMappingMode;
 use crate::{DeltaResult, Engine, EngineData, Error, FileMeta, SnapshotRef, Version};
 
 use self::log_replay::scan_action_iter;
@@ -151,6 +152,7 @@ impl PhysicalPredicate {
     pub(crate) fn try_new(
         predicate: &Predicate,
         logical_schema: &Schema,
+        column_mapping_mode: ColumnMappingMode,
     ) -> DeltaResult<PhysicalPredicate> {
         if can_statically_skip_all_files(predicate) {
             return Ok(PhysicalPredicate::StaticSkipAll);
@@ -160,6 +162,7 @@ impl PhysicalPredicate {
             column_mappings: HashMap::new(),
             logical_path: vec![],
             physical_path: vec![],
+            column_mapping_mode,
         };
         let schema_opt = get_referenced_fields.transform_struct(logical_schema);
         let mut unresolved = get_referenced_fields.unresolved_references.into_iter();
@@ -210,6 +213,7 @@ struct GetReferencedFields<'a> {
     column_mappings: HashMap<ColumnName, ColumnName>,
     logical_path: Vec<String>,
     physical_path: Vec<String>,
+    column_mapping_mode: ColumnMappingMode,
 }
 impl<'a> SchemaTransform<'a> for GetReferencedFields<'a> {
     // Capture the path mapping for this leaf field
@@ -235,7 +239,7 @@ impl<'a> SchemaTransform<'a> for GetReferencedFields<'a> {
     }
 
     fn transform_struct_field(&mut self, field: &'a StructField) -> Option<Cow<'a, StructField>> {
-        let physical_name = field.physical_name();
+        let physical_name = field.physical_name(self.column_mapping_mode);
         self.logical_path.push(field.name.clone());
         self.physical_path.push(physical_name.to_string());
         let field = self.recurse_into_struct_field(field);
@@ -729,6 +733,7 @@ pub(crate) mod test_utils {
 
     use super::state::ScanCallback;
     use super::PhysicalPredicate;
+    use crate::table_features::ColumnMappingMode;
     use crate::transforms::TransformSpec;
 
     // Generates a batch of sidecar actions with the given paths.
@@ -843,6 +848,7 @@ pub(crate) mod test_utils {
             physical_schema: logical_schema,
             physical_predicate: PhysicalPredicate::None,
             transform_spec,
+            column_mapping_mode: ColumnMappingMode::None,
         });
         let iter = scan_action_iter(
             &SyncEngine::new(),
@@ -1053,7 +1059,9 @@ mod tests {
         ];
 
         for (predicate, expected) in test_cases {
-            let result = PhysicalPredicate::try_new(&predicate, &logical_schema).ok();
+            let result =
+                PhysicalPredicate::try_new(&predicate, &logical_schema, ColumnMappingMode::Name)
+                    .ok();
             assert_eq!(
                 result, expected,
                 "Failed for predicate: {predicate:#?}, expected {expected:#?}, got {result:#?}"

kernel/src/scan/state_info.rs

@@ -25,6 +25,8 @@ pub(crate) struct StateInfo {
     pub(crate) physical_predicate: PhysicalPredicate,
     /// Transform specification for converting physical to logical data
     pub(crate) transform_spec: Option<Arc<TransformSpec>>,
+    /// The column mapping mode for this scan
+    pub(crate) column_mapping_mode: ColumnMappingMode,
 }
 
 /// Validating the metadata columns also extracts information needed to properly construct the full
@@ -192,7 +194,7 @@ impl StateInfo {
         let physical_schema = Arc::new(StructType::try_new(read_fields)?);
 
         let physical_predicate = match predicate {
-            Some(pred) => PhysicalPredicate::try_new(&pred, &logical_schema)?,
+            Some(pred) => PhysicalPredicate::try_new(&pred, &logical_schema, column_mapping_mode)?,
             None => PhysicalPredicate::None,
         };
 
@@ -208,6 +210,7 @@ impl StateInfo {
             physical_schema,
             physical_predicate,
             transform_spec,
+            column_mapping_mode,
         })
     }
 }

kernel/src/schema/mod.rs

@@ -314,17 +314,26 @@ impl StructField {
 
     /// Get the physical name for this field as it should be read from parquet.
     ///
+    /// When `column_mapping_mode` is `None`, always returns the logical name (even if physical
+    /// name metadata is present). When mode is `Id` or `Name`, returns the physical name from
+    /// metadata if present, otherwise returns the logical name.
+    ///
     /// NOTE: Caller affirms that the schema was already validated by
     /// [`crate::table_features::validate_schema_column_mapping`], to ensure that annotations are
     /// always and only present when column mapping mode is enabled.
     #[internal_api]
-    pub(crate) fn physical_name(&self) -> &str {
-        match self
-            .metadata
-            .get(ColumnMetadataKey::ColumnMappingPhysicalName.as_ref())
-        {
-            Some(MetadataValue::String(physical_name)) => physical_name,
-            _ => &self.name,
+    pub(crate) fn physical_name(&self, column_mapping_mode: ColumnMappingMode) -> &str {
+        match column_mapping_mode {
+            ColumnMappingMode::None => &self.name,
+            ColumnMappingMode::Id | ColumnMappingMode::Name => {
+                match self
+                    .metadata
+                    .get(ColumnMetadataKey::ColumnMappingPhysicalName.as_ref())
+                {
+                    Some(MetadataValue::String(physical_name)) => physical_name,
+                    _ => &self.name,
+                }
+            }
         }
     }
 
@@ -422,7 +431,7 @@ impl StructField {
                                     .is_some_and(|x| matches!(x, MetadataValue::String(_))));
                             }
                         }
-                        field.physical_name().to_owned()
+                        field.physical_name(self.column_mapping_mode).to_owned()
                     }
                 };
 
@@ -1901,7 +1910,7 @@ mod tests {
                 assert!(matches!(col_id, MetadataValue::Number(num) if *num == 4));
                 assert!(matches!(id_start, MetadataValue::Number(num) if *num == 2147483648i64));
                 assert_eq!(
-                    field.physical_name(),
+                    field.physical_name(mode),
                     "col-5f422f40-de70-45b2-88ab-1d5c90e94db1"
                 );
                 let physical_field = field.make_physical(mode);

kernel/src/table_changes/log_replay/tests.rs

@@ -10,7 +10,7 @@ use crate::scan::state::DvInfo;
 use crate::scan::PhysicalPredicate;
 use crate::schema::{DataType, StructField, StructType};
 use crate::table_changes::log_replay::LogReplayScanner;
-use crate::table_features::TableFeature;
+use crate::table_features::{ColumnMappingMode, TableFeature};
 use crate::utils::test_utils::{assert_result_error_with_message, Action, LocalMockTable};
 use crate::Predicate;
 use crate::{DeltaResult, Engine, Error, Version};
@@ -539,10 +539,11 @@ async fn data_skipping_filter() {
         Scalar::from(4),
     );
     let logical_schema = get_schema();
-    let predicate = match PhysicalPredicate::try_new(&predicate, &logical_schema) {
-        Ok(PhysicalPredicate::Some(p, s)) => Some((p, s)),
-        other => panic!("Unexpected result: {other:?}"),
-    };
+    let predicate =
+        match PhysicalPredicate::try_new(&predicate, &logical_schema, ColumnMappingMode::None) {
+            Ok(PhysicalPredicate::Some(p, s)) => Some((p, s)),
+            other => panic!("Unexpected result: {other:?}"),
+        };
     let commits = get_segment(engine.as_ref(), mock_table.table_root(), 0, None)
         .unwrap()
         .into_iter();

kernel/src/table_changes/physical_to_logical.rs

@@ -107,6 +107,7 @@ pub(crate) fn get_cdf_transform_expr(
         &state_info.logical_schema,
         transform_spec,
         &scan_file.partition_values,
+        state_info.column_mapping_mode,
     )?;
     partition_values.extend(parsed_values);
 
@@ -131,6 +132,7 @@ mod tests {
     use crate::scan::state_info::StateInfo;
     use crate::scan::PhysicalPredicate;
     use crate::schema::{DataType, StructField, StructType};
+    use crate::table_features::ColumnMappingMode;
     use crate::transforms::FieldTransformSpec;
     use std::collections::HashMap;
     use std::sync::Arc;
@@ -180,6 +182,7 @@ mod tests {
             physical_schema: physical_schema.into(),
             physical_predicate: PhysicalPredicate::None,
             transform_spec: Some(Arc::new(transform_spec)),
+            column_mapping_mode: ColumnMappingMode::None,
         }
     }
 
@@ -397,6 +400,7 @@ mod tests {
             physical_schema: physical_schema.clone().into(),
             physical_predicate: PhysicalPredicate::None,
             transform_spec: Some(Arc::new(transform_spec)),
+            column_mapping_mode: ColumnMappingMode::None,
         };
 
         let result = get_cdf_transform_expr(&scan_file, &state_info, &physical_schema);

kernel/src/transforms.rs

@@ -13,6 +13,7 @@ use crate::expressions::{
     BinaryExpressionOp, Expression, ExpressionRef, Scalar, Transform, VariadicExpressionOp,
 };
 use crate::schema::{DataType, SchemaRef, StructType};
+use crate::table_features::ColumnMappingMode;
 use crate::{DeltaResult, Error};
 
 /// A list of field transforms that describes a transform expression to be created at scan time.
@@ -70,13 +71,14 @@ pub(crate) fn parse_partition_value(
     field_idx: usize,
     logical_schema: &SchemaRef,
     partition_values: &HashMap<String, String>,
+    column_mapping_mode: ColumnMappingMode,
 ) -> DeltaResult<(usize, (String, Scalar))> {
     let Some(field) = logical_schema.field_at_index(field_idx) else {
         return Err(Error::InternalError(format!(
             "out of bounds partition column field index {field_idx}"
         )));
     };
-    let name = field.physical_name();
+    let name = field.physical_name(column_mapping_mode);
     let partition_value = parse_partition_value_raw(partition_values.get(name), field.data_type())?;
     Ok((field_idx, (name.to_string(), partition_value)))
 }
@@ -86,13 +88,19 @@ pub(crate) fn parse_partition_values(
     logical_schema: &SchemaRef,
     transform_spec: &TransformSpec,
     partition_values: &HashMap<String, String>,
+    column_mapping_mode: ColumnMappingMode,
 ) -> DeltaResult<HashMap<usize, (String, Scalar)>> {
     transform_spec
         .iter()
         .filter_map(|field_transform| match field_transform {
-            FieldTransformSpec::MetadataDerivedColumn { field_index, .. } => Some(
-                parse_partition_value(*field_index, logical_schema, partition_values),
-            ),
+            FieldTransformSpec::MetadataDerivedColumn { field_index, .. } => {
+                Some(parse_partition_value(
+                    *field_index,
+                    logical_schema,
+                    partition_values,
+                    column_mapping_mode,
+                ))
+            }
             FieldTransformSpec::DynamicColumn { .. }
             | FieldTransformSpec::StaticInsert { .. }
             | FieldTransformSpec::GenerateRowId { .. }
@@ -221,7 +229,7 @@ mod tests {
         )]));
         let partition_values = HashMap::new();
 
-        let result = parse_partition_value(5, &schema, &partition_values);
+        let result = parse_partition_value(5, &schema, &partition_values, ColumnMappingMode::None);
         assert_result_error_with_message(result, "out of bounds");
     }
 
@@ -256,7 +264,13 @@ mod tests {
         partition_values.insert("id".to_string(), "test".to_string());
         partition_values.insert("_change_type".to_string(), "insert".to_string());
 
-        let result = parse_partition_values(&schema, &transform_spec, &partition_values).unwrap();
+        let result = parse_partition_values(
+            &schema,
+            &transform_spec,
+            &partition_values,
+            ColumnMappingMode::None,
+        )
+        .unwrap();
         assert_eq!(result.len(), 2);
         assert!(result.contains_key(&0));
         assert!(result.contains_key(&1));
@@ -276,7 +290,13 @@ mod tests {
         let transform_spec = vec![];
         let partition_values = HashMap::new();
 
-        let result = parse_partition_values(&schema, &transform_spec, &partition_values).unwrap();
+        let result = parse_partition_values(
+            &schema,
+            &transform_spec,
+            &partition_values,
+            ColumnMappingMode::None,
+        )
+        .unwrap();
         assert!(result.is_empty());
     }