fix

Author: gotocoding-DB

kernel/src/actions/mod.rs

@@ -1712,7 +1712,7 @@ mod tests {
     fn test_commit_info_into_engine_data() {
         let engine = ExprEngine::new();
 
-        let commit_info = CommitInfo::new(0, None, None);
+        let commit_info = CommitInfo::new(0, None, None, None);
         let commit_info_txn_id = commit_info.txn_id.clone();
 
         let engine_data = commit_info.into_engine_data(CommitInfo::to_schema().into(), &engine);

kernel/src/checkpoint/tests.rs

@@ -60,11 +60,7 @@ fn test_deleted_file_retention_timestamp() -> DeltaResult<()> {
 #[test]
 fn test_create_checkpoint_metadata_batch() -> DeltaResult<()> {
     let (store, _) = new_in_memory_store();
-    let engine = DefaultEngine::new(
-        store.clone(),
-        Arc::new(TokioBackgroundExecutor::new()),
-        None,
-    );
+    let engine = DefaultEngine::new(store.clone(), Arc::new(TokioBackgroundExecutor::new()));
 
     // 1st commit (version 0) - metadata and protocol actions
     // Protocol action does not include the v2Checkpoint reader/writer feature.
@@ -120,11 +116,7 @@ fn test_create_last_checkpoint_data() -> DeltaResult<()> {
     let add_actions_counter = 75;
     let size_in_bytes: i64 = 1024 * 1024; // 1MB
     let (store, _) = new_in_memory_store();
-    let engine = DefaultEngine::new(
-        store.clone(),
-        Arc::new(TokioBackgroundExecutor::new()),
-        None,
-    );
+    let engine = DefaultEngine::new(store.clone(), Arc::new(TokioBackgroundExecutor::new()));
 
     // Create last checkpoint metadata
     let last_checkpoint_batch = create_last_checkpoint_data(
@@ -285,11 +277,7 @@ fn read_last_checkpoint_file(store: &Arc<InMemory>) -> DeltaResult<Value> {
 #[test]
 fn test_v1_checkpoint_latest_version_by_default() -> DeltaResult<()> {
     let (store, _) = new_in_memory_store();
-    let engine = DefaultEngine::new(
-        store.clone(),
-        Arc::new(TokioBackgroundExecutor::new()),
-        None,
-    );
+    let engine = DefaultEngine::new(store.clone(), Arc::new(TokioBackgroundExecutor::new()));
 
     // 1st commit: adds `fake_path_1`
     write_commit_to_store(&store, vec![create_add_action("fake_path_1")], 0)?;
@@ -359,11 +347,7 @@ fn test_v1_checkpoint_latest_version_by_default() -> DeltaResult<()> {
 #[test]
 fn test_v1_checkpoint_specific_version() -> DeltaResult<()> {
     let (store, _) = new_in_memory_store();
-    let engine = DefaultEngine::new(
-        store.clone(),
-        Arc::new(TokioBackgroundExecutor::new()),
-        None,
-    );
+    let engine = DefaultEngine::new(store.clone(), Arc::new(TokioBackgroundExecutor::new()));
 
     // 1st commit (version 0) - metadata and protocol actions
     // Protocol action does not include the v2Checkpoint reader/writer feature.
@@ -425,11 +409,7 @@ fn test_v1_checkpoint_specific_version() -> DeltaResult<()> {
 #[test]
 fn test_finalize_errors_if_checkpoint_data_iterator_is_not_exhausted() -> DeltaResult<()> {
     let (store, _) = new_in_memory_store();
-    let engine = DefaultEngine::new(
-        store.clone(),
-        Arc::new(TokioBackgroundExecutor::new()),
-        None,
-    );
+    let engine = DefaultEngine::new(store.clone(), Arc::new(TokioBackgroundExecutor::new()));
 
     // 1st commit (version 0) - metadata and protocol actions
     write_commit_to_store(
@@ -471,11 +451,7 @@ fn test_finalize_errors_if_checkpoint_data_iterator_is_not_exhausted() -> DeltaR
 #[test]
 fn test_v2_checkpoint_supported_table() -> DeltaResult<()> {
     let (store, _) = new_in_memory_store();
-    let engine = DefaultEngine::new(
-        store.clone(),
-        Arc::new(TokioBackgroundExecutor::new()),
-        None,
-    );
+    let engine = DefaultEngine::new(store.clone(), Arc::new(TokioBackgroundExecutor::new()));
 
     // 1st commit: adds `fake_path_2` & removes `fake_path_1`
     write_commit_to_store(
@@ -547,11 +523,7 @@ fn test_v2_checkpoint_supported_table() -> DeltaResult<()> {
 #[test]
 fn test_no_checkpoint_staged_commits() -> DeltaResult<()> {
     let (store, _) = new_in_memory_store();
-    let engine = DefaultEngine::new(
-        store.clone(),
-        Arc::new(TokioBackgroundExecutor::new()),
-        None,
-    );
+    let engine = DefaultEngine::new(store.clone(), Arc::new(TokioBackgroundExecutor::new()));
 
     // normal commit
     write_commit_to_store(

kernel/src/committer.rs

@@ -212,11 +212,7 @@ mod tests {
     async fn catalog_managed_tables_block_transactions() {
         let storage = Arc::new(InMemory::new());
         let table_root = Url::parse("memory:///").unwrap();
-        let engine = DefaultEngine::new(
-            storage.clone(),
-            Arc::new(TokioBackgroundExecutor::new()),
-            None,
-        );
+        let engine = DefaultEngine::new(storage.clone(), Arc::new(TokioBackgroundExecutor::new()));
 
         let actions = [
             r#"{"commitInfo":{"timestamp":12345678900,"inCommitTimestamp":12345678900}}"#,

kernel/src/engine/default/filesystem.rs

@@ -393,7 +393,7 @@ mod tests {
         store.put(&name, data.clone().into()).await.unwrap();
 
         let table_root = Url::parse("memory:///").expect("valid url");
-        let engine = DefaultEngine::new(store, Arc::new(TokioBackgroundExecutor::new()), None);
+        let engine = DefaultEngine::new(store, Arc::new(TokioBackgroundExecutor::new()));
         let files: Vec<_> = engine
             .storage_handler()
             .list_from(&table_root.join("_delta_log").unwrap().join("0").unwrap())
@@ -423,7 +423,7 @@ mod tests {
 
         let url = Url::from_directory_path(tmp.path()).unwrap();
         let store = Arc::new(LocalFileSystem::new());
-        let engine = DefaultEngine::new(store, Arc::new(TokioBackgroundExecutor::new()), None);
+        let engine = DefaultEngine::new(store, Arc::new(TokioBackgroundExecutor::new()));
         let files = engine
             .storage_handler()
             .list_from(&url.join("_delta_log").unwrap().join("0").unwrap())

kernel/src/kernel_predicates/tests.rs

@@ -387,7 +387,7 @@ fn test_eval_junction() {
         (&[], Some(true), Some(false)),
         (&[Some(true)], Some(true), Some(true)),
         (&[Some(false)], Some(false), Some(false)),
-        (&[None], None),
+        (&[None], None, None),
         (&[Some(true), Some(false)], Some(false), Some(true)),
         (&[Some(false), Some(true)], Some(false), Some(true)),
         (&[Some(true), None], None, Some(true)),

kernel/src/listed_log_files.rs

@@ -682,7 +682,7 @@ mod list_log_files_with_log_tail_tests {
         ];
 
         let (storage, log_root) = create_storage(log_files);
-        let result: Vec<_> = list_log_files(storage.as_ref(), &log_root, vec![], None)
+        let result: Vec<_> = list_log_files(storage.as_ref(), &log_root, vec![], None, None)
             .unwrap()
             .try_collect()
             .unwrap();

kernel/src/log_compaction/tests.rs

@@ -237,11 +237,7 @@ fn test_no_compaction_staged_commits() {
 
     // Set up in-memory store
     let store = Arc::new(InMemory::new());
-    let engine = DefaultEngine::new(
-        store.clone(),
-        Arc::new(TokioBackgroundExecutor::new()),
-        None,
-    );
+    let engine = DefaultEngine::new(store.clone(), Arc::new(TokioBackgroundExecutor::new()));
 
     // Create basic commits with proper metadata and protocol
     use crate::actions::{Metadata, Protocol};

kernel/src/log_segment/tests.rs

@@ -960,11 +960,7 @@ fn test_checkpoint_batch_with_no_sidecars_returns_none() -> DeltaResult<()> {
 #[test]
 fn test_checkpoint_batch_with_sidecars_returns_sidecar_batches() -> DeltaResult<()> {
     let (store, log_root) = new_in_memory_store();
-    let engine = DefaultEngine::new(
-        store.clone(),
-        Arc::new(TokioBackgroundExecutor::new()),
-        None,
-    );
+    let engine = DefaultEngine::new(store.clone(), Arc::new(TokioBackgroundExecutor::new()));
     let read_schema = get_all_actions_schema().project(&[ADD_NAME, REMOVE_NAME, SIDECAR_NAME])?;
 
     add_sidecar_to_store(
@@ -1004,11 +1000,7 @@ fn test_checkpoint_batch_with_sidecars_returns_sidecar_batches() -> DeltaResult<
 #[test]
 fn test_checkpoint_batch_with_sidecar_files_that_do_not_exist() -> DeltaResult<()> {
     let (store, log_root) = new_in_memory_store();
-    let engine = DefaultEngine::new(
-        store.clone(),
-        Arc::new(TokioBackgroundExecutor::new()),
-        None,
-    );
+    let engine = DefaultEngine::new(store.clone(), Arc::new(TokioBackgroundExecutor::new()));
 
     let checkpoint_batch = sidecar_batch_with_given_paths(
         vec!["sidecarfile1.parquet", "sidecarfile2.parquet"],
@@ -1035,11 +1027,7 @@ fn test_checkpoint_batch_with_sidecar_files_that_do_not_exist() -> DeltaResult<(
 #[test]
 fn test_reading_sidecar_files_with_predicate() -> DeltaResult<()> {
     let (store, log_root) = new_in_memory_store();
-    let engine = DefaultEngine::new(
-        store.clone(),
-        Arc::new(TokioBackgroundExecutor::new()),
-        None,
-    );
+    let engine = DefaultEngine::new(store.clone(), Arc::new(TokioBackgroundExecutor::new()));
     let read_schema = get_all_actions_schema().project(&[ADD_NAME, REMOVE_NAME, SIDECAR_NAME])?;
 
     let checkpoint_batch =
@@ -1079,11 +1067,7 @@ fn test_reading_sidecar_files_with_predicate() -> DeltaResult<()> {
 fn test_create_checkpoint_stream_returns_checkpoint_batches_as_is_if_schema_has_no_file_actions(
 ) -> DeltaResult<()> {
     let (store, log_root) = new_in_memory_store();
-    let engine = DefaultEngine::new(
-        store.clone(),
-        Arc::new(TokioBackgroundExecutor::new()),
-        None,
-    );
+    let engine = DefaultEngine::new(store.clone(), Arc::new(TokioBackgroundExecutor::new()));
     add_checkpoint_to_store(
         &store,
         // Create a checkpoint batch with sidecar actions to verify that the sidecar actions are not read.
@@ -1130,11 +1114,7 @@ fn test_create_checkpoint_stream_returns_checkpoint_batches_as_is_if_schema_has_
 fn test_create_checkpoint_stream_returns_checkpoint_batches_if_checkpoint_is_multi_part(
 ) -> DeltaResult<()> {
     let (store, log_root) = new_in_memory_store();
-    let engine = DefaultEngine::new(
-        store.clone(),
-        Arc::new(TokioBackgroundExecutor::new()),
-        None,
-    );
+    let engine = DefaultEngine::new(store.clone(), Arc::new(TokioBackgroundExecutor::new()));
 
     // Multi-part checkpoints should never contain sidecar actions.
     // This test intentionally includes batches with sidecar actions in multi-part checkpoints
@@ -1202,11 +1182,7 @@ fn test_create_checkpoint_stream_returns_checkpoint_batches_if_checkpoint_is_mul
 fn test_create_checkpoint_stream_reads_parquet_checkpoint_batch_without_sidecars() -> DeltaResult<()>
 {
     let (store, log_root) = new_in_memory_store();
-    let engine = DefaultEngine::new(
-        store.clone(),
-        Arc::new(TokioBackgroundExecutor::new()),
-        None,
-    );
+    let engine = DefaultEngine::new(store.clone(), Arc::new(TokioBackgroundExecutor::new()));
 
     add_checkpoint_to_store(
         &store,
@@ -1249,11 +1225,7 @@ fn test_create_checkpoint_stream_reads_parquet_checkpoint_batch_without_sidecars
 #[test]
 fn test_create_checkpoint_stream_reads_json_checkpoint_batch_without_sidecars() -> DeltaResult<()> {
     let (store, log_root) = new_in_memory_store();
-    let engine = DefaultEngine::new(
-        store.clone(),
-        Arc::new(TokioBackgroundExecutor::new()),
-        None,
-    );
+    let engine = DefaultEngine::new(store.clone(), Arc::new(TokioBackgroundExecutor::new()));
 
     let filename = "00000000000000000010.checkpoint.80a083e8-7026-4e79-81be-64bd76c43a11.json";
 
@@ -1311,11 +1283,7 @@ fn test_create_checkpoint_stream_reads_json_checkpoint_batch_without_sidecars()
 fn test_create_checkpoint_stream_reads_checkpoint_file_and_returns_sidecar_batches(
 ) -> DeltaResult<()> {
     let (store, log_root) = new_in_memory_store();
-    let engine = DefaultEngine::new(
-        store.clone(),
-        Arc::new(TokioBackgroundExecutor::new()),
-        None,
-    );
+    let engine = DefaultEngine::new(store.clone(), Arc::new(TokioBackgroundExecutor::new()));
 
     add_checkpoint_to_store(
         &store,

kernel/src/path.rs

@@ -958,11 +958,7 @@ mod tests {
     #[tokio::test]
     async fn test_read_in_commit_timestamp_success() {
         let store = Arc::new(InMemory::new());
-        let engine = DefaultEngine::new(
-            store.clone(),
-            Arc::new(TokioBackgroundExecutor::new()),
-            None,
-        );
+        let engine = DefaultEngine::new(store.clone(), Arc::new(TokioBackgroundExecutor::new()));
         let table_url = url::Url::parse("memory://test/").unwrap();
 
         // Create a commit file with ICT using add_commit
@@ -991,11 +987,7 @@ mod tests {
     #[tokio::test]
     async fn test_read_in_commit_timestamp_missing_ict() {
         let store = Arc::new(InMemory::new());
-        let engine = DefaultEngine::new(
-            store.clone(),
-            Arc::new(TokioBackgroundExecutor::new()),
-            None,
-        );
+        let engine = DefaultEngine::new(store.clone(), Arc::new(TokioBackgroundExecutor::new()));
         let table_url = url::Url::parse("memory://test/").unwrap();
 
         // Create a commit file without ICT

kernel/src/scan/data_skipping/tests.rs

@@ -305,19 +305,19 @@ fn test_sql_where() {
 
     let pred = &Pred::is_not_null(col.clone());
     do_test(ALL_NULL, pred, PRESENT, Some(false), Some(false));
-    do_test(ALL_NULL, pred, MISSING, None);
+    do_test(ALL_NULL, pred, MISSING, None, None);
 
     // SQL WHERE allows a present-but-all-null column to be pruned, but not a missing column.
     let pred = &Pred::lt(col.clone(), VAL);
     do_test(NO_NULL, pred, PRESENT, Some(true), Some(true));
     do_test(SOME_NULL, pred, PRESENT, Some(true), Some(true));
     do_test(ALL_NULL, pred, PRESENT, None, Some(false));
-    do_test(ALL_NULL, pred, MISSING, None);
+    do_test(ALL_NULL, pred, MISSING, None, None);
 
     // Comparison inside AND works
     let pred = &Pred::and(TRUE, Pred::lt(VAL, col.clone()));
     do_test(ALL_NULL, pred, PRESENT, None, Some(false));
-    do_test(ALL_NULL, pred, MISSING, None);
+    do_test(ALL_NULL, pred, MISSING, None, None);
 
     // NULL inside AND allows static skipping under SQL semantics
     let pred = &Pred::and(NULL, Pred::lt(col.clone(), VAL));
@@ -327,17 +327,17 @@ fn test_sql_where() {
     // Comparison inside AND inside AND works
     let pred = &Pred::and(TRUE, Pred::and(TRUE, Pred::lt(col.clone(), VAL)));
     do_test(ALL_NULL, pred, PRESENT, None, Some(false));
-    do_test(ALL_NULL, pred, MISSING, None);
+    do_test(ALL_NULL, pred, MISSING, None, None);
 
     // Comparison inside OR works
     let pred = &Pred::or(FALSE, Pred::lt(col.clone(), VAL));
     do_test(ALL_NULL, pred, PRESENT, None, Some(false));
-    do_test(ALL_NULL, pred, MISSING, None);
+    do_test(ALL_NULL, pred, MISSING, None, None);
 
     // Comparison inside AND inside OR works
     let pred = &Pred::or(FALSE, Pred::and(TRUE, Pred::lt(col.clone(), VAL)));
     do_test(ALL_NULL, pred, PRESENT, None, Some(false));
-    do_test(ALL_NULL, pred, MISSING, None);
+    do_test(ALL_NULL, pred, MISSING, None, None);
 }
 
 // TODO(#1002): we currently don't support file skipping on timestamp columns' max stat since they