This is an automated email from the ASF dual-hosted git repository. github-merge-queue[bot] pushed a commit to branch gh-readonly-queue/main/pr-25092-4f0faac25d218d75a9a9583a96a31b2ca8340878 in repository https://gitbox.apache.org/repos/asf/datafusion.git
commit 1ec9ede5da5c67cbc2d3e0110a7ddbc2a47796a6 Author: Oleks V <[email protected]> AuthorDate: Thu Sep 10 17:42:45 2026 +0000 minor: make MinIO tests more stable (#25092) ## Rationale for this change I observed PRs sometimes got kicked out of merge queue because flaky `datafusion-cli` tests, which require full CI again The `datafusion-cli` storage integration tests (`test_cli`, `test_aws_options`, `test_s3_url_fallback`, `test_object_store_profiling`) each start their own MinIO container, so a single CI run pulls the image several times concurrently. Any transient Docker failure during that pull fails the test immediately: ``` thread 'test_aws_options' panicked at datafusion-cli/tests/cli_integration.rs:607:25: called `Result::unwrap()` on an `Err` value: "Failed to start MinIO container. Ensure Docker is running and accessible: failed to pull the image 'minio/minio:RELEASE.2025-02-28T09-55-16Z', error: bytes remaining on stream" ``` These errors are transient and clear on a second attempt, so a flaky pull should not fail the job. ## Rationale for this change The `datafusion-cli` storage integration tests (`test_cli`, `test_aws_options`, `test_s3_url_fallback`, `test_object_store_profiling`) each start their own MinIO container, so a single CI run pulls the image several times concurrently. Any transient Docker failure during that pull fails the test immediately: ``` thread 'test_aws_options' panicked at datafusion-cli/tests/cli_integration.rs:607:25: called `Result::unwrap()` on an `Err` value: "Failed to start MinIO container. Ensure Docker is running and accessible: failed to pull the image 'minio/minio:RELEASE.2025-02-28T09-55-16Z', error: bytes remaining on stream" ``` These errors are transient and clear on a second attempt, so a flaky pull should not fail the job. Co-authored-by: Andrew Lamb <[email protected]> --- .github/workflows/rust.yml | 15 ++ datafusion-cli/tests/cli_integration.rs | 294 ++++++++++++++++++++------------ 2 files changed, 204 insertions(+), 105 deletions(-) diff --git a/.github/workflows/rust.yml b/.github/workflows/rust.yml index 86fb33a82c..e51215f991 100644 --- a/.github/workflows/rust.yml +++ b/.github/workflows/rust.yml @@ -395,6 +395,21 @@ jobs: with: save-if: false # set in linux-test shared-key: "amd-ci" + # The storage integration tests start MinIO containers. Pulling the image + # once up front keeps the pull off the critical path of the tests, which + # would otherwise pull it several times concurrently and occasionally fail + # with transient Docker transport errors. The tests retry the pull + # themselves, so a failure here is only a warning. + # + # MINIO_IMAGE must match the image used by the `minio` module of the + # `testcontainers-modules` crate. The `minio_image_matches_ci_prepull` + # test in `datafusion-cli/tests/cli_integration.rs` fails if it drifts. + - name: Pre-pull MinIO image + env: + MINIO_IMAGE: minio/minio:RELEASE.2025-02-28T09-55-16Z + run: | + ci/scripts/retry timeout 120 docker pull "$MINIO_IMAGE" \ + || echo "::warning::Could not pre-pull $MINIO_IMAGE, the tests will pull it themselves" - name: Run tests (excluding doctests) env: RUST_BACKTRACE: 1 diff --git a/datafusion-cli/tests/cli_integration.rs b/datafusion-cli/tests/cli_integration.rs index 9bfeb65278..a6591eaea8 100644 --- a/datafusion-cli/tests/cli_integration.rs +++ b/datafusion-cli/tests/cli_integration.rs @@ -24,12 +24,13 @@ use insta::internals::SettingsBindDropGuard; use insta::{Settings, glob}; use insta_cmd::{assert_cmd_snapshot, get_cargo_bin}; use std::path::PathBuf; +use std::time::Duration; use std::{env, fs}; use testcontainers_modules::minio; use testcontainers_modules::testcontainers::core::{CmdWaitFor, ExecCommand, Mount}; use testcontainers_modules::testcontainers::runners::AsyncRunner; use testcontainers_modules::testcontainers::{ - ContainerAsync, ImageExt, TestcontainersError, + ContainerAsync, Image, ImageExt, TestcontainersError, }; fn cli() -> Command { @@ -45,10 +46,110 @@ fn make_settings() -> Settings { settings } +const MINIO_ROOT_USER: &str = "TEST-DataFusionLogin"; +const MINIO_ROOT_PASSWORD: &str = "TEST-DataFusionPassword"; + +/// How many times to try bringing up the MinIO container before failing. +/// +/// Both the Docker Hub image pull and the `mc` calls that provision the bucket +/// fail intermittently on CI with transient errors such as +/// `bytes remaining on stream`. Retrying is much cheaper than a flaky run. +const MINIO_SETUP_ATTEMPTS: u32 = 3; + +/// Delay before the first retry of the MinIO setup, doubled on each attempt. +const MINIO_SETUP_RETRY_DELAY: Duration = Duration::from_secs(5); + +/// Time budget for a single MinIO setup attempt. A stalled image pull or `mc` +/// invocation is retried instead of hanging the whole test run. +const MINIO_SETUP_TIMEOUT: Duration = Duration::from_mins(3); + +/// Starts a MinIO container preloaded with the test data, retrying transient +/// Docker failures. +/// +/// Returns `None` when the test should be skipped, that is when +/// `TEST_STORAGE_INTEGRATION` is unset or Docker Hub is rate limiting the image +/// pull. Panics if the container cannot be started for any other reason. +async fn start_minio_or_skip() -> Option<ContainerAsync<minio::MinIO>> { + if env::var("TEST_STORAGE_INTEGRATION").is_err() { + eprintln!("Skipping external storages integration tests"); + return None; + } + + match setup_minio_container().await { + Ok(container) => Some(container), + Err(e) if is_docker_pull_rate_limit(&e) => { + eprintln!("Skipping test: Docker pull rate limit reached: {e}"); + None + } + Err(e) => panic!("{e}"), + } +} + +/// A Docker Hub pull rate limit does not clear up within a test run, so the +/// affected tests are skipped rather than retried. +fn is_docker_pull_rate_limit(error: &str) -> bool { + error.contains("toomanyrequests") +} + +/// Retrying only pays off for transient failures. An exhausted pull quota or a +/// Docker daemon that cannot be reached at all stays broken for the whole run. +fn is_retryable(error: &str) -> bool { + !is_docker_pull_rate_limit(error) + && !error.contains("failed to initialize a docker client") +} + async fn setup_minio_container() -> Result<ContainerAsync<minio::MinIO>, String> { - const MINIO_ROOT_USER: &str = "TEST-DataFusionLogin"; - const MINIO_ROOT_PASSWORD: &str = "TEST-DataFusionPassword"; + let mut delay = MINIO_SETUP_RETRY_DELAY; + let mut last_error = String::from("MinIO container setup was not attempted at all"); + for attempt in 1..=MINIO_SETUP_ATTEMPTS { + last_error = match tokio::time::timeout( + MINIO_SETUP_TIMEOUT, + try_setup_minio_container(), + ) + .await + { + Ok(Ok(container)) => return Ok(container), + Ok(Err(e)) => e, + Err(_) => format!( + "Timed out after {MINIO_SETUP_TIMEOUT:?} while starting the MinIO container" + ), + }; + + if attempt == MINIO_SETUP_ATTEMPTS || !is_retryable(&last_error) { + break; + } + + eprintln!( + "MinIO container setup failed (attempt {attempt}/{MINIO_SETUP_ATTEMPTS}), \ + retrying in {delay:?}: {last_error}" + ); + tokio::time::sleep(delay).await; + delay *= 2; + } + + Err(last_error) +} + +/// A single attempt at starting and provisioning a MinIO container. +/// +/// The container is removed again if provisioning fails, so that the next +/// attempt starts from a clean state. +async fn try_setup_minio_container() -> Result<ContainerAsync<minio::MinIO>, String> { + let container = start_minio_container().await?; + + match provision_minio_container(&container).await { + Ok(()) => Ok(container), + Err(e) => { + if let Err(rm_error) = container.rm().await { + eprintln!("Failed to remove the MinIO container: {rm_error}"); + } + Err(e) + } + } +} + +async fn start_minio_container() -> Result<ContainerAsync<minio::MinIO>, String> { let data_path = PathBuf::from(env!("CARGO_MANIFEST_DIR")).join("../datafusion/core/tests/data"); @@ -56,7 +157,7 @@ async fn setup_minio_container() -> Result<ContainerAsync<minio::MinIO>, String> .canonicalize() .expect("Failed to get absolute path for test data"); - let container = minio::MinIO::default() + minio::MinIO::default() .with_env_var("MINIO_ROOT_USER", MINIO_ROOT_USER) .with_env_var("MINIO_ROOT_PASSWORD", MINIO_ROOT_PASSWORD) .with_mount(Mount::bind_mount( @@ -64,60 +165,81 @@ async fn setup_minio_container() -> Result<ContainerAsync<minio::MinIO>, String> "/source", )) .start() - .await; - - match container { - Ok(container) => { - // We wait for MinIO to be healthy and prepare test files. We do it via CLI to avoid s3 dependency - let commands = [ - ExecCommand::new(["/usr/bin/mc", "ready", "local"]), - ExecCommand::new([ - "/usr/bin/mc", - "alias", - "set", - "localminio", - "http://localhost:9000";, - MINIO_ROOT_USER, - MINIO_ROOT_PASSWORD, - ]), - ExecCommand::new(["/usr/bin/mc", "mb", "localminio/data"]), - ExecCommand::new([ - "/usr/bin/mc", - "cp", - "-r", - "/source/", - "localminio/data/", - ]), - ]; - - for command in commands { - let command = - command.with_cmd_ready_condition(CmdWaitFor::Exit { code: Some(0) }); - - let cmd_ref = format!("{command:?}"); - - if let Err(e) = container.exec(command).await { - let stdout = container.stdout_to_vec().await.unwrap_or_default(); - let stderr = container.stderr_to_vec().await.unwrap_or_default(); - - return Err(format!( - "Failed to execute command: {}\nError: {}\nStdout: {:?}\nStderr: {:?}", - cmd_ref, - e, - String::from_utf8_lossy(&stdout), - String::from_utf8_lossy(&stderr) - )); - } - } + .await + .map_err(|e| match e { + TestcontainersError::Client(e) => format!( + "Failed to start MinIO container. Ensure Docker is running and accessible: {e}" + ), + e => format!("Failed to start MinIO container: {e}"), + }) +} - Ok(container) +/// Waits for MinIO to be healthy and uploads the test files. +/// +/// This is done via the `mc` CLI shipped in the image to avoid an s3 dependency. +async fn provision_minio_container( + container: &ContainerAsync<minio::MinIO>, +) -> Result<(), String> { + let commands = [ + ExecCommand::new(["/usr/bin/mc", "ready", "local"]), + ExecCommand::new([ + "/usr/bin/mc", + "alias", + "set", + "localminio", + "http://localhost:9000";, + MINIO_ROOT_USER, + MINIO_ROOT_PASSWORD, + ]), + ExecCommand::new(["/usr/bin/mc", "mb", "localminio/data"]), + ExecCommand::new(["/usr/bin/mc", "cp", "-r", "/source/", "localminio/data/"]), + ]; + + for command in commands { + let command = + command.with_cmd_ready_condition(CmdWaitFor::Exit { code: Some(0) }); + + let cmd_ref = format!("{command:?}"); + + if let Err(e) = container.exec(command).await { + let stdout = container.stdout_to_vec().await.unwrap_or_default(); + let stderr = container.stderr_to_vec().await.unwrap_or_default(); + + return Err(format!( + "Failed to execute command: {}\nError: {}\nStdout: {:?}\nStderr: {:?}", + cmd_ref, + e, + String::from_utf8_lossy(&stdout), + String::from_utf8_lossy(&stderr) + )); } - - Err(TestcontainersError::Client(e)) => Err(format!( - "Failed to start MinIO container. Ensure Docker is running and accessible: {e}" - )), - Err(e) => Err(format!("Failed to start MinIO container: {e}")), } + + Ok(()) +} + +/// CI pre-pulls the MinIO image so that the storage integration tests do not +/// have to pull it themselves. Guard against that pre-pull going stale when +/// `testcontainers-modules` bumps the image it uses. +#[test] +fn minio_image_matches_ci_prepull() { + let workflow = + PathBuf::from(env!("CARGO_MANIFEST_DIR")).join("../.github/workflows/rust.yml"); + + // The workflow is not shipped with the published crate. + let Ok(contents) = fs::read_to_string(&workflow) else { + return; + }; + + let image = minio::MinIO::default(); + let image_ref = format!("{}:{}", image.name(), image.tag()); + + assert!( + contents.contains(&image_ref), + "{} does not pre-pull `{image_ref}`. Update MINIO_IMAGE in the \ + `Pre-pull MinIO image` step to match the image used by the tests.", + workflow.display() + ); } #[cfg(test)] @@ -553,18 +675,8 @@ fn test_cli_wide_result_set_no_crash() { #[tokio::test] async fn test_cli() { - if env::var("TEST_STORAGE_INTEGRATION").is_err() { - eprintln!("Skipping external storages integration tests"); + let Some(container) = start_minio_or_skip().await else { return; - } - - let container = match setup_minio_container().await { - Ok(c) => c, - Err(e) if e.contains("toomanyrequests") => { - eprintln!("Skipping test: Docker pull rate limit reached: {e}"); - return; - } - e @ Err(_) => e.unwrap(), }; let settings = make_settings(); @@ -577,8 +689,8 @@ async fn test_cli() { assert_cmd_snapshot!( cli() .env_clear() - .env("AWS_ACCESS_KEY_ID", "TEST-DataFusionLogin") - .env("AWS_SECRET_ACCESS_KEY", "TEST-DataFusionPassword") + .env("AWS_ACCESS_KEY_ID", MINIO_ROOT_USER) + .env("AWS_SECRET_ACCESS_KEY", MINIO_ROOT_PASSWORD) .env("AWS_ENDPOINT", format!("http://localhost:{port}";)) .env("AWS_ALLOW_HTTP", "true") .pass_stdin(input) @@ -590,22 +702,13 @@ async fn test_cli() { async fn test_aws_options() { // Separate test is needed to pass aws as options in sql and not via env - if env::var("TEST_STORAGE_INTEGRATION").is_err() { - eprintln!("Skipping external storages integration tests"); + let Some(container) = start_minio_or_skip().await else { return; - } + }; let settings = make_settings(); let _bound = settings.bind_to_scope(); - let container = match setup_minio_container().await { - Ok(c) => c, - Err(e) if e.contains("toomanyrequests") => { - eprintln!("Skipping test: Docker pull rate limit reached: {e}"); - return; - } - e @ Err(_) => e.unwrap(), - }; let port = container.get_host_port_ipv4(9000).await.unwrap(); let input = format!( @@ -613,8 +716,8 @@ async fn test_aws_options() { STORED AS CSV LOCATION 's3://data/cars.csv' OPTIONS( - 'aws.access_key_id' 'TEST-DataFusionLogin', - 'aws.secret_access_key' 'TEST-DataFusionPassword', + 'aws.access_key_id' '{MINIO_ROOT_USER}', + 'aws.secret_access_key' '{MINIO_ROOT_PASSWORD}', 'aws.endpoint' 'http://localhost:{port}', 'aws.allow_http' 'true' ); @@ -689,18 +792,8 @@ fn test_backtrace_output(#[case] query: &str) { #[tokio::test] async fn test_s3_url_fallback() { - if env::var("TEST_STORAGE_INTEGRATION").is_err() { - eprintln!("Skipping external storages integration tests"); + let Some(container) = start_minio_or_skip().await else { return; - } - - let container = match setup_minio_container().await { - Ok(c) => c, - Err(e) if e.contains("toomanyrequests") => { - eprintln!("Skipping test: Docker pull rate limit reached: {e}"); - return; - } - e @ Err(_) => e.unwrap(), }; let mut settings = make_settings(); @@ -726,19 +819,10 @@ SELECT * FROM partitioned_data ORDER BY column_1, column_2 LIMIT 5; /// Validate object store profiling output #[tokio::test] async fn test_object_store_profiling() { - if env::var("TEST_STORAGE_INTEGRATION").is_err() { - eprintln!("Skipping external storages integration tests"); + let Some(container) = start_minio_or_skip().await else { return; - } - - let container = match setup_minio_container().await { - Ok(c) => c, - Err(e) if e.contains("toomanyrequests") => { - eprintln!("Skipping test: Docker pull rate limit reached: {e}"); - return; - } - e @ Err(_) => e.unwrap(), }; + let mut settings = make_settings(); // as the object store profiling contains timestamps and durations, we must @@ -800,8 +884,8 @@ impl MinioCommandExt for Command { let port = container.get_host_port_ipv4(9000).await.unwrap(); self.env_clear() - .env("AWS_ACCESS_KEY_ID", "TEST-DataFusionLogin") - .env("AWS_SECRET_ACCESS_KEY", "TEST-DataFusionPassword") + .env("AWS_ACCESS_KEY_ID", MINIO_ROOT_USER) + .env("AWS_SECRET_ACCESS_KEY", MINIO_ROOT_PASSWORD) .env("AWS_ENDPOINT", format!("http://localhost:{port}";)) .env("AWS_ALLOW_HTTP", "true") } --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
