Helper methods for CDF Physical to Logical Transformation

kernel/src/expressions/scalars.rs

@@ -152,14 +152,14 @@ impl Scalar {
         matches!(self, Self::Null(_))
     }
 
-    /// Constructs a Scalar timestamp from an `i64` millisecond since unix epoch
-    pub fn timestamp_from_millis(millis: i64) -> DeltaResult<Self> {
+    /// Constructs a Scalar timestamp with no timezone from an `i64` millisecond since unix epoch
+    pub(crate) fn timestamp_ntz_from_millis(millis: i64) -> DeltaResult<Self> {
         let Some(timestamp) = DateTime::from_timestamp_millis(millis) else {
             return Err(Error::generic(format!(
                 "Failed to create millisecond timestamp from {millis}"
             )));
         };
-        Ok(Self::Timestamp(timestamp.timestamp_micros()))
+        Ok(Self::TimestampNtz(timestamp.timestamp_micros()))
     }
 }
 
@@ -472,6 +472,7 @@ impl PrimitiveType {
 
 #[cfg(test)]
 mod tests {
+    use core::panic;
     use std::f32::consts::PI;
 
     use crate::expressions::{column_expr, BinaryOperator};

kernel/src/table_changes/mod.rs

@@ -21,11 +21,16 @@ mod resolve_dvs;
 pub mod scan;
 mod scan_file;
 
+static CHANGE_TYPE_COL_NAME: &str = "_change_type";
+static COMMIT_VERSION_COL_NAME: &str = "_commit_version";
+static COMMIT_TIMESTAMP_COL_NAME: &str = "_commit_timestamp";
+static ADD_CHANGE_TYPE: &str = "insert";
+static REMOVE_CHANGE_TYPE: &str = "delete";
 static CDF_FIELDS: LazyLock<[StructField; 3]> = LazyLock::new(|| {
     [
-        StructField::new("_change_type", DataType::STRING, false),
-        StructField::new("_commit_version", DataType::LONG, false),
-        StructField::new("_commit_timestamp", DataType::TIMESTAMP, false),
+        StructField::new(CHANGE_TYPE_COL_NAME, DataType::STRING, false),
+        StructField::new(COMMIT_VERSION_COL_NAME, DataType::LONG, false),
+        StructField::new(COMMIT_TIMESTAMP_COL_NAME, DataType::TIMESTAMP, false),
     ]
 });
 

kernel/src/table_changes/physical_to_logical.rs

@@ -2,51 +2,50 @@ use std::collections::HashMap;
 use std::iter;
 
 use itertools::Itertools;
-use url::Url;
 
-use crate::actions::deletion_vector::split_vector;
-use crate::expressions::{column_expr, Scalar};
-use crate::scan::state::GlobalScanState;
-use crate::scan::{parse_partition_value, ColumnType, ScanResult};
+use crate::expressions::Scalar;
+use crate::scan::{parse_partition_value, ColumnType};
 use crate::schema::{ColumnName, DataType, SchemaRef, StructField, StructType};
-use crate::{DeltaResult, Engine, Error, Expression, ExpressionRef, FileMeta};
+use crate::{DeltaResult, Error, Expression};
 
-use super::resolve_dvs::ResolvedCdfScanFile;
 use super::scan_file::{CdfScanFile, CdfScanFileType};
+use super::{
+    ADD_CHANGE_TYPE, CHANGE_TYPE_COL_NAME, COMMIT_TIMESTAMP_COL_NAME, COMMIT_VERSION_COL_NAME,
+    REMOVE_CHANGE_TYPE,
+};
 
 /// Returns a map from change data feed column name to an expression that generates the row data.
 #[allow(unused)]
-fn get_generated_columns(scan_file: &CdfScanFile) -> DeltaResult<HashMap<&str, Expression>> {
-    let timestamp = Scalar::timestamp_from_millis(scan_file.commit_timestamp)?;
+fn get_cdf_columns(scan_file: &CdfScanFile) -> DeltaResult<HashMap<&str, Expression>> {
+    let timestamp = Scalar::timestamp_ntz_from_millis(scan_file.commit_timestamp)?;
     let version = scan_file.commit_version;
     let change_type: Expression = match scan_file.scan_type {
-        CdfScanFileType::Cdc => column_expr!("_change_type"),
-        CdfScanFileType::Add => "insert".into(),
-
-        CdfScanFileType::Remove => "delete".into(),
+        CdfScanFileType::Cdc => Expression::column([CHANGE_TYPE_COL_NAME]),
+        CdfScanFileType::Add => ADD_CHANGE_TYPE.into(),
+        CdfScanFileType::Remove => REMOVE_CHANGE_TYPE.into(),
     };
     let expressions = [
-        ("_change_type", change_type),
-        ("_commit_version", Expression::literal(version)),
-        ("_commit_timestamp", timestamp.into()),
+        (CHANGE_TYPE_COL_NAME, change_type),
+        (COMMIT_VERSION_COL_NAME, Expression::literal(version)),
+        (COMMIT_TIMESTAMP_COL_NAME, timestamp.into()),
     ];
     Ok(expressions.into_iter().collect())
 }
 
 /// Generates the expression used to convert physical data from the `scan_file` path into logical
-/// data matching the `global_state.logical_schema`
+/// data matching the `logical_schema`
 #[allow(unused)]
-fn get_expression(
+fn physical_to_logical_expr(
     scan_file: &CdfScanFile,
-    global_state: &GlobalScanState,
+    logical_schema: &StructType,
     all_fields: &[ColumnType],
 ) -> DeltaResult<Expression> {
-    let mut generated_columns = get_generated_columns(scan_file)?;
+    let mut cdf_columns = get_cdf_columns(scan_file)?;
     let all_fields = all_fields
         .iter()
         .map(|field| match field {
             ColumnType::Partition(field_idx) => {
-                let field = global_state.logical_schema.fields.get_index(*field_idx);
+                let field = logical_schema.fields.get_index(*field_idx);
                 let Some((_, field)) = field else {
                     return Err(Error::generic(
                         "logical schema did not contain expected field, can't transform data",
@@ -59,7 +58,7 @@ fn get_expression(
             }
             ColumnType::Selected(field_name) => {
                 // Remove to take ownership
-                let generated_column = generated_columns.remove(field_name.as_str());
+                let generated_column = cdf_columns.remove(field_name.as_str());
                 Ok(generated_column.unwrap_or_else(|| ColumnName::new([field_name]).into()))
             }
         })
@@ -69,72 +68,72 @@ fn get_expression(
 
 /// Gets the physical schema that will be used to read data in the `scan_file` path.
 #[allow(unused)]
-fn get_read_schema(scan_file: &CdfScanFile, global_scan_state: &GlobalScanState) -> SchemaRef {
+fn scan_file_read_schema(scan_file: &CdfScanFile, read_schema: &StructType) -> SchemaRef {
     if scan_file.scan_type == CdfScanFileType::Cdc {
-        let change_type = StructField::new("_change_type", DataType::STRING, false);
-        let fields = global_scan_state
-            .read_schema
-            .fields()
-            .cloned()
-            .chain(iter::once(change_type));
+        let change_type = StructField::new(CHANGE_TYPE_COL_NAME, DataType::STRING, false);
+        let fields = read_schema.fields().cloned().chain(iter::once(change_type));
         StructType::new(fields).into()
     } else {
-        global_scan_state.read_schema.clone()
+        read_schema.clone().into()
     }
 }
 
-/// Reads the data at the `resolved_scan_file` and transforms the data from physical to logical.
-/// The result is a fallible iterator of [`ScanResult`] containing the logical data.
-#[allow(unused)]
-pub(crate) fn read_scan_data(
-    engine: &dyn Engine,
-    resolved_scan_file: ResolvedCdfScanFile,
-    global_state: &GlobalScanState,
-    all_fields: &[ColumnType],
-    predicate: Option<ExpressionRef>,
-) -> DeltaResult<impl Iterator<Item = DeltaResult<ScanResult>>> {
-    let ResolvedCdfScanFile {
-        scan_file,
-        mut selection_vector,
-    } = resolved_scan_file;
-
-    let expression = get_expression(&scan_file, global_state, all_fields)?;
-    let schema = get_read_schema(&scan_file, global_state);
-    let evaluator = engine.get_expression_handler().get_evaluator(
-        schema.clone(),
-        expression,
-        global_state.logical_schema.clone().into(),
-    );
+#[cfg(test)]
+mod tests {
+    use std::collections::HashMap;
 
-    let table_root = Url::parse(&global_state.table_root)?;
-    let location = table_root.join(&scan_file.path)?;
-    let file = FileMeta {
-        last_modified: 0,
-        size: 0,
-        location,
+    use crate::expressions::{column_expr, Expression, Scalar};
+    use crate::scan::ColumnType;
+    use crate::schema::{DataType, StructField, StructType};
+    use crate::table_changes::physical_to_logical::physical_to_logical_expr;
+    use crate::table_changes::scan_file::{CdfScanFile, CdfScanFileType};
+    use crate::table_changes::{
+        ADD_CHANGE_TYPE, CHANGE_TYPE_COL_NAME, COMMIT_TIMESTAMP_COL_NAME, COMMIT_VERSION_COL_NAME,
+        REMOVE_CHANGE_TYPE,
     };
-    let read_result_iter =
-        engine
-            .get_parquet_handler()
-            .read_parquet_files(&[file], schema, predicate)?;
 
-    let result = read_result_iter.map(move |batch| -> DeltaResult<_> {
-        let batch = batch?;
-        // to transform the physical data into the correct logical form
-        let logical = evaluator.evaluate(batch.as_ref());
-        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
-        // trying to return a captured variable. We're going to reassign `selection_vector`
-        // to `rest` in a moment anyway
-        let mut sv = selection_vector.take();
-        let rest = split_vector(sv.as_mut(), len, None);
-        let result = ScanResult {
-            raw_data: logical,
-            raw_mask: sv,
+    #[test]
+    fn verify_physical_to_logical_expression() {
+        let test = |scan_type, expected_expr| {
+            let scan_file = CdfScanFile {
+                scan_type,
+                path: "fake_path".to_string(),
+                dv_info: Default::default(),
+                remove_dv: None,
+                partition_values: HashMap::from([("age".to_string(), "20".to_string())]),
+                commit_version: 42,
+                commit_timestamp: 1234,
+            };
+            let logical_schema = StructType::new([
+                StructField::new("id", DataType::STRING, true),
+                StructField::new("age", DataType::LONG, false),
+                StructField::new(CHANGE_TYPE_COL_NAME, DataType::STRING, false),
+                StructField::new(COMMIT_VERSION_COL_NAME, DataType::LONG, false),
+                StructField::new(COMMIT_TIMESTAMP_COL_NAME, DataType::TIMESTAMP, false),
+            ]);
+            let all_fields = vec![
+                ColumnType::Selected("id".to_string()),
+                ColumnType::Partition(1),
+                ColumnType::Selected(CHANGE_TYPE_COL_NAME.to_string()),
+                ColumnType::Selected(COMMIT_VERSION_COL_NAME.to_string()),
+                ColumnType::Selected(COMMIT_TIMESTAMP_COL_NAME.to_string()),
+            ];
+            let phys_to_logical_expr =
+                physical_to_logical_expr(&scan_file, &logical_schema, &all_fields).unwrap();
+            let expected_expr = Expression::struct_from([
+                column_expr!("id"),
+                Scalar::Long(20).into(),
+                expected_expr,
+                Expression::literal(42i64),
+                Scalar::TimestampNtz(1234000).into(), // Microsecond is 1000x millisecond
+            ]);
+
+            assert_eq!(phys_to_logical_expr, expected_expr)
         };
-        selection_vector = rest;
-        Ok(result)
-    });
-    Ok(result)
+
+        let cdc_change_type = Expression::column([CHANGE_TYPE_COL_NAME]);
+        test(CdfScanFileType::Add, ADD_CHANGE_TYPE.into());
+        test(CdfScanFileType::Remove, REMOVE_CHANGE_TYPE.into());
+        test(CdfScanFileType::Cdc, cdc_change_type);
+    }
 }

kernel/src/table_changes/scan.rs

@@ -4,14 +4,11 @@ use itertools::Itertools;
 use tracing::debug;
 
 use crate::scan::state::GlobalScanState;
-use crate::scan::{ColumnType, ScanResult};
+use crate::scan::ColumnType;
 use crate::schema::{SchemaRef, StructType};
 use crate::{DeltaResult, Engine, ExpressionRef};
 
 use super::log_replay::{table_changes_action_iter, TableChangesScanData};
-use super::physical_to_logical::read_scan_data;
-use super::resolve_dvs::resolve_scan_file_dv;
-use super::scan_file::scan_data_to_scan_file;
 use super::{TableChanges, CDF_FIELDS};
 
 /// The result of building a [`TableChanges`] scan over a table. This can be used to get a change
@@ -216,6 +213,9 @@ mod tests {
     use crate::expressions::{column_expr, Scalar};
     use crate::scan::ColumnType;
     use crate::schema::{DataType, StructField, StructType};
+    use crate::table_changes::{
+        CHANGE_TYPE_COL_NAME, COMMIT_TIMESTAMP_COL_NAME, COMMIT_VERSION_COL_NAME,
+    };
     use crate::{Expression, Table};
 
     #[test]
@@ -234,9 +234,9 @@ mod tests {
             vec![
                 ColumnType::Selected("part".to_string()),
                 ColumnType::Selected("id".to_string()),
-                ColumnType::Selected("_change_type".to_string()),
-                ColumnType::Selected("_commit_version".to_string()),
-                ColumnType::Selected("_commit_timestamp".to_string()),
+                ColumnType::Selected(CHANGE_TYPE_COL_NAME.to_string()),
+                ColumnType::Selected(COMMIT_VERSION_COL_NAME.to_string()),
+                ColumnType::Selected(COMMIT_TIMESTAMP_COL_NAME.to_string()),
             ]
         );
         assert_eq!(scan.predicate, None);
@@ -254,7 +254,7 @@ mod tests {
 
         let schema = table_changes
             .schema()
-            .project(&["id", "_commit_version"])
+            .project(&["id", COMMIT_VERSION_COL_NAME])
             .unwrap();
         let predicate = Arc::new(Expression::gt(column_expr!("id"), Scalar::from(10)));
         let scan = table_changes
@@ -267,14 +267,14 @@ mod tests {
             scan.all_fields,
             vec![
                 ColumnType::Selected("id".to_string()),
-                ColumnType::Selected("_commit_version".to_string()),
+                ColumnType::Selected(COMMIT_VERSION_COL_NAME.to_string()),
             ]
         );
         assert_eq!(
             scan.logical_schema,
             StructType::new([
                 StructField::new("id", DataType::INTEGER, true),
-                StructField::new("_commit_version", DataType::LONG, false),
+                StructField::new(COMMIT_VERSION_COL_NAME, DataType::LONG, false),
             ])
             .into()
         );