extract file deduplicator

OussamaSaoudi · OussamaSaoudi · commit 493cca18719a · 2025-12-08T13:13:45.000-08:00
diff --git a/kernel/src/action_reconciliation/log_replay.rs b/kernel/src/action_reconciliation/log_replay.rs
@@ -31,6 +31,7 @@
 //!      actions selected
 //!
 use crate::engine_data::{FilteredEngineData, GetData, RowVisitor, TypedGetData as _};
+use crate::log_replay::deduplicator::Deduplicator;
 use crate::log_replay::{
     ActionsBatch, FileActionDeduplicator, FileActionKey, HasSelectionVector, LogReplayProcessor,
 };
diff --git a/kernel/src/log_replay.rs b/kernel/src/log_replay.rs
@@ -15,6 +15,7 @@
 //! to minimize memory usage for tables with extensive history.
 use crate::actions::deletion_vector::DeletionVectorDescriptor;
 use crate::engine_data::{GetData, TypedGetData};
+use crate::log_replay::deduplicator::Deduplicator;
 use crate::scan::data_skipping::DataSkippingFilter;
 use crate::{DeltaResult, EngineData};
 
@@ -24,6 +25,8 @@ use std::collections::HashSet;
 
 use tracing::debug;
 
+pub(crate) mod deduplicator;
+
 /// The subset of file action fields that uniquely identifies it in the log, used for deduplication
 /// of adds and removes during log replay.
 #[derive(Debug, Hash, Eq, PartialEq, serde::Serialize, serde::Deserialize, Clone)]
@@ -87,36 +90,6 @@ impl<'seen> FileActionDeduplicator<'seen> {
         }
     }
 
-    /// Checks if log replay already processed this logical file (in which case the current action
-    /// should be ignored). If not already seen, register it so we can recognize future duplicates.
-    /// Returns `true` if we have seen the file and should ignore it, `false` if we have not seen it
-    /// and should process it.
-    pub(crate) fn check_and_record_seen(&mut self, key: FileActionKey) -> bool {
-        // Note: each (add.path + add.dv_unique_id()) pair has a
-        // unique Add + Remove pair in the log. For example:
-        // https://github.com/delta-io/delta/blob/master/spark/src/test/resources/delta/table-with-dv-large/_delta_log/00000000000000000001.json
-
-        if self.seen_file_keys.contains(&key) {
-            debug!(
-                "Ignoring duplicate ({}, {:?}) in scan, is log {}",
-                key.path, key.dv_unique_id, self.is_log_batch
-            );
-            true
-        } else {
-            debug!(
-                "Including ({}, {:?}) in scan, is log {}",
-                key.path, key.dv_unique_id, self.is_log_batch
-            );
-            if self.is_log_batch {
-                // Remember file actions from this batch so we can ignore duplicates as we process
-                // batches from older commit and/or checkpoint files. We don't track checkpoint
-                // batches because they are already the oldest actions and never replace anything.
-                self.seen_file_keys.insert(key);
-            }
-            false
-        }
-    }
-
     /// Extracts the deletion vector unique ID if it exists.
     ///
     /// This function retrieves the necessary fields for constructing a deletion vector unique ID
@@ -145,6 +118,38 @@ impl<'seen> FileActionDeduplicator<'seen> {
             None => Ok(None),
         }
     }
+}
+impl<'seen> Deduplicator for FileActionDeduplicator<'seen> {
+    type Key = FileActionKey;
+    /// Checks if log replay already processed this logical file (in which case the current action
+    /// should be ignored). If not already seen, register it so we can recognize future duplicates.
+    /// Returns `true` if we have seen the file and should ignore it, `false` if we have not seen it
+    /// and should process it.
+    fn check_and_record_seen(&mut self, key: FileActionKey) -> bool {
+        // Note: each (add.path + add.dv_unique_id()) pair has a
+        // unique Add + Remove pair in the log. For example:
+        // https://github.com/delta-io/delta/blob/master/spark/src/test/resources/delta/table-with-dv-large/_delta_log/00000000000000000001.json
+
+        if self.seen_file_keys.contains(&key) {
+            debug!(
+                "Ignoring duplicate ({}, {:?}) in scan, is log {}",
+                key.path, key.dv_unique_id, self.is_log_batch
+            );
+            true
+        } else {
+            debug!(
+                "Including ({}, {:?}) in scan, is log {}",
+                key.path, key.dv_unique_id, self.is_log_batch
+            );
+            if self.is_log_batch {
+                // Remember file actions from this batch so we can ignore duplicates as we process
+                // batches from older commit and/or checkpoint files. We don't track checkpoint
+                // batches because they are already the oldest actions and never replace anything.
+                self.seen_file_keys.insert(key);
+            }
+            false
+        }
+    }
 
     /// Extracts a file action key and determines if it's an add operation.
     /// This method examines the data at the given index using the provided getters
@@ -159,7 +164,7 @@ impl<'seen> FileActionDeduplicator<'seen> {
     /// - `Ok(Some((key, is_add)))`: When a file action is found, returns the key and whether it's an add operation
     /// - `Ok(None)`: When no file action is found
     /// - `Err(...)`: On any error during extraction
-    pub(crate) fn extract_file_action<'a>(
+    fn extract_file_action<'a>(
         &self,
         i: usize,
         getters: &[&'a dyn GetData<'a>],
@@ -190,7 +195,7 @@ impl<'seen> FileActionDeduplicator<'seen> {
     ///
     /// `true` indicates we are processing a batch from a commit file.
     /// `false` indicates we are processing a batch from a checkpoint.
-    pub(crate) fn is_log_batch(&self) -> bool {
+    fn is_log_batch(&self) -> bool {
         self.is_log_batch
     }
 }
@@ -362,6 +367,7 @@ pub(crate) trait HasSelectionVector {
 mod tests {
     use super::*;
     use crate::engine_data::GetData;
+    use crate::log_replay::deduplicator::Deduplicator;
     use crate::DeltaResult;
     use std::collections::{HashMap, HashSet};
 
diff --git a/kernel/src/log_replay/deduplicator.rs b/kernel/src/log_replay/deduplicator.rs
@@ -0,0 +1,13 @@
+use crate::{engine_data::GetData, DeltaResult};
+
+pub(crate) trait Deduplicator {
+    type Key;
+    fn extract_file_action<'a>(
+        &self,
+        i: usize,
+        getters: &[&'a dyn GetData<'a>],
+        skip_removes: bool,
+    ) -> DeltaResult<Option<(Self::Key, bool)>>;
+    fn check_and_record_seen(&mut self, key: Self::Key) -> bool;
+    fn is_log_batch(&self) -> bool;
+}
diff --git a/kernel/src/scan/log_replay.rs b/kernel/src/scan/log_replay.rs
@@ -12,6 +12,7 @@ use crate::actions::get_log_add_schema;
 use crate::engine_data::{GetData, RowVisitor, TypedGetData as _};
 use crate::expressions::{column_name, ColumnName, Expression, ExpressionRef, PredicateRef};
 use crate::kernel_predicates::{DefaultKernelPredicateEvaluator, KernelPredicateEvaluator as _};
+use crate::log_replay::deduplicator::Deduplicator;
 use crate::log_replay::{ActionsBatch, FileActionDeduplicator, FileActionKey, LogReplayProcessor};
 use crate::scan::Scalar;
 use crate::schema::ToSchema as _;
@@ -46,6 +47,15 @@ pub struct SerializableScanState {
     pub seen_file_keys: HashSet<FileActionKey>,
 }
 
+// These index positions correspond to the order of columns defined in
+// `selected_column_names_and_types()`
+pub(crate) const ADD_PATH_INDEX: usize = 0; // Position of "add.path" in getters
+const ADD_PARTITION_VALUES_INDEX: usize = 1; // Position of "add.partitionValues" in getters
+pub(crate) const ADD_DV_START_INDEX: usize = 2; // Start position of add deletion vector columns
+const BASE_ROW_ID_INDEX: usize = 5; // Position of add.baseRowId in getters
+pub(crate) const REMOVE_PATH_INDEX: usize = 6; // Position of "remove.path" in getters
+pub(crate) const REMOVE_DV_START_INDEX: usize = 7; // Start position of remove deletion vector columns
+
 /// [`ScanLogReplayProcessor`] performs log replay (processes actions) specifically for doing a table scan.
 ///
 /// During a table scan, the processor reads batches of log actions (in reverse chronological order)
@@ -226,40 +236,23 @@ impl ScanLogReplayProcessor {
 /// replay visits actions newest-first, so once we've seen a file action for a given (path, dvId)
 /// pair, we should ignore all subsequent (older) actions for that same (path, dvId) pair. If the
 /// first action for a given file is a remove, then that file does not show up in the result at all.
-struct AddRemoveDedupVisitor<'seen> {
-    deduplicator: FileActionDeduplicator<'seen>,
+struct AddRemoveDedupVisitor<D: Deduplicator> {
+    deduplicator: D,
     selection_vector: Vec<bool>,
     state_info: Arc<StateInfo>,
     partition_filter: Option<PredicateRef>,
     row_transform_exprs: Vec<Option<ExpressionRef>>,
 }
 
-impl AddRemoveDedupVisitor<'_> {
-    // These index positions correspond to the order of columns defined in
-    // `selected_column_names_and_types()`
-    const ADD_PATH_INDEX: usize = 0; // Position of "add.path" in getters
-    const ADD_PARTITION_VALUES_INDEX: usize = 1; // Position of "add.partitionValues" in getters
-    const ADD_DV_START_INDEX: usize = 2; // Start position of add deletion vector columns
-    const BASE_ROW_ID_INDEX: usize = 5; // Position of add.baseRowId in getters
-    const REMOVE_PATH_INDEX: usize = 6; // Position of "remove.path" in getters
-    const REMOVE_DV_START_INDEX: usize = 7; // Start position of remove deletion vector columns
-
+impl<D: Deduplicator> AddRemoveDedupVisitor<D> {
     fn new(
-        seen: &mut HashSet<FileActionKey>,
+        deduplicator: D,
         selection_vector: Vec<bool>,
         state_info: Arc<StateInfo>,
         partition_filter: Option<PredicateRef>,
-        is_log_batch: bool,
-    ) -> AddRemoveDedupVisitor<'_> {
+    ) -> AddRemoveDedupVisitor<D> {
         AddRemoveDedupVisitor {
-            deduplicator: FileActionDeduplicator::new(
-                seen,
-                is_log_batch,
-                Self::ADD_PATH_INDEX,
-                Self::REMOVE_PATH_INDEX,
-                Self::ADD_DV_START_INDEX,
-                Self::REMOVE_DV_START_INDEX,
-            ),
+            deduplicator,
             selection_vector,
             state_info,
             partition_filter,
@@ -312,7 +305,7 @@ impl AddRemoveDedupVisitor<'_> {
         let partition_values = match &self.state_info.transform_spec {
             Some(transform) if is_add => {
                 let partition_values =
-                    getters[Self::ADD_PARTITION_VALUES_INDEX].get(i, "add.partitionValues")?;
+                    getters[ADD_PARTITION_VALUES_INDEX].get(i, "add.partitionValues")?;
                 let partition_values = parse_partition_values(
                     &self.state_info.logical_schema,
                     transform,
@@ -331,8 +324,7 @@ impl AddRemoveDedupVisitor<'_> {
         if self.deduplicator.check_and_record_seen(file_key) || !is_add {
             return Ok(false);
         }
-        let base_row_id: Option<i64> =
-            getters[Self::BASE_ROW_ID_INDEX].get_opt(i, "add.baseRowId")?;
+        let base_row_id: Option<i64> = getters[BASE_ROW_ID_INDEX].get_opt(i, "add.baseRowId")?;
         let transform = self
             .state_info
             .transform_spec
@@ -355,7 +347,7 @@ impl AddRemoveDedupVisitor<'_> {
     }
 }
 
-impl RowVisitor for AddRemoveDedupVisitor<'_> {
+impl<D: Deduplicator> RowVisitor for AddRemoveDedupVisitor<D> {
     fn selected_column_names_and_types(&self) -> (&'static [ColumnName], &'static [DataType]) {
         // NOTE: The visitor assumes a schema with adds first and removes optionally afterward.
         static NAMES_AND_TYPES: LazyLock<ColumnNamesAndTypes> = LazyLock::new(|| {
@@ -501,12 +493,19 @@ impl LogReplayProcessor for ScanLogReplayProcessor {
         let selection_vector = self.build_selection_vector(actions.as_ref())?;
         assert_eq!(selection_vector.len(), actions.len());
 
-        let mut visitor = AddRemoveDedupVisitor::new(
+        let deduplicator = FileActionDeduplicator::new(
             &mut self.seen_file_keys,
+            is_log_batch,
+            ADD_PATH_INDEX,
+            REMOVE_PATH_INDEX,
+            ADD_DV_START_INDEX,
+            REMOVE_DV_START_INDEX,
+        );
+        let mut visitor = AddRemoveDedupVisitor::new(
+            deduplicator,
             selection_vector,
             self.state_info.clone(),
             self.partition_filter.clone(),
-            is_log_batch,
         );
         visitor.visit_rows_of(actions.as_ref())?;
 
