chore: use shared containers for integration tests (#924)

Closes #923

This PR makes integration tests re-use the same set of docker
containers, hence allowing them to be run concurrently

1. all existing tests are moved to a `shared_tests` module, with a
single top-level `shared.rs` file, thus requiring only [one
compilation](https://corrode.dev/blog/tips-for-faster-rust-compile-times/#combine-all-integration-tests-into-a-single-binary)
step (as opposed to compiling every test file individually)
2. `set_test_fixture` is made sync, since there's no async code in it
3. given 1 and 2, the test fixture is wrapped in a `OnceLock`, so that
it is invoked only once for the resulting (shared) test binary; it also
has a corresponding destructor to spin down the containers after the
tests run
4. if there's a need for integration tests which don't use the shared
docker containers, their test files should be placed alongside
`shared.rs` (top-level); otherwise they should be created in the
`shared_tests` module
5. one (shared) file can now contain an arbitrary number of tests that
utilize the shared set of containers (without having to resort to
decorating them with `#[serial]`)

Finally, with this change my integration test runs go from taking ~3.5
minutes down to 30 seconds.

EDIT: The CI `unit` workflow run duration also seems [~10min shorter
now](#924 (comment)).

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/shared.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_tests;
+
+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_tests/append_data_file_test.rs

@@ -17,60 +17,38 @@
 
 //! Integration tests for rest catalog.
 
-use std::collections::HashMap;
 use std::sync::Arc;
 
 use arrow_array::{ArrayRef, BooleanArray, Int32Array, RecordBatch, StringArray};
 use futures::TryStreamExt;
-use iceberg::spec::{NestedField, PrimitiveType, Schema, Type};
 use iceberg::transaction::Transaction;
 use iceberg::writer::base_writer::data_file_writer::DataFileWriterBuilder;
 use iceberg::writer::file_writer::location_generator::{
     DefaultFileNameGenerator, DefaultLocationGenerator,
 };
 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, TableCreation};
+use iceberg_catalog_rest::RestCatalog;
 use parquet::arrow::arrow_reader::ArrowReaderOptions;
 use parquet::file::properties::WriterProperties;
 
+use crate::get_shared_containers;
+use crate::shared_tests::{random_ns, test_schema};
+
 #[tokio::test]
 async fn test_append_data_file() {
-    let fixture = set_test_fixture("test_create_table").await;
-
-    let ns = Namespace::with_properties(
-        NamespaceIdent::from_strs(["apple", "ios"]).unwrap(),
-        HashMap::from([
-            ("owner".to_string(), "ray".to_string()),
-            ("community".to_string(), "apache".to_string()),
-        ]),
-    );
-
-    fixture
-        .rest_catalog
-        .create_namespace(ns.name(), ns.properties().clone())
-        .await
-        .unwrap();
-
-    let schema = Schema::builder()
-        .with_schema_id(1)
-        .with_identifier_field_ids(vec![2])
-        .with_fields(vec![
-            NestedField::optional(1, "foo", Type::Primitive(PrimitiveType::String)).into(),
-            NestedField::required(2, "bar", Type::Primitive(PrimitiveType::Int)).into(),
-            NestedField::optional(3, "baz", Type::Primitive(PrimitiveType::Boolean)).into(),
-        ])
-        .build()
-        .unwrap();
+    let fixture = get_shared_containers();
+    let rest_catalog = RestCatalog::new(fixture.catalog_config.clone());
+    let ns = random_ns().await;
+    let schema = test_schema();
 
     let table_creation = TableCreation::builder()
         .name("t1".to_string())
         .schema(schema.clone())
         .build();
 
-    let table = fixture
-        .rest_catalog
+    let table = rest_catalog
         .create_table(ns.name(), table_creation)
         .await
         .unwrap();
@@ -137,7 +115,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 +135,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_tests/append_partition_data_file_test.rs

@@ -17,15 +17,11 @@
 
 //! Integration test for partition data file
 
-use std::collections::HashMap;
 use std::sync::Arc;
 
 use arrow_array::{ArrayRef, BooleanArray, Int32Array, RecordBatch, StringArray};
 use futures::TryStreamExt;
-use iceberg::spec::{
-    Literal, NestedField, PrimitiveLiteral, PrimitiveType, Schema, Struct, Transform, Type,
-    UnboundPartitionSpec,
-};
+use iceberg::spec::{Literal, PrimitiveLiteral, Struct, Transform, UnboundPartitionSpec};
 use iceberg::table::Table;
 use iceberg::transaction::Transaction;
 use iceberg::writer::base_writer::data_file_writer::DataFileWriterBuilder;
@@ -34,38 +30,19 @@ 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, TableCreation};
+use iceberg_catalog_rest::RestCatalog;
 use parquet::file::properties::WriterProperties;
 
+use crate::get_shared_containers;
+use crate::shared_tests::{random_ns, test_schema};
+
 #[tokio::test]
 async fn test_append_partition_data_file() {
-    let fixture = set_test_fixture("test_partition_data_file").await;
-
-    let ns = Namespace::with_properties(
-        NamespaceIdent::from_strs(["iceberg", "rust"]).unwrap(),
-        HashMap::from([
-            ("owner".to_string(), "ray".to_string()),
-            ("community".to_string(), "apache".to_string()),
-        ]),
-    );
-
-    fixture
-        .rest_catalog
-        .create_namespace(ns.name(), ns.properties().clone())
-        .await
-        .unwrap();
-
-    let schema = Schema::builder()
-        .with_schema_id(1)
-        .with_identifier_field_ids(vec![2])
-        .with_fields(vec![
-            NestedField::optional(1, "foo", Type::Primitive(PrimitiveType::String)).into(),
-            NestedField::required(2, "bar", Type::Primitive(PrimitiveType::Int)).into(),
-            NestedField::optional(3, "baz", Type::Primitive(PrimitiveType::Boolean)).into(),
-        ])
-        .build()
-        .unwrap();
+    let fixture = get_shared_containers();
+    let rest_catalog = RestCatalog::new(fixture.catalog_config.clone());
+    let ns = random_ns().await;
+    let schema = test_schema();
 
     let unbound_partition_spec = UnboundPartitionSpec::builder()
         .add_partition_field(2, "id", Transform::Identity)
@@ -82,8 +59,7 @@ async fn test_append_partition_data_file() {
         .partition_spec(partition_spec)
         .build();
 
-    let table = fixture
-        .rest_catalog
+    let table = rest_catalog
         .create_table(ns.name(), table_creation)
         .await
         .unwrap();
@@ -148,7 +124,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