LukaZdravic commented on code in PR #58823:
URL: https://github.com/apache/spark/pull/58823#discussion_r4018054540
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/basicLogicalOperators.scala:
##########
@@ -2688,12 +2688,39 @@ object AsOfJoin {
rightOperand: Expression,
normalizedOp: MatchComparisonOperator)
: (Expression, Expression, Seq[Expression], Seq[Expression]) = {
+ val (coercedLeft, coercedRight) = coerceMatchLeafOperands(leftOperand,
rightOperand)
val (asOfCondition, orderExpression) =
- buildMatchExpressions(leftOperand, rightOperand, normalizedOp)
- val (leftSortExprs, rightSortExprs) = matchSortExpressions(leftOperand,
rightOperand)
+ buildMatchExpressions(coercedLeft, coercedRight, normalizedOp)
+ val (leftSortExprs, rightSortExprs) = matchSortExpressions(coercedLeft,
coercedRight)
(asOfCondition, orderExpression, leftSortExprs, rightSortExprs)
}
+ /**
+ * Casts a scalar operand pair to the common type the comparison operator
would use, so the
+ * comparison, ordering distance, and per-side sort keys all agree. This is
required for a
+ * string vs DATE/TIMESTAMP/number pair: the sort-merge scan sorts the right
buffer by its raw
+ * sort key, and only a shared type keeps that order consistent with the
coerced comparison.
+ * STRUCT and ARRAY operands are left untouched; their sort key is the whole
value, so a
+ * per-field cast could not reach it.
Review Comment:
Both modes are covered end-to-end in `AsOfJoinSQLSuite`: the `DATE vs
STRING` and `INT vs STRING` tests each run under `ansi=true` and `ansi=false`
via `withSQLConf`, and the INT case asserts the per-mode sort-key type (LONG
under ANSI, INT otherwise), which pins the mode-dependent common-type
selection. Those tests set the conf explicitly, so they do not depend on the
SQLConf default — a default change would not hide a regression. I left the
catalyst unit assertions as-is to avoid duplicating that coverage.
##########
sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/plans/logical/AsOfJoinMatchConditionTypesSuite.scala:
##########
@@ -29,9 +29,39 @@ class AsOfJoinMatchConditionTypesSuite extends SparkFunSuite
{
assert(!MatchConditionTypes.usesStructDecomposition(IntegerType, LongType))
}
- test("string and temporal types are incompatible") {
- assert(!MatchConditionTypes.areOperandsCompatible(StringType,
TimestampType))
- assert(!MatchConditionTypes.areOperandsCompatible(DateType, StringType))
+ test("scalar string and temporal types coerce like the comparison operator")
{
+ // SPARK-59527: a scalar string vs DATE/TIMESTAMP pair is accepted and
coerced, matching `>=`.
+ assert(MatchConditionTypes.areOperandsCompatible(StringType,
TimestampType))
+ assert(MatchConditionTypes.areOperandsCompatible(DateType, StringType))
+ // The shared common type is the temporal type (string is cast to it), not
string, so the
+ // sort-merge sort key and the comparison order agree.
+ assert(MatchConditionTypes.matchComparisonCommonType(DateType,
StringType).contains(DateType))
+ assert(
+ MatchConditionTypes.matchComparisonCommonType(StringType, TimestampType)
+ .contains(TimestampType))
+ }
Review Comment:
`areFieldTypesCompatible` is also called recursively on struct fields and
array elements (via `arePositionalStructsCompatible` /
`areArrayElementsCompatible`), which the top-level guard in
`areOperandsCompatible` has not validated — so it has to keep its own
`isValidOperandType` check. The outer guard is still needed for the scalar
branches, because `matchComparisonCommonType` / `findWiderTypeForTwo` do not
verify operand orderability themselves. Neither check is safely removable, so I
kept both; the top-level double-check on composite operands is a cheap,
intentional overlap.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/basicLogicalOperators.scala:
##########
@@ -2703,7 +2730,56 @@ object AsOfJoin {
def isValidOperandType(dataType: DataType): Boolean =
RowOrdering.isOrderable(dataType) && !containsEmptyStructType(dataType)
+ /**
+ * Top-level operand compatibility. A scalar pair is compatible when the
comparison operator
+ * would accept it, so DATE/TIMESTAMP vs STRING (and string vs number)
coerce like `>=`.
+ * STRUCT/ARRAY operands keep the stricter widening rule via
[[areFieldTypesCompatible]],
+ * because their sort key is the whole value and cannot carry a per-field
string cast.
+ */
def areOperandsCompatible(leftType: DataType, rightType: DataType):
Boolean = {
+ if (!isValidOperandType(leftType) || !isValidOperandType(rightType)) {
+ false
+ } else {
+ (leftType, rightType) match {
+ case (_: StructType, _) | (_, _: StructType) | (_: ArrayType, _) |
(_, _: ArrayType) =>
+ areFieldTypesCompatible(leftType, rightType)
+ case _ =>
+ matchComparisonCommonType(leftType, rightType).isDefined ||
+ TypeCoercion.findWiderTypeForTwo(leftType, rightType).isDefined
+ }
Review Comment:
`collectFirst { case j: AsOfJoin => j }.get` is the established pattern
throughout `AsOfJoinSQLSuite` (every analysis test uses it), so I kept it for
consistency with the surrounding tests. If the suite owners prefer
`getOrElse(fail(...))`, I am happy to switch them all in a follow-up.
--
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]
