make dedup compatible

Author: OussamaSaoudi

kernel/src/log_replay.rs

@@ -14,7 +14,7 @@
 //! deduplication with `FileActionDeduplicator` which tracks unique files across log batches
 //! to minimize memory usage for tables with extensive history.
 use crate::engine_data::GetData;
-use crate::log_replay::deduplicator::Deduplicator;
+use crate::log_replay::deduplicator::{extract_dv_unique_id, Deduplicator};
 use crate::scan::data_skipping::DataSkippingFilter;
 use crate::{DeltaResult, EngineData};
 
@@ -144,7 +144,7 @@ impl<'seen> Deduplicator for FileActionDeduplicator<'seen> {
     ) -> DeltaResult<Option<(FileActionKey, bool)>> {
         // Try to extract an add action by the required path column
         if let Some(path) = getters[self.add_path_index].get_str(i, "add.path")? {
-            let dv_unique_id = self.extract_dv_unique_id(i, getters, self.add_dv_start_index)?;
+            let dv_unique_id = extract_dv_unique_id(i, getters, self.add_dv_start_index)?;
             return Ok(Some((FileActionKey::new(path, dv_unique_id), true)));
         }
 
@@ -155,7 +155,7 @@ impl<'seen> Deduplicator for FileActionDeduplicator<'seen> {
 
         // Try to extract a remove action by the required path column
         if let Some(path) = getters[self.remove_path_index].get_str(i, "remove.path")? {
-            let dv_unique_id = self.extract_dv_unique_id(i, getters, self.remove_dv_start_index)?;
+            let dv_unique_id = extract_dv_unique_id(i, getters, self.remove_dv_start_index)?;
             return Ok(Some((FileActionKey::new(path, dv_unique_id), false)));
         }
 

kernel/src/log_replay/deduplicator.rs

@@ -71,22 +71,25 @@ pub(crate) trait Deduplicator {
 pub(crate) struct CheckpointDeduplicator<'a> {
     seen_file_keys: &'a HashSet<FileActionKey>,
     add_path_index: usize,
+    add_dv_start_index: usize,
 }
 impl CheckpointDeduplicator<'_> {
     #[allow(unused)]
     pub(crate) fn try_new<'a>(
         seen_file_keys: &'a HashSet<FileActionKey>,
         add_path_index: usize,
+        add_dv_start_index: usize,
     ) -> DeltaResult<CheckpointDeduplicator<'a>> {
         Ok(CheckpointDeduplicator {
             seen_file_keys,
             add_path_index,
+            add_dv_start_index,
         })
     }
 }
 
 impl Deduplicator for CheckpointDeduplicator<'_> {
-    type Key = String;
+    type Key = FileActionKey;
 
     fn extract_file_action<'a>(
         &self,
@@ -96,7 +99,8 @@ impl Deduplicator for CheckpointDeduplicator<'_> {
     ) -> DeltaResult<Option<(Self::Key, bool)>> {
         // Try to extract an add action by the required path column
         if let Some(path) = getters[self.add_path_index].get_str(i, "add.path")? {
-            Ok(Some((path.to_string(), true)))
+            let dv_unique_id = extract_dv_unique_id(i, getters, self.add_dv_start_index)?;
+            Ok(Some((FileActionKey::new(path, dv_unique_id), true)))
         } else {
             Ok(None)
         }
@@ -110,3 +114,31 @@ impl Deduplicator for CheckpointDeduplicator<'_> {
         false
     }
 }
+
+/// Extracts the deletion vector unique ID if it exists.
+///
+/// This function retrieves the necessary fields for constructing a deletion vector unique ID
+/// by accessing `getters` at `dv_start_index` and the following two indices. Specifically:
+/// - `dv_start_index` retrieves the storage type (`deletionVector.storageType`).
+/// - `dv_start_index + 1` retrieves the path or inline deletion vector (`deletionVector.pathOrInlineDv`).
+/// - `dv_start_index + 2` retrieves the optional offset (`deletionVector.offset`).
+pub(crate) fn extract_dv_unique_id<'a>(
+    i: usize,
+    getters: &[&'a dyn GetData<'a>],
+    dv_start_index: usize,
+) -> DeltaResult<Option<String>> {
+    match getters[dv_start_index].get_opt(i, "deletionVector.storageType")? {
+        Some(storage_type) => {
+            let path_or_inline =
+                getters[dv_start_index + 1].get(i, "deletionVector.pathOrInlineDv")?;
+            let offset = getters[dv_start_index + 2].get_opt(i, "deletionVector.offset")?;
+
+            Ok(Some(DeletionVectorDescriptor::unique_id_from_parts(
+                storage_type,
+                path_or_inline,
+                offset,
+            )))
+        }
+        None => Ok(None),
+    }
+}

kernel/src/scan/log_replay.rs

@@ -495,7 +495,11 @@ impl ParallelizableLogReplayProcessor for ScanLogReplayProcessor {
         let selection_vector = self.build_selection_vector(actions.as_ref())?;
         assert_eq!(selection_vector.len(), actions.len());
 
-        let deduplicator = CheckpointDeduplicator::try_new(&self.seen_file_keys, ADD_PATH_INDEX)?;
+        let deduplicator = CheckpointDeduplicator::try_new(
+            &self.seen_file_keys,
+            ADD_PATH_INDEX,
+            ADD_DV_START_INDEX,
+        )?;
         let mut visitor = AddRemoveDedupVisitor::new(
             deduplicator,
             selection_vector,