spark-test commented on code in PR #60198:
URL: https://github.com/apache/spark/pull/60198#discussion_r4502438282
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/Optimizer.scala:
##########
@@ -3199,6 +3215,35 @@ object PushPredicateThroughJoin extends
Rule[LogicalPlan] with PredicateHelper {
(leftEvaluateCondition, rightEvaluateCondition, commonCondition ++ stayUp)
}
+ /**
+ * [[split]] for an external lookup join, which also returns the conjuncts
that must not move.
+ * Moving a conjunct onto the lookup side must not turn a query that runs
without it into one
+ * that fails to plan:
+ * - A comparison that lookup planning can offer as a lookup candidate,
such as `t.dept = 'x'`,
+ * stays with the common conditions, in the join condition, so that the
connector can still
+ * select it as an access path.
+ * - A conjunct over lookup columns that a lookup-side stage cannot
evaluate, such as one with
+ * a Python UDF or a subquery, is pinned: it is neither pushed nor moved
into the condition.
+ * - A conjunct without references, such as a literal, belongs to neither
side. It stays with
+ * the common conditions, so that a condition that folds to false keeps
its result without a
+ * lookup, instead of leaving a join whose stream side is empty.
+ *
+ * @return (canEvaluateInLeft, canEvaluateInRight, haveToEvaluateInBoth,
pinned)
+ */
+ private def splitLookupJoin(condition: Seq[Expression], join:
ExternalLookupJoin) = {
+ val (pinned, movable) = condition.partition { cond =>
+ cond.references.exists(join.right.outputSet.contains) &&
+ !ExternalLookupJoinShapes.canEvaluateInLookupStage(cond)
Review Comment:
**Non-blocking (P2):** This check lets through a conjunct that a lookup-side
stage cannot evaluate. `canEvaluateInLookupStage` accepts `UpdateFields`, which
`Column.withField` and `Column.dropFields` produce. `UpdateFields` is
`Unevaluable`, and only `ReplaceUpdateFieldsExpression` lowers it, in the
optimizer's last batch. By then `PrepareExternalLookupJoin` has folded
lookup-side filters into `PreparedExternalLookupJoin.lookupTransforms`, which
no expression rule reaches.
So a lookup-only conjunct that still holds a whole-struct
`withField`/`dropFields` value now moves onto the lookup side: from the join
condition for any join type, or from a filter above an inner join. The
micro-batch then fails with an internal error that `update_fields` cannot be
evaluated. For example, `stream.join(lookup, stream("id") === lookup("id") &&
lookup("info").dropFields("b") === struct(lit(1).as("a")), "left_outer")`
returned its rows before this commit, because the conjunct stayed in the
residual condition, where `ReplaceUpdateFieldsExpression` lowered it. Field
reads through `withField` are not affected, since `SimplifyExtractValueOps`
removes them earlier.
Pinning conjuncts that contain `UpdateFields` here would keep them where
they ran before. Rejecting `UpdateFields` in `canEvaluateInLookupStage`, in the
stacked PR that adds lookup-side stages, would also cover lookup sides written
with `withField`.
**Verification:**
- **Regression:** A plan test in the predicate pushdown suite for external
lookup joins shows that a conjunct over lookup columns that contains
UpdateFields stays in the join condition for every join type and stays above an
inner join. An end-to-end test shows that a lookup join whose condition
compares a withField or dropFields value of a lookup struct column returns the
same rows as with PushDownPredicates excluded.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/ExternalLookupJoin.scala:
##########
@@ -0,0 +1,338 @@
+/*
+ * 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.plans.logical
+
+import org.apache.spark.SparkException
+import org.apache.spark.sql.catalyst.expressions.{Attribute, AttributeSet,
BinaryComparison,
+ Cast, EqualNullSafe, EqualTo, Expression, ExprUtils,
ExternalUserDefinedFunction, GreaterThan,
+ GreaterThanOrEqual, LessThan, LessThanOrEqual, NamedExpression, PythonUDF,
SubqueryExpression}
+import org.apache.spark.sql.catalyst.plans.{Inner, JoinType, LeftAnti,
LeftOuter, LeftSemi}
+import org.apache.spark.sql.catalyst.trees.TreePattern.{EXTERNAL_LOOKUP_HINT,
EXTERNAL_LOOKUP_JOIN,
+ TreePattern}
+import org.apache.spark.sql.catalyst.util.CharVarcharUtils
+import org.apache.spark.sql.connector.read.{ExternalLookupReaderFactory, Scan}
+import org.apache.spark.sql.types.{BinaryType, BooleanType, ByteType,
CharType, DataType,
+ DateType, DecimalType, IntegerType, IntegralType, LongType, ShortType,
StringType,
+ TimestampNTZType, TimestampType, VarcharType}
+
+/**
+ * The cache settings of one external lookup join, from the
+ * `spark.sql.join.externalLookup.cache.*` configurations unless the join's
`EXTERNAL_LOOKUP` hint
+ * overrides them. A cache lives within a task, per input partition, so these
bound each such cache:
+ * `maxEntries` keys, each kept for `ttlMillis` (0 keeps it until evicted). A
key with more than
+ * `maxRowsPerKey` result rows is still joined but not cached, so that one
large result cannot
+ * fill the cache. At present, a join that selects a range comparison (`<`,
`<=`, `>`, `>=`) caches
+ * nothing, whatever these settings say.
+ */
+case class ExternalLookupCacheOptions(
+ enabled: Boolean,
+ maxEntries: Int,
+ ttlMillis: Long,
+ maxRowsPerKey: Int)
+
+/**
+ * Marks the plan below it as the lookup side that the `EXTERNAL_LOOKUP` hint
asks an external
+ * lookup join to read, with the hint's cache options. The external lookup
join rules consume it
+ * when they form the join.
+ *
+ * Join strategy hints are [[ResolvedHint]] nodes with a [[JoinStrategyHint]].
This hint is not,
+ * because that representation cannot keep two of its guarantees:
+ * - A strategy hint that conflicts with it must fail the query. [[HintInfo]]
holds one strategy,
+ * and `ResolveJoinStrategyHints` merges a strategy hint on an already
hinted relation into the
+ * existing [[ResolvedHint]] with `HintInfo.merge`, which keeps one
strategy and only warns.
+ * - It selects how the lookup side is read, not a physical join.
`EliminateResolvedHint` moves
+ * every [[ResolvedHint]] it can attach into `Join.hint`, for physical join
selection.
+ */
+case class ExternalLookupHint(
+ child: LogicalPlan,
+ cacheOptions: ExternalLookupCacheOptions) extends UnaryNode {
+ override def output: Seq[Attribute] = child.output
+
+ final override val nodePatterns: Seq[TreePattern] = Seq(EXTERNAL_LOOKUP_HINT)
+
+ override protected def withNewChildInternal(newChild: LogicalPlan):
ExternalLookupHint =
+ copy(child = newChild)
+}
+
+/**
+ * A lookup join retained through ordinary expression and operator
optimization.
+ *
+ * @param requiredLookupOutput Lookup-side attributes still required above
this join. They are
+ * part of this node's references, so column pruning keeps them on the
lookup side.
+ * `ColumnPruning` narrows them to what the parent reads, and
`PrepareExternalLookupJoin` prunes
+ * the lookup scan to them.
+ */
+case class ExternalLookupJoin(
+ left: LogicalPlan,
+ right: LogicalPlan,
+ joinType: JoinType,
+ condition: Option[Expression],
+ cacheOptions: ExternalLookupCacheOptions,
+ requiredLookupOutput: Seq[Attribute]) extends BinaryNode {
+
+ final override val nodePatterns: Seq[TreePattern] = Seq(EXTERNAL_LOOKUP_JOIN)
+
+ override def output: Seq[Attribute] = joinType match {
+ case Inner => left.output ++ right.output
+ case LeftOuter => left.output ++ right.output.map(_.withNullability(true))
+ case LeftSemi | LeftAnti => left.output
+ case other => throw SparkException.internalError(s"Unexpected external
lookup join type $other")
+ }
+
+ override def maxRows: Option[Long] = joinType match {
+ case LeftSemi | LeftAnti => left.maxRows
+ case _ => None
+ }
+
+ override protected def withNewChildrenInternal(
+ newLeft: LogicalPlan,
+ newRight: LogicalPlan): ExternalLookupJoin = copy(left = newLeft, right
= newRight)
+}
+
+/**
+ * The join-condition and lookup-side shapes that an external lookup join can
run. Analysis,
+ * lookup planning and the optimizer rules for lookup joins share them.
+ */
+object ExternalLookupJoinShapes {
+ /**
+ * A lookup candidate: a comparison of a lookup column with a value, written
with the lookup
+ * column on the left. `predicateName` is the name of the DSv2 `Predicate`
that is offered to the
+ * connector for it: the SQL operator of the comparison, as
`V2ExpressionBuilder` names it.
+ */
+ case class LookupComparison(
+ lookupField: Attribute,
+ parameter: Expression,
+ predicateName: String,
+ nullSafe: Boolean)
+
+ /**
+ * `conjunct` of the condition of a lookup join as a lookup candidate: `=`,
`<=>`, `<`, `<=`, `>`
+ * or `>=` between a bare lookup column and a value, in either operand
order. When the lookup
+ * column is written on the right, the comparison is rebuilt with it on the
left, so
+ * `s.ts < t.event_time` is offered as `event_time > ?`.
+ *
+ * The value is the other operand, which Spark evaluates for each stream row
to fill the
+ * comparison's parameter. It may read only stream columns, and it may read
none, as a literal
+ * does. A value that reads lookup columns, as in `t.a = t.b`, cannot be a
parameter.
+ *
+ * The lookup column must be of a type on which [[ExternalLookupKeyTypes]]
lets Spark offer the
+ * comparison, and not a CHAR or VARCHAR column. A comparison on another
column is an ordinary
+ * conjunct.
+ *
+ * @param streamOutput The output of the join's left side, the stream side.
+ * @param lookupOutput The output of the join's right side, the lookup side.
+ */
+ def lookupComparison(
+ conjunct: Expression,
+ streamOutput: AttributeSet,
+ lookupOutput: AttributeSet): Option[LookupComparison] = {
+ def isLookupComparison(lookupField: Attribute, parameter: Expression):
Boolean =
+ lookupOutput.contains(lookupField) &&
parameter.references.subsetOf(streamOutput)
+ conjunct match {
+ case comparison: BinaryComparison if isLookupOperator(comparison) =>
+ val nullSafe = comparison.isInstanceOf[EqualNullSafe]
+ Seq(comparison, withOperandsSwapped(comparison)).collectFirst {
+ case candidate @ BinaryComparison(lookupField: Attribute, parameter)
+ if isLookupComparison(lookupField, parameter) &&
+ isOfferedOn(comparison, lookupField, parameter) =>
+ LookupComparison(lookupField, parameter, candidate.sqlOperator,
nullSafe)
+ }
+ case _ => None
+ }
+ }
+
+ private def isLookupOperator(comparison: BinaryComparison): Boolean =
comparison match {
+ case _: EqualTo | _: EqualNullSafe | _: LessThan | _: LessThanOrEqual | _:
GreaterThan |
+ _: GreaterThanOrEqual => true
+ case _ => false
+ }
+
+ /**
+ * Whether Spark offers the comparison on this lookup column, with this
parameter. Spark offers
+ * no comparison on a CHAR or VARCHAR column, whose data type alone does not
show it.
+ *
+ * Spark evaluates every parameter on every stream row, before the other
join conditions. That
+ * is safe for an equality, as a hash join evaluates its keys on every row
too. A range is a
+ * join condition, which a join evaluates only after its keys match, so its
parameter must be
+ * one that cannot fail, as `cannotFail` decides.
+ */
+ private def isOfferedOn(
+ comparison: BinaryComparison,
+ lookupField: Attribute,
+ parameter: Expression): Boolean =
+ !ExternalLookupKeyTypes.isCharOrVarchar(lookupField) && (comparison match {
+ case _: EqualTo | _: EqualNullSafe =>
+ ExternalLookupKeyTypes.supportsEquality(lookupField.dataType)
+ case _ =>
+ ExternalLookupKeyTypes.supportsRange(lookupField.dataType) &&
cannotFail(parameter)
+ })
+
+ /**
+ * Whether a range parameter cannot fail:
ExprUtils.canEvaluateUnconditionally accepts it, or it
+ * is a widening numeric cast of such a value. Analysis adds such a cast
when the stream side has
+ * a narrower type than the lookup column, as an INT column compared with a
BIGINT key.
+ */
+ private def cannotFail(parameter: Expression): Boolean = parameter match {
+ case cast: Cast =>
+ isTotalNumericWidening(cast.child.dataType, cast.dataType) &&
+ ExprUtils.canEvaluateUnconditionally(cast.child)
+ case _ => ExprUtils.canEvaluateUnconditionally(parameter)
+ }
+
+ /**
+ * Whether a cast converts every value of its source type: an integral type
to a wider one, or
+ * an integral type or DECIMAL to a DECIMAL with room for all of its digits.
`Cast.canUpCast`
+ * is not enough: it also accepts casts that fail on some values, as a
far-future DATE overflows
+ * a TIMESTAMP, and 9999999999999999999 as DECIMAL(19, 0) exceeds BIGINT
under ANSI.
+ */
+ private def isTotalNumericWidening(from: DataType, to: DataType): Boolean =
(from, to) match {
+ case (from: IntegralType, to: IntegralType) => from.defaultSize <
to.defaultSize
+ case (from: IntegralType, to: DecimalType) => to.isWiderThan(from)
+ case (from: DecimalType, to: DecimalType) => to.isWiderThan(from)
+ case _ => false
+ }
+
+ /** The same comparison with its operands swapped, so that `a < b` becomes
`b > a`. */
+ private def withOperandsSwapped(comparison: BinaryComparison):
BinaryComparison =
+ comparison match {
+ case EqualTo(left, right) => EqualTo(right, left)
+ case EqualNullSafe(left, right) => EqualNullSafe(right, left)
+ case LessThan(left, right) => GreaterThan(right, left)
+ case LessThanOrEqual(left, right) => GreaterThanOrEqual(right, left)
+ case GreaterThan(left, right) => LessThan(right, left)
+ case GreaterThanOrEqual(left, right) => LessThanOrEqual(right, left)
+ }
+
+ /**
+ * Whether a lookup-side projection or filter can evaluate `expression` on
the lookup results,
+ * inside the join. Python and external UDFs and subqueries are planned as
separate operators,
+ * which a lookup-side stage cannot contain. A non-deterministic expression
is not allowed
+ * either: the join evaluates a stage on each returned row, so one stored
row could get
+ * different values for different stream rows, which a read of the table
never does.
+ */
+ def canEvaluateInLookupStage(expression: Expression): Boolean =
+ expression.deterministic && !expression.exists {
+ case _: PythonUDF | _: ExternalUserDefinedFunction | _:
SubqueryExpression => true
+ case _ => false
+ }
+}
+
+/** One optimized lookup-side operation, retained in evaluation order. */
+sealed trait ExternalLookupTransform {
+ def input: Seq[Attribute]
+ def output: Seq[Attribute]
+}
+
+case class ExternalLookupProject(
+ projectList: Seq[NamedExpression],
+ input: Seq[Attribute]) extends ExternalLookupTransform {
+ override def output: Seq[Attribute] = projectList.map(_.toAttribute)
+}
+
+case class ExternalLookupFilter(
+ condition: Expression,
+ input: Seq[Attribute]) extends ExternalLookupTransform {
+ override def output: Seq[Attribute] = input
+}
+
+/**
+ * A lookup join after its optimized lookup side has been converted to an
external operation.
+ *
+ * @param lookupKeys The table column that each selected lookup predicate
compares with the
+ * parameter at the same position. Spark checks each row that the connector
returns against
+ * them, so the read schema must hold these columns.
+ * @param lookupKeyPredicates The name of each selected lookup predicate, such
as `=` or `<`,
+ * with the lookup column on the left. It tells how Spark checks the key.
+ */
+case class PreparedExternalLookupJoin(
+ child: LogicalPlan,
+ tableName: String,
+ @transient scan: Scan,
+ readerFactory: ExternalLookupReaderFactory,
+ maxParameterRowsPerRequest: Int,
+ lookupInput: Seq[Attribute],
+ lookupOutput: Seq[Attribute],
+ lookupTransforms: Seq[ExternalLookupTransform],
+ lookupParameters: Seq[Expression],
+ nullSafeParameters: Seq[Boolean],
+ lookupKeys: Seq[Attribute],
+ lookupKeyPredicates: Seq[String],
+ joinType: JoinType,
+ residualCondition: Option[Expression],
+ cacheOptions: ExternalLookupCacheOptions) extends UnaryNode {
+
+ override def output: Seq[Attribute] = joinType match {
+ case Inner => child.output ++ lookupOutput
+ case LeftOuter => child.output ++ lookupOutput.map(_.withNullability(true))
+ case LeftSemi | LeftAnti => child.output
+ case other => throw SparkException.internalError(s"Unexpected external
lookup join type $other")
+ }
+
+ override def references: AttributeSet =
+ AttributeSet(lookupParameters.flatMap(_.references)) ++
+ residualCondition.map(_.references.intersect(child.outputSet))
+ .getOrElse(AttributeSet.empty)
+
+ override protected def withNewChildInternal(newChild: LogicalPlan):
PreparedExternalLookupJoin =
+ copy(child = newChild)
+
+ // Connector descriptions may contain secrets. The physical node renders a
redacted description.
+ override protected def stringArgs: Iterator[Any] = Iterator(child)
+}
+
+/**
+ * The types of lookup columns on which Spark offers a comparison to the
connector as a lookup
+ * candidate. The connector evaluates a selected candidate in its store, so
Spark offers only types
+ * whose values a store compares as Spark does. A comparison on any other type
stays in Spark's
Review Comment:
**Nit (P3):** With this PR's pushdown, this sentence no longer holds for a
comparison that reads only lookup columns. Such a comparison, for example
`t.score = 5.0` on a DOUBLE, now moves onto the lookup side: from the join
condition for every join type, and from a filter above an inner join. Spark
evaluates it there in a lookup-side stage, and the connector may also receive
it as a pushed filter. So the connector's filter handling now matters for
exactly the types this object keeps out of the offered candidates, while the
doc says they stay with Spark. The PR description and the plan test 'move a
comparison on a type that is never offered as a lookup key like any conjunct'
already describe the new behavior.
**Verification:**
- **Inspection:** The ExternalLookupKeyTypes doc's account of where a
comparison on a never-offered type is evaluated matches how predicate pushdown
handles a lookup-only comparison that is not a lookup candidate, including the
existing plan coverage that moves DOUBLE and STRING-range comparisons onto the
lookup side.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/Optimizer.scala:
##########
@@ -3293,6 +3338,62 @@ object PushPredicateThroughJoin extends
Rule[LogicalPlan] with PredicateHelper {
case other =>
throw SparkException.internalError(s"Unexpected join type: $other")
}
+ // The same for an external lookup join. The conjuncts keep their written
order where they stay.
+ case f @ Filter(filterCondition, j: ExternalLookupJoin) =>
+ val conjuncts = splitConjunctivePredicates(filterCondition)
+ val (leftFilterConditions, rightFilterConditions, commonFilterCondition,
_) =
+ splitLookupJoin(conjuncts, j)
+ val (newJoinConditions, rightFilterConditionsToPush) = j.joinType match {
+ case Inner =>
+ // don't push throwable expressions into join condition
+ (commonFilterCondition.filter(cond => canEvaluateWithinJoin(cond) &&
!cond.throwable),
+ rightFilterConditions)
+ // A filter on lookup columns above a left outer join also removes the
null-padded rows of
+ // stream rows without a match, so it stays above. EliminateOuterJoin
first makes the join
+ // inner when the filter rejects nulls. A semi or anti join outputs no
lookup column.
+ case LeftOuter | LeftSemi | LeftAnti => (Nil, Nil)
+ case other =>
+ throw SparkException.internalError(s"Unexpected external lookup join
type: $other")
+ }
+ val pushed = leftFilterConditions ++ rightFilterConditionsToPush ++
newJoinConditions
+ if (pushed.isEmpty) {
+ f
+ } else {
+ val newJoin = j.copy(
+ left = leftFilterConditions.
+ reduceLeftOption(And).map(Filter(_, j.left)).getOrElse(j.left),
+ right = rightFilterConditionsToPush.
+ reduceLeftOption(And).map(Filter(_, j.right)).getOrElse(j.right),
+ condition = (j.condition.toSeq ++
newJoinConditions).reduceLeftOption(And))
+ conjuncts.filterNot(pushed.contains).
+ reduceLeftOption(And).map(Filter(_, newJoin)).getOrElse(newJoin)
+ }
+
+ // Push the join filter of an external lookup join into its children.
+ case j: ExternalLookupJoin =>
+ val conjuncts =
j.condition.map(splitConjunctivePredicates).getOrElse(Nil)
+ val (leftJoinConditions, rightJoinConditions, _, _) =
splitLookupJoin(conjuncts, j)
+ val leftJoinConditionsToPush = j.joinType match {
+ case Inner | LeftSemi => leftJoinConditions
+ // A stream-side conjunct decides which stream rows find a match, so
it stays in the
+ // condition. The lookup join checks it on each stream row before the
lookup.
Review Comment:
**Nit (P3):** This holds only for conjuncts that
`ExprUtils.canEvaluateUnconditionally` accepts.
`ExternalLookupJoinExec.splitResidualCondition` checks a stream-only conjunct
before the lookup only in that case, and evaluates every other one on joined
rows after the lookup. So for a conjunct such as `upper(s.region) = 'EU'` or
`s.amount * 2 > 100`, every stream row it rejects is still looked up.
`PushPredicateThroughExternalLookupJoinSuite` repeats the claim ("The join
checks it there before the lookup."); its `streamKey > 1` example happens to be
a shape that is checked early.
**Verification:**
- **Inspection:** Both comments describe when a stream-side conjunct of a
left outer or left anti lookup join is checked before the lookup in the same
way the operator splits its residual into the pre-lookup stream condition.
--
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]
