Prevent exponential planning time for Window functions - v2

[berkaysynnada]

Which issue does this PR close?

• Closes Exponential planning time when window function is partitioned by multiple columns #17401.

Rationale for this change

There is a complexity issue with window ordering calculations. Rather than computing all possibilities and eliminating the failing ones, we incrementally refine the ordering, keeping only elements that satisfy requirements.

related content: #14813, #17401, #17563

What changes are included in this PR?

This PR implements the algorithm mentioned in the issue. There is also one minor change: the sort_options_resolving_constant function now generates candidates according to the intended usage. This enables a minor optimization possibility.

Are these changes tested?

Are there any user-facing changes?

[berkaysynnada]

@findepi can you please review this? I've implemented something similar to what I'd in my mind, and I hope it will solve your problems without sacrificing anything, I don't know how to measure the improvement on my computer though

[berkaysynnada]

cc @alamb also, if you are interested in this and want to review

[alamb]

🤖 ./gh_compare_branch_bench.sh Benchmark Script Running
Linux aal-dev 6.14.0-1014-gcp #15~24.04.1-Ubuntu SMP Fri Jul 25 23:26:08 UTC 2025 x86_64 x86_64 x86_64 GNU/Linux
Comparing fix/exponential-window-ordering-calc (43f7592) to 03b6789 diff
BENCH_NAME=sql_planner
BENCH_COMMAND=cargo bench --bench sql_planner
BENCH_FILTER=physical_window_function_partition
BENCH_BRANCH_NAME=fix_exponential-window-ordering-calc
Results will be posted here when complete

[berkaysynnada]

@findepi do you want me to take bench and slt changes from #17563 ?

[comphead]

Thanks @berkaysynnada nice to see benches though

[findepi]

@findepi do you want me to take bench and slt changes from #17563 ?

yes, please

[findepi]

This doesn't solve the original problem.

I checked out this PR and run cargo run --bin datafusion-cli

Partitioning over 4 columns

DataFusion CLI v50.0.0
> 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
    )
FROM source;
+----------------------------------------------------------------------------------------------------------------------------------+
| sum(source.n) PARTITION BY [source.a1, source.a2, source.a3, source.a4] ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING |
+----------------------------------------------------------------------------------------------------------------------------------+
| 1                                                                                                                                |
+----------------------------------------------------------------------------------------------------------------------------------+
1 row(s) fetched.
Elapsed 0.673 seconds.

Partitioning over 5 columns

> 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
    )
FROM source;
+---------------------------------------------------------------------------------------------------------------------------------------------+
| sum(source.n) PARTITION BY [source.a1, source.a2, source.a3, source.a4, source.a5] ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING |
+---------------------------------------------------------------------------------------------------------------------------------------------+
| 1                                                                                                                                           |
+---------------------------------------------------------------------------------------------------------------------------------------------+
1 row(s) fetched.
Elapsed 10.065 seconds.

Notice the time!

[findepi]

This condition looks redundant.

[findepi]

This condition still looks redundant. IIUC, when partitioning_exprs.is_empty() the code inside the THEN arm will not do anything

[findepi]

this looks exponential still. when is this code path taken?

[berkaysynnada]

Yes, this part is for calculating orderings of function results, and similar approach can also be applied here. If you think the partitioning expr update reasonable now, I'll apply the same here as well, in addition to the valid parts of your PR

[findepi]

let's maybe file issue / follow-up PR, so that we don't need to re-review code unnecessarily

[findepi]

Does this look potentially expensive?

[findepi]

In #17563 we have this example query

# 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;

with current PR code it never completes, but let's ignore this for a moment.

if i read the code correctly, it "prunes early" when next partitioning expression doesn't provide more benefits. So i thought that it won't prune in some cases. Perhaps when input is pre-partitioned?

Let's add a test query

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

[findepi]

However, I am not sure if i understand correctly the pruning condition and what's the condition for not pruning. Please elaborate.

[alamb]

This seems to actually have made planning time worse -- the benchmark run from yesterday is still running

Benchmarking physical_window_function_partition_by_7_on_values: Warming up for 3.0000 s
Warning: Unable to complete 100 samples in 5.0s. You may wish to increase target time to 30504.9s, or reduce sample count to 10.
physical_window_function_partition_by_7_on_values
                        time:   [303.87 s 304.07 s 304.28 s]
Found 5 outliers among 100 measurements (5.00%)
  5 (5.00%) high mild

Benchmarking physical_window_function_partition_by_8_on_values: Warming up for 3.0000 s
Warning: Unable to complete 100 samples in 5.0s. You may wish to increase target time to 696351.6s, or reduce sample count to 10.
Benchmarking physical_window_function_partition_by_8_on_values: Collecting 100 samples in estimated 696352 s (100 iterations)
...

(note I think 696351.6s means 8 days!)

.

[berkaysynnada]

DataFusion CLI v50.0.0
> 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
    )
FROM source;
+----------------------------------------------------------------------------------------------------------------------------------+
| sum(source.n) PARTITION BY [source.a1, source.a2, source.a3, source.a4] ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING |
+----------------------------------------------------------------------------------------------------------------------------------+
| 1                                                                                                                                |
+----------------------------------------------------------------------------------------------------------------------------------+
1 row(s) fetched. 
Elapsed 0.039 seconds.

> 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
    )
FROM source;
+---------------------------------------------------------------------------------------------------------------------------------------------+
| sum(source.n) PARTITION BY [source.a1, source.a2, source.a3, source.a4, source.a5] ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING |
+---------------------------------------------------------------------------------------------------------------------------------------------+
| 1                                                                                                                                           |
+---------------------------------------------------------------------------------------------------------------------------------------------+
1 row(s) fetched. 
Elapsed 0.018 seconds.

I missed the case where there are no ordering constraints. In that scenario, ordering_satisfy() returns true for nearly all combinations, so we were still keeping O(4^n) orderings despite pruning. I've now updated the approach as greedy.

1. For each partition column, try 4 sort options and pick the first that works
2. Move to the next column only with the chosen option
3. Stop immediately if any column has no valid option

Can you check this updated version @findepi ?

[findepi]

@findepi do you want me to take bench and slt changes from #17563 ?

The PR is currently light on tests. Can you please consider pulling synnada-ai#74 ?

[findepi]

the changed piece of code no longer looks like exhibiting exponential complexity

thank you!

please include tests (#17684 (comment))

[findepi]

This condition still looks redundant. IIUC, when partitioning_exprs.is_empty() the code inside the THEN arm will not do anything

[findepi]

nit:

ordering.clone() is avoidable.
logically, it's an expression being pushed-tried-else-popped:

// Find a single valid ordering using a greedy approach
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.iter() {
        candidate_ordering.push(sort_expr.clone());

        if let Some(lex) = LexOrdering::new(candidate_ordering.clone()) {
            if window_eq_properties.ordering_satisfy(lex)? {
                found = true;
                break;
            }
        } else {
            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
if candidate_ordering.len() == partitioning_exprs.len() {
    if let Some(lex) = LexOrdering::new(ordering) {
        all_satisfied_lexs.push(lex);
    }
}

[findepi]

With every new window sort expression iterated over, we seem to be processing all previous sort expressions. This feels quadratic.

[berkaysynnada]

Yes, it's quadratic as we validate orderings of length 1, 2, 3, n. But that's unavoidable since we need to check that each partial ordering stays valid as we extend it.

Still way better than before. Old version is O(4^n), the new one is O(n^2 * 4)

The quadratic cost is the price of correctness

[findepi]

No doubt quadratic is so much better than exponential.

What could make turn valid partial ordering into an invalid, when extended with a new expression?
Is it only about duplicates?
IIRC, LexOrdering::new checks for dups, so something we could easily do O(n) overall. But I don't know yet what window_eq_properties.ordering_satisfy does.

[findepi]

let's maybe file issue / follow-up PR, so that we don't need to re-review code unnecessarily

[alamb]

🤖 ./gh_compare_branch_bench.sh Benchmark Script Running
Linux aal-dev 6.14.0-1014-gcp #15~24.04.1-Ubuntu SMP Fri Jul 25 23:26:08 UTC 2025 x86_64 x86_64 x86_64 GNU/Linux
Comparing fix/exponential-window-ordering-calc (43a323f) to 03b6789 diff
BENCH_NAME=sql_planner
BENCH_COMMAND=cargo bench --bench sql_planner
BENCH_FILTER=physical_window_function_partition
BENCH_BRANCH_NAME=fix_exponential-window-ordering-calc
Results will be posted here when complete

[alamb]

🤖: Benchmark completed

Details

group                                                fix_exponential-window-ordering-calc    main
-----                                                ------------------------------------    ----
physical_window_function_partition_by_4_on_values    1.00    606.4±5.29µs        ? ?/sec     2.19   1327.3±5.88µs        ? ?/sec
physical_window_function_partition_by_7_on_values    1.00    736.8±4.67µs        ? ?/sec     48.32    35.6±0.16ms        ? ?/sec
physical_window_function_partition_by_8_on_values    1.00    797.4±6.45µs        ? ?/sec     173.51   138.4±0.54ms        ? ?/sec

[alamb]

🤖 ./gh_compare_branch_bench.sh Benchmark Script Running
Linux aal-dev 6.14.0-1014-gcp #15~24.04.1-Ubuntu SMP Fri Jul 25 23:26:08 UTC 2025 x86_64 x86_64 x86_64 GNU/Linux
Comparing fix/exponential-window-ordering-calc (43a323f) to 03b6789 diff
BENCH_NAME=sql_planner
BENCH_COMMAND=cargo bench --bench sql_planner
BENCH_FILTER=physical_window_function_partition
BENCH_BRANCH_NAME=fix_exponential-window-ordering-calc
Results will be posted here when complete

[alamb]

physical_window_function_partition_by_4_on_values    1.00    606.4±5.29µs        ? ?/sec     2.19   1327.3±5.88µs        ? ?/sec
physical_window_function_partition_by_7_on_values    1.00    736.8±4.67µs        ? ?/sec     48.32    35.6±0.16ms        ? ?/sec
physical_window_function_partition_by_8_on_values    1.00    797.4±6.45µs        ? ?/sec     173.51   138.4±0.54ms        ? ?/sec

🚀 -- it is not often we see a 173x speedup 🐈 🥳

[berkaysynnada]

the changed piece of code no longer looks like exhibiting exponential complexity

thank you!

please include tests (#17684 (comment))

I'll drive this to the finish line, probably tomorrow morning (all the stuff we've talked about). Thank you :)

[alamb]

🤖: Benchmark completed

Details

group                                                fix_exponential-window-ordering-calc    main
-----                                                ------------------------------------    ----
physical_window_function_partition_by_4_on_values    1.00    598.6±4.75µs        ? ?/sec     2.21  1323.9±14.77µs        ? ?/sec
physical_window_function_partition_by_7_on_values    1.00    751.3±5.72µs        ? ?/sec     47.12    35.4±0.15ms        ? ?/sec
physical_window_function_partition_by_8_on_values    1.00    798.7±3.28µs        ? ?/sec     173.17   138.3±0.48ms        ? ?/sec

[findepi]

🤖: Benchmark completed

thanks @alamb . these lgtm

[berkaysynnada]

I think it's ready. @findepi feel free to commit directly if you'd like to make any changes

DataFusion CLI v50.0.0
> 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
    )
FROM source;
+---------------------------------------------------------------------------------------------------------------------------------------------+
| sum(source.n) PARTITION BY [source.a1, source.a2, source.a3, source.a4, source.a5] ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING |
+---------------------------------------------------------------------------------------------------------------------------------------------+
| 1                                                                                                                                           |
+---------------------------------------------------------------------------------------------------------------------------------------------+
1 row(s) fetched. 
Elapsed 0.039 seconds.

[alamb]

Thank you @berkaysynnada -- I went through this PR carefully and I think it is a nice improvement over what is on main. Thank you for working on this and thank you to @findepi and @comphead for the reviews

[alamb]

Similarly here you could potentially use sort_options.into_iter() and save the clone below. Here is what I used

@@ -505,13 +505,14 @@ pub(crate) fn window_equivalence_properties(
                             sort_options_resolving_constant(Arc::clone(expr), false);

                         // Try each option and pick the first that works
-                        for sort_expr in sort_options.iter() {
-                            candidate_order.push(sort_expr.clone());
+                        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 {
-                                        asc = !sort_expr.options.descending;
+                                        asc = is_asc;
                                     }
                                     found = true;
                                     break;

[alamb]

FWIW this seems very similar to the loop above, I wonder if there is some way (as a follow on PR) to factor it out to reduce the replication

[alamb]

I ran this locally, and found

both main and This branch -- took 2 seconds

Main

(venv) andrewlamb@Andrews-MacBook-Pro-3:~/Software/datafusion$ cargo test --profile=ci --test sqllogictests  --  window.slt
    Finished `ci` profile [unoptimized + debuginfo] target(s) in 0.20s
     Running bin/sqllogictests.rs (target/ci/deps/sqllogictests-4dcb99f83e94c047)
Completed 2 test files in 2 seconds

This branch

(venv) andrewlamb@Andrews-MacBook-Pro-3:~/Software/datafusion$ cargo test --profile=ci --test sqllogictests  --  window.slt
    Finished `ci` profile [unoptimized + debuginfo] target(s) in 0.19s
     Running bin/sqllogictests.rs (target/ci/deps/sqllogictests-fbba93e4275b7826)
Completed 2 test files in 2 seconds

[findepi]

I ran this locally, and found

both main and This branch -- took 2 seconds

Main

Do you mean you run window.slt unmodified as currently in main, or did you apply changes from this PR first?

I tried to reproduce the latter and for me execution "hangs" at

[00:00:01] #######################################-     442/454     "window.slt"

[findepi]

&& !candidate_ordering.is_empty() is redundant.
in the empty case, the LexOrdering::new returns None

Suggested change

&& !candidate_ordering.is_empty()

[findepi]

why first is special? worth a code comment

(i know the logic is pre-existing and did not have a comment, but you seem to know what this is about and I do not)

[findepi]

Thanks for solving this exponential case too!

I am not sure the regression tests cover this case.
Do we need some more test queries?

[berkaysynnada]

datafusion/datafusion/core/tests/physical_optimizer/enforce_sorting.rs

Line 2470 in 6ec14e9

async fn test_window_partial_constant_and_set_monotonicity() -> Result<()> {

highly covers the correctness issue, but for the performance concerns, a similar regression test should be added, like this:

    SUM(c1 + c2 + c3 + c4 + c5 + c6 + c7 + c8 + c9 + c10 + ... + cN) 
        OVER (ORDER BY c1, c2, ... cN and a causal window)

[alamb]

Are there any remaining issues preventing merging this PR?

[alamb]

Ok, this looks good to me, so I'll merge it in now. Thank you @berkaysynnada and @findepi

[alamb]

Backport PR

• [branch-50] Backport Prevent exponential planning time for Window functions - v2 #17684  #17778