You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@datafusion.apache.org by gi...@apache.org on 2026/10/19 20:10:09 UTC
(datafusion) branch gh-readonly-queue/main/pr-26386-481a1f84395a4ffea637f2c8144e237e00bbe24a created (now 742cfae0d7)
This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a change to branch gh-readonly-queue/main/pr-26386-481a1f84395a4ffea637f2c8144e237e00bbe24a
in repository https://gitbox.apache.org/repos/asf/datafusion.git
at 742cfae0d7 test: add NULL values to Fuzz window (#26386)
This branch includes the following new commits:
new 742cfae0d7 test: add NULL values to Fuzz window (#26386)
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@datafusion.apache.org
For additional commands, e-mail: commits-help@datafusion.apache.org
(datafusion) 01/01: test: add NULL values to Fuzz window (#26386)
Posted by gi...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a commit to branch gh-readonly-queue/main/pr-26386-481a1f84395a4ffea637f2c8144e237e00bbe24a
in repository https://gitbox.apache.org/repos/asf/datafusion.git
commit 742cfae0d7f0db9a4b870dddc4b8076527010b46
Author: Nagato Yuzuru <na...@yuzuru.me>
AuthorDate: Mon Oct 19 20:09:46 2026 +0000
test: add NULL values to Fuzz window (#26386)
<!--
We generally require a GitHub issue to be filed for all bug fixes and
enhancements and this helps us generate change logs for our releases.
You can link an issue to this PR using the GitHub syntax. For example
`Closes #123` indicates that this PR will close issue #123.
-->
Part of #24243.
This PR adds NULL values to the window fuzz test, as planned.
## Rationale for this change
<!--
Why are you proposing this change? If this is already explained clearly
in the issue then this section is not needed.
Explaining clearly why changes are proposed helps reviewers understand
your changes and offer better suggestions for fixes.
Please explain the problem you are trying to solve in terms of the
user-visible
behavior, rather than the implementation.
For example, "The code in `foo.rs` doesn't handle nulls" is a symptom of
the
implementation. "COUNT(DISTINCT) returns wrong results when the column
contains
nulls" is the user-visible problem.
-->
The window fuzz test had two gaps:
- The test data had no NULL values.
- The test used only one sort order: `ASC NULLS FIRST`.
With these gaps closed, the test finds wrong results on `main`. All of
them
come from one known error: #25859.
## What changes are included in this PR?
<!--
There is no need to duplicate the description in the issue here, but it
is sometimes worth providing a summary of the individual changes in this
PR.
-->
- Add NULL values to the test data. The NULL rate changes between 0,
0.01, 0.1
and 0.5.
- Give each sort column a random sort direction and a random NULL
position.
Note: a seed now gives different data than before this PR.
## What is the testing strategy for this PR?
<!--
We typically require tests for all PRs in order to:
1. Prevent the code from being accidentally broken by subsequent changes
2. Serve as another way to document the expected behavior of the code
Briefly describe how this PR is tested, and point to the specific tests
you added. For example: 'This new feature is covered by the
`sqllogictest` cases added in `foo.slt`'.
If this PR does not add tests, explain why. For example, if the change
is already covered by existing tests, please mention it.
You should also check the `codecov` bot reply on this PR to confirm the
changed code is exercised.
-->
```bash
cargo test --profile ci -p datafusion --features extended_tests --test fuzz -- window_fuzz
```
On `main`, the test fails. For example:
```
WindowFrame { units: Range, start_bound: Preceding(Int32(6)), end_bound: Preceding(Int32(5)), .. },
window_fn: max(x), random_seed: 267, null_pct: 0.5, pb_cols: ["a"], ..
```
All failures have a `RANGE` frame that ends in `N PRECEDING`, and NULL
values in
the data. They occur with all four sort orders (ASC/DESC, NULLS
FIRST/LAST).
## Additional notes
#25860 fixes #25859. With #25860, this test passes. This PR must merge
after
#25860. Or, if the author of #25860 agrees, these commits can go into
#25860.
## Are there any user-facing changes?
<!--
If there are user-facing changes then we may require documentation to be
updated before approving the PR.
If there are any breaking changes to public APIs, please add the `api
change` label.
-->
No
---
datafusion/core/tests/fuzz_cases/window_fuzz.rs | 129 +++++++++++++++---------
1 file changed, 83 insertions(+), 46 deletions(-)
diff --git a/datafusion/core/tests/fuzz_cases/window_fuzz.rs b/datafusion/core/tests/fuzz_cases/window_fuzz.rs
index f69b5e9a41..56576241a5 100644
--- a/datafusion/core/tests/fuzz_cases/window_fuzz.rs
+++ b/datafusion/core/tests/fuzz_cases/window_fuzz.rs
@@ -18,7 +18,7 @@
use std::sync::Arc;
use arrow::array::{ArrayRef, Int32Array, StringArray};
-use arrow::compute::{SortOptions, concat_batches};
+use arrow::compute::{SortColumn, SortOptions, concat_batches, lexsort_to_indices, take};
use arrow::datatypes::SchemaRef;
use arrow::record_batch::RecordBatch;
use arrow::util::pretty::pretty_format_batches;
@@ -133,10 +133,22 @@ async fn window_bounded_window_random_comparison() -> Result<()> {
let mut handles = Vec::new();
for i in 0..n {
let idx = i % test_cases.len();
+ // Change the NULL rate after each pass through `test_cases`, so that
+ // each test case runs with each NULL rate.
+ let null_pct = NULL_PCTS[(i / test_cases.len()) % NULL_PCTS.len()];
+ let sort_options = random_sort_options(i as u64);
let (pb_cols, ob_cols, search_mode) = test_cases[idx].clone();
let job = SpawnedTask::spawn(run_window_test(
- make_staggered_batches::<true>(1000, n_distinct, i as u64),
+ make_staggered_batches::<true>(
+ 1000,
+ n_distinct,
+ null_pct,
+ sort_options,
+ i as u64,
+ ),
i as u64,
+ null_pct,
+ sort_options,
pb_cols,
ob_cols,
search_mode,
@@ -156,7 +168,15 @@ async fn window_bounded_window_random_comparison() -> Result<()> {
async fn bounded_window_causal_non_causal() -> Result<()> {
let session_config = SessionConfig::new();
let ctx = SessionContext::new_with_config(session_config);
- let mut batches = make_staggered_batches::<true>(1000, 10, 23_u64);
+ // This test examines only the output batch sizes and has no ORDER BY.
+ // NULL values do not change the result of this test.
+ let mut batches = make_staggered_batches::<true>(
+ 1000,
+ 10,
+ 0.0,
+ [SortOptions::default(); 3],
+ 23_u64,
+ );
// Remove empty batches:
batches.retain(|batch| batch.num_rows() > 0);
let schema = batches[0].schema();
@@ -585,6 +605,8 @@ fn convert_bound_to_current_row_if_applicable(
async fn run_window_test(
input1: Vec<RecordBatch>,
random_seed: u64,
+ null_pct: f64,
+ sort_options: [SortOptions; 3],
partition_by_columns: Vec<&str>,
orderby_columns: Vec<&str>,
search_mode: InputOrderMode,
@@ -596,12 +618,18 @@ async fn run_window_test(
let ctx = SessionContext::new_with_config(session_config);
let (window_fn, args, fn_name) = get_random_function(&schema, &mut rng, is_linear);
let window_frame = get_random_window_frame(&mut rng, is_linear);
- let mut orderby_exprs = vec![];
- for column in &orderby_columns {
- orderby_exprs.push(PhysicalSortExpr {
+ // Use the sort options of the input for each column. Then the input
+ // satisfies the ordering of the `Sorted` and `PartiallySorted` modes.
+ let sort_expr = |column: &str| -> Result<PhysicalSortExpr> {
+ let idx = ["a", "b", "c"].iter().position(|c| *c == column).unwrap();
+ Ok(PhysicalSortExpr {
expr: col(column, &schema)?,
- options: SortOptions::default(),
+ options: sort_options[idx],
})
+ };
+ let mut orderby_exprs = vec![];
+ for column in &orderby_columns {
+ orderby_exprs.push(sort_expr(column)?)
}
if orderby_exprs.len() > 1 && !window_frame.can_accept_multi_orderby() {
orderby_exprs.truncate(1);
@@ -611,11 +639,8 @@ async fn run_window_test(
partitionby_exprs.push(col(column, &schema)?);
}
let mut sort_keys = vec![];
- for partition_by_expr in &partitionby_exprs {
- sort_keys.push(PhysicalSortExpr {
- expr: partition_by_expr.clone(),
- options: SortOptions::default(),
- })
+ for column in &partition_by_columns {
+ sort_keys.push(sort_expr(column)?)
}
for order_by_expr in &orderby_exprs {
if !sort_keys.contains(order_by_expr) {
@@ -624,21 +649,8 @@ async fn run_window_test(
}
let concat_input_record = concat_batches(&schema, &input1)?;
- let source_sort_keys: LexOrdering = [
- PhysicalSortExpr {
- expr: col("a", &schema)?,
- options: Default::default(),
- },
- PhysicalSortExpr {
- expr: col("b", &schema)?,
- options: Default::default(),
- },
- PhysicalSortExpr {
- expr: col("c", &schema)?,
- options: Default::default(),
- },
- ]
- .into();
+ let source_sort_keys: LexOrdering =
+ [sort_expr("a")?, sort_expr("b")?, sort_expr("c")?].into();
let mut exec1 = DataSourceExec::from_data_source(
MemorySourceConfig::try_new(&[vec![concat_input_record]], schema.clone(), None)?
.try_with_sort_information(vec![source_sort_keys.clone()])?,
@@ -699,7 +711,7 @@ async fn run_window_test(
// BoundedWindowAggExec should produce more chunk than the usual WindowAggExec.
// Otherwise it means that we cannot generate result in running mode.
let err_msg = format!(
- "Inconsistent result for window_frame: {window_frame:?}, window_fn: {window_fn:?}, args:{args:?}, random_seed: {random_seed:?}, search_mode: {search_mode:?}, partition_by_columns:{partition_by_columns:?}, orderby_columns: {orderby_columns:?}"
+ "Inconsistent result for window_frame: {window_frame:?}, window_fn: {window_fn:?}, args:{args:?}, random_seed: {random_seed:?}, null_pct: {null_pct:?}, sort_options: {sort_options:?}, search_mode: {search_mode:?}, partition_by_columns:{partition_by_columns:?}, orderby_columns: {orderby_columns:?}"
);
// Below check makes sure that, streaming execution generates more chunks than the bulk execution.
// Since algorithms and operators works on sliding windows in the streaming execution.
@@ -733,7 +745,7 @@ async fn run_window_test(
{
if !usual_line.eq(running_line) {
println!(
- "Inconsistent result for window_frame at line:{i:?}: {window_frame:?}, window_fn: {window_fn:?}, args:{args:?}, pb_cols:{partition_by_columns:?}, ob_cols:{orderby_columns:?}, search_mode:{search_mode:?}"
+ "Inconsistent result for window_frame at line:{i:?}: {window_frame:?}, window_fn: {window_fn:?}, args:{args:?}, random_seed: {random_seed:?}, null_pct: {null_pct:?}, sort_options: {sort_options:?}, pb_cols:{partition_by_columns:?}, ob_cols:{orderby_columns:?}, search_mode:{search_mode:?}"
);
println!(
"--------usual_formatted_sorted----------------running_formatted_sorted--------"
@@ -749,6 +761,15 @@ async fn run_window_test(
Ok(())
}
+/// NULL rates for the random window tests
+const NULL_PCTS: [f64; 4] = [0.0, 0.01, 0.1, 0.5];
+
+/// Return a random sort direction and NULL position for 'a', 'b' and 'c'
+fn random_sort_options(seed: u64) -> [SortOptions; 3] {
+ let mut rng = StdRng::seed_from_u64(seed);
+ std::array::from_fn(|_| SortOptions::new(rng.random_bool(0.5), rng.random_bool(0.5)))
+}
+
fn generate_random_string(rng: &mut StdRng, length: usize) -> String {
rng.sample_iter(&Alphanumeric)
.take(length)
@@ -759,40 +780,56 @@ fn generate_random_string(rng: &mut StdRng, length: usize) -> String {
/// Return randomly sized record batches with:
/// three sorted int32 columns 'a', 'b', 'c' ranged from 0..DISTINCT as columns
/// one random int32 column x
+///
+/// `sort_options` sets the sort order of 'a', 'b' and 'c', in this sequence.
+/// Each value in 'a', 'b', 'c' and 'x' is NULL with the probability `null_pct`.
pub(crate) fn make_staggered_batches<const STREAM: bool>(
len: usize,
n_distinct: usize,
+ null_pct: f64,
+ sort_options: [SortOptions; 3],
random_seed: u64,
) -> Vec<RecordBatch> {
// use a random number generator to pick a random sized output
let mut rng = StdRng::seed_from_u64(random_seed);
- let mut input123: Vec<(i32, i32, i32)> = vec![(0, 0, 0); len];
- let mut input4: Vec<i32> = vec![0; len];
+ let mut input4: Vec<Option<i32>> = vec![None; len];
let mut input5: Vec<String> = vec!["".to_string(); len];
- for v in &mut input123 {
- *v = (
- rng.random_range(0..n_distinct) as i32,
- rng.random_range(0..n_distinct) as i32,
- rng.random_range(0..n_distinct) as i32,
- )
+ let random_key = |rng: &mut StdRng| {
+ let value = rng.random_range(0..n_distinct) as i32;
+ (!rng.random_bool(null_pct)).then_some(value)
+ };
+ let keys: [ArrayRef; 3] = std::array::from_fn(|_| {
+ Arc::new(Int32Array::from_iter(
+ (0..len).map(|_| random_key(&mut rng)),
+ )) as _
+ });
+ for v in &mut input4 {
+ let value = rng.random();
+ *v = (!rng.random_bool(null_pct)).then_some(value);
}
- input123.sort_unstable();
- rng.fill(&mut input4[..]);
for v in &mut input5 {
*v = generate_random_string(&mut rng, 1);
}
input5.sort();
- let input1 = Int32Array::from_iter_values(input123.iter().map(|k| k.0));
- let input2 = Int32Array::from_iter_values(input123.iter().map(|k| k.1));
- let input3 = Int32Array::from_iter_values(input123.iter().map(|k| k.2));
- let input4 = Int32Array::from_iter_values(input4);
+ let sort_columns: Vec<SortColumn> = keys
+ .iter()
+ .zip(sort_options)
+ .map(|(values, options)| SortColumn {
+ values: Arc::clone(values),
+ options: Some(options),
+ })
+ .collect();
+ let indices = lexsort_to_indices(&sort_columns, None).unwrap();
+ let [input1, input2, input3] =
+ keys.map(|values| take(values.as_ref(), &indices, None).unwrap());
+ let input4 = Int32Array::from_iter(input4);
let input5 = StringArray::from_iter_values(input5);
// split into several record batches
let mut remainder = RecordBatch::try_from_iter(vec![
- ("a", Arc::new(input1) as ArrayRef),
- ("b", Arc::new(input2) as ArrayRef),
- ("c", Arc::new(input3) as ArrayRef),
+ ("a", input1),
+ ("b", input2),
+ ("c", input3),
("x", Arc::new(input4) as ArrayRef),
("string_field", Arc::new(input5) as ArrayRef),
])
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@datafusion.apache.org
For additional commands, e-mail: commits-help@datafusion.apache.org