From ff4c630345d032e0d1112649bf76f7a8499f660e Mon Sep 17 00:00:00 2001 From: Andrew Lamb Date: Thu, 12 Mar 2026 06:56:50 -0400 Subject: [PATCH 01/12] [branch-52] fix: maintain inner list nullability for (#19948) (#20878) - Part of https://github.com/apache/datafusion/issues/20855 - Closes https://github.com/apache/datafusion/issues/19947 on branch-52 This PR: - Backports https://github.com/apache/datafusion/pull/19948 from @Jefffrey to the branch-52 line Co-authored-by: Jeffrey Vo --- datafusion/functions-nested/src/sort.rs | 36 +++++--------------- datafusion/sqllogictest/test_files/array.slt | 25 ++++++++++++++ 2 files changed, 34 insertions(+), 27 deletions(-) diff --git a/datafusion/functions-nested/src/sort.rs b/datafusion/functions-nested/src/sort.rs index ba2da0f760eee..cbe101f111b26 100644 --- a/datafusion/functions-nested/src/sort.rs +++ b/datafusion/functions-nested/src/sort.rs @@ -18,16 +18,14 @@ //! [`ScalarUDFImpl`] definitions for array_sort function. use crate::utils::make_scalar_function; -use arrow::array::{ - Array, ArrayRef, GenericListArray, NullBufferBuilder, OffsetSizeTrait, new_null_array, -}; +use arrow::array::{Array, ArrayRef, GenericListArray, OffsetSizeTrait, new_null_array}; use arrow::buffer::OffsetBuffer; use arrow::compute::SortColumn; use arrow::datatypes::{DataType, FieldRef}; use arrow::{compute, compute::SortOptions}; use datafusion_common::cast::{as_large_list_array, as_list_array, as_string_array}; use datafusion_common::utils::ListCoercion; -use datafusion_common::{Result, exec_err, plan_err}; +use datafusion_common::{Result, exec_err}; use datafusion_expr::{ ArrayFunctionArgument, ArrayFunctionSignature, ColumnarValue, Documentation, ScalarUDFImpl, Signature, TypeSignature, Volatility, @@ -134,18 +132,7 @@ impl ScalarUDFImpl for ArraySort { } fn return_type(&self, arg_types: &[DataType]) -> Result { - match &arg_types[0] { - DataType::Null => Ok(DataType::Null), - DataType::List(field) => { - Ok(DataType::new_list(field.data_type().clone(), true)) - } - DataType::LargeList(field) => { - Ok(DataType::new_large_list(field.data_type().clone(), true)) - } - arg_type => { - plan_err!("{} does not support type {arg_type}", self.name()) - } - } + Ok(arg_types[0].clone()) } fn invoke_with_args( @@ -206,11 +193,11 @@ fn array_sort_inner(args: &[ArrayRef]) -> Result { } DataType::List(field) => { let array = as_list_array(&args[0])?; - array_sort_generic(array, field, sort_options) + array_sort_generic(array, Arc::clone(field), sort_options) } DataType::LargeList(field) => { let array = as_large_list_array(&args[0])?; - array_sort_generic(array, field, sort_options) + array_sort_generic(array, Arc::clone(field), sort_options) } // Signature should prevent this arm ever occurring _ => exec_err!("array_sort expects list for first argument"), @@ -219,18 +206,16 @@ fn array_sort_inner(args: &[ArrayRef]) -> Result { fn array_sort_generic( list_array: &GenericListArray, - field: &FieldRef, + field: FieldRef, sort_options: Option, ) -> Result { let row_count = list_array.len(); let mut array_lengths = vec![]; let mut arrays = vec![]; - let mut valid = NullBufferBuilder::new(row_count); for i in 0..row_count { if list_array.is_null(i) { array_lengths.push(0); - valid.append_null(); } else { let arr_ref = list_array.value(i); @@ -253,25 +238,22 @@ fn array_sort_generic( }; array_lengths.push(sorted_array.len()); arrays.push(sorted_array); - valid.append_non_null(); } } - let buffer = valid.finish(); - let elements = arrays .iter() .map(|a| a.as_ref()) .collect::>(); let list_arr = if elements.is_empty() { - GenericListArray::::new_null(Arc::clone(field), row_count) + GenericListArray::::new_null(field, row_count) } else { GenericListArray::::new( - Arc::clone(field), + field, OffsetBuffer::from_lengths(array_lengths), Arc::new(compute::concat(elements.as_slice())?), - buffer, + list_array.nulls().cloned(), ) }; Ok(Arc::new(list_arr)) diff --git a/datafusion/sqllogictest/test_files/array.slt b/datafusion/sqllogictest/test_files/array.slt index c31f3d0702358..65df1e50b8981 100644 --- a/datafusion/sqllogictest/test_files/array.slt +++ b/datafusion/sqllogictest/test_files/array.slt @@ -2577,6 +2577,31 @@ NULL NULL NULL NULL NULL NULL +# maintains inner nullability +query ?T +select array_sort(column1), arrow_typeof(array_sort(column1)) +from values + (arrow_cast([], 'List(non-null Int32)')), + (arrow_cast(NULL, 'List(non-null Int32)')), + (arrow_cast([1, 3, 5, -5], 'List(non-null Int32)')) +; +---- +[] List(non-null Int32) +NULL List(non-null Int32) +[-5, 1, 3, 5] List(non-null Int32) + +query ?T +select column1, arrow_typeof(column1) +from values (array_sort(arrow_cast([1, 3, 5, -5], 'LargeList(non-null Int32)'))); +---- +[-5, 1, 3, 5] LargeList(non-null Int32) + +query ?T +select column1, arrow_typeof(column1) +from values (array_sort(arrow_cast([1, 3, 5, -5], 'FixedSizeList(4 x non-null Int32)'))); +---- +[-5, 1, 3, 5] List(non-null Int32) + query ? select array_sort([struct('foo', 3), struct('foo', 1), struct('bar', 1)]) ---- From 822f3fa5ed551d746f31b7fec65fde84c207db11 Mon Sep 17 00:00:00 2001 From: Andrew Lamb Date: Thu, 12 Mar 2026 06:57:02 -0400 Subject: [PATCH 02/12] [branch-52] fix: Ensure columns are casted to the correct names with Unions (#20146) (#20879) - Part of https://github.com/apache/datafusion/issues/20855 - Closes https://github.com/apache/datafusion/issues/20123 on branch-52 This PR: - Backports https://github.com/apache/datafusion/pull/20146 from @nuno-faria to the branch-52 line --------- Co-authored-by: Nuno Faria --- datafusion/expr/src/expr_rewriter/mod.rs | 13 +- .../optimizer/tests/optimizer_integration.rs | 6 +- .../substrait/tests/cases/logical_plans.rs | 24 +++ .../duplicate_name_in_union.substrait.json | 171 ++++++++++++++++++ 4 files changed, 208 insertions(+), 6 deletions(-) create mode 100644 datafusion/substrait/tests/testdata/test_plans/duplicate_name_in_union.substrait.json diff --git a/datafusion/expr/src/expr_rewriter/mod.rs b/datafusion/expr/src/expr_rewriter/mod.rs index a0faca76e91e4..32a88ab8cf310 100644 --- a/datafusion/expr/src/expr_rewriter/mod.rs +++ b/datafusion/expr/src/expr_rewriter/mod.rs @@ -261,9 +261,16 @@ fn coerce_exprs_for_schema( #[expect(deprecated)] Expr::Wildcard { .. } => Ok(expr), _ => { - // maintain the original name when casting - let name = dst_schema.field(idx).name(); - Ok(expr.cast_to(new_type, src_schema)?.alias(name)) + match expr { + // maintain the original name when casting a column, to avoid the + // tablename being added to it when not explicitly set by the query + // (see: https://github.com/apache/datafusion/issues/18818) + Expr::Column(ref column) => { + let name = column.name().to_owned(); + Ok(expr.cast_to(new_type, src_schema)?.alias(name)) + } + _ => Ok(expr.cast_to(new_type, src_schema)?), + } } } } else { diff --git a/datafusion/optimizer/tests/optimizer_integration.rs b/datafusion/optimizer/tests/optimizer_integration.rs index 36a6df54ddaf0..fd4991c24413f 100644 --- a/datafusion/optimizer/tests/optimizer_integration.rs +++ b/datafusion/optimizer/tests/optimizer_integration.rs @@ -543,7 +543,7 @@ fn recursive_cte_projection_pushdown() -> Result<()> { RecursiveQuery: is_distinct=false Projection: test.col_int32 AS id TableScan: test projection=[col_int32] - Projection: CAST(CAST(nodes.id AS Int64) + Int64(1) AS Int32) AS id + Projection: CAST(CAST(nodes.id AS Int64) + Int64(1) AS Int32) Filter: nodes.id < Int32(3) TableScan: nodes projection=[id] " @@ -567,7 +567,7 @@ fn recursive_cte_with_aliased_self_reference() -> Result<()> { RecursiveQuery: is_distinct=false Projection: test.col_int32 AS id TableScan: test projection=[col_int32] - Projection: CAST(CAST(child.id AS Int64) + Int64(1) AS Int32) AS id + Projection: CAST(CAST(child.id AS Int64) + Int64(1) AS Int32) SubqueryAlias: child Filter: nodes.id < Int32(3) TableScan: nodes projection=[id] @@ -630,7 +630,7 @@ fn recursive_cte_projection_pushdown_baseline() -> Result<()> { Projection: test.col_int32 AS n Filter: test.col_int32 = Int32(5) TableScan: test projection=[col_int32] - Projection: CAST(CAST(countdown.n AS Int64) - Int64(1) AS Int32) AS n + Projection: CAST(CAST(countdown.n AS Int64) - Int64(1) AS Int32) Filter: countdown.n > Int32(1) TableScan: countdown projection=[n] " diff --git a/datafusion/substrait/tests/cases/logical_plans.rs b/datafusion/substrait/tests/cases/logical_plans.rs index 41f08c579f471..e3e45193e7e0c 100644 --- a/datafusion/substrait/tests/cases/logical_plans.rs +++ b/datafusion/substrait/tests/cases/logical_plans.rs @@ -220,6 +220,30 @@ mod tests { // Trigger execution to ensure plan validity DataFrame::new(ctx.state(), plan).show().await?; + Ok(()) + } + #[tokio::test] + async fn duplicate_name_in_union() -> Result<()> { + let proto_plan = + read_json("tests/testdata/test_plans/duplicate_name_in_union.substrait.json"); + let ctx = add_plan_schemas_to_ctx(SessionContext::new(), &proto_plan)?; + let plan = from_substrait_plan(&ctx.state(), &proto_plan).await?; + + assert_snapshot!( + plan, + @r" + Projection: foo AS col1, bar AS col2 + Union + Projection: foo, bar + Values: (Int64(100), Int64(200)) + Projection: x, foo + Values: (Int32(300), Int64(400)) + " + ); + + // Trigger execution to ensure plan validity + DataFrame::new(ctx.state(), plan).show().await?; + Ok(()) } } diff --git a/datafusion/substrait/tests/testdata/test_plans/duplicate_name_in_union.substrait.json b/datafusion/substrait/tests/testdata/test_plans/duplicate_name_in_union.substrait.json new file mode 100644 index 0000000000000..1da2ff6131368 --- /dev/null +++ b/datafusion/substrait/tests/testdata/test_plans/duplicate_name_in_union.substrait.json @@ -0,0 +1,171 @@ +{ + "version": { + "minorNumber": 54, + "producer": "datafusion-test" + }, + "relations": [ + { + "root": { + "input": { + "set": { + "common": { + "direct": {} + }, + "inputs": [ + { + "project": { + "common": { + "emit": { + "outputMapping": [2, 3] + } + }, + "input": { + "read": { + "common": { + "direct": {} + }, + "baseSchema": { + "names": ["foo", "bar"], + "struct": { + "types": [ + { + "i64": { + "nullability": "NULLABILITY_REQUIRED" + } + }, + { + "i64": { + "nullability": "NULLABILITY_REQUIRED" + } + } + ], + "nullability": "NULLABILITY_REQUIRED" + } + }, + "virtualTable": { + "expressions": [ + { + "fields": [ + { + "literal": { + "i64": "100" + } + }, + { + "literal": { + "i64": "200" + } + } + ] + } + ] + } + } + }, + "expressions": [ + { + "selection": { + "directReference": { + "structField": { + "field": 0 + } + }, + "rootReference": {} + } + }, + { + "selection": { + "directReference": { + "structField": { + "field": 1 + } + }, + "rootReference": {} + } + } + ] + } + }, + { + "project": { + "common": { + "emit": { + "outputMapping": [2, 3] + } + }, + "input": { + "read": { + "common": { + "direct": {} + }, + "baseSchema": { + "names": ["x", "foo"], + "struct": { + "types": [ + { + "i32": { + "nullability": "NULLABILITY_REQUIRED" + } + }, + { + "i64": { + "nullability": "NULLABILITY_REQUIRED" + } + } + ], + "nullability": "NULLABILITY_REQUIRED" + } + }, + "virtualTable": { + "expressions": [ + { + "fields": [ + { + "literal": { + "i32": 300 + } + }, + { + "literal": { + "i64": "400" + } + } + ] + } + ] + } + } + }, + "expressions": [ + { + "selection": { + "directReference": { + "structField": { + "field": 0 + } + }, + "rootReference": {} + } + }, + { + "selection": { + "directReference": { + "structField": { + "field": 1 + } + }, + "rootReference": {} + } + } + ] + } + } + ], + "op": "SET_OP_UNION_ALL" + } + }, + "names": ["col1", "col2"] + } + } + ] +} From a5f6fbb4cd89a47e1036986abe201def15542093 Mon Sep 17 00:00:00 2001 From: Andrew Lamb Date: Thu, 12 Mar 2026 06:57:13 -0400 Subject: [PATCH 03/12] [branch-52] fix: interval analysis error when have two filterexec that inner filter proves zero selectivity (#20743) (#20880) - Part of https://github.com/apache/datafusion/issues/20855 - Closes https://github.com/apache/datafusion/issues/20742 on branch-52 This PR: - Backports https://github.com/apache/datafusion/pull/20743 from @haohuaijin to the branch-52 line Co-authored-by: Huaijin --- .../partition_statistics.rs | 24 ++--- datafusion/physical-plan/src/filter.rs | 91 +++++++++++++++++-- 2 files changed, 93 insertions(+), 22 deletions(-) diff --git a/datafusion/core/tests/physical_optimizer/partition_statistics.rs b/datafusion/core/tests/physical_optimizer/partition_statistics.rs index 36fa95aa9f57d..00827edc2413a 100644 --- a/datafusion/core/tests/physical_optimizer/partition_statistics.rs +++ b/datafusion/core/tests/physical_optimizer/partition_statistics.rs @@ -387,17 +387,17 @@ mod test { column_statistics: vec![ ColumnStatistics { null_count: Precision::Exact(0), - max_value: Precision::Exact(ScalarValue::Null), - min_value: Precision::Exact(ScalarValue::Null), - sum_value: Precision::Exact(ScalarValue::Null), + max_value: Precision::Exact(ScalarValue::Int32(None)), + min_value: Precision::Exact(ScalarValue::Int32(None)), + sum_value: Precision::Exact(ScalarValue::Int32(None)), distinct_count: Precision::Exact(0), byte_size: Precision::Exact(16), }, ColumnStatistics { null_count: Precision::Exact(0), - max_value: Precision::Exact(ScalarValue::Null), - min_value: Precision::Exact(ScalarValue::Null), - sum_value: Precision::Exact(ScalarValue::Null), + max_value: Precision::Exact(ScalarValue::Date32(None)), + min_value: Precision::Exact(ScalarValue::Date32(None)), + sum_value: Precision::Exact(ScalarValue::Date32(None)), distinct_count: Precision::Exact(0), byte_size: Precision::Exact(16), // 4 rows * 4 bytes (Date32) }, @@ -416,17 +416,17 @@ mod test { column_statistics: vec![ ColumnStatistics { null_count: Precision::Exact(0), - max_value: Precision::Exact(ScalarValue::Null), - min_value: Precision::Exact(ScalarValue::Null), - sum_value: Precision::Exact(ScalarValue::Null), + max_value: Precision::Exact(ScalarValue::Int32(None)), + min_value: Precision::Exact(ScalarValue::Int32(None)), + sum_value: Precision::Exact(ScalarValue::Int32(None)), distinct_count: Precision::Exact(0), byte_size: Precision::Exact(8), }, ColumnStatistics { null_count: Precision::Exact(0), - max_value: Precision::Exact(ScalarValue::Null), - min_value: Precision::Exact(ScalarValue::Null), - sum_value: Precision::Exact(ScalarValue::Null), + max_value: Precision::Exact(ScalarValue::Date32(None)), + min_value: Precision::Exact(ScalarValue::Date32(None)), + sum_value: Precision::Exact(ScalarValue::Date32(None)), distinct_count: Precision::Exact(0), byte_size: Precision::Exact(8), // 2 rows * 4 bytes (Date32) }, diff --git a/datafusion/physical-plan/src/filter.rs b/datafusion/physical-plan/src/filter.rs index f6ed0c91eca8f..b96b99933cff5 100644 --- a/datafusion/physical-plan/src/filter.rs +++ b/datafusion/physical-plan/src/filter.rs @@ -232,6 +232,7 @@ impl FilterExec { let total_byte_size = total_byte_size.with_estimated_selectivity(selectivity); let column_statistics = collect_new_statistics( + schema, &input_stats.column_statistics, analysis_ctx.boundaries, ); @@ -637,6 +638,7 @@ impl EmbeddedProjection for FilterExec { /// is adjusted by using the next/previous value for its data type to convert /// it into a closed bound. fn collect_new_statistics( + schema: &SchemaRef, input_column_stats: &[ColumnStatistics], analysis_boundaries: Vec, ) -> Vec { @@ -653,12 +655,17 @@ fn collect_new_statistics( }, )| { let Some(interval) = interval else { - // If the interval is `None`, we can say that there are no rows: + // If the interval is `None`, we can say that there are no rows. + // Use a typed null to preserve the column's data type, so that + // downstream interval analysis can still intersect intervals + // of the same type. + let typed_null = ScalarValue::try_from(schema.field(idx).data_type()) + .unwrap_or(ScalarValue::Null); return ColumnStatistics { null_count: Precision::Exact(0), - max_value: Precision::Exact(ScalarValue::Null), - min_value: Precision::Exact(ScalarValue::Null), - sum_value: Precision::Exact(ScalarValue::Null), + max_value: Precision::Exact(typed_null.clone()), + min_value: Precision::Exact(typed_null.clone()), + sum_value: Precision::Exact(typed_null), distinct_count: Precision::Exact(0), byte_size: input_column_stats[idx].byte_size, }; @@ -1351,17 +1358,17 @@ mod tests { statistics.column_statistics, vec![ ColumnStatistics { - min_value: Precision::Exact(ScalarValue::Null), - max_value: Precision::Exact(ScalarValue::Null), - sum_value: Precision::Exact(ScalarValue::Null), + min_value: Precision::Exact(ScalarValue::Int32(None)), + max_value: Precision::Exact(ScalarValue::Int32(None)), + sum_value: Precision::Exact(ScalarValue::Int32(None)), distinct_count: Precision::Exact(0), null_count: Precision::Exact(0), byte_size: Precision::Absent, }, ColumnStatistics { - min_value: Precision::Exact(ScalarValue::Null), - max_value: Precision::Exact(ScalarValue::Null), - sum_value: Precision::Exact(ScalarValue::Null), + min_value: Precision::Exact(ScalarValue::Int32(None)), + max_value: Precision::Exact(ScalarValue::Int32(None)), + sum_value: Precision::Exact(ScalarValue::Int32(None)), distinct_count: Precision::Exact(0), null_count: Precision::Exact(0), byte_size: Precision::Absent, @@ -1372,6 +1379,70 @@ mod tests { Ok(()) } + /// Regression test: stacking two FilterExecs where the inner filter + /// proves zero selectivity should not panic with a type mismatch + /// during interval intersection. + /// + /// Previously, when a filter proved no rows could match, the column + /// statistics used untyped `ScalarValue::Null` (data type `Null`). + /// If an outer FilterExec then tried to analyze its own predicate + /// against those statistics, `Interval::intersect` would fail with: + /// "Only intervals with the same data type are intersectable, lhs:Null, rhs:Int32" + #[tokio::test] + async fn test_nested_filter_with_zero_selectivity_inner() -> Result<()> { + // Inner table: a: [1, 100], b: [1, 3] + let schema = Schema::new(vec![ + Field::new("a", DataType::Int32, false), + Field::new("b", DataType::Int32, false), + ]); + let input = Arc::new(StatisticsExec::new( + Statistics { + num_rows: Precision::Inexact(1000), + total_byte_size: Precision::Inexact(4000), + column_statistics: vec![ + ColumnStatistics { + min_value: Precision::Inexact(ScalarValue::Int32(Some(1))), + max_value: Precision::Inexact(ScalarValue::Int32(Some(100))), + ..Default::default() + }, + ColumnStatistics { + min_value: Precision::Inexact(ScalarValue::Int32(Some(1))), + max_value: Precision::Inexact(ScalarValue::Int32(Some(3))), + ..Default::default() + }, + ], + }, + schema, + )); + + // Inner filter: a > 200 (impossible given a max=100 → zero selectivity) + let inner_predicate: Arc = Arc::new(BinaryExpr::new( + Arc::new(Column::new("a", 0)), + Operator::Gt, + Arc::new(Literal::new(ScalarValue::Int32(Some(200)))), + )); + let inner_filter: Arc = + Arc::new(FilterExec::try_new(inner_predicate, input)?); + + // Outer filter: a = 50 + // Before the fix, this would panic because the inner filter's + // zero-selectivity statistics produced Null-typed intervals for + // column `a`, which couldn't intersect with the Int32 literal. + let outer_predicate: Arc = Arc::new(BinaryExpr::new( + Arc::new(Column::new("a", 0)), + Operator::Eq, + Arc::new(Literal::new(ScalarValue::Int32(Some(50)))), + )); + let outer_filter: Arc = + Arc::new(FilterExec::try_new(outer_predicate, inner_filter)?); + + // Should succeed without error + let statistics = outer_filter.partition_statistics(None)?; + assert_eq!(statistics.num_rows, Precision::Inexact(0)); + + Ok(()) + } + #[tokio::test] async fn test_filter_statistics_more_inputs() -> Result<()> { let schema = Schema::new(vec![ From 41acbf8e4bb4ac15003bd5365661e6b17551f7f0 Mon Sep 17 00:00:00 2001 From: Andrew Lamb Date: Thu, 12 Mar 2026 11:57:34 -0400 Subject: [PATCH 04/12] [branch-52] fix: Return `probe_side.len()` for RightMark/Anti count(*) queries (#20710) (#20881) - Part of https://github.com/apache/datafusion/issues/20855 - Closes https://github.com/apache/datafusion/issues/20669 on branch-52 This PR: - Backports https://github.com/apache/datafusion/pull/20710 from @jonathanc-n to the branch-52 line Co-authored-by: Jonathan Chen --- .../src/joins/hash_join/stream.rs | 3 ++ .../src/joins/symmetric_hash_join.rs | 3 ++ datafusion/physical-plan/src/joins/utils.rs | 15 ++++++- datafusion/sqllogictest/test_files/joins.slt | 43 +++++++++++++++++++ 4 files changed, 63 insertions(+), 1 deletion(-) diff --git a/datafusion/physical-plan/src/joins/hash_join/stream.rs b/datafusion/physical-plan/src/joins/hash_join/stream.rs index e6735675125bd..c5c794f5a8c68 100644 --- a/datafusion/physical-plan/src/joins/hash_join/stream.rs +++ b/datafusion/physical-plan/src/joins/hash_join/stream.rs @@ -639,6 +639,7 @@ impl HashJoinStream { filter, JoinSide::Left, None, + self.join_type, )? } else { (left_indices, right_indices) @@ -707,6 +708,7 @@ impl HashJoinStream { &right_indices, &self.column_indices, join_side, + self.join_type, )?; self.output_buffer.push_batch(batch)?; @@ -770,6 +772,7 @@ impl HashJoinStream { &right_side, &self.column_indices, JoinSide::Left, + self.join_type, )?; self.output_buffer.push_batch(batch)?; } diff --git a/datafusion/physical-plan/src/joins/symmetric_hash_join.rs b/datafusion/physical-plan/src/joins/symmetric_hash_join.rs index 1f6bc703a0300..a75a9893e9f1a 100644 --- a/datafusion/physical-plan/src/joins/symmetric_hash_join.rs +++ b/datafusion/physical-plan/src/joins/symmetric_hash_join.rs @@ -930,6 +930,7 @@ pub(crate) fn build_side_determined_results( &probe_indices, column_indices, build_hash_joiner.build_side, + join_type, ) .map(|batch| (batch.num_rows() > 0).then_some(batch)) } else { @@ -993,6 +994,7 @@ pub(crate) fn join_with_probe_batch( filter, build_hash_joiner.build_side, None, + join_type, )? } else { (build_indices, probe_indices) @@ -1031,6 +1033,7 @@ pub(crate) fn join_with_probe_batch( &probe_indices, column_indices, build_hash_joiner.build_side, + join_type, ) .map(|batch| (batch.num_rows() > 0).then_some(batch)) } diff --git a/datafusion/physical-plan/src/joins/utils.rs b/datafusion/physical-plan/src/joins/utils.rs index a9243fe04e28d..53b4c4f80236e 100644 --- a/datafusion/physical-plan/src/joins/utils.rs +++ b/datafusion/physical-plan/src/joins/utils.rs @@ -910,6 +910,7 @@ pub(crate) fn get_final_indices_from_bit_map( (left_indices, right_indices) } +#[expect(clippy::too_many_arguments)] pub(crate) fn apply_join_filter_to_indices( build_input_buffer: &RecordBatch, probe_batch: &RecordBatch, @@ -918,6 +919,7 @@ pub(crate) fn apply_join_filter_to_indices( filter: &JoinFilter, build_side: JoinSide, max_intermediate_size: Option, + join_type: JoinType, ) -> Result<(UInt64Array, UInt32Array)> { if build_indices.is_empty() && probe_indices.is_empty() { return Ok((build_indices, probe_indices)); @@ -938,6 +940,7 @@ pub(crate) fn apply_join_filter_to_indices( &probe_indices.slice(i, len), filter.column_indices(), build_side, + join_type, )?; let filter_result = filter .expression() @@ -959,6 +962,7 @@ pub(crate) fn apply_join_filter_to_indices( &probe_indices, filter.column_indices(), build_side, + join_type, )?; filter @@ -979,6 +983,7 @@ pub(crate) fn apply_join_filter_to_indices( /// Returns a new [RecordBatch] by combining the `left` and `right` according to `indices`. /// The resulting batch has [Schema] `schema`. +#[expect(clippy::too_many_arguments)] pub(crate) fn build_batch_from_indices( schema: &Schema, build_input_buffer: &RecordBatch, @@ -987,11 +992,19 @@ pub(crate) fn build_batch_from_indices( probe_indices: &UInt32Array, column_indices: &[ColumnIndex], build_side: JoinSide, + join_type: JoinType, ) -> Result { if schema.fields().is_empty() { + // For RightAnti and RightSemi joins, after `adjust_indices_by_join_type` + // the build_indices were untouched so only probe_indices hold the actual + // row count. + let row_count = match join_type { + JoinType::RightAnti | JoinType::RightSemi => probe_indices.len(), + _ => build_indices.len(), + }; let options = RecordBatchOptions::new() .with_match_field_names(true) - .with_row_count(Some(build_indices.len())); + .with_row_count(Some(row_count)); return Ok(RecordBatch::try_new_with_options( Arc::new(schema.clone()), diff --git a/datafusion/sqllogictest/test_files/joins.slt b/datafusion/sqllogictest/test_files/joins.slt index 42441fe787dbe..1155bc4f3b2bf 100644 --- a/datafusion/sqllogictest/test_files/joins.slt +++ b/datafusion/sqllogictest/test_files/joins.slt @@ -5226,3 +5226,46 @@ DROP TABLE issue_20437_small; statement count 0 DROP TABLE issue_20437_large; + +# Test count(*) with right semi/anti joins returns correct row counts +# issue: https://github.com/apache/datafusion/issues/20669 + +statement ok +CREATE TABLE t1 (k INT, v INT); + +statement ok +CREATE TABLE t2 (k INT, v INT); + +statement ok +INSERT INTO t1 SELECT i AS k, i AS v FROM generate_series(1, 100) t(i); + +statement ok +INSERT INTO t2 VALUES (1, 1); + +query I +WITH t AS ( + SELECT * + FROM t1 + LEFT ANTI JOIN t2 ON t1.k = t2.k +) +SELECT count(*) +FROM t; +---- +99 + +query I +WITH t AS ( + SELECT * + FROM t1 + LEFT SEMI JOIN t2 ON t1.k = t2.k +) +SELECT count(*) +FROM t; +---- +1 + +statement count 0 +DROP TABLE t1; + +statement count 0 +DROP TABLE t2; From 2947378e9ef9dbdda75b4ff047edcfc1a06ef0d2 Mon Sep 17 00:00:00 2001 From: Andrew Lamb Date: Thu, 12 Mar 2026 14:06:47 -0400 Subject: [PATCH 05/12] [branch-52] fix: disable dynamic filter pushdown for non min/max aggregates (#20279) (#20877) - Part of https://github.com/apache/datafusion/issues/20855 - Closes https://github.com/apache/datafusion/issues/20267 on branch-52 This PR: - Backports https://github.com/apache/datafusion/pull/20279 from @notashes to the branch-52 line Co-authored-by: notashes Co-authored-by: Adrian Garcia Badaracco <1755071+adriangb@users.noreply.github.com> --- .../physical-plan/src/aggregates/mod.rs | 2 +- .../dynamic_filter_pushdown_config.slt | 19 +++++++++++++++++++ 2 files changed, 20 insertions(+), 1 deletion(-) diff --git a/datafusion/physical-plan/src/aggregates/mod.rs b/datafusion/physical-plan/src/aggregates/mod.rs index 06f12a90195d2..c030507ea3dfa 100644 --- a/datafusion/physical-plan/src/aggregates/mod.rs +++ b/datafusion/physical-plan/src/aggregates/mod.rs @@ -980,7 +980,7 @@ impl AggregateExec { } else if fun_name.eq_ignore_ascii_case("max") { DynamicFilterAggregateType::Max } else { - continue; + return; }; // 2. arg should be only 1 column reference diff --git a/datafusion/sqllogictest/test_files/dynamic_filter_pushdown_config.slt b/datafusion/sqllogictest/test_files/dynamic_filter_pushdown_config.slt index 3e403171e0718..6ee6ba0695cb8 100644 --- a/datafusion/sqllogictest/test_files/dynamic_filter_pushdown_config.slt +++ b/datafusion/sqllogictest/test_files/dynamic_filter_pushdown_config.slt @@ -257,6 +257,25 @@ physical_plan 04)------RepartitionExec: partitioning=RoundRobinBatch(4), input_partitions=1 05)--------DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/dynamic_filter_pushdown_config/agg_data.parquet]]}, projection=[score], file_type=parquet, predicate=category@0 = alpha AND DynamicFilter [ empty ], pruning_predicate=category_null_count@2 != row_count@3 AND category_min@0 <= alpha AND alpha <= category_max@1, required_guarantees=[category in (alpha)] +# Test 4b: COUNT + MAX — DynamicFilter should NOT appear here in mixed aggregates + +query TT +EXPLAIN SELECT COUNT(*), MAX(score) FROM agg_parquet WHERE category = 'alpha'; +---- +logical_plan +01)Projection: count(Int64(1)) AS count(*), max(agg_parquet.score) +02)--Aggregate: groupBy=[[]], aggr=[[count(Int64(1)), max(agg_parquet.score)]] +03)----Projection: agg_parquet.score +04)------Filter: agg_parquet.category = Utf8View("alpha") +05)--------TableScan: agg_parquet projection=[category, score], partial_filters=[agg_parquet.category = Utf8View("alpha")] +physical_plan +01)ProjectionExec: expr=[count(Int64(1))@0 as count(*), max(agg_parquet.score)@1 as max(agg_parquet.score)] +02)--AggregateExec: mode=Final, gby=[], aggr=[count(Int64(1)), max(agg_parquet.score)] +03)----CoalescePartitionsExec +04)------AggregateExec: mode=Partial, gby=[], aggr=[count(Int64(1)), max(agg_parquet.score)] +05)--------RepartitionExec: partitioning=RoundRobinBatch(4), input_partitions=1 +06)----------DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/dynamic_filter_pushdown_config/agg_data.parquet]]}, projection=[score], file_type=parquet, predicate=category@0 = alpha, pruning_predicate=category_null_count@2 != row_count@3 AND category_min@0 <= alpha AND alpha <= category_max@1, required_guarantees=[category in (alpha)] + # Disable aggregate dynamic filters only statement ok SET datafusion.optimizer.enable_aggregate_dynamic_filter_pushdown = false; From e5547e2772fbaed693e7472f38feab690a7fe3ef Mon Sep 17 00:00:00 2001 From: Andrew Lamb Date: Fri, 13 Mar 2026 14:25:33 -0400 Subject: [PATCH 06/12] [branch-52] Fix duplicate group keys after hash aggregation spill (#20724) (#20858) (#20917) - Part of https://github.com/apache/datafusion/issues/20855 - Closes https://github.com/apache/datafusion/issues/20724 on branch-52 This PR: - Backports https://github.com/apache/datafusion/pull/20858 from @gboucher90 to the branch-52 line Co-authored-by: gboucher90 --- datafusion/physical-plan/src/aggregates/row_hash.rs | 12 ++++++++++++ 1 file changed, 12 insertions(+) diff --git a/datafusion/physical-plan/src/aggregates/row_hash.rs b/datafusion/physical-plan/src/aggregates/row_hash.rs index 7cc59b44a301f..a6fc275723702 100644 --- a/datafusion/physical-plan/src/aggregates/row_hash.rs +++ b/datafusion/physical-plan/src/aggregates/row_hash.rs @@ -1233,6 +1233,18 @@ impl GroupedHashAggregateStream { // on the grouping columns. self.group_ordering = GroupOrdering::Full(GroupOrderingFull::new()); + // Recreate group_values to use streaming mode (GroupValuesColumn + // with scalarized_intern) which preserves input row order, as required + // by GroupOrderingFull. This is only needed for multi-column group by, + // since single-column uses GroupValuesPrimitive which is always safe. + let group_schema = self + .spill_state + .merging_group_by + .group_schema(&self.spill_state.spill_schema)?; + if group_schema.fields().len() > 1 { + self.group_values = new_group_values(group_schema, &self.group_ordering)?; + } + // Use `OutOfMemoryMode::ReportError` from this point on // to ensure we don't spill the spilled data to disk again. self.oom_mode = OutOfMemoryMode::ReportError; From 7e20eb7ddb3acf8174af7adec52859e28333d570 Mon Sep 17 00:00:00 2001 From: Matt Butrovich Date: Fri, 13 Mar 2026 19:16:45 -0400 Subject: [PATCH 07/12] [branch-52] perf: Cache num_output_rows in sort merge join to avoid O(n) recount (#20478) (#20936) --- .../src/joins/sort_merge_join/stream.rs | 29 +++++++++---------- 1 file changed, 13 insertions(+), 16 deletions(-) diff --git a/datafusion/physical-plan/src/joins/sort_merge_join/stream.rs b/datafusion/physical-plan/src/joins/sort_merge_join/stream.rs index 3a57dc6b41a57..dca55c720ef62 100644 --- a/datafusion/physical-plan/src/joins/sort_merge_join/stream.rs +++ b/datafusion/physical-plan/src/joins/sort_merge_join/stream.rs @@ -128,6 +128,8 @@ pub(super) struct StreamedBatch { pub join_arrays: Vec, /// Chunks of indices from buffered side (may be nulls) joined to streamed pub output_indices: Vec, + /// Total number of output rows across all chunks in `output_indices` + pub num_output_rows: usize, /// Index of currently scanned batch from buffered data pub buffered_batch_idx: Option, /// Indices that found a match for the given join filter @@ -144,6 +146,7 @@ impl StreamedBatch { idx: 0, join_arrays, output_indices: vec![], + num_output_rows: 0, buffered_batch_idx: None, join_filter_matched_idxs: HashSet::new(), } @@ -155,6 +158,7 @@ impl StreamedBatch { idx: 0, join_arrays: vec![], output_indices: vec![], + num_output_rows: 0, buffered_batch_idx: None, join_filter_matched_idxs: HashSet::new(), } @@ -162,10 +166,7 @@ impl StreamedBatch { /// Number of unfrozen output pairs in this streamed batch fn num_output_rows(&self) -> usize { - self.output_indices - .iter() - .map(|chunk| chunk.streamed_indices.len()) - .sum() + self.num_output_rows } /// Appends new pair consisting of current streamed index and `buffered_idx` @@ -175,7 +176,6 @@ impl StreamedBatch { buffered_batch_idx: Option, buffered_idx: Option, batch_size: usize, - num_unfrozen_pairs: usize, ) { // If no current chunk exists or current chunk is not for current buffered batch, // create a new chunk @@ -183,12 +183,13 @@ impl StreamedBatch { { // Compute capacity only when creating a new chunk (infrequent operation). // The capacity is the remaining space to reach batch_size. - // This should always be >= 1 since we only call this when num_unfrozen_pairs < batch_size. + // This should always be >= 1 since we only call this when num_output_rows < batch_size. debug_assert!( - batch_size > num_unfrozen_pairs, - "batch_size ({batch_size}) must be > num_unfrozen_pairs ({num_unfrozen_pairs})" + batch_size > self.num_output_rows, + "batch_size ({batch_size}) must be > num_output_rows ({})", + self.num_output_rows ); - let capacity = batch_size - num_unfrozen_pairs; + let capacity = batch_size - self.num_output_rows; self.output_indices.push(StreamedJoinedChunk { buffered_batch_idx, streamed_indices: UInt64Builder::with_capacity(capacity), @@ -205,6 +206,7 @@ impl StreamedBatch { } else { current_chunk.buffered_indices.append_null(); } + self.num_output_rows += 1; } } @@ -1134,13 +1136,10 @@ impl SortMergeJoinStream { let scanning_idx = self.buffered_data.scanning_idx(); if join_streamed { // Join streamed row and buffered row - // Pass batch_size and num_unfrozen_pairs to compute capacity only when - // creating a new chunk (when buffered_batch_idx changes), not on every iteration. self.streamed_batch.append_output_pair( Some(self.buffered_data.scanning_batch_idx), Some(scanning_idx), self.batch_size, - self.num_unfrozen_pairs(), ); } else { // Join nulls and buffered row for FULL join @@ -1166,13 +1165,10 @@ impl SortMergeJoinStream { // For Mark join we store a dummy id to indicate the row has a match let scanning_idx = mark_row_as_match.then_some(0); - // Pass batch_size=1 and num_unfrozen_pairs=0 to get capacity of 1, - // since we only append a single null-joined pair here (not in a loop). self.streamed_batch.append_output_pair( scanning_batch_idx, scanning_idx, - 1, - 0, + self.batch_size, ); self.buffered_data.scanning_finish(); self.streamed_joined = true; @@ -1469,6 +1465,7 @@ impl SortMergeJoinStream { } self.streamed_batch.output_indices.clear(); + self.streamed_batch.num_output_rows = 0; Ok(()) } From 5881edec5d937036891bbec9e7cb01837d9155a5 Mon Sep 17 00:00:00 2001 From: Andrew Lamb Date: Fri, 13 Mar 2026 20:29:33 -0400 Subject: [PATCH 08/12] [branch-52] fix: SanityCheckPlan error with window functions and NVL filter (#20231) (#20931) - Part of https://github.com/apache/datafusion/issues/19692 - Closes https://github.com/apache/datafusion/issues/20194 on branch-52 This PR: - Backports https://github.com/apache/datafusion/pull/20231 from @EeshanBembi to the branch-52 line --------- Co-authored-by: EeshanBembi <33062610+EeshanBembi@users.noreply.github.com> --- datafusion/physical-plan/src/filter.rs | 100 +++++++++++++++--- datafusion/sqllogictest/test_files/window.slt | 46 ++++++++ 2 files changed, 133 insertions(+), 13 deletions(-) diff --git a/datafusion/physical-plan/src/filter.rs b/datafusion/physical-plan/src/filter.rs index b96b99933cff5..9e849759ca653 100644 --- a/datafusion/physical-plan/src/filter.rs +++ b/datafusion/physical-plan/src/filter.rs @@ -55,12 +55,12 @@ use datafusion_common::{ use datafusion_execution::TaskContext; use datafusion_expr::Operator; use datafusion_physical_expr::equivalence::ProjectionMapping; -use datafusion_physical_expr::expressions::{BinaryExpr, Column, lit}; +use datafusion_physical_expr::expressions::{BinaryExpr, Column, Literal, lit}; use datafusion_physical_expr::intervals::utils::check_support; use datafusion_physical_expr::utils::{collect_columns, reassign_expr_columns}; use datafusion_physical_expr::{ - AcrossPartitions, AnalysisContext, ConstExpr, ExprBoundaries, PhysicalExpr, analyze, - conjunction, split_conjunction, + AcrossPartitions, AnalysisContext, ConstExpr, EquivalenceProperties, ExprBoundaries, + PhysicalExpr, analyze, conjunction, split_conjunction, }; use datafusion_physical_expr_common::physical_expr::fmt_sql; @@ -243,6 +243,20 @@ impl FilterExec { }) } + /// Returns the `AcrossPartitions` value for `expr` if it is constant: + /// either already known constant in `input_eqs`, or a `Literal` + /// (which is inherently constant across all partitions). + fn expr_constant_or_literal( + expr: &Arc, + input_eqs: &EquivalenceProperties, + ) -> Option { + input_eqs.is_expr_constant(expr).or_else(|| { + expr.as_any() + .downcast_ref::() + .map(|l| AcrossPartitions::Uniform(Some(l.value().clone()))) + }) + } + fn extend_constants( input: &Arc, predicate: &Arc, @@ -255,18 +269,24 @@ impl FilterExec { if let Some(binary) = conjunction.as_any().downcast_ref::() && binary.op() == &Operator::Eq { - // Filter evaluates to single value for all partitions - if input_eqs.is_expr_constant(binary.left()).is_some() { - let across = input_eqs - .is_expr_constant(binary.right()) - .unwrap_or_default(); + // Check if either side is constant — either already known + // constant from the input equivalence properties, or a literal + // value (which is inherently constant across all partitions). + let left_const = Self::expr_constant_or_literal(binary.left(), input_eqs); + let right_const = + Self::expr_constant_or_literal(binary.right(), input_eqs); + + if let Some(left_across) = left_const { + // LEFT is constant, so RIGHT must also be constant. + // Use RIGHT's known across value if available, otherwise + // propagate LEFT's (e.g. Uniform from a literal). + let across = right_const.unwrap_or(left_across); res_constants .push(ConstExpr::new(Arc::clone(binary.right()), across)); - } else if input_eqs.is_expr_constant(binary.right()).is_some() { - let across = input_eqs - .is_expr_constant(binary.left()) - .unwrap_or_default(); - res_constants.push(ConstExpr::new(Arc::clone(binary.left()), across)); + } else if let Some(right_across) = right_const { + // RIGHT is constant, so LEFT must also be constant. + res_constants + .push(ConstExpr::new(Arc::clone(binary.left()), right_across)); } } } @@ -866,6 +886,19 @@ fn collect_columns_from_predicate_inner( let predicates = split_conjunction(predicate); predicates.into_iter().for_each(|p| { if let Some(binary) = p.as_any().downcast_ref::() { + // Only extract pairs where at least one side is a Column reference. + // Pairs like `complex_expr = literal` should not create equivalence + // classes — the literal could appear in many unrelated expressions + // (e.g. sort keys), and normalize_expr's deep traversal would + // replace those occurrences with the complex expression, corrupting + // sort orderings. Constant propagation for such pairs is handled + // separately by `extend_constants`. + let has_direct_column_operand = + binary.left().as_any().downcast_ref::().is_some() + || binary.right().as_any().downcast_ref::().is_some(); + if !has_direct_column_operand { + return; + } match binary.op() { Operator::Eq => { eq_predicate_columns.push((binary.left(), binary.right())) @@ -1700,6 +1733,47 @@ mod tests { from output schema (c@0) to input schema (c@2)" ); + Ok(()) + } + /// Regression test for https://github.com/apache/datafusion/issues/20194 + /// + /// `collect_columns_from_predicate_inner` should only extract equality + /// pairs where at least one side is a Column. Pairs like + /// `complex_expr = literal` must not create equivalence classes because + /// `normalize_expr`'s deep traversal would replace the literal inside + /// unrelated expressions (e.g. sort keys) with the complex expression. + #[test] + fn test_collect_columns_skips_non_column_pairs() -> Result<()> { + let schema = test::aggr_test_schema(); + + // Simulate: nvl(c2, 0) = 0 → (c2 IS DISTINCT FROM 0) = 0 + // Neither side is a Column, so this should NOT be extracted. + let complex_expr: Arc = binary( + col("c2", &schema)?, + Operator::IsDistinctFrom, + lit(0u32), + &schema, + )?; + let predicate: Arc = + binary(complex_expr, Operator::Eq, lit(0u32), &schema)?; + + let (equal_pairs, _) = collect_columns_from_predicate_inner(&predicate); + assert_eq!( + 0, + equal_pairs.len(), + "Should not extract equality pairs where neither side is a Column" + ); + + // But col = literal should still be extracted + let predicate: Arc = + binary(col("c2", &schema)?, Operator::Eq, lit(0u32), &schema)?; + let (equal_pairs, _) = collect_columns_from_predicate_inner(&predicate); + assert_eq!( + 1, + equal_pairs.len(), + "Should extract equality pairs where one side is a Column" + ); + Ok(()) } } diff --git a/datafusion/sqllogictest/test_files/window.slt b/datafusion/sqllogictest/test_files/window.slt index 8ac8724683a8a..b1329bf55346f 100644 --- a/datafusion/sqllogictest/test_files/window.slt +++ b/datafusion/sqllogictest/test_files/window.slt @@ -6081,3 +6081,49 @@ WHERE acctbal > ( ); ---- 1 + +# Regression test for https://github.com/apache/datafusion/issues/20194 +# Window function with CASE WHEN in ORDER BY combined with NVL filter +# should not trigger SanityCheckPlan error from equivalence normalization +# replacing literals in sort expressions with complex filter expressions. +statement ok +CREATE TABLE issue_20194_t1 ( + value_1_1 decimal(25) NULL, + value_1_2 int NULL, + value_1_3 bigint NULL +); + +statement ok +CREATE TABLE issue_20194_t2 ( + value_2_1 bigint NULL, + value_2_2 varchar(140) NULL, + value_2_3 varchar(140) NULL +); + +statement ok +INSERT INTO issue_20194_t1 (value_1_1, value_1_2, value_1_3) VALUES (6774502793, 10040029, 1120); + +statement ok +INSERT INTO issue_20194_t2 (value_2_1, value_2_2, value_2_3) VALUES (1120, '0', '0'); + +query RII +SELECT + t1.value_1_1, t1.value_1_2, + ROW_NUMBER() OVER ( + PARTITION BY t1.value_1_1, t1.value_1_2 + ORDER BY + CASE WHEN t2.value_2_2 = '0' THEN 1 ELSE 0 END ASC, + CASE WHEN t2.value_2_3 = '0' THEN 1 ELSE 0 END ASC + ) AS ord +FROM issue_20194_t1 t1 +INNER JOIN issue_20194_t2 t2 + ON t1.value_1_3 = t2.value_2_1 + AND nvl(t2.value_2_3, '0') = '0'; +---- +6774502793 10040029 1 + +statement ok +DROP TABLE issue_20194_t1; + +statement ok +DROP TABLE issue_20194_t2; From 74aaa65001afd7bc649f471bcf634d52744c46fd Mon Sep 17 00:00:00 2001 From: Oleks V Date: Wed, 18 Mar 2026 04:13:09 -0700 Subject: [PATCH 09/12] [branch-52] chore: Ignore RUSTSEC-2024-0014 (#20862) (#21020) - Closes #. ``` Crate: generational-arena Version: 0.2.9 Warning: unmaintained Title: `generational-arena` is unmaintained Date: 2024-02-11 ID: RUSTSEC-2024-0014 URL: https://rustsec.org/advisories/RUSTSEC-2024-0014 ``` ## Which issue does this PR close? - Closes #. ## Rationale for this change ## What changes are included in this PR? ## Are these changes tested? ## Are there any user-facing changes? --- .github/workflows/audit.yml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/.github/workflows/audit.yml b/.github/workflows/audit.yml index 37f8f8a525407..9ba1fe39d324b 100644 --- a/.github/workflows/audit.yml +++ b/.github/workflows/audit.yml @@ -48,4 +48,4 @@ jobs: - name: Run audit check # Note: you can ignore specific RUSTSEC issues using the `--ignore` flag ,for example: # run: cargo audit --ignore RUSTSEC-2026-0001 - run: cargo audit + run: cargo audit --ignore RUSTSEC-2024-0436 --ignore RUSTSEC-2024-0014 From 664099b60640097a982e63174a96d8828fe1dc0d Mon Sep 17 00:00:00 2001 From: Andrew Lamb Date: Wed, 18 Mar 2026 07:16:23 -0400 Subject: [PATCH 10/12] [branch-52] fix: InList Dictionary filter pushdown type mismatch (#20962) (#20997) - Part of https://github.com/apache/datafusion/issues/20855 - Closes #20997 on branch-52 This PR: - Backports https://github.com/apache/datafusion/pull/20962 from @erratic-pattern to the branch-52 line - Backports the related tests from https://github.com/apache/datafusion/pull/20960 Co-authored-by: Adam Curtis --- .../physical-expr/src/expressions/in_list.rs | 545 +++++++++++++++++- .../test_files/parquet_filter_pushdown.slt | 99 ++++ 2 files changed, 641 insertions(+), 3 deletions(-) diff --git a/datafusion/physical-expr/src/expressions/in_list.rs b/datafusion/physical-expr/src/expressions/in_list.rs index 5c2f1adcd0cf3..379bd7edf7e69 100644 --- a/datafusion/physical-expr/src/expressions/in_list.rs +++ b/datafusion/physical-expr/src/expressions/in_list.rs @@ -98,11 +98,18 @@ impl StaticFilter for ArrayStaticFilter { )); } + // Unwrap dictionary-encoded needles when the value type matches + // in_array, evaluating against the dictionary values and mapping + // back via keys. downcast_dictionary_array! { v => { - let values_contains = self.contains(v.values().as_ref(), negated)?; - let result = take(&values_contains, v.keys(), None)?; - return Ok(downcast_array(result.as_ref())) + // Only unwrap when the haystack (in_array) type matches + // the dictionary value type + if v.values().data_type() == self.in_array.data_type() { + let values_contains = self.contains(v.values().as_ref(), negated)?; + let result = take(&values_contains, v.keys(), None)?; + return Ok(downcast_array(result.as_ref())); + } } _ => {} } @@ -3507,4 +3514,536 @@ mod tests { Ok(()) } + /// Helper: creates an InListExpr with `static_filter = None` + /// to force the column-reference evaluation path. + fn make_in_list_with_columns( + expr: Arc, + list: Vec>, + negated: bool, + ) -> Arc { + Arc::new(InListExpr::new(expr, list, negated, None)) + } + + #[test] + fn test_in_list_with_columns_int32_scalars() -> Result<()> { + // Column-reference path with scalar literals (bypassing static filter) + let schema = Schema::new(vec![Field::new("a", DataType::Int32, true)]); + let col_a = col("a", &schema)?; + let batch = RecordBatch::try_new( + Arc::new(schema), + vec![Arc::new(Int32Array::from(vec![ + Some(1), + Some(2), + Some(3), + None, + ]))], + )?; + + let list = vec![ + lit(ScalarValue::Int32(Some(1))), + lit(ScalarValue::Int32(Some(3))), + ]; + let expr = make_in_list_with_columns(col_a, list, false); + + let result = expr.evaluate(&batch)?.into_array(batch.num_rows())?; + let result = as_boolean_array(&result); + assert_eq!( + result, + &BooleanArray::from(vec![Some(true), Some(false), Some(true), None,]) + ); + Ok(()) + } + + #[test] + fn test_in_list_with_columns_int32_column_refs() -> Result<()> { + // IN list with column references + let schema = Schema::new(vec![ + Field::new("a", DataType::Int32, true), + Field::new("b", DataType::Int32, true), + Field::new("c", DataType::Int32, true), + ]); + let batch = RecordBatch::try_new( + Arc::new(schema.clone()), + vec![ + Arc::new(Int32Array::from(vec![Some(1), Some(2), Some(3), None])), + Arc::new(Int32Array::from(vec![ + Some(1), + Some(99), + Some(99), + Some(99), + ])), + Arc::new(Int32Array::from(vec![Some(99), Some(99), Some(3), None])), + ], + )?; + + let col_a = col("a", &schema)?; + let list = vec![col("b", &schema)?, col("c", &schema)?]; + let expr = make_in_list_with_columns(col_a, list, false); + + let result = expr.evaluate(&batch)?.into_array(batch.num_rows())?; + let result = as_boolean_array(&result); + // row 0: 1 IN (1, 99) → true + // row 1: 2 IN (99, 99) → false + // row 2: 3 IN (99, 3) → true + // row 3: NULL IN (99, NULL) → NULL + assert_eq!( + result, + &BooleanArray::from(vec![Some(true), Some(false), Some(true), None,]) + ); + Ok(()) + } + + #[test] + fn test_in_list_with_columns_utf8_column_refs() -> Result<()> { + // IN list with Utf8 column references + let schema = Schema::new(vec![ + Field::new("a", DataType::Utf8, false), + Field::new("b", DataType::Utf8, false), + ]); + let batch = RecordBatch::try_new( + Arc::new(schema.clone()), + vec![ + Arc::new(StringArray::from(vec!["x", "y", "z"])), + Arc::new(StringArray::from(vec!["x", "x", "z"])), + ], + )?; + + let col_a = col("a", &schema)?; + let list = vec![col("b", &schema)?]; + let expr = make_in_list_with_columns(col_a, list, false); + + let result = expr.evaluate(&batch)?.into_array(batch.num_rows())?; + let result = as_boolean_array(&result); + // row 0: "x" IN ("x") → true + // row 1: "y" IN ("x") → false + // row 2: "z" IN ("z") → true + assert_eq!(result, &BooleanArray::from(vec![true, false, true])); + Ok(()) + } + + #[test] + fn test_in_list_with_columns_negated() -> Result<()> { + // NOT IN with column references + let schema = Schema::new(vec![ + Field::new("a", DataType::Int32, false), + Field::new("b", DataType::Int32, false), + ]); + let batch = RecordBatch::try_new( + Arc::new(schema.clone()), + vec![ + Arc::new(Int32Array::from(vec![1, 2, 3])), + Arc::new(Int32Array::from(vec![1, 99, 3])), + ], + )?; + + let col_a = col("a", &schema)?; + let list = vec![col("b", &schema)?]; + let expr = make_in_list_with_columns(col_a, list, true); + + let result = expr.evaluate(&batch)?.into_array(batch.num_rows())?; + let result = as_boolean_array(&result); + // row 0: 1 NOT IN (1) → false + // row 1: 2 NOT IN (99) → true + // row 2: 3 NOT IN (3) → false + assert_eq!(result, &BooleanArray::from(vec![false, true, false])); + Ok(()) + } + + #[test] + fn test_in_list_with_columns_null_in_list() -> Result<()> { + // IN list with NULL scalar (column-reference path) + let schema = Schema::new(vec![Field::new("a", DataType::Int32, false)]); + let col_a = col("a", &schema)?; + let batch = RecordBatch::try_new( + Arc::new(schema), + vec![Arc::new(Int32Array::from(vec![1, 2]))], + )?; + + let list = vec![ + lit(ScalarValue::Int32(None)), + lit(ScalarValue::Int32(Some(1))), + ]; + let expr = make_in_list_with_columns(col_a, list, false); + + let result = expr.evaluate(&batch)?.into_array(batch.num_rows())?; + let result = as_boolean_array(&result); + // row 0: 1 IN (NULL, 1) → true (true OR null = true) + // row 1: 2 IN (NULL, 1) → NULL (false OR null = null) + assert_eq!(result, &BooleanArray::from(vec![Some(true), None])); + Ok(()) + } + + #[test] + fn test_in_list_with_columns_float_nan() -> Result<()> { + // Verify NaN == NaN is true in the column-reference path + // (consistent with Arrow's totalOrder semantics) + let schema = Schema::new(vec![ + Field::new("a", DataType::Float64, false), + Field::new("b", DataType::Float64, false), + ]); + let batch = RecordBatch::try_new( + Arc::new(schema.clone()), + vec![ + Arc::new(Float64Array::from(vec![f64::NAN, 1.0, f64::NAN])), + Arc::new(Float64Array::from(vec![f64::NAN, 2.0, 0.0])), + ], + )?; + + let col_a = col("a", &schema)?; + let list = vec![col("b", &schema)?]; + let expr = make_in_list_with_columns(col_a, list, false); + + let result = expr.evaluate(&batch)?.into_array(batch.num_rows())?; + let result = as_boolean_array(&result); + // row 0: NaN IN (NaN) → true + // row 1: 1.0 IN (2.0) → false + // row 2: NaN IN (0.0) → false + assert_eq!(result, &BooleanArray::from(vec![true, false, false])); + Ok(()) + } + + /// Tests that short-circuit evaluation produces correct results. + /// When all rows match after the first list item, remaining items + /// should be skipped without affecting correctness. + #[test] + fn test_in_list_with_columns_short_circuit() -> Result<()> { + // a IN (b, c) where b already matches every row of a + // The short-circuit should skip evaluating c + let schema = Schema::new(vec![ + Field::new("a", DataType::Int32, false), + Field::new("b", DataType::Int32, false), + Field::new("c", DataType::Int32, false), + ]); + let batch = RecordBatch::try_new( + Arc::new(schema.clone()), + vec![ + Arc::new(Int32Array::from(vec![1, 2, 3])), + Arc::new(Int32Array::from(vec![1, 2, 3])), // b == a for all rows + Arc::new(Int32Array::from(vec![99, 99, 99])), + ], + )?; + + let col_a = col("a", &schema)?; + let list = vec![col("b", &schema)?, col("c", &schema)?]; + let expr = make_in_list_with_columns(col_a, list, false); + + let result = expr.evaluate(&batch)?.into_array(batch.num_rows())?; + let result = as_boolean_array(&result); + assert_eq!(result, &BooleanArray::from(vec![true, true, true])); + Ok(()) + } + + /// Short-circuit must NOT skip when nulls are present (three-valued logic). + /// Even if all non-null values are true, null rows keep the result as null. + #[test] + fn test_in_list_with_columns_short_circuit_with_nulls() -> Result<()> { + // a IN (b, c) where a has nulls + // Even if b matches all non-null rows, result should preserve nulls + let schema = Schema::new(vec![ + Field::new("a", DataType::Int32, true), + Field::new("b", DataType::Int32, false), + Field::new("c", DataType::Int32, false), + ]); + let batch = RecordBatch::try_new( + Arc::new(schema.clone()), + vec![ + Arc::new(Int32Array::from(vec![Some(1), None, Some(3)])), + Arc::new(Int32Array::from(vec![1, 2, 3])), // matches non-null rows + Arc::new(Int32Array::from(vec![99, 99, 99])), + ], + )?; + + let col_a = col("a", &schema)?; + let list = vec![col("b", &schema)?, col("c", &schema)?]; + let expr = make_in_list_with_columns(col_a, list, false); + + let result = expr.evaluate(&batch)?.into_array(batch.num_rows())?; + let result = as_boolean_array(&result); + // row 0: 1 IN (1, 99) → true + // row 1: NULL IN (2, 99) → NULL + // row 2: 3 IN (3, 99) → true + assert_eq!( + result, + &BooleanArray::from(vec![Some(true), None, Some(true)]) + ); + Ok(()) + } + + /// Tests the make_comparator + collect_bool fallback path using + /// struct column references (nested types don't support arrow_eq). + #[test] + fn test_in_list_with_columns_struct() -> Result<()> { + let struct_fields = Fields::from(vec![ + Field::new("x", DataType::Int32, false), + Field::new("y", DataType::Utf8, false), + ]); + let struct_dt = DataType::Struct(struct_fields.clone()); + + let schema = Schema::new(vec![ + Field::new("a", struct_dt.clone(), true), + Field::new("b", struct_dt.clone(), false), + Field::new("c", struct_dt.clone(), false), + ]); + + // a: [{1,"a"}, {2,"b"}, NULL, {4,"d"}] + // b: [{1,"a"}, {9,"z"}, {3,"c"}, {4,"d"}] + // c: [{9,"z"}, {2,"b"}, {9,"z"}, {9,"z"}] + let a = Arc::new(StructArray::new( + struct_fields.clone(), + vec![ + Arc::new(Int32Array::from(vec![1, 2, 3, 4])), + Arc::new(StringArray::from(vec!["a", "b", "c", "d"])), + ], + Some(vec![true, true, false, true].into()), + )); + let b = Arc::new(StructArray::new( + struct_fields.clone(), + vec![ + Arc::new(Int32Array::from(vec![1, 9, 3, 4])), + Arc::new(StringArray::from(vec!["a", "z", "c", "d"])), + ], + None, + )); + let c = Arc::new(StructArray::new( + struct_fields.clone(), + vec![ + Arc::new(Int32Array::from(vec![9, 2, 9, 9])), + Arc::new(StringArray::from(vec!["z", "b", "z", "z"])), + ], + None, + )); + + let batch = RecordBatch::try_new(Arc::new(schema.clone()), vec![a, b, c])?; + + let col_a = col("a", &schema)?; + let list = vec![col("b", &schema)?, col("c", &schema)?]; + let expr = make_in_list_with_columns(col_a, list, false); + + let result = expr.evaluate(&batch)?.into_array(batch.num_rows())?; + let result = as_boolean_array(&result); + // row 0: {1,"a"} IN ({1,"a"}, {9,"z"}) → true (matches b) + // row 1: {2,"b"} IN ({9,"z"}, {2,"b"}) → true (matches c) + // row 2: NULL IN ({3,"c"}, {9,"z"}) → NULL + // row 3: {4,"d"} IN ({4,"d"}, {9,"z"}) → true (matches b) + assert_eq!( + result, + &BooleanArray::from(vec![Some(true), Some(true), None, Some(true)]) + ); + + // Also test NOT IN + let col_a = col("a", &schema)?; + let list = vec![col("b", &schema)?, col("c", &schema)?]; + let expr = make_in_list_with_columns(col_a, list, true); + + let result = expr.evaluate(&batch)?.into_array(batch.num_rows())?; + let result = as_boolean_array(&result); + // row 0: {1,"a"} NOT IN ({1,"a"}, {9,"z"}) → false + // row 1: {2,"b"} NOT IN ({9,"z"}, {2,"b"}) → false + // row 2: NULL NOT IN ({3,"c"}, {9,"z"}) → NULL + // row 3: {4,"d"} NOT IN ({4,"d"}, {9,"z"}) → false + assert_eq!( + result, + &BooleanArray::from(vec![Some(false), Some(false), None, Some(false)]) + ); + Ok(()) + } + + // ----------------------------------------------------------------------- + // Tests for try_new_from_array: evaluates `needle IN in_array`. + // + // This exercises the code path used by HashJoin dynamic filter pushdown, + // where in_array is built directly from the join's build-side arrays. + // Unlike try_new (used by SQL IN expressions), which always produces a + // non-Dictionary in_array because evaluate_list() flattens Dictionary + // scalars, try_new_from_array passes the array directly and can produce + // a Dictionary in_array. + // ----------------------------------------------------------------------- + + fn wrap_in_dict(array: ArrayRef) -> ArrayRef { + let keys = Int32Array::from((0..array.len() as i32).collect::>()); + Arc::new(DictionaryArray::new(keys, array)) + } + + /// Evaluates `needle IN in_array` via try_new_from_array, the same + /// path used by HashJoin dynamic filter pushdown (not the SQL literal + /// IN path which goes through try_new). + fn eval_in_list_from_array( + needle: ArrayRef, + in_array: ArrayRef, + ) -> Result { + let schema = + Schema::new(vec![Field::new("a", needle.data_type().clone(), false)]); + let col_a = col("a", &schema)?; + let expr = Arc::new(InListExpr::try_new_from_array(col_a, in_array, false)?) + as Arc; + let batch = RecordBatch::try_new(Arc::new(schema), vec![needle])?; + let result = expr.evaluate(&batch)?.into_array(batch.num_rows())?; + Ok(as_boolean_array(&result).clone()) + } + + #[test] + fn test_in_list_from_array_type_combinations() -> Result<()> { + use arrow::compute::cast; + + // All cases: needle[0] and needle[2] match, needle[1] does not. + let expected = BooleanArray::from(vec![Some(true), Some(false), Some(true)]); + + // Base arrays cast to each target type + let base_in = Arc::new(Int64Array::from(vec![1i64, 2, 3])) as ArrayRef; + let base_needle = Arc::new(Int64Array::from(vec![1i64, 4, 2])) as ArrayRef; + + // Test all specializations in instantiate_static_filter + let primitive_types = vec![ + DataType::Int8, + DataType::Int16, + DataType::Int32, + DataType::Int64, + DataType::UInt8, + DataType::UInt16, + DataType::UInt32, + DataType::UInt64, + DataType::Float32, + DataType::Float64, + ]; + + for dt in &primitive_types { + let in_array = cast(&base_in, dt)?; + let needle = cast(&base_needle, dt)?; + + // T in_array, T needle + assert_eq!( + expected, + eval_in_list_from_array(Arc::clone(&needle), Arc::clone(&in_array))?, + "same-type failed for {dt:?}" + ); + + // T in_array, Dict(Int32, T) needle + assert_eq!( + expected, + eval_in_list_from_array(wrap_in_dict(needle), in_array)?, + "dict-needle failed for {dt:?}" + ); + } + + // Utf8 (falls through to ArrayStaticFilter) + let utf8_in = Arc::new(StringArray::from(vec!["a", "b", "c"])) as ArrayRef; + let utf8_needle = Arc::new(StringArray::from(vec!["a", "d", "b"])) as ArrayRef; + + // Utf8 in_array, Utf8 needle + assert_eq!( + expected, + eval_in_list_from_array(Arc::clone(&utf8_needle), Arc::clone(&utf8_in),)? + ); + + // Utf8 in_array, Dict(Utf8) needle + assert_eq!( + expected, + eval_in_list_from_array( + wrap_in_dict(Arc::clone(&utf8_needle)), + Arc::clone(&utf8_in), + )? + ); + + // Dict(Utf8) in_array, Dict(Utf8) needle: the #20937 bug + assert_eq!( + expected, + eval_in_list_from_array( + wrap_in_dict(Arc::clone(&utf8_needle)), + wrap_in_dict(Arc::clone(&utf8_in)), + )? + ); + + // Struct in_array, Struct needle: multi-column join + let struct_fields = Fields::from(vec![ + Field::new("c0", DataType::Utf8, true), + Field::new("c1", DataType::Int64, true), + ]); + let make_struct = |c0: ArrayRef, c1: ArrayRef| -> ArrayRef { + let pairs: Vec<(FieldRef, ArrayRef)> = + struct_fields.iter().cloned().zip([c0, c1]).collect(); + Arc::new(StructArray::from(pairs)) + }; + assert_eq!( + expected, + eval_in_list_from_array( + make_struct( + Arc::clone(&utf8_needle), + Arc::new(Int64Array::from(vec![1, 4, 2])), + ), + make_struct( + Arc::clone(&utf8_in), + Arc::new(Int64Array::from(vec![1, 2, 3])), + ), + )? + ); + + // Struct with Dict fields: multi-column Dict join + let dict_struct_fields = Fields::from(vec![ + Field::new( + "c0", + DataType::Dictionary(Box::new(DataType::Int32), Box::new(DataType::Utf8)), + true, + ), + Field::new("c1", DataType::Int64, true), + ]); + let make_dict_struct = |c0: ArrayRef, c1: ArrayRef| -> ArrayRef { + let pairs: Vec<(FieldRef, ArrayRef)> = + dict_struct_fields.iter().cloned().zip([c0, c1]).collect(); + Arc::new(StructArray::from(pairs)) + }; + assert_eq!( + expected, + eval_in_list_from_array( + make_dict_struct( + wrap_in_dict(Arc::clone(&utf8_needle)), + Arc::new(Int64Array::from(vec![1, 4, 2])), + ), + make_dict_struct( + wrap_in_dict(Arc::clone(&utf8_in)), + Arc::new(Int64Array::from(vec![1, 2, 3])), + ), + )? + ); + + Ok(()) + } + + #[test] + fn test_in_list_from_array_type_mismatch_errors() -> Result<()> { + // Utf8 needle, Dict(Utf8) in_array + let err = eval_in_list_from_array( + Arc::new(StringArray::from(vec!["a", "d", "b"])), + wrap_in_dict(Arc::new(StringArray::from(vec!["a", "b", "c"]))), + ) + .unwrap_err() + .to_string(); + assert!( + err.contains("Can't compare arrays of different types"), + "{err}" + ); + + // Dict(Utf8) needle, Int64 in_array: specialized Int64StaticFilter + // rejects the Utf8 dictionary values at construction time + let err = eval_in_list_from_array( + wrap_in_dict(Arc::new(StringArray::from(vec!["a", "d", "b"]))), + Arc::new(Int64Array::from(vec![1, 2, 3])), + ) + .unwrap_err() + .to_string(); + assert!(err.contains("Failed to downcast"), "{err}"); + + // Dict(Int64) needle, Dict(Utf8) in_array: both Dict but different + // value types, make_comparator rejects the comparison + let err = eval_in_list_from_array( + wrap_in_dict(Arc::new(Int64Array::from(vec![1, 4, 2]))), + wrap_in_dict(Arc::new(StringArray::from(vec!["a", "b", "c"]))), + ) + .unwrap_err() + .to_string(); + assert!( + err.contains("Can't compare arrays of different types"), + "{err}" + ); + Ok(()) + } } diff --git a/datafusion/sqllogictest/test_files/parquet_filter_pushdown.slt b/datafusion/sqllogictest/test_files/parquet_filter_pushdown.slt index 5e643273baed4..d306f94ae310c 100644 --- a/datafusion/sqllogictest/test_files/parquet_filter_pushdown.slt +++ b/datafusion/sqllogictest/test_files/parquet_filter_pushdown.slt @@ -663,3 +663,102 @@ DROP TABLE o2_parquet_20696; # Cleanup settings statement ok set datafusion.execution.parquet.pushdown_filters = false; + +########## +# Regression test: filter pushdown with Struct columns in schema +# +# When a schema has Struct columns, Arrow field indices diverge from Parquet +# leaf indices (Struct children become separate leaves). A filter on a +# primitive column *after* a Struct must use the correct Parquet leaf index. +# +# Schema: +# Arrow: col_a=0 struct_col=1 col_b=2 +# Parquet: col_a=0 struct_col.x=1 struct_col.y=2 col_b=3 +########## + +statement ok +set datafusion.execution.parquet.pushdown_filters = true; + +statement ok +COPY ( + SELECT + column1 as col_a, + column2 as struct_col, + column3 as col_b + FROM VALUES + (1, {x: 10, y: 100}, 'aaa'), + (2, {x: 20, y: 200}, 'target'), + (3, {x: 30, y: 300}, 'zzz') +) TO 'test_files/scratch/parquet_filter_pushdown/struct_filter.parquet' +STORED AS PARQUET; + +statement ok +CREATE EXTERNAL TABLE t_struct_filter +STORED AS PARQUET +LOCATION 'test_files/scratch/parquet_filter_pushdown/struct_filter.parquet'; + +# Filter on col_b (the primitive column after the struct). +# Before the fix, this returned 0 rows because the filter read struct_col.y +# (Parquet leaf 2) instead of col_b (Parquet leaf 3). +query IT +SELECT col_a, col_b FROM t_struct_filter WHERE col_b = 'target'; +---- +2 target + +# Clean up +statement ok +set datafusion.execution.parquet.pushdown_filters = false; + +statement ok +DROP TABLE t_struct_filter; + +########## +# Regression test for https://github.com/apache/datafusion/issues/20937 +# +# Dynamic filter pushdown fails when joining VALUES against +# Dictionary-encoded Parquet columns. The InListExpr's ArrayStaticFilter +# unwraps the needle Dictionary but not the stored in_array, causing a +# make_comparator(Utf8, Dictionary) type mismatch. +########## + +statement ok +set datafusion.execution.parquet.pushdown_filters = true; + +statement ok +set datafusion.execution.parquet.reorder_filters = true; + +statement ok +COPY ( + SELECT + arrow_cast(chr(65 + (row_num % 26)), 'Dictionary(Int32, Utf8)') as tag1, + row_num * 1.0 as value + FROM (SELECT unnest(range(0, 10000)) as row_num) +) TO 'test_files/scratch/parquet_filter_pushdown/dict_filter_bug.parquet'; + +statement ok +CREATE EXTERNAL TABLE dict_filter_bug +STORED AS PARQUET +LOCATION 'test_files/scratch/parquet_filter_pushdown/dict_filter_bug.parquet'; + +query TR +SELECT t.tag1, t.value +FROM dict_filter_bug t +JOIN (VALUES ('A'), ('B')) AS v(c1) +ON t.tag1 = v.c1 +ORDER BY t.tag1, t.value +LIMIT 4; +---- +A 0 +A 26 +A 52 +A 78 + +# Cleanup +statement ok +set datafusion.execution.parquet.pushdown_filters = false; + +statement ok +set datafusion.execution.parquet.reorder_filters = false; + +statement ok +DROP TABLE dict_filter_bug; From e034c6b0b103c674c4576644007b30480565bec3 Mon Sep 17 00:00:00 2001 From: Andrew Lamb Date: Wed, 18 Mar 2026 08:14:46 -0400 Subject: [PATCH 11/12] [branch-52] Update to use lz4_flex 0.12.1 and quinn-proto 0.11.14 (#21009) ## Which issue does this PR close? - part of https://github.com/apache/datafusion/issues/20855 ## Rationale for this change `cargo audit` is failing on on branch-52 like this: ``` ... Crate: lz4_flex Version: 0.12.0 Warning: yanked error: 2 vulnerabilities found! warning: 4 allowed warnings found ``` here is an example of that heppening on CI: https://github.com/apache/datafusion/actions/runs/23209529148/job/67454157529?pr=21004 ## What changes are included in this PR? - Update lz4_flex 50 0.12.1 (non yanked) ## Are these changes tested? ## Are there any user-facing changes? --- Cargo.lock | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index dc938853f3aad..689e07cc77f09 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -4042,9 +4042,9 @@ checksum = "112b39cec0b298b6c1999fee3e31427f74f676e4cb9879ed1a121b43661a4154" [[package]] name = "lz4_flex" -version = "0.12.0" +version = "0.12.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "ab6473172471198271ff72e9379150e9dfd70d8e533e0752a27e515b48dd375e" +checksum = "98c23545df7ecf1b16c303910a69b079e8e251d60f7dd2cc9b4177f2afaf1746" dependencies = [ "twox-hash", ] @@ -4864,9 +4864,9 @@ dependencies = [ [[package]] name = "quinn-proto" -version = "0.11.13" +version = "0.11.14" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "f1906b49b0c3bc04b5fe5d86a77925ae6524a19b816ae38ce1e426255f1d8a31" +checksum = "434b42fec591c96ef50e21e886936e66d3cc3f737104fdb9b737c40ffb94c098" dependencies = [ "bytes", "getrandom 0.3.4", From e5bad58716cf74612ff3b245010411425063c3ec Mon Sep 17 00:00:00 2001 From: Andrew Lamb Date: Wed, 18 Mar 2026 09:01:57 -0400 Subject: [PATCH 12/12] [branch-52] Update version to 52.4.0 and update changelog (#21004) --- Cargo.lock | 84 +++++++++++++++---------------- Cargo.toml | 76 ++++++++++++++-------------- dev/changelog/52.4.0.md | 57 +++++++++++++++++++++ docs/source/user-guide/configs.md | 2 +- 4 files changed, 138 insertions(+), 81 deletions(-) create mode 100644 dev/changelog/52.4.0.md diff --git a/Cargo.lock b/Cargo.lock index 689e07cc77f09..cdf81366bc022 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1767,7 +1767,7 @@ dependencies = [ [[package]] name = "datafusion" -version = "52.3.0" +version = "52.4.0" dependencies = [ "arrow", "arrow-schema", @@ -1839,7 +1839,7 @@ dependencies = [ [[package]] name = "datafusion-benchmarks" -version = "52.3.0" +version = "52.4.0" dependencies = [ "arrow", "clap", @@ -1864,7 +1864,7 @@ dependencies = [ [[package]] name = "datafusion-catalog" -version = "52.3.0" +version = "52.4.0" dependencies = [ "arrow", "async-trait", @@ -1887,7 +1887,7 @@ dependencies = [ [[package]] name = "datafusion-catalog-listing" -version = "52.3.0" +version = "52.4.0" dependencies = [ "arrow", "async-trait", @@ -1909,7 +1909,7 @@ dependencies = [ [[package]] name = "datafusion-cli" -version = "52.3.0" +version = "52.4.0" dependencies = [ "arrow", "async-trait", @@ -1940,7 +1940,7 @@ dependencies = [ [[package]] name = "datafusion-common" -version = "52.3.0" +version = "52.4.0" dependencies = [ "ahash", "apache-avro", @@ -1967,7 +1967,7 @@ dependencies = [ [[package]] name = "datafusion-common-runtime" -version = "52.3.0" +version = "52.4.0" dependencies = [ "futures", "log", @@ -1976,7 +1976,7 @@ dependencies = [ [[package]] name = "datafusion-datasource" -version = "52.3.0" +version = "52.4.0" dependencies = [ "arrow", "async-compression", @@ -2011,7 +2011,7 @@ dependencies = [ [[package]] name = "datafusion-datasource-arrow" -version = "52.3.0" +version = "52.4.0" dependencies = [ "arrow", "arrow-ipc", @@ -2034,7 +2034,7 @@ dependencies = [ [[package]] name = "datafusion-datasource-avro" -version = "52.3.0" +version = "52.4.0" dependencies = [ "apache-avro", "arrow", @@ -2053,7 +2053,7 @@ dependencies = [ [[package]] name = "datafusion-datasource-csv" -version = "52.3.0" +version = "52.4.0" dependencies = [ "arrow", "async-trait", @@ -2074,7 +2074,7 @@ dependencies = [ [[package]] name = "datafusion-datasource-json" -version = "52.3.0" +version = "52.4.0" dependencies = [ "arrow", "async-trait", @@ -2094,7 +2094,7 @@ dependencies = [ [[package]] name = "datafusion-datasource-parquet" -version = "52.3.0" +version = "52.4.0" dependencies = [ "arrow", "async-trait", @@ -2123,11 +2123,11 @@ dependencies = [ [[package]] name = "datafusion-doc" -version = "52.3.0" +version = "52.4.0" [[package]] name = "datafusion-examples" -version = "52.3.0" +version = "52.4.0" dependencies = [ "arrow", "arrow-flight", @@ -2166,7 +2166,7 @@ dependencies = [ [[package]] name = "datafusion-execution" -version = "52.3.0" +version = "52.4.0" dependencies = [ "arrow", "async-trait", @@ -2187,7 +2187,7 @@ dependencies = [ [[package]] name = "datafusion-expr" -version = "52.3.0" +version = "52.4.0" dependencies = [ "arrow", "async-trait", @@ -2211,7 +2211,7 @@ dependencies = [ [[package]] name = "datafusion-expr-common" -version = "52.3.0" +version = "52.4.0" dependencies = [ "arrow", "datafusion-common", @@ -2222,7 +2222,7 @@ dependencies = [ [[package]] name = "datafusion-ffi" -version = "52.3.0" +version = "52.4.0" dependencies = [ "abi_stable", "arrow", @@ -2256,7 +2256,7 @@ dependencies = [ [[package]] name = "datafusion-functions" -version = "52.3.0" +version = "52.4.0" dependencies = [ "arrow", "arrow-buffer", @@ -2289,7 +2289,7 @@ dependencies = [ [[package]] name = "datafusion-functions-aggregate" -version = "52.3.0" +version = "52.4.0" dependencies = [ "ahash", "arrow", @@ -2310,7 +2310,7 @@ dependencies = [ [[package]] name = "datafusion-functions-aggregate-common" -version = "52.3.0" +version = "52.4.0" dependencies = [ "ahash", "arrow", @@ -2323,7 +2323,7 @@ dependencies = [ [[package]] name = "datafusion-functions-nested" -version = "52.3.0" +version = "52.4.0" dependencies = [ "arrow", "arrow-ord", @@ -2346,7 +2346,7 @@ dependencies = [ [[package]] name = "datafusion-functions-table" -version = "52.3.0" +version = "52.4.0" dependencies = [ "arrow", "async-trait", @@ -2360,7 +2360,7 @@ dependencies = [ [[package]] name = "datafusion-functions-window" -version = "52.3.0" +version = "52.4.0" dependencies = [ "arrow", "datafusion-common", @@ -2376,7 +2376,7 @@ dependencies = [ [[package]] name = "datafusion-functions-window-common" -version = "52.3.0" +version = "52.4.0" dependencies = [ "datafusion-common", "datafusion-physical-expr-common", @@ -2384,7 +2384,7 @@ dependencies = [ [[package]] name = "datafusion-macros" -version = "52.3.0" +version = "52.4.0" dependencies = [ "datafusion-doc", "quote", @@ -2393,7 +2393,7 @@ dependencies = [ [[package]] name = "datafusion-optimizer" -version = "52.3.0" +version = "52.4.0" dependencies = [ "arrow", "async-trait", @@ -2420,7 +2420,7 @@ dependencies = [ [[package]] name = "datafusion-physical-expr" -version = "52.3.0" +version = "52.4.0" dependencies = [ "ahash", "arrow", @@ -2447,7 +2447,7 @@ dependencies = [ [[package]] name = "datafusion-physical-expr-adapter" -version = "52.3.0" +version = "52.4.0" dependencies = [ "arrow", "datafusion-common", @@ -2460,7 +2460,7 @@ dependencies = [ [[package]] name = "datafusion-physical-expr-common" -version = "52.3.0" +version = "52.4.0" dependencies = [ "ahash", "arrow", @@ -2475,7 +2475,7 @@ dependencies = [ [[package]] name = "datafusion-physical-optimizer" -version = "52.3.0" +version = "52.4.0" dependencies = [ "arrow", "datafusion-common", @@ -2495,7 +2495,7 @@ dependencies = [ [[package]] name = "datafusion-physical-plan" -version = "52.3.0" +version = "52.4.0" dependencies = [ "ahash", "arrow", @@ -2531,7 +2531,7 @@ dependencies = [ [[package]] name = "datafusion-proto" -version = "52.3.0" +version = "52.4.0" dependencies = [ "arrow", "async-trait", @@ -2568,7 +2568,7 @@ dependencies = [ [[package]] name = "datafusion-proto-common" -version = "52.3.0" +version = "52.4.0" dependencies = [ "arrow", "datafusion-common", @@ -2580,7 +2580,7 @@ dependencies = [ [[package]] name = "datafusion-pruning" -version = "52.3.0" +version = "52.4.0" dependencies = [ "arrow", "datafusion-common", @@ -2598,7 +2598,7 @@ dependencies = [ [[package]] name = "datafusion-session" -version = "52.3.0" +version = "52.4.0" dependencies = [ "async-trait", "datafusion-common", @@ -2610,7 +2610,7 @@ dependencies = [ [[package]] name = "datafusion-spark" -version = "52.3.0" +version = "52.4.0" dependencies = [ "arrow", "bigdecimal", @@ -2632,7 +2632,7 @@ dependencies = [ [[package]] name = "datafusion-sql" -version = "52.3.0" +version = "52.4.0" dependencies = [ "arrow", "bigdecimal", @@ -2658,7 +2658,7 @@ dependencies = [ [[package]] name = "datafusion-sqllogictest" -version = "52.3.0" +version = "52.4.0" dependencies = [ "arrow", "async-trait", @@ -2689,7 +2689,7 @@ dependencies = [ [[package]] name = "datafusion-substrait" -version = "52.3.0" +version = "52.4.0" dependencies = [ "async-recursion", "async-trait", @@ -2710,7 +2710,7 @@ dependencies = [ [[package]] name = "datafusion-wasmtest" -version = "52.3.0" +version = "52.4.0" dependencies = [ "chrono", "console_error_panic_hook", diff --git a/Cargo.toml b/Cargo.toml index 9a0766efcb364..f8a1538358953 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -79,7 +79,7 @@ repository = "https://github.com/apache/datafusion" # Define Minimum Supported Rust Version (MSRV) rust-version = "1.88.0" # Define DataFusion version -version = "52.3.0" +version = "52.4.0" [workspace.dependencies] # We turn off default-features for some dependencies here so the workspaces which inherit them can @@ -112,43 +112,43 @@ chrono = { version = "0.4.42", default-features = false } criterion = "0.8" ctor = "0.6.3" dashmap = "6.0.1" -datafusion = { path = "datafusion/core", version = "52.3.0", default-features = false } -datafusion-catalog = { path = "datafusion/catalog", version = "52.3.0" } -datafusion-catalog-listing = { path = "datafusion/catalog-listing", version = "52.3.0" } -datafusion-common = { path = "datafusion/common", version = "52.3.0", default-features = false } -datafusion-common-runtime = { path = "datafusion/common-runtime", version = "52.3.0" } -datafusion-datasource = { path = "datafusion/datasource", version = "52.3.0", default-features = false } -datafusion-datasource-arrow = { path = "datafusion/datasource-arrow", version = "52.3.0", default-features = false } -datafusion-datasource-avro = { path = "datafusion/datasource-avro", version = "52.3.0", default-features = false } -datafusion-datasource-csv = { path = "datafusion/datasource-csv", version = "52.3.0", default-features = false } -datafusion-datasource-json = { path = "datafusion/datasource-json", version = "52.3.0", default-features = false } -datafusion-datasource-parquet = { path = "datafusion/datasource-parquet", version = "52.3.0", default-features = false } -datafusion-doc = { path = "datafusion/doc", version = "52.3.0" } -datafusion-execution = { path = "datafusion/execution", version = "52.3.0", default-features = false } -datafusion-expr = { path = "datafusion/expr", version = "52.3.0", default-features = false } -datafusion-expr-common = { path = "datafusion/expr-common", version = "52.3.0" } -datafusion-ffi = { path = "datafusion/ffi", version = "52.3.0" } -datafusion-functions = { path = "datafusion/functions", version = "52.3.0" } -datafusion-functions-aggregate = { path = "datafusion/functions-aggregate", version = "52.3.0" } -datafusion-functions-aggregate-common = { path = "datafusion/functions-aggregate-common", version = "52.3.0" } -datafusion-functions-nested = { path = "datafusion/functions-nested", version = "52.3.0", default-features = false } -datafusion-functions-table = { path = "datafusion/functions-table", version = "52.3.0" } -datafusion-functions-window = { path = "datafusion/functions-window", version = "52.3.0" } -datafusion-functions-window-common = { path = "datafusion/functions-window-common", version = "52.3.0" } -datafusion-macros = { path = "datafusion/macros", version = "52.3.0" } -datafusion-optimizer = { path = "datafusion/optimizer", version = "52.3.0", default-features = false } -datafusion-physical-expr = { path = "datafusion/physical-expr", version = "52.3.0", default-features = false } -datafusion-physical-expr-adapter = { path = "datafusion/physical-expr-adapter", version = "52.3.0", default-features = false } -datafusion-physical-expr-common = { path = "datafusion/physical-expr-common", version = "52.3.0", default-features = false } -datafusion-physical-optimizer = { path = "datafusion/physical-optimizer", version = "52.3.0" } -datafusion-physical-plan = { path = "datafusion/physical-plan", version = "52.3.0" } -datafusion-proto = { path = "datafusion/proto", version = "52.3.0" } -datafusion-proto-common = { path = "datafusion/proto-common", version = "52.3.0" } -datafusion-pruning = { path = "datafusion/pruning", version = "52.3.0" } -datafusion-session = { path = "datafusion/session", version = "52.3.0" } -datafusion-spark = { path = "datafusion/spark", version = "52.3.0" } -datafusion-sql = { path = "datafusion/sql", version = "52.3.0" } -datafusion-substrait = { path = "datafusion/substrait", version = "52.3.0" } +datafusion = { path = "datafusion/core", version = "52.4.0", default-features = false } +datafusion-catalog = { path = "datafusion/catalog", version = "52.4.0" } +datafusion-catalog-listing = { path = "datafusion/catalog-listing", version = "52.4.0" } +datafusion-common = { path = "datafusion/common", version = "52.4.0", default-features = false } +datafusion-common-runtime = { path = "datafusion/common-runtime", version = "52.4.0" } +datafusion-datasource = { path = "datafusion/datasource", version = "52.4.0", default-features = false } +datafusion-datasource-arrow = { path = "datafusion/datasource-arrow", version = "52.4.0", default-features = false } +datafusion-datasource-avro = { path = "datafusion/datasource-avro", version = "52.4.0", default-features = false } +datafusion-datasource-csv = { path = "datafusion/datasource-csv", version = "52.4.0", default-features = false } +datafusion-datasource-json = { path = "datafusion/datasource-json", version = "52.4.0", default-features = false } +datafusion-datasource-parquet = { path = "datafusion/datasource-parquet", version = "52.4.0", default-features = false } +datafusion-doc = { path = "datafusion/doc", version = "52.4.0" } +datafusion-execution = { path = "datafusion/execution", version = "52.4.0", default-features = false } +datafusion-expr = { path = "datafusion/expr", version = "52.4.0", default-features = false } +datafusion-expr-common = { path = "datafusion/expr-common", version = "52.4.0" } +datafusion-ffi = { path = "datafusion/ffi", version = "52.4.0" } +datafusion-functions = { path = "datafusion/functions", version = "52.4.0" } +datafusion-functions-aggregate = { path = "datafusion/functions-aggregate", version = "52.4.0" } +datafusion-functions-aggregate-common = { path = "datafusion/functions-aggregate-common", version = "52.4.0" } +datafusion-functions-nested = { path = "datafusion/functions-nested", version = "52.4.0", default-features = false } +datafusion-functions-table = { path = "datafusion/functions-table", version = "52.4.0" } +datafusion-functions-window = { path = "datafusion/functions-window", version = "52.4.0" } +datafusion-functions-window-common = { path = "datafusion/functions-window-common", version = "52.4.0" } +datafusion-macros = { path = "datafusion/macros", version = "52.4.0" } +datafusion-optimizer = { path = "datafusion/optimizer", version = "52.4.0", default-features = false } +datafusion-physical-expr = { path = "datafusion/physical-expr", version = "52.4.0", default-features = false } +datafusion-physical-expr-adapter = { path = "datafusion/physical-expr-adapter", version = "52.4.0", default-features = false } +datafusion-physical-expr-common = { path = "datafusion/physical-expr-common", version = "52.4.0", default-features = false } +datafusion-physical-optimizer = { path = "datafusion/physical-optimizer", version = "52.4.0" } +datafusion-physical-plan = { path = "datafusion/physical-plan", version = "52.4.0" } +datafusion-proto = { path = "datafusion/proto", version = "52.4.0" } +datafusion-proto-common = { path = "datafusion/proto-common", version = "52.4.0" } +datafusion-pruning = { path = "datafusion/pruning", version = "52.4.0" } +datafusion-session = { path = "datafusion/session", version = "52.4.0" } +datafusion-spark = { path = "datafusion/spark", version = "52.4.0" } +datafusion-sql = { path = "datafusion/sql", version = "52.4.0" } +datafusion-substrait = { path = "datafusion/substrait", version = "52.4.0" } doc-comment = "0.3" env_logger = "0.11" diff --git a/dev/changelog/52.4.0.md b/dev/changelog/52.4.0.md new file mode 100644 index 0000000000000..04fba07cde9e6 --- /dev/null +++ b/dev/changelog/52.4.0.md @@ -0,0 +1,57 @@ + + +# Apache DataFusion 52.4.0 Changelog + +This release consists of 11 commits from 10 contributors. See credits at the end of this changelog for more information. + +See the [upgrade guide](https://datafusion.apache.org/library-user-guide/upgrading.html) for information on how to upgrade from previous versions. + +**Other:** + +- [branch-52] fix: maintain inner list nullability for (#19948) [#20878](https://github.com/apache/datafusion/pull/20878) (Jefffrey) +- [branch-52] fix: Ensure columns are casted to the correct names with Unions (#20146) [#20879](https://github.com/apache/datafusion/pull/20879) (nuno-faria) +- [branch-52] fix: interval analysis error when have two filterexec that inner filter proves zero selectivity (#20743) [#20880](https://github.com/apache/datafusion/pull/20880) (haohuaijin) +- [branch-52] fix: Return `probe_side.len()` for RightMark/Anti count(\*) queries (#20710) [#20881](https://github.com/apache/datafusion/pull/20881) (jonathanc-n) +- [branch-52] fix: disable dynamic filter pushdown for non min/max aggregates (#20279) [#20877](https://github.com/apache/datafusion/pull/20877) (notashes) +- [branch-52] Fix duplicate group keys after hash aggregation spill (#20724) (#20858) [#20917](https://github.com/apache/datafusion/pull/20917) (gboucher90) +- [branch-52] perf: Cache num_output_rows in sort merge join to avoid O(n) recount (#20478) [#20936](https://github.com/apache/datafusion/pull/20936) (andygrove) +- [branch-52] fix: SanityCheckPlan error with window functions and NVL filter (#20231) [#20931](https://github.com/apache/datafusion/pull/20931) (EeshanBembi) +- [branch-52] chore: Ignore RUSTSEC-2024-0014 (#20862) [#21020](https://github.com/apache/datafusion/pull/21020) (comphead) +- [branch-52] fix: InList Dictionary filter pushdown type mismatch (#20962) [#20997](https://github.com/apache/datafusion/pull/20997) (alamb) +- [branch-52] Update to use lz4_flex 0.12.1 and quinn-proto 0.11.14 [#21009](https://github.com/apache/datafusion/pull/21009) (alamb) + +## Credits + +Thank you to everyone who contributed to this release. Here is a breakdown of commits (PRs merged) per contributor. + +``` + 2 Andrew Lamb + 1 Andy Grove + 1 EeshanBembi + 1 Guillaume Boucher + 1 Huaijin + 1 Jeffrey Vo + 1 Jonathan Chen + 1 Nuno Faria + 1 Oleks V + 1 notashes +``` + +Thank you also to everyone who contributed in other ways such as filing issues, reviewing PRs, and providing feedback on this release. diff --git a/docs/source/user-guide/configs.md b/docs/source/user-guide/configs.md index 8423480d53305..ebd9ef728ae7b 100644 --- a/docs/source/user-guide/configs.md +++ b/docs/source/user-guide/configs.md @@ -99,7 +99,7 @@ The following configuration settings are available: | datafusion.execution.parquet.dictionary_page_size_limit | 1048576 | (writing) Sets best effort maximum dictionary page size, in bytes | | datafusion.execution.parquet.statistics_enabled | page | (writing) Sets if statistics are enabled for any column Valid values are: "none", "chunk", and "page" These values are not case sensitive. If NULL, uses default parquet writer setting | | datafusion.execution.parquet.max_row_group_size | 1048576 | (writing) Target maximum number of rows in each row group (defaults to 1M rows). Writing larger row groups requires more memory to write, but can get better compression and be faster to read. | -| datafusion.execution.parquet.created_by | datafusion version 52.3.0 | (writing) Sets "created by" property | +| datafusion.execution.parquet.created_by | datafusion version 52.4.0 | (writing) Sets "created by" property | | datafusion.execution.parquet.column_index_truncate_length | 64 | (writing) Sets column index truncate length | | datafusion.execution.parquet.statistics_truncate_length | 64 | (writing) Sets statistics truncate length. If NULL, uses default parquet writer setting | | datafusion.execution.parquet.data_page_row_count_limit | 20000 | (writing) Sets best effort maximum number of rows in data page |