feat: Introduce Deduplicator trait to unify mutable and immutable deduplication (delta-io#1537)

## What changes are proposed in this pull request?
<!--
Please clarify what changes you are proposing and why the changes are
needed.
The purpose of this section is to outline the changes, why they are
needed, and how this PR fixes the issue.
If the reason for the change is already explained clearly in an issue,
then it does not need to be restated here.
1. If you propose a new API or feature, clarify the use case for a new
API or feature.
  2. If you fix a bug, you can clarify why it is a bug.
-->

This PR introduces the `Deduplicator` trait which will unify Commit and
Checkpoint deduplication under a single API. This is done to move the
`&mut HashSet<..>` out of `AddRemoveDedup` and behind the `Deduplicator`
trait. That way, we can reuse `AddRemoveDedupVisitor` in an immutable
scenario.

<!--
Uncomment this section if there are any changes affecting public APIs:
### This PR affects the following public APIs

If there are breaking changes, please ensure the `breaking-changes`
label gets added by CI, and describe why the changes are needed.

Note that _new_ public APIs are not considered breaking.
-->


## How was this change tested?
<!--
Please make sure to add test cases that check the changes thoroughly
including negative and positive cases if possible.
If it was tested in a way different from regular unit tests, please
clarify how you tested, ideally via a reproducible test documented in
the PR description.
-->

---------

Co-authored-by: Nick Lanham <nicklan@users.noreply.github.com>

Author: OussamaSaoudi

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 as _;
 use crate::log_replay::{
     ActionsBatch, FileActionDeduplicator, FileActionKey, HasSelectionVector, LogReplayProcessor,
 };

kernel/src/expressions/mod.rs

@@ -273,14 +273,14 @@ fn fail_serialize_opaque_predicate<S>(
 where
     S: Serializer,
 {
-    Err(ser::Error::custom("Cannot serialize Opaque Expression"))
+    Err(ser::Error::custom("Cannot serialize an Opaque Predicate"))
 }
 
 fn fail_deserialize_opaque_predicate<'de, D>(_deserializer: D) -> Result<OpaquePredicate, D::Error>
 where
     D: Deserializer<'de>,
 {
-    Err(de::Error::custom("Cannot deserialize Opaque Expression"))
+    Err(de::Error::custom("Cannot deserialize an Opaque Predicate"))
 }
 
 impl OpaquePredicate {
@@ -316,7 +316,7 @@ fn fail_serialize_opaque_expression<S>(
 where
     S: Serializer,
 {
-    Err(ser::Error::custom("Cannot serialize Opaque Expression"))
+    Err(ser::Error::custom("Cannot serialize an Opaque Expression"))
 }
 
 fn fail_deserialize_opaque_expression<'de, D>(
@@ -325,7 +325,7 @@ fn fail_deserialize_opaque_expression<'de, D>(
 where
     D: Deserializer<'de>,
 {
-    Err(de::Error::custom("Cannot deserialize Opaque Expression"))
+    Err(de::Error::custom("Cannot deserialize an Opaque Expression"))
 }
 
 /// A transformation affecting a single field (one pieces of a [`Transform`]). The transformation
@@ -1161,6 +1161,7 @@ mod tests {
             Expression, Predicate, Scalar, Transform, UnaryExpressionOp, VariadicExpressionOp,
         };
         use crate::schema::{ArrayType, DataType, DecimalType, MapType, StructField};
+        use crate::utils::test_utils::assert_result_error_with_message;
 
         use super::assert_roundtrip;
 
@@ -1552,17 +1553,7 @@ mod tests {
 
             let expr = Expression::opaque(TestOpaqueExprOp, [Expression::literal(1)]);
             let result = serde_json::to_string(&expr);
-            assert!(
-                result.is_err(),
-                "Opaque expression serialization should fail"
-            );
-            assert!(
-                result
-                    .unwrap_err()
-                    .to_string()
-                    .contains("Cannot serialize Opaque Expression"),
-                "Error should mention opaque expression"
-            );
+            assert_result_error_with_message(result, "Cannot serialize an Opaque Expression");
         }
 
         #[test]
@@ -1610,17 +1601,7 @@ mod tests {
 
             let pred = Predicate::opaque(TestOpaquePredOp, [Expression::literal(1)]);
             let result = serde_json::to_string(&pred);
-            assert!(
-                result.is_err(),
-                "Opaque predicate serialization should fail"
-            );
-            assert!(
-                result
-                    .unwrap_err()
-                    .to_string()
-                    .contains("Cannot serialize Opaque Expression"),
-                "Error should mention opaque expression"
-            );
+            assert_result_error_with_message(result, "Cannot serialize an Opaque Predicate");
         }
     }
 }

kernel/src/log_replay.rs

@@ -13,8 +13,8 @@
 //! This module provides structures for efficient batch processing, focusing on file action
 //! deduplication with `FileActionDeduplicator` which tracks unique files across log batches
 //! to minimize memory usage for tables with extensive history.
-use crate::actions::deletion_vector::DeletionVectorDescriptor;
-use crate::engine_data::{GetData, TypedGetData};
+use crate::engine_data::GetData;
+use crate::log_replay::deduplicator::Deduplicator;
 use crate::scan::data_skipping::DataSkippingFilter;
 use crate::{DeltaResult, EngineData};
 
@@ -24,6 +24,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)]
@@ -56,7 +58,8 @@ pub(crate) struct FileActionDeduplicator<'seen> {
     seen_file_keys: &'seen mut HashSet<FileActionKey>,
     // TODO: Consider renaming to `is_commit_batch`, `deduplicate_batch`, or `save_batch`
     // to better reflect its role in deduplication logic.
-    /// Whether we're processing a log batch (as opposed to a checkpoint)
+    /// Whether we're processing a commit log JSON file (`true`) or a checkpoint file (`false`).
+    /// When `true`, file actions are added to `seen_file_keys` as they're processed.
     is_log_batch: bool,
     /// Index of the getter containing the add.path column
     add_path_index: usize,
@@ -86,12 +89,14 @@ impl<'seen> FileActionDeduplicator<'seen> {
             remove_dv_start_index,
         }
     }
+}
 
+impl<'seen> Deduplicator for 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 {
+    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
@@ -117,35 +122,6 @@ impl<'seen> FileActionDeduplicator<'seen> {
         }
     }
 
-    /// 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`).
-    fn extract_dv_unique_id<'a>(
-        &self,
-        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),
-        }
-    }
-
     /// 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
     /// to identify whether a file action exists and what type it is.
@@ -159,7 +135,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 +166,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
     }
 }

kernel/src/log_replay/deduplicator.rs

@@ -0,0 +1,65 @@
+//! Deduplication abstraction for log replay processors.
+//!
+//! The [`Deduplicator`] trait supports two deduplication strategies:
+//!
+//! - **JSON commit files** (`is_log_batch = true`): Tracks (path, dv_unique_id) and updates
+//!   the hashmap as files are seen. Implementation: [`FileActionDeduplicator`]
+//!
+//! - **Checkpoint files** (`is_log_batch = false`): Uses (path, dv_unique_id) to filter actions
+//!   using a read-only hashmap pre-populated from the commit log phase. Future implementation.
+//!
+//! [`FileActionDeduplicator`]: crate::log_replay::FileActionDeduplicator
+
+use crate::actions::deletion_vector::DeletionVectorDescriptor;
+use crate::engine_data::{GetData, TypedGetData};
+use crate::log_replay::FileActionKey;
+use crate::DeltaResult;
+
+pub(crate) trait Deduplicator {
+    /// Extracts a file action key from the data. Returns `(key, is_add)` if found.
+    ///
+    /// TODO: Remove the skip_removes field in the future. The caller is responsible for using the
+    /// correct Deduplicator instance depeding on whether the batch belongs to a commit or to a
+    /// checkpoint.
+    fn extract_file_action<'a>(
+        &self,
+        i: usize,
+        getters: &[&'a dyn GetData<'a>],
+        skip_removes: bool,
+    ) -> DeltaResult<Option<(FileActionKey, bool)>>;
+
+    /// Checks if this file has been seen. When `is_log_batch() = true`, updates the hashmap
+    /// to track new files. Returns `true` if the file should be filtered out.
+    fn check_and_record_seen(&mut self, key: FileActionKey) -> bool;
+
+    /// Returns `true` for commit log batches (updates hashmap), `false` for checkpoints (read-only).
+    fn is_log_batch(&self) -> bool;
+
+    /// 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`).
+    fn extract_dv_unique_id<'a>(
+        &self,
+        i: usize,
+        getters: &[&'a dyn GetData<'a>],
+        dv_start_index: usize,
+    ) -> DeltaResult<Option<String>> {
+        let Some(storage_type) =
+            getters[dv_start_index].get_opt(i, "deletionVector.storageType")?
+        else {
+            return Ok(None);
+        };
+        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,
+        )))
+    }
+}

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 _;
@@ -80,6 +81,15 @@ pub(crate) struct ScanLogReplayProcessor {
 }
 
 impl ScanLogReplayProcessor {
+    // 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
+
     /// Create a new [`ScanLogReplayProcessor`] instance
     pub(crate) fn new(engine: &dyn Engine, state_info: Arc<StateInfo>) -> DeltaResult<Self> {
         Self::new_with_seen_files(engine, state_info, Default::default())
@@ -234,40 +244,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,
@@ -319,8 +312,8 @@ impl AddRemoveDedupVisitor<'_> {
         // encounter if the table's schema was replaced after the most recent checkpoint.
         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")?;
+                let partition_values = getters[ScanLogReplayProcessor::ADD_PARTITION_VALUES_INDEX]
+                    .get(i, "add.partitionValues")?;
                 let partition_values = parse_partition_values(
                     &self.state_info.logical_schema,
                     transform,
@@ -340,7 +333,7 @@ impl AddRemoveDedupVisitor<'_> {
             return Ok(false);
         }
         let base_row_id: Option<i64> =
-            getters[Self::BASE_ROW_ID_INDEX].get_opt(i, "add.baseRowId")?;
+            getters[ScanLogReplayProcessor::BASE_ROW_ID_INDEX].get_opt(i, "add.baseRowId")?;
         let transform = self
             .state_info
             .transform_spec
@@ -363,7 +356,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(|| {
@@ -514,12 +507,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,
+            Self::ADD_PATH_INDEX,
+            Self::REMOVE_PATH_INDEX,
+            Self::ADD_DV_START_INDEX,
+            Self::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())?;