refactor: Make get_cdf_transform_expr return Option<ExpressionRef> (delta-io#1401)

<!--
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.
-->


<!--
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?
<!--
Please clarify what changes you are proposing and why the changes are
needed.
The purpose of this section is to outline the changes, why they are
needed, and how this PR fixes the issue.
If the reason for the change is already explained clearly in an issue,
then it does not need to be restated here.
1. If you propose a new API or feature, clarify the use case for a new
API or feature.
  2. If you fix a bug, you can clarify why it is a bug.
-->
delta-io#1378
Make get_cdf_transform_expr return Option<ExpressionRef> to avoid
creating unnecessary identity expressions when no transformation is
needed for CDF scans.


<!--
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.
-->


## How was this change tested?
<!--
Please make sure to add test cases that check the changes thoroughly
including negative and positive cases if possible.
If it was tested in a way different from regular unit tests, please
clarify how you tested, ideally via a reproducible test documented in
the PR description.
-->

Existing unit tests

Author: DrakeLin

kernel/src/table_changes/physical_to_logical.rs

@@ -72,6 +72,8 @@ pub(crate) fn scan_file_physical_schema(
 
 // Get the transform expression for a CDF scan file
 //
+// Returns None when no transformation is needed (identity transform), otherwise returns Some(expr).
+//
 // Note: parse_partition_values returns null values for missing partition columns,
 // and CDF metadata columns (commit_timestamp, commit_version, change_type) are then
 // added to overwrite any conflicting values. This behavior can be made more strict by changing
@@ -81,7 +83,7 @@ pub(crate) fn get_cdf_transform_expr(
     scan_file: &CdfScanFile,
     state_info: &StateInfo,
     physical_schema: &StructType,
-) -> DeltaResult<ExpressionRef> {
+) -> DeltaResult<Option<ExpressionRef>> {
     let mut partition_values = HashMap::new();
 
     // Get the transform spec from StateInfo (if present)
@@ -92,6 +94,11 @@ pub(crate) fn get_cdf_transform_expr(
         .map(|ts| ts.as_ref())
         .unwrap_or(&empty_spec);
 
+    // Return None for identity transforms to avoid unnecessary expression evaluation
+    if transform_spec.is_empty() {
+        return Ok(None);
+    }
+
     // Handle regular partition values using parse_partition_values
     let parsed_values = parse_partition_values(
         &state_info.logical_schema,
@@ -104,7 +111,9 @@ pub(crate) fn get_cdf_transform_expr(
     let cdf_values = get_cdf_columns(&state_info.logical_schema, scan_file)?;
     partition_values.extend(cdf_values);
 
-    get_transform_expr(transform_spec, partition_values, physical_schema)
+    let expr = get_transform_expr(transform_spec, partition_values, physical_schema)?;
+
+    Ok(Some(expr))
 }
 
 #[cfg(test)]
@@ -192,7 +201,9 @@ mod tests {
         let result = get_cdf_transform_expr(&scan_file, &state_info, &physical_schema);
         assert!(result.is_ok());
 
-        let expr = result.unwrap();
+        let expr_opt = result.unwrap();
+        assert!(expr_opt.is_some(), "Expected Some(expr) but got None");
+        let expr = expr_opt.unwrap();
         let Expression::Transform(transform) = expr.as_ref() else {
             panic!("Expected Transform expression");
         };
@@ -236,7 +247,9 @@ mod tests {
         let result = get_cdf_transform_expr(&scan_file, &state_info, &physical_schema);
         assert!(result.is_ok());
 
-        let expr = result.unwrap();
+        let expr_opt = result.unwrap();
+        assert!(expr_opt.is_some(), "Expected Some(expr) but got None");
+        let expr = expr_opt.unwrap();
         let Expression::Transform(transform) = expr.as_ref() else {
             panic!("Expected Transform expression");
         };
@@ -283,7 +296,9 @@ mod tests {
         let result = get_cdf_transform_expr(&scan_file, &state_info, &physical_schema);
         assert!(result.is_ok());
 
-        let expr = result.unwrap();
+        let expr_opt = result.unwrap();
+        assert!(expr_opt.is_some(), "Expected Some(expr) but got None");
+        let expr = expr_opt.unwrap();
         let Expression::Transform(transform) = expr.as_ref() else {
             panic!("Expected Transform expression");
         };
@@ -340,4 +355,49 @@ mod tests {
         let result = scan_file_physical_schema(&remove_file, &physical_schema);
         assert_eq!(result.fields().len(), 2); // No change
     }
+
+    #[test]
+    fn test_get_cdf_transform_expr_returns_none_for_identity() {
+        // When there's no transform spec and no CDF metadata columns in the schema,
+        // the function should return None (identity transform)
+        let scan_file = CdfScanFile {
+            path: "test/file.parquet".to_string(),
+            partition_values: HashMap::new(),
+            scan_type: CdfScanFileType::Add,
+            commit_version: 100,
+            commit_timestamp: 1000000000000,
+            dv_info: DvInfo::default(),
+            remove_dv: None,
+        };
+
+        // Create a simple schema without CDF metadata columns
+        let logical_schema = Arc::new(StructType::new_unchecked(vec![
+            StructField::nullable("id", DataType::STRING),
+            StructField::nullable("name", DataType::STRING),
+        ]));
+
+        let physical_schema = StructType::new_unchecked(vec![
+            StructField::nullable("id", DataType::STRING),
+            StructField::nullable("name", DataType::STRING),
+        ]);
+
+        // Empty transform spec - no transformations needed
+        let transform_spec = vec![];
+
+        let state_info = StateInfo {
+            logical_schema,
+            physical_schema: physical_schema.clone().into(),
+            physical_predicate: PhysicalPredicate::None,
+            transform_spec: Some(Arc::new(transform_spec)),
+        };
+
+        let result = get_cdf_transform_expr(&scan_file, &state_info, &physical_schema);
+        assert!(result.is_ok());
+
+        let expr_opt = result.unwrap();
+        assert!(
+            expr_opt.is_none(),
+            "Expected None for identity transform but got Some(expr)"
+        );
+    }
 }

kernel/src/table_changes/scan.rs

@@ -239,11 +239,14 @@ fn read_scan_file(
         scan_file_physical_schema(&scan_file, state_info.physical_schema.as_ref());
     let transform_expr = get_cdf_transform_expr(&scan_file, state_info, physical_schema.as_ref())?;
 
-    let phys_to_logical_eval = engine.evaluation_handler().new_expression_evaluator(
-        physical_schema.clone(),
-        transform_expr,
-        state_info.logical_schema.clone().into(),
-    );
+    // Only create an evaluator if transformation is needed
+    let phys_to_logical_eval = transform_expr.map(|expr| {
+        engine.evaluation_handler().new_expression_evaluator(
+            physical_schema.clone(),
+            expr,
+            state_info.logical_schema.clone().into(),
+        )
+    });
     // Determine if the scan file was derived from a deletion vector pair
     let is_dv_resolved_pair = scan_file.remove_dv.is_some();
 
@@ -261,8 +264,13 @@ fn read_scan_file(
 
     let result = read_result_iter.map(move |batch| -> DeltaResult<_> {
         let batch = batch?;
-        // to transform the physical data into the correct logical form
-        let logical = phys_to_logical_eval.evaluate(batch.as_ref());
+        // Transform the physical data into the correct logical form, or pass through unchanged
+        let logical = if let Some(ref eval) = phys_to_logical_eval {
+            eval.evaluate(batch.as_ref())
+        } else {
+            // No transformation needed - pass through the batch as-is
+            Ok(batch)
+        };
         let len = logical.as_ref().map_or(0, |res| res.len());
         // need to split the dv_mask. what's left in dv_mask covers this result, and rest
         // will cover the following results. we `take()` out of `selection_vector` to avoid