Make integration tests run concurrently

By extension make them use the same set of docker containers.

crates/catalog/glue/tests/glue_catalog_test.rs

@@ -44,7 +44,7 @@ fn before_all() {
         normalize_test_name(module_path!()),
         format!("{}/testdata/glue_catalog", env!("CARGO_MANIFEST_DIR")),
     );
-    docker_compose.run();
+    docker_compose.up();
     guard.replace(docker_compose);
 }
 

crates/catalog/hms/tests/hms_catalog_test.rs

@@ -43,7 +43,7 @@ fn before_all() {
         normalize_test_name(module_path!()),
         format!("{}/testdata/hms_catalog", env!("CARGO_MANIFEST_DIR")),
     );
-    docker_compose.run();
+    docker_compose.up();
     guard.replace(docker_compose);
 }
 

crates/catalog/rest/tests/rest_catalog_test.rs

@@ -41,7 +41,7 @@ fn before_all() {
         normalize_test_name(module_path!()),
         format!("{}/testdata/rest_catalog", env!("CARGO_MANIFEST_DIR")),
     );
-    docker_compose.run();
+    docker_compose.up();
     guard.replace(docker_compose);
 }
 

crates/iceberg/tests/file_io_gcs_test.rs

@@ -40,7 +40,7 @@ mod tests {
             normalize_test_name(module_path!()),
             format!("{}/testdata/file_io_gcs", env!("CARGO_MANIFEST_DIR")),
         );
-        docker_compose.run();
+        docker_compose.up();
         guard.replace(docker_compose);
     }
 

crates/iceberg/tests/file_io_s3_test.rs

@@ -38,7 +38,7 @@ mod tests {
             normalize_test_name(module_path!()),
             format!("{}/testdata/file_io_s3", env!("CARGO_MANIFEST_DIR")),
         );
-        docker_compose.run();
+        docker_compose.up();
         guard.replace(docker_compose);
     }
 

crates/integration_tests/Cargo.toml

@@ -27,6 +27,7 @@ rust-version = { workspace = true }
 [dependencies]
 arrow-array = { workspace = true }
 arrow-schema = { workspace = true }
+ctor = { workspace = true }
 datafusion = { workspace = true }
 futures = { workspace = true }
 iceberg = { workspace = true }

crates/integration_tests/src/lib.rs

@@ -18,31 +18,32 @@
 use std::collections::HashMap;
 
 use iceberg::io::{S3_ACCESS_KEY_ID, S3_ENDPOINT, S3_REGION, S3_SECRET_ACCESS_KEY};
-use iceberg_catalog_rest::{RestCatalog, RestCatalogConfig};
+use iceberg_catalog_rest::RestCatalogConfig;
 use iceberg_test_utils::docker::DockerCompose;
 use iceberg_test_utils::{normalize_test_name, set_up};
 
 const REST_CATALOG_PORT: u16 = 8181;
 
 pub struct TestFixture {
     pub _docker_compose: DockerCompose,
-    pub rest_catalog: RestCatalog,
+    pub catalog_config: RestCatalogConfig,
 }
 
-pub async fn set_test_fixture(func: &str) -> TestFixture {
+pub fn set_test_fixture(func: &str) -> TestFixture {
     set_up();
     let docker_compose = DockerCompose::new(
         normalize_test_name(format!("{}_{func}", module_path!())),
         format!("{}/testdata", env!("CARGO_MANIFEST_DIR")),
     );
 
-    // Start docker compose
-    docker_compose.run();
+    // Stop any containers from previous runs and start new ones
+    docker_compose.down();
+    docker_compose.up();
 
     let rest_catalog_ip = docker_compose.get_container_ip("rest");
     let minio_ip = docker_compose.get_container_ip("minio");
 
-    let config = RestCatalogConfig::builder()
+    let catalog_config = RestCatalogConfig::builder()
         .uri(format!("http://{}:{}", rest_catalog_ip, REST_CATALOG_PORT))
         .props(HashMap::from([
             (
@@ -54,10 +55,9 @@ pub async fn set_test_fixture(func: &str) -> TestFixture {
             (S3_REGION.to_string(), "us-east-1".to_string()),
         ]))
         .build();
-    let rest_catalog = RestCatalog::new(config);
 
     TestFixture {
         _docker_compose: docker_compose,
-        rest_catalog,
+        catalog_config,
     }
 }

crates/integration_tests/tests/main.rs

@@ -0,0 +1,36 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+use std::sync::{Arc, OnceLock};
+
+use ctor::dtor;
+use iceberg_integration_tests::{set_test_fixture, TestFixture};
+
+pub mod shared;
+
+static DOCKER_CONTAINERS: OnceLock<Arc<TestFixture>> = OnceLock::new();
+
+pub fn get_shared_containers() -> &'static Arc<TestFixture> {
+    DOCKER_CONTAINERS.get_or_init(|| Arc::new(set_test_fixture("shared_tests")))
+}
+
+#[dtor]
+fn shutdown() {
+    if let Some(fixture) = DOCKER_CONTAINERS.get() {
+        fixture._docker_compose.down()
+    }
+}

crates/integration_tests/tests/append_data_file_test.rs → crates/integration_tests/tests/shared/append_data_file_test.rs

@@ -31,13 +31,16 @@ use iceberg::writer::file_writer::location_generator::{
 use iceberg::writer::file_writer::ParquetWriterBuilder;
 use iceberg::writer::{IcebergWriter, IcebergWriterBuilder};
 use iceberg::{Catalog, Namespace, NamespaceIdent, TableCreation};
-use iceberg_integration_tests::set_test_fixture;
+use iceberg_catalog_rest::RestCatalog;
 use parquet::arrow::arrow_reader::ArrowReaderOptions;
 use parquet::file::properties::WriterProperties;
 
+use crate::get_shared_containers;
+
 #[tokio::test]
 async fn test_append_data_file() {
-    let fixture = set_test_fixture("test_create_table").await;
+    let fixture = get_shared_containers();
+    let rest_catalog = RestCatalog::new(fixture.catalog_config.clone());
 
     let ns = Namespace::with_properties(
         NamespaceIdent::from_strs(["apple", "ios"]).unwrap(),
@@ -47,11 +50,9 @@ async fn test_append_data_file() {
         ]),
     );
 
-    fixture
-        .rest_catalog
+    let _ = rest_catalog
         .create_namespace(ns.name(), ns.properties().clone())
-        .await
-        .unwrap();
+        .await;
 
     let schema = Schema::builder()
         .with_schema_id(1)
@@ -69,8 +70,7 @@ async fn test_append_data_file() {
         .schema(schema.clone())
         .build();
 
-    let table = fixture
-        .rest_catalog
+    let table = rest_catalog
         .create_table(ns.name(), table_creation)
         .await
         .unwrap();
@@ -137,7 +137,7 @@ async fn test_append_data_file() {
     let mut append_action = tx.fast_append(None, vec![]).unwrap();
     append_action.add_data_files(data_file.clone()).unwrap();
     let tx = append_action.apply().await.unwrap();
-    let table = tx.commit(&fixture.rest_catalog).await.unwrap();
+    let table = tx.commit(&rest_catalog).await.unwrap();
 
     // check result
     let batch_stream = table
@@ -157,7 +157,7 @@ async fn test_append_data_file() {
     let mut append_action = tx.fast_append(None, vec![]).unwrap();
     append_action.add_data_files(data_file.clone()).unwrap();
     let tx = append_action.apply().await.unwrap();
-    let table = tx.commit(&fixture.rest_catalog).await.unwrap();
+    let table = tx.commit(&rest_catalog).await.unwrap();
 
     // check result again
     let batch_stream = table

crates/integration_tests/tests/append_partition_data_file_test.rs → crates/integration_tests/tests/shared/append_partition_data_file_test.rs

@@ -35,12 +35,15 @@ use iceberg::writer::file_writer::location_generator::{
 use iceberg::writer::file_writer::ParquetWriterBuilder;
 use iceberg::writer::{IcebergWriter, IcebergWriterBuilder};
 use iceberg::{Catalog, Namespace, NamespaceIdent, TableCreation};
-use iceberg_integration_tests::set_test_fixture;
+use iceberg_catalog_rest::RestCatalog;
 use parquet::file::properties::WriterProperties;
 
+use crate::get_shared_containers;
+
 #[tokio::test]
 async fn test_append_partition_data_file() {
-    let fixture = set_test_fixture("test_partition_data_file").await;
+    let fixture = get_shared_containers();
+    let rest_catalog = RestCatalog::new(fixture.catalog_config.clone());
 
     let ns = Namespace::with_properties(
         NamespaceIdent::from_strs(["iceberg", "rust"]).unwrap(),
@@ -50,11 +53,9 @@ async fn test_append_partition_data_file() {
         ]),
     );
 
-    fixture
-        .rest_catalog
+    let _ = rest_catalog
         .create_namespace(ns.name(), ns.properties().clone())
-        .await
-        .unwrap();
+        .await;
 
     let schema = Schema::builder()
         .with_schema_id(1)
@@ -77,13 +78,12 @@ async fn test_append_partition_data_file() {
         .expect("could not bind to schema");
 
     let table_creation = TableCreation::builder()
-        .name("t1".to_string())
+        .name("t2".to_string())
         .schema(schema.clone())
         .partition_spec(partition_spec)
         .build();
 
-    let table = fixture
-        .rest_catalog
+    let table = rest_catalog
         .create_table(ns.name(), table_creation)
         .await
         .unwrap();
@@ -148,7 +148,7 @@ async fn test_append_partition_data_file() {
         .add_data_files(data_file_valid.clone())
         .unwrap();
     let tx = append_action.apply().await.unwrap();
-    let table = tx.commit(&fixture.rest_catalog).await.unwrap();
+    let table = tx.commit(&rest_catalog).await.unwrap();
 
     // check result
     let batch_stream = table

crates/integration_tests/tests/conflict_commit_test.rs → crates/integration_tests/tests/shared/conflict_commit_test.rs

@@ -31,12 +31,15 @@ use iceberg::writer::file_writer::location_generator::{
 use iceberg::writer::file_writer::ParquetWriterBuilder;
 use iceberg::writer::{IcebergWriter, IcebergWriterBuilder};
 use iceberg::{Catalog, Namespace, NamespaceIdent, TableCreation};
-use iceberg_integration_tests::set_test_fixture;
+use iceberg_catalog_rest::RestCatalog;
 use parquet::file::properties::WriterProperties;
 
+use crate::get_shared_containers;
+
 #[tokio::test]
 async fn test_append_data_file_conflict() {
-    let fixture = set_test_fixture("test_create_table").await;
+    let fixture = get_shared_containers();
+    let rest_catalog = RestCatalog::new(fixture.catalog_config.clone());
 
     let ns = Namespace::with_properties(
         NamespaceIdent::from_strs(["apple", "ios"]).unwrap(),
@@ -46,11 +49,9 @@ async fn test_append_data_file_conflict() {
         ]),
     );
 
-    fixture
-        .rest_catalog
+    let _ = rest_catalog
         .create_namespace(ns.name(), ns.properties().clone())
-        .await
-        .unwrap();
+        .await;
 
     let schema = Schema::builder()
         .with_schema_id(1)
@@ -64,12 +65,11 @@ async fn test_append_data_file_conflict() {
         .unwrap();
 
     let table_creation = TableCreation::builder()
-        .name("t1".to_string())
+        .name("t3".to_string())
         .schema(schema.clone())
         .build();
 
-    let table = fixture
-        .rest_catalog
+    let table = rest_catalog
         .create_table(ns.name(), table_creation)
         .await
         .unwrap();
@@ -120,7 +120,7 @@ async fn test_append_data_file_conflict() {
     append_action.add_data_files(data_file.clone()).unwrap();
     let tx2 = append_action.apply().await.unwrap();
     let table = tx2
-        .commit(&fixture.rest_catalog)
+        .commit(&rest_catalog)
         .await
         .expect("The first commit should not fail.");
 
@@ -138,5 +138,5 @@ async fn test_append_data_file_conflict() {
     assert_eq!(batches[0], batch);
 
     // another commit should fail
-    assert!(tx1.commit(&fixture.rest_catalog).await.is_err());
+    assert!(tx1.commit(&rest_catalog).await.is_err());
 }

crates/integration_tests/tests/datafusion.rs → crates/integration_tests/tests/shared/datafusion.rs

@@ -25,17 +25,18 @@ use datafusion::catalog::TableProvider;
 use datafusion::error::DataFusionError;
 use datafusion::prelude::SessionContext;
 use iceberg::{Catalog, TableIdent};
+use iceberg_catalog_rest::RestCatalog;
 use iceberg_datafusion::IcebergTableProvider;
-use iceberg_integration_tests::set_test_fixture;
 use parquet::arrow::PARQUET_FIELD_ID_META_KEY;
 
+use crate::get_shared_containers;
+
 #[tokio::test]
 async fn test_basic_queries() -> Result<(), DataFusionError> {
-    let fixture = set_test_fixture("datafusion_basic_read").await;
-
-    let catalog = fixture.rest_catalog;
+    let fixture = get_shared_containers();
+    let rest_catalog = RestCatalog::new(fixture.catalog_config.clone());
 
-    let table = catalog
+    let table = rest_catalog
         .load_table(&TableIdent::from_strs(["default", "types_test"]).unwrap())
         .await
         .unwrap();

crates/integration_tests/tests/shared/mod.rs

@@ -0,0 +1,24 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+mod append_data_file_test;
+mod append_partition_data_file_test;
+mod conflict_commit_test;
+mod datafusion;
+mod read_evolved_schema;
+mod read_positional_deletes;
+mod scan_all_type;