You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@spark.apache.org by "juliuszsompolski (via GitHub)" <gi...@apache.org> on 2026/09/29 19:17:45 UTC
[PR] [SPARK-59871][SQL] Balance wide predicate trees in optimizer to avoid stack overflow [spark]
juliuszsompolski opened a new pull request, #59143:
URL: https://github.com/apache/spark/pull/59143
### What changes were proposed in this pull request?
Several optimizer rules combine a wide set of predicates into a single conjunction or disjunction with a linear `reduce(And)` / `reduceLeft(And)` / `reduceOption(And)` chain. That produces a left-deep `And`/`Or` tree of depth `O(n)` in the number of predicates, and a later recursive traversal of the tree (by another rule, or by a generic `transform`) then recurses `O(n)` deep.
This PR replaces those linear combinations with the balanced-tree builder `PredicateHelper.buildBalancedPredicate`, which produces a tree of depth `O(log n)`. Concretely:
- Adds `PredicateHelper.buildBalancedPredicateOption`, the `Option`-returning companion to the existing `buildBalancedPredicate` (returns `None` for an empty input, mirroring `reduceOption` / `reduceLeftOption`).
- Switches the wide-input predicate-combination sites in the following rules from `reduce*`/`reduceOption*` to `buildBalancedPredicate` / `buildBalancedPtersFromConstraints`, `CombineFilters`, `PruneFilters`, `PushPredicateThroughNonJoin`, `PushPredicateThroughJoin`, `PushDownLeftSemiAntiJoin`,`PushExtraPredicateThroughJoin`, `ExtractPythonUDFFromJoinCondition`, `ReorderJoin`, and `ExtractEquiJoinKeys`.
`buildBalancedPredicate` already exists and is used the same way in `BooleanSimplification` and `LikeSimplification`; this extends the same treatment to the remaining predicate-combination sites. Bounded-arity combinations (e.g. an expression's own children) and sites where the shape is irrelevant are left unchanged.
### Why are the changes needed?
A deeply nested `And`/`Or` tree built from a wide predicate set can cause a `StackOverflowError` during optimization when a downstream rule traverses it recursively. Building the tree balanced at construction time bounds the recursion depth to `O(log n)` and removes that failure mode at the source.
### Does this PR introduce _any_ user-facing change?
No. `And` and `Or` are associative and the left-to-right evaluation order of the leaf predicates is preserved, so the combined predicate is semantically identical (including short-circuiting). Only the internal shape of the `And`/`Or` tree changes: for two or three operands it is identical to the old linear tree, and for four or more it becomes balanced rather than left-deep.
### How was this patch tested?
Added `PredicateHelperSuite` covering `buildBalancedPredicate` (balanced shape) and `buildBalancedPredicateOption` (empty input and balanced shape). The ercised by the existing catalyst optimizer suites (constraint inference, filter pushdown, join reordering, equi-join key extraction).
### Was this patch authored or co-authored using generative AI tooling?
Yes - Claude Code Opus 5.5
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-59871][SQL] Balance wide predicate trees in optimizer to avoid stack overflow [spark]
Posted by "juliuszsompolski (via GitHub)" <gi...@apache.org>.
juliuszsompolski commented on PR #59143:
URL: https://github.com/apache/spark/pull/59143#issuecomment-5912887804
@HyukjinKwon thanks for the review and pointing to additional application spots. It seems I didn't go deep enough looking for them. My fix was investigating it from the angle of InferFiltersFromConstraints and then ConstantPropagation, because there are cases that can create really huge predicate trees there and I hit some stack overflows in my internal testing last week so decided to tackle it.
I now applied building balanced trees in more places, added a test and regenerated the q58 golden files.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-59871][SQL] Balance wide predicate trees in optimizer to avoid stack overflow [spark]
Posted by "dongjoon-hyun (via GitHub)" <gi...@apache.org>.
dongjoon-hyun commented on code in PR #59143:
URL: https://github.com/apache/spark/pull/59143#discussion_r4328857778
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/predicates.scala:
##########
@@ -185,6 +185,16 @@ trait PredicateHelper extends AliasHelper with Logging {
currentResult.head
}
+ /**
+ * Option-returning form of [[buildBalancedPredicate]]: `None` for an empty input, otherwise the
+ * combined predicate.
+ */
+ protected def buildBalancedPredicateOption(
+ expressions: Seq[Expression],
+ op: (Expression, Expression) => Expression): Option[Expression] = {
+ if (expressions.isEmpty) None else Some(buildBalancedPredicate(expressions, op))
Review Comment:
**The first pass of `buildBalancedPredicate` is quadratic on a `List`, and this wrapper passes its input through unchanged.** HyukjinKwon asked about this in review 5405036205, and the helper is unchanged at 774871c. At L171-183, `currentResult` starts as the caller's `Seq`, and the first pass calls `currentResult.size` in both loop conditions, in the array length and in `i + 1 == currentResult.size`, plus `currentResult(i)` and `currentResult(i + 1)`. On a `List`, each of these walks the list, so the first pass costs about 1.5 * n^2 node steps, where the replaced `reduceLeft` was O(n).
The `ExpressionSet` callers pass `List`s too. `ExpressionSet` does not override `toSeq`, so it is `immutable.Seq.from(this)`, which builds a `List`. That covers `newFilters.toSeq` and `newPredicates.toSeq` in `InferFiltersFromConstraints` (Optimizer.scala:2255 and :2306), the `CombineFilters` difference (Optimizer.scala:2407) and `joinConds.toSeq` (CostBasedJoinReorder.scala:308).
The cost is also paid repeatedly. The `Join` arm of `PushPredicateThroughJoin` rebuilds every join condition on every fixed-point iteration, and `ExtractEquiJoinKeys.unapply` (patterns.scala:223-224) builds both `otherPredicates` and `predicatesOfJoinKeys` on every match, although none of its 16 match sites in `sql/` reads the latter. Where master re-split a left-deep tree, its recursive split was already quadratic, so those paths only move the cost. But a `FullOuter` join condition from SQL, which the parser builds balanced and `PushPredicateThroughJoin` never rebuilds, now costs O(n^2) instead of O(n log n) on each `ExtractEquiJoinKeys` match. I traced this by reading the code and did not measure it.
Could you change L171 to `var currentResult: Seq[Expression] = expressions.toIndexedSeq`? It is a no-op for an `ArraySeq` or a `Vector`, keeps the same pairing and tree shape, and makes the helper O(n) for every caller.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-59871][SQL] Balance wide predicate trees in optimizer to avoid stack overflow [spark]
Posted by "HyukjinKwon (via GitHub)" <gi...@apache.org>.
HyukjinKwon commented on PR #59143:
URL: https://github.com/apache/spark/pull/59143#issuecomment-5978075467
Thanks for extending the balanced-tree construction to the subquery, decorrelation, cost-based join reorder, CTE-definition, and INTERSECT/EXCEPT sites, and for adding the rule-level FilterPushdownSuite depth assertion. The description now naming the sites left linear (bounded-operand folds and plan canonicalization) addresses my earlier question about the remaining linear sites. Since leaf order is preserved, short-circuiting and NULL semantics are unchanged. This resolves my earlier comments.
<!-- SPARK_DEV_REVIEW_REPLY {"feedback_id":"conversation:5912887804","thread_id":"conversation:5912887804","verdict_sha256":"00c82329ca7a864c988dfabc966ab5df8cf7a943ac28e29b0c9fb14ce83ac1a3"} -->
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-59871][SQL] Balance wide predicate trees in optimizer to avoid stack overflow [spark]
Posted by "dongjoon-hyun (via GitHub)" <gi...@apache.org>.
dongjoon-hyun commented on code in PR #59143:
URL: https://github.com/apache/spark/pull/59143#discussion_r4328857799
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/Optimizer.scala:
##########
@@ -2252,7 +2252,7 @@ object InferFiltersFromConstraints extends Rule[LogicalPlan]
val newFilters = filter.constraints --
(child.constraints ++ splitConjunctivePredicates(condition))
if (newFilters.nonEmpty) {
- Filter(And(newFilters.reduce(And), condition), child)
+ Filter(And(buildBalancedPredicate(newFilters.toSeq, And), condition), child)
Review Comment:
**Filters inferred here on a file scan are folded left-deep again later in the same optimizer run.** They usually end up in the `Filter` right above a scan. The "Early Filter and Projection Push-Down" batch (L252), which runs after `operatorOptimizationBatch` (L244), splits that `Filter` and folds all of its conjuncts back into a left-deep `And`: `filtersPushDown.reduceOption(And)` in SchemaPruning.scala:45, `postScanFilters.reduceLeftOption(And)` in V2ScanRelationPushDown.scala:163 (and :1001), `filters.reduceLeft(And)` in PruneFileSourcePartitions.scala:42, and the same pattern in PruneHiveTablePartitions.scala:117 and PushVariantIntoScan.scala:424.
For the default V1 Parquet source, `SchemaPruning` does this for every `HadoopFsRelation` scan with filters and walks the chain recursively in `normalizeAttributeRefNames` (SchemaPruning.scala:69 and :163) and `identifyRootFields` (:70). The `canPruneDataSchema` check, which reads `nestedSchemaPruningEnabled`, only comes after that (:76), and `.getOrElse(op)` (:51) means the walk happens even when nothing is pruned. So a wide `WHERE`, or the large inferred constraint sets that motivated this PR, on a Parquet table still produces an n-deep tree that is walked inside `Optimizer.execute`, and a DSv2 scan keeps a left-deep `Filter` in the optimized plan. I traced this by reading the code and did not run it.
Could you convert these early scan push-down folds too (`SchemaPruning` and `PruneFileSourcePartitions` would need `with PredicateHelper`), or, if they are out of scope, say so in the description and file a follow-up JIRA?
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/PushdownPredicatesAndPruneColumnsForCTEDef.scala:
##########
@@ -147,10 +147,10 @@ object PushdownPredicatesAndPruneColumnsForCTEDef extends Rule[LogicalPlan] with
// used as-is: re-pushing the combined predicate is redundant but always
// semantics-preserving, since the disjunction of the reference predicates is
// valid for every row of the CTE definition.
- removePushedDownFilter(child, prevPreds.reduce(Or))
+ removePushedDownFilter(child, buildBalancedPredicate(prevPreds, Or))
case _ => child
}
- val newCombinedPred = newPreds.reduce(Or)
+ val newCombinedPred = buildBalancedPredicate(newPreds, Or)
Review Comment:
**`BooleanSimplification`'s common-factor branches still fold unbounded lists left-deep.** The description says `buildBalancedPredicate` "is used the same way in `BooleanSimplification`", but only its de-duplication branches use it (expressions.scala:500 and :538). The common-factor branches still use `reduce`: `common.reduce(Or)` and `(common :+ And(ldiff.reduce(Or), rdiff.reduce(Or))).reduce(Or)` at expressions.scala:485 and :489, and the `And` mirror at :523 and :527. They run in the same fixed-point batch as the rules converted here.
For example, `WHERE (k > 0 AND a1 > 0 AND ... AND a3000 > 0) OR (k > 0 AND b1 > 0 AND ... AND b3000 > 0)` is parsed balanced, but expressions.scala:527 turns it into `And(k > 0, Or(<3000-deep chain>, <3000-deep chain>))`. The converted rules only rebuild top-level conjunct lists, so the nested chains survive into the optimized plan. The `Or` built here can take the same path when the references of a non-inlined CTE share conjuncts. I traced this by reading the code and did not run it.
Could you build these four folds with `buildBalancedPredicate` as well, as expressions.scala:500 and :538 already do?
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/Optimizer.scala:
##########
@@ -2303,7 +2303,7 @@ object InferFiltersFromConstraints extends Rule[LogicalPlan]
if (newPredicates.isEmpty) {
plan
} else {
- Filter(newPredicates.reduce(And), plan)
+ Filter(buildBalancedPredicate(newPredicates.toSeq, And), plan)
Review Comment:
Minor: `inferNewFilter` balances the conjunction here, but one inferred constraint can itself be an `Or` as deep as the number of union branches. `Union.validConstraints` (basicLogicalOperators.scala:577-581) merges the children's constraints with `.reduce(merge(_, _))`, and `merge` builds `Or(othera(attr).reduceLeft(And), otherb(attr).reduceLeft(And))` (:572), so each branch adds one level. For `(SELECT k FROM t WHERE k = 1 UNION ALL ... UNION ALL SELECT k FROM t WHERE k = 3000) u JOIN d ON u.k = d.k`, the union's constraint is a 3000-deep left-deep `Or`. `inferAdditionalConstraints` rewrites it with a recursive `transform` (QueryPlanConstraints.scala:134), and if that survives, this line installs the rewritten `Or` on `d` as a single leaf, which the balanced `And` cannot flatten. I traced this by reading the code and did not run it. Could you balance the folds in `Union.merge` as well, or list it as a known remaining site?
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/Optimizer.scala:
##########
@@ -3401,13 +3414,14 @@ object ReplaceIntersectWithSemiJoin extends Rule[LogicalPlan] {
* 2. This rule has to be done after de-duplicating the attributes; otherwise, the generated
* join conditions will be incorrect.
*/
-object ReplaceExceptWithAntiJoin extends Rule[LogicalPlan] {
+object ReplaceExceptWithAntiJoin extends Rule[LogicalPlan] with PredicateHelper {
def apply(plan: LogicalPlan): LogicalPlan = plan.transformWithPruning(
_.containsPattern(EXCEPT), ruleId) {
case Except(left, right, false) =>
assert(left.output.size == right.output.size)
val joinCond = left.output.zip(right.output).map { case (l, r) => EqualNullSafe(l, r) }
- Distinct(Join(left, right, LeftAnti, joinCond.reduceLeftOption(And), JoinHint.NONE))
+ Distinct(
+ Join(left, right, LeftAnti, buildBalancedPredicateOption(joinCond, And), JoinHint.NONE))
Review Comment:
Minor: This balances the one-`EqualNullSafe`-per-column condition of `EXCEPT`, but its analyzer counterpart is unchanged. NaturalAndUsingJoinResolution.scala:73 builds `(condition ++ joinPairs.map(EqualTo.tupled)).reduceOption(And)`, one `EqualTo` per USING column, for `JOIN ... USING`, `NATURAL JOIN` and `df1.join(df2, usingColumns, joinType)`. So `df1.join(df2, df1.columns.toSeq, "left_anti")` on a wide schema, a common way to diff two DataFrames, gets an N-deep left-deep condition from analysis on. `CheckAnalysis` walks it, and it is rebalanced only when `PushPredicateThroughJoin` first reaches the join. For `FULL OUTER JOIN ... USING`, `canPushThrough` excludes `FullOuter`, so the condition stays left-deep through the whole optimizer. Could you convert it too (the object would need `with PredicateHelper`), or mention it as a follow-up?
##########
sql/core/src/test/resources/tpcds-plan-stability/approved-plans-v1_4/q58/explain.txt:
##########
@@ -199,7 +199,7 @@ Arguments: HashedRelationBroadcastMode(List(input[0, string, true]),false), [pla
Left keys [1]: [item_id#11]
Right keys [1]: [item_id#22]
Join type: Inner
-Join condition: ((((cast(ss_item_rev#12 as decimal(19,3)) >= (0.9 * cs_item_rev#23)) AND (cast(ss_item_rev#12 as decimal(20,3)) <= (1.1 * cs_item_rev#23))) AND (cast(cs_item_rev#23 as decimal(19,3)) >= (0.9 * ss_item_rev#12))) AND (cast(cs_item_rev#23 as decimal(20,3)) <= (1.1 * ss_item_rev#12)))
+Join condition: (((cast(ss_item_rev#12 as decimal(19,3)) >= (0.9 * cs_item_rev#23)) AND (cast(ss_item_rev#12 as decimal(20,3)) <= (1.1 * cs_item_rev#23))) AND ((cast(cs_item_rev#23 as decimal(19,3)) >= (0.9 * ss_item_rev#12)) AND (cast(cs_item_rev#23 as decimal(20,3)) <= (1.1 * ss_item_rev#12))))
Review Comment:
Minor: Only join conditions planned through `ExtractEquiJoinKeys` change shape in these goldens. `Filter` conditions over scans are still left-deep, e.g. `approved-plans-v1_4/q19/explain.txt` L55, `Condition : ((((isnotnull(d_moy#3) AND isnotnull(d_year#2)) AND (d_moy#3 = 11)) AND (d_year#2 = 1998)) AND isnotnull(d_date_sk#1))`, and q64/explain.txt L656. Physical planning folds every after-scan filter again: `afterScanFilters.toSeq.reduceOption(expressions.And)` in FileSourceStrategy.scala:362, and the same pattern in DataSourceStrategy.scala:489, DataSourceV2Strategy.scala:1041 and SparkPlanner.scala:100. So for `spark.read.parquet(path).where(<wide AND>)`, the condition the optimizer now keeps balanced becomes n-deep again in `FilterExec`, whose constructor splits it recursively (basicPhysicalOperators.scala:250), and AQE canonicalization and the explain string walk it too. The `StackOverflowError` would move from optimization to planning. I traced this by reading the code and d
id not run it. Could you convert these planner sites, or state in the description that the PR covers the optimizer only and track them in a follow-up?
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/predicates.scala:
##########
@@ -185,6 +185,16 @@ trait PredicateHelper extends AliasHelper with Logging {
currentResult.head
}
+ /**
+ * Option-returning form of [[buildBalancedPredicate]]: `None` for an empty input, otherwise the
Review Comment:
Minor: The description keeps the plan-canonicalization `reduce` sites "to preserve their canonical form", but that does not apply to `QueryPlan.normalizePredicates` (QueryPlan.scala:871). There `predicates.reduce(And)` is only an intermediate value: `normalizeExpressions` ends with `.canonicalized`, which gathers the operands, sorts them by hash code and builds a flat `MultiCommutativeOp` (for 3 or more operands by default), and the result is split again. Both sides of a comparison go through this same function, so any deterministic fold gives a valid canonical form. Meanwhile, the left-deep fold goes through the recursive `transformUp` in `normalizeExpressions` (QueryPlan.scala:839) each time a `FileSourceScanExec` is canonicalized, e.g. for exchange reuse or AQE, since its `doCanonicalize` passes all `dataFilters` (DataSourceScanExec.scala:971). Since `object QueryPlan extends PredicateHelper`, could you use `buildBalancedPredicate(predicates, And)` there?
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/subquery.scala:
##########
@@ -999,7 +1000,7 @@ object RewriteCorrelatedScalarSubquery extends Rule[LogicalPlan] with AliasHelpe
currentChild.output :+ caseExpr,
Join(currentChild,
Project(subqueryRoot.output :+ alwaysTrueExpr, subqueryRoot),
- joinType, conditions.reduceOption(And), joinHint))
+ joinType, buildBalancedPredicateOption(conditions, And), joinHint))
Review Comment:
Minor: The description says `RewriteLateralSubquery` is left as `reduce` because of its "at-most-two-way condition", but `(condition ++ joinCond).reduceOption(And)` at L1096 folds `joinCond`, which has one entry per correlated predicate. `PullupCorrelatedPredicates` fills it from `DecorrelateInnerQuery` (L664-666, and DecorrelateInnerQuery.scala:629 and :643), the same kind of list as the `conditions` converted here in `constructLeftJoins`. For example, `SELECT * FROM t1 JOIN LATERAL (SELECT * FROM t2 WHERE t2.k1 = t1.k1 AND ... AND t2.kN = t1.kN) ON t1.x > 0` gets an N-deep chain. The impact is small, since `PushPredicateThroughJoin` rebalances lateral join conditions (Inner, LeftOuter or Cross) on the next iteration. Could you either convert L1096 (the rule would need `with PredicateHelper`) or correct the description?
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/joins.scala:
##########
@@ -270,11 +270,11 @@ object ExtractPythonUDFFromJoinCondition extends Rule[LogicalPlan] with Predicat
log" it will be moved out and the join plan will be turned to cross join.")
None
} else {
- Some(rest.reduceLeft(And))
+ Some(buildBalancedPredicate(rest, And))
Review Comment:
Nit: `PlanExternalUDFs` documents that its "Join-condition handling mirrors `ExtractPythonUDFFromJoinCondition`" (PlanExternalUDFs.scala:35) and already mixes in `PredicateHelper`, but its copies of L273 and L277 still use `otherConditions.reduceLeft(And)` and `udfConditions.reduceLeft(And)` (PlanExternalUDFs.scala:90 and :92). Both rules run in the same "Extract UDFs" batch, so the two now produce different shapes. The external UDF path is behind the experimental `spark.sql.execution.udf.unified.execution.enabled`, so the impact is low. Could you convert those two lines as well, to keep the mirror in sync?
##########
sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/optimizer/FilterPushdownSuite.scala:
##########
@@ -2082,4 +2082,21 @@ class FilterPushdownSuite extends PlanTest {
val optimizedC = Optimize.execute(queryC)
comparePlans(optimizedC, queryC, checkAnalysis = false)
}
+
+ test("SPARK-59871: predicates pushed through a join form a balanced tree") {
Review Comment:
Nit: This is about the PR description, not this line. Its generative AI section says "Yes - Claude Code Opus 5.5", while the template asks to "include the phrase: 'Generated-by: ' followed by the name of the tool and its version". `dev/merge_spark_pr.py` copies the description into the commit message, so the merged commit would have no `Generated-by:` line. Your other recent commits use it, e.g. c79c1a3f8b9 (`Generated-by: Claude Code (claude-opus-5-5)`). Could you change it to `Generated-by: Claude Code (Opus 5.5)`?
##########
sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/expressions/PredicateHelperSuite.scala:
##########
@@ -0,0 +1,45 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.catalyst.expressions
+
+import org.apache.spark.SparkFunSuite
+import org.apache.spark.sql.types.BooleanType
+
+class PredicateHelperSuite extends SparkFunSuite with PredicateHelper {
+
+ test("buildBalancedPredicate builds a balanced (log-depth) tree") {
+ val a = AttributeReference("a", BooleanType)()
+ val b = AttributeReference("b", BooleanType)()
+ val c = AttributeReference("c", BooleanType)()
+ val d = AttributeReference("d", BooleanType)()
+ assert(buildBalancedPredicate(Seq(a), And) == a)
+ assert(buildBalancedPredicate(Seq(a, b), And) == And(a, b))
+ assert(buildBalancedPredicate(Seq(a, b, c, d), And) == And(And(a, b), And(c, d)))
+ assert(buildBalancedPredicate(Seq(a, b, c, d), Or) == Or(Or(a, b), Or(c, d)))
+ }
+
+ test("buildBalancedPredicateOption returns None for empty input, balanced tree otherwise") {
+ val a = AttributeReference("a", BooleanType)()
Review Comment:
Nit: These four attributes repeat L26-29. Could you make them class-level `private val`s, as `ExtractPredicatesWithinOutputSetSuite` does? An n = 3 assertion (`And(And(a, b), c)`) would also pin the description's statement that two or three operands keep the old left-deep shape.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/predicates.scala:
##########
@@ -185,6 +185,16 @@ trait PredicateHelper extends AliasHelper with Logging {
currentResult.head
}
+ /**
+ * Option-returning form of [[buildBalancedPredicate]]: `None` for an empty input, otherwise the
+ * combined predicate.
+ */
+ protected def buildBalancedPredicateOption(
Review Comment:
Minor: The remaining sites in my other comments show how easy it is to miss one, and nothing prevents new linear folds. At 774871c, a search for `reduce`, `reduceLeft`, `reduceOption` or `reduceLeftOption` over `And`/`Or` still finds 55 lines of main code (25 in catalyst, 27 in core, 3 in hive). Master has also gained one since this branch's base: `boundExpressions.reduce(And)` in `ParquetStorageFilter.create` (SPARK-59620), which is harmless because its input is bounded. Also, this helper is a `protected` member of a trait, so `DataSourceV2Strategy`, `SparkPlanner`, `SchemaPruning`, `PruneFileSourcePartitions` and `NaturalAndUsingJoinResolution` would have to change their supertypes to call it. As a follow-up, could you consider a public helper that these trait methods delegate to (catalyst is excluded from MiMa), plus a scalastyle `RegexChecker` rule against linear `And`/`Or` folds, with `// scalastyle:off` at the deliberate sites?
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/subquery.scala:
##########
@@ -444,15 +444,15 @@ object RewritePredicateSubquery extends Rule[LogicalPlan] with PredicateHelper {
// Deduplicate conflicting attributes if any.
val newSub = dedupSubqueryOnSelfJoin(newPlan, sub, Some(values))
val inConditions = values.zip(newSub.output).map(EqualTo.tupled)
- val newConditions = (inConditions ++ conditions).reduceLeftOption(And)
+ val newConditions = buildBalancedPredicateOption(inConditions ++ conditions, And)
val joinHint = JoinHint(None, subHint)
newPlan = Join(newPlan, rewriteDomainJoinsIfPresent(newPlan, newSub, newConditions),
ExistenceJoin(exists), newConditions, joinHint)
introducedAttrs += exists
exists
}
}
- (newExprs.reduceOption(And), newPlan, introducedAttrs.toSeq)
+ (buildBalancedPredicateOption(newExprs.toSeq, And), newPlan, introducedAttrs.toSeq)
Review Comment:
Nit: `newExprs` is already an immutable `Seq` (`exprs.map { ... }` at L406), so `.toSeq` is a no-op here, unlike the `ExpressionSet` conversions at Optimizer.scala:2255, :2306 and :2407 and CostBasedJoinReorder.scala:308. Could you drop it, i.e. `buildBalancedPredicateOption(newExprs, And)`?
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-59871][SQL] Balance wide predicate trees in optimizer to avoid stack overflow [spark]
Posted by "juliuszsompolski (via GitHub)" <gi...@apache.org>.
juliuszsompolski commented on PR #59143:
URL: https://github.com/apache/spark/pull/59143#issuecomment-6002649783
Not flaked CI at last 🎉
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org