fix: make `UnKnownColumn`s not equal to others physical exprs by jonahgao · Pull Request #11536 · apache/datafusion

jonahgao · 2024-07-18T15:29:20Z

Which issue does this PR close?

Rationale for this change

~~After certain optimizations such as ProjectionPushdown, the children of InterleaveExec can no longer meet the interleave condition because their output_partitioning becomes inconsistent.~~
~~Perhaps in this situation, we could consider falling back to UnionExec from InterleaveExec instead of throwing an error.~~

Update

We determine that union inputs can interleave when they have the same hash partition spec.

datafusion/datafusion/physical-plan/src/union.rs

Line 499 in 12d82c4

pub fn can_interleave<T: Borrow<Arc<dyn ExecutionPlan>>>(

But this may be incorrect if the expressions in the partition spec contain UnKnownColumn.

When the partitioning expression no longer appears in the plan's output, for example, if the projection expressions in ProjectionExec no longer include it, this expression is converted into an UnknownColumn, using expr::to_string() as its name. This expression comes from the child of current plan. It is invalid for the current plan and shouldn't be used for evaluation or making decisions against the current plan.

datafusion/datafusion/physical-plan/src/projection.rs

Line 138 in 12d82c4

Arc::new(UnKnownColumn::new(&expr.to_string()))

Even if two UnknownColumns have the same name, their source expressions might not be the same.
For example, in issue #11409, the partition expressions of both inputs of InterleaveExec are UnKnownColumn { name: "v0@0" }, which corresponds to a Column expression named v0 with index 0. However, this Column expression references the output of the child plan of these inputs, and their corresponding source expressions are CAST(t1.v0 AS Float64) and t2.v0, respectively.

# InterleaveExec Input-0
ProjectionExec: expr=[v2@2 as v2, v0@1 as v0], schema=[v2:Int64;N, v0:Int64;N], O=Hash([UnKnownColumn { name: "v0@0" }], 2)
  ProjectionExec: expr=[v0@0 as v0, v0@1 as v0, v2@2 as v2], schema=[v0:Float64;N, v0:Int64;N, v2:Int64;N], O=Hash([Column { name: "v0", index: 0 }], 2)
    HashJoinExec: mode=Partitioned, join_type=Inner, on=[(v0@0, CAST(t1.v0 AS Float64)@2)], schema=[v0:Float64;N, v0:Int64;N, v2:Int64;N, CAST(t1.v0 AS Float64):Float64;N], O=Hash([Column { name: "CAST(t1.v0 AS Float64)", index: 3 }], 2)
      RepartitionExec: partitioning=Hash([v0@0], 2), input_partitions=1, schema=[v0:Float64;N], O=Hash([Column { name: "v0", index: 0 }], 2)
        MemoryExec: partitions=1, partition_sizes=[1], schema=[v0:Float64;N], O=UnknownPartitioning(1)
      RepartitionExec: partitioning=Hash([CAST(t1.v0 AS Float64)@2], 2), input_partitions=1, schema=[v0:Int64;N, v2:Int64;N, CAST(t1.v0 AS Float64):Float64;N], O=Hash([Column { name: "CAST(t1.v0 AS Float64)", index: 2 }], 2)
        ProjectionExec: expr=[v0@0 as v0, v2@1 as v2, CAST(v0@0 AS Float64) as CAST(t1.v0 AS Float64)], schema=[v0:Int64;N, v2:Int64;N, CAST(t1.v0 AS Float64):Float64;N], O=UnknownPartitioning(1)
          MemoryExec: partitions=1, partition_sizes=[4], schema=[v0:Int64;N, v2:Int64;N], O=UnknownPartitioning(1)

# InterleaveExec Input-1
ProjectionExec: expr=[v2@2 as v2, v0@1 as v0], schema=[v2:Int64;N, v0:Int64;N], O=Hash([UnKnownColumn { name: "v0@0" }], 2)
  ProjectionExec: expr=[v0@0 as v0, v0@1 as v0, v2@2 as v2], schema=[v0:Float64;N, v0:Int64;N, v2:Int64;N], O=Hash([Column { name: "v0", index: 0 }], 2)
    ProjectionExec: expr=[v0@3 as v0, v0@0 as v0, v2@1 as v2, CAST(t1.v0 AS Float64)@2 as CAST(t1.v0 AS Float64)], schema=[v0:Float64;N, v0:Int64;N, v2:Int64;N, CAST(t1.v0 AS Float64):Float64;N], O=Hash([Column { name: "v0", index: 0 }], 2)
      HashJoinExec: mode=Partitioned, join_type=Inner, on=[(CAST(t1.v0 AS Float64)@2, v0@0)], schema=[v0:Int64;N, v2:Int64;N, CAST(t1.v0 AS Float64):Float64;N, v0:Float64;N], O=Hash([Column { name: "v0", index: 3 }], 2)
        RepartitionExec: partitioning=Hash([CAST(t1.v0 AS Float64)@2], 2), input_partitions=2, schema=[v0:Int64;N, v2:Int64;N, CAST(t1.v0 AS Float64):Float64;N], O=Hash([Column { name: "CAST(t1.v0 AS Float64)", index: 2 }], 2)
          ProjectionExec: expr=[v0@0 as v0, v2@1 as v2, CAST(v0@0 AS Float64) as CAST(t1.v0 AS Float64)], schema=[v0:Int64;N, v2:Int64;N, CAST(t1.v0 AS Float64):Float64;N], O=RoundRobinBatch(2)
            RepartitionExec: partitioning=RoundRobinBatch(2), input_partitions=1, schema=[v0:Int64;N, v2:Int64;N], O=RoundRobinBatch(2)
              FilterExec: v2@1 IS NULL, schema=[v0:Int64;N, v2:Int64;N], O=UnknownPartitioning(1)
                MemoryExec: partitions=1, partition_sizes=[4], schema=[v0:Int64;N, v2:Int64;N], O=UnknownPartitioning(1)
        RepartitionExec: partitioning=Hash([v0@0], 2), input_partitions=1, schema=[v0:Float64;N], O=Hash([Column { name: "v0", index: 0 }], 2)
          MemoryExec: partitions=1, partition_sizes=[1], schema=[v0:Float64;N], O=UnknownPartitioning(1)

After ProjectionPushdown creates new ProjectionExec, it performs some degree of restoration on the UnKnownColumn, they become different and no longer meet the interleave condition.

# New InterleaveExec Input-0:
ProjectionExec: expr=[v2@1 as v2, v0@0 as v0], schema=[v2:Int64;N, v0:Int64;N], O=Hash([UnKnownColumn { name: "CAST(t1.v0 AS Float64)@3" }], 2)
  HashJoinExec: mode=Partitioned, join_type=Inner, on=[(v0@0, CAST(t1.v0 AS Float64)@2)], projection=[v0@1, v2@2], schema=[v0:Int64;N, v2:Int64;N], O=Hash([UnKnownColumn { name: "CAST(t1.v0 AS Float64)@3" }], 2)
    RepartitionExec: partitioning=Hash([v0@0], 2), input_partitions=1, schema=[v0:Float64;N], O=Hash([Column { name: "v0", index: 0 }], 2)
      MemoryExec: partitions=1, partition_sizes=[1], schema=[v0:Float64;N], O=UnknownPartitioning(1)
    RepartitionExec: partitioning=Hash([CAST(t1.v0 AS Float64)@2], 2), input_partitions=1, schema=[v0:Int64;N, v2:Int64;N, CAST(t1.v0 AS Float64):Float64;N], O=Hash([Column { name: "CAST(t1.v0 AS Float64)", index: 2 }], 2)
      ProjectionExec: expr=[v0@0 as v0, v2@1 as v2, CAST(v0@0 AS Float64) as CAST(t1.v0 AS Float64)], schema=[v0:Int64;N, v2:Int64;N, CAST(t1.v0 AS Float64):Float64;N], O=UnknownPartitioning(1)
        MemoryExec: partitions=1, partition_sizes=[4], schema=[v0:Int64;N, v2:Int64;N], O=UnknownPartitioning(1)

# New InterleaveExec Input-1:
ProjectionExec: expr=[v2@1 as v2, v0@0 as v0], schema=[v2:Int64;N, v0:Int64;N], O=Hash([UnKnownColumn { name: "v0@3" }], 2)
  HashJoinExec: mode=Partitioned, join_type=Inner, on=[(CAST(t1.v0 AS Float64)@2, v0@0)], projection=[v0@0, v2@1], schema=[v0:Int64;N, v2:Int64;N], O=Hash([UnKnownColumn { name: "v0@3" }], 2)
    RepartitionExec: partitioning=Hash([CAST(t1.v0 AS Float64)@2], 2), input_partitions=2, schema=[v0:Int64;N, v2:Int64;N, CAST(t1.v0 AS Float64):Float64;N], O=Hash([Column { name: "CAST(t1.v0 AS Float64)", index: 2 }], 2)
      ProjectionExec: expr=[v0@0 as v0, v2@1 as v2, CAST(v0@0 AS Float64) as CAST(t1.v0 AS Float64)], schema=[v0:Int64;N, v2:Int64;N, CAST(t1.v0 AS Float64):Float64;N], O=RoundRobinBatch(2)
        RepartitionExec: partitioning=RoundRobinBatch(2), input_partitions=1, schema=[v0:Int64;N, v2:Int64;N], O=RoundRobinBatch(2)
          FilterExec: v2@1 IS NULL, schema=[v0:Int64;N, v2:Int64;N], O=UnknownPartitioning(1)
            MemoryExec: partitions=1, partition_sizes=[4], schema=[v0:Int64;N, v2:Int64;N], O=UnknownPartitioning(1)
    RepartitionExec: partitioning=Hash([v0@0], 2), input_partitions=1, schema=[v0:Float64;N], O=Hash([Column { name: "v0", index: 0 }], 2)
      MemoryExec: partitions=1, partition_sizes=[1], schema=[v0:Float64;N], O=UnknownPartitioning(1)

What changes are included in this PR?

Are these changes tested?

Yes

Are there any user-facing changes?

No

alamb

I am approving this PR as it fixes a bug and a query that didn't run before will now run

However, I think the proposed fix might be treating the symptom rather than the root cause and I think more investigation is warranted. We could merge this PR and file a follow on ticket to investigate for example

Thank you very much @jonahgao

alamb · 2024-07-18T20:26:13Z

        children: Vec<Arc<dyn ExecutionPlan>>,
    ) -> Result<Arc<dyn ExecutionPlan>> {
-        Ok(Arc::new(InterleaveExec::try_new(children)?))
+        // New children may not be able to interleave; in that case, we fall back to UnionExec.


I see how this particular change is beneficial as now the plan runs where it didn't before, so that is hard to argue with and I would be ok merging it in

However, I feel like this might be silently masking some bug in the planner. My personal preference would be to make this an error

if !can_interleave(children.iter()) { return internal_err!("Can not create InterleaveExec: new children can not be interleaved"); }

And then track down / fix whatever optimizer pass is causing the children to no longer be interleavable

FYI @mustafasrepo

Applied. Thank you @alamb

I see how this particular change is beneficial as now the plan runs where it didn't before, so that is hard to argue with and I would be ok merging it in

However, I feel like this might be silently masking some bug in the planner. My personal preference would be to make this an error

if !can_interleave(children.iter()) { return internal_err!("Can not create InterleaveExec: new children can not be interleaved"); }

And then track down / fix whatever optimizer pass is causing the children to no longer be interleavable

FYI @mustafasrepo

This makes sense, invariant should be handled outside the operator during planning. This checks the whether assumptions met. Thanks for this.

jonahgao · 2024-07-19T08:01:15Z

However, I think the proposed fix might be treating the symptom rather than the root cause and I think more investigation is warranted. We could merge this PR and file a follow on ticket to investigate for example

@alamb I think the root clause should be UnKnownColumn. I changed to an alternative fix and updated the description in the PR.

alamb

Makes sense to me -- thank you @jonahgao and @mustafasrepo

alamb · 2024-07-19T20:04:42Z

-            .map(|x| self == x)
-            .unwrap_or(false)
+    fn eq(&self, _other: &dyn Any) -> bool {
+        // UnknownColumn is not a valid expression, so it should not be equal to any other expression.


jonahgao · 2024-07-20T00:50:33Z

Thank you @alamb for the review.

…#11536) * fix: fall back to `UnionExec` if can't interleave * alternative fix * check interleavable in with_new_children * link to pr

fix: fall back to UnionExec if can't interleave

ddb1f34

github-actions Bot added the sqllogictest SQL Logic Tests (.slt) label Jul 18, 2024

alamb approved these changes Jul 18, 2024

View reviewed changes

alternative fix

d243f43

github-actions Bot added the physical-expr Changes to the physical-expr crates label Jul 19, 2024

jonahgao changed the title ~~fix: fall back to UnionExec if can't interleave~~ fix: make UnKnownColumns not equal to each other Jul 19, 2024

jonahgao changed the title ~~fix: make UnKnownColumns not equal to each other~~ fix: make UnKnownColumns not equal to each other Jul 19, 2024

check interleavable in with_new_children

d4317b7

jonahgao marked this pull request as draft July 19, 2024 07:33

link to pr

d668b4f

jonahgao marked this pull request as ready for review July 19, 2024 08:01

jonahgao changed the title ~~fix: make UnKnownColumns not equal to each other~~ fix: make UnKnownColumns not equal to others physical exprs Jul 19, 2024

alamb approved these changes Jul 19, 2024

View reviewed changes

alamb merged commit ebe61ba into apache:main Jul 19, 2024

jonahgao deleted the union_fallback branch July 20, 2024 00:50

jonahgao mentioned this pull request Apr 27, 2026

fix: InterleaveExec fallback to UnionExec when children partitioning diverges #21827

Closed

Navigation Menu

Repository navigation

fix: make `UnKnownColumn`s not equal to others physical exprs - #11536

fix: make `UnKnownColumn`s not equal to others physical exprs#11536
alamb merged 4 commits into
apache:mainfrom
jonahgao:union_fallback

jonahgao commented Jul 18, 2024 •

edited

Loading

Uh oh!

alamb left a comment

Uh oh!

alamb Jul 18, 2024

Uh oh!

jonahgao Jul 19, 2024

Uh oh!

mustafasrepo Jul 19, 2024

Uh oh!

jonahgao commented Jul 19, 2024

Uh oh!

alamb left a comment

Uh oh!

alamb Jul 19, 2024

Uh oh!

jonahgao commented Jul 20, 2024

Uh oh!

Reviewers

Uh oh!

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

Conversation

jonahgao commented Jul 18, 2024 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Which issue does this PR close?

Rationale for this change

Update

What changes are included in this PR?

Are these changes tested?

Are there any user-facing changes?

Uh oh!

alamb left a comment

Choose a reason for hiding this comment

Uh oh!

alamb Jul 18, 2024

Choose a reason for hiding this comment

Uh oh!

jonahgao Jul 19, 2024

Choose a reason for hiding this comment

Uh oh!

mustafasrepo Jul 19, 2024

Choose a reason for hiding this comment

Uh oh!

jonahgao commented Jul 19, 2024

Uh oh!

alamb left a comment

Choose a reason for hiding this comment

Uh oh!

alamb Jul 19, 2024

Choose a reason for hiding this comment

Uh oh!

jonahgao commented Jul 20, 2024

Uh oh!

Reviewers

Uh oh!

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

jonahgao commented Jul 18, 2024 •

edited

Loading