test: introduce the new MockJsonHandler and MockParquetHandler

kernel/src/lib.rs

@@ -106,6 +106,9 @@ pub use error::{DeltaResult, Error};
 pub use expressions::{Expression, ExpressionRef};
 pub use table::Table;
 
+#[cfg(test)]
+use path::ParsedLogPath;
+
 #[cfg(any(
     feature = "default-engine",
     feature = "sync-engine",
@@ -183,6 +186,12 @@ impl FileMeta {
             size,
         }
     }
+
+    /// Create a ParsedLogPath from this FileMeta
+    #[cfg(test)]
+    fn to_parsed_log_path(&self) -> ParsedLogPath {
+        ParsedLogPath::try_from(self.clone()).unwrap().unwrap()
+    }
 }
 
 /// Extension trait that makes it easier to work with traits objects that implement [`Any`],

kernel/src/log_segment/tests.rs

@@ -4,14 +4,106 @@ use itertools::Itertools;
 use object_store::{memory::InMemory, path::Path, ObjectStore};
 use url::Url;
 
+use crate::actions::{get_log_add_schema, PROTOCOL_NAME};
 use crate::engine::default::executor::tokio::TokioBackgroundExecutor;
 use crate::engine::default::filesystem::ObjectStoreFileSystemClient;
 use crate::engine::sync::SyncEngine;
 use crate::log_segment::LogSegment;
+use crate::scan::test_utils::add_batch_simple;
 use crate::snapshot::CheckpointMetadata;
-use crate::{FileSystemClient, Table};
+use crate::utils::test_utils::MockEngineContext;
+use crate::{EngineData, Expression, FileMeta, FileSystemClient, Table};
 use test_utils::delta_path_for_version;
 
+fn create_file_meta(path: &str) -> FileMeta {
+    FileMeta {
+        location: Url::parse(path).expect("Invalid file URL"),
+        last_modified: 0,
+        size: 0,
+    }
+}
+
+#[test]
+fn test_log_replay() {
+    // Retrieve the engine context
+    let engine_context = MockEngineContext::new();
+
+    // Define checkpoint and commit files
+    let checkpoint_meta = create_file_meta("file:///00000000000000000001.checkpoint1.json");
+    let checkpoint_parts = vec![checkpoint_meta.to_parsed_log_path()];
+    let commit_files = [
+        "file:///00000000000000000001.version1.json",
+        "file:///00000000000000000002.version2.json",
+    ]
+    .into_iter()
+    .map(|path| create_file_meta(path).to_parsed_log_path())
+    .collect::<Vec<_>>();
+
+    // Define log segment
+    let log_segment = LogSegment {
+        end_version: 0,
+        log_root: Url::parse("file:///root.json").expect("Invalid log root URL"),
+        ascending_commit_files: commit_files.clone(),
+        checkpoint_parts,
+    };
+
+    // Expected commit files should be read in reverse order
+    let expected_commit_files_read = [
+        "file:///00000000000000000002.version2.json",
+        "file:///00000000000000000001.version1.json",
+    ]
+    .into_iter()
+    .map(create_file_meta)
+    .collect::<Vec<_>>();
+
+    // Define predicate and projected schemas
+    let predicate = Some(Arc::new(
+        Expression::column([PROTOCOL_NAME, "minReaderVersion"]).is_not_null(),
+    ));
+    let commit_schema = get_log_add_schema().clone();
+    let checkpoint_schema = get_log_add_schema().clone();
+
+    // Expect the JSON and Parquet handlers to receive the correct files, schemas, and predicates
+    engine_context.json_handler.expect_read_json_files(
+        expected_commit_files_read.clone(),
+        commit_schema.clone(),
+        predicate.clone(),
+        Ok(Box::new(std::iter::once(Ok(
+            add_batch_simple() as Box<dyn EngineData>
+        )))),
+    );
+    engine_context.parquet_handler.expect_read_parquet_files(
+        vec![checkpoint_meta],
+        checkpoint_schema.clone(),
+        predicate.clone(),
+        Ok(Box::new(std::iter::once(Ok(
+            add_batch_simple() as Box<dyn EngineData>
+        )))),
+    );
+
+    // Run the log replay
+    let mut iter = log_segment
+        .replay(
+            &engine_context.engine,
+            commit_schema,
+            checkpoint_schema,
+            predicate,
+        )
+        .unwrap()
+        .into_iter();
+
+    // Verify the commit batch and checkpoint batch are chained together in order
+    assert!(
+        iter.next().unwrap().unwrap().1,
+        "First element should be a commit batch"
+    );
+    assert!(
+        !iter.next().unwrap().unwrap().1,
+        "Second element should be a checkpoint batch"
+    );
+    assert!(iter.next().is_none(), "Iterator should only have 2 batches");
+}
+
 // NOTE: In addition to testing the meta-predicate for metadata replay, this test also verifies
 // that the parquet reader properly infers nullcount = rowcount for missing columns. The two
 // checkpoint part files that contain transaction app ids have truncated schemas that would