Resolve deletion vectors to find inserted and removed rows for CDF

kernel/src/actions/deletion_vector.rs

@@ -215,7 +215,19 @@ fn slice_to_u32(buf: &[u8], endian: Endian) -> DeltaResult<u32> {
 
 /// helper function to convert a treemap into a boolean vector where, for index i, if the bit is
 /// set, the vector will be false, and otherwise at index i the vector will be true
-pub(crate) fn treemap_to_bools(treemap: RoaringTreemap) -> Vec<bool> {
+pub(crate) fn deletion_treemap_to_bools(treemap: RoaringTreemap) -> Vec<bool> {
+    treemap_to_bools_with(treemap, false)
+}
+
+/// helper function to convert a treemap into a boolean vector where, for index i, if the bit is
+/// set, the vector will be true, and otherwise at index i the vector will be false
+pub(crate) fn selection_treemap_to_bools(treemap: RoaringTreemap) -> Vec<bool> {
+    treemap_to_bools_with(treemap, true)
+}
+
+/// helper function to generate vectors of bools from treemap. If `set_bit` is `true`, this is
+/// [`selection_treemap_to_bools`]. If `set_bit` is false, this is [`deletion_treemap_to_bools`]
+fn treemap_to_bools_with(treemap: RoaringTreemap, set_bit: bool) -> Vec<bool> {
     fn combine(high_bits: u32, low_bits: u32) -> usize {
         ((u64::from(high_bits) << 32) | u64::from(low_bits)) as usize
     }
@@ -224,12 +236,12 @@ pub(crate) fn treemap_to_bools(treemap: RoaringTreemap) -> Vec<bool> {
         Some(max) => {
             // there are values in the map
             //TODO(nick) panic if max is > MAX_USIZE
-            let mut result = vec![true; max as usize + 1];
+            let mut result = vec![!set_bit; max as usize + 1];
             let bitmaps = treemap.bitmaps();
             for (index, bitmap) in bitmaps {
                 for bit in bitmap.iter() {
                     let vec_index = combine(index, bit);
-                    result[vec_index] = false;
+                    result[vec_index] = set_bit;
                 }
             }
             result
@@ -380,7 +392,7 @@ mod tests {
     }
 
     // this test is ignored by default as it's expensive to allocate such big vecs full of `true`. you can run it via:
-    // cargo test actions::action_definitions::tests::test_dv_to_bools
+    // cargo test actions::deletion_vector::tests::test_dv_to_bools -- --ignored
     #[test]
     #[ignore]
     fn test_dv_to_bools() {
@@ -391,7 +403,7 @@ mod tests {
         rb.insert(30854);
         rb.insert(4294967297);
         rb.insert(4294967300);
-        let bools = super::treemap_to_bools(rb);
+        let bools = super::deletion_treemap_to_bools(rb);
         let mut expected = vec![true; 4294967301];
         expected[0] = false;
         expected[2] = false;
@@ -402,6 +414,26 @@ mod tests {
         assert_eq!(bools, expected);
     }
 
+    #[test]
+    fn test_sv_to_bools() {
+        let mut rb = RoaringTreemap::new();
+        rb.insert(0);
+        rb.insert(2);
+        rb.insert(7);
+        rb.insert(30854);
+        rb.insert(4294967297);
+        rb.insert(4294967300);
+        let bools = super::selection_treemap_to_bools(rb);
+        let mut expected = vec![false; 4294967301];
+        expected[0] = true;
+        expected[2] = true;
+        expected[7] = true;
+        expected[30854] = true;
+        expected[4294967297] = true;
+        expected[4294967300] = true;
+        assert_eq!(bools, expected);
+    }
+
     #[test]
     fn test_dv_row_indexes() {
         let example = dv_inline();

kernel/src/scan/mod.rs

@@ -7,7 +7,9 @@ use itertools::Itertools;
 use tracing::debug;
 use url::Url;
 
-use crate::actions::deletion_vector::{split_vector, treemap_to_bools, DeletionVectorDescriptor};
+use crate::actions::deletion_vector::{
+    deletion_treemap_to_bools, split_vector, DeletionVectorDescriptor,
+};
 use crate::actions::{get_log_add_schema, get_log_schema, ADD_NAME, REMOVE_NAME};
 use crate::expressions::{ColumnName, Expression, ExpressionRef, Scalar};
 use crate::scan::state::{DvInfo, Stats};
@@ -438,7 +440,7 @@ pub fn selection_vector(
 ) -> DeltaResult<Vec<bool>> {
     let fs_client = engine.get_file_system_client();
     let dv_treemap = descriptor.read(fs_client, table_root)?;
-    Ok(treemap_to_bools(dv_treemap))
+    Ok(deletion_treemap_to_bools(dv_treemap))
 }
 
 /// Transform the raw data read from parquet into the correct logical form, based on the provided

kernel/src/scan/state.rs

@@ -3,17 +3,16 @@
 use std::collections::HashMap;
 use std::sync::LazyLock;
 
+use crate::actions::deletion_vector::deletion_treemap_to_bools;
 use crate::utils::require;
 use crate::{
-    actions::{
-        deletion_vector::{treemap_to_bools, DeletionVectorDescriptor},
-        visitors::visit_deletion_vector_at,
-    },
+    actions::{deletion_vector::DeletionVectorDescriptor, visitors::visit_deletion_vector_at},
     engine_data::{GetData, RowVisitor, TypedGetData as _},
     schema::{ColumnName, ColumnNamesAndTypes, DataType, SchemaRef},
     table_features::ColumnMappingMode,
     DeltaResult, Engine, EngineData, Error,
 };
+use roaring::RoaringTreemap;
 use serde::{Deserialize, Serialize};
 use tracing::warn;
 
@@ -30,11 +29,18 @@ pub struct GlobalScanState {
 }
 
 /// this struct can be used by an engine to materialize a selection vector
-#[derive(Debug, Clone, PartialEq, Eq)]
+#[derive(Default, Debug, Clone, PartialEq, Eq)]
 pub struct DvInfo {
     pub(crate) deletion_vector: Option<DeletionVectorDescriptor>,
 }
 
+impl From<DeletionVectorDescriptor> for DvInfo {
+    fn from(deletion_vector: DeletionVectorDescriptor) -> Self {
+        let deletion_vector = Some(deletion_vector);
+        DvInfo { deletion_vector }
+    }
+}
+
 /// Give engines an easy way to consume stats
 #[derive(Debug, Clone, PartialEq, Eq, Deserialize)]
 #[serde(rename_all = "camelCase")]
@@ -53,20 +59,27 @@ impl DvInfo {
         self.deletion_vector.is_some()
     }
 
-    pub fn get_selection_vector(
+    pub(crate) fn get_treemap(
         &self,
         engine: &dyn Engine,
         table_root: &url::Url,
-    ) -> DeltaResult<Option<Vec<bool>>> {
-        let dv_treemap = self
-            .deletion_vector
+    ) -> DeltaResult<Option<RoaringTreemap>> {
+        self.deletion_vector
             .as_ref()
             .map(|dv_descriptor| {
                 let fs_client = engine.get_file_system_client();
                 dv_descriptor.read(fs_client, table_root)
             })
-            .transpose()?;
-        Ok(dv_treemap.map(treemap_to_bools))
+            .transpose()
+    }
+
+    pub fn get_selection_vector(
+        &self,
+        engine: &dyn Engine,
+        table_root: &url::Url,
+    ) -> DeltaResult<Option<Vec<bool>>> {
+        let dv_treemap = self.get_treemap(engine, table_root)?;
+        Ok(dv_treemap.map(deletion_treemap_to_bools))
     }
 
     /// Returns a vector of row indexes that should be *removed* from the result set
@@ -112,11 +125,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

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