feat: add CheckpointLogReplayProcessor in new checkpoints mod

kernel/src/actions/visitors.rs

@@ -3,9 +3,11 @@
 
 use std::collections::{HashMap, HashSet};
 use std::sync::LazyLock;
+
 use tracing::debug;
 
 use crate::engine_data::{GetData, RowVisitor, TypedGetData as _};
+use crate::log_replay::FileActionKey;
 use crate::schema::{column_name, ColumnName, ColumnNamesAndTypes, DataType};
 use crate::utils::require;
 use crate::{DeltaResult, Error};
@@ -502,27 +504,29 @@ impl RowVisitor for SidecarVisitor {
 #[cfg_attr(feature = "developer-visibility", visibility::make(pub))]
 pub(crate) struct CheckpointVisitor<'seen> {
     // File actions deduplication state
-    deduplicator: FileActionDeduplicator<'seen>,
-    total_file_actions: usize,
-    total_add_actions: usize,
-    minimum_file_retention_timestamp: i64,
+    pub(crate) deduplicator: FileActionDeduplicator<'seen>,
+    pub(crate) total_file_actions: usize,
+    pub(crate) total_add_actions: usize,
+    pub(crate) minimum_file_retention_timestamp: i64,
 
     // Non-file actions deduplication state
-    seen_protocol: bool,
-    seen_metadata: bool,
-    seen_txns: &'seen mut HashSet<String>,
-    total_non_file_actions: usize,
+    pub(crate) seen_protocol: bool,
+    pub(crate) seen_metadata: bool,
+    pub(crate) seen_txns: &'seen mut HashSet<String>,
+    pub(crate) total_non_file_actions: usize,
 }
 
 #[allow(unused)]
 impl CheckpointVisitor<'_> {
     /// Create a new CheckpointVisitor
-    fn new<'seen>(
+    pub(crate) fn new<'seen>(
         seen_file_keys: &'seen mut HashSet<FileActionKey>,
-        seen_txns: &'seen mut HashSet<String>,
         selection_vector: Vec<bool>,
         is_log_batch: bool,
         minimum_file_retention_timestamp: i64,
+        seen_protocol: bool,
+        seen_metadata: bool,
+        seen_txns: &'seen mut HashSet<String>,
     ) -> CheckpointVisitor<'seen> {
         CheckpointVisitor {
             deduplicator: FileActionDeduplicator::new(
@@ -534,8 +538,8 @@ impl CheckpointVisitor<'_> {
             total_add_actions: 0,
             minimum_file_retention_timestamp,
 
-            seen_protocol: false,
-            seen_metadata: false,
+            seen_protocol,
+            seen_metadata,
             seen_txns,
             total_non_file_actions: 0,
         }
@@ -703,20 +707,6 @@ impl RowVisitor for CheckpointVisitor<'_> {
     }
 }
 
-/// 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)]
-pub(crate) struct FileActionKey {
-    pub(crate) path: String,
-    pub(crate) dv_unique_id: Option<String>,
-}
-impl FileActionKey {
-    pub(crate) fn new(path: impl Into<String>, dv_unique_id: Option<String>) -> Self {
-        let path = path.into();
-        Self { path, dv_unique_id }
-    }
-}
-
 /// This struct contains indices and configuration options needed to
 /// extract file actions from action batches in the Delta log.
 pub(crate) struct FileActionExtractConfig {
@@ -923,25 +913,11 @@ pub(crate) fn visit_deletion_vector_at<'a>(
 
 #[cfg(test)]
 mod tests {
-    use std::sync::Arc;
-
-    use crate::arrow::array::{RecordBatch, StringArray};
-    use crate::arrow::datatypes::{DataType, Field, Schema as ArrowSchema};
+    use crate::arrow::array::StringArray;
+    use crate::utils::test_utils::parse_json_batch;
+    use crate::EngineData;
 
     use super::*;
-    use crate::{
-        actions::get_log_schema, engine::arrow_data::ArrowEngineData, engine::sync::SyncEngine,
-        Engine, EngineData,
-    };
-
-    // TODO(nick): Merge all copies of this into one "test utils" thing
-    fn string_array_to_engine_data(string_array: StringArray) -> Box<dyn EngineData> {
-        let string_field = Arc::new(Field::new("a", DataType::Utf8, true));
-        let schema = Arc::new(ArrowSchema::new(vec![string_field]));
-        let batch = RecordBatch::try_new(schema, vec![Arc::new(string_array)])
-            .expect("Can't convert to record batch");
-        Box::new(ArrowEngineData::new(batch))
-    }
 
     fn action_batch() -> Box<dyn EngineData> {
         let json_strings: StringArray = vec![
@@ -958,15 +934,6 @@ mod tests {
         parse_json_batch(json_strings)
     }
 
-    fn parse_json_batch(json_strings: StringArray) -> Box<dyn EngineData> {
-        let engine = SyncEngine::new();
-        let json_handler = engine.get_json_handler();
-        let output_schema = get_log_schema().clone();
-        json_handler
-            .parse_json(string_array_to_engine_data(json_strings), output_schema)
-            .unwrap()
-    }
-
     #[test]
     fn test_parse_protocol() -> DeltaResult<()> {
         let data = action_batch();
@@ -1180,10 +1147,12 @@ mod tests {
         let mut seen_txns = HashSet::new();
         let mut visitor = CheckpointVisitor::new(
             &mut seen_file_keys,
-            &mut seen_txns,
             vec![false; 8],
             true,
-            0, // minimum_file_retention_timestamp (no expired tombstones)
+            0,     // minimum_file_retention_timestamp (no expired tombstones)
+            false, // seen_protocol
+            false, // seen_metadata
+            &mut seen_txns,
         );
 
         visitor.visit_rows_of(data.as_ref())?;
@@ -1226,10 +1195,12 @@ mod tests {
         let mut seen_txns = HashSet::new();
         let mut visitor = CheckpointVisitor::new(
             &mut seen_file_keys,
-            &mut seen_txns,
             vec![false; 4],
             true,
-            100, // minimum_file_retention_timestamp (threshold set to 100)
+            100,   // minimum_file_retention_timestamp (threshold set to 100)
+            false, // seen_protocol
+            false, // seen_metadata
+            &mut seen_txns,
         );
 
         visitor.visit_rows_of(batch.as_ref())?;
@@ -1247,16 +1218,23 @@ mod tests {
     fn test_checkpoint_visitor_conflicting_file_actions_in_log_batch() -> DeltaResult<()> {
         let json_strings: StringArray = vec![
             r#"{"add":{"path":"file1","partitionValues":{"c1":"6","c2":"a"},"size":452,"modificationTime":1670892998137,"dataChange":true}}"#,
-             // Duplicate path
+            // Duplicate path
             r#"{"remove":{"path":"file1","deletionTimestamp":100,"dataChange":true,"partitionValues":{}}}"#,
         ]
         .into();
         let batch = parse_json_batch(json_strings);
 
         let mut seen_file_keys = HashSet::new();
         let mut seen_txns = HashSet::new();
-        let mut visitor =
-            CheckpointVisitor::new(&mut seen_file_keys, &mut seen_txns, vec![false; 2], true, 0);
+        let mut visitor = CheckpointVisitor::new(
+            &mut seen_file_keys,
+            vec![false; 2],
+            true,
+            0,     // minimum_file_retention_timestamp
+            false, // seen_protocol
+            false, // seen_metadata
+            &mut seen_txns,
+        );
 
         visitor.visit_rows_of(batch.as_ref())?;
 
@@ -1285,10 +1263,12 @@ mod tests {
         let mut seen_txns = HashSet::new();
         let mut visitor = CheckpointVisitor::new(
             &mut seen_file_keys,
-            &mut seen_txns,
             vec![false; 2],
             false, // is_log_batch = false (checkpoint batch)
-            0,
+            0,     // minimum_file_retention_timestamp
+            false, // seen_protocol
+            false, // seen_metadata
+            &mut seen_txns,
         );
 
         visitor.visit_rows_of(batch.as_ref())?;
@@ -1316,8 +1296,15 @@ mod tests {
 
         let mut seen_file_keys = HashSet::new();
         let mut seen_txns = HashSet::new();
-        let mut visitor =
-            CheckpointVisitor::new(&mut seen_file_keys, &mut seen_txns, vec![false; 3], true, 0);
+        let mut visitor = CheckpointVisitor::new(
+            &mut seen_file_keys,
+            vec![false; 3],
+            true,  // is_log_batch
+            0,     // minimum_file_retention_timestamp
+            false, // seen_protocol
+            false, // seen_metadata
+            &mut seen_txns,
+        );
 
         visitor.visit_rows_of(batch.as_ref())?;
 
@@ -1341,8 +1328,15 @@ mod tests {
 
         let mut seen_file_keys = HashSet::new();
         let mut seen_txns = HashSet::new();
-        let mut visitor =
-            CheckpointVisitor::new(&mut seen_file_keys, &mut seen_txns, vec![false; 3], true, 0);
+        let mut visitor = CheckpointVisitor::new(
+            &mut seen_file_keys,
+            vec![false; 3],
+            true,  // is_log_batch
+            0,     // minimum_file_retention_timestamp
+            false, // seen_protocol
+            false, // seen_metadata
+            &mut seen_txns,
+        );
 
         visitor.visit_rows_of(batch.as_ref())?;
 
@@ -1373,16 +1367,14 @@ mod tests {
 
         let mut visitor = CheckpointVisitor::new(
             &mut seen_file_keys,
-            &mut seen_txns, // Pre-populated transaction
             vec![false; 3],
-            true,
-            0,
+            true,           // is_log_batch
+            0,              // minimum_file_retention_timestamp
+            true,           // seen_protocol - Already seen
+            true,           // seen_metadata - Already seen
+            &mut seen_txns, // Pre-populated transaction
         );
 
-        // Mark these as already seen
-        visitor.seen_protocol = true;
-        visitor.seen_metadata = true;
-
         visitor.visit_rows_of(batch.as_ref())?;
 
         // All actions should be skipped as they have already been seen
@@ -1413,10 +1405,12 @@ mod tests {
         let mut seen_txns = HashSet::new();
         let mut visitor = CheckpointVisitor::new(
             &mut seen_file_keys,
-            &mut seen_txns,
             vec![false; 7],
-            true, // is_log_batch
-            0,    // minimum_file_retention_timestamp
+            true,  // is_log_batch
+            0,     // minimum_file_retention_timestamp
+            false, // seen_protocol
+            false, // seen_metadata
+            &mut seen_txns,
         );
 
         visitor.visit_rows_of(batch.as_ref())?;
@@ -1425,7 +1419,7 @@ mod tests {
         let expected = vec![true, false, true, true, false, true, false];
         assert_eq!(visitor.deduplicator.selection_vector, expected);
         assert_eq!(visitor.seen_txns.len(), 2); // Two different app IDs
-        assert_eq!(visitor.total_non_file_actions, 4); // 2 txns + 1 protocol + 1 metadata
+        assert_eq!(visitor.total_non_file_actions, 4);
         assert_eq!(visitor.total_file_actions, 0);
 
         Ok(())