This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/datafusion.git
The following commit(s) were added to refs/heads/main by this push:
new 1ec9ede5da minor: make MinIO tests more stable (#25092)
1ec9ede5da is described below
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]
