jayzhan211 commented on code in PR #26747:
URL: https://github.com/apache/datafusion/pull/26747#discussion_r4373486186
##########
datafusion/physical-plan/src/repartition/mod.rs:
##########
@@ -2200,15 +2200,17 @@ impl RepartitionExec {
input: &Arc<dyn ExecutionPlan>,
preserve_order: bool,
) -> EquivalenceProperties {
- // Equivalence Properties
let mut eq_properties = input.equivalence_properties().clone();
- // If the ordering is lost, reset the ordering equivalence class:
+ let input_partitions = input.output_partitioning().partition_count();
Review Comment:
Optional: both branches only matter when there is more than one input
partition, so this could be one block, and it would keep a comment like the one
in `SortPreservingMergeExec`:
```rs
// Fusing several input partitions keeps only the merge key ordering,
// and per-partition constants no longer hold.
if input.output_partitioning().partition_count() > 1 {
eq_properties.clear_orderings();
if preserve_order && let Some(ordering) = input.output_ordering() {
eq_properties.add_ordering(ordering.clone());
}
eq_properties.clear_per_partition_constants();
}
```
##########
datafusion/physical-expr/src/equivalence/properties/mod.rs:
##########
@@ -1402,6 +1412,28 @@ impl EquivalenceProperties {
/// Returns an [`ExprProperties`] object containing the ordering and range
/// information for the given expression.
pub fn get_expr_properties(&self, expr: Arc<dyn PhysicalExpr>) ->
ExprProperties {
Review Comment:
Optional: could the doc comment mention that this now falls back to the sort
properties of an equivalent expression? A short comment on the new block in
`class.rs` saying why the equality is recorded would also help: it keeps
orderings derived through strict functions after a merge resets the orderings.
##########
datafusion/physical-expr/src/equivalence/class.rs:
##########
@@ -752,6 +753,13 @@ impl EquivalenceGroup {
for (target, _) in targets.iter() {
cls.push(Arc::clone(target));
}
+ if source.is::<CastExpr>() {
Review Comment:
Limiting this to `CastExpr` brings back the original planning failure for
other strictly order-preserving functions. If I swap the cast in the unbounded
test for `to_unixtime(a)` or `from_unixtime(CAST(a AS BIGINT))`, main plans it
with a sorted window, but this branch fails with `SanityCheckPlan ... Cannot
execute pipeline breaking queries`:
```sql
EXPLAIN SELECT a, x, SUM(c) OVER (PARTITION BY x ORDER BY b) s
FROM (SELECT to_unixtime(a) x, a, b, c FROM t_inf);
```
The equality holds for any deterministic expression, so recording it for
every non-volatile source fixes these. The full sqllogictest suite passes with
no expected-plan changes:
```diff
- if source.is::<CastExpr>() {
+ if !source.children().is_empty() && !is_volatile(source) {
```
Could you add one of these queries next to the existing unbounded test?
##########
datafusion/physical-expr/src/equivalence/class.rs:
##########
@@ -752,6 +753,13 @@ impl EquivalenceGroup {
for (target, _) in targets.iter() {
cls.push(Arc::clone(target));
}
+ if source.is::<CastExpr>() {
Review Comment:
This records `x = CAST(r)` even when the source is volatile, and then the
planner picks the wrong rows. With `random()` the two calls are independent
draws:
```sql
SET datafusion.execution.target_partitions = 1;
CREATE TABLE g AS SELECT value AS v FROM generate_series(1, 2000);
SELECT r, x FROM (
SELECT random() AS r, CAST(random() AS FLOAT) AS x FROM g ORDER BY r LIMIT
1000
) ORDER BY x;
```
On main the plan is `Sort[x]` over `TopK[r]`. On this branch it becomes a
single `TopK(fetch=1000), expr=[x]`, so it returns the 1000 smallest `x`
instead of the 1000 smallest `r` (max `r` in the output is 1.00 instead of
about 0.50). Skipping volatile sources fixes it locally:
```diff
- if source.is::<CastExpr>() {
+ if source.is::<CastExpr>() && !is_volatile(source) {
```
with `use datafusion_physical_expr_common::physical_expr::is_volatile;`.
--
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]
