use immutable addrmdedup

Author: OussamaSaoudi

kernel/src/log_replay.rs

@@ -204,6 +204,10 @@ impl ActionsBatch {
     }
 }
 
+pub(crate) trait ParallelizableLogReplayProcessor: LogReplayProcessor {
+    fn process_actions_batch(&self, actions_batch: ActionsBatch) -> DeltaResult<Self::Output>;
+}
+
 /// A trait for processing batches of actions from Delta transaction logs during log replay.
 ///
 /// Log replay processors scan transaction logs in **reverse chronological order** (newest to oldest),

kernel/src/log_replay/deduplicator.rs

@@ -10,11 +10,12 @@
 //!
 //! [`FileActionDeduplicator`]: crate::log_replay::FileActionDeduplicator
 
-use crate::{
-    actions::deletion_vector::DeletionVectorDescriptor,
-    engine_data::{GetData, TypedGetData},
-    DeltaResult,
-};
+use std::collections::HashSet;
+
+use crate::actions::deletion_vector::DeletionVectorDescriptor;
+use crate::engine_data::{GetData, TypedGetData};
+use crate::log_replay::FileActionKey;
+use crate::DeltaResult;
 
 pub(crate) trait Deduplicator {
     /// Key type for identifying file actions. JSON deduplicators use `FileActionKey`
@@ -67,24 +68,24 @@ pub(crate) trait Deduplicator {
 }
 
 #[allow(unused)]
-pub(crate) struct CheckpointDeduplicator {
-    seen_file_keys: HashSet<String>,
+pub(crate) struct CheckpointDeduplicator<'a> {
+    seen_file_keys: &'a HashSet<FileActionKey>,
     add_path_index: usize,
 }
-impl CheckpointDeduplicator {
+impl CheckpointDeduplicator<'_> {
     #[allow(unused)]
-    pub(crate) fn try_new(
-        seen_file_keys: HashSet<String>,
+    pub(crate) fn try_new<'a>(
+        seen_file_keys: &'a HashSet<FileActionKey>,
         add_path_index: usize,
-    ) -> DeltaResult<Self> {
-        Ok(Self {
+    ) -> DeltaResult<CheckpointDeduplicator<'a>> {
+        Ok(CheckpointDeduplicator {
             seen_file_keys,
             add_path_index,
         })
     }
 }
 
-impl Deduplicator for CheckpointDeduplicator {
+impl Deduplicator for CheckpointDeduplicator<'_> {
     type Key = String;
 
     fn extract_file_action<'a>(

kernel/src/scan/log_replay.rs

@@ -12,8 +12,11 @@ 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::log_replay::deduplicator::{self, CheckpointDeduplicator, Deduplicator};
+use crate::log_replay::{
+    ActionsBatch, FileActionDeduplicator, FileActionKey, LogReplayProcessor,
+    ParallelizableLogReplayProcessor,
+};
 use crate::scan::Scalar;
 use crate::schema::ToSchema as _;
 use crate::schema::{ColumnNamesAndTypes, DataType, MapType, StructField, StructType};
@@ -480,6 +483,37 @@ pub(crate) fn get_scan_metadata_transform_expr() -> ExpressionRef {
     EXPR.clone()
 }
 
+impl ParallelizableLogReplayProcessor for ScanLogReplayProcessor {
+    fn process_actions_batch(&self, actions_batch: ActionsBatch) -> DeltaResult<Self::Output> {
+        let ActionsBatch {
+            actions,
+            is_log_batch: _,
+        } = actions_batch;
+        // Build an initial selection vector for the batch which has had the data skipping filter
+        // applied. The selection vector is further updated by the deduplication visitor to remove
+        // rows that are not valid adds.
+        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 mut visitor = AddRemoveDedupVisitor::new(
+            deduplicator,
+            selection_vector,
+            self.state_info.clone(),
+            self.partition_filter.clone(),
+        );
+        visitor.visit_rows_of(actions.as_ref())?;
+
+        // TODO: Teach expression eval to respect the selection vector we just computed so carefully!
+        let result = self.add_transform.evaluate(actions.as_ref())?;
+        ScanMetadata::try_new(
+            result,
+            visitor.selection_vector,
+            visitor.row_transform_exprs,
+        )
+    }
+}
+
 impl LogReplayProcessor for ScanLogReplayProcessor {
     type Output = ScanMetadata;