cloud-fan commented on code in PR #59258:
URL: https://github.com/apache/spark/pull/59258#discussion_r4237673518
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/QueryPlanConstraints.scala:
##########
@@ -126,6 +125,92 @@ trait ConstraintHelper {
inferred -- constraints
}
+ /**
+ * Infers the full additional constraint set for a given set of constraints
by combining the
+ * equality-based inferences above ([[inferAdditionalConstraints]],
+ * [[inferConstraintsFromLiteralBindings]]) with transitive inequality
chaining
+ * ([[inferInequalityConstraints]]). The inequality step is iterated to a
fixed point because a
+ * single pass only chains pairs present in the input set: e.g. `a < b && b
< c && c < 5`
+ * requires one pass to derive `b < 5` and a second to derive `a < 5` from
it.
+ */
+ def inferAllAdditionalConstraints(constraints: ExpressionSet): ExpressionSet
= {
+ @tailrec
+ def chainInequalities(inferred: ExpressionSet): ExpressionSet = {
+ val newlyInferred = inferInequalityConstraints(constraints ++ inferred)
+ if (newlyInferred.isEmpty) inferred else chainInequalities(inferred ++
newlyInferred)
+ }
+
+ chainInequalities(
+ inferAdditionalConstraints(constraints) ++
+ inferConstraintsFromLiteralBindings(constraints))
+ }
+
+ /**
+ * Infers an additional set of constraints from a given set of inequality
constraints.
+ * For e.g., if an operator has constraints of the form (`a > b`, `b > 5`),
this returns an
+ * additional constraint of the form `a > 5`.
+ */
+ private def inferInequalityConstraints(constraints: ExpressionSet):
ExpressionSet = {
+ // Substituting through a comparison is only safe under binary-stable
collations, for the
+ // same reason as in inferConstraintsFromLiteralBindings:
non-binary-stable collations may
+ // order values differently than their binary representation would
suggest. Attribute-to-
+ // attribute equality is excluded here too: its transitivity is already
handled by
+ // inferAdditionalConstraints via substitution, so chaining it here would
only re-derive the
+ // same fact as a redundant <= / >= pair instead of the clean substituted
equality that path
+ // already produces.
+ val validComparisons = constraints.filter {
+ case b: BinaryComparison =>
+ b.deterministic && !b.throwable && !b.isInstanceOf[EqualNullSafe] &&
Review Comment:
**Non-blocking (P2):** `!b.throwable` is not positive evidence that the
operand is safe to evaluate on additional rows: ordinary ANSI casts and
arithmetic can fail while inheriting `throwable = false`. A cross-side residual
such as `CAST(x.s AS INT) < y.b`, combined with `y.b < 5`, now derives a child
filter `CAST(x.s AS INT) < 5`; an unmatched `x` row that never evaluated the
residual before can therefore raise `CAST_INVALID_INPUT`. This changes query
behavior for inner, left-outer, and left-anti paths rather than merely adding a
pushdown.
**Verification:**
- **Regression:** Execute an ANSI parsing-cast range predicate with
unmatched keys through inner, outer, and anti join relocation paths and verify
that enabling constraint propagation neither evaluates the cast on previously
unevaluated rows nor changes the result.
- **Compatibility:** Verify that a lossless numeric widening comparison
still derives and pushes its range bound, while parsing, narrowing, arithmetic,
and unknown-expression operands do not.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/QueryPlanConstraints.scala:
##########
@@ -126,6 +125,92 @@ trait ConstraintHelper {
inferred -- constraints
}
+ /**
+ * Infers the full additional constraint set for a given set of constraints
by combining the
+ * equality-based inferences above ([[inferAdditionalConstraints]],
+ * [[inferConstraintsFromLiteralBindings]]) with transitive inequality
chaining
+ * ([[inferInequalityConstraints]]). The inequality step is iterated to a
fixed point because a
+ * single pass only chains pairs present in the input set: e.g. `a < b && b
< c && c < 5`
+ * requires one pass to derive `b < 5` and a second to derive `a < 5` from
it.
+ */
+ def inferAllAdditionalConstraints(constraints: ExpressionSet): ExpressionSet
= {
+ @tailrec
+ def chainInequalities(inferred: ExpressionSet): ExpressionSet = {
+ val newlyInferred = inferInequalityConstraints(constraints ++ inferred)
Review Comment:
**Blocking (P1):** This fixed point rescans the full candidates-by-bounds
product after every newly discovered bound. For `x1 < x2 < ... < xn < L`, only
one bound advances per round, but all `n - 1` candidates are compared with
every settled bound each time, yielding Theta(n^3) semantic comparisons (about
500 million at `n = 1000`) before execution. Because constraint propagation is
enabled by default and the input width is unbounded, wide generated predicates
can make planning impractical even when the inferred filters provide no I/O
benefit.
**Recommended change:** Replace the recursive full-set fixed point with an
endpoint-indexed directed work list. Normalize each inequality into a directed
edge with strictness metadata, expand a foldable equality into both directed
bounds outside ExpressionSet canonical membership, seed the queue with
equality- and literal-derived bounds, and propagate only newly discovered
semantic bounds along adjacent edges. Convert results back to ExpressionSet
only at the output boundary. Add mixed-chain spelling regressions and a wide
sparse-chain workload that detects settled-pair rescanning.
**Why this works:** A directed record preserves the endpoint orientation
that commutative EqualTo canonicalization loses. An adjacency index finds only
edges whose endpoint matches a newly queued upper or lower bound; the derived
edge is strict when either input edge is strict, and a semantic-key set
prevents cycles and duplicate work. Each discovered bound is processed once
instead of rebuilding and crossing the complete closure on every depth step.
**Scope:** Make inequality closure explicitly directed and incremental while
preserving the public ExpressionSet result and both existing production call
routes.
**Compatibility:** Constraint propagation continues to infer the complete
supported transitive range closure when enabled, including safe
literal-equality and widening-cast chains, and produces no added bounds when
disabled.
**Risks:** An incorrect semantic key can merge distinct typed or collated
bounds or fail to terminate on cycles. Strictness can be weakened if a path
containing one strict comparison is combined incorrectly. Multiple incomparable
literal bounds can still require output-proportional work even after avoidable
rescans are removed.
**Constraints:** Preserve all currently supported strict, non-strict, upper,
lower, equality-bound, cast-coercion, and collation admission semantics except
where a separate safety repair narrows unsafe expressions. Keep
spark.sql.constraintPropagation.enabled as the existing switch and preserve
legacy no-inference behavior when it is disabled. Return canonical
ExpressionSet predicates to callers so downstream constraint and filter
consumers keep their existing contract.
**Success:** Mixed equality/inequality chains derive the same bound for both
SQL equality operand orders. Every newly discovered directed bound is expanded
once; advancing chain depth does not trigger a Cartesian rescan of all settled
candidates and bounds. Strictness and direction of every emitted upper or lower
bound match the composed path. Plan-local constraints and join-side pushed
filters observe the same complete canonical closure.
**Verification:**
- **Regression:** Exercise equality-derived literal bounds followed by
inequality edges in both equality operand orders and verify identical inferred
upper and lower bounds through both plan-local and join-side propagation.
- **Behavior:** Cover combinations of strict and non-strict edges, cycles,
multiple paths, and multiple foldable bounds and verify the exact directed
closure without duplicate predicates.
- **Regression:** Build a wide sparse chain with one terminal literal bound
and verify planning completes within a bounded workload that would expose full
settled-set rescanning rather than relying on a small three-edge example.
- **Inspection:** Confirm the traversal indexes edges by endpoint, queues
only newly discovered semantic bounds, and has no loop that recomputes the full
candidates-by-bounds product at each fixed-point depth.
--
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]
