test: avoid circular dependency with core/test crates

Signed-off-by: Robert Pack <robstar.pack@gmail.com>

Author: roeap

crates/core/Cargo.toml

@@ -99,7 +99,7 @@ humantime = { version = "2.1.0" }
 criterion = "0.5"
 ctor = "0"
 datatest-stable = "0.2"
-deltalake-test = { path = "../test", features = ["datafusion"] }
+deltalake-test = { path = "../test" }
 dotenvy = "0"
 fs_extra = "1.2.0"
 maplit = "1"
@@ -133,6 +133,8 @@ cloud = ["object_store/cloud"]
 # enable caching some file I/O operations when scanning delta logs
 delta-cache = ["foyer", "tempfile", "url/serde"]
 
+integration_test = []
+
 [[test]]
 name = "dat"
 harness = false

crates/core/src/kernel/scalars.rs

@@ -9,7 +9,7 @@ use delta_kernel::{
     schema::StructField,
 };
 use object_store::path::Path;
-#[cfg(test)]
+#[cfg(any(test, feature = "integration_test"))]
 use serde_json::Value;
 use urlencoding::encode;
 
@@ -24,7 +24,7 @@ pub trait ScalarExt: Sized {
     /// Create a [`Scalar`] from an arrow array row
     fn from_array(arr: &dyn Array, index: usize) -> Option<Self>;
     /// Serialize as serde_json::Value
-    #[cfg(test)]
+    #[cfg(any(test, feature = "integration_test"))]
     fn to_json(&self) -> serde_json::Value;
 }
 
@@ -235,7 +235,7 @@ impl ScalarExt for Scalar {
     }
 
     /// Serializes this scalar as a serde_json::Value.
-    #[cfg(test)]
+    #[cfg(any(test, feature = "integration_test"))]
     fn to_json(&self) -> serde_json::Value {
         match self {
             Self::String(s) => Value::String(s.to_owned()),

crates/core/src/kernel/snapshot/log_segment.rs

@@ -579,7 +579,6 @@ pub(super) async fn list_log_files(
 #[cfg(test)]
 pub(super) mod tests {
     use delta_kernel::table_features::{ReaderFeature, WriterFeature};
-    use deltalake_test::utils::*;
     use maplit::hashset;
     use tokio::task::JoinHandle;
 
@@ -588,22 +587,23 @@ pub(super) mod tests {
         kernel::transaction::{CommitBuilder, TableReference},
         kernel::{Action, Add, Format, Remove},
         protocol::{DeltaOperation, SaveMode},
+        test_utils::{TestResult, TestTables},
         DeltaTableBuilder,
     };
 
     use super::*;
 
-    pub(crate) async fn test_log_segment(context: &IntegrationContext) -> TestResult {
-        read_log_files(context).await?;
-        read_metadata(context).await?;
-        log_segment_serde(context).await?;
+    pub(crate) async fn test_log_segment() -> TestResult {
+        read_log_files().await?;
+        read_metadata().await?;
+        log_segment_serde().await?;
 
         Ok(())
     }
 
-    async fn log_segment_serde(context: &IntegrationContext) -> TestResult {
-        let store = context
-            .table_builder(TestTables::Simple)
+    async fn log_segment_serde() -> TestResult {
+        let store = TestTables::Simple
+            .table_builder()
             .build_storage()?
             .object_store(None);
 
@@ -620,9 +620,9 @@ pub(super) mod tests {
         Ok(())
     }
 
-    async fn read_log_files(context: &IntegrationContext) -> TestResult {
-        let store = context
-            .table_builder(TestTables::SimpleWithCheckpoint)
+    async fn read_log_files() -> TestResult {
+        let store = TestTables::SimpleWithCheckpoint
+            .table_builder()
             .build_storage()?
             .object_store(None);
 
@@ -654,8 +654,8 @@ pub(super) mod tests {
         assert_eq!(segment.commit_files.len(), 9);
         assert_eq!(segment.checkpoint_files.len(), 0);
 
-        let store = context
-            .table_builder(TestTables::Simple)
+        let store = TestTables::Simple
+            .table_builder()
             .build_storage()?
             .object_store(None);
 
@@ -670,9 +670,9 @@ pub(super) mod tests {
         Ok(())
     }
 
-    async fn read_metadata(context: &IntegrationContext) -> TestResult {
-        let store = context
-            .table_builder(TestTables::WithDvSmall)
+    async fn read_metadata() -> TestResult {
+        let store = TestTables::WithDvSmall
+            .table_builder()
             .build_storage()?
             .object_store(None);
         let segment = LogSegment::try_new(&Path::default(), None, store.as_ref()).await?;
@@ -692,16 +692,13 @@ pub(super) mod tests {
         Ok(())
     }
 
-    pub(crate) async fn concurrent_checkpoint(context: &IntegrationContext) -> TestResult {
-        context
-            .load_table(TestTables::LatestNotCheckpointed)
-            .await?;
-        let table_to_checkpoint = context
-            .table_builder(TestTables::LatestNotCheckpointed)
+    pub(crate) async fn concurrent_checkpoint() -> TestResult {
+        let table_to_checkpoint = TestTables::LatestNotCheckpointed
+            .table_builder()
             .load()
             .await?;
-        let store = context
-            .table_builder(TestTables::LatestNotCheckpointed)
+        let store = TestTables::LatestNotCheckpointed
+            .table_builder()
             .build_storage()?
             .object_store(None);
         let slow_list_store = Arc::new(slow_store::SlowListStore { store });

crates/core/src/kernel/snapshot/mod.rs

@@ -816,44 +816,34 @@ fn find_nested_field<'a>(
 mod tests {
     use std::collections::HashMap;
 
-    use chrono::Utc;
-    use deltalake_test::utils::*;
     use futures::TryStreamExt;
     use itertools::Itertools;
 
     use super::log_segment::tests::{concurrent_checkpoint, test_log_segment};
     use super::replay::tests::test_log_replay;
     use super::*;
-    use crate::kernel::Remove;
     use crate::protocol::{DeltaOperation, SaveMode};
-    use crate::test_utils::ActionFactory;
+    use crate::test_utils::{assert_batches_sorted_eq, ActionFactory, TestResult, TestTables};
 
     #[tokio::test]
     async fn test_snapshots() -> TestResult {
-        let context = IntegrationContext::new(Box::<LocalStorageIntegration>::default())?;
-        context.load_table(TestTables::Checkpoints).await?;
-        context.load_table(TestTables::Simple).await?;
-        context.load_table(TestTables::SimpleWithCheckpoint).await?;
-        context.load_table(TestTables::WithDvSmall).await?;
-
-        test_log_segment(&context).await?;
-        test_log_replay(&context).await?;
-        test_snapshot(&context).await?;
-        test_eager_snapshot(&context).await?;
+        test_log_segment().await?;
+        test_log_replay().await?;
+        test_snapshot().await?;
+        test_eager_snapshot().await?;
 
         Ok(())
     }
 
     #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
     async fn test_concurrent_checkpoint() -> TestResult {
-        let context = IntegrationContext::new(Box::<LocalStorageIntegration>::default())?;
-        concurrent_checkpoint(&context).await?;
+        concurrent_checkpoint().await?;
         Ok(())
     }
 
-    async fn test_snapshot(context: &IntegrationContext) -> TestResult {
-        let store = context
-            .table_builder(TestTables::Simple)
+    async fn test_snapshot() -> TestResult {
+        let store = TestTables::Simple
+            .table_builder()
             .build_storage()?
             .object_store(None);
 
@@ -900,8 +890,8 @@ mod tests {
         ];
         assert_batches_sorted_eq!(expected, &batches);
 
-        let store = context
-            .table_builder(TestTables::Checkpoints)
+        let store = TestTables::Checkpoints
+            .table_builder()
             .build_storage()?
             .object_store(None);
 
@@ -924,9 +914,9 @@ mod tests {
         Ok(())
     }
 
-    async fn test_eager_snapshot(context: &IntegrationContext) -> TestResult {
-        let store = context
-            .table_builder(TestTables::Simple)
+    async fn test_eager_snapshot() -> TestResult {
+        let store = TestTables::Simple
+            .table_builder()
             .build_storage()?
             .object_store(None);
 
@@ -942,8 +932,8 @@ mod tests {
         let expected: StructType = serde_json::from_str(schema_string)?;
         assert_eq!(snapshot.schema(), &expected);
 
-        let store = context
-            .table_builder(TestTables::Checkpoints)
+        let store = TestTables::Checkpoints
+            .table_builder()
             .build_storage()?
             .object_store(None);
 
@@ -962,76 +952,6 @@ mod tests {
         Ok(())
     }
 
-    #[tokio::test]
-    async fn test_eager_snapshot_advance() -> TestResult {
-        let context = IntegrationContext::new(Box::<LocalStorageIntegration>::default())?;
-        context.load_table(TestTables::Simple).await?;
-
-        let store = context
-            .table_builder(TestTables::Simple)
-            .build_storage()?
-            .object_store(None);
-
-        let mut snapshot =
-            EagerSnapshot::try_new(&Path::default(), store.clone(), Default::default(), None)
-                .await?;
-
-        let version = snapshot.version();
-
-        let files = snapshot.file_actions()?.enumerate().collect_vec();
-        let num_files = files.len();
-
-        let split = files.split(|(idx, _)| *idx == num_files / 2).collect_vec();
-        assert!(split.len() == 2 && !split[0].is_empty() && !split[1].is_empty());
-        let (first, second) = split.into_iter().next_tuple().unwrap();
-
-        let removes = first
-            .iter()
-            .map(|(_, add)| {
-                Remove {
-                    path: add.path.clone(),
-                    size: Some(add.size),
-                    data_change: add.data_change,
-                    deletion_timestamp: Some(Utc::now().timestamp_millis()),
-                    extended_file_metadata: Some(true),
-                    partition_values: Some(add.partition_values.clone()),
-                    tags: add.tags.clone(),
-                    deletion_vector: add.deletion_vector.clone(),
-                    base_row_id: add.base_row_id,
-                    default_row_commit_version: add.default_row_commit_version,
-                }
-                .into()
-            })
-            .collect_vec();
-
-        let operation = DeltaOperation::Write {
-            mode: SaveMode::Append,
-            partition_by: None,
-            predicate: None,
-        };
-
-        let actions = vec![CommitData::new(
-            removes,
-            operation,
-            HashMap::new(),
-            Vec::new(),
-        )];
-
-        let new_version = snapshot.advance(&actions)?;
-        assert_eq!(new_version, version + 1);
-
-        let new_files = snapshot.file_actions()?.map(|f| f.path).collect::<Vec<_>>();
-        assert_eq!(new_files.len(), num_files - first.len());
-        assert!(first
-            .iter()
-            .all(|(_, add)| { !new_files.contains(&add.path) }));
-        assert!(second
-            .iter()
-            .all(|(_, add)| { new_files.contains(&add.path) }));
-
-        Ok(())
-    }
-
     #[test]
     fn test_partition_schema() {
         let schema = StructType::new(vec![

crates/core/src/kernel/snapshot/replay.rs

@@ -603,7 +603,6 @@ pub(super) mod tests {
 
     use arrow_select::concat::concat_batches;
     use delta_kernel::schema::DataType;
-    use deltalake_test::utils::*;
     use futures::TryStreamExt;
     use object_store::path::Path;
 
@@ -613,16 +612,16 @@ pub(super) mod tests {
     use crate::kernel::{models::ActionType, StructType};
     use crate::protocol::DeltaOperation;
     use crate::table::config::TableConfig;
-    use crate::test_utils::{ActionFactory, TestResult, TestSchemas};
+    use crate::test_utils::{ActionFactory, TestResult, TestSchemas, TestTables};
 
-    pub(crate) async fn test_log_replay(context: &IntegrationContext) -> TestResult {
+    pub(crate) async fn test_log_replay() -> TestResult {
         let log_schema = Arc::new(StructType::new(vec![
             ActionType::Add.schema_field().clone(),
             ActionType::Remove.schema_field().clone(),
         ]));
 
-        let store = context
-            .table_builder(TestTables::SimpleWithCheckpoint)
+        let store = TestTables::SimpleWithCheckpoint
+            .table_builder()
             .build_storage()?
             .object_store(None);
 
@@ -643,8 +642,8 @@ pub(super) mod tests {
         let filtered = scanner.process_files_batch(&batch, true)?;
         assert_eq!(filtered.schema().fields().len(), 1);
 
-        let store = context
-            .table_builder(TestTables::Simple)
+        let store = TestTables::Simple
+            .table_builder()
             .build_storage()?
             .object_store(None);
         let segment = LogSegment::try_new(&Path::default(), None, store.as_ref()).await?;

crates/core/src/lib.rs

@@ -77,7 +77,7 @@ pub mod protocol;
 pub mod schema;
 pub mod table;
 
-#[cfg(test)]
+#[cfg(any(test, feature = "integration_test"))]
 pub mod test_utils;
 
 #[cfg(feature = "datafusion")]