[SPARK-37670][SQL] Support predicate pushdown and column pruning for de-duped CTEs #34929

maryannxue · 2021-12-17T05:18:06Z

What changes were proposed in this pull request?

This PR adds predicate push-down and column pruning to CTEs that are not inlined as well as fixes a few potential correctness issues:

Replace (previously not inlined) CTE refs with Repartition operations at the end of logical plan optimization so that WithCTE is not carried over to physical plan. As a result, we can simplify the logic of physical planning, as well as avoid a correctness issue where the logical link of a physical plan node can point to WithCTE and lead to unexpected behaviors in AQE, e.g., class cast exceptions in DPP.
Pull (not inlined) CTE defs from subqueries up to the main query level, in order to avoid creating copies of the same CTE def during predicate push-downs and other transformations.
Make CTE IDs more deterministic by starting from 0 for each query.

Why are the changes needed?

Improve de-duped CTEs' performance with predicate pushdown and column pruning; fixes de-duped CTEs' correctness issues.

Does this PR introduce any user-facing change?

No.

How was this patch tested?

Added UTs.

SparkQA · 2021-12-17T06:14:27Z

Kubernetes integration test starting
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/50788/

SparkQA · 2021-12-17T07:16:01Z

Kubernetes integration test status failure
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/50788/

SparkQA · 2021-12-17T07:36:40Z

Test build #146315 has finished for PR 34929 at commit 841aa2a.

This patch fails Spark unit tests.
This patch merges cleanly.
This patch adds no public classes.

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala

cloud-fan · 2022-01-06T15:44:56Z

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

        val (cteDef, refCount) = cteMap(ref.cteId)
-        val newRef = if (forceInline || shouldInline(cteDef, refCount)) {
+        if (shouldInline(cteDef, refCount)) {
          if (ref.outputSet == cteDef.outputSet) {


why do we compare output set, not output? is it possible that the ref and def have different output column order?

...ala/org/apache/spark/sql/catalyst/optimizer/PushdownPredicatesAndPruneColumnsForCTEDef.scala

...st/src/main/scala/org/apache/spark/sql/catalyst/optimizer/ReplaceCTERefWithRepartition.scala

peter-toth · 2022-04-12T19:11:43Z

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/CTESubstitution.scala

        traverseAndSubstituteCTE(relation, isCommand, cteDefs)._1
      }
+
+      if (cteDefs.length > lastCTEDefCount) {


Please consider #36146 as an alternative to substituting and changing accumulated cteDefs so far.

Hi @peter-toth , this PR contains all the CTE bug fixes we have found so far internally. Can you rebase #36146 after this one gets merged if you think your fix is cleaner? thanks!

I can rebase my #36146, no problem with that.

But I'm more concerned about #32298. This PR seems to contain a mix of improvements and bugfixes and changes a lot in CTE handling and conflicts with my PR. As mine is on the 3.3 whitelist do you think we can merge that first and rebase this on that?

This PR will be merged to 3.2 because of the fixes for the correctness bugs and performance regressions. It doesn't really improve the performance compared to Spark 3.1.

But can we merge it after #32298?

#32298 won't go to 3.2, right? If we really need a different CTE handling in master/3.3 for the merging scalar subqueries feature, we should still merge this PR first and make a followup PR to change CTE in master/3.3

Oh I though you just made a typo and want to merge it 3.3+ only and maybe backport the bugfix parts to 3.2...

peter-toth · 2022-04-12T19:17:51Z

@maryannxue do you think we can merge #32298 as it is targeted to Spark 3.3 and rebease this PR on the top of that?

cc @sigmod, @cloud-fan, @tgravescs

cloud-fan · 2022-04-18T18:19:24Z

The test build: https://github.com/maryannxue/spark/actions/runs/2184936574

cloud-fan · 2022-04-19T02:50:05Z

thanks, merging to master/3.3/3.2!

@peter-toth hopefully #32298 answers your question. I can work with you together to adjust your PR with the new CTE changes.

…de-duped CTEs This PR adds predicate push-down and column pruning to CTEs that are not inlined as well as fixes a few potential correctness issues: 1) Replace (previously not inlined) CTE refs with Repartition operations at the end of logical plan optimization so that WithCTE is not carried over to physical plan. As a result, we can simplify the logic of physical planning, as well as avoid a correctness issue where the logical link of a physical plan node can point to `WithCTE` and lead to unexpected behaviors in AQE, e.g., class cast exceptions in DPP. 2) Pull (not inlined) CTE defs from subqueries up to the main query level, in order to avoid creating copies of the same CTE def during predicate push-downs and other transformations. 3) Make CTE IDs more deterministic by starting from 0 for each query. Improve de-duped CTEs' performance with predicate pushdown and column pruning; fixes de-duped CTEs' correctness issues. No. Added UTs. Closes #34929 from maryannxue/cte-followup. Lead-authored-by: Maryann Xue <[email protected]> Co-authored-by: Wenchen Fan <[email protected]> Signed-off-by: Wenchen Fan <[email protected]> (cherry picked from commit 175e429) Signed-off-by: Wenchen Fan <[email protected]>

peter-toth · 2022-04-19T05:49:50Z

@peter-toth hopefully #32298 answers your question. I can work with you together to adjust your PR with the new CTE changes.

Thanks, I will try to rebase mine today.

…s an outer CTE ### What changes were proposed in this pull request? Please note that the bug in the [SPARK-38404](https://issues.apache.org/jira/browse/SPARK-38404) is fixed already with #34929. This PR is a minor improvement to the current implementation by collecting already resolved outer CTEs to avoid re-substituting already collected CTE definitions. ### Why are the changes needed? Small improvement + additional tests. ### Does this PR introduce _any_ user-facing change? No. ### How was this patch tested? Added new test case. Closes #36146 from peter-toth/SPARK-38404-nested-cte-references-outer-cte. Authored-by: Peter Toth <[email protected]> Signed-off-by: Wenchen Fan <[email protected]>

…terSuite ### What changes were proposed in this pull request? To remove unnecessary changes from `InjectRuntimeFilterSuite` after apache#32298. These are not needed after apache#34929 as the final optimized plan does'n contain any `WithCTE` nodes. ### Why are the changes needed? No need for those changes. ### Does this PR introduce _any_ user-facing change? No. ### How was this patch tested? Added new test. Closes apache#36361 from peter-toth/SPARK-34079-multi-column-scalar-subquery-follow-up-2. Authored-by: Peter Toth <[email protected]> Signed-off-by: Wenchen Fan <[email protected]>

…terSuite To remove unnecessary changes from `InjectRuntimeFilterSuite` after #32298. These are not needed after #34929 as the final optimized plan does'n contain any `WithCTE` nodes. No need for those changes. No. Added new test. Closes #36361 from peter-toth/SPARK-34079-multi-column-scalar-subquery-follow-up-2. Authored-by: Peter Toth <[email protected]> Signed-off-by: Wenchen Fan <[email protected]> (cherry picked from commit d05e01d) Signed-off-by: Wenchen Fan <[email protected]>

dongjoon-hyun

Hi, @maryannxue and @cloud-fan .

This seems to break branch-3.2 with two failures.

TPCDSV1_4_PlanStabilitySuite.check simplified (tpcds-v1.4/q4)
TPCDSV1_4_PlanStabilitySuite.check simplified (tpcds-v1.4/q5)

dongjoon-hyun · 2022-06-09T02:27:01Z

Here is a followup.

[SPARK-37670][FOLLOWUP][SQL][TESTS][3.2] Update TPCDS golden files #36815

dongjoon-hyun · 2022-06-09T07:41:29Z

sql/core/src/main/scala/org/apache/spark/sql/execution/QueryExecution.scala

  protected def planner = sparkSession.sessionState.planner

-  // The CTE map for the planner shared by the main query and all subqueries.
-  private val cteMap = mutable.HashMap.empty[Long, CTERelationDef]


This move (from QueryExecutor to the other location) could be the root cause of the UT failure in GitHub Action.

As I posted on #36815 , this PR seems to generate the golden answer files with the following correctly.

SPARK_GENERATE_GOLDEN_FILES=1 build/sbt "sql/testOnly *PlanStabilitySuite"

However, unfortunately, we are experiencing GitHub Action failures. In addition, the result is also different from the one when we run individual query.

SPARK_GENERATE_GOLDEN_FILES=1 build/sbt "sql/testOnly *PlanStabilitySuite -- -z (tpcds-v1.4/q4)"

Given that, this PR seems to introduce indeterministic logic in terms of expression IDs.

Could you take a look together with me, @maryannxue and @cloud-fan ?

If there is no easy solution, I'd like to recommend to revert this from branch-3.2 first because it has been broken for 2 months already. How do you want to proceed this?

Here is the result generation status.

$ SPARK_GENERATE_GOLDEN_FILES=1 build/sbt "sql/testOnly *PlanStabilitySuite" &> /dev/null $ git status On branch branch-3.2 Your branch is up to date with 'apache/branch-3.2'. nothing to commit, working tree clean $ SPARK_GENERATE_GOLDEN_FILES=1 build/sbt "sql/testOnly *PlanStabilitySuite -- -z (tpcds-v1.4/q4)" &> /dev/null $ SPARK_GENERATE_GOLDEN_FILES=1 build/sbt "sql/testOnly *PlanStabilitySuite -- -z (tpcds-v1.4/q5)" &> /dev/null $ git diff --stat sql/core/src/test/resources/tpcds-plan-stability/approved-plans-v1_4/q4/explain.txt | 238 ++++++++++++++++++++++++++++++++++++++++++++++++++++---------------------------------------------------- sql/core/src/test/resources/tpcds-plan-stability/approved-plans-v1_4/q5/explain.txt | 212 ++++++++++++++++++++++++++++++++++++++++++++++---------------------------------------------- 2 files changed, 225 insertions(+), 225 deletions(-)

It turns out to be a bug in the plan stability test suite.

The test suite normalizes the expr IDs, by using regex "#\\d+L?".r to match the explain string. However, The exchange node has a special string arg id=#...: https://github.com/apache/spark/blob/master/sql/core/src/main/scala/org/apache/spark/sql/execution/exchange/Exchange.scala#L40

The regex can't distinguish between expr ID and exchange plan id, and may normalize the plan wrongly.

I'll try to fix it tomorrow.

Great! Thank you for your investigation, @cloud-fan .

It sounds like a general bug on all branches. In that case, do you know why only branch-3.2 is so flaky like this?

if the plans are different between 3.2 and 3.3, then the plan ids are different and we may not trigger the bug.

dongjoon-hyun · 2022-06-09T08:22:16Z

As a backup, I created a reverting PR too. Since it's been two month already, we need to check if there is no other dependent PRs.

dongjoon-hyun · 2022-06-10T04:39:16Z

Any updates, @cloud-fan ? We also can revert this first and land it back after fixing the root cause inside PlanStabilitySuite.

cloud-fan · 2022-06-10T05:04:02Z

Just created: #36827

…s an outer CTE ### What changes were proposed in this pull request? Please note that the bug in the [SPARK-38404](https://issues.apache.org/jira/browse/SPARK-38404) is fixed already with apache#34929. This PR is a minor improvement to the current implementation by collecting already resolved outer CTEs to avoid re-substituting already collected CTE definitions. ### Why are the changes needed? Small improvement + additional tests. ### Does this PR introduce _any_ user-facing change? No. ### How was this patch tested? Added new test case. Closes apache#36146 from peter-toth/SPARK-38404-nested-cte-references-outer-cte. Authored-by: Peter Toth <[email protected]> Signed-off-by: Wenchen Fan <[email protected]>

…rences an outer CTE ### What changes were proposed in this pull request? Please note that the bug in the [SPARK-38404](https://issues.apache.org/jira/browse/SPARK-38404) is fixed already with #34929. This PR is a minor improvement to the current implementation by collecting already resolved outer CTEs to avoid re-substituting already collected CTE definitions. ### Why are the changes needed? Small improvement + additional tests. ### Does this PR introduce _any_ user-facing change? No. ### How was this patch tested? Added new test case. Closes #37760 from peter-toth/SPARK-38404-nested-cte-references-outer-cte-3.3. Authored-by: Peter Toth <[email protected]> Signed-off-by: Wenchen Fan <[email protected]>

…de-duped CTEs This PR adds predicate push-down and column pruning to CTEs that are not inlined as well as fixes a few potential correctness issues: 1) Replace (previously not inlined) CTE refs with Repartition operations at the end of logical plan optimization so that WithCTE is not carried over to physical plan. As a result, we can simplify the logic of physical planning, as well as avoid a correctness issue where the logical link of a physical plan node can point to `WithCTE` and lead to unexpected behaviors in AQE, e.g., class cast exceptions in DPP. 2) Pull (not inlined) CTE defs from subqueries up to the main query level, in order to avoid creating copies of the same CTE def during predicate push-downs and other transformations. 3) Make CTE IDs more deterministic by starting from 0 for each query. Improve de-duped CTEs' performance with predicate pushdown and column pruning; fixes de-duped CTEs' correctness issues. No. Added UTs. Closes apache#34929 from maryannxue/cte-followup. Lead-authored-by: Maryann Xue <[email protected]> Co-authored-by: Wenchen Fan <[email protected]> Signed-off-by: Wenchen Fan <[email protected]> (cherry picked from commit 175e429) Signed-off-by: Wenchen Fan <[email protected]> (cherry picked from commit 1a35685) Signed-off-by: Dongjoon Hyun <[email protected]>

maryannxue added 5 commits December 16, 2021 22:33

pred push down

386a427

fix id conflict

8642220

fix InlineCTE

645d3ac

add more tests

392d748

fix compilation

841aa2a

github-actions bot added the SQL label Dec 17, 2021