Copilot commented on code in PR #58823:
URL: https://github.com/apache/spark/pull/58823#discussion_r4017168553
##########
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:
`matchComparisonCommonType` depends on `SQLConf.get.ansiEnabled`, but this
suite doesn’t explicitly exercise both ANSI and non-ANSI modes. Since the PR’s
behavior is explicitly mode-dependent (and the SQL suite covers both), consider
adding ANSI-enabled/disabled variants here as well to pin the per-mode
common-type selection at the unit level and avoid regressions if SQLConf
defaults change.
##########
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:
The rationale in this Scaladoc is not always accurate for STRUCT operands:
`matchSortExpressions` can decompose SQL tuple structs into per-field sort
expressions (i.e., the sort key is not necessarily the whole struct value).
Since the code intentionally keeps STRUCT/ARRAY strict anyway, consider
rewording this comment to reflect the actual reason (e.g., consistency with
existing strict rules / inability to safely apply comparison-style string casts
in composite ordering semantics), rather than claiming per-field casts cannot
reach the sort key.
##########
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:
`areOperandsCompatible` validates operand types up front, but
`areFieldTypesCompatible` (called in the STRUCT/ARRAY branch) also performs
`isValidOperandType` checks internally. Since `areFieldTypesCompatible` is
private, consider consolidating the validity check in one place (either rely on
the outer check and remove the inner one, or remove the outer check for the
STRUCT/ARRAY branch) to reduce duplicated logic.
##########
sql/core/src/test/scala/org/apache/spark/sql/AsOfJoinSQLSuite.scala:
##########
@@ -200,14 +201,41 @@ class AsOfJoinSQLSuite extends QueryTest with
SharedSparkSession {
sqlState = Some("42K09"),
parameters = Map(
"type1" -> "\"TIMESTAMP\"",
- "type2" -> "\"STRING\""),
+ "type2" -> "\"DECIMAL(5,2)\""),
queryContext = Array(
ExpectedContext(
fragment = """ASOF JOIN quotes q
- | MATCH_CONDITION (t.trade_time >= q.symbol)
+ | MATCH_CONDITION (t.trade_time >= q.bid_price)
| ON t.symbol = q.symbol""".stripMargin,
start = 35,
- stop = 122)))
+ stop = 125)))
+ }
+
+ // SPARK-59527 reproduces with ANSI on and off, so both are checked.
+ Seq(true, false).foreach { ansi =>
+ test(s"MATCH_CONDITION coerces DATE vs STRING like the comparison operator
(ansi=$ansi)") {
+ withSQLConf(SQLConf.ANSI_ENABLED.key -> ansi.toString) {
+ // `l.d >= r.s` type-checks: the string is coerced to DATE, the same
as a bare `>=`
+ // comparison. Before SPARK-59527 this failed with INVALID_TYPE.
+ val sqlText =
+ """
+ |SELECT l.d, r.s
+ |FROM VALUES (DATE '2024-01-03') AS l(d) ASOF JOIN
+ | VALUES ('2024-01-01') AS r(s)
+ | MATCH_CONDITION (l.d >= r.s)
+ |""".stripMargin
+ val asOfJoin = sql(sqlText).queryExecution.analyzed.collectFirst {
+ case j: AsOfJoin => j
+ }.get
Review Comment:
Using `.get` here will fail with a `NoSuchElementException` if the analyzed
plan shape changes, which makes failures harder to diagnose. Consider using
`getOrElse(fail(...))` (or an equivalent assertion) to produce a clearer test
failure message indicating that an `AsOfJoin` node was not found.
--
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]
