mithuncy opened a new pull request, #23416:
URL: https://github.com/apache/datafusion/pull/23416
## Which issue does this PR close?
- Closes #23289.
Related:
- EPIC: #22395
- Same pattern for aggregations: #23239
- Stacked on (and requires): #23184 — only the last commit here is this PR;
the first commit is that PR's base. Will rebase and mark ready once it merges.
- Sibling rollout for PartitionedTopK on the same base: #23355. Note that PR
currently also opts the two window execs into range satisfaction as part of its
PartitionedTopK work; this PR is the dedicated #23289 change with the
window-specific test suite the issue asks for, and scopes the opt-in to the
keyed branch only. Happy to coordinate / rebase whichever lands second (the slt
test numbering in `range_partitioning.slt` will need reconciling too).
- General `Range` satisfaction (future work): #23266
## Rationale for this change
`WindowAggExec` and `BoundedWindowAggExec` require
`Distribution::KeyPartitioned(partition_keys)`: all rows with the same
`PARTITION BY` key must be in the same partition. An input that is already
range partitioned on a compatible key (e.g. `Range([a])` for
`PARTITION BY a`) provides exactly that colocation, but only aggregates
currently opt in to range satisfaction. As a result, an unnecessary
`RepartitionExec: partitioning=Hash(...)` is inserted in front of windows
even when the input's range partitioning already satisfies the
requirement.
This PR is stacked on #23184, which introduces
`InputDistributionRequirements` and routes distribution satisfaction
through an operator-declared policy (replacing the optimizer-side private
gate from #23239). On that base, enabling windows is a declaration-level
change on the operators themselves.
## What changes are included in this PR?
Core change (two files, no optimizer changes):
- `datafusion/physical-plan/src/windows/window_agg_exec.rs` and
`datafusion/physical-plan/src/windows/bounded_window_agg_exec.rs`:
when `partition_keys()` is non-empty, the
`input_distribution_requirements()` declaration now appends
`.allow_range_satisfaction_for_key_partitioning()`, mirroring
`AggregateExec` on the base branch. The `ensure_distribution` and
`SanityCheckPlan` machinery honor the declared policy with no
operator-specific code.
Notes on scope:
- Windows with no `PARTITION BY`, or with
`datafusion.optimizer.repartition_windows = false`, have empty
`partition_keys()` and keep requiring `SinglePartition`; the opt-in is
not applied to them.
- Exact satisfaction (`Range([a])` for `PARTITION BY a`), subset
satisfaction (`Range([a])` for `PARTITION BY a, b` when enabled), and
hash fallback for incompatible keys all come from the base branch's
satisfaction machinery, unchanged.
- The per-operator opt-in is a rollout marker: once `Range` satisfies
`KeyPartitioned` generally (#23266), these opt-ins get deleted.
- Joins/co-partitioning are entirely the base PR's (#23184) concern; this
PR only touches single-input window operators.
## Are these changes tested?
Yes.
- SLT (`range_partitioning.slt`, TESTs 19–26, appended after the base
branch's join tests), each `EXPLAIN` paired with a result query where
its aggregate-section precedent has one:
- exact reuse: `PARTITION BY range_key` over `Range([range_key])` plans
with no repartition (window over a per-partition sort over the range
source)
- a non-streamable frame (`ROWS BETWEEN UNBOUNDED PRECEDING AND
UNBOUNDED FOLLOWING`) planning as `WindowAggExec` — covers the
non-bounded operator
- incompatible key: `PARTITION BY non_range_key` still hash repartitions
- subset satisfaction enabled: `PARTITION BY range_key, non_range_key`
avoids the hash repartition
- subset threshold not met: the same query hash repartitions
- `preserve_file_partitions` met: range preserved even when
`target_partitions` exceeds the input partition count
- subset threshold not met with higher `target_partitions`:
repartitions to increase parallelism
- no `PARTITION BY`: single-partition plan unchanged
- Plan-shape tests in `enforce_distribution.rs`: exact-positive
(no hash repartition over compatible range input) and negative
(incompatible key still rehashes), mirroring the aggregate tests from
#23239.
- Sanity checker tests: a window over a compatible range-partitioned
input passes `SanityCheckPlan`; an incompatible one is rejected. These
reuse the file-local `range_partitioned_exec` fixture the base branch
added for its join sanity test.
- Test infra: a `bounded_window_exec_with_can_repartition` builder
variant (the pre-existing builder hardcodes `can_repartition = false`,
under which window distribution tests pass vacuously).
- Full local run: fmt, `./dev/rust_lint.sh`, the extended workspace test
suite (`--features avro,json,backtrace,extended_tests,
recursive_protection,parquet_encryption`), and the `range_partitioning`
slt file all pass on this branch.
## Are there any user-facing changes?
No API changes. Plans over range-partitioned sources may now avoid a hash
repartition in front of window operators (a performance improvement); the
behavior is controlled by the existing
`datafusion.optimizer.subset_repartition_threshold`,
`datafusion.optimizer.preserve_file_partitions`, and
`datafusion.execution.target_partitions` settings.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
