SortQueryFuzzer found a failing case on main

Describe the bug

Fuzzer failed during an unrelated change - https://github.com/apache/datafusion/actions/runs/15741542523/job/44367876525?pr=16449.

Not sure how long GitHub retains logs, so I'm also adding the actual output:

---- fuzz_cases::sort_query_fuzz::sort_query_fuzzer_runner stdout ----
[SortQueryFuzzer] Round 0, Query 0 (Config 0)
  Seeds:
    init_seed   = 10313160656544581998
    query_seed  = 9760675044397831413
    config_seed = 2456252688354296803
  Dataset schema:
    [u8_low:UInt8;N, binary:Binary;N, timestamp_us:Timestamp(Microsecond, None);N, time64_ns:Time64(Nanosecond);N, u8:UInt8;N, i32:Int32;N, duration_nanosecond:Duration(Nanosecond);N, utf8_low:Utf8;N, duration_milliseconds:Duration(Millisecond);N]
  Query:
    SELECT * FROM sort_fuzz_table ORDER BY time64_ns ASC
  Config: 
    Dataset size: 286.6 KB
    Number of partitions: 3
    Batch size: 6
    Memory limit: Unbounded
    Per partition memory limit: Unbounded
    Sort spill reservation bytes: 48.4 KB
    Sort in place threshold bytes: 2.3 KB


[SortQueryFuzzer] Round 0, Query 0 (Config 1)
  Seeds:
    init_seed   = 10313160656544581998
    query_seed  = 9760675044397831413
    config_seed = 13112719340419033750
  Dataset schema:
    [u8_low:UInt8;N, binary:Binary;N, timestamp_us:Timestamp(Microsecond, None);N, time64_ns:Time64(Nanosecond);N, u8:UInt8;N, i32:Int32;N, duration_nanosecond:Duration(Nanosecond);N, utf8_low:Utf8;N, duration_milliseconds:Duration(Millisecond);N]
  Query:
    SELECT * FROM sort_fuzz_table ORDER BY time64_ns ASC
  Config: 
    Dataset size: 286.6 KB
    Number of partitions: 3
    Batch size: 6
    Memory limit: Unbounded
    Per partition memory limit: Unbounded
    Sort spill reservation bytes: 42.3 KB
    Sort in place threshold bytes: 3.7 KB


[SortQueryFuzzer] Round 0, Query 0 (Config 2)
  Seeds:
    init_seed   = 10313160656544581998
    query_seed  = 9760675044397831413
    config_seed = 11553131583763959573
  Dataset schema:
    [u8_low:UInt8;N, binary:Binary;N, timestamp_us:Timestamp(Microsecond, None);N, time64_ns:Time64(Nanosecond);N, u8:UInt8;N, i32:Int32;N, duration_nanosecond:Duration(Nanosecond);N, utf8_low:Utf8;N, duration_milliseconds:Duration(Millisecond);N]
  Query:
    SELECT * FROM sort_fuzz_table ORDER BY time64_ns ASC
  Config: 
    Dataset size: 286.6 KB
    Number of partitions: 3
    Batch size: 6
    Memory limit: Unbounded
    Per partition memory limit: Unbounded
    Sort spill reservation bytes: 21.4 KB
    Sort in place threshold bytes: 486.0 B


[SortQueryFuzzer] Round 0, Query 0 (Config 3)
  Seeds:
    init_seed   = 10313160656544581998
    query_seed  = 9760675044397831413
    config_seed = 7181[2061](https://github.com/apache/datafusion/actions/runs/15741542523/job/44367876525?pr=16449#step:4:2062)06220866377
  Dataset schema:
    [u8_low:UInt8;N, binary:Binary;N, timestamp_us:Timestamp(Microsecond, None);N, time64_ns:Time64(Nanosecond);N, u8:UInt8;N, i32:Int32;N, duration_nanosecond:Duration(Nanosecond);N, utf8_low:Utf8;N, duration_milliseconds:Duration(Millisecond);N]
  Query:
    SELECT * FROM sort_fuzz_table ORDER BY time64_ns ASC
  Config: 
    Dataset size: 286.6 KB
    Number of partitions: 3
    Batch size: 6
    Memory limit: Unbounded
    Per partition memory limit: Unbounded
    Sort spill reservation bytes: 43.8 KB
    Sort in place threshold bytes: 3.2 KB


[SortQueryFuzzer] Round 0, Query 0 (Config 4)
  Seeds:
    init_seed   = 10313160656544581998
    query_seed  = 9760675044397831413
    config_seed = 17269272874247070475
  Dataset schema:
    [u8_low:UInt8;N, binary:Binary;N, timestamp_us:Timestamp(Microsecond, None);N, time64_ns:Time64(Nanosecond);N, u8:UInt8;N, i32:Int32;N, duration_nanosecond:Duration(Nanosecond);N, utf8_low:Utf8;N, duration_milliseconds:Duration(Millisecond);N]
  Query:
    SELECT * FROM sort_fuzz_table ORDER BY time64_ns ASC
  Config: 
    Dataset size: 286.6 KB
    Number of partitions: 3
    Batch size: 6
    Memory limit: Unbounded
    Per partition memory limit: Unbounded
    Sort spill reservation bytes: 45.0 KB
    Sort in place threshold bytes: 2.3 KB


[SortQueryFuzzer] Round 0, Query 1 (Config 0)
  Seeds:
    init_seed   = 10313160656544581998
    query_seed  = 15004039071976572201
    config_seed = 11807432710583113300
  Dataset schema:
    [u8_low:UInt8;N, binary:Binary;N, timestamp_us:Timestamp(Microsecond, None);N, time64_ns:Time64(Nanosecond);N, u8:UInt8;N, i32:Int32;N, duration_nanosecond:Duration(Nanosecond);N, utf8_low:Utf8;N, duration_milliseconds:Duration(Millisecond);N]
  Query:
    SELECT * FROM sort_fuzz_table ORDER BY timestamp_us DESC LIMIT 3
  Config: 
    Dataset size: 286.6 KB
    Number of partitions: 3
    Batch size: 6
    Memory limit: Unbounded
    Per partition memory limit: Unbounded
    Sort spill reservation bytes: 25.8 KB
    Sort in place threshold bytes: 2.8 KB


[SortQueryFuzzer] Round 0, Query 1 (Config 1)
  Seeds:
    init_seed   = 10313160656544581998
    query_seed  = 15004039071976572201
    config_seed = 759937414670321802
  Dataset schema:
    [u8_low:UInt8;N, binary:Binary;N, timestamp_us:Timestamp(Microsecond, None);N, time64_ns:Time64(Nanosecond);N, u8:UInt8;N, i32:Int32;N, duration_nanosecond:Duration(Nanosecond);N, utf8_low:Utf8;N, duration_milliseconds:Duration(Millisecond);N]
  Query:
    SELECT * FROM sort_fuzz_table ORDER BY timestamp_us DESC LIMIT 3
  Config: 
    Dataset size: 286.6 KB
    Number of partitions: 3
    Batch size: 6
    Memory limit: Unbounded
    Per partition memory limit: Unbounded
    Sort spill reservation bytes: 47.4 KB
    Sort in place threshold bytes: 2.5 KB



thread 'fuzz_cases::sort_query_fuzz::sort_query_fuzzer_runner' panicked at datafusion/core/tests/fuzz_cases/sort_query_fuzz.rs:232:71:
called `Result::unwrap()` on an `Err` value: InconsistentResult { row_idx: 0, lhs_row: "+--------+--------------------------------------+--------------+---------------------------------------------------------------------------------------------+-----+-----------+-------------------------+-------------+-------------------------+", rhs_row: "+--------+------------------------------------------------+--------------+---------------------------------------------------------------------------------------------+-----+-------------+--------------------------+-------------+--------------------------+" }

To Reproduce

No response

Expected behavior

No response

Additional context

No response

Thanks for reporting!

~~This failure should be reproducable by~~ I didn't check it carefully and this reproducer is incorrect...See #16452 (comment) for the correct reproducer.

    /// Reproduce the bug with specific seeds from the failing test case
    #[tokio::test]
    async fn test_reproduce_sort_query_bug() {
        // Seeds from the failing test case
        let init_seed = 10313160656544581998u64;
        let query_seed = 15004039071976572201u64;
        let config_seed_1 = 11807432710583113300u64;
        let config_seed_2 = 759937414670321802u64;

        // Use a fixed seed to replicate the original behavior more closely
        let random_seed = 1u64; // Use a fixed seed to ensure consistent behavior

        println!("Creating test generator with same config as original runner...");
        let test_generator = SortFuzzerTestGenerator::new(
            2000,
            3,
            "sort_fuzz_table".to_string(),
            get_supported_types_columns(random_seed),
            false,
            random_seed,
        );

        // Create a fuzzer with the same configuration as the original
        let mut fuzzer = SortQueryFuzzer::new(random_seed)
            .with_max_rounds(Some(1))
            .with_queries_per_round(2)
            .with_config_variations_per_query(2)
            .with_test_generator(test_generator);

        // Try to run and catch any errors
        match fuzzer.run().await {
            Ok(_) => println!("Fuzzer completed successfully - bug may have been fixed"),
            Err(e) => {
                println!("Fuzzer failed with error: {}", e);
                panic!("Reproduced the bug: {}", e);
            }
        }
    }

However, I’ve run it several times and it always passes.
This could be either:

a bug in the fuzzer, or
a heisenbug that depends on Tokio’s scheduling order.

I got a very similar failure in #16447 - https://github.com/apache/datafusion/actions/runs/15744174093/job/44376646866?pr=16447

Thanks for the repro! I'll try and spend some time hacking on it to see if I can reproduce it, maybe trying out loom will help

@adriangb

Perhaps this is related to the recent work from @adriangb with topk limit pushdown (once we figure out how to reliably reproduce it, we can verify by turning off the feature)

I've built a repro that fails consistently, and the most interesting part of it is that when using the default test runtime (which is a tokio current thread variant), the test passes.
It only fails with the multithreaded runtime which makes me think this is indeed a tokio futures interleaving issue. I'll give it some more time later tonight.

#[tokio::test(flavor = "multi_thread")]
async fn test_reproduce_sort_query_issue_16452() {
    // Seeds from the failing test case
    let init_seed = 10313160656544581998u64;
    let query_seed = 15004039071976572201u64;
    let config_seed_1 = 11807432710583113300u64;
    let config_seed_2 = 759937414670321802u64;

    // Use a fixed seed to replicate the original behavior more closely
    let random_seed = 1u64; // Use a fixed seed to ensure consistent behavior

    println!("Creating test generator with same config as original runner...");
    let mut test_generator = SortFuzzerTestGenerator::new(
        2000,
        3,
        "sort_fuzz_table".to_string(),
        get_supported_types_columns(random_seed),
        false,
        random_seed,
    );

    let mut results = vec![];

    for config_seed in [config_seed_1, config_seed_2] {
        let r = test_generator
            .fuzzer_run(init_seed, query_seed, config_seed)
            .await
            .unwrap();

        results.push(r);
    }

    for (lhs, rhs) in results.iter().tuple_windows() {
        check_equality_of_batches(lhs, rhs).unwrap();
    }
}

Btw to disable the new topk changes you can set datafusion.optimizer.enable_dynamic_filter_pushdown = false

Some more findings:

datafusion.optimizer.enable_dynamic_filter_pushdown doesn't seem to make a difference
Played around with the seeds, seems like the only one that's important to reproduce the issue is query_seed, it doesn't reproduce every time with it but changing it seems to make it to not reproduce in a reasonable time. The query it generates is:

SELECT * FROM sort_fuzz_table ORDER BY interval_month_day_nano DESC LIMIT 3;

with interval_month_day_nano's being a nullable Interval(MonthDayNano).

I also took a deeper look at the actual data and noticed some surprising things that might explain the failure:

The top values in the column the query sorts on are often null, which makes me think there's a sort stability issue here (the implementation of check_equality_of_batches also points in that direction).
A lot of the value in the table isn't really valid? when displayed we get a lot of values that are conversion errors of numbers into all kind of temporal types, like Cast error: Failed to convert -6727098022243200000 to temporal for Date64

I wonder if what's happening all comes down to an unstable sort + when running on a multithreaded runtime events interleave in different ways which result in different overall outcomes.

I think this is a reproducer? https://github.com/apache/datafusion/actions/runs/15764484967/job/44438090077?pr=16461

@Dandandan

datafusion.optimizer.enable_dynamic_filter_pushdown doesn't seem to make a difference

@Dandandan I wonder if it's the filtering being done inside of the TopK?

@AdamGS could you try commenting out these lines?

datafusion/datafusion/physical-plan/src/topk/mod.rs

Lines 217 to 251 in 2d7ae09

    
           if let Some(filter) = self.filter.as_ref() { 
        
               // If a filter is provided, update it with the new rows 
        
               let filter = filter.current()?; 
        
               let filtered = filter.evaluate(&batch)?; 
        
               let num_rows = batch.num_rows(); 
        
               let array = filtered.into_array(num_rows)?; 
        
               let mut filter = array.as_boolean().clone(); 
        
               let true_count = filter.true_count(); 
        
               if true_count == 0 { 
        
                   // nothing to filter, so no need to update 
        
                   return Ok(()); 
        
               } 
        
               // only update the keys / rows if the filter does not match all rows 
        
               if true_count < num_rows { 
        
                   // Indices in `set_indices` should be correct if filter contains nulls 
        
                   // So we prepare the filter here. Note this is also done in the `FilterBuilder` 
        
                   // so there is no overhead to do this here. 
        
                   if filter.nulls().is_some() { 
        
                       filter = prep_null_mask_filter(&filter); 
        
                   } 
        
                   let filter_predicate = FilterBuilder::new(&filter); 
        
                   let filter_predicate = if sort_keys.len() > 1 { 
        
                       // Optimize filter when it has multiple sort keys 
        
                       filter_predicate.optimize().build() 
        
                   } else { 
        
                       filter_predicate.build() 
        
                   }; 
        
                   selected_rows = Some(filter); 
        
                   sort_keys = sort_keys 
        
                       .iter() 
        
                       .map(|key| filter_predicate.filter(key).map_err(|x| x.into())) 
        
                       .collect::<Result<Vec<_>>>()?; 
        
               } 
        
           };

that indeed makes the failure to go away

Let's merge that ASAP. I'm AFK for the next two hours? Could you prepare a PR by chance?

Fuzzer-based reproduction and a temporary fix for apache#16452

done - https://github.com/apache/datafusion/pull/16465/files

@alamb

For transparency, @alamb noticed another failure: https://github.com/apache/datafusion/actions/runs/15780623790/job/44485106315

Right -- so in summary

We tried to back out what we thought was the offending code: Temporarily fix bug in dynamic top-k optimization #16465
However, when we started running the tests again on main it failed: Revert "Ignore sort_query_fuzzer_runner (#16462)" #16470
So we disabled sort_query_fuzzer_runner again: Reapply "Ignore sort_query_fuzzer_runner (#16462)" (#16470) #16485

I'm continuing to investigate. Unfortunately this is very flaky and slow to test and I've run out of time for today. I will continue tomorrow morning.

@alamb

I can confirm it's related to #15770. It seems like there may be multiple errors. This is one I found tonight:

thread 'tokio-runtime-worker' panicked at /.cargo/registry/src/index.crates.io-1949cf8c6b5b557f/chrono-0.4.41/src/naive/date/mod.rs:1911:38:
`NaiveDate + TimeDelta` overflowed
stack backtrace:
   0: __rustc::rust_begin_unwind
             at /rustc/17067e9ac6d7ecb70e50f92c1944e545188d2359/library/std/src/panicking.rs:697:5
   1: core::panicking::panic_fmt
             at /rustc/17067e9ac6d7ecb70e50f92c1944e545188d2359/library/core/src/panicking.rs:75:14
   2: core::panicking::panic_display
             at /rustc/17067e9ac6d7ecb70e50f92c1944e545188d2359/library/core/src/panicking.rs:261:5
   3: core::option::expect_failed
             at /rustc/17067e9ac6d7ecb70e50f92c1944e545188d2359/library/core/src/option.rs:2024:5
   4: core::option::Option<T>::expect
             at /.rustup/toolchains/1.87.0-aarch64-apple-darwin/lib/rustlib/src/rust/library/core/src/option.rs:933:21
   5: <chrono::naive::date::NaiveDate as core::ops::arith::Add<chrono::time_delta::TimeDelta>>::add
             at /.cargo/registry/src/index.crates.io-1949cf8c6b5b557f/chrono-0.4.41/src/naive/date/mod.rs:1911:9
   6: arrow_array::types::Date64Type::to_naive_date
             at /.cargo/registry/src/index.crates.io-1949cf8c6b5b557f/arrow-array-55.1.0/src/types.rs:1036:9
   7: <datafusion_common::scalar::ScalarValue as core::fmt::Display>::fmt::{{closure}}
             at /GitHub/datafusion/datafusion/common/src/scalar/mod.rs:3823:45
   8: core::option::Option<T>::map
             at /.rustup/toolchains/1.87.0-aarch64-apple-darwin/lib/rustlib/src/rust/library/core/src/option.rs:1119:29
   9: <datafusion_common::scalar::ScalarValue as core::fmt::Display>::fmt
             at /GitHub/datafusion/datafusion/common/src/scalar/mod.rs:3823:35
  10: core::fmt::rt::Argument::fmt
             at /rustc/17067e9ac6d7ecb70e50f92c1944e545188d2359/library/core/src/fmt/rt.rs:184:76
  11: core::fmt::write
             at /rustc/17067e9ac6d7ecb70e50f92c1944e545188d2359/library/core/src/fmt/mod.rs:1481:21
  12: <&mut W as core::fmt::Write::write_fmt::SpecWriteFmt>::spec_write_fmt
             at /rustc/17067e9ac6d7ecb70e50f92c1944e545188d2359/library/core/src/fmt/mod.rs:229:21
  13: core::fmt::Write::write_fmt
             at /rustc/17067e9ac6d7ecb70e50f92c1944e545188d2359/library/core/src/fmt/mod.rs:234:9
  14: alloc::fmt::format::format_inner
             at /rustc/17067e9ac6d7ecb70e50f92c1944e545188d2359/library/alloc/src/fmt.rs:649:14
  15: alloc::fmt::format::{{closure}}
             at /.rustup/toolchains/1.87.0-aarch64-apple-darwin/lib/rustlib/src/rust/library/alloc/src/fmt.rs:654:34
  16: core::option::Option<T>::map_or_else
             at /.rustup/toolchains/1.87.0-aarch64-apple-darwin/lib/rustlib/src/rust/library/core/src/option.rs:1225:21
  17: alloc::fmt::format
             at /.rustup/toolchains/1.87.0-aarch64-apple-darwin/lib/rustlib/src/rust/library/alloc/src/fmt.rs:654:5
  18: datafusion_physical_expr::expressions::literal::Literal::new_with_metadata
             at /GitHub/datafusion/datafusion/physical-expr/src/expressions/literal.rs:70:24
  19: datafusion_physical_expr::expressions::literal::Literal::new
             at /GitHub/datafusion/datafusion/physical-expr/src/expressions/literal.rs:61:9
  20: datafusion_physical_expr::expressions::literal::lit
             at /GitHub/datafusion/datafusion/physical-expr/src/expressions/literal.rs:143:41
  21: datafusion_physical_plan::topk::TopK::update_filter
             at /GitHub/datafusion/datafusion/physical-plan/src/topk/mod.rs:314:17

But if I comment out update_filters() there are no failures. I think the update_filters() function may be the culprit of all of the evils and it's manifesting via some sort of unstable sort / mishandling of nulls and this other random bug with Display for Date64.

Bizarre... will keep pushing to find a root cause and see if we can address it. If I can't get it sorted over the weekend @alamb I'd propose we revert the TopK specific part of that PR to unblock the rest of the project.

@adriangb

Sounds good -- thank you for the investigation @adriangb

So yeah that's a trivially reproducible bug that is completely unrelated to this mess but I guess just happens to be hit by the new code paths:

Literal::new(ScalarValue::Date64(Some(-790179464505600000)))

Panics because it calls:

https://github.com/apache/arrow-rs/blob/1ededfe024e6da1dd08bd0aee9411d1fb04523ac/arrow-array/src/types.rs#L1036

Which by as per the documentation will panic:

https://github.com/chronotope/chrono/blob/bab97905ccfa5aa7ceae1ee131b41b7113ea4f6a/src/naive/date/mod.rs#L1876-L1877

@blaginin

This is actually the same error that @blaginin reported in #16452 (comment): https://github.com/apache/datafusion/actions/runs/15780623790/job/44485106315#step:4:1382

#16491 now has the fix!

We think we have fixed this in

re-enable sort_query_fuzzer_runner #16491

Let's keep this PR open until

We have some more time where the tests are passing successfully on main
We revert the code changes in Temporarily fix bug in dynamic top-k optimization #16465 that were unrelated. I'll make a PR to do that (update restore topk pre-filtering of batches and make sort query fuzzer less sensitive to expected non determinism #16501)

We revert the code changes in Temporarily fix bug in dynamic top-k optimization #16465 that were unrelated. I'll make a PR to do that (update Restore topk filtering tests #16501)

I think that part may have its own issues!

Navigation Menu

Repository navigation

SortQueryFuzzer found a failing case on main #16452

Describe the bug

To Reproduce

Expected behavior

Additional context

Metadata

Assignees

Labels

Type

Projects

Milestone

Relationships

Development

	if let Some(filter) = self.filter.as_ref() {
	// If a filter is provided, update it with the new rows
	let filter = filter.current()?;
	let filtered = filter.evaluate(&batch)?;
	let num_rows = batch.num_rows();
	let array = filtered.into_array(num_rows)?;
	let mut filter = array.as_boolean().clone();
	let true_count = filter.true_count();
	if true_count == 0 {
	// nothing to filter, so no need to update
	return Ok(());
	}
	// only update the keys / rows if the filter does not match all rows
	if true_count < num_rows {
	// Indices in `set_indices` should be correct if filter contains nulls
	// So we prepare the filter here. Note this is also done in the `FilterBuilder`
	// so there is no overhead to do this here.
	if filter.nulls().is_some() {
	filter = prep_null_mask_filter(&filter);
	}

	let filter_predicate = FilterBuilder::new(&filter);
	let filter_predicate = if sort_keys.len() > 1 {
	// Optimize filter when it has multiple sort keys
	filter_predicate.optimize().build()
	} else {
	filter_predicate.build()
	};
	selected_rows = Some(filter);
	sort_keys = sort_keys
	.iter()
	.map(\|key\| filter_predicate.filter(key).map_err(\|x\| x.into()))
	.collect::<Result<Vec<_>>>()?;
	}
	};

SortQueryFuzzer found a failing case on main #16452

Description

Describe the bug

To Reproduce

Expected behavior

Additional context

Activity

2010YOUY01 commented on Jun 19, 2025

AdamGS commented on Jun 19, 2025

alamb commented on Jun 19, 2025

AdamGS commented on Jun 19, 2025

adriangb commented on Jun 19, 2025

AdamGS commented on Jun 19, 2025

adriangb commented on Jun 19, 2025

adriangb commented on Jun 19, 2025

AdamGS commented on Jun 19, 2025

adriangb commented on Jun 19, 2025

AdamGS commented on Jun 19, 2025

blaginin commented on Jun 20, 2025

alamb commented on Jun 20, 2025

adriangb commented on Jun 20, 2025

adriangb commented on Jun 21, 2025

alamb commented on Jun 21, 2025

adriangb commented on Jun 21, 2025

adriangb commented on Jun 21, 2025

adriangb commented on Jun 21, 2025

alamb commented on Jun 22, 2025

adriangb commented on Jun 22, 2025

Metadata

Metadata

Assignees

Labels

Type

Projects

Milestone

Relationships

Development

Issue actions