vranes commented on code in PR #59834:
URL: https://github.com/apache/spark/pull/59834#discussion_r4402733857
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/basicLogicalOperators.scala:
##########
@@ -2965,200 +2993,32 @@ object AsOfJoin {
}
}
- /**
- * Returns the `asOfCondition` and `orderExpression` for one MATCH_CONDITION
comparison. The
- * operands are compared as whole values, so a NULL struct never matches,
while a struct whose
- * fields are all NULL can.
- */
- private def buildMatchExpressions(
- leftOperand: Expression,
- rightOperand: Expression,
- operator: MatchComparisonOperator): (Expression, Expression) = {
- val orderExpression = buildOrderExpression(leftOperand, rightOperand,
operator)
- operator match {
- case GreaterThanOrEqualOp =>
- (GreaterThanOrEqual(leftOperand, rightOperand), orderExpression)
- case GreaterThanOp =>
- (GreaterThan(leftOperand, rightOperand), orderExpression)
- case LessThanOrEqualOp =>
- (LessThanOrEqual(leftOperand, rightOperand), orderExpression)
- case LessThanOp =>
- (LessThan(leftOperand, rightOperand), orderExpression)
- }
- }
-
- private def buildOrderExpression(
- leftOperand: Expression,
- rightOperand: Expression,
- operator: MatchComparisonOperator): Expression = {
- (leftOperand.dataType, rightOperand.dataType) match {
- case (_: ArrayType, _)
- if MatchConditionTypes.usesArrayOrderExpression(
- leftOperand.dataType, rightOperand.dataType) =>
- // MATCH_CONDITION array comparison uses Spark lexicographic ordering
(including length).
- // The ordering distance below is element-wise via ZipWith, padding
the shorter side with
- // null when lengths differ (e.g. [0, null]), not a lexicographic
length tie-break.
- buildArrayOrderExpression(leftOperand, rightOperand, operator)
- case (leftType, rightType)
- if MatchConditionTypes.usesStructDecomposition(leftType, rightType)
=>
- buildFlattenedStructOrderExpression(
- leftOperand, rightOperand, leftType.asInstanceOf[StructType],
operator)
- case _ =>
- buildLeafOrderExpression(leftOperand, rightOperand, operator)
- }
- }
-
- private def buildLeafOrderExpression(
- leftOperand: Expression,
- rightOperand: Expression,
- operator: MatchComparisonOperator): Expression = {
- if (supportsSubtract(leftOperand.dataType)) {
- operator match {
- case GreaterThanOrEqualOp | GreaterThanOp =>
- Subtract(leftOperand, rightOperand)
- case LessThanOrEqualOp | LessThanOp =>
- Subtract(rightOperand, leftOperand)
- }
- } else {
- buildSignedComparisonDistance(leftOperand, rightOperand, operator)
- }
- }
-
- private[catalyst] def supportsSubtract(dataType: DataType): Boolean = {
- dataType match {
- case _: NumericType | _: DayTimeIntervalType | _: YearMonthIntervalType |
- _: CalendarIntervalType | _: TimestampType | _: TimestampNTZType |
_: DateType =>
- true
- case _ =>
- false
- }
- }
-
- private def buildSignedComparisonDistance(
- leftOperand: Expression,
- rightOperand: Expression,
- operator: MatchComparisonOperator): Expression = {
- val (greaterValue, lesserValue) = operator match {
- case GreaterThanOrEqualOp | GreaterThanOp => (Literal(1), Literal(-1))
- case LessThanOrEqualOp | LessThanOp => (Literal(-1), Literal(1))
- }
- If(
- EqualTo(leftOperand, rightOperand),
- Literal(0),
- If(GreaterThan(leftOperand, rightOperand), greaterValue, lesserValue))
- }
-
- private def buildArrayOrderExpression(
- leftOperand: Expression,
- rightOperand: Expression,
- operator: MatchComparisonOperator): Expression = {
- val leftElementType =
leftOperand.dataType.asInstanceOf[ArrayType].elementType
- val rightElementType =
rightOperand.dataType.asInstanceOf[ArrayType].elementType
- // Both array inputs and the ZipWith lambda variables must share the
element type the binary
- // comparison uses. comparisonCommonType is the single source the type
check also reads, so
- // the two cannot disagree. castArrayElementType widens a coercible side
(e.g. INT to BIGINT)
- // and leaves a structurally equal side untouched.
- val elementType =
- MatchConditionTypes.comparisonCommonType(leftElementType,
rightElementType) match {
- case Some(commonElementType) => commonElementType
- case None =>
- // Unreachable: usesArrayOrderExpression already required a common
element type here.
- throw SparkException.internalError(
- "MATCH_CONDITION array elements have no common type")
- }
- val leftArray = castArrayElementType(leftOperand, elementType)
- val rightArray = castArrayElementType(rightOperand, elementType)
- elementType match {
- // An empty struct has no fields to split into, so it takes the
whole-value arm below.
- case struct: StructType if struct.nonEmpty =>
- val leftElement = NamedLambdaVariable("left_elem", struct, nullable =
true)
- val rightElement = NamedLambdaVariable("right_elem", struct, nullable
= true)
- val leafDiffs = collectStructLeafPairs(leftElement, rightElement,
struct).map {
- case (left, right) => buildLeafOrderExpression(left, right, operator)
- }
- val elementOrder = wrapCompositeOrderExpression(leafDiffs)
- ZipWith(
- leftArray,
- rightArray,
- LambdaFunction(elementOrder, Seq(leftElement, rightElement)))
- case _ =>
- val leftElement = NamedLambdaVariable("left_elem", elementType,
nullable = true)
- val rightElement = NamedLambdaVariable("right_elem", elementType,
nullable = true)
- val elementOrder = buildLeafOrderExpression(leftElement, rightElement,
operator)
- ZipWith(
- leftArray,
- rightArray,
- LambdaFunction(elementOrder, Seq(leftElement, rightElement)))
- }
- }
-
- /** Cast an array operand to the given element type, keeping its own
`containsNull`. */
- private def castArrayElementType(operand: Expression, elementType:
DataType): Expression = {
- val arrayType = operand.dataType.asInstanceOf[ArrayType]
- // Skip the cast when the element already matches structurally, the rule
the type check uses,
- // so a structurally equal side keeps its own field names instead of being
renamed by a cast.
- val alreadyMatches =
- DataType.equalsStructurally(arrayType.elementType, elementType,
ignoreNullability = true)
- if (alreadyMatches) {
- operand
- } else {
- Cast(operand, ArrayType(elementType, arrayType.containsNull))
- }
- }
-
- private def buildFlattenedStructOrderExpression(
- leftOperand: Expression,
- rightOperand: Expression,
- structType: StructType,
- operator: MatchComparisonOperator): Expression = {
- val leafDiffs = collectStructLeafPairs(leftOperand, rightOperand,
structType).map {
- case (left, right) => buildLeafOrderExpression(left, right, operator)
- }
- wrapCompositeOrderExpression(leafDiffs)
- }
-
- private def collectStructLeafPairs(
- leftOperand: Expression,
- rightOperand: Expression,
- structType: StructType): Seq[(Expression, Expression)] = {
- structFieldExprs(leftOperand, structType)
- .zip(structFieldExprs(rightOperand, structType))
- .flatMap {
- case (left, right) =>
- if (MatchConditionTypes.usesStructDecomposition(left.dataType,
right.dataType)) {
- collectStructLeafPairs(
- left, right, left.dataType.asInstanceOf[StructType])
- } else {
- Seq((left, right))
- }
- }
- }
-
- /** Groups multiple field distances in a struct because their types may
differ. */
- private def wrapCompositeOrderExpression(diffs: Seq[Expression]): Expression
= {
- diffs match {
- case Seq(single) => single
- case _ => CreateStruct(diffs)
+ /** Builds the subtraction-based distance for a DataFrame fallback join. */
+ private[catalyst] def makeOrderingExpr(
+ leftAsOf: Expression,
+ rightAsOf: Expression,
+ direction: AsOfJoinDirection,
+ evalContext: NumericEvalContext): Expression = {
+ def subtract(left: Expression, right: Expression): Expression = {
+ Subtract(left = left, right = right, evalContext = evalContext)
}
- }
- private def structFieldExprs(
- operand: Expression,
- structType: StructType): Seq[Expression] = {
- operand match {
- case ns: CreateNamedStruct => ns.valExprs
- case _ =>
- structType.indices.map(index =>
- GetStructField(operand, index, Some(structType(index).name)))
+ direction match {
+ case Backward => subtract(leftAsOf, rightAsOf)
+ case Forward => subtract(rightAsOf, leftAsOf)
+ case Nearest =>
+ If(GreaterThan(leftAsOf, rightAsOf),
+ subtract(leftAsOf, rightAsOf), subtract(rightAsOf, leftAsOf))
}
}
- private def makeAsOfCond(
+ private[catalyst] def makeAsOfCond(
leftAsOf: Expression,
rightAsOf: Expression,
tolerance: Option[Expression],
allowExactMatches: Boolean,
- direction: AsOfJoinDirection): Expression = {
+ direction: AsOfJoinDirection,
+ evalContext: NumericEvalContext =
NumericEvalContext.fromSQLConf(SQLConf.get)): Expression = {
Review Comment:
Done, removed the default. Thanks!
--
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]
