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: [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]
