This is an automated email from the ASF dual-hosted git repository.
Yicong-Huang pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/master by this push:
new 43826cde7f7f [SPARK-58224][SQL] ASOF JOIN analysis fixes found during
FVT review
43826cde7f7f is described below
commit 43826cde7f7f5c57c335932d4d89d3bcb1989b76
Author: Serge Rielau <[email protected]>
AuthorDate: Mon Jul 20 22:29:54 2026 +0000
[SPARK-58224][SQL] ASOF JOIN analysis fixes found during FVT review
### What changes were proposed in this pull request?
Product fixes uncovered while landing ASOF JOIN golden-file tests
(SPARK-58173):
1. **Non-boolean `ON`** — report `JOIN_CONDITION_IS_NOT_BOOLEAN_TYPE`
(`CheckAnalysis`)
2. **Invalid `MATCH_CONDITION` operator errors** — recover SQL operator
text (`!=` vs `<>`, `IS [NOT] DISTINCT FROM`) via parse-tree inspection instead
of Catalyst pretty names (`AstBuilder`)
3. **Foldable `MATCH_CONDITION` operands** — assign literals,
`CURRENT_TIMESTAMP()`, and session variables to join sides by syntactic
position (`AsOfJoin.normalizeMatchOperands`)
4. **Subquery resolution** — resolve subqueries in `AsOfJoin` `ON` /
`MATCH_CONDITION` like regular `Join` (`Analyzer.ResolveSubquery`,
`ValidateSubqueryExpression`)
5. **Struct/array `MATCH_CONDITION` analysis** — reject empty structs;
require identical array element types so analysis matches order-expression
construction (`ResolveAsOfJoin`)
### Why are the changes needed?
Without these fixes, FVT and integration tests surfaced misleading errors
(e.g. `TABLE_OR_VIEW_NOT_FOUND` for unresolved subqueries) or
analysis/execution mismatches on array operands.
### Does this PR introduce _any_ user-facing change?
Yes — clearer analysis errors and correct behavior for foldable operands
and subqueries in ASOF JOIN.
### How was this patch tested?
- `PlanParserSuite` — invalid `MATCH_CONDITION` operator text
- `AsOfJoinSQLSuite` — foldable operands, subqueries, struct/array operands
```bash
build/sbt -Dscalastyle.skip=true \
"sql/testOnly org.apache.spark.sql.catalyst.parser.PlanParserSuite -- -z
asof" \
"sql/testOnly org.apache.spark.sql.AsOfJoinSQLSuite"
```
Stacked PR: [SPARK-58173](https://github.com/apache/spark/pull/57306) (FVT
goldens) builds on this branch.
### Was this patch authored or co-authored using generative AI tooling?
No.
Closes #57380 from srielau/SPARK-58224.
Authored-by: Serge Rielau <[email protected]>
Signed-off-by: Yicong-Huang <[email protected]>
---
.../spark/sql/catalyst/analysis/Analyzer.scala | 2 +
.../sql/catalyst/analysis/CheckAnalysis.scala | 10 +-
.../sql/catalyst/analysis/ResolveAsOfJoin.scala | 43 +-----
.../analysis/ValidateSubqueryExpression.scala | 2 +-
.../spark/sql/catalyst/parser/AstBuilder.scala | 71 ++++++++-
.../plans/logical/basicLogicalOperators.scala | 148 ++++++++++++++----
.../sql/catalyst/parser/PlanParserSuite.scala | 38 +++++
.../logical/AsOfJoinMatchConditionTypesSuite.scala | 120 ++++++++++++++
.../org/apache/spark/sql/AsOfJoinSQLSuite.scala | 172 ++++++++++++++++++++-
9 files changed, 523 insertions(+), 83 deletions(-)
diff --git
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala
index 51882a9dd983..f8e79f9d8da9 100644
---
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala
+++
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala
@@ -2633,6 +2633,8 @@ class Analyzer(
resolveSubQueries(r, r)
case j: Join if j.childrenResolved && j.duplicateResolved =>
resolveSubQueries(j, j)
+ case j: AsOfJoin if j.childrenResolved && j.duplicateResolved =>
+ resolveSubQueries(j, j)
case tvf: UnresolvedTableValuedFunction =>
resolveSubQueries(tvf, tvf)
case s: SupportsSubquery if s.childrenResolved =>
diff --git
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/CheckAnalysis.scala
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/CheckAnalysis.scala
index 86e789b0b2b1..1d78d455084a 100644
---
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/CheckAnalysis.scala
+++
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/CheckAnalysis.scala
@@ -678,11 +678,11 @@ trait CheckAnalysis extends LookupCatalog with
QueryErrorsBase with PlanToString
case j @ AsOfJoin(_, _, _, Some(condition), _, _, _, _, _, _, _, _,
_, _)
if condition.dataType != BooleanType =>
- throw SparkException.internalError(
- msg = s"join condition '${toSQLExpr(condition)}' " +
- s"of type ${toSQLType(condition.dataType)} is not a boolean.",
- context = j.origin.getQueryContext,
- summary = j.origin.context.summary)
+ j.failAnalysis(
+ errorClass = "JOIN_CONDITION_IS_NOT_BOOLEAN_TYPE",
+ messageParameters = Map(
+ "joinCondition" -> toSQLExpr(condition),
+ "conditionType" -> toSQLType(condition.dataType)))
case j @ AsOfJoin(_, _, _, _, _, _, Some(toleranceAssertion), _, _,
_, _, _, _, _) =>
if (!toleranceAssertion.foldable) {
diff --git
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolveAsOfJoin.scala
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolveAsOfJoin.scala
index fb3b5a425e78..f09612d5902b 100644
---
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolveAsOfJoin.scala
+++
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolveAsOfJoin.scala
@@ -20,18 +20,17 @@ package org.apache.spark.sql.catalyst.analysis
import org.apache.spark.sql.catalyst.SQLConfHelper
import org.apache.spark.sql.catalyst.expressions.{
Expression,
- RowOrdering,
SubqueryExpression,
WindowExpression
}
import org.apache.spark.sql.catalyst.expressions.AttributeSet
import org.apache.spark.sql.catalyst.expressions.aggregate.AggregateExpression
import org.apache.spark.sql.catalyst.plans.logical.{AsOfJoin, LogicalPlan,
Project}
+import org.apache.spark.sql.catalyst.plans.logical.AsOfJoin.MatchConditionTypes
import org.apache.spark.sql.catalyst.rules.Rule
import org.apache.spark.sql.catalyst.trees.TreePattern.{AS_OF_JOIN, GENERATOR}
import org.apache.spark.sql.catalyst.util._
import org.apache.spark.sql.errors.QueryErrorsBase
-import org.apache.spark.sql.types.{ArrayType, DataType, DatetimeType,
StringType, StructType}
/**
* Resolves SQL [[AsOfJoin]] operators: materializes `MATCH_CONDITION` into
`asOfCondition` and
@@ -147,9 +146,9 @@ private[analysis] object AsOfJoinValidation extends
QueryErrorsBase {
messageParameters = Map("expr" -> toSQLExpr(invalidExpr)))
}
}
- if (!RowOrdering.isOrderable(leftExpr.dataType) ||
- !RowOrdering.isOrderable(rightExpr.dataType) ||
- !areMatchConditionTypesCompatible(leftExpr.dataType,
rightExpr.dataType)) {
+ if (!MatchConditionTypes.isValidOperandType(leftExpr.dataType) ||
+ !MatchConditionTypes.isValidOperandType(rightExpr.dataType) ||
+ !MatchConditionTypes.areOperandsCompatible(leftExpr.dataType,
rightExpr.dataType)) {
join.failAnalysis(
errorClass = "ASOF_JOIN_MATCH_CONDITION_INVALID_TYPE",
messageParameters = Map(
@@ -158,40 +157,6 @@ private[analysis] object AsOfJoinValidation extends
QueryErrorsBase {
}
}
- /**
- * Tuple/struct operands may use different field names on each side; compare
field-wise by
- * position when [[TypeCoercion.findWiderTypeForTwo]] does not apply.
- */
- private def areMatchConditionTypesCompatible(t1: DataType, t2: DataType):
Boolean = {
- if (isIncompatibleMatchConditionPair(t1, t2)) {
- false
- } else {
- TypeCoercion.findWiderTypeForTwo(t1, t2).isDefined ||
- areStructurallyComparableTypes(t1, t2)
- }
- }
-
- private def isIncompatibleMatchConditionPair(t1: DataType, t2: DataType):
Boolean = {
- def isString(dt: DataType): Boolean = dt.isInstanceOf[StringType]
- def isTemporal(dt: DataType): Boolean = dt.isInstanceOf[DatetimeType]
- (isTemporal(t1) && isString(t2)) || (isString(t1) && isTemporal(t2))
- }
-
- private def areStructurallyComparableTypes(t1: DataType, t2: DataType):
Boolean = {
- (t1, t2) match {
- case (s1: StructType, s2: StructType) if s1.length == s2.length =>
- s1.zip(s2).forall { case (f1, f2) =>
- RowOrdering.isOrderable(f1.dataType) &&
- RowOrdering.isOrderable(f2.dataType) &&
- areMatchConditionTypesCompatible(f1.dataType, f2.dataType)
- }
- case (ArrayType(e1, _), ArrayType(e2, _)) =>
- RowOrdering.isOrderable(e1) && RowOrdering.isOrderable(e2) &&
- areMatchConditionTypesCompatible(e1, e2)
- case _ => false
- }
- }
-
private def findInvalidMatchConditionExpression(expr: Expression):
Option[Expression] = {
expr.collect {
case e: SubqueryExpression => e
diff --git
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/ValidateSubqueryExpression.scala
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/ValidateSubqueryExpression.scala
index f88e2d112b55..7f41a980a382 100644
---
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/ValidateSubqueryExpression.scala
+++
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/ValidateSubqueryExpression.scala
@@ -225,7 +225,7 @@ object ValidateSubqueryExpression
case inSubqueryOrExistsSubquery =>
plan match {
- case _: Filter | _: SupportsSubquery | _: Join |
+ case _: Filter | _: SupportsSubquery | _: Join | _: AsOfJoin |
_: Project | _: Aggregate | _: Window => // Ok
case _ =>
expr.failAnalysis(
diff --git
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/parser/AstBuilder.scala
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/parser/AstBuilder.scala
index fe3e7dd25d33..0d3f678da456 100644
---
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/parser/AstBuilder.scala
+++
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/parser/AstBuilder.scala
@@ -2562,17 +2562,72 @@ class AstBuilder extends DataTypeAstBuilder
case LessThan(left, right) => (left, LessThanOp, right)
case _ =>
throw QueryParsingErrors.sqlAsOfJoinMatchConditionInvalidOperator(
- asOfMatchConditionInvalidOperatorText(expr), ctx)
+ asOfMatchConditionInvalidOperatorText(expr, ctx), ctx)
}
}
- private def asOfMatchConditionInvalidOperatorText(expr: Expression): String
= expr match {
- case EqualTo(_, _) => "="
- case Not(EqualTo(_, _)) => "<>"
- case EqualNullSafe(_, _) => "<=>"
- case And(_, _) => "AND"
- case Or(_, _) => "OR"
- case _ => expr.prettyName
+ /**
+ * Map a rejected MATCH_CONDITION expression back to the SQL operator the
user wrote.
+ *
+ * Catalyst has no `NotEqualTo` class: both `<>` and `!=` parse to
`Not(EqualTo(...))`, and
+ * both `<=>` and `IS NOT DISTINCT FROM` parse to `EqualNullSafe(...)`. Walk
the match
+ * expression parse tree (not `prettyName`) to recover the token the user
wrote.
+ */
+ private def asOfMatchConditionInvalidOperatorText(
+ expr: Expression,
+ ctx: ParserRuleContext): String = {
+ expr match {
+ case And(_, _) => "AND"
+ case Or(_, _) => "OR"
+ case _ =>
+ findFirstComparisonContext(ctx)
+ .map(comparisonOperatorText)
+ .orElse(findPredicatedContext(ctx).flatMap(distinctFromOperatorText))
+ .getOrElse(getOriginalText(ctx).trim)
+ }
+ }
+
+ private def findFirstComparisonContext(ctx: ParserRuleContext):
Option[ComparisonContext] = {
+ ctx match {
+ case comparison: ComparisonContext => Some(comparison)
+ case _ =>
+ Option(ctx.children).iterator.flatMap(_.asScala).collectFirst {
+ case child: ParserRuleContext => findFirstComparisonContext(child)
+ }.flatten
+ }
+ }
+
+ private def findPredicatedContext(ctx: ParserRuleContext):
Option[PredicatedContext] = {
+ ctx match {
+ case predicated: PredicatedContext => Some(predicated)
+ case _ =>
+ Option(ctx.children).iterator.flatMap(_.asScala).collectFirst {
+ case child: ParserRuleContext => findPredicatedContext(child)
+ }.flatten
+ }
+ }
+
+ private def comparisonOperatorText(ctx: ComparisonContext): String = {
+ val operator =
ctx.comparisonOperator().getChild(0).asInstanceOf[TerminalNode]
+ operator.getSymbol.getType match {
+ case SqlBaseParser.EQ => "="
+ case SqlBaseParser.NSEQ => "<=>"
+ case SqlBaseParser.NEQ => "<>"
+ case SqlBaseParser.NEQJ => "!="
+ case SqlBaseParser.LT => "<"
+ case SqlBaseParser.LTE => "<="
+ case SqlBaseParser.GT => ">"
+ case SqlBaseParser.GTE => ">="
+ case _ => source(ctx.comparisonOperator())
+ }
+ }
+
+ private def distinctFromOperatorText(predicated: PredicatedContext):
Option[String] = {
+ Option(predicated.predicate).flatMap { predicate =>
+ Option(predicate.kind).filter(_.getType == SqlBaseParser.DISTINCT).map {
_ =>
+ if (predicate.errorCapturingNot != null) "IS NOT DISTINCT FROM" else
"IS DISTINCT FROM"
+ }
+ }
}
/**
diff --git
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/basicLogicalOperators.scala
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/basicLogicalOperators.scala
index c2e42df00b68..3cff953cbbca 100644
---
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/basicLogicalOperators.scala
+++
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/basicLogicalOperators.scala
@@ -2639,19 +2639,103 @@ object AsOfJoin {
(asOfCondition, orderExpression, leftSortExprs, rightSortExprs)
}
+ /**
+ * Shared MATCH_CONDITION operand type rules used by analysis validation and
by expression
+ * materialization so the two paths cannot drift.
+ */
+ private[catalyst] object MatchConditionTypes {
+
+ def isValidOperandType(dataType: DataType): Boolean =
+ RowOrdering.isOrderable(dataType) && !containsEmptyStructType(dataType)
+
+ def areOperandsCompatible(leftType: DataType, rightType: DataType):
Boolean = {
+ if (!isValidOperandType(leftType) || !isValidOperandType(rightType)) {
+ false
+ } else if (isStringTemporalMismatch(leftType, rightType)) {
+ false
+ } else {
+ (leftType, rightType) match {
+ case (ArrayType(_, _), ArrayType(_, _)) =>
+ usesArrayOrderExpression(leftType, rightType)
+ case _ =>
+ TypeCoercion.findWiderTypeForTwo(leftType, rightType).isDefined ||
+ arePositionalStructsCompatible(leftType, rightType)
+ }
+ }
+ }
+
+ def usesArrayOrderExpression(leftType: DataType, rightType: DataType):
Boolean =
+ (leftType, rightType) match {
+ case (ArrayType(leftElem, _), ArrayType(rightElem, _)) =>
+ areArrayElementsCompatible(leftElem, rightElem)
+ case _ => false
+ }
+
+ private def areArrayElementsCompatible(leftElem: DataType, rightElem:
DataType): Boolean = {
+ if (DataTypeUtils.sameType(leftElem, rightElem)) {
+ RowOrdering.isOrderable(leftElem)
+ } else {
+ arePositionalStructsCompatible(leftElem, rightElem)
+ }
+ }
+
+ /** Positional struct operands with the same field count (names may
differ). */
+ def usesStructDecomposition(leftType: DataType, rightType: DataType):
Boolean =
+ (leftType, rightType) match {
+ case (leftStruct: StructType, rightStruct: StructType) =>
+ leftStruct.length == rightStruct.length && leftStruct.nonEmpty
+ case _ => false
+ }
+
+ /** Whole struct columns with identical schemas sort as a single struct
value. */
+ def usesIdenticalStructSort(leftType: DataType, rightType: DataType):
Boolean =
+ (leftType, rightType) match {
+ case (leftStruct: StructType, rightStruct: StructType) =>
+ leftStruct.sameType(rightStruct) && leftStruct.nonEmpty
+ case _ => false
+ }
+
+ private def isStringTemporalMismatch(leftType: DataType, rightType:
DataType): Boolean = {
+ def isString(dataType: DataType): Boolean =
dataType.isInstanceOf[StringType]
+ def isTemporal(dataType: DataType): Boolean =
dataType.isInstanceOf[DatetimeType]
+ (isTemporal(leftType) && isString(rightType)) || (isString(leftType) &&
isTemporal(rightType))
+ }
+
+ private def arePositionalStructsCompatible(
+ leftType: DataType,
+ rightType: DataType): Boolean = {
+ (leftType, rightType) match {
+ case (leftStruct: StructType, rightStruct: StructType)
+ if usesStructDecomposition(leftType, rightType) =>
+ leftStruct.zip(rightStruct).forall { case (leftField, rightField) =>
+ areOperandsCompatible(leftField.dataType, rightField.dataType)
+ }
+ case _ => false
+ }
+ }
+
+ private def containsEmptyStructType(dataType: DataType): Boolean =
dataType match {
+ case struct: StructType =>
+ struct.isEmpty || struct.exists(field =>
containsEmptyStructType(field.dataType))
+ case ArrayType(elementType, _) => containsEmptyStructType(elementType)
+ case _ => false
+ }
+ }
+
/**
* Sort-merge ASOF join sorts each side by these expressions (after
equi-keys) so the
* right-side buffer is ordered consistently with MATCH_CONDITION
lexicographic comparison.
*
* SQL tuple literals `(t.a, t.b)` are flattened to scalar leaves. Whole
struct columns
* (`t.k >= r.k`) sort by the struct value directly so nested struct shapes
stay intact.
+ * Array operands sort element-wise; length mismatches follow Spark array
ordering semantics.
*/
def matchSortExpressions(
leftOperand: Expression,
rightOperand: Expression): (Seq[Expression], Seq[Expression]) = {
(leftOperand.dataType, rightOperand.dataType) match {
case (leftStruct: StructType, rightStruct: StructType)
- if leftStruct.sameType(rightStruct) && leftStruct.nonEmpty =>
+ if MatchConditionTypes.usesIdenticalStructSort(leftStruct,
rightStruct) =>
if (isSqlTupleStructOperand(leftOperand) ||
isSqlTupleStructOperand(rightOperand)) {
val pairs = collectStructLeafPairs(leftOperand, rightOperand,
leftStruct)
(pairs.map(_._1), pairs.map(_._2))
@@ -2675,8 +2759,8 @@ object AsOfJoin {
expr2: Expression): (Expression, Expression, MatchComparisonOperator) = {
val leftSet = left.outputSet
val rightSet = right.outputSet
- val expr1Side = operandJoinSide(expr1, leftSet, rightSet)
- val expr2Side = operandJoinSide(expr2, leftSet, rightSet)
+ val expr1Side = operandJoinSide(expr1, leftSet, rightSet, syntacticIsLeft
= true)
+ val expr2Side = operandJoinSide(expr2, leftSet, rightSet, syntacticIsLeft
= false)
(expr1Side, expr2Side) match {
case (Some(true), Some(false)) => (expr1, expr2, operator)
case (Some(false), Some(true)) => (expr2, expr1, operator.flip)
@@ -2688,10 +2772,13 @@ object AsOfJoin {
private def operandJoinSide(
expr: Expression,
leftSet: AttributeSet,
- rightSet: AttributeSet): Option[Boolean] = {
+ rightSet: AttributeSet,
+ syntacticIsLeft: Boolean): Option[Boolean] = {
val refs = expr.references
if (refs.isEmpty) {
- None
+ // Literals, CURRENT_TIMESTAMP(), session variables, etc. have no column
refs;
+ // use MATCH_CONDITION syntactic position (expr1/expr2) for join-side
assignment.
+ Some(syntacticIsLeft)
} else if (refs.subsetOf(leftSet)) {
Some(true)
} else if (refs.subsetOf(rightSet)) {
@@ -2725,13 +2812,17 @@ object AsOfJoin {
rightOperand: Expression,
operator: MatchComparisonOperator): Expression = {
(leftOperand.dataType, rightOperand.dataType) match {
- case (ArrayType(leftElem, _), ArrayType(rightElem, _))
- if DataTypeUtils.sameType(leftElem, rightElem) =>
- buildArrayOrderExpression(leftOperand, rightOperand, leftElem,
operator)
- case (leftStruct: StructType, rightStruct: StructType)
- if leftStruct.length == rightStruct.length && leftStruct.nonEmpty =>
+ case (ArrayType(elementType, _), _)
+ 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, elementType,
operator)
+ case (leftType, rightType)
+ if MatchConditionTypes.usesStructDecomposition(leftType, rightType)
=>
buildFlattenedStructOrderExpression(
- leftOperand, rightOperand, leftStruct, operator)
+ leftOperand, rightOperand, leftType.asInstanceOf[StructType],
operator)
case _ =>
buildLeafOrderExpression(leftOperand, rightOperand, operator)
}
@@ -2844,12 +2935,11 @@ object AsOfJoin {
.zip(structFieldExprs(rightOperand, structType))
.flatMap {
case (left, right) =>
- (left.dataType, right.dataType) match {
- case (leftStruct: StructType, rightStruct: StructType)
- if leftStruct.length == rightStruct.length &&
leftStruct.nonEmpty =>
- collectStructLeafPairs(left, right, leftStruct)
- case _ =>
- Seq((left, right))
+ if (MatchConditionTypes.usesStructDecomposition(left.dataType,
right.dataType)) {
+ collectStructLeafPairs(
+ left, right, left.dataType.asInstanceOf[StructType])
+ } else {
+ Seq((left, right))
}
}
}
@@ -2871,17 +2961,19 @@ object AsOfJoin {
private def decomposeStructOperands(
leftOperand: Expression,
rightOperand: Expression): Option[Seq[(Expression, Expression)]] = {
- (leftOperand.dataType, rightOperand.dataType) match {
- case (leftStruct: StructType, rightStruct: StructType)
- if leftStruct.length == rightStruct.length && leftStruct.nonEmpty =>
- val leftFields = structFieldExprs(leftOperand, leftStruct)
- val rightFields = structFieldExprs(rightOperand, rightStruct)
- if (leftFields.length == rightFields.length) {
- Some(leftFields.zip(rightFields))
- } else {
- None
- }
- case _ => None
+ if (MatchConditionTypes.usesStructDecomposition(
+ leftOperand.dataType, rightOperand.dataType)) {
+ val leftStruct = leftOperand.dataType.asInstanceOf[StructType]
+ val rightStruct = rightOperand.dataType.asInstanceOf[StructType]
+ val leftFields = structFieldExprs(leftOperand, leftStruct)
+ val rightFields = structFieldExprs(rightOperand, rightStruct)
+ if (leftFields.length == rightFields.length) {
+ Some(leftFields.zip(rightFields))
+ } else {
+ None
+ }
+ } else {
+ None
}
}
diff --git
a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/parser/PlanParserSuite.scala
b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/parser/PlanParserSuite.scala
index 278c9a1bc1e4..2d8e60aee22c 100644
---
a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/parser/PlanParserSuite.scala
+++
b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/parser/PlanParserSuite.scala
@@ -1044,6 +1044,17 @@ class PlanParserSuite extends AnalysisTest {
fragment = "asof join u match_condition (t.a <> u.a)",
start = 16,
stop = 55)))
+ checkError(
+ exception = parseException(
+ "select * from t asof join u match_condition (t.a != u.a)"),
+ condition = "ASOF_JOIN_MATCH_CONDITION_INVALID_OPERATOR",
+ sqlState = Some("42K0E"),
+ parameters = Map("operator" -> "!="),
+ queryContext = Array(
+ ExpectedContext(
+ fragment = "asof join u match_condition (t.a != u.a)",
+ start = 16,
+ stop = 55)))
}
}
@@ -1060,6 +1071,33 @@ class PlanParserSuite extends AnalysisTest {
fragment = "asof join u match_condition (t.a <=> u.a)",
start = 16,
stop = 56)))
+ checkError(
+ exception = parseException(
+ "select * from t asof join u match_condition (t.a is not distinct
from u.a)"),
+ condition = "ASOF_JOIN_MATCH_CONDITION_INVALID_OPERATOR",
+ sqlState = Some("42K0E"),
+ parameters = Map("operator" -> "IS NOT DISTINCT FROM"),
+ queryContext = Array(
+ ExpectedContext(
+ fragment = "asof join u match_condition (t.a is not distinct from
u.a)",
+ start = 16,
+ stop = 73)))
+ }
+ }
+
+ test("asof join - is distinct from match operator") {
+ withSQLConf(SQLConf.SQL_ASOF_JOIN_ENABLED.key -> "true") {
+ checkError(
+ exception = parseException(
+ "select * from t asof join u match_condition (t.a is distinct from
u.a)"),
+ condition = "ASOF_JOIN_MATCH_CONDITION_INVALID_OPERATOR",
+ sqlState = Some("42K0E"),
+ parameters = Map("operator" -> "IS DISTINCT FROM"),
+ queryContext = Array(
+ ExpectedContext(
+ fragment = "asof join u match_condition (t.a is distinct from
u.a)",
+ start = 16,
+ stop = 69)))
}
}
diff --git
a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/plans/logical/AsOfJoinMatchConditionTypesSuite.scala
b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/plans/logical/AsOfJoinMatchConditionTypesSuite.scala
new file mode 100644
index 000000000000..048eac353df9
--- /dev/null
+++
b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/plans/logical/AsOfJoinMatchConditionTypesSuite.scala
@@ -0,0 +1,120 @@
+/*
+ * 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.SparkFunSuite
+import org.apache.spark.sql.catalyst.plans.logical.AsOfJoin.MatchConditionTypes
+import org.apache.spark.sql.types._
+
+class AsOfJoinMatchConditionTypesSuite extends SparkFunSuite {
+
+ test("scalar types widen via TypeCoercion") {
+ assert(MatchConditionTypes.areOperandsCompatible(IntegerType, LongType))
+ assert(!MatchConditionTypes.usesArrayOrderExpression(IntegerType,
LongType))
+ assert(!MatchConditionTypes.usesStructDecomposition(IntegerType, LongType))
+ }
+
+ test("string and temporal types are incompatible") {
+ assert(!MatchConditionTypes.areOperandsCompatible(StringType,
TimestampType))
+ assert(!MatchConditionTypes.areOperandsCompatible(DateType, StringType))
+ }
+
+ test("positional struct operands with different field names are compatible")
{
+ val leftStruct = StructType(
+ StructField("a", IntegerType) ::
+ StructField("b", LongType) ::
+ Nil)
+ val rightStruct = StructType(
+ StructField("x", IntegerType) ::
+ StructField("y", LongType) ::
+ Nil)
+ assert(MatchConditionTypes.areOperandsCompatible(leftStruct, rightStruct))
+ assert(MatchConditionTypes.usesStructDecomposition(leftStruct,
rightStruct))
+ assert(!MatchConditionTypes.usesIdenticalStructSort(leftStruct,
rightStruct))
+ }
+
+ test("nested struct operands are compatible when fields match positionally")
{
+ val innerLeft = StructType(StructField("k", IntegerType) :: Nil)
+ val innerRight = StructType(StructField("z", IntegerType) :: Nil)
+ val leftStruct = StructType(StructField("outer", innerLeft) :: Nil)
+ val rightStruct = StructType(StructField("other", innerRight) :: Nil)
+ assert(MatchConditionTypes.areOperandsCompatible(leftStruct, rightStruct))
+ assert(MatchConditionTypes.usesStructDecomposition(leftStruct,
rightStruct))
+ }
+
+ test("array operands require identical element types") {
+ val leftArray = ArrayType(IntegerType)
+ val rightArray = ArrayType(IntegerType)
+ assert(MatchConditionTypes.areOperandsCompatible(leftArray, rightArray))
+ assert(MatchConditionTypes.usesArrayOrderExpression(leftArray, rightArray))
+ }
+
+ test("array operands with different struct element field names are
compatible") {
+ val leftArray = ArrayType(
+ StructType(
+ StructField("x", IntegerType, nullable = false) ::
+ StructField("y", IntegerType, nullable = false) ::
+ Nil))
+ val rightArray = ArrayType(
+ StructType(
+ StructField("p", IntegerType, nullable = false) ::
+ StructField("q", IntegerType, nullable = false) ::
+ Nil))
+ assert(MatchConditionTypes.areOperandsCompatible(leftArray, rightArray))
+ assert(MatchConditionTypes.usesArrayOrderExpression(leftArray, rightArray))
+ }
+
+ test("array operands with different element types are rejected") {
+ val leftArray = ArrayType(IntegerType)
+ val rightArray = ArrayType(StringType)
+ assert(!MatchConditionTypes.areOperandsCompatible(leftArray, rightArray))
+ assert(!MatchConditionTypes.usesArrayOrderExpression(leftArray,
rightArray))
+ }
+
+ test("empty struct operands are invalid") {
+ val emptyStruct = StructType(Nil)
+ assert(!MatchConditionTypes.isValidOperandType(emptyStruct))
+ assert(!MatchConditionTypes.areOperandsCompatible(emptyStruct,
emptyStruct))
+ assert(!MatchConditionTypes.usesStructDecomposition(emptyStruct,
emptyStruct))
+ }
+
+ test("nested empty struct operands are invalid") {
+ val nestedEmptyStruct = StructType(StructField("x", StructType(Nil)) ::
Nil)
+ assert(!MatchConditionTypes.isValidOperandType(nestedEmptyStruct))
+ assert(!MatchConditionTypes.areOperandsCompatible(nestedEmptyStruct,
nestedEmptyStruct))
+ }
+
+ test("identical struct schemas enable whole-struct sort") {
+ val structType = StructType(
+ StructField("ts", TimestampType) ::
+ StructField("seq", IntegerType) ::
+ Nil)
+ assert(MatchConditionTypes.usesIdenticalStructSort(structType, structType))
+ assert(MatchConditionTypes.usesStructDecomposition(structType, structType))
+ }
+
+ test("struct field count mismatch is not decomposable") {
+ val leftStruct = StructType(StructField("a", IntegerType) :: Nil)
+ val rightStruct = StructType(
+ StructField("a", IntegerType) ::
+ StructField("b", IntegerType) ::
+ Nil)
+ assert(!MatchConditionTypes.usesStructDecomposition(leftStruct,
rightStruct))
+ assert(!MatchConditionTypes.areOperandsCompatible(leftStruct, rightStruct))
+ }
+}
diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/AsOfJoinSQLSuite.scala
b/sql/core/src/test/scala/org/apache/spark/sql/AsOfJoinSQLSuite.scala
index e75ebbe04a04..36962df5af45 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/AsOfJoinSQLSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/AsOfJoinSQLSuite.scala
@@ -49,8 +49,8 @@ class AsOfJoinSQLSuite extends QueryTest with
SharedSparkSession {
|""".stripMargin)
sql(
"""
- |CREATE OR REPLACE TEMP VIEW quotes(quote_time, symbol) AS
- |VALUES (TIMESTAMP '2026-06-29 10:00:00', 'AAPL')
+ |CREATE OR REPLACE TEMP VIEW quotes(quote_time, symbol, bid_price) AS
+ |VALUES (TIMESTAMP '2026-06-29 10:00:00', 'AAPL', 180.10)
|""".stripMargin)
}
@@ -209,4 +209,172 @@ class AsOfJoinSQLSuite extends QueryTest with
SharedSparkSession {
start = 35,
stop = 122)))
}
+
+ test("MATCH_CONDITION accepts CURRENT_TIMESTAMP as left operand") {
+ setupTradeQuoteViews()
+ val sqlText =
+ """
+ |SELECT count(*)
+ |FROM trades t ASOF JOIN quotes q
+ | MATCH_CONDITION (current_timestamp() >= q.quote_time)
+ | ON t.symbol = q.symbol
+ |""".stripMargin
+ val asOfJoin = sql(sqlText).queryExecution.analyzed.collectFirst {
+ case j: AsOfJoin => j
+ }.get
+ assert(asOfJoin.asOfCondition.resolved)
+ assert(asOfJoin.leftSortExprs.nonEmpty)
+ assert(asOfJoin.rightSortExprs.nonEmpty)
+ }
+
+ test("MATCH_CONDITION accepts literal constant as right operand") {
+ setupTradeQuoteViews()
+ val sqlText =
+ """
+ |SELECT count(*)
+ |FROM trades t ASOF JOIN quotes q
+ | MATCH_CONDITION (t.trade_time >= TIMESTAMP '2026-06-29 10:00:00')
+ | ON t.symbol = q.symbol
+ |""".stripMargin
+ val asOfJoin = sql(sqlText).queryExecution.analyzed.collectFirst {
+ case j: AsOfJoin => j
+ }.get
+ assert(asOfJoin.asOfCondition.resolved)
+ }
+
+ test("MATCH_CONDITION rejects scalar subquery operand") {
+ setupTradeQuoteViews()
+ val sqlText =
+ """
+ |SELECT * FROM trades t ASOF JOIN quotes q
+ | MATCH_CONDITION (t.trade_time >= (SELECT max(trade_time) FROM
trades))
+ | ON t.symbol = q.symbol
+ |""".stripMargin
+ checkError(
+ exception = intercept[AnalysisException](sql(sqlText)),
+ condition = "ASOF_JOIN_MATCH_CONDITION_INVALID_EXPRESSION",
+ sqlState = Some("42903"),
+ parameters = Map("expr" -> "\"scalarsubquery()\""),
+ queryContext = Array(
+ ExpectedContext(
+ fragment = """ASOF JOIN quotes q
+ | MATCH_CONDITION (t.trade_time >= (SELECT
max(trade_time) FROM trades))
+ | ON t.symbol = q.symbol""".stripMargin,
+ start = 24,
+ stop = 139)))
+ }
+
+ test("uncorrelated scalar subquery in ON clause is allowed") {
+ setupTradeQuoteViews()
+ checkAnswer(
+ sql(
+ """
+ |SELECT t.symbol
+ |FROM trades t ASOF JOIN quotes q
+ | MATCH_CONDITION (t.trade_time >= q.quote_time)
+ | ON t.symbol = q.symbol
+ | AND t.trade_time >= (SELECT max(trade_time) FROM trades)
+ |""".stripMargin),
+ Row("AAPL"))
+ }
+
+ test("correlated scalar subquery in ON clause is rejected like regular
JOIN") {
+ setupTradeQuoteViews()
+ val sqlText =
+ """
+ |SELECT t.symbol, q.bid_price
+ |FROM trades t ASOF JOIN quotes q
+ | MATCH_CONDITION (t.trade_time >= q.quote_time)
+ | ON t.symbol = q.symbol
+ | AND q.bid_price > (
+ | SELECT avg(bid_price) FROM quotes qq WHERE qq.symbol = q.symbol)
+ |""".stripMargin
+ val e = intercept[AnalysisException](sql(sqlText))
+ assert(
+ e.getCondition ===
+
"UNSUPPORTED_SUBQUERY_EXPRESSION_CATEGORY.UNSUPPORTED_CORRELATED_SCALAR_SUBQUERY")
+ }
+
+ test("MATCH_CONDITION accepts STRUCT tuple operands") {
+ sql(
+ """
+ |CREATE OR REPLACE TEMP VIEW deploys(deploy_ts, seq, service, version)
AS
+ |VALUES (TIMESTAMP '2026-06-29 10:00:00', 1, 'api', 'v1.0')
+ |""".stripMargin)
+ sql(
+ """
+ |CREATE OR REPLACE TEMP VIEW requests(req_ts, seq, service) AS
+ |VALUES (TIMESTAMP '2026-06-29 10:03:00', 1, 'api')
+ |""".stripMargin)
+ val sqlText =
+ """
+ |SELECT r.req_ts, d.version
+ |FROM requests r ASOF JOIN deploys d
+ | MATCH_CONDITION ((r.req_ts, r.seq) >= (d.deploy_ts, d.seq))
+ | ON r.service = d.service
+ |""".stripMargin
+ val asOfJoin = sql(sqlText).queryExecution.analyzed.collectFirst {
+ case j: AsOfJoin => j
+ }.get
+ assert(asOfJoin.asOfCondition.resolved)
+ assert(asOfJoin.leftSortExprs.nonEmpty)
+ assert(asOfJoin.rightSortExprs.nonEmpty)
+ }
+
+ test("MATCH_CONDITION accepts ARRAY operands") {
+ val sqlText =
+ """
+ |SELECT t.a
+ |FROM VALUES (ARRAY(1, 3)) AS t(a) ASOF JOIN VALUES (ARRAY(1, 2)) AS
r(a)
+ | MATCH_CONDITION (t.a >= r.a)
+ |""".stripMargin
+ val asOfJoin = sql(sqlText).queryExecution.analyzed.collectFirst {
+ case j: AsOfJoin => j
+ }.get
+ assert(asOfJoin.asOfCondition.resolved)
+ assert(asOfJoin.leftSortExprs.nonEmpty)
+ assert(asOfJoin.rightSortExprs.nonEmpty)
+ }
+
+ test("MATCH_CONDITION accepts nested STRUCT column operands") {
+ val sqlText =
+ """
+ |SELECT r.k.tag
+ |FROM VALUES (named_struct(
+ | 'inner_ts', TIMESTAMP '2026-06-29 10:00:00',
+ | 'inner_seq', 2,
+ | 'tag', 'a')) AS t(k)
+ |ASOF JOIN VALUES (named_struct(
+ | 'inner_ts', TIMESTAMP '2026-06-29 09:00:00',
+ | 'inner_seq', 1,
+ | 'tag', 'a')) AS r(k)
+ | MATCH_CONDITION (t.k >= r.k)
+ |""".stripMargin
+ val asOfJoin = sql(sqlText).queryExecution.analyzed.collectFirst {
+ case j: AsOfJoin => j
+ }.get
+ assert(asOfJoin.asOfCondition.resolved)
+ }
+
+ test("MATCH_CONDITION rejects empty STRUCT operands") {
+ val sqlText =
+ """
+ |SELECT *
+ |FROM VALUES (named_struct()) AS t(s) ASOF JOIN VALUES
(named_struct()) AS r(s)
+ | MATCH_CONDITION (t.s >= r.s)
+ |""".stripMargin
+ checkError(
+ exception = intercept[AnalysisException](sql(sqlText)),
+ condition = "ASOF_JOIN_MATCH_CONDITION_INVALID_TYPE",
+ sqlState = Some("42K09"),
+ parameters = Map(
+ "type1" -> "\"STRUCT<>\"",
+ "type2" -> "\"STRUCT<>\""),
+ queryContext = Array(
+ ExpectedContext(
+ fragment = """ASOF JOIN VALUES (named_struct()) AS r(s)
+ | MATCH_CONDITION (t.s >= r.s)""".stripMargin,
+ start = 47,
+ stop = 118)))
+ }
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
