[SPARK-46487][SQL] Push down part of filter through aggregate with nondeterministic field by zml1206 · Pull Request #44460 · apache/spark

zml1206 · 2023-12-22T09:15:55Z

What changes were proposed in this pull request?

Push down part of filter which is deterministic and references are subset of aggregate's child through aggregate with nondeterministic field.
For example

testRelation
      .groupBy($"a")($"a", Rand(10).as("rand"))
      .where($"a" > 5 && $"rand" > 5)

We can push down $"a" > 5 and do not push down $"rand" > 5. Because $"rand" > 5 is non-deterministic, push down it will change the evaluation result in aggregate.

Why are the changes needed?

Filter earlier to improve performance.

Does this PR introduce any user-facing change?

No.

How was this patch tested?

UT.

Was this patch authored or co-authored using generative AI tooling?

No.

beliefer · 2023-12-25T02:04:48Z

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/Optimizer.scala

      project.copy(child = Filter(replaceAlias(condition, aliasMap), grandChild))

    case filter @ Filter(condition, aggregate: Aggregate)
-      if aggregate.aggregateExpressions.forall(_.deterministic)


Do you investigate the issue https://issues.apache.org/jira/browse/SPARK-13473 ?
It seems we can't release the restrictions.

I investigated before,
Similar ones are https://issues.apache.org/jira/browse/SPARK-20246.
At that time, it was because it was impossible to determine whether the filter expression was deterministic for example $"rand" > 5, therefore, it will incorrectly push down $"rand" > 5.
However, after the filter expression is replaced, it can be judged whether it is a deterministic expression.

If so, how about keep
if aggregate.aggregateExpressions.exists(_.deterministic)

beliefer

LGTM cc @cloud-fan

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/Optimizer.scala

cloud-fan · 2024-01-31T03:03:22Z

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/Optimizer.scala

    // implies that, for a given input row, the output are determined by the expression's initial
    // state and all the input rows processed before. In another word, the order of input rows
    // matters for non-deterministic expressions, while pushing down predicates changes the order.
-    // This also applies to Aggregate.


what's the rationale for not applying this to Aggregate?

When "Aggregate" contains both non-deterministic and deterministic expression fields, it can push down some deterministic filters after replaced. Because if the condition after replaced is deterministic, it certainly has no association with the non-deterministic expression field.

For Aggregate, only filter which reference to deterministic groupExpression is pushed down, so it will only filter the group. I can't think of what kind of Non-deterministic expressions will be affected by this push down.

cloud-fan · 2024-01-31T06:28:22Z

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/Optimizer.scala

+    // Push `Filter` operators through `Aggregate` operators. Parts of the predicates that can
+    // be beneath must satisfy the following conditions:
+    // 1. Grouping expressions are not empty.
+    // 2. Predicate expression is deterministic.


do you mean de-aliased predicate?

Yes, I will change the description to make it clearer.

cloud-fan · 2024-01-31T06:28:52Z

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/Optimizer.scala

      val (pushDown, stayUp) = splitConjunctivePredicates(condition).partition { cond =>
        val replaced = replaceAlias(cond, aliasMap)
-        cond.deterministic && !cond.throwable &&
+        replaced.deterministic && !cond.throwable &&


shall we also use replaced.throwable?

I think can push down throwable filter through aggregate, it seems does not affect exception thrown
. What do you think? @cloud-fan

I will make a PR later.

cloud-fan · 2024-01-31T13:56:31Z

I checked the aggregate execution implementation. Basically, it has an embedded Project operator and applies the result projection for aggregated values. That said, I think it has the same property as Project and we shouldn't relax the requirement.

On the other hand, the aggregate operator itself is already non-deterministic, as the output row order is different between hash and sort aggregate. But we shouldn't make it worse.

zml1206 · 2024-01-31T15:17:12Z

I checked the aggregate execution implementation. Basically, it has an embedded Project operator and applies the result projection for aggregated values. That said, I think it has the same property as Project and we shouldn't relax the requirement.

On the other hand, the aggregate operator itself is already non-deterministic, as the output row order is different between hash and sort aggregate. But we shouldn't make it worse.

I understand, thank you, for example

Seq((1, 1), (1, 2), (2, 1), (3, 1)).toDF("a", "b").createOrReplaceTempView("t1")
sql("select a,monotonically_increasing_id() as c from t1 group by a").filter("a = 2").show()

Push down will make the results wrong, correct: (2, 2), error(2, 0).

github-actions bot added the SQL label Dec 22, 2023

zml1206 force-pushed the limit_pushdown branch from 92bf43c to 80697a8 Compare December 22, 2023 10:56

zml1206 closed this Dec 25, 2023

zml1206 deleted the limit_pushdown branch December 25, 2023 00:39

zml1206 restored the limit_pushdown branch December 25, 2023 00:39

zml1206 reopened this Dec 25, 2023

Push down part of filter through aggregate with nondeterministic field

f122265

zml1206 force-pushed the limit_pushdown branch from 80697a8 to f122265 Compare December 25, 2023 00:46

beliefer reviewed Dec 25, 2023

View reviewed changes

update

0a7ec21

beliefer approved these changes Dec 25, 2023

View reviewed changes

Update comments

16d33f6

HyukjinKwon reviewed Dec 26, 2023

View reviewed changes

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/Optimizer.scala Outdated Show resolved Hide resolved

zml1206 added 2 commits December 26, 2023 10:09

Update comments

35282ee

Merge branch 'master' into limit_pushdown

b08a3ab

cloud-fan reviewed Jan 31, 2024

View reviewed changes

update

43f1596

zml1206 closed this Jan 31, 2024

Conversation

zml1206 commented Dec 22, 2023 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

What changes were proposed in this pull request?

Why are the changes needed?

Does this PR introduce any user-facing change?

How was this patch tested?

Was this patch authored or co-authored using generative AI tooling?

Uh oh!

beliefer Dec 25, 2023 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

zml1206 Dec 25, 2023 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

beliefer left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

zml1206 Jan 31, 2024 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

zml1206 Jan 31, 2024 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

cloud-fan commented Jan 31, 2024

Uh oh!

zml1206 commented Jan 31, 2024

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

4 participants

zml1206 commented Dec 22, 2023 •

edited

Loading

beliefer Dec 25, 2023 •

edited

Loading

zml1206 Dec 25, 2023 •

edited

Loading

zml1206 Jan 31, 2024 •

edited

Loading

zml1206 Jan 31, 2024 •

edited

Loading