From cbacc8371cbe15953dceecdee5ec0ddc9399162b Mon Sep 17 00:00:00 2001 From: Yongting You <2010youy01@gmail.com> Date: Tue, 3 Nov 2026 20:33:31 +0800 Subject: [PATCH 1/6] test: Add a shared framework for reproducible fuzz tests --- .github/workflows/extended.yml | 6 +- Cargo.lock | 10 + Cargo.toml | 1 + datafusion/fuzz/Cargo.toml | 43 ++++ datafusion/fuzz/tests/fuzz.rs | 63 ++++++ datafusion/fuzz/tests/suites/mod.rs | 7 + datafusion/fuzz/tests/suites/spill_pool.rs | 27 +++ datafusion/physical-plan/Cargo.toml | 3 +- .../physical-plan/src/spill/spill_pool.rs | 64 +++--- test-utils/src/fuzz.rs | 186 ++++++++++++++++++ test-utils/src/lib.rs | 1 + xtask/src/ci_steps.rs | 83 +++----- 12 files changed, 406 insertions(+), 88 deletions(-) create mode 100644 datafusion/fuzz/Cargo.toml create mode 100644 datafusion/fuzz/tests/fuzz.rs create mode 100644 datafusion/fuzz/tests/suites/mod.rs create mode 100644 datafusion/fuzz/tests/suites/spill_pool.rs create mode 100644 test-utils/src/fuzz.rs diff --git a/.github/workflows/extended.yml b/.github/workflows/extended.yml index a240f2c3e2aa5..c4b22940c669a 100644 --- a/.github/workflows/extended.yml +++ b/.github/workflows/extended.yml @@ -82,10 +82,10 @@ jobs: - name: Show available disk space run: | df -h - - name: Run tests (excluding doctests) + - name: Run extended tests run: cargo xtask ci step test extended - - name: Run spill pool fuzzer with more iterations - run: cargo xtask ci step test spill-pool-fuzz + - name: Run fuzz tests + run: cargo xtask ci step test fuzz - parallel: - name: Verify Working Directory Clean run: git diff --exit-code diff --git a/Cargo.lock b/Cargo.lock index b581b3c8d42d3..35479f2e706c9 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2412,6 +2412,16 @@ dependencies = [ "datafusion-physical-expr-common", ] +[[package]] +name = "datafusion-fuzz" +version = "0.1.0" +dependencies = [ + "datafusion-common", + "datafusion-physical-plan", + "test-utils", + "tokio", +] + [[package]] name = "datafusion-macros" version = "55.1.0" diff --git a/Cargo.toml b/Cargo.toml index 3070ed33f3231..b4a188cbfdeb4 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -33,6 +33,7 @@ members = [ "datafusion/execution", "datafusion/ffi", "datafusion/ffi/integration-tests", + "datafusion/fuzz", "datafusion/functions", "datafusion/functions-aggregate", "datafusion/functions-aggregate-common", diff --git a/datafusion/fuzz/Cargo.toml b/datafusion/fuzz/Cargo.toml new file mode 100644 index 0000000000000..5a669ef10993c --- /dev/null +++ b/datafusion/fuzz/Cargo.toml @@ -0,0 +1,43 @@ +# 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. + +# A private crate providing a single entry point for all fuzz tests. + +[package] +name = "datafusion-fuzz" +version = "0.1.0" +edition = { workspace = true } +# Ensure this crate is private +publish = false +autotests = false + +[lints] +workspace = true + +[features] +extended_tests = ["datafusion-physical-plan/test_utils"] + +[dev-dependencies] +datafusion-common = { workspace = true } +datafusion-physical-plan = { workspace = true } +test-utils = { path = "../../test-utils" } +tokio = { workspace = true, features = ["rt-multi-thread"] } + +[[test]] +name = "fuzz" +path = "tests/fuzz.rs" +required-features = ["extended_tests"] diff --git a/datafusion/fuzz/tests/fuzz.rs b/datafusion/fuzz/tests/fuzz.rs new file mode 100644 index 0000000000000..a857e3842a63c --- /dev/null +++ b/datafusion/fuzz/tests/fuzz.rs @@ -0,0 +1,63 @@ +// SPDX-License-Identifier: Apache-2.0 +// SPDX-FileCopyrightText: See the NOTICE file distributed with this work for additional information regarding copyright ownership +// SPDX-FileContributor: Licensed to the Apache Software Foundation (ASF) under one or more contributor license agreements + +//! One integration entrypoint for DataFusion's registered fuzz suites. + +use std::future::Future; +use std::pin::Pin; + +use datafusion_common::Result; +use test_utils::fuzz::{FuzzTest, RunConfig, run}; + +mod suites; + +/// The single registration point for project-wide fuzz campaigns. +#[tokio::test(flavor = "multi_thread")] +async fn fuzz_tests() -> Result<()> { + let spill_pool = suites::spill_pool::SpillPoolFuzzer; + let suites = [Suite::new( + &spill_pool, + RunConfig { + run_seed: Some(0), + max_cases: Some(1000), + time_limit: None, + cases_per_dataset: 1000, + }, + )]; + run_suites(suites).await +} + +async fn run_suites<'a>(suites: impl IntoIterator>) -> Result<()> { + for suite in suites { + eprintln!("Running fuzz suite: {}", suite.name); + suite.campaign.await?; + } + Ok(()) +} + +/// Erase only a campaign's future, retaining each adapter's concrete dataset +/// type. Borrowed fixtures and futures that are not `Send` remain supported. +struct Suite<'a> { + name: &'a str, + campaign: Pin> + 'a>>, +} + +impl<'a> Suite<'a> { + fn new(test: &'a T, config: RunConfig) -> Self { + Self { + name: test.name(), + campaign: Box::pin(async move { + let stats = run(test, config).await?; + eprintln!( + "Fuzz suite {}: {} cases, {} datasets, {:?}", + test.name(), + stats.cases_passed, + stats.datasets_prepared, + stats.elapsed + ); + Ok(()) + }), + } + } +} diff --git a/datafusion/fuzz/tests/suites/mod.rs b/datafusion/fuzz/tests/suites/mod.rs new file mode 100644 index 0000000000000..886985c2286b1 --- /dev/null +++ b/datafusion/fuzz/tests/suites/mod.rs @@ -0,0 +1,7 @@ +// SPDX-License-Identifier: Apache-2.0 +// SPDX-FileCopyrightText: See the NOTICE file distributed with this work for additional information regarding copyright ownership +// SPDX-FileContributor: Licensed to the Apache Software Foundation (ASF) under one or more contributor license agreements + +//! Adapters for component-local fuzz scenarios. + +pub(crate) mod spill_pool; diff --git a/datafusion/fuzz/tests/suites/spill_pool.rs b/datafusion/fuzz/tests/suites/spill_pool.rs new file mode 100644 index 0000000000000..c1b4897c6e9ec --- /dev/null +++ b/datafusion/fuzz/tests/suites/spill_pool.rs @@ -0,0 +1,27 @@ +// SPDX-License-Identifier: Apache-2.0 +// SPDX-FileCopyrightText: See the NOTICE file distributed with this work for additional information regarding copyright ownership +// SPDX-FileContributor: Licensed to the Apache Software Foundation (ASF) under one or more contributor license agreements + +//! Connect the component-local spill pool scenario to the shared fuzz interface. + +use datafusion_common::Result; +use datafusion_physical_plan::spill::spill_pool::run_spill_pool_scenario; +use test_utils::fuzz::FuzzTest; + +pub(crate) struct SpillPoolFuzzer; + +impl FuzzTest for SpillPoolFuzzer { + type Dataset = (); + + fn name(&self) -> &str { + "spill_pool_scenario" + } + + fn prepare_dataset(&self, _: u64) -> impl Future> { + std::future::ready(Ok(())) + } + + async fn run_case(&self, _: &(), case_seed: u64) -> Result<()> { + run_spill_pool_scenario(case_seed).await + } +} diff --git a/datafusion/physical-plan/Cargo.toml b/datafusion/physical-plan/Cargo.toml index 82a2215581f22..8572f2320e4a9 100644 --- a/datafusion/physical-plan/Cargo.toml +++ b/datafusion/physical-plan/Cargo.toml @@ -39,7 +39,7 @@ workspace = true [features] force_hash_collisions = [] -test_utils = ["arrow/test_utils"] +test_utils = ["arrow/test_utils", "dep:rand"] tokio_coop = [] tokio_coop_fallback = [] # Enables `PhysicalExpr::try_to_proto` / `try_from_proto` hooks on the @@ -87,6 +87,7 @@ log = { workspace = true } num-traits = { workspace = true } parking_lot = { workspace = true } pin-project-lite = { workspace = true } +rand = { workspace = true, optional = true } serde = { workspace = true } serde_json = { workspace = true } tokio = { workspace = true, features = ["time"] } diff --git a/datafusion/physical-plan/src/spill/spill_pool.rs b/datafusion/physical-plan/src/spill/spill_pool.rs index 6d50245ea60af..c4089354f4ab4 100644 --- a/datafusion/physical-plan/src/spill/spill_pool.rs +++ b/datafusion/physical-plan/src/spill/spill_pool.rs @@ -814,24 +814,40 @@ impl RecordBatchStream for SpillPoolReader { } } -#[cfg(test)] +/// Internal fuzz entrypoint for the centralized test driver. +/// +/// This is not a public API; see [API policy] for details. +/// +/// [API policy]: https://datafusion.apache.org/contributor-guide/api-health.html#internal-public-apis +#[cfg(any(test, feature = "test_utils"))] +#[doc(hidden)] +pub use tests::run_spill_pool_scenario; + +#[cfg(any(test, feature = "test_utils"))] mod tests { use super::*; use crate::metrics::{ExecutionPlanMetricsSet, SpillMetrics}; use arrow::array::{ArrayRef, Int32Array}; use arrow::datatypes::{DataType, Field, Schema}; + use datafusion_common::DataFusionError; + #[cfg(test)] + use datafusion_common::exec_datafusion_err; use datafusion_common::instant::Instant; - use datafusion_common::{DataFusionError, exec_datafusion_err}; + #[cfg(test)] use datafusion_common_runtime::{JoinSet, SpawnedTask}; use datafusion_execution::disk_manager::{ DiskManager, DiskManagerBuilder, DiskManagerMode, }; + #[cfg(test)] use datafusion_execution::runtime_env::RuntimeEnv; use datafusion_execution::runtime_env::RuntimeEnvBuilder; use datafusion_execution::{SpillFile, SpillWriter, TempFileFactory}; use std::pin::Pin; + #[cfg(test)] use std::sync::Barrier; + #[cfg(test)] use std::sync::atomic::{AtomicBool, Ordering}; + #[cfg(test)] use std::sync::mpsc::{self, Sender}; use std::time::Duration; @@ -839,6 +855,7 @@ mod tests { Arc::new(Schema::new(vec![Field::new("a", DataType::Int32, false)])) } + #[cfg(test)] fn create_test_batch(start: i32, count: usize) -> RecordBatch { let schema = create_test_schema(); let a: ArrayRef = Arc::new(Int32Array::from( @@ -847,6 +864,7 @@ mod tests { RecordBatch::try_new(schema, vec![a]).unwrap() } + #[cfg(test)] fn create_spill_channel( max_file_size: usize, ) -> (SpillPoolSink, SendableRecordBatchStream) { @@ -858,6 +876,7 @@ mod tests { spsc_channel(max_file_size, spill_manager) } + #[cfg(test)] fn create_shared_spill_channel( max_file_size: usize, ) -> (SpillPoolWriter, SendableRecordBatchStream) { @@ -869,6 +888,7 @@ mod tests { mpsc_channel(max_file_size, spill_manager) } + #[cfg(test)] fn create_spill_channel_with_metrics( max_file_size: usize, ) -> (SpillPoolSink, SendableRecordBatchStream, SpillMetrics) { @@ -1300,6 +1320,7 @@ mod tests { Ok(()) } + #[cfg(test)] #[tokio::test(flavor = "multi_thread", worker_threads = 10)] async fn test_concurrent_writers() -> Result<()> { let (writer, mut reader) = create_shared_spill_channel(1024 * 1024); @@ -1690,6 +1711,10 @@ mod tests { /// different error paths of [`SpillPoolFile::poll_file`]: one fails while it /// builds the stream, the other fails while it polls a stream that is /// already built. + #[cfg_attr( + not(test), + expect(dead_code, reason = "Read failures are injected by unit tests") + )] enum ReadBehavior { /// Read the real bytes, after this delay. Delay(Duration), @@ -1813,6 +1838,7 @@ mod tests { /// Holds the first disk write on the pool until the test releases it, and /// reports every other write so a test can wait for a second writer to /// reach a file of its own instead of sleeping. + #[cfg(test)] struct WriteGate { taken: AtomicBool, entered: Barrier, @@ -1820,6 +1846,7 @@ mod tests { other_writes: Sender<()>, } + #[cfg(test)] impl WriteGate { fn hold_if_first(&self) { if self.taken.swap(true, Ordering::SeqCst) { @@ -2049,8 +2076,10 @@ mod tests { .value(0) } - /// One randomly generated scenario, see [`spill_pool_scenario_fuzz`]. - async fn run_spill_pool_scenario(seed: u64) -> Result<()> { + /// Run one seeded scenario checking concurrent writes, reader progress, + /// ordering, and spill-file cleanup. Seeds reproduce the logical workload + /// and delay policy; thread scheduling may still vary. + pub async fn run_spill_pool_scenario(seed: u64) -> Result<()> { use rand::rngs::StdRng; use rand::{Rng, SeedableRng}; @@ -2229,31 +2258,4 @@ mod tests { ); Ok(()) } - - /// Randomized scenarios against the pool's contract: after any set of - /// overlapping pushes the reader can drain everything without further - /// writer activity (no lost wakeups, no parking on one file while another - /// has data), it never signals EOF while a writer is alive, it returns - /// exactly the pushed batches (in order for a single writer), and the - /// files are released when it is dropped. - /// - /// `DATAFUSION_SPILL_POOL_FUZZ_ITERATIONS` and - /// `DATAFUSION_SPILL_POOL_FUZZ_SEED` select how many scenarios run and - /// from which seed; a failure names its seed so it can be replayed. - #[tokio::test] - async fn spill_pool_scenario_fuzz() -> Result<()> { - let env_u64 = |name: &str, default: u64| { - std::env::var(name) - .ok() - .and_then(|v| v.parse().ok()) - .unwrap_or(default) - }; - let first_seed = env_u64("DATAFUSION_SPILL_POOL_FUZZ_SEED", 0); - let iterations = env_u64("DATAFUSION_SPILL_POOL_FUZZ_ITERATIONS", 50); - // `saturating_add` so a replay seed near `u64::MAX` cannot overflow. - for seed in first_seed..first_seed.saturating_add(iterations) { - run_spill_pool_scenario(seed).await?; - } - Ok(()) - } } diff --git a/test-utils/src/fuzz.rs b/test-utils/src/fuzz.rs new file mode 100644 index 0000000000000..9de1d27980cd4 --- /dev/null +++ b/test-utils/src/fuzz.rs @@ -0,0 +1,186 @@ +// SPDX-License-Identifier: Apache-2.0 +// SPDX-FileCopyrightText: See the NOTICE file distributed with this work for additional information regarding copyright ownership +// SPDX-FileContributor: Licensed to the Apache Software Foundation (ASF) under one or more contributor license agreements + +//! Shared execution and replay for [`FuzzTest`] experiments. + +use std::future::Future; +use std::time::Duration; + +use datafusion_common::instant::Instant; +use datafusion_common::{Result, exec_err}; + +/// A component's fuzz experiment, run by the centralized driver. +/// +/// The component prepares fixtures and checks one case; the driver handles +/// repetition, budgets, and seed reporting. Each case can be replayed alone. +/// +/// See the driver in `datafusion-fuzz` for how to use it. At a high level: +/// 1. Generate one dataset with `prepare_dataset()`. +/// 2. Run several cases on that dataset with `run_case()`. +/// 3. Repeat steps 1 and 2 until the time budget is reached. +/// (All configuration, such as the total time budget, lives in [`RunConfig`].) +#[expect(async_fn_in_trait)] // Serial execution deliberately does not require Send futures. +pub trait FuzzTest { + /// Reusable fixtures; `()` is valid when no preparation can be shared. + type Dataset; + + /// Test identity, including any static variant needed for replay. + fn name(&self) -> &str; + + /// Build all fixtures from this seed and the test's fixed configuration. + async fn prepare_dataset(&self, dataset_seed: u64) -> Result; + + /// Execute one complete, independent validation and clean up its resources. + /// + /// The implementation should ensure: + /// - The test case is reproducible given the same dataset and seed. + /// - If the test fails, the error message includes all the information needed + /// to reproduce it (by providing the data seed and case seed). + async fn run_case(&self, dataset: &Self::Dataset, case_seed: u64) -> Result<()>; + + /// Reproduce the failing case deterministically. + /// + /// The input seed should be printed in the error message from [`Self::run_case`] + /// when the test fails. + async fn reproduce(&self, dataset_seed: u64, case_seed: u64) -> Result<()> { + let dataset = with_context(self.name(), None, dataset_seed, None, || { + self.prepare_dataset(dataset_seed) + }) + .await?; + with_context(self.name(), None, dataset_seed, Some(case_seed), || { + self.run_case(&dataset, case_seed) + }) + .await + } +} + +/// Execution limits for [`run`]. At least one stopping condition is required. +#[derive(Debug, Clone)] +pub struct RunConfig { + /// Initial seed, or `None` to choose a fresh seed once per run. + pub run_seed: Option, + /// Maximum successful cases; must be positive when supplied. + pub max_cases: Option, + /// Soft budget including preparation. The first complete case always runs, + /// even with a zero budget. Active preparation and cases are never cancelled. + pub time_limit: Option, + /// Maximum cases sharing one fixture; must be positive. + pub cases_per_dataset: usize, +} + +impl RunConfig { + fn validate(&self) -> Result<()> { + if self.max_cases.is_none() && self.time_limit.is_none() { + return exec_err!("Fuzz run requires max_cases or time_limit"); + } + if self.max_cases == Some(0) { + return exec_err!("Fuzz run max_cases must be greater than zero"); + } + if self.cases_per_dataset == 0 { + return exec_err!("Fuzz run cases_per_dataset must be greater than zero"); + } + Ok(()) + } + + // Keep admission separate from the clock so deadline boundaries can be + // tested without sleeping. Only the first case may start past the deadline. + fn should_stop(&self, cases_passed: usize, elapsed: Duration) -> bool { + self.max_cases.is_some_and(|limit| cases_passed >= limit) + || (cases_passed > 0 && self.time_limit.is_some_and(|limit| elapsed >= limit)) + } +} + +/// Counters returned by a successful campaign. Failures stop the run immediately. +#[derive(Debug)] +pub struct RunStats { + /// Resolved seed, including when the caller requested a fresh one. + pub run_seed: u64, + /// Number of successfully prepared fixtures. + pub datasets_prepared: usize, + /// Number of complete validations that returned successfully. + pub cases_passed: usize, + /// Total execution time, including fixture preparation and cleanup. + pub elapsed: Duration, +} + +/// Run cases serially until either limit is reached or a failure occurs. +/// +/// Case seeds are `run_seed + case_index`, wrapping at `u64::MAX`. Dataset seeds +/// independently use `run_seed + dataset_index`. Neither sequence depends on +/// randomness consumed by the test. This also lets existing consecutive-seed +/// fuzzers retain their coverage. A timed run may finish different case counts +/// on different machines; use a fixed count for a repeatable seed sequence. +pub async fn run(test: &T, config: RunConfig) -> Result { + config.validate()?; + let run_seed = config.run_seed.unwrap_or_else(rand::random); + let start = Instant::now(); + let mut stats = RunStats { + run_seed, + datasets_prepared: 0, + cases_passed: 0, + elapsed: Duration::ZERO, + }; + while !config.should_stop(stats.cases_passed, start.elapsed()) { + let dataset_seed = run_seed.wrapping_add(stats.datasets_prepared as u64); + let dataset = + with_context(test.name(), Some(run_seed), dataset_seed, None, || { + test.prepare_dataset(dataset_seed) + }) + .await?; + stats.datasets_prepared += 1; + for _ in 0..config.cases_per_dataset { + if config.should_stop(stats.cases_passed, start.elapsed()) { + break; + } + let case_seed = run_seed.wrapping_add(stats.cases_passed as u64); + with_context( + test.name(), + Some(run_seed), + dataset_seed, + Some(case_seed), + || test.run_case(&dataset, case_seed), + ) + .await?; + stats.cases_passed += 1; + } + } + stats.elapsed = start.elapsed(); + Ok(stats) +} + +/// Print before invoking the operation so libtest captures context even for +/// assertion panics. Returned errors retain their original source; panics unwind +/// normally with their original payload and hook, without process-global changes. +async fn with_context>>( + name: &str, + run_seed: Option, + dataset_seed: u64, + case_seed: Option, + operation: impl FnOnce() -> F, +) -> Result { + let phase = if case_seed.is_some() { + "run_case" + } else { + "prepare_dataset" + }; + let run_context = run_seed + .map(|seed| format!("Run seed: {seed}\n")) + .unwrap_or_default(); + let instruction = if let Some(case_seed) = case_seed { + format!( + "Case seed: {case_seed}\nReproduce with this test's configuration: \ + test.reproduce({dataset_seed}, {case_seed}).await" + ) + } else { + format!( + "Retry with this test's configuration: \ + test.prepare_dataset({dataset_seed}).await" + ) + }; + let context = format!( + "Fuzz test: {name}\nPhase: {phase}\n{run_context}Dataset seed: {dataset_seed}\n{instruction}" + ); + eprintln!("{context}"); + operation().await.map_err(|error| error.context(context)) +} diff --git a/test-utils/src/lib.rs b/test-utils/src/lib.rs index 55d801e3b5ba2..be0f4ff806f3f 100644 --- a/test-utils/src/lib.rs +++ b/test-utils/src/lib.rs @@ -12,6 +12,7 @@ use rand::{Rng, SeedableRng}; pub mod array_gen; mod data_gen; +pub mod fuzz; mod string_gen; pub mod tpcds; pub mod tpch; diff --git a/xtask/src/ci_steps.rs b/xtask/src/ci_steps.rs index c6b605d36ed72..a38d5122e2529 100644 --- a/xtask/src/ci_steps.rs +++ b/xtask/src/ci_steps.rs @@ -15,8 +15,6 @@ type StepRunner = fn(&StepContext, &[String]) -> StepResult; const CI_COMMAND: &str = "cargo xtask ci step"; const CI_SHORTCUT: &str = "cargo ci-step"; const TPCH_DATA_DIR: &str = "datafusion/sqllogictest/test_files/tpch/data"; -/// The `datafusion-physical-plan` unit test that the `spill-pool-fuzz` step runs. -const SPILL_POOL_FUZZ_TEST: &str = "spill::spill_pool::tests::spill_pool_scenario_fuzz"; /// Metadata and implementation for one CI step. /// @@ -40,7 +38,7 @@ static CI_STEPS: &[StepInfo] = &[ }, StepInfo { command: "test", - help_usage: "test [--explain]", + help_usage: "test [--explain]", help_examples: &["test cli", "test workspace"], help_description: "Run a CI test suite", runner: StepContext::run_test, @@ -438,10 +436,7 @@ impl StepContext { } } "extended" => { - cmd_builder.doc( - "Run tests behind the `extended_tests` feature. The `workspace`\n\ - step runs the remaining tests.", - ); + cmd_builder.doc("Run the core extended tests."); cmd_builder .program("cargo") .args([ @@ -450,6 +445,8 @@ impl StepContext { "ci", "-p", "datafusion", + "-p", + "datafusion-fuzz", "--test", "extended_tests", "--features", @@ -457,11 +454,9 @@ impl StepContext { ]) .env("RUST_BACKTRACE", "1"); } - "spill-pool-fuzz" => { - cmd_builder.doc( - "Run more random spill pool fuzzer scenarios than the `workspace`\n\ - step. This takes about a minute.", - ); + "fuzz" => { + cmd_builder.doc("Run the centralized fuzz tests."); + // Match extended's packages and features to reuse compiled dependencies. cmd_builder .program("cargo") .args([ @@ -469,14 +464,15 @@ impl StepContext { "--profile", "ci", "-p", - "datafusion-physical-plan", - "--lib", - "--", - "--exact", - SPILL_POOL_FUZZ_TEST, + "datafusion", + "-p", + "datafusion-fuzz", + "--test", + "fuzz", + "--features", + "avro,backtrace,extended_tests,parquet_encryption", ]) - .env("RUST_BACKTRACE", "1") - .env("DATAFUSION_SPILL_POOL_FUZZ_ITERATIONS", "1000"); + .env("RUST_BACKTRACE", "1"); } "hash-collisions" => { cmd_builder.doc( @@ -557,7 +553,7 @@ mod tests { use super::*; use std::ffi::OsStr; - const TEST_USAGE: &str = "usage: cargo xtask ci step test [--explain]"; + const TEST_USAGE: &str = "usage: cargo xtask ci step test [--explain]"; fn args(args: &[&str]) -> Vec { args.iter().map(|arg| (*arg).to_string()).collect() @@ -606,28 +602,9 @@ mod tests { assert_eq!(error, TEST_USAGE); } - /// A filter that matches no test passes without running anything, so a - /// renamed or moved spill pool fuzzer would silently stop running in CI. - #[test] - fn spill_pool_fuzz_test_exists() { - let (module, test) = SPILL_POOL_FUZZ_TEST - .rsplit_once("::tests::") - .expect("test path has a `tests` module"); - let file = Path::new(env!("CARGO_MANIFEST_DIR")) - .join("../datafusion/physical-plan/src") - .join(format!("{}.rs", module.replace("::", "/"))); - let source = std::fs::read_to_string(&file) - .unwrap_or_else(|e| panic!("cannot read {}: {e}", file.display())); - assert!( - source.contains("mod tests {") && source.contains(&format!("fn {test}(")), - "{} has no test `{test}` in `mod tests`; update SPILL_POOL_FUZZ_TEST", - file.display() - ); - } - - /// The `extended` step runs only the `extended_tests` test target, which - /// requires the `extended_tests` feature. A test gated on the feature with a - /// `cfg` in any other target would never run in CI. + /// Extended workflows explicitly select targets gated by Cargo's + /// `required-features`. A test hidden behind a `cfg` in another target + /// would never run in those workflows. #[test] fn extended_tests_feature_only_gates_whole_targets() { fn visit(dir: &Path, found: &mut Vec) { @@ -651,8 +628,9 @@ mod tests { ); assert!( found.is_empty(), - "put tests behind `extended_tests` in the `extended_tests` test target \ - (see datafusion/core/Cargo.toml), not behind a `cfg` in {found:?}" + "gate complete test targets with `required-features = [\"extended_tests\"]` \ + (see datafusion/core/Cargo.toml and datafusion/fuzz/Cargo.toml), \ + not with a `cfg` in {found:?}" ); } @@ -926,34 +904,33 @@ cargo test \ command: &["test", "extended"], expected: |actual| { insta::assert_snapshot!(actual, @r" -# Run tests behind the `extended_tests` feature. The `workspace` -# step runs the remaining tests. +# Run the core extended tests. cd /workspace && \ RUST_BACKTRACE=1 \ cargo test \ --profile ci \ -p datafusion \ +-p datafusion-fuzz \ --test extended_tests \ --features avro,backtrace,extended_tests,parquet_encryption ") }, }, Cmd { - command: &["test", "spill-pool-fuzz"], + command: &["test", "fuzz"], expected: |actual| { insta::assert_snapshot!(actual, @r" -# Run more random spill pool fuzzer scenarios than the `workspace` -# step. This takes about a minute. +# Run the centralized fuzz tests. cd /workspace && \ -DATAFUSION_SPILL_POOL_FUZZ_ITERATIONS=1000 \ RUST_BACKTRACE=1 \ cargo test \ --profile ci \ --p datafusion-physical-plan \ ---lib \ --- --exact spill::spill_pool::tests::spill_pool_scenario_fuzz +-p datafusion \ +-p datafusion-fuzz \ +--test fuzz \ +--features avro,backtrace,extended_tests,parquet_encryption ") }, }, From 2b56ad97cfdaa44b6c48ba816299da63d814e743 Mon Sep 17 00:00:00 2001 From: Yongting You <2010youy01@gmail.com> Date: Tue, 3 Nov 2026 21:03:35 +0800 Subject: [PATCH 2/6] cleanup --- .../check_asf_yaml_status_checks.cpython-314.pyc | Bin 6229 -> 0 bytes 1 file changed, 0 insertions(+), 0 deletions(-) delete mode 100644 ci/scripts/__pycache__/check_asf_yaml_status_checks.cpython-314.pyc diff --git a/ci/scripts/__pycache__/check_asf_yaml_status_checks.cpython-314.pyc b/ci/scripts/__pycache__/check_asf_yaml_status_checks.cpython-314.pyc deleted file mode 100644 index a8a2f27f7b01a81b50148c9fa1ac42da442bebbf..0000000000000000000000000000000000000000 GIT binary patch literal 0 HcmV?d00001 literal 6229 zcmd5=TWlN06`kesv0N@6dRX#9kJTfVXuV>|actRfBTJSnOO_QjrV~^NOOZ=yZHl|u zT}Bp@y8I|wst6z{2r8vX5jqGO83O*n`;;|SXn_z_)D+^N`JL;2L+3~0lx`JD$IB9~Dz3nvJ!zwePzzPxzM~QOz!xldx zJIzOVNLjk(`0uP%JC)3)SwidX{Sd*z;7ejphhBO z-hLlilW5H7^P?lV`UxUH#7%KBU|gR_jOh;qjhYQwMx_q<34OL`@)7uV1XjVR7Ih$z ze8Dt|SKgI;rVex@N4aP}neP|Pe)Fh(7XkRxpO^>rKZnpXP?P+AYf)&OS*)eA-7;>jbY%#>drC zcrgmR7!aj#DHy}jqgE7udMc%p^V(FiAaXI)jD=CP%fSd2hzh~5N=|T!YT*@bN>*u^ zQ^H&%7>_EdH7LUPu-C3SL52f(S^$fbN<>sEkd5#YP-;S~QpQHC*_uiLB?ZJrmaM=T zl?*9YWqjoa*<4b8*l|(jB)MZoh)v*;9lb)B>rke|4p1h+NL&W#Z;QkeiMF`X7Lh;$ zV3F?#@f~tV;zdR7&=wCEo77?a(qR0*5FfB83%CDA>*Gc~ntuUfNL zrHRdwJ&QGO)})C{Nqx%tkVJI#mKC`?$^H%Zp1H(qVyQWOI^%9ycekv%TUJ`1{kT5k zK9fAVQPud~$lD`#s#;g6y(#NInew^e+2NIahi`Ren4{}V_bStUyZd7)!<^e9k-cQw zipr}}y&FvFI#a#MR4;|nF!10y)3(aAEl2J!Pi>Ph;=UbKw{5u)Q<^;ga35?2_W6yB zAMjh1r%ze_c*@c1vhPp{6!d8-0hWQofw&TRwNN1Vh6;f3xm1V740SMy?`&5pCa`|6 z1G>=bBBGZbbcX2v6$qVa%n&)hN$)M9n8-0kUvku?*YYJOCQ_d*g2lqWykdI6tP5Zm zcu<9Wq~ClTlwcV6P=$sD5CWA`lBcHEd{`9~Yq#VTjDAAVGFGO~^IOKs^}60~>;Ohp zA-|<4U2C51c>?aoI))#kmI|m_zfJuFVbE{E^lK5Q#$r3S>0HXsm;j9M9Wy9>hF*%c z$GnAo`kwP;{%%3d7UyY!2^c+)pQ+ah)c@-__5aHPp3vXsiO!LU>@8QrEz?DLV)xti zTM`|Ye%56Y89y-!<98>&qreo@?(PNpjLxH4bm^UA(+K6({sHtAr)dB0DGpuB8kIA| zP&#K`0Be1>c{xK2|2Su0=)~`Og=iE^gJxg=ILUxLlt~OTcnJWE62z)CYgu-oEFDf^ z9L321a(tW;mjHeR*)R_Tr@$Iu6okVTB0Y ztdk62lCA7yP}b}R{sg2Z!|=GMt(CAGYM+^kj;Ln5<>ytC5K|d39*t_so0GvjX;?17 z6hb=HjMw&hRI4mVU<_6BL{u1u$+BwJ>=aPrW0DOnQNnsr!sbW9xLF0v$j883#6p~E z#>Z35zRGs>1Ig zp7bQNH^X2438=p$+F}snO7?B;tNpM;}jQ4vl2k3+wwv zQ_f9i$$Z7LEX1B=(*yp|4TDw|Vh!4SIY9$JxS@6yrby=w;!-(||NU96ZJ zn;lzfOj92+*O}$#!Pv1!GtQ?|rp>NrZjb+|alLEc)2@Me-=crPzcjLZdTmd~UwtWh z^FZw;Q!$_T;7EpPeuxO?FtMfO8zl!8YZhwKwoFO$ddaz!l5@-FZjRg-xqT|rdT!f7 zR?#ce-Ul=)uULF~;pz3VhSjo$^!ZF#XNuZrX!-Eu^^>=(w@sOb6YC8h73D)Z#6@-^n@hP&d9yMDt{zUW?Xr)|rn8BgoFr)$;Ib*t}==lBCRYV0HK z*P;E-5nG*z@g#>H9tMs9GgqAKH!^-kf2gb+Xte#M(J}CpBLv?@!tW8Z7uxJ!3->7yV4kzI4$Y=@=>MuxHB8*L8l0s7Q9+RpA(PaK=JuR9z6g; zF+hKT^F#Vc!@d9>sX}8|z_sKvkO+9g<_0uo*x>>zXl_S45Q8xXoRPl)F0%S8#hj76 z#k+L073+{(+RZ=Fb}6hO9I&Js+SsFy%$Gi^-^w5#I6#n7orP{pFG8343+a2%Me3qG zPPU`CsUOaxxnHw>yTJo^rGTV>2YzbAGz1t2yFnr?QlkBTqm;vByD2#^B|p7uy+v5h zp||B`UPq`n@BFmRC9v<7(Z_Mb?q`56bf&hA=_HDB3v_YQrH+QJvX6uV2m z!SP~qbiA{Jj`|9K1b+anS9+_lCT?oYr@u^b6f4Yos#2cU>hUC zqdee<5Qjj|j@Z$^l5gQ4&m=}505V?9MV#745~fx*VPV-ewm#vgXPdL$XE&Q@@pfm! zLOQWvY-V?4N#9U{fx-F*o0wwbG6bYFtg$nQltV5!KvW1%c86}Jg;|Mz4wLZtHQ;7AuRDC(9lYtLRf0}fK-o1VNVC3oWw(L5W^Iy24`6Js-!#vunpEyR5HrN zye`!)aiS2A1VK?L-Nl!%P^xZ&D>;{517lWBk_1VXu+U53?;zDQ9Z_kFxw6-$mS#%` zn;6_s%i|Eo$OaAzT5Z#j0EV#=_7G(3jOKzCxgra`%u_+YmwnJXgoi!U@YbfMbkVk8 z!$@J1qUUV0wmJ8#`(4v}wzqBXx!-nwWV&g)VY}(R;r>09Y3|AFJ)NO?w@FipH97bI zAXe$z^z8K9>$9&f4W?hqxVw^PHv#SpEDS6TFAS&mEf3y$EmPi`Jb$mkyWE_q=t^Gr zhhzUH)HT%-JVDc3zm)zV8DM@zDA!|6`;-*fMo7w)(ZrpS%5%EjXg$KUm*FMQT` zEK_|vQ`QZ@$>DnAdk~12J3o7VqqK5~oR2RJEmP?PK(fvgkd>T#K$1?^eF{1EY|-f8 zpU-Lh~pXu?jz{fqVz$mG3auBP&EuEhrS{$u(2lOX?l1BQ7(-O z&#h5DsHDPk{4Bp@&A4jUC>9#tuxZ`L*QoBz^8HIkeleWt+hj`S`V6n}HYJcF64QUy-E6`t|l?Pxd=Q= From dd6f892d874447bc38a7f718f5e00484363cef2a Mon Sep 17 00:00:00 2001 From: Yongting You <2010youy01@gmail.com> Date: Wed, 4 Nov 2026 14:48:05 +0800 Subject: [PATCH 3/6] review: move spill_pool_fuzz implementation to `datafusion-fuzz` crage --- Cargo.lock | 6 + datafusion/fuzz/Cargo.toml | 10 +- datafusion/fuzz/tests/fuzz.rs | 10 +- datafusion/fuzz/tests/suites/mod.rs | 2 +- datafusion/fuzz/tests/suites/spill_pool.rs | 391 +++++++++++++++++- datafusion/physical-plan/Cargo.toml | 3 +- .../physical-plan/src/spill/spill_pool.rs | 285 +------------ 7 files changed, 410 insertions(+), 297 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 871883357639a..a5b661bccba31 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2415,8 +2415,14 @@ dependencies = [ name = "datafusion-fuzz" version = "0.1.0" dependencies = [ + "arrow", + "bytes", "datafusion-common", + "datafusion-execution", "datafusion-physical-plan", + "futures", + "parking_lot 0.13.0", + "rand 0.9.4", "test-utils", "tokio", ] diff --git a/datafusion/fuzz/Cargo.toml b/datafusion/fuzz/Cargo.toml index 5a669ef10993c..8ad5c759ec21d 100644 --- a/datafusion/fuzz/Cargo.toml +++ b/datafusion/fuzz/Cargo.toml @@ -29,13 +29,19 @@ autotests = false workspace = true [features] -extended_tests = ["datafusion-physical-plan/test_utils"] +extended_tests = [] [dev-dependencies] +arrow = { workspace = true } +bytes = { workspace = true } datafusion-common = { workspace = true } +datafusion-execution = { workspace = true } datafusion-physical-plan = { workspace = true } +futures = { workspace = true } +parking_lot = { workspace = true } +rand = { workspace = true } test-utils = { path = "../../test-utils" } -tokio = { workspace = true, features = ["rt-multi-thread"] } +tokio = { workspace = true, features = ["rt-multi-thread", "time"] } [[test]] name = "fuzz" diff --git a/datafusion/fuzz/tests/fuzz.rs b/datafusion/fuzz/tests/fuzz.rs index a857e3842a63c..cba3ed62fb749 100644 --- a/datafusion/fuzz/tests/fuzz.rs +++ b/datafusion/fuzz/tests/fuzz.rs @@ -31,23 +31,23 @@ async fn fuzz_tests() -> Result<()> { async fn run_suites<'a>(suites: impl IntoIterator>) -> Result<()> { for suite in suites { eprintln!("Running fuzz suite: {}", suite.name); - suite.campaign.await?; + suite.run.await?; } Ok(()) } -/// Erase only a campaign's future, retaining each adapter's concrete dataset -/// type. Borrowed fixtures and futures that are not `Send` remain supported. +/// Wrapper for one fuzzing test suite struct Suite<'a> { name: &'a str, - campaign: Pin> + 'a>>, + /// Run the test suite. Report an error on failure or execution statistics on success. + run: Pin> + 'a>>, } impl<'a> Suite<'a> { fn new(test: &'a T, config: RunConfig) -> Self { Self { name: test.name(), - campaign: Box::pin(async move { + run: Box::pin(async move { let stats = run(test, config).await?; eprintln!( "Fuzz suite {}: {} cases, {} datasets, {:?}", diff --git a/datafusion/fuzz/tests/suites/mod.rs b/datafusion/fuzz/tests/suites/mod.rs index 886985c2286b1..a443d433bd089 100644 --- a/datafusion/fuzz/tests/suites/mod.rs +++ b/datafusion/fuzz/tests/suites/mod.rs @@ -2,6 +2,6 @@ // SPDX-FileCopyrightText: See the NOTICE file distributed with this work for additional information regarding copyright ownership // SPDX-FileContributor: Licensed to the Apache Software Foundation (ASF) under one or more contributor license agreements -//! Adapters for component-local fuzz scenarios. +//! Implementation for fuzz test cases. pub(crate) mod spill_pool; diff --git a/datafusion/fuzz/tests/suites/spill_pool.rs b/datafusion/fuzz/tests/suites/spill_pool.rs index c1b4897c6e9ec..d6669bfe8e265 100644 --- a/datafusion/fuzz/tests/suites/spill_pool.rs +++ b/datafusion/fuzz/tests/suites/spill_pool.rs @@ -2,10 +2,29 @@ // SPDX-FileCopyrightText: See the NOTICE file distributed with this work for additional information regarding copyright ownership // SPDX-FileContributor: Licensed to the Apache Software Foundation (ASF) under one or more contributor license agreements -//! Connect the component-local spill pool scenario to the shared fuzz interface. +//! Seeded scenarios checking concurrent writes, reader progress, ordering, and +//! spill-file cleanup. Seeds reproduce the workload and delay policy; thread +//! scheduling may still vary. +use std::pin::Pin; +use std::sync::Arc; +use std::time::Duration; + +use arrow::array::{ArrayRef, Int32Array}; +use arrow::datatypes::{DataType, Field, Schema, SchemaRef}; +use arrow::record_batch::RecordBatch; use datafusion_common::Result; -use datafusion_physical_plan::spill::spill_pool::run_spill_pool_scenario; +use datafusion_common::instant::Instant; +use datafusion_execution::disk_manager::{ + DiskManager, DiskManagerBuilder, DiskManagerMode, +}; +use datafusion_execution::runtime_env::RuntimeEnvBuilder; +use datafusion_execution::{SpillFile, SpillWriter, TempFileFactory}; +use datafusion_physical_plan::metrics::{ExecutionPlanMetricsSet, SpillMetrics}; +use datafusion_physical_plan::spill::SpillManager; +use datafusion_physical_plan::spill::spill_pool::{SpillPoolSink, mpsc_channel}; +use futures::{Stream, StreamExt}; +use parking_lot::Mutex; use test_utils::fuzz::FuzzTest; pub(crate) struct SpillPoolFuzzer; @@ -21,7 +40,371 @@ impl FuzzTest for SpillPoolFuzzer { std::future::ready(Ok(())) } - async fn run_case(&self, _: &(), case_seed: u64) -> Result<()> { - run_spill_pool_scenario(case_seed).await + async fn run_case(&self, _: &(), seed: u64) -> Result<()> { + use rand::rngs::StdRng; + use rand::{Rng, SeedableRng}; + + const READ_TIMEOUT: Duration = Duration::from_secs(5); + + let mut rng = StdRng::seed_from_u64(seed); + let n_writers = rng.random_range(1..=4); + let batch_bytes = batch_with_id(0, 10).get_array_memory_size(); + // Rotate after every batch, after a few batches, or never. + let max_file_size = match rng.random_range(0..3) { + 0 => batch_bytes / 2, + 1 => batch_bytes * 3, + _ => usize::MAX / 2, + }; + let n_phases = rng.random_range(1..=6); + let context = format!( + "seed {seed}: {n_writers} writers, max_file_size {max_file_size}, {n_phases} phases" + ); + + let pauser = Arc::new(PhasePauser::default()); + let write_hook: WriteHook = { + let pauser = Arc::clone(&pauser); + Arc::new(move || pauser.on_io()) + }; + // Some files are slow to read back, so an older file's bytes can arrive + // after a newer file's. Batches must still come back in file order for a + // single writer. + let read_hook: ReadHook = { + let rng = Mutex::new(StdRng::seed_from_u64(seed ^ 0x5EED_4EAD)); + Arc::new(move || { + let mut rng = rng.lock(); + if rng.random_bool(0.5) { + Duration::ZERO + } else { + Duration::from_millis(rng.random_range(1..=20)) + } + }) + }; + let (spill_manager, disk_manager) = + spill_manager_with_io_hooks(write_hook, read_hook)?; + let (writer, mut reader) = mpsc_channel(max_file_size, spill_manager); + let mut sinks: Vec> = + (0..n_writers).map(|_| Some(writer.new_sink())).collect(); + drop(writer); + + let mut pushed: Vec = vec![]; + let mut read: Vec = vec![]; + + for phase in 0..n_phases { + let alive: Vec = + (0..n_writers).filter(|w| sinks[*w].is_some()).collect(); + if alive.is_empty() { + break; + } + + // A random non-empty subset of the live writers each pushes a few + // batches, concurrently, held so that their pushes overlap. + let mut pushing: Vec = alive + .iter() + .copied() + .filter(|_| rng.random_bool(0.6)) + .collect(); + if pushing.is_empty() { + pushing.push(alive[rng.random_range(0..alive.len())]); + } + pauser.begin_phase(pushing.len()); + let mut threads = Vec::with_capacity(pushing.len()); + for w in pushing { + let sink = sinks[w].take().unwrap(); + let batches: Vec = (0..rng.random_range(1..=3)) + .map(|_| { + let id = pushed.len() as i32 + 1; + pushed.push(id); + batch_with_id(id, rng.random_range(1..=40)) + }) + .collect(); + threads.push(( + w, + std::thread::spawn(move || { + for batch in &batches { + sink.push_batch(batch).unwrap(); + } + sink + }), + )); + } + pauser.release_phase(); + for (w, thread) in threads { + sinks[w] = Some(thread.join().unwrap()); + } + + // Quiescent point: no push is in progress and none will happen until + // the reader is done. Every batch pushed so far must be readable now, + // in any amount. `RepartitionExec` relies on exactly this: it blocks + // on the reader once per pushed batch while its channel gate keeps + // the writers parked. Sometimes leave a backlog for later phases. + let read_now = rng.random_range(0..=pushed.len() - read.len()); + for _ in 0..read_now { + let next = tokio::time::timeout(READ_TIMEOUT, reader.next()) + .await + .unwrap_or_else(|_| { + panic!( + "{context}, phase {phase}: reader stalled with {} of {} \ + batches unread and {} writers alive", + pushed.len() - read.len(), + pushed.len(), + alive.len() + ) + }); + let batch = next.unwrap_or_else(|| { + panic!("{context}, phase {phase}: EOF while writers are alive") + })?; + read.push(id_of(&batch)); + } + + // With nothing left to read the reader must wait, not signal EOF. + if read.len() == pushed.len() && rng.random_bool(0.3) { + let probe = + tokio::time::timeout(Duration::from_millis(50), reader.next()).await; + assert!( + probe.is_err(), + "{context}, phase {phase}: reader yielded {probe:?} with nothing \ + pushed and writers alive" + ); + } + + // Dropping one of several writers must not disturb anything. + if alive.len() > 1 && rng.random_bool(0.3) { + sinks[alive[rng.random_range(0..alive.len())]] = None; + } + } + + // Once the last writer is gone the reader must hand out the backlog and + // then signal EOF. + sinks.clear(); + while read.len() < pushed.len() { + let next = tokio::time::timeout(READ_TIMEOUT, reader.next()) + .await + .unwrap_or_else(|_| { + panic!( + "{context}: reader stalled with {} of {} batches unread after \ + all writers were dropped", + pushed.len() - read.len(), + pushed.len() + ) + }); + let batch = next.unwrap_or_else(|| { + panic!( + "{context}: EOF with {} batches unread", + pushed.len() - read.len() + ) + })?; + read.push(id_of(&batch)); + } + let eof = tokio::time::timeout(READ_TIMEOUT, reader.next()) + .await + .unwrap_or_else(|_| { + panic!("{context}: reader stalled instead of signalling EOF after all writers were dropped") + }); + assert!(eof.is_none(), "{context}: batch after everything was read"); + + let mut expected = pushed.clone(); + expected.sort_unstable(); + let mut actual = read.clone(); + actual.sort_unstable(); + assert_eq!(actual, expected, "{context}: batches lost or duplicated"); + if n_writers == 1 { + assert_eq!(read, pushed, "{context}: single-writer pool must be FIFO"); + } + + drop(reader); + assert_eq!( + disk_manager.used_disk_space(), + 0, + "{context}: spill files not released" + ); + Ok(()) } } + +fn create_test_schema() -> SchemaRef { + Arc::new(Schema::new(vec![Field::new("a", DataType::Int32, false)])) +} + +type WriteHook = Arc; +type ReadHook = Arc Duration + Send + Sync>; + +/// Test double for a spill file that runs `write_hook` before every disk +/// write or flush, and consults `read_hook` before every read. Writers write +/// while holding their file's lock, so a hook that blocks pauses a +/// `push_batch` at exactly the point where it has a file checked out for +/// writing; a read delay makes an older file's bytes arrive after a newer +/// file's. Data still goes to real temporary files. +struct IoHookFactory { + inner: Arc, + write_hook: WriteHook, + read_hook: ReadHook, +} + +impl TempFileFactory for IoHookFactory { + fn create_temp_file(&self, description: &str) -> Result> { + Ok(Arc::new(IoHookFile { + inner: self.inner.create_tmp_file(description)?, + write_hook: Arc::clone(&self.write_hook), + read_hook: Arc::clone(&self.read_hook), + })) + } +} + +struct IoHookFile { + inner: Arc, + write_hook: WriteHook, + read_hook: ReadHook, +} + +impl SpillFile for IoHookFile { + fn path(&self) -> Option<&std::path::Path> { + self.inner.path() + } + + fn size(&self) -> Option { + self.inner.size() + } + + fn read_stream( + &self, + ) -> Result> + Send>>> { + let delay = (self.read_hook)(); + let mut inner = Some(self.inner.read_stream()?); + Ok(Box::pin( + futures::stream::once(tokio::time::sleep(delay)) + .flat_map(move |_| inner.take().expect("polled once")), + )) + } + + fn open_writer(&self) -> Result> { + Ok(Box::new(IoHookWriter { + inner: self.inner.open_writer()?, + hook: Arc::clone(&self.write_hook), + })) + } +} + +struct IoHookWriter { + inner: Box, + hook: WriteHook, +} + +impl std::io::Write for IoHookWriter { + fn write(&mut self, buf: &[u8]) -> std::io::Result { + (self.hook)(); + self.inner.write(buf) + } + + fn flush(&mut self) -> std::io::Result<()> { + (self.hook)(); + self.inner.flush() + } +} + +impl SpillWriter for IoHookWriter { + fn finish(&mut self) -> Result<()> { + self.inner.finish() + } +} + +/// A `SpillManager` whose files run `write_hook` before every disk write and +/// consult `read_hook` before every read, plus the `DiskManager` that owns +/// the files so tests can check disk usage. +fn spill_manager_with_io_hooks( + write_hook: WriteHook, + read_hook: ReadHook, +) -> Result<(Arc, Arc)> { + let disk_manager = Arc::new(DiskManagerBuilder::default().build()?); + let runtime = RuntimeEnvBuilder::new() + .with_disk_manager_builder(DiskManagerBuilder::default().with_mode( + DiskManagerMode::Custom(Arc::new(IoHookFactory { + inner: Arc::clone(&disk_manager), + write_hook, + read_hook, + })), + )) + .build_arc()?; + let metrics = SpillMetrics::new(&ExecutionPlanMetricsSet::new(), 0); + let spill_manager = + Arc::new(SpillManager::new(runtime, metrics, create_test_schema())); + Ok((spill_manager, disk_manager)) +} + +/// Pauses every writer of a "phase" inside its first disk write, so that +/// several `push_batch` calls overlap at the point where each holds a file +/// checked out for writing. A writer that blocks earlier (for example on +/// another writer's file lock) never arrives; `release_phase` stops waiting +/// for it after a short while. +#[derive(Default)] +struct PhasePauser { + state: Mutex, + changed: parking_lot::Condvar, +} + +#[derive(Default)] +struct PhaseState { + phase: u64, + expected: usize, + held: usize, + released: bool, +} + +thread_local! { + /// The phase in which the current writer thread has already been held. + static HELD_IN_PHASE: std::cell::Cell = const { std::cell::Cell::new(0) }; +} + +impl PhasePauser { + fn begin_phase(&self, expected: usize) { + let mut state = self.state.lock(); + state.phase += 1; + state.expected = expected; + state.held = 0; + state.released = false; + } + + /// Called by writer threads before every disk write. + fn on_io(&self) { + let mut state = self.state.lock(); + if state.released + || state.held >= state.expected + || HELD_IN_PHASE.get() == state.phase + { + return; + } + HELD_IN_PHASE.set(state.phase); + state.held += 1; + self.changed.notify_all(); + while !state.released { + self.changed.wait(&mut state); + } + } + + /// Waits until every writer of the phase is held (or 300 ms have + /// passed), then releases them all at once. + fn release_phase(&self) { + let mut state = self.state.lock(); + let deadline = Instant::now() + Duration::from_millis(300); + while state.held < state.expected { + if self.changed.wait_until(&mut state, deadline).timed_out() { + break; + } + } + state.released = true; + self.changed.notify_all(); + } +} + +fn batch_with_id(id: i32, rows: usize) -> RecordBatch { + let a: ArrayRef = Arc::new(Int32Array::from(vec![id; rows])); + RecordBatch::try_new(create_test_schema(), vec![a]).unwrap() +} + +fn id_of(batch: &RecordBatch) -> i32 { + batch + .column(0) + .as_any() + .downcast_ref::() + .unwrap() + .value(0) +} diff --git a/datafusion/physical-plan/Cargo.toml b/datafusion/physical-plan/Cargo.toml index 8572f2320e4a9..82a2215581f22 100644 --- a/datafusion/physical-plan/Cargo.toml +++ b/datafusion/physical-plan/Cargo.toml @@ -39,7 +39,7 @@ workspace = true [features] force_hash_collisions = [] -test_utils = ["arrow/test_utils", "dep:rand"] +test_utils = ["arrow/test_utils"] tokio_coop = [] tokio_coop_fallback = [] # Enables `PhysicalExpr::try_to_proto` / `try_from_proto` hooks on the @@ -87,7 +87,6 @@ log = { workspace = true } num-traits = { workspace = true } parking_lot = { workspace = true } pin-project-lite = { workspace = true } -rand = { workspace = true, optional = true } serde = { workspace = true } serde_json = { workspace = true } tokio = { workspace = true, features = ["time"] } diff --git a/datafusion/physical-plan/src/spill/spill_pool.rs b/datafusion/physical-plan/src/spill/spill_pool.rs index a38d2ffdb9222..f1a426ccd59bd 100644 --- a/datafusion/physical-plan/src/spill/spill_pool.rs +++ b/datafusion/physical-plan/src/spill/spill_pool.rs @@ -814,40 +814,23 @@ impl RecordBatchStream for SpillPoolReader { } } -/// Internal fuzz entrypoint for the centralized test driver. -/// -/// This is not a public API; see [API policy] for details. -/// -/// [API policy]: https://datafusion.apache.org/contributor-guide/api-health.html#internal-public-apis -#[cfg(any(test, feature = "test_utils"))] -#[doc(hidden)] -pub use tests::run_spill_pool_scenario; - -#[cfg(any(test, feature = "test_utils"))] +#[cfg(test)] mod tests { use super::*; use crate::metrics::{ExecutionPlanMetricsSet, SpillMetrics}; use arrow::array::{ArrayRef, Int32Array}; use arrow::datatypes::{DataType, Field, Schema}; - use datafusion_common::DataFusionError; - #[cfg(test)] - use datafusion_common::exec_datafusion_err; - use datafusion_common::instant::Instant; - #[cfg(test)] + use datafusion_common::{DataFusionError, exec_datafusion_err}; use datafusion_common_runtime::{JoinSet, SpawnedTask}; use datafusion_execution::disk_manager::{ DiskManager, DiskManagerBuilder, DiskManagerMode, }; - #[cfg(test)] use datafusion_execution::runtime_env::RuntimeEnv; use datafusion_execution::runtime_env::RuntimeEnvBuilder; use datafusion_execution::{SpillFile, SpillWriter, TempFileFactory}; use std::pin::Pin; - #[cfg(test)] use std::sync::Barrier; - #[cfg(test)] use std::sync::atomic::{AtomicBool, Ordering}; - #[cfg(test)] use std::sync::mpsc::{self, Sender}; use std::time::Duration; @@ -855,7 +838,6 @@ mod tests { Arc::new(Schema::new(vec![Field::new("a", DataType::Int32, false)])) } - #[cfg(test)] fn create_test_batch(start: i32, count: usize) -> RecordBatch { let schema = create_test_schema(); let a: ArrayRef = Arc::new(Int32Array::from( @@ -864,7 +846,6 @@ mod tests { RecordBatch::try_new(schema, vec![a]).unwrap() } - #[cfg(test)] fn create_spill_channel( max_file_size: usize, ) -> (SpillPoolSink, SendableRecordBatchStream) { @@ -876,7 +857,6 @@ mod tests { spsc_channel(max_file_size, spill_manager) } - #[cfg(test)] fn create_shared_spill_channel( max_file_size: usize, ) -> (SpillPoolWriter, SendableRecordBatchStream) { @@ -888,7 +868,6 @@ mod tests { mpsc_channel(max_file_size, spill_manager) } - #[cfg(test)] fn create_spill_channel_with_metrics( max_file_size: usize, ) -> (SpillPoolSink, SendableRecordBatchStream, SpillMetrics) { @@ -1320,7 +1299,6 @@ mod tests { Ok(()) } - #[cfg(test)] #[tokio::test(flavor = "multi_thread", worker_threads = 10)] async fn test_concurrent_writers() -> Result<()> { let (writer, mut reader) = create_shared_spill_channel(1024 * 1024); @@ -1711,10 +1689,6 @@ mod tests { /// different error paths of [`SpillPoolFile::poll_file`]: one fails while it /// builds the stream, the other fails while it polls a stream that is /// already built. - #[cfg_attr( - not(test), - expect(dead_code, reason = "Read failures are injected by unit tests") - )] enum ReadBehavior { /// Read the real bytes, after this delay. Delay(Duration), @@ -1838,7 +1812,6 @@ mod tests { /// Holds the first disk write on the pool until the test releases it, and /// reports every other write so a test can wait for a second writer to /// reach a file of its own instead of sleeping. - #[cfg(test)] struct WriteGate { taken: AtomicBool, entered: Barrier, @@ -1846,7 +1819,6 @@ mod tests { other_writes: Sender<()>, } - #[cfg(test)] impl WriteGate { fn hold_if_first(&self) { if self.taken.swap(true, Ordering::SeqCst) { @@ -1997,76 +1969,6 @@ mod tests { Ok(()) } - /// Pauses every writer of a "phase" inside its first disk write, so that - /// several `push_batch` calls overlap at the point where each holds a file - /// checked out for writing. A writer that blocks earlier (for example on - /// another writer's file lock) never arrives; `release_phase` stops waiting - /// for it after a short while. - #[derive(Default)] - struct PhasePauser { - state: Mutex, - changed: parking_lot::Condvar, - } - - #[derive(Default)] - struct PhaseState { - phase: u64, - expected: usize, - held: usize, - released: bool, - } - - thread_local! { - /// The phase in which the current writer thread has already been held. - static HELD_IN_PHASE: std::cell::Cell = const { std::cell::Cell::new(0) }; - } - - impl PhasePauser { - fn begin_phase(&self, expected: usize) { - let mut state = self.state.lock(); - state.phase += 1; - state.expected = expected; - state.held = 0; - state.released = false; - } - - /// Called by writer threads before every disk write. - fn on_io(&self) { - let mut state = self.state.lock(); - if state.released - || state.held >= state.expected - || HELD_IN_PHASE.get() == state.phase - { - return; - } - HELD_IN_PHASE.set(state.phase); - state.held += 1; - self.changed.notify_all(); - while !state.released { - self.changed.wait(&mut state); - } - } - - /// Waits until every writer of the phase is held (or 300 ms have - /// passed), then releases them all at once. - fn release_phase(&self) { - let mut state = self.state.lock(); - let deadline = Instant::now() + Duration::from_millis(300); - while state.held < state.expected { - if self.changed.wait_until(&mut state, deadline).timed_out() { - break; - } - } - state.released = true; - self.changed.notify_all(); - } - } - - fn batch_with_id(id: i32, rows: usize) -> RecordBatch { - let a: ArrayRef = Arc::new(Int32Array::from(vec![id; rows])); - RecordBatch::try_new(create_test_schema(), vec![a]).unwrap() - } - fn id_of(batch: &RecordBatch) -> i32 { batch .column(0) @@ -2075,187 +1977,4 @@ mod tests { .unwrap() .value(0) } - - /// Run one seeded scenario checking concurrent writes, reader progress, - /// ordering, and spill-file cleanup. Seeds reproduce the logical workload - /// and delay policy; thread scheduling may still vary. - pub async fn run_spill_pool_scenario(seed: u64) -> Result<()> { - use rand::rngs::StdRng; - use rand::{Rng, SeedableRng}; - - const READ_TIMEOUT: Duration = Duration::from_secs(5); - - let mut rng = StdRng::seed_from_u64(seed); - let n_writers = rng.random_range(1..=4); - let batch_bytes = batch_with_id(0, 10).get_array_memory_size(); - // Rotate after every batch, after a few batches, or never. - let max_file_size = match rng.random_range(0..3) { - 0 => batch_bytes / 2, - 1 => batch_bytes * 3, - _ => usize::MAX / 2, - }; - let n_phases = rng.random_range(1..=6); - let context = format!( - "seed {seed}: {n_writers} writers, max_file_size {max_file_size}, {n_phases} phases" - ); - - let pauser = Arc::new(PhasePauser::default()); - let write_hook: WriteHook = { - let pauser = Arc::clone(&pauser); - Arc::new(move || pauser.on_io()) - }; - // Some files are slow to read back, so an older file's bytes can arrive - // after a newer file's. Batches must still come back in file order for a - // single writer. - let read_hook: ReadHook = { - let rng = Mutex::new(StdRng::seed_from_u64(seed ^ 0x5EED_4EAD)); - Arc::new(move || { - let mut rng = rng.lock(); - if rng.random_bool(0.5) { - ReadBehavior::Delay(Duration::ZERO) - } else { - ReadBehavior::Delay(Duration::from_millis(rng.random_range(1..=20))) - } - }) - }; - let (spill_manager, disk_manager) = - spill_manager_with_io_hooks(write_hook, read_hook)?; - let (writer, mut reader) = mpsc_channel(max_file_size, spill_manager); - let mut sinks: Vec> = - (0..n_writers).map(|_| Some(writer.new_sink())).collect(); - drop(writer); - - let mut pushed: Vec = vec![]; - let mut read: Vec = vec![]; - - for phase in 0..n_phases { - let alive: Vec = - (0..n_writers).filter(|w| sinks[*w].is_some()).collect(); - if alive.is_empty() { - break; - } - - // A random non-empty subset of the live writers each pushes a few - // batches, concurrently, held so that their pushes overlap. - let mut pushing: Vec = alive - .iter() - .copied() - .filter(|_| rng.random_bool(0.6)) - .collect(); - if pushing.is_empty() { - pushing.push(alive[rng.random_range(0..alive.len())]); - } - pauser.begin_phase(pushing.len()); - let mut threads = Vec::with_capacity(pushing.len()); - for w in pushing { - let sink = sinks[w].take().unwrap(); - let batches: Vec = (0..rng.random_range(1..=3)) - .map(|_| { - let id = pushed.len() as i32 + 1; - pushed.push(id); - batch_with_id(id, rng.random_range(1..=40)) - }) - .collect(); - threads.push(( - w, - std::thread::spawn(move || { - for batch in &batches { - sink.push_batch(batch).unwrap(); - } - sink - }), - )); - } - pauser.release_phase(); - for (w, thread) in threads { - sinks[w] = Some(thread.join().unwrap()); - } - - // Quiescent point: no push is in progress and none will happen until - // the reader is done. Every batch pushed so far must be readable now, - // in any amount. `RepartitionExec` relies on exactly this: it blocks - // on the reader once per pushed batch while its channel gate keeps - // the writers parked. Sometimes leave a backlog for later phases. - let read_now = rng.random_range(0..=pushed.len() - read.len()); - for _ in 0..read_now { - let next = tokio::time::timeout(READ_TIMEOUT, reader.next()) - .await - .unwrap_or_else(|_| { - panic!( - "{context}, phase {phase}: reader stalled with {} of {} \ - batches unread and {} writers alive", - pushed.len() - read.len(), - pushed.len(), - alive.len() - ) - }); - let batch = next.unwrap_or_else(|| { - panic!("{context}, phase {phase}: EOF while writers are alive") - })?; - read.push(id_of(&batch)); - } - - // With nothing left to read the reader must wait, not signal EOF. - if read.len() == pushed.len() && rng.random_bool(0.3) { - let probe = - tokio::time::timeout(Duration::from_millis(50), reader.next()).await; - assert!( - probe.is_err(), - "{context}, phase {phase}: reader yielded {probe:?} with nothing \ - pushed and writers alive" - ); - } - - // Dropping one of several writers must not disturb anything. - if alive.len() > 1 && rng.random_bool(0.3) { - sinks[alive[rng.random_range(0..alive.len())]] = None; - } - } - - // Once the last writer is gone the reader must hand out the backlog and - // then signal EOF. - sinks.clear(); - while read.len() < pushed.len() { - let next = tokio::time::timeout(READ_TIMEOUT, reader.next()) - .await - .unwrap_or_else(|_| { - panic!( - "{context}: reader stalled with {} of {} batches unread after \ - all writers were dropped", - pushed.len() - read.len(), - pushed.len() - ) - }); - let batch = next.unwrap_or_else(|| { - panic!( - "{context}: EOF with {} batches unread", - pushed.len() - read.len() - ) - })?; - read.push(id_of(&batch)); - } - let eof = tokio::time::timeout(READ_TIMEOUT, reader.next()) - .await - .unwrap_or_else(|_| { - panic!("{context}: reader stalled instead of signalling EOF after all writers were dropped") - }); - assert!(eof.is_none(), "{context}: batch after everything was read"); - - let mut expected = pushed.clone(); - expected.sort_unstable(); - let mut actual = read.clone(); - actual.sort_unstable(); - assert_eq!(actual, expected, "{context}: batches lost or duplicated"); - if n_writers == 1 { - assert_eq!(read, pushed, "{context}: single-writer pool must be FIFO"); - } - - drop(reader); - assert_eq!( - disk_manager.used_disk_space(), - 0, - "{context}: spill files not released" - ); - Ok(()) - } } From e205a1b0a62f910a32bba969692c1e6fab43a1d5 Mon Sep 17 00:00:00 2001 From: Yongting You <2010youy01@gmail.com> Date: Wed, 4 Nov 2026 15:08:40 +0800 Subject: [PATCH 4/6] review: add tracking issue in comments --- .github/workflows/extended.yml | 1 + datafusion/fuzz/tests/fuzz.rs | 4 ++++ test-utils/src/fuzz.rs | 6 ++++-- 3 files changed, 9 insertions(+), 2 deletions(-) diff --git a/.github/workflows/extended.yml b/.github/workflows/extended.yml index f6b2a55efe68e..b6a328cdfa1a0 100644 --- a/.github/workflows/extended.yml +++ b/.github/workflows/extended.yml @@ -87,6 +87,7 @@ jobs: - name: Run fuzz tests run: cargo xtask ci step test fuzz # Migrate it to the above unified fuzz runner over time + # Tracker: https://github.com/apache/datafusion/issues/27082 - name: Run unmigrated fuzzers with more iterations run: cargo xtask ci step test fuzz-legacy - parallel: diff --git a/datafusion/fuzz/tests/fuzz.rs b/datafusion/fuzz/tests/fuzz.rs index cba3ed62fb749..6eae5f9127eb5 100644 --- a/datafusion/fuzz/tests/fuzz.rs +++ b/datafusion/fuzz/tests/fuzz.rs @@ -13,6 +13,10 @@ use test_utils::fuzz::{FuzzTest, RunConfig, run}; mod suites; /// The single registration point for project-wide fuzz campaigns. +/// +/// Migration of the remaining fuzz tests is tracked in [#27082]. +/// +/// [#27082]: https://github.com/apache/datafusion/issues/27082 #[tokio::test(flavor = "multi_thread")] async fn fuzz_tests() -> Result<()> { let spill_pool = suites::spill_pool::SpillPoolFuzzer; diff --git a/test-utils/src/fuzz.rs b/test-utils/src/fuzz.rs index f42fdace53adf..2796119102bc6 100644 --- a/test-utils/src/fuzz.rs +++ b/test-utils/src/fuzz.rs @@ -10,17 +10,19 @@ use std::time::Duration; use datafusion_common::instant::Instant; use datafusion_common::{Result, exec_err}; -/// A component's fuzz experiment, run by the centralized driver. +/// A component's fuzz experiment, run by the [centralized driver]. /// /// The component prepares fixtures and checks one case; the driver handles /// repetition, budgets, and seed reporting. Each case can be replayed alone. /// -/// See the driver in `datafusion-fuzz` for how to use it. At a high level: +/// See the [centralized driver] for how to use it. At a high level: /// 1. Generate one dataset with `prepare_dataset()`. /// 2. Run several cases on that dataset with `run_case()`. /// 3. Repeat steps 1 and 2 until the time budget is reached. /// /// (All configuration, such as the total time budget, lives in [`RunConfig`].) +/// +/// [centralized driver]: https://github.com/apache/datafusion/blob/main/datafusion/fuzz/tests/fuzz.rs #[expect(async_fn_in_trait)] // Serial execution deliberately does not require Send futures. pub trait FuzzTest { /// Reusable fixtures; `()` is valid when no preparation can be shared. From 00019d388671bab2af157e2c47b04f127b8b8adf Mon Sep 17 00:00:00 2001 From: Yongting You <2010youy01@gmail.com> Date: Wed, 4 Nov 2026 15:20:47 +0800 Subject: [PATCH 5/6] fix lint --- datafusion/fuzz/Cargo.toml | 19 +++---------------- 1 file changed, 3 insertions(+), 16 deletions(-) diff --git a/datafusion/fuzz/Cargo.toml b/datafusion/fuzz/Cargo.toml index 8ad5c759ec21d..661c4be58b276 100644 --- a/datafusion/fuzz/Cargo.toml +++ b/datafusion/fuzz/Cargo.toml @@ -1,19 +1,6 @@ -# 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. +# SPDX-License-Identifier: Apache-2.0 +# SPDX-FileCopyrightText: See the NOTICE file distributed with this work for additional information regarding copyright ownership +# SPDX-FileContributor: Licensed to the Apache Software Foundation (ASF) under one or more contributor license agreements # A private crate providing a single entry point for all fuzz tests. From a35deb525e83c5620e8c879bb9b3fb0ecd34ee73 Mon Sep 17 00:00:00 2001 From: Yongting You <2010youy01@gmail.com> Date: Wed, 4 Nov 2026 15:27:38 +0800 Subject: [PATCH 6/6] improve wording --- test-utils/src/fuzz.rs | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/test-utils/src/fuzz.rs b/test-utils/src/fuzz.rs index 2796119102bc6..f932f75f4fb34 100644 --- a/test-utils/src/fuzz.rs +++ b/test-utils/src/fuzz.rs @@ -10,10 +10,10 @@ use std::time::Duration; use datafusion_common::instant::Instant; use datafusion_common::{Result, exec_err}; -/// A component's fuzz experiment, run by the [centralized driver]. +/// Implement this trait to add a fuzzing unit test to DataFusion. /// -/// The component prepares fixtures and checks one case; the driver handles -/// repetition, budgets, and seed reporting. Each case can be replayed alone. +/// The fuzz implementation prepares the data and checks one case. The [centralized driver] +/// repeats cases, manages the time budget, and reports failures. /// /// See the [centralized driver] for how to use it. At a high level: /// 1. Generate one dataset with `prepare_dataset()`.