fix: resolve merge conflicts in checkpoint tests

Author: DrakeLin

kernel/src/checkpoint/tests.rs

@@ -9,14 +9,9 @@ use crate::arrow::{
     array::{create_array, RecordBatch},
     datatypes::Field,
 };
-<<<<<<< HEAD
-use crate::checkpoint::create_last_checkpoint_data;
-use crate::engine::arrow_data::ArrowEngineData;
-use crate::engine::default::executor::tokio::TokioMultiThreadExecutor;
-=======
 use crate::checkpoint::{create_last_checkpoint_data, CheckpointDataIterator};
 use crate::engine::arrow_data::{ArrowEngineData, EngineDataArrowExt};
->>>>>>> 9df95006 (write-integration)
+use crate::engine::default::executor::tokio::TokioMultiThreadExecutor;
 use crate::engine::default::DefaultEngineBuilder;
 use crate::log_replay::HasSelectionVector;
 use crate::schema::{DataType as KernelDataType, StructField, StructType};
@@ -629,22 +624,6 @@ async fn test_no_checkpoint_staged_commits() -> DeltaResult<()> {
     Ok(())
 }
 
-<<<<<<< HEAD
-#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
-async fn test_snapshot_checkpoint() -> DeltaResult<()> {
-    let (store, _) = new_in_memory_store();
-    let executor = Arc::new(TokioMultiThreadExecutor::new(
-        tokio::runtime::Handle::current(),
-    ));
-    let engine = DefaultEngineBuilder::new(store.clone())
-        .with_task_executor(executor)
-        .build();
-
-    // Version 0: metadata & protocol
-    write_commit_to_store(
-        &store,
-        vec![create_metadata_action(), create_basic_protocol_action()],
-=======
 /// Create a Metadata action with writeStatsAsStruct enabled
 fn create_metadata_action_with_stats_struct() -> Action {
     let mut config = HashMap::new();
@@ -682,47 +661,36 @@ fn create_add_action_with_stats(path: &str, num_records: i64) -> Action {
     })
 }
 
-/// Tests checkpoint_data with writeStatsAsStruct enabled.
-/// Verifies that the output schema includes stats_parsed.
-#[tokio::test]
-async fn test_checkpoint_data_struct_enabled() -> DeltaResult<()> {
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn test_snapshot_checkpoint() -> DeltaResult<()> {
     let (store, _) = new_in_memory_store();
-    let engine = DefaultEngineBuilder::new(store.clone()).build();
+    let executor = Arc::new(TokioMultiThreadExecutor::new(
+        tokio::runtime::Handle::current(),
+    ));
+    let engine = DefaultEngineBuilder::new(store.clone())
+        .with_task_executor(executor)
+        .build();
 
-    // 1st commit: protocol + metadata with writeStatsAsStruct=true
+    // Version 0: metadata & protocol
     write_commit_to_store(
         &store,
-        vec![
-            create_basic_protocol_action(),
-            create_metadata_action_with_stats_struct(),
-        ],
->>>>>>> 9df95006 (write-integration)
+        vec![create_metadata_action(), create_basic_protocol_action()],
         0,
     )
     .await?;
 
-<<<<<<< HEAD
     // Version 1: add 3 files
     write_commit_to_store(
         &store,
         vec![
             create_add_action("file1.parquet"),
             create_add_action("file2.parquet"),
             create_add_action("file3.parquet"),
-=======
-    // 2nd commit: add actions with JSON stats
-    write_commit_to_store(
-        &store,
-        vec![
-            create_add_action_with_stats("file1.parquet", 100),
-            create_add_action_with_stats("file2.parquet", 200),
->>>>>>> 9df95006 (write-integration)
         ],
         1,
     )
     .await?;
 
-<<<<<<< HEAD
     // Version 2: add 2 more files, remove 1
     write_commit_to_store(
         &store,
@@ -755,7 +723,75 @@ async fn test_checkpoint_data_struct_enabled() -> DeltaResult<()> {
             create_add_action("file8.parquet"),
         ],
         4,
-=======
+    )
+    .await?;
+
+    let table_root = Url::parse("memory:///")?;
+    let snapshot = Snapshot::builder_for(table_root.clone()).build(&engine)?;
+
+    snapshot.checkpoint(&engine)?;
+
+    // First checkpoint: 1 metadata + 1 protocol + 5 add + 3 remove = 10, numOfAddFiles = 5
+    let checkpoint_path = Path::from("_delta_log/00000000000000000004.checkpoint.parquet");
+    let checkpoint_size = store.head(&checkpoint_path).await?.size;
+    assert_last_checkpoint_contents(&store, 4, 10, 5, checkpoint_size).await?;
+
+    // Version 5: add 2 files, remove 1
+    write_commit_to_store(
+        &store,
+        vec![
+            create_add_action("file9.parquet"),
+            create_add_action("file10.parquet"),
+            create_remove_action("file4.parquet"),
+        ],
+        5,
+    )
+    .await?;
+
+    // Version 6: add 1 file
+    write_commit_to_store(&store, vec![create_add_action("file11.parquet")], 6).await?;
+
+    let snapshot = Snapshot::builder_for(table_root).build(&engine)?;
+
+    snapshot.checkpoint(&engine)?;
+
+    // Second checkpoint: 1 metadata + 1 protocol + 7 add + 4 remove = 13, numOfAddFiles = 7
+    let checkpoint_path = Path::from("_delta_log/00000000000000000006.checkpoint.parquet");
+    let checkpoint_size = store.head(&checkpoint_path).await?.size;
+    assert_last_checkpoint_contents(&store, 6, 13, 7, checkpoint_size).await?;
+
+    Ok(())
+}
+
+/// Tests checkpoint_data with writeStatsAsStruct enabled.
+/// Verifies that the output schema includes stats_parsed.
+#[tokio::test]
+async fn test_checkpoint_data_struct_enabled() -> DeltaResult<()> {
+    let (store, _) = new_in_memory_store();
+    let engine = DefaultEngineBuilder::new(store.clone()).build();
+
+    // 1st commit: protocol + metadata with writeStatsAsStruct=true
+    write_commit_to_store(
+        &store,
+        vec![
+            create_basic_protocol_action(),
+            create_metadata_action_with_stats_struct(),
+        ],
+        0,
+    )
+    .await?;
+
+    // 2nd commit: add actions with JSON stats
+    write_commit_to_store(
+        &store,
+        vec![
+            create_add_action_with_stats("file1.parquet", 100),
+            create_add_action_with_stats("file2.parquet", 200),
+        ],
+        1,
+    )
+    .await?;
+
     let table_root = Url::parse("memory:///")?;
     let snapshot = Snapshot::builder_for(table_root).build(&engine)?;
     let writer = snapshot.create_checkpoint_writer()?;
@@ -812,45 +848,10 @@ async fn test_checkpoint_data_default_settings() -> DeltaResult<()> {
         &store,
         vec![create_add_action_with_stats("file1.parquet", 100)],
         1,
->>>>>>> 9df95006 (write-integration)
     )
     .await?;
 
     let table_root = Url::parse("memory:///")?;
-<<<<<<< HEAD
-    let snapshot = Snapshot::builder_for(table_root.clone()).build(&engine)?;
-
-    snapshot.checkpoint(&engine)?;
-
-    // First checkpoint: 1 metadata + 1 protocol + 5 add + 3 remove = 10, numOfAddFiles = 5
-    let checkpoint_path = Path::from("_delta_log/00000000000000000004.checkpoint.parquet");
-    let checkpoint_size = store.head(&checkpoint_path).await?.size;
-    assert_last_checkpoint_contents(&store, 4, 10, 5, checkpoint_size).await?;
-
-    // Version 5: add 2 files, remove 1
-    write_commit_to_store(
-        &store,
-        vec![
-            create_add_action("file9.parquet"),
-            create_add_action("file10.parquet"),
-            create_remove_action("file4.parquet"),
-        ],
-        5,
-    )
-    .await?;
-
-    // Version 6: add 1 file
-    write_commit_to_store(&store, vec![create_add_action("file11.parquet")], 6).await?;
-
-    let snapshot = Snapshot::builder_for(table_root).build(&engine)?;
-
-    snapshot.checkpoint(&engine)?;
-
-    // Second checkpoint: 1 metadata + 1 protocol + 7 add + 4 remove = 13, numOfAddFiles = 7
-    let checkpoint_path = Path::from("_delta_log/00000000000000000006.checkpoint.parquet");
-    let checkpoint_size = store.head(&checkpoint_path).await?.size;
-    assert_last_checkpoint_contents(&store, 6, 13, 7, checkpoint_size).await?;
-=======
     let snapshot = Snapshot::builder_for(table_root).build(&engine)?;
     let writer = snapshot.create_checkpoint_writer()?;
 
@@ -1096,7 +1097,6 @@ async fn test_all_stats_config_combinations() -> DeltaResult<()> {
             struct2
         );
     }
->>>>>>> 9df95006 (write-integration)
 
     Ok(())
 }

kernel/src/snapshot.rs

@@ -10,7 +10,7 @@ use crate::actions::domain_metadata::{
 };
 use crate::actions::set_transaction::SetTransactionScanner;
 use crate::actions::INTERNAL_DOMAIN_PREFIX;
-use crate::checkpoint::CheckpointWriter;
+use crate::checkpoint::{CheckpointDataIterator, CheckpointWriter};
 use crate::committer::Committer;
 use crate::listed_log_files::{ListedLogFiles, ListedLogFilesBuilder};
 use crate::log_segment::LogSegment;