SortQueryFuzzer found a failing case on main

[AdamGS]

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

[2010YOUY01]

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.

[AdamGS]

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

[alamb]

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)

[AdamGS]

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();
    }
}

[adriangb]

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

[AdamGS]

Some more findings:

1. datafusion.optimizer.enable_dynamic_filter_pushdown doesn't seem to make a difference
2. 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:

1. 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).
2. 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.

[adriangb]

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

[adriangb]

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<_>>>()?;
	}
	};

[AdamGS]

that indeed makes the failure to go away