You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@datafusion.apache.org by al...@apache.org on 2025/09/26 18:37:23 UTC
(datafusion) branch branch-50 updated: Prevent exponential planning time for Window functions - v2 (#17684) (#17778)
This is an automated email from the ASF dual-hosted git repository.
alamb pushed a commit to branch branch-50
in repository https://gitbox.apache.org/repos/asf/datafusion.git
The following commit(s) were added to refs/heads/branch-50 by this push:
new 0aa293891a Prevent exponential planning time for Window functions - v2 (#17684) (#17778)
0aa293891a is described below
commit 0aa293891aaf540db854848bc517b5c915c78d1d
Author: Andrew Lamb <an...@nerdnetworks.org>
AuthorDate: Fri Sep 26 11:37:19 2025 -0700
Prevent exponential planning time for Window functions - v2 (#17684) (#17778)
* fix
* Update mod.rs
* Update mod.rs
* Update mod.rs
* tests copied from v1 pr
* test case from review comment
https://github.com/apache/datafusion/pull/17684#discussion_r2366146307
* one more test case
* Update mod.rs
* Update datafusion/physical-plan/src/windows/mod.rs
* Update datafusion/physical-plan/src/windows/mod.rs
* Update mod.rs
* Update mod.rs
---------
Co-authored-by: Berkay Şahin <12...@users.noreply.github.com>
Co-authored-by: Piotr Findeisen <pi...@gmail.com>
---
datafusion/core/benches/sql_planner.rs | 3 +-
datafusion/physical-plan/src/windows/mod.rs | 148 ++++++++++++++++++++------
datafusion/sqllogictest/test_files/window.slt | 89 ++++++++++++++++
3 files changed, 209 insertions(+), 31 deletions(-)
diff --git a/datafusion/core/benches/sql_planner.rs b/datafusion/core/benches/sql_planner.rs
index c71191507f..9ae6e1f570 100644
--- a/datafusion/core/benches/sql_planner.rs
+++ b/datafusion/core/benches/sql_planner.rs
@@ -476,7 +476,8 @@ fn criterion_benchmark(c: &mut Criterion) {
});
});
- for partitioning_columns in [4, 7, 8] {
+ // It was observed in production that queries with window functions sometimes partition over more than 30 columns
+ for partitioning_columns in [4, 7, 8, 12, 30] {
c.bench_function(
&format!(
"physical_window_function_partition_by_{partitioning_columns}_on_values"
diff --git a/datafusion/physical-plan/src/windows/mod.rs b/datafusion/physical-plan/src/windows/mod.rs
index dccd9200fc..ddc2bfa10e 100644
--- a/datafusion/physical-plan/src/windows/mod.rs
+++ b/datafusion/physical-plan/src/windows/mod.rs
@@ -371,17 +371,40 @@ pub(crate) fn window_equivalence_properties(
for (i, expr) in window_exprs.iter().enumerate() {
let partitioning_exprs = expr.partition_by();
let no_partitioning = partitioning_exprs.is_empty();
- // Collect columns defining partitioning, and construct all `SortOptions`
- // variations for them. Then, we will check each one whether it satisfies
- // the existing ordering provided by the input plan.
+
+ // Find "one" valid ordering for partition columns to avoid exponential complexity.
+ // see https://github.com/apache/datafusion/issues/17401
let mut all_satisfied_lexs = vec![];
- for lex in partitioning_exprs
- .iter()
- .map(|pb_order| sort_options_resolving_constant(Arc::clone(pb_order)))
- .multi_cartesian_product()
- .filter_map(LexOrdering::new)
- {
- if window_eq_properties.ordering_satisfy(lex.clone())? {
+ let mut candidate_ordering = vec![];
+
+ for partition_expr in partitioning_exprs.iter() {
+ let sort_options =
+ sort_options_resolving_constant(Arc::clone(partition_expr), true);
+
+ // Try each sort option and pick the first one that works
+ let mut found = false;
+ for sort_expr in sort_options.into_iter() {
+ candidate_ordering.push(sort_expr);
+ if let Some(lex) = LexOrdering::new(candidate_ordering.clone()) {
+ if window_eq_properties.ordering_satisfy(lex)? {
+ found = true;
+ break;
+ }
+ }
+ // This option didn't work, remove it and try the next one
+ candidate_ordering.pop();
+ }
+ // If no sort option works for this column, we can't build a valid ordering
+ if !found {
+ candidate_ordering.clear();
+ break;
+ }
+ }
+
+ // If we successfully built an ordering for all columns, use it
+ // When there are no partition expressions, candidate_ordering will be empty and won't be added
+ if candidate_ordering.len() == partitioning_exprs.len() {
+ if let Some(lex) = LexOrdering::new(candidate_ordering) {
all_satisfied_lexs.push(lex);
}
}
@@ -410,8 +433,10 @@ pub(crate) fn window_equivalence_properties(
// Window function results in a partial constant value in
// some ordering. Adjust the ordering equivalences accordingly:
let new_lexs = all_satisfied_lexs.into_iter().flat_map(|lex| {
- let new_partial_consts =
- sort_options_resolving_constant(Arc::clone(&window_col));
+ let new_partial_consts = sort_options_resolving_constant(
+ Arc::clone(&window_col),
+ false,
+ );
new_partial_consts.into_iter().map(move |partial| {
let mut existing = lex.clone();
@@ -467,23 +492,52 @@ pub(crate) fn window_equivalence_properties(
// utilize set-monotonicity since the set shrinks as the frame
// boundary starts "touching" the end of the table.
else if frame.is_causal() {
- let args_all_lexs = sliding_expr
- .get_aggregate_expr()
- .expressions()
- .into_iter()
- .map(sort_options_resolving_constant)
- .multi_cartesian_product();
-
- let (mut asc, mut satisfied) = (false, false);
- for order in args_all_lexs {
- if let Some(f) = order.first() {
- asc = !f.options.descending;
+ // Find one valid ordering for aggregate arguments instead of
+ // checking all combinations
+ let aggregate_exprs = sliding_expr.get_aggregate_expr().expressions();
+ let mut candidate_order = vec![];
+ let mut asc = false;
+
+ for (idx, expr) in aggregate_exprs.iter().enumerate() {
+ let mut found = false;
+ let sort_options =
+ sort_options_resolving_constant(Arc::clone(expr), false);
+
+ // Try each option and pick the first that works
+ for sort_expr in sort_options.into_iter() {
+ let is_asc = !sort_expr.options.descending;
+ candidate_order.push(sort_expr);
+
+ if let Some(lex) = LexOrdering::new(candidate_order.clone()) {
+ if window_eq_properties.ordering_satisfy(lex)? {
+ if idx == 0 {
+ // The first column's ordering direction determines the overall
+ // monotonicity behavior of the window result.
+ // - If the aggregate has increasing set monotonicity (e.g., MAX, COUNT)
+ // and the first arg is ascending, the window result is increasing
+ // - If the aggregate has decreasing set monotonicity (e.g., MIN)
+ // and the first arg is ascending, the window result is also increasing
+ // This flag is used to determine the final window column ordering.
+ asc = is_asc;
+ }
+ found = true;
+ break;
+ }
+ }
+ // This option didn't work, remove it and try the next one
+ candidate_order.pop();
}
- if window_eq_properties.ordering_satisfy(order)? {
- satisfied = true;
+
+ // If we couldn't extend the ordering, stop trying
+ if !found {
break;
}
}
+
+ // Check if we successfully built a complete ordering
+ let satisfied = candidate_order.len() == aggregate_exprs.len()
+ && !aggregate_exprs.is_empty();
+
if satisfied {
let increasing =
set_monotonicity.eq(&SetMonotonicity::Increasing);
@@ -634,11 +688,45 @@ pub fn get_window_mode(
Ok(None)
}
-fn sort_options_resolving_constant(expr: Arc<dyn PhysicalExpr>) -> Vec<PhysicalSortExpr> {
- vec![
- PhysicalSortExpr::new(Arc::clone(&expr), SortOptions::new(false, false)),
- PhysicalSortExpr::new(expr, SortOptions::new(true, true)),
- ]
+/// Generates sort option variations for a given expression.
+///
+/// This function is used to handle constant columns in window operations. Since constant
+/// columns can be considered as having any ordering, we generate multiple sort options
+/// to explore different ordering possibilities.
+///
+/// # Parameters
+/// - `expr`: The physical expression to generate sort options for
+/// - `only_monotonic`: If false, generates all 4 possible sort options (ASC/DESC × NULLS FIRST/LAST).
+/// If true, generates only 2 options that preserve set monotonicity.
+///
+/// # When to use `only_monotonic = false`:
+/// Use for PARTITION BY columns where we want to explore all possible orderings to find
+/// one that matches the existing data ordering.
+///
+/// # When to use `only_monotonic = true`:
+/// Use for aggregate/window function arguments where set monotonicity needs to be preserved.
+/// Only generates ASC NULLS LAST and DESC NULLS FIRST because:
+/// - Set monotonicity is broken if data has increasing order but nulls come first
+/// - Set monotonicity is broken if data has decreasing order but nulls come last
+fn sort_options_resolving_constant(
+ expr: Arc<dyn PhysicalExpr>,
+ only_monotonic: bool,
+) -> Vec<PhysicalSortExpr> {
+ if only_monotonic {
+ // Generate only the 2 options that preserve set monotonicity
+ vec![
+ PhysicalSortExpr::new(Arc::clone(&expr), SortOptions::new(false, false)), // ASC NULLS LAST
+ PhysicalSortExpr::new(expr, SortOptions::new(true, true)), // DESC NULLS FIRST
+ ]
+ } else {
+ // Generate all 4 possible sort options for partition columns
+ vec![
+ PhysicalSortExpr::new(Arc::clone(&expr), SortOptions::new(false, false)), // ASC NULLS LAST
+ PhysicalSortExpr::new(Arc::clone(&expr), SortOptions::new(false, true)), // ASC NULLS FIRST
+ PhysicalSortExpr::new(Arc::clone(&expr), SortOptions::new(true, false)), // DESC NULLS LAST
+ PhysicalSortExpr::new(expr, SortOptions::new(true, true)), // DESC NULLS FIRST
+ ]
+ }
}
#[cfg(test)]
diff --git a/datafusion/sqllogictest/test_files/window.slt b/datafusion/sqllogictest/test_files/window.slt
index c302582344..e81662a753 100644
--- a/datafusion/sqllogictest/test_files/window.slt
+++ b/datafusion/sqllogictest/test_files/window.slt
@@ -6034,3 +6034,92 @@ LIMIT 5
0 2 NULL NULL 0 NULL NULL
0 3 NULL NULL 0 NULL NULL
0 4 NULL NULL 0 NULL NULL
+
+# regression test for https://github.com/apache/datafusion/issues/17401
+query I
+WITH source AS (
+ SELECT
+ 1 AS n,
+ '' AS a1, '' AS a2, '' AS a3, '' AS a4, '' AS a5, '' AS a6, '' AS a7, '' AS a8,
+ '' AS a9, '' AS a10, '' AS a11, '' AS a12
+)
+SELECT
+ sum(n) OVER (PARTITION BY
+ a1, a2, a3, a4, a5, a6, a7, a8, a9, a10, a11, a12
+ )
+FROM source;
+----
+1
+
+# regression test for https://github.com/apache/datafusion/issues/17401
+query I
+WITH source AS (
+ SELECT
+ 1 AS n,
+ '' AS a1, '' AS a2, '' AS a3, '' AS a4, '' AS a5, '' AS a6, '' AS a7, '' AS a8,
+ '' AS a9, '' AS a10, '' AS a11, '' AS a12, '' AS a13, '' AS a14, '' AS a15, '' AS a16,
+ '' AS a17, '' AS a18, '' AS a19, '' AS a20, '' AS a21, '' AS a22, '' AS a23, '' AS a24,
+ '' AS a25, '' AS a26, '' AS a27, '' AS a28, '' AS a29, '' AS a30, '' AS a31, '' AS a32,
+ '' AS a33, '' AS a34, '' AS a35, '' AS a36, '' AS a37, '' AS a38, '' AS a39, '' AS a40
+)
+SELECT
+ sum(n) OVER (PARTITION BY
+ a1, a2, a3, a4, a5, a6, a7, a8, a9, a10, a11, a12, a13, a14, a15, a16, a17, a18, a19, a20,
+ a21, a22, a23, a24, a25, a26, a27, a28, a29, a30, a31, a32, a33, a34, a35, a36, a37, a38, a39, a40
+ )
+FROM source;
+----
+1
+
+# regression test for https://github.com/apache/datafusion/issues/17401
+query I
+WITH source AS (
+ SELECT
+ 1 AS n,
+ '' AS a1, '' AS a2, '' AS a3, '' AS a4, '' AS a5, '' AS a6, '' AS a7, '' AS a8,
+ '' AS a9, '' AS a10, '' AS a11, '' AS a12, '' AS a13, '' AS a14, '' AS a15, '' AS a16,
+ '' AS a17, '' AS a18, '' AS a19, '' AS a20, '' AS a21, '' AS a22, '' AS a23, '' AS a24,
+ '' AS a25, '' AS a26, '' AS a27, '' AS a28, '' AS a29, '' AS a30, '' AS a31, '' AS a32,
+ '' AS a33, '' AS a34, '' AS a35, '' AS a36, '' AS a37, '' AS a38, '' AS a39, '' AS a40
+)
+SELECT
+ sum(n) OVER (PARTITION BY
+ a1, a2, a3, a4, a5, a6, a7, a8, a9, a10, a11, a12, a13, a14, a15, a16, a17, a18, a19, a20,
+ a21, a22, a23, a24, a25, a26, a27, a28, a29, a30, a31, a32, a33, a34, a35, a36, a37, a38, a39, a40
+ )
+FROM (
+ SELECT * FROM source
+ ORDER BY a1, a2, a3, a4, a5, a6, a7, a8, a9, a10, a11, a12, a13, a14, a15, a16, a17, a18, a19, a20,
+ a21, a22, a23, a24, a25, a26, a27, a28, a29, a30, a31, a32, a33, a34, a35, a36, a37, a38, a39, a40
+);
+----
+1
+
+# regression test for https://github.com/apache/datafusion/issues/17401
+query I
+WITH source AS (
+ SELECT
+ 1 AS n,
+ '' AS a1, '' AS a2, '' AS a3, '' AS a4, '' AS a5, '' AS a6, '' AS a7, '' AS a8,
+ '' AS a9, '' AS a10, '' AS a11, '' AS a12, '' AS a13, '' AS a14, '' AS a15, '' AS a16,
+ '' AS a17, '' AS a18, '' AS a19, '' AS a20, '' AS a21, '' AS a22, '' AS a23, '' AS a24,
+ '' AS a25, '' AS a26, '' AS a27, '' AS a28, '' AS a29, '' AS a30, '' AS a31, '' AS a32,
+ '' AS a33, '' AS a34, '' AS a35, '' AS a36, '' AS a37, '' AS a38, '' AS a39, '' AS a40
+)
+SELECT
+ sum(n) OVER (PARTITION BY
+ a1, a2, a3, a4, a5, a6, a7, a8, a9, a10, a11, a12, a13, a14, a15, a16, a17, a18, a19, a20,
+ a21, a22, a23, a24, a25, a26, a27, a28, a29, a30, a31, a32, a33, a34, a35, a36, a37, a38, a39, a40
+ )
+FROM (
+ SELECT * FROM source
+ WHERE a1 = '' AND a2 = '' AND a3 = '' AND a4 = '' AND a5 = '' AND a6 = '' AND a7 = '' AND a8 = ''
+ AND a9 = '' AND a10 = '' AND a11 = '' AND a12 = '' AND a13 = '' AND a14 = '' AND a15 = '' AND a16 = ''
+ AND a17 = '' AND a18 = '' AND a19 = '' AND a20 = '' AND a21 = '' AND a22 = '' AND a23 = '' AND a24 = ''
+ AND a25 = '' AND a26 = '' AND a27 = '' AND a28 = '' AND a29 = '' AND a30 = '' AND a31 = '' AND a32 = ''
+ AND a33 = '' AND a34 = '' AND a35 = '' AND a36 = '' AND a37 = '' AND a38 = '' AND a39 = '' AND a40 = ''
+ ORDER BY a1, a2, a3, a4, a5, a6, a7, a8, a9, a10, a11, a12, a13, a14, a15, a16, a17, a18, a19, a20,
+ a21, a22, a23, a24, a25, a26, a27, a28, a29, a30, a31, a32, a33, a34, a35, a36, a37, a38, a39, a40
+);
+----
+1
\ No newline at end of file
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@datafusion.apache.org
For additional commands, e-mail: commits-help@datafusion.apache.org