[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](#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 <serge@rielau.com>
Signed-off-by: Yicong-Huang <17627829+Yicong-Huang@users.noreply.github.com>

Author: srielau

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 =>

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) {

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

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(

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"
+      }
+    }
   }
 
   /**

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
     }
   }