TableChangesScan::execute and end to end testing for CDF

kernel/src/expressions/scalars.rs

@@ -151,6 +151,16 @@ impl Scalar {
     pub fn is_null(&self) -> bool {
         matches!(self, Self::Null(_))
     }
+
+    /// 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::TimestampNtz(timestamp.timestamp_micros()))
+    }
 }
 
 impl Display for Scalar {

kernel/src/log_segment.rs

@@ -145,7 +145,6 @@ impl LogSegment {
     /// `start_version` and `end_version`: Its LogSegment is made of zero checkpoints and all commits
     /// between versions `start_version` (inclusive) and `end_version` (inclusive). If no `end_version`
     /// is specified it will be the most recent version by default.
-    #[allow(unused)]
     #[cfg_attr(feature = "developer-visibility", visibility::make(pub))]
     pub(crate) fn for_table_changes(
         fs_client: &dyn FileSystemClient,

kernel/src/scan/mod.rs

@@ -125,7 +125,7 @@ pub struct ScanResult {
     pub raw_data: DeltaResult<Box<dyn EngineData>>,
     /// Raw row mask.
     // TODO(nick) this should be allocated by the engine
-    raw_mask: Option<Vec<bool>>,
+    pub(crate) raw_mask: Option<Vec<bool>>,
 }
 
 impl ScanResult {
@@ -160,7 +160,7 @@ impl ScanResult {
 /// store the name of the column, as that's all that's needed during the actual query. For
 /// `Partition` we store an index into the logical schema for this query since later we need the
 /// data type as well to materialize the partition column.
-#[derive(PartialEq, Debug)]
+#[derive(Clone, PartialEq, Debug)]
 pub enum ColumnType {
     // A column, selected from the data, as is
     Selected(String),
@@ -384,7 +384,10 @@ pub fn scan_row_schema() -> Schema {
     log_replay::SCAN_ROW_SCHEMA.as_ref().clone()
 }
 
-fn parse_partition_value(raw: Option<&String>, data_type: &DataType) -> DeltaResult<Scalar> {
+pub(crate) fn parse_partition_value(
+    raw: Option<&String>,
+    data_type: &DataType,
+) -> DeltaResult<Scalar> {
     match (raw, data_type.as_primitive_opt()) {
         (Some(v), Some(primitive)) => primitive.parse_scalar(v),
         (Some(_), None) => Err(Error::generic(format!(

kernel/src/table_changes/log_replay.rs

@@ -28,7 +28,6 @@ use itertools::Itertools;
 mod tests;
 
 /// Scan data for a Change Data Feed query. This holds metadata that is needed to read data rows.
-#[allow(unused)]
 pub(crate) struct TableChangesScanData {
     /// Engine data with the schema defined in [`scan_row_schema`]
     ///
@@ -127,7 +126,6 @@ struct LogReplayScanner {
     //
     // Note: This will be used once an expression is introduced to transform the engine data in
     // [`TableChangesScanData`]
-    #[allow(unused)]
     timestamp: i64,
 }
 

kernel/src/table_changes/mod.rs

@@ -16,15 +16,21 @@ use crate::utils::require;
 use crate::{DeltaResult, Engine, Error, Version};
 
 mod log_replay;
+mod physical_to_logical;
 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),
     ]
 });
 
@@ -172,7 +178,6 @@ impl TableChanges {
         &self.table_root
     }
     /// The partition columns that will be read.
-    #[allow(unused)]
     pub(crate) fn partition_columns(&self) -> &Vec<String> {
         &self.end_snapshot.metadata().partition_columns
     }

kernel/src/table_changes/physical_to_logical.rs

@@ -0,0 +1,139 @@
+use std::collections::HashMap;
+use std::iter;
+
+use itertools::Itertools;
+
+use crate::expressions::Scalar;
+use crate::scan::{parse_partition_value, ColumnType};
+use crate::schema::{ColumnName, DataType, SchemaRef, StructField, StructType};
+use crate::{DeltaResult, Error, Expression};
+
+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.
+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 => Expression::column([CHANGE_TYPE_COL_NAME]),
+        CdfScanFileType::Add => ADD_CHANGE_TYPE.into(),
+        CdfScanFileType::Remove => REMOVE_CHANGE_TYPE.into(),
+    };
+    let expressions = [
+        (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 `logical_schema`
+pub(crate) fn physical_to_logical_expr(
+    scan_file: &CdfScanFile,
+    logical_schema: &StructType,
+    all_fields: &[ColumnType],
+) -> DeltaResult<Expression> {
+    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 = 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",
+                    ));
+                };
+                let name = field.physical_name();
+                let value_expression =
+                    parse_partition_value(scan_file.partition_values.get(name), field.data_type())?;
+                Ok(value_expression.into())
+            }
+            ColumnType::Selected(field_name) => {
+                // Remove to take ownership
+                let generated_column = cdf_columns.remove(field_name.as_str());
+                Ok(generated_column.unwrap_or_else(|| ColumnName::new([field_name]).into()))
+            }
+        })
+        .try_collect()?;
+    Ok(Expression::Struct(all_fields))
+}
+
+/// Gets the physical schema that will be used to read data in the `scan_file` path.
+pub(crate) 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_COL_NAME, DataType::STRING, false);
+        let fields = read_schema.fields().cloned().chain(iter::once(change_type));
+        StructType::new(fields).into()
+    } else {
+        read_schema.clone().into()
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use std::collections::HashMap;
+
+    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,
+    };
+
+    #[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)
+        };
+
+        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/resolve_dvs.rs

@@ -7,16 +7,15 @@ use crate::{DeltaResult, Engine, Error};
 
 /// A [`CdfScanFile`] with its associated `selection_vector`. The `scan_type` is resolved to
 /// match the `_change_type` that its rows will have in the change data feed.
-#[allow(unused)]
-struct ResolvedCdfScanFile {
+pub(crate) struct ResolvedCdfScanFile {
     /// The scan file that holds the path the data file to be read. The `scan_type` field is
     /// resolved to the `_change_type` of the rows for this data file.
-    scan_file: CdfScanFile,
+    pub(crate) scan_file: CdfScanFile,
     /// Optional vector of bools. If `selection_vector[i] = true`, then that row must be included
     /// in the CDF output. Otherwise the row must be filtered out. The vector may be shorter than
     /// the data file. In this case, all the remaining rows are *not* selected. If `selection_vector`
     /// is `None`, then all rows are selected.
-    selection_vector: Option<Vec<bool>>,
+    pub(crate) selection_vector: Option<Vec<bool>>,
 }
 
 /// Resolves the deletion vectors for a [`CdfScanFile`]. This function handles two
@@ -33,8 +32,7 @@ struct ResolvedCdfScanFile {
 /// 2. The second case handles all other add, remove, and cdc [`CdfScanFile`]s. These will simply
 ///    read the deletion vector (if present), and each is converted into a [`ResolvedCdfScanFile`].
 ///    No changes are made to the `scan_type`.
-#[allow(unused)]
-fn resolve_scan_file_dv(
+pub(crate) fn resolve_scan_file_dv(
     engine: &dyn Engine,
     table_root: &Url,
     scan_file: CdfScanFile,