ScanFile expression and visitor for CDF

kernel/src/scan/state.rs

@@ -112,11 +112,11 @@ pub type ScanCallback<T> = fn(
 /// ## Example
 /// ```ignore
 /// let mut context = [my context];
-/// for res in scan_data { // scan data from scan.get_scan_data()
+/// for res in scan_data { // scan data from scan.scan_data()
 ///     let (data, vector) = res?;
 ///     context = delta_kernel::scan::state::visit_scan_files(
 ///        data.as_ref(),
-///        vector,
+///        selection_vector,
 ///        context,
 ///        my_callback,
 ///     )?;

kernel/src/table_changes/log_replay.rs

@@ -11,16 +11,17 @@ use crate::actions::{
     PROTOCOL_NAME, REMOVE_NAME,
 };
 use crate::engine_data::{GetData, TypedGetData};
-use crate::expressions::{column_expr, column_name, ColumnName, Expression};
+use crate::expressions::{column_name, ColumnName};
 use crate::path::ParsedLogPath;
 use crate::scan::data_skipping::DataSkippingFilter;
-use crate::scan::scan_row_schema;
 use crate::scan::state::DvInfo;
 use crate::schema::{ArrayType, ColumnNamesAndTypes, DataType, MapType, SchemaRef, StructType};
+use crate::table_changes::scan_file::{cdf_scan_row_expression, cdf_scan_row_schema};
 use crate::table_changes::{check_cdf_table_properties, ensure_cdf_read_supported};
 use crate::table_properties::TableProperties;
 use crate::utils::require;
 use crate::{DeltaResult, Engine, EngineData, Error, ExpressionRef, RowVisitor};
+
 use itertools::Itertools;
 
 #[cfg(test)]
@@ -36,7 +37,7 @@ pub(crate) struct TableChangesScanData {
     pub(crate) scan_data: Box<dyn EngineData>,
     /// The selection vector used to filter the `scan_data`.
     pub(crate) selection_vector: Vec<bool>,
-    /// An map from a remove action's path to its deletion vector
+    /// A map from a remove action's path to its deletion vector
     pub(crate) remove_dvs: Arc<HashMap<String, DvInfo>>,
 }
 
@@ -65,21 +66,6 @@ pub(crate) fn table_changes_action_iter(
     Ok(result)
 }
 
-// Gets the expression for generating the engine data in [`TableChangesScanData`].
-//
-// TODO: This expression is temporary. In the future it will also select `cdc` and `remove` actions
-// fields.
-fn add_transform_expr() -> Expression {
-    Expression::Struct(vec![
-        column_expr!("add.path"),
-        column_expr!("add.size"),
-        column_expr!("add.modificationTime"),
-        column_expr!("add.stats"),
-        column_expr!("add.deletionVector"),
-        Expression::Struct(vec![column_expr!("add.partitionValues")]),
-    ])
-}
-
 /// Processes a single commit file from the log to generate an iterator of [`TableChangesScanData`].
 /// The scanner operates in two phases that _must_ be performed in the following order:
 /// 1. Prepare phase [`LogReplayScanner::try_new`]: This iterates over every action in the commit.
@@ -237,7 +223,7 @@ impl LogReplayScanner {
             remove_dvs,
             commit_file,
             // TODO: Add the timestamp as a column with an expression
-            timestamp: _,
+            timestamp,
         } = self;
         let remove_dvs = Arc::new(remove_dvs);
 
@@ -247,10 +233,14 @@ impl LogReplayScanner {
             schema,
             None,
         )?;
+        let commit_version = commit_file
+            .version
+            .try_into()
+            .map_err(|_| Error::generic("Failed to convert commit version to i64"))?;
         let evaluator = engine.get_expression_handler().get_evaluator(
             get_log_add_schema().clone(),
-            add_transform_expr(),
-            scan_row_schema().into(),
+            cdf_scan_row_expression(timestamp, commit_version),
+            cdf_scan_row_schema().into(),
         );
 
         let result = action_iter.map(move |actions| -> DeltaResult<_> {

kernel/src/table_changes/mod.rs

@@ -17,6 +17,7 @@ use crate::{DeltaResult, Engine, Error, Version};
 
 mod log_replay;
 pub mod scan;
+mod scan_file;
 
 static CDF_FIELDS: LazyLock<[StructField; 3]> = LazyLock::new(|| {
     [