diff --git a/common/utils/src/main/resources/error/error-conditions.json b/common/utils/src/main/resources/error/error-conditions.json index f04c5a4d5053f..ffb64afbe9f9e 100644 --- a/common/utils/src/main/resources/error/error-conditions.json +++ b/common/utils/src/main/resources/error/error-conditions.json @@ -156,6 +156,30 @@ ], "sqlState" : "42713" }, + "ASOF_JOIN_MATCH_CONDITION_INVALID_EXPRESSION" : { + "message" : [ + "The MATCH_CONDITION operand contains an invalid expression: . Rewrite the query to avoid subqueries, aggregate functions, window functions, generator functions, or non-deterministic functions in MATCH_CONDITION operands." + ], + "sqlState" : "42903" + }, + "ASOF_JOIN_MATCH_CONDITION_INVALID_OPERATOR" : { + "message" : [ + "The MATCH_CONDITION operator must be one of >=, >, <=, <. Got: ." + ], + "sqlState" : "42K0E" + }, + "ASOF_JOIN_MATCH_CONDITION_INVALID_TYPE" : { + "message" : [ + "The MATCH_CONDITION operands must be of an orderable, mutually comparable type. Got: and ." + ], + "sqlState" : "42K09" + }, + "ASOF_JOIN_MATCH_CONDITION_TABLE_REFERENCE" : { + "message" : [ + "Each operand of the MATCH_CONDITION comparison must reference only columns of one join input. Got: and ." + ], + "sqlState" : "42K0E" + }, "ASSIGNMENT_ARITY_MISMATCH" : { "message" : [ "The number of columns or variables assigned or aliased: does not match the number of source expressions: ." @@ -167,6 +191,11 @@ "Invalid as-of join." ], "subClass" : { + "SORT_MERGE_REQUIRED" : { + "message" : [ + "SQL ASOF JOIN requires the sort-merge physical operator. Set to true." + ] + }, "TOLERANCE_IS_NON_NEGATIVE" : { "message" : [ "The input argument `tolerance` must be non-negative." @@ -8231,6 +8260,11 @@ "The ANALYZE TABLE command does not support views." ] }, + "ASOF_JOIN" : { + "message" : [ + "SQL ASOF JOIN syntax. Set to true to enable it." + ] + }, "BIN_BY" : { "message" : [ "The BIN BY relation operator is not yet supported." diff --git a/docs/sql-ref-ansi-compliance.md b/docs/sql-ref-ansi-compliance.md index c3eec6cc5ce3e..367abce6cb8d3 100644 --- a/docs/sql-ref-ansi-compliance.md +++ b/docs/sql-ref-ansi-compliance.md @@ -430,6 +430,7 @@ Below is a list of all the keywords in Spark SQL. |AS|reserved|non-reserved|reserved| |ASC|non-reserved|non-reserved|non-reserved| |ASENSITIVE|non-reserved|non-reserved|non-reserved| +|ASOF|non-reserved|non-reserved|non-reserved| |AT|non-reserved|non-reserved|reserved| |ATOMIC|non-reserved|non-reserved|non-reserved| |AUTHORIZATION|reserved|non-reserved|reserved| @@ -641,6 +642,7 @@ Below is a list of all the keywords in Spark SQL. |MACRO|non-reserved|non-reserved|non-reserved| |MAP|non-reserved|non-reserved|non-reserved| |MATCHED|non-reserved|non-reserved|non-reserved| +|MATCH_CONDITION|non-reserved|non-reserved|non-reserved| |MATERIALIZED|non-reserved|non-reserved|non-reserved| |MAX|non-reserved|non-reserved|non-reserved| |MEASURE|non-reserved|non-reserved|non-reserved| diff --git a/sql/api/src/main/antlr4/org/apache/spark/sql/catalyst/parser/SqlBaseLexer.g4 b/sql/api/src/main/antlr4/org/apache/spark/sql/catalyst/parser/SqlBaseLexer.g4 index 3e5c79613f198..64a99e82053b1 100644 --- a/sql/api/src/main/antlr4/org/apache/spark/sql/catalyst/parser/SqlBaseLexer.g4 +++ b/sql/api/src/main/antlr4/org/apache/spark/sql/catalyst/parser/SqlBaseLexer.g4 @@ -145,6 +145,7 @@ APPROX: 'APPROX'; ARCHIVE: 'ARCHIVE'; ARRAY: 'ARRAY' {incComplexTypeLevelCounter();}; AS: 'AS'; +ASOF: 'ASOF'; ASC: 'ASC'; ASENSITIVE: 'ASENSITIVE'; AT: 'AT'; @@ -358,6 +359,7 @@ LOOP: 'LOOP'; MACRO: 'MACRO'; MAP: 'MAP' {incComplexTypeLevelCounter();}; MATCHED: 'MATCHED'; +MATCH_CONDITION: 'MATCH_CONDITION'; MATERIALIZED: 'MATERIALIZED'; MAX: 'MAX'; MEASURE: 'MEASURE'; diff --git a/sql/api/src/main/antlr4/org/apache/spark/sql/catalyst/parser/SqlBaseParser.g4 b/sql/api/src/main/antlr4/org/apache/spark/sql/catalyst/parser/SqlBaseParser.g4 index 6b2237f15cc49..53ea9a3ceb329 100644 --- a/sql/api/src/main/antlr4/org/apache/spark/sql/catalyst/parser/SqlBaseParser.g4 +++ b/sql/api/src/main/antlr4/org/apache/spark/sql/catalyst/parser/SqlBaseParser.g4 @@ -75,7 +75,7 @@ options { tokenVocab = SqlBaseLexer; } la == AS || la == WHERE || la == PIVOT || la == UNPIVOT || la == TABLESAMPLE || la == INNER || la == CROSS || la == LEFT || la == RIGHT || la == FULL || la == NATURAL || la == SEMI || - la == ANTI || la == JOIN || la == UNION || la == EXCEPT || + la == ANTI || la == ASOF || la == JOIN || la == UNION || la == EXCEPT || la == SETMINUS || la == INTERSECT || la == ORDER || la == CLUSTER || la == DISTRIBUTE || la == SORT || la == LIMIT || la == OFFSET || la == AGGREGATE || la == WINDOW || la == LATERAL || la == BIN; @@ -1104,8 +1104,24 @@ relationExtension ; joinRelation - : (joinType) JOIN LATERAL? right=relationPrimary (joinCriteria | nearestByClause)? + : (joinType) JOIN LATERAL? right=relationPrimary asofJoinPostfix? | NATURAL joinType JOIN LATERAL? right=relationPrimary + | asofJoinType ASOF JOIN right=relationPrimary asofJoinCriteria + ; + +asofJoinType + : INNER? + | LEFT OUTER? + ; + +asofJoinPostfix + : joinCriteria + | nearestByClause + ; + +asofJoinCriteria + : MATCH_CONDITION LEFT_PAREN matchExpr=booleanExpression RIGHT_PAREN + ( ON onExpr=booleanExpression | USING identifierList )? ; joinType @@ -2013,6 +2029,7 @@ ansiNonReserved | ARRAY | ASC | ASENSITIVE + | ASOF | AT | ATOMIC | AUTO @@ -2180,6 +2197,7 @@ ansiNonReserved | MACRO | MAP | MATCHED + | MATCH_CONDITION | MATERIALIZED | MAX | MEASURE @@ -2404,6 +2422,7 @@ nonReserved | ARRAY | AS | ASC + | ASOF | ASENSITIVE | AT | ATOMIC @@ -2610,6 +2629,7 @@ nonReserved | MACRO | MAP | MATCHED + | MATCH_CONDITION | MATERIALIZED | MAX | MEASURE diff --git a/sql/api/src/main/scala/org/apache/spark/sql/errors/QueryParsingErrors.scala b/sql/api/src/main/scala/org/apache/spark/sql/errors/QueryParsingErrors.scala index f73b0457f87f7..06cbe9aad3e4b 100644 --- a/sql/api/src/main/scala/org/apache/spark/sql/errors/QueryParsingErrors.scala +++ b/sql/api/src/main/scala/org/apache/spark/sql/errors/QueryParsingErrors.scala @@ -902,6 +902,22 @@ private[sql] object QueryParsingErrors extends DataTypeErrorsBase { ctx) } + def sqlAsofJoinDisabled(configKey: String, ctx: ParserRuleContext): Throwable = { + new ParseException( + errorClass = "UNSUPPORTED_FEATURE.ASOF_JOIN", + messageParameters = Map("config" -> toSQLConf(configKey)), + ctx) + } + + def sqlAsOfJoinMatchConditionInvalidOperator( + operator: String, + ctx: ParserRuleContext): Throwable = { + new ParseException( + errorClass = "ASOF_JOIN_MATCH_CONDITION_INVALID_OPERATOR", + messageParameters = Map("operator" -> operator), + ctx) + } + def invalidNameForDropTempFunc(name: Seq[String], ctx: ParserRuleContext): Throwable = { new ParseException( errorClass = "INVALID_SQL_SYNTAX.MULTI_PART_NAME", 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 49fdbd45b91c2..541a83c1a8b72 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 @@ -636,7 +636,8 @@ class Analyzer( typeCoercionRules() ++ Seq( ResolveWithCTE, - ExtractDistributedSequenceID) ++ + ExtractDistributedSequenceID, + ResolveAsOfJoin) ++ Seq(ResolveUpdateEventTimeWatermarkColumn) ++ extendedResolutionRules ++ Seq(NameStreamingSources) : _*), @@ -1931,6 +1932,29 @@ class Analyzer( case d: DataFrameDropColumns if !d.resolved => resolveDataFrameDropColumns(d) + case j @ AsOfJoin(left, right, _, condition, _, _, _, _, matchCmp, _, _, _) + if left.resolved && right.resolved && + (matchCmp.exists(mc => !mc.left.resolved || !mc.right.resolved) || + condition.exists(!_.resolved)) => + var updated = j + matchCmp.foreach { mc => + val resolvedLeft = resolveExpressionByPlanChildren(mc.left, j, includeLastResort = true) + val resolvedRight = resolveExpressionByPlanChildren(mc.right, j, includeLastResort = true) + if (!resolvedLeft.fastEquals(mc.left) || !resolvedRight.fastEquals(mc.right)) { + updated = updated.copy( + matchComparison = Some(AsOfMatchCondition(resolvedLeft, mc.operator, resolvedRight))) + } + } + condition.foreach { cond => + if (!cond.resolved) { + val resolvedCond = resolveExpressionByPlanChildren(cond, j, includeLastResort = true) + if (!resolvedCond.fastEquals(cond)) { + updated = updated.copy(condition = Some(resolvedCond)) + } + } + } + if (updated.fastEquals(j)) j else updated + case q: LogicalPlan => logTrace(s"Attempting to resolve ${q.simpleString(conf.maxToStringFields)}") q.mapExpressions(resolveExpressionByPlanChildren(_, q, includeLastResort = true)) 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 323dc49426123..10438eebc913e 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 @@ -676,7 +676,14 @@ trait CheckAnalysis extends LookupCatalog with QueryErrorsBase with PlanToString "joinCondition" -> toSQLExpr(condition), "conditionType" -> toSQLType(condition.dataType))) - case j @ AsOfJoin(_, _, _, Some(condition), _, _, _) + case j @ AsOfJoin(_, _, _, _, _, _, _, _, _, _, _, true) + if !SQLConf.get.sortMergeAsOfJoinEnabled => + j.failAnalysis( + errorClass = "AS_OF_JOIN.SORT_MERGE_REQUIRED", + messageParameters = Map( + "config" -> SQLConf.SORT_MERGE_AS_OF_JOIN_ENABLED.key)) + + case j @ AsOfJoin(_, _, _, Some(condition), _, _, _, _, _, _, _, _) if condition.dataType != BooleanType => throw SparkException.internalError( msg = s"join condition '${toSQLExpr(condition)}' " + @@ -684,7 +691,7 @@ trait CheckAnalysis extends LookupCatalog with QueryErrorsBase with PlanToString context = j.origin.getQueryContext, summary = j.origin.context.summary) - case j @ AsOfJoin(_, _, _, _, _, _, Some(toleranceAssertion)) => + case j @ AsOfJoin(_, _, _, _, _, _, Some(toleranceAssertion), _, _, _, _, _) => if (!toleranceAssertion.foldable) { j.failAnalysis( errorClass = "AS_OF_JOIN.TOLERANCE_IS_UNFOLDABLE", diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/DeduplicateRelations.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/DeduplicateRelations.scala index 1fb703814fb91..f4fd5ddc55fab 100644 --- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/DeduplicateRelations.scala +++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/DeduplicateRelations.scala @@ -48,7 +48,7 @@ object DeduplicateRelations extends Rule[LogicalPlan] { if right.resolved && !j.duplicateResolved && noMissingInput(right.plan) => j.copy(right = right.withNewPlan(dedupRight(left, right.plan))) // Resolve duplicate output for AsOfJoin. - case j @ AsOfJoin(left, right, _, _, _, _, _) + case j @ AsOfJoin(left, right, _, _, _, _, _, _, _, _, _, _) if !j.duplicateResolved && noMissingInput(right) => j.copy(right = dedupRight(left, right)) // Resolve duplicate output for NearestByJoin. 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 new file mode 100644 index 0000000000000..e7d739442c580 --- /dev/null +++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolveAsOfJoin.scala @@ -0,0 +1,186 @@ +/* + * 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.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.AsOfMatchCondition +import org.apache.spark.sql.catalyst.plans.logical.{AsOfJoin, LogicalPlan, Project} +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 + * `orderExpression`, and expands `USING` column lists into equi-join predicates. + */ +object ResolveAsOfJoin extends Rule[LogicalPlan] with SQLConfHelper { + + override def apply(plan: LogicalPlan): LogicalPlan = plan.resolveOperatorsUpWithPruning( + _.containsPattern(AS_OF_JOIN), ruleId) { + case j @ AsOfJoin(left, right, _, condition, _, _, _, usingColumns, matchCmp, _, _, _) + if left.resolved && right.resolved && condition.forall(_.resolved) => + val (joinBase, usingProjection) = usingColumns match { + case Some(cols) if condition.isEmpty => + val (projectList, hiddenList, newCondition) = + NaturalAndUsingJoinResolution.computeJoinOutputsAndNewCondition( + left, + left.output, + right, + right.output, + j.joinType, + cols, + None, + (l, r) => conf.resolver(l, r)) + (j.copy(condition = newCondition, usingColumns = None), Some((projectList, hiddenList))) + case _ => (j, None) + } + val resolvedJoin = matchCmp match { + case Some(AsOfMatchCondition(leftExpr, operator, rightExpr)) => + AsOfJoinValidation.validateMatchConditionTableReferences( + joinBase, left, right, leftExpr, rightExpr) + if (leftExpr.resolved && rightExpr.resolved) { + AsOfJoinValidation.validateMatchConditionOperands(joinBase, leftExpr, rightExpr) + val (asOfCondition, orderExpression, leftSortExprs, rightSortExprs) = + AsOfJoin.resolveMatchComparison(left, right, leftExpr, operator, rightExpr) + joinBase.copy( + asOfCondition = asOfCondition, + orderExpression = orderExpression, + leftSortExprs = leftSortExprs, + rightSortExprs = rightSortExprs, + matchComparison = None) + } else { + joinBase + } + case _ => joinBase + } + usingProjection match { + case Some((projectList, hiddenList)) => + val project = Project(projectList, resolvedJoin) + project.setTagValue( + Project.hiddenOutputTag, + hiddenList.map(_.markAsQualifiedAccessOnly())) + project + case None => resolvedJoin + } + } +} + +private[analysis] object AsOfJoinValidation extends QueryErrorsBase { + + def validateMatchConditionTableReferences( + join: AsOfJoin, + left: LogicalPlan, + right: LogicalPlan, + leftExpr: Expression, + rightExpr: Expression): Unit = { + val leftSet = left.outputSet + val rightSet = right.outputSet + + def referencesBothJoinSides(refs: AttributeSet): Boolean = { + refs.nonEmpty && + refs.intersect(leftSet).nonEmpty && + refs.intersect(rightSet).nonEmpty + } + + val leftRefs = leftExpr.references + val rightRefs = rightExpr.references + if (referencesBothJoinSides(leftRefs) || referencesBothJoinSides(rightRefs)) { + join.failAnalysis( + errorClass = "ASOF_JOIN_MATCH_CONDITION_TABLE_REFERENCE", + messageParameters = Map( + "refs1" -> toSQLExpr(leftExpr), + "refs2" -> toSQLExpr(rightExpr))) + } + } + + def validateMatchConditionOperands( + join: AsOfJoin, + leftExpr: Expression, + rightExpr: Expression): Unit = { + Seq(leftExpr, rightExpr).foreach { expr => + findInvalidMatchConditionExpression(expr).foreach { invalidExpr => + join.failAnalysis( + errorClass = "ASOF_JOIN_MATCH_CONDITION_INVALID_EXPRESSION", + messageParameters = Map("expr" -> toSQLExpr(invalidExpr))) + } + } + if (!RowOrdering.isOrderable(leftExpr.dataType) || + !RowOrdering.isOrderable(rightExpr.dataType) || + !areMatchConditionTypesCompatible(leftExpr.dataType, rightExpr.dataType)) { + join.failAnalysis( + errorClass = "ASOF_JOIN_MATCH_CONDITION_INVALID_TYPE", + messageParameters = Map( + "type1" -> toSQLType(leftExpr.dataType), + "type2" -> toSQLType(rightExpr.dataType))) + } + } + + /** + * 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 + case e: AggregateExpression => e + case e: WindowExpression => e + case e if e.containsPattern(GENERATOR) => e + case e if !e.deterministic => e + }.headOption + } +} diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/RewriteAsOfJoin.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/RewriteAsOfJoin.scala index 6f10cad108f72..bdaa616c2ec75 100644 --- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/RewriteAsOfJoin.scala +++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/RewriteAsOfJoin.scala @@ -58,7 +58,8 @@ object RewriteAsOfJoin extends Rule[LogicalPlan] { if (conf.sortMergeAsOfJoinEnabled) return plan plan.transformUpWithNewOutput { - case j @ AsOfJoin(left, right, asOfCondition, condition, joinType, orderExpression, _) => + case j @ AsOfJoin( + left, right, asOfCondition, condition, joinType, orderExpression, _, _, _, _, _, _) => val conditionWithOuterReference = condition.map(And(_, asOfCondition)).getOrElse(asOfCondition).transformUp { case a: AttributeReference if left.outputSet.contains(a) => @@ -69,7 +70,7 @@ object RewriteAsOfJoin extends Rule[LogicalPlan] { val orderExpressionWithOuterReference = orderExpression.transformUp { case a: AttributeReference if left.outputSet.contains(a) => OuterReference(a) - } + } val rightStruct = CreateStruct(right.output) val nearestRight = MinBy(rightStruct, orderExpressionWithOuterReference) .toAggregateExpression() 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 032e032fa1a95..881791acb11f1 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 @@ -2505,11 +2505,17 @@ class AstBuilder extends DataTypeAstBuilder } } - if (ctx.nearestByClause != null) { - withNearestByJoin(ctx, base, baseJoinType) + val joinPostfix = Option(ctx.asofJoinPostfix) + val joinCriteriaCtx = joinPostfix.flatMap(p => Option(p.joinCriteria)) + val nearestByClauseCtx = joinPostfix.flatMap(p => Option(p.nearestByClause)) + + if (ctx.asofJoinCriteria != null) { + withAsOfJoin(ctx, base, ctx.asofJoinCriteria) + } else if (nearestByClauseCtx.isDefined) { + withNearestByJoin(ctx, base, baseJoinType, nearestByClauseCtx.get) } else { // Resolve the join type and join condition - val (joinType, condition) = Option(ctx.joinCriteria) match { + val (joinType, condition) = joinCriteriaCtx match { case Some(c) if c.USING != null => if (ctx.LATERAL != null) { throw QueryParsingErrors.lateralJoinWithUsingJoinUnsupportedError(ctx) @@ -2546,6 +2552,56 @@ class AstBuilder extends DataTypeAstBuilder } } + private def asOfMatchConditionFromExpression( + expr: Expression, + ctx: ParserRuleContext): (Expression, MatchComparisonOperator, Expression) = { + expr match { + case GreaterThanOrEqual(left, right) => (left, GreaterThanOrEqualOp, right) + case GreaterThan(left, right) => (left, GreaterThanOp, right) + case LessThanOrEqual(left, right) => (left, LessThanOrEqualOp, right) + case LessThan(left, right) => (left, LessThanOp, right) + case _ => + throw QueryParsingErrors.sqlAsOfJoinMatchConditionInvalidOperator( + asOfMatchConditionInvalidOperatorText(expr), ctx) + } + } + + private def asOfMatchConditionInvalidOperatorText(expr: Expression): String = expr match { + case EqualTo(_, _) => "=" + case And(_, _) => "AND" + case Or(_, _) => "OR" + case _ => expr.prettyName + } + + /** + * Build an [[AsOfJoin]] from the parsed `ASOF JOIN ... MATCH_CONDITION` clause. + */ + private def withAsOfJoin( + ctx: JoinRelationContext, + base: LogicalPlan, + criteria: AsofJoinCriteriaContext): AsOfJoin = { + if (!conf.getConf(SQLConf.SQL_ASOF_JOIN_ENABLED)) { + throw QueryParsingErrors.sqlAsofJoinDisabled(SQLConf.SQL_ASOF_JOIN_ENABLED.key, ctx) + } + val joinType = Option(ctx.asofJoinType) match { + case None => Inner + case Some(jt) if jt.LEFT != null => LeftOuter + case _ => Inner + } + val (leftExpr, operator, rightExpr) = + asOfMatchConditionFromExpression(expression(criteria.matchExpr), criteria.matchExpr) + val (condition, usingColumns) = + (Option(criteria.onExpr), Option(criteria.identifierList)) match { + case (Some(expr), None) => (Some(expression(expr)), None) + case (None, Some(ids)) => (None, Some(visitIdentifierList(ids))) + case (None, None) => (None, None) + case _ => + throw SparkException.internalError(s"Unimplemented asofJoinCriteria: $criteria") + } + AsOfJoin.fromMatchCondition( + base, plan(ctx.right), leftExpr, operator, rightExpr, condition, joinType, usingColumns) + } + /** * Build a [[NearestByJoin]] from the parsed `NEAREST BY` clause attached to a join relation. * Validates that the clause is not combined with `LATERAL` and that the base join type is one @@ -2555,7 +2611,8 @@ class AstBuilder extends DataTypeAstBuilder private def withNearestByJoin( ctx: JoinRelationContext, base: LogicalPlan, - baseJoinType: JoinType): NearestByJoin = { + baseJoinType: JoinType, + nearestByClause: NearestByClauseContext): NearestByJoin = { if (ctx.LATERAL != null) { throw QueryParsingErrors.nearestByJoinWithLateralUnsupportedError(ctx) } @@ -2563,7 +2620,7 @@ class AstBuilder extends DataTypeAstBuilder throw QueryParsingErrors.unsupportedNearestByJoinTypeError( ctx, baseJoinType.sql, NearestByJoinType.supportedDisplay) } - val clause = ctx.nearestByClause + val clause = nearestByClause val approx = clause.APPROX != null val numResults = Option(clause.num).map { n => // Guard against literals that overflow Long. diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/joinTypes.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/joinTypes.scala index 790307e44ec94..7dc0cba7625b9 100644 --- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/joinTypes.scala +++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/joinTypes.scala @@ -21,7 +21,7 @@ import java.util.Locale import org.apache.spark.SparkUnsupportedOperationException import org.apache.spark.sql.AnalysisException -import org.apache.spark.sql.catalyst.expressions.Attribute +import org.apache.spark.sql.catalyst.expressions.{Attribute, Expression} object JoinType { @@ -160,6 +160,41 @@ case object Forward extends AsOfJoinDirection case object Backward extends AsOfJoinDirection case object Nearest extends AsOfJoinDirection +sealed abstract class MatchComparisonOperator { + def sql: String + def flip: MatchComparisonOperator +} + +case object GreaterThanOrEqualOp extends MatchComparisonOperator { + override def sql: String = ">=" + override def flip: MatchComparisonOperator = LessThanOrEqualOp +} + +case object GreaterThanOp extends MatchComparisonOperator { + override def sql: String = ">" + override def flip: MatchComparisonOperator = LessThanOp +} + +case object LessThanOrEqualOp extends MatchComparisonOperator { + override def sql: String = "<=" + override def flip: MatchComparisonOperator = GreaterThanOrEqualOp +} + +case object LessThanOp extends MatchComparisonOperator { + override def sql: String = "<" + override def flip: MatchComparisonOperator = GreaterThanOp +} + +/** + * Parsed SQL `MATCH_CONDITION (left op right)` before analysis materializes it into + * [[org.apache.spark.sql.catalyst.plans.logical.AsOfJoin.asOfCondition]]. + * Must be a case class (not a tuple) so expression rewriting does not destroy the shape. + */ +case class AsOfMatchCondition( + left: Expression, + operator: MatchComparisonOperator, + right: Expression) + object LateralJoinType { val supported = Seq( @@ -226,6 +261,11 @@ object NearestByJoinType { } } +object AsOfJoinType { + + val supportedDisplay: String = "'INNER', 'LEFT OUTER'" +} + object NearestByJoinMode { /** @see [[NearestByJoinValidation.SupportedModes]] */ 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 b7c441063874d..7e9b7192a5165 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 @@ -26,6 +26,7 @@ import org.apache.spark.sql.catalyst.expressions.aggregate.{AggregateExpression, import org.apache.spark.sql.catalyst.plans._ import org.apache.spark.sql.catalyst.plans.physical.{HashPartitioning, Partitioning, RangePartitioning, RoundRobinPartitioning, ShufflePartitionIdPassThrough, SinglePartition} import org.apache.spark.sql.catalyst.trees.TreeNodeTag +import org.apache.spark.sql.catalyst.trees.TreePattern import org.apache.spark.sql.catalyst.trees.TreePattern._ import org.apache.spark.sql.catalyst.types.DataTypeUtils import org.apache.spark.sql.catalyst.types.DataTypeUtils.toAttributes @@ -35,6 +36,8 @@ import org.apache.spark.sql.internal.SQLConf import org.apache.spark.sql.types._ import org.apache.spark.sql.util.CaseInsensitiveStringMap import org.apache.spark.util.ArrayImplicits._ +import org.apache.spark.util.BestEffortLazyVal +import org.apache.spark.util.collection.BitSet import org.apache.spark.util.collection.Utils import org.apache.spark.util.random.RandomSampler @@ -2533,7 +2536,13 @@ case class AsOfJoin( condition: Option[Expression], joinType: JoinType, orderExpression: Expression, - toleranceAssertion: Option[Expression]) extends BinaryNode { + toleranceAssertion: Option[Expression], + usingColumns: Option[Seq[String]] = None, + matchComparison: Option[AsOfMatchCondition] = None, + leftSortExprs: Seq[Expression] = Nil, + rightSortExprs: Seq[Expression] = Nil, + requiresSortMergeAsOfJoin: Boolean = false) + extends BinaryNode { require(Seq(Inner, LeftOuter).contains(joinType), s"Unsupported as-of join type $joinType") @@ -2551,9 +2560,25 @@ case class AsOfJoin( def duplicateResolved: Boolean = left.outputSet.intersect(right.outputSet).isEmpty + private val _asOfJoinTreePatternBits = new BestEffortLazyVal[BitSet](() => { + val bits = new BitSet(TreePattern.maxId) + bits.union(super.treePatternBits) + matchComparison.foreach { mc => + bits.union(mc.left.treePatternBits) + bits.union(mc.right.treePatternBits) + } + bits + }) + + override def treePatternBits: BitSet = _asOfJoinTreePatternBits() + override lazy val resolved: Boolean = { childrenResolved && + usingColumns.isEmpty && + matchComparison.isEmpty && expressions.forall(_.resolved) && + leftSortExprs.forall(_.resolved) && + rightSortExprs.forall(_.resolved) && duplicateResolved && asOfCondition.dataType == BooleanType && condition.forall(_.dataType == BooleanType) && @@ -2564,6 +2589,40 @@ case class AsOfJoin( final override val nodePatterns: Seq[TreePattern] = Seq(AS_OF_JOIN) + /** + * [[AsOfMatchCondition]] is not an [[Expression]], so the default [[mapExpressions]] path does + * not rewrite its operand trees. Analysis rules (reference resolution, type coercion, etc.) + * must still reach those operands before [[ResolveAsOfJoin]] materializes them. + */ + override def mapExpressions(f: Expression => Expression): this.type = { + val mapped = super.mapExpressions(f) + val newLeftSortExprs = leftSortExprs.map(f) + val newRightSortExprs = rightSortExprs.map(f) + val withMatchComparison = matchComparison match { + case Some(mc) => + val newLeft = f(mc.left) + val newRight = f(mc.right) + if (newLeft.fastEquals(mc.left) && newRight.fastEquals(mc.right)) { + mapped + } else { + mapped.copy(matchComparison = Some(mc.copy(left = newLeft, right = newRight))) + } + case None => mapped + } + val sortExprsChanged = !newLeftSortExprs.zip(leftSortExprs).forall { + case (l, r) => l.fastEquals(r) + } || !newRightSortExprs.zip(rightSortExprs).forall { + case (l, r) => l.fastEquals(r) + } + if (sortExprsChanged) { + withMatchComparison.copy( + leftSortExprs = newLeftSortExprs, + rightSortExprs = newRightSortExprs).asInstanceOf[this.type] + } else { + withMatchComparison.asInstanceOf[this.type] + } + } + override protected def withNewChildrenInternal( newLeft: LogicalPlan, newRight: LogicalPlan): AsOfJoin = { copy(left = newLeft, right = newRight) @@ -2584,8 +2643,306 @@ object AsOfJoin { direction: AsOfJoinDirection): AsOfJoin = { val asOfCond = makeAsOfCond(leftAsOf, rightAsOf, tolerance, allowExactMatches, direction) val orderingExpr = makeOrderingExpr(leftAsOf, rightAsOf, direction) + val (leftSortExprs, rightSortExprs) = matchSortExpressions(leftAsOf, rightAsOf) AsOfJoin(left, right, asOfCond, condition, joinType, - orderingExpr, tolerance.map(t => GreaterThanOrEqual(t, Literal.default(t.dataType)))) + orderingExpr, tolerance.map(t => GreaterThanOrEqual(t, Literal.default(t.dataType))), + leftSortExprs = leftSortExprs, rightSortExprs = rightSortExprs) + } + + /** + * Build an [[AsOfJoin]] from a SQL `MATCH_CONDITION (left_expr op right_expr)` clause. + * Operand normalization is deferred until analysis when join inputs are resolved. + */ + def fromMatchCondition( + left: LogicalPlan, + right: LogicalPlan, + leftExpr: Expression, + operator: MatchComparisonOperator, + rightExpr: Expression, + condition: Option[Expression], + joinType: JoinType, + usingColumns: Option[Seq[String]] = None): AsOfJoin = { + AsOfJoin( + left, + right, + asOfCondition = Literal.TrueLiteral, + condition = condition, + joinType = joinType, + orderExpression = Literal(0), + toleranceAssertion = None, + usingColumns = usingColumns, + matchComparison = Some(AsOfMatchCondition(leftExpr, operator, rightExpr)), + requiresSortMergeAsOfJoin = true) + } + + def resolveMatchComparison( + left: LogicalPlan, + right: LogicalPlan, + leftExpr: Expression, + operator: MatchComparisonOperator, + rightExpr: Expression): (Expression, Expression, Seq[Expression], Seq[Expression]) = { + val (leftOperand, rightOperand, normalizedOp) = + normalizeMatchOperands(left, right, leftExpr, operator, rightExpr) + val (asOfCondition, orderExpression) = + buildMatchExpressions(leftOperand, rightOperand, normalizedOp) + val (leftSortExprs, rightSortExprs) = matchSortExpressions(leftOperand, rightOperand) + (asOfCondition, orderExpression, leftSortExprs, rightSortExprs) + } + + /** + * 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. + */ + 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 (isSqlTupleStructOperand(leftOperand) || isSqlTupleStructOperand(rightOperand)) { + val pairs = collectStructLeafPairs(leftOperand, rightOperand, leftStruct) + (pairs.map(_._1), pairs.map(_._2)) + } else { + (Seq(leftOperand), Seq(rightOperand)) + } + case _ => + (Seq(leftOperand), Seq(rightOperand)) + } + } + + /** True for SQL `(col1, col2, ...)` tuple operands, which become [[CreateNamedStruct]]. */ + private def isSqlTupleStructOperand(operand: Expression): Boolean = operand.isInstanceOf[CreateNamedStruct] + + private def normalizeMatchOperands( + left: LogicalPlan, + right: LogicalPlan, + expr1: Expression, + operator: MatchComparisonOperator, + 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) + (expr1Side, expr2Side) match { + case (Some(true), Some(false)) => (expr1, expr2, operator) + case (Some(false), Some(true)) => (expr2, expr1, operator.flip) + case _ => + throw QueryCompilationErrors.asOfJoinMatchConditionTableReferenceError(expr1, expr2) + } + } + + private def operandJoinSide( + expr: Expression, + leftSet: AttributeSet, + rightSet: AttributeSet): Option[Boolean] = { + val refs = expr.references + if (refs.isEmpty) { + None + } else if (refs.subsetOf(leftSet)) { + Some(true) + } else if (refs.subsetOf(rightSet)) { + Some(false) + } else { + None + } + } + + private def buildMatchExpressions( + leftOperand: Expression, + rightOperand: Expression, + operator: MatchComparisonOperator): (Expression, Expression) = { + val (leftForCompare, rightForCompare) = + alignOperandsForComparison(leftOperand, rightOperand) + val orderExpression = buildOrderExpression(leftOperand, rightOperand, operator) + operator match { + case GreaterThanOrEqualOp => + (GreaterThanOrEqual(leftForCompare, rightForCompare), orderExpression) + case GreaterThanOp => + (GreaterThan(leftForCompare, rightForCompare), orderExpression) + case LessThanOrEqualOp => + (LessThanOrEqual(leftForCompare, rightForCompare), orderExpression) + case LessThanOp => + (LessThan(leftForCompare, rightForCompare), orderExpression) + } + } + + private def buildOrderExpression( + leftOperand: Expression, + 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 => + buildFlattenedStructOrderExpression( + leftOperand, rightOperand, leftStruct, operator) + case _ => + buildLeafOrderExpression(leftOperand, rightOperand, operator) + } + } + + /** + * Tuple/struct operands may use different field names on each side. Rewrite them to positional + * structs with matching schemas so comparison and ordering type-check. + */ + private def alignOperandsForComparison( + leftOperand: Expression, + rightOperand: Expression): (Expression, Expression) = { + decomposeStructOperands(leftOperand, rightOperand) match { + case Some(pairs) => + val aligned = pairs.map { case (left, right) => + alignOperandsForComparison(left, right) + } + (CreateStruct(aligned.map(_._1)), CreateStruct(aligned.map(_._2))) + case None => + (leftOperand, rightOperand) + } + } + + 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 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, + elementType: DataType, + operator: MatchComparisonOperator): Expression = { + elementType match { + case struct: StructType => + 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, + ArrayType(struct, containsNull = true)) + ZipWith( + leftOperand, + rightOperand, + 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( + leftOperand, + rightOperand, + LambdaFunction(elementOrder, Seq(leftElement, rightElement))) + } + } + + 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, structType) + } + + private def collectStructLeafPairs( + leftOperand: Expression, + rightOperand: Expression, + structType: StructType): Seq[(Expression, Expression)] = { + structFieldExprs(leftOperand, structType) + .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)) + } + } + } + + private def wrapCompositeOrderExpression( + diffs: Seq[Expression], + compositeType: DataType): Expression = { + diffs match { + case Seq(single) => single + case _ => + compositeType match { + case _: ArrayType => CreateArray(diffs) + case _ => CreateStruct(diffs) + } + } + } + + /** Positional struct fields when both operands are the same struct shape. */ + 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 + } + } + + 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))) + } } private def makeAsOfCond( diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/rules/RuleIdCollection.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/rules/RuleIdCollection.scala index 12fcf3ed9aa3d..2953c09e183dc 100644 --- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/rules/RuleIdCollection.scala +++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/rules/RuleIdCollection.scala @@ -101,6 +101,7 @@ object RuleIdCollection { "org.apache.spark.sql.catalyst.analysis.ResolveCursors" :: "org.apache.spark.sql.catalyst.analysis.ResolveFetchCursor" :: "org.apache.spark.sql.catalyst.analysis.ResolveBinBy" :: + "org.apache.spark.sql.catalyst.analysis.ResolveAsOfJoin" :: "org.apache.spark.sql.catalyst.analysis.ResolveSetVariable" :: "org.apache.spark.sql.catalyst.analysis.ResolveTableConstraints" :: "org.apache.spark.sql.catalyst.analysis.ResolveTableSpec" :: diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/errors/QueryCompilationErrors.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/errors/QueryCompilationErrors.scala index 30cbad712d35d..d919f11587bdb 100644 --- a/sql/catalyst/src/main/scala/org/apache/spark/sql/errors/QueryCompilationErrors.scala +++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/errors/QueryCompilationErrors.scala @@ -4962,6 +4962,16 @@ private[sql] object QueryCompilationErrors extends QueryErrorsBase with Compilat messageParameters = Map.empty) } + def asOfJoinMatchConditionTableReferenceError( + expr1: Expression, + expr2: Expression): Throwable = { + new AnalysisException( + errorClass = "ASOF_JOIN_MATCH_CONDITION_TABLE_REFERENCE", + messageParameters = Map( + "refs1" -> toSQLExpr(expr1), + "refs2" -> toSQLExpr(expr2))) + } + def nestedSequentialStreamingUnionError(): Throwable = { new AnalysisException( errorClass = "NESTED_SEQUENTIAL_STREAMING_UNION", diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala index 22f5c8416e050..497a034e64ed1 100644 --- a/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala +++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala @@ -945,6 +945,15 @@ object SQLConf { .booleanConf .createWithDefault(false) + val SQL_ASOF_JOIN_ENABLED = + buildConf("spark.sql.join.asofJoin.enabled") + .doc("When true, enable SQL ASOF JOIN syntax with MATCH_CONDITION. When false, " + + "ASOF JOIN fails at parse time.") + .version("4.3.0") + .withBindingPolicy(ConfigBindingPolicy.SESSION) + .booleanConf + .createWithDefault(false) + val REQUIRE_ALL_CLUSTER_KEYS_FOR_CO_PARTITION = buildConf("spark.sql.requireAllClusterKeysForCoPartition") .internal() @@ -8466,6 +8475,8 @@ class SQLConf extends Serializable with Logging with SqlApiConf { def sortMergeAsOfJoinEnabled: Boolean = getConf(SORT_MERGE_AS_OF_JOIN_ENABLED) + def sqlAsOfJoinEnabled: Boolean = getConf(SQL_ASOF_JOIN_ENABLED) + def enableRadixSort: Boolean = getConf(RADIX_SORT_ENABLED) def isParquetSchemaMergingEnabled: Boolean = getConf(PARQUET_SCHEMA_MERGING_ENABLED) 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 e9f1a94f7010b..a51b8b1c7ab40 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 @@ -953,17 +953,102 @@ class PlanParserSuite extends AnalysisTest { stop = 73)) } + test("asof join") { + withSQLConf(SQLConf.SQL_ASOF_JOIN_ENABLED.key -> "true") { + assertEqual( + "select * from t asof join u match_condition (t.a >= u.a)", + AsOfJoin.fromMatchCondition( + table("t"), + table("u"), + $"t.a", + GreaterThanOrEqualOp, + $"u.a", + None, + Inner).select(star())) + + assertEqual( + "select * from t left asof join u match_condition (t.a >= u.a) on t.b = u.b", + AsOfJoin.fromMatchCondition( + table("t"), + table("u"), + $"t.a", + GreaterThanOrEqualOp, + $"u.a", + Some($"t.b" === $"u.b"), + LeftOuter).select(star())) + + assertEqual( + "select * from t asof join u match_condition (t.a >= u.a) using (b)", + AsOfJoin.fromMatchCondition( + table("t"), + table("u"), + $"t.a", + GreaterThanOrEqualOp, + $"u.a", + None, + Inner, + usingColumns = Some(Seq("b"))).select(star())) + + assertEqual( + "select * from t asof join u match_condition (u.a <= t.a)", + AsOfJoin.fromMatchCondition( + table("t"), + table("u"), + $"u.a", + LessThanOrEqualOp, + $"t.a", + None, + Inner).select(star())) + } + } + + test("asof join - invalid match operator") { + withSQLConf(SQLConf.SQL_ASOF_JOIN_ENABLED.key -> "true") { + val sql = + "select * from t asof join u match_condition (t.a = u.a)" + checkError( + exception = parseException(sql), + condition = "ASOF_JOIN_MATCH_CONDITION_INVALID_OPERATOR", + sqlState = "42K0E", + parameters = Map("operator" -> "=")) + } + } + + test("asof join - compound match condition rejected") { + withSQLConf(SQLConf.SQL_ASOF_JOIN_ENABLED.key -> "true") { + checkError( + exception = parseException( + "select * from t asof join u match_condition (t.a >= u.a and t.b >= u.b)"), + condition = "ASOF_JOIN_MATCH_CONDITION_INVALID_OPERATOR", + sqlState = "42K0E", + parameters = Map("operator" -> "AND")) + } + } + + test("asof join disabled by default") { + withSQLConf(SQLConf.SQL_ASOF_JOIN_ENABLED.key -> "false") { + checkError( + exception = parseException("select * from t asof join u match_condition (t.a >= u.a)"), + condition = "UNSUPPORTED_FEATURE.ASOF_JOIN", + sqlState = "0A000", + parameters = Map("config" -> "\"spark.sql.join.asofJoin.enabled\""), + context = ExpectedContext( + fragment = "asof join u match_condition (t.a >= u.a)", + start = 16, + stop = 55)) + } + } + test("nearest-by keywords are non-reserved (usable as identifiers)") { - // The five new keywords (APPROX, DISTANCE, EXACT, NEAREST, SIMILARITY) must remain - // non-reserved so they can continue to be used as column or table identifiers. - Seq("approx", "distance", "exact", "nearest", "similarity").foreach { kw => + // Spark-specific join keywords must remain non-reserved so they can be used as identifiers. + Seq("approx", "asof", "distance", "exact", "nearest", "similarity").foreach { kw => // As a column identifier in the SELECT list. parsePlan(s"select $kw from t") // As a table identifier in the FROM clause. parsePlan(s"select * from $kw") } // All five together in a single SELECT list. - parsePlan("select approx, distance, exact, nearest, similarity from t") + parsePlan("select approx, asof, distance, exact, nearest, similarity from t") } test("sampled relations") { diff --git a/sql/connect/client/jdbc/src/test/scala/org/apache/spark/sql/connect/client/jdbc/SparkConnectDatabaseMetaDataSuite.scala b/sql/connect/client/jdbc/src/test/scala/org/apache/spark/sql/connect/client/jdbc/SparkConnectDatabaseMetaDataSuite.scala index 6c35f73f5742f..babac5dd94863 100644 --- a/sql/connect/client/jdbc/src/test/scala/org/apache/spark/sql/connect/client/jdbc/SparkConnectDatabaseMetaDataSuite.scala +++ b/sql/connect/client/jdbc/src/test/scala/org/apache/spark/sql/connect/client/jdbc/SparkConnectDatabaseMetaDataSuite.scala @@ -210,7 +210,7 @@ class SparkConnectDatabaseMetaDataSuite extends ConnectFunSuite with RemoteSpark val metadata = conn.getMetaData // scalastyle:off line.size.limit // CURRENT_PATH and SYSTEM are excluded: getSQLKeywords drops SQL:2003 reserved words (see companion). - assert(metadata.getSQLKeywords === "ADD,AFTER,AGGREGATE,ALIGN,ALWAYS,ANALYZE,ANTI,ANY_VALUE,APPLY,APPROX,ARCHIVE,ASC,AUTO,BERNOULLI,BIN,BINDING,BIN_DISTRIBUTE_RATIO,BIN_END,BIN_START,BUCKET,BUCKETS,BYTE,CACHE,CASCADE,CATALOG,CATALOGS,CDC,CHANGE,CHANGES,CLEAR,CLUSTER,CLUSTERED,CODEGEN,COLLATION,COLLATIONS,COLLECTION,COLUMNS,COMMENT,COMPACT,COMPACTIONS,COMPENSATION,COMPUTE,CONCATENATE,CONTAINS,CONTINUE,COST,CURRENT_DATABASE,CURRENT_SCHEMA,DATA,DATABASE,DATABASES,DATEADD,DATEDIFF,DATE_ADD,DATE_DIFF,DAYOFYEAR,DAYS,DBPROPERTIES,DEFAULT_PATH,DEFINED,DEFINER,DELAY,DELIMITED,DESC,DFS,DIRECTORIES,DIRECTORY,DISTANCE,DISTRIBUTE,DIV,DO,ELSEIF,ENFORCED,ESCAPED,EVOLUTION,EXACT,EXCHANGE,EXCLUDE,EXCLUSIVE,EXIT,EXPLAIN,EXPORT,EXTEND,EXTENDED,FIELDS,FILEFORMAT,FIRST,FLOW,FOLLOWING,FORMAT,FORMATTED,FOUND,FUNCTIONS,GENERATED,GEOGRAPHY,GEOMETRY,HANDLER,HOURS,IDENTIFIED,IDENTIFIER,IF,IGNORE,ILIKE,IMMEDIATE,INCLUDE,INCLUSIVE,INCREMENT,INDEX,INDEXES,INPATH,INPUT,INPUTFORMAT,INVOKER,ITEMS,ITERATE,JSON,KEY,KEYS,LAST,LAZY,LEAVE,LEVEL,LIMIT,LINES,LIST,LOAD,LOCATION,LOCK,LOCKS,LOGICAL,LONG,LOOP,MACRO,MAP,MATCHED,MATERIALIZED,MEASURE,METRICS,MICROSECOND,MICROSECONDS,MILLISECOND,MILLISECONDS,MINUS,MINUTES,MONTHS,MSCK,NAME,NAMESPACE,NAMESPACES,NANOSECOND,NANOSECONDS,NEAREST,NORELY,NULLS,OFFSET,OPTION,OPTIONS,OUTPUTFORMAT,OVERWRITE,PARTITIONED,PARTITIONS,PATH,PERCENT,PIVOT,PLACING,PRECEDING,PRINCIPALS,PROCEDURES,PROPERTIES,PURGE,QUALIFY,QUARTER,QUERY,RECORDREADER,RECORDWRITER,RECOVER,RECURSION,REDUCE,REFRESH,RELY,RENAME,REPAIR,REPEAT,REPEATABLE,REPLACE,RESET,RESPECT,RESTRICT,ROLE,ROLES,SCHEMA,SCHEMAS,SECONDS,SECURITY,SEMI,SEPARATED,SEQUENCE,SERDE,SERDEPROPERTIES,SETS,SHORT,SHOW,SIMILARITY,SINGLE,SKEWED,SORT,SORTED,SOURCE,STATISTICS,STORED,STRATIFY,STREAM,STREAMING,STRING,STRUCT,SUBSTR,SYNC,SYSTEM_PATH,SYSTEM_TIME,SYSTEM_VERSION,TABLES,TARGET,TBLPROPERTIES,TERMINATED,TIMEDIFF,TIMESTAMPADD,TIMESTAMPDIFF,TIMESTAMP_LTZ,TIMESTAMP_NTZ,TINYINT,TOUCH,TRANSACTION,TRANSACTIONS,TRANSFORM,TRUNCATE,TRY_CAST,TYPE,UNARCHIVE,UNBOUNDED,UNCACHE,UNIFORM,UNLOCK,UNPIVOT,UNSET,UNTIL,USE,VAR,VARIABLE,VARIANT,VERSION,VIEW,VIEWS,VOID,WATERMARK,WEEK,WEEKS,WHILE,WIDTH,X,YEARS,ZONE") + assert(metadata.getSQLKeywords === "ADD,AFTER,AGGREGATE,ALIGN,ALWAYS,ANALYZE,ANTI,ANY_VALUE,APPLY,APPROX,ARCHIVE,ASC,ASOF,AUTO,BERNOULLI,BIN,BINDING,BIN_DISTRIBUTE_RATIO,BIN_END,BIN_START,BUCKET,BUCKETS,BYTE,CACHE,CASCADE,CATALOG,CATALOGS,CDC,CHANGE,CHANGES,CLEAR,CLUSTER,CLUSTERED,CODEGEN,COLLATION,COLLATIONS,COLLECTION,COLUMNS,COMMENT,COMPACT,COMPACTIONS,COMPENSATION,COMPUTE,CONCATENATE,CONTAINS,CONTINUE,COST,CURRENT_DATABASE,CURRENT_SCHEMA,DATA,DATABASE,DATABASES,DATEADD,DATEDIFF,DATE_ADD,DATE_DIFF,DAYOFYEAR,DAYS,DBPROPERTIES,DEFAULT_PATH,DEFINED,DEFINER,DELAY,DELIMITED,DESC,DFS,DIRECTORIES,DIRECTORY,DISTANCE,DISTRIBUTE,DIV,DO,ELSEIF,ENFORCED,ESCAPED,EVOLUTION,EXACT,EXCHANGE,EXCLUDE,EXCLUSIVE,EXIT,EXPLAIN,EXPORT,EXTEND,EXTENDED,FIELDS,FILEFORMAT,FIRST,FLOW,FOLLOWING,FORMAT,FORMATTED,FOUND,FUNCTIONS,GENERATED,GEOGRAPHY,GEOMETRY,HANDLER,HOURS,IDENTIFIED,IDENTIFIER,IF,IGNORE,ILIKE,IMMEDIATE,INCLUDE,INCLUSIVE,INCREMENT,INDEX,INDEXES,INPATH,INPUT,INPUTFORMAT,INVOKER,ITEMS,ITERATE,JSON,KEY,KEYS,LAST,LAZY,LEAVE,LEVEL,LIMIT,LINES,LIST,LOAD,LOCATION,LOCK,LOCKS,LOGICAL,LONG,LOOP,MACRO,MAP,MATCHED,MATCH_CONDITION,MATERIALIZED,MEASURE,METRICS,MICROSECOND,MICROSECONDS,MILLISECOND,MILLISECONDS,MINUS,MINUTES,MONTHS,MSCK,NAME,NAMESPACE,NAMESPACES,NANOSECOND,NANOSECONDS,NEAREST,NORELY,NULLS,OFFSET,OPTION,OPTIONS,OUTPUTFORMAT,OVERWRITE,PARTITIONED,PARTITIONS,PATH,PERCENT,PIVOT,PLACING,PRECEDING,PRINCIPALS,PROCEDURES,PROPERTIES,PURGE,QUALIFY,QUARTER,QUERY,RECORDREADER,RECORDWRITER,RECOVER,RECURSION,REDUCE,REFRESH,RELY,RENAME,REPAIR,REPEAT,REPEATABLE,REPLACE,RESET,RESPECT,RESTRICT,ROLE,ROLES,SCHEMA,SCHEMAS,SECONDS,SECURITY,SEMI,SEPARATED,SEQUENCE,SERDE,SERDEPROPERTIES,SETS,SHORT,SHOW,SIMILARITY,SINGLE,SKEWED,SORT,SORTED,SOURCE,STATISTICS,STORED,STRATIFY,STREAM,STREAMING,STRING,STRUCT,SUBSTR,SYNC,SYSTEM_PATH,SYSTEM_TIME,SYSTEM_VERSION,TABLES,TARGET,TBLPROPERTIES,TERMINATED,TIMEDIFF,TIMESTAMPADD,TIMESTAMPDIFF,TIMESTAMP_LTZ,TIMESTAMP_NTZ,TINYINT,TOUCH,TRANSACTION,TRANSACTIONS,TRANSFORM,TRUNCATE,TRY_CAST,TYPE,UNARCHIVE,UNBOUNDED,UNCACHE,UNIFORM,UNLOCK,UNPIVOT,UNSET,UNTIL,USE,VAR,VARIABLE,VARIANT,VERSION,VIEW,VIEWS,VOID,WATERMARK,WEEK,WEEKS,WHILE,WIDTH,X,YEARS,ZONE") // scalastyle:on line.size.limit } } diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/SparkStrategies.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/SparkStrategies.scala index d89f7a919269a..cfb6784f402f5 100644 --- a/sql/core/src/main/scala/org/apache/spark/sql/execution/SparkStrategies.scala +++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/SparkStrategies.scala @@ -432,16 +432,22 @@ abstract class SparkStrategies extends QueryPlanner[SparkPlan] { object AsOfJoinSelection extends Strategy with PredicateHelper { def apply(plan: LogicalPlan): Seq[SparkPlan] = plan match { case j @ AsOfJoin(left, right, asOfCondition, condition, joinType, - orderExpression, _) if conf.sortMergeAsOfJoinEnabled => + orderExpression, _, _, _, leftSortExprs, rightSortExprs, _) + if conf.sortMergeAsOfJoinEnabled => val (leftKeys, rightKeys, residual) = condition match { case Some(cond) => extractEquiJoinKeys(cond, left, right) case None => (Seq.empty[Expression], Seq.empty[Expression], None) } - val (leftAsOf, rightAsOf) = extractAsOfExprs( - asOfCondition, orderExpression, left, right) + val (leftSort, rightSort) = + if (leftSortExprs.nonEmpty && rightSortExprs.nonEmpty) { + (leftSortExprs, rightSortExprs) + } else { + val (leftAsOf, rightAsOf) = extractAsOfExprs(orderExpression, left, right) + (Seq(leftAsOf), Seq(rightAsOf)) + } joins.SortMergeAsOfJoinExec( - leftKeys, rightKeys, leftAsOf, rightAsOf, + leftKeys, rightKeys, leftSort, rightSort, asOfCondition, orderExpression, joinType, residual, planLater(left), planLater(right)) :: Nil case _ => Nil @@ -493,7 +499,6 @@ abstract class SparkStrategies extends QueryPlanner[SparkPlan] { * allowExactMatches variations that complicate asOfCondition's shape. */ private def extractAsOfExprs( - asOfCondition: Expression, orderExpression: Expression, left: LogicalPlan, right: LogicalPlan): (Expression, Expression) = { diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/joins/SortMergeAsOfJoinExec.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/joins/SortMergeAsOfJoinExec.scala index 955fc7a185c82..5bdfbca78abfa 100644 --- a/sql/core/src/main/scala/org/apache/spark/sql/execution/joins/SortMergeAsOfJoinExec.scala +++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/joins/SortMergeAsOfJoinExec.scala @@ -45,8 +45,8 @@ import org.apache.spark.sql.execution.metric.{SQLMetric, SQLMetrics} case class SortMergeAsOfJoinExec( leftKeys: Seq[Expression], rightKeys: Seq[Expression], - leftAsOfExpr: Expression, - rightAsOfExpr: Expression, + leftSortExprs: Seq[Expression], + rightSortExprs: Seq[Expression], asOfCondition: Expression, orderExpression: Expression, joinType: JoinType, @@ -55,6 +55,10 @@ case class SortMergeAsOfJoinExec( right: SparkPlan, isSkewJoin: Boolean = false) extends ShuffledJoin { + require(leftSortExprs.nonEmpty && rightSortExprs.nonEmpty && + leftSortExprs.length == rightSortExprs.length, + s"$nodeName requires matching non-empty sort expressions on both sides") + require(Seq(Inner, LeftOuter).exists(joinType == _), s"$nodeName does not support join type: $joinType") @@ -79,27 +83,37 @@ case class SortMergeAsOfJoinExec( } override def requiredChildOrdering: Seq[Seq[SortOrder]] = { - val leftOrdering = leftKeys.map(SortOrder(_, Ascending)) :+ - SortOrder(leftAsOfExpr, Ascending) - val rightOrdering = rightKeys.map(SortOrder(_, Ascending)) :+ - SortOrder(rightAsOfExpr, Ascending) + val leftOrdering = leftKeys.map(SortOrder(_, Ascending)) ++ + leftSortExprs.map(SortOrder(_, Ascending)) + val rightOrdering = rightKeys.map(SortOrder(_, Ascending)) ++ + rightSortExprs.map(SortOrder(_, Ascending)) leftOrdering :: rightOrdering :: Nil } override def outputOrdering: Seq[SortOrder] = left.outputOrdering - // Determine scan direction based on the order expression (distance metric). - // This is a performance heuristic only -- if it misclassifies, the scan - // still produces the correct result; only the early-termination shortcut - // is lost. - // - // orderExpression is direction-unique by construction: - // Backward: Subtract(leftAsOf, rightAsOf) -> backward mode - // Forward: Subtract(rightAsOf, leftAsOf) -> forward mode - // Nearest: If(...) -> forward mode - private val isBackwardJoin: Boolean = orderExpression match { - case Subtract(l, _, _) if l.semanticEquals(leftAsOfExpr) => true - case _ => false + // Backward joins scan the right-side buffer forward and keep the last match. + // Detect from asOfCondition so composite MATCH_CONDITION sort keys still work. + private val isBackwardJoin: Boolean = + isBackwardAsOfCondition(asOfCondition, left.output, right.output) + + private def isBackwardAsOfCondition( + condition: Expression, + leftOutput: Seq[Attribute], + rightOutput: Seq[Attribute]): Boolean = { + val leftAttrs = AttributeSet(leftOutput) + val rightAttrs = AttributeSet(rightOutput) + def isBackwardPredicate(expr: Expression): Boolean = expr match { + case GreaterThanOrEqual(l, r) + if l.references.subsetOf(leftAttrs) && r.references.subsetOf(rightAttrs) => + true + case GreaterThan(l, r) + if l.references.subsetOf(leftAttrs) && r.references.subsetOf(rightAttrs) => + true + case And(c, _) => isBackwardPredicate(c) + case _ => false + } + isBackwardPredicate(condition) } protected override def doExecute(): RDD[InternalRow] = { @@ -165,8 +179,14 @@ private[joins] class SortMergeAsOfJoinScanner( private val joinedOutput = leftOutput ++ rightOutput private val joinedRow = new JoinedRow() - private val resultProjection = - UnsafeProjection.create(joinedOutput, joinedOutput) + // Use nullable bound references so outer-join null padding is safe even when + // right attributes are NOT NULL in the catalog schema. + private val resultProjection = { + val nullableRefs = joinedOutput.zipWithIndex.map { case (attr, i) => + BoundReference(i, attr.dataType, nullable = true) + } + UnsafeProjection.create(nullableRefs, joinedOutput) + } private val boundAsOfCond = bindReference(asOfCondition, joinedOutput) private val boundOrderExpr = bindReference(orderExpression, joinedOutput) @@ -190,7 +210,15 @@ private[joins] class SortMergeAsOfJoinScanner( private val distanceOrdering = TypeUtils.getInterpretedOrdering(orderExpression.dataType) - private val nullRightRow = new GenericInternalRow(rightOutput.length) + // Materialize an all-null right row as UnsafeRow. GenericInternalRow cannot be + // passed through identity UnsafeProjection when right columns are NOT NULL. + private val nullRightRow: InternalRow = { + val nullableRefs = rightOutput.zipWithIndex.map { case (attr, i) => + BoundReference(i, attr.dataType, nullable = true) + } + val proj = UnsafeProjection.create(nullableRefs, rightOutput) + proj(new GenericInternalRow(rightOutput.length)) + } // Spill-backed right-side buffer private val rightGroupBuffer = new ExternalAppendOnlyUnsafeRowArray( diff --git a/sql/core/src/test/resources/sql-tests/results/keywords-enforced.sql.out b/sql/core/src/test/resources/sql-tests/results/keywords-enforced.sql.out index 0c57178048ebe..989bfb7251117 100644 --- a/sql/core/src/test/resources/sql-tests/results/keywords-enforced.sql.out +++ b/sql/core/src/test/resources/sql-tests/results/keywords-enforced.sql.out @@ -23,6 +23,7 @@ ARRAY false AS true ASC false ASENSITIVE false +ASOF false AT false ATOMIC false AUTHORIZATION true @@ -234,6 +235,7 @@ LOOP false MACRO false MAP false MATCHED false +MATCH_CONDITION false MATERIALIZED false MAX false MEASURE false diff --git a/sql/core/src/test/resources/sql-tests/results/keywords.sql.out b/sql/core/src/test/resources/sql-tests/results/keywords.sql.out index e0528a75c6ef5..ee127e584f172 100644 --- a/sql/core/src/test/resources/sql-tests/results/keywords.sql.out +++ b/sql/core/src/test/resources/sql-tests/results/keywords.sql.out @@ -23,6 +23,7 @@ ARRAY false AS false ASC false ASENSITIVE false +ASOF false AT false ATOMIC false AUTHORIZATION false @@ -234,6 +235,7 @@ LOOP false MACRO false MAP false MATCHED false +MATCH_CONDITION false MATERIALIZED false MAX false MEASURE false diff --git a/sql/core/src/test/resources/sql-tests/results/nonansi/keywords.sql.out b/sql/core/src/test/resources/sql-tests/results/nonansi/keywords.sql.out index e0528a75c6ef5..ee127e584f172 100644 --- a/sql/core/src/test/resources/sql-tests/results/nonansi/keywords.sql.out +++ b/sql/core/src/test/resources/sql-tests/results/nonansi/keywords.sql.out @@ -23,6 +23,7 @@ ARRAY false AS false ASC false ASENSITIVE false +ASOF false AT false ATOMIC false AUTHORIZATION false @@ -234,6 +235,7 @@ LOOP false MACRO false MAP false MATCHED false +MATCH_CONDITION false MATERIALIZED false MAX false MEASURE false 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 new file mode 100644 index 0000000000000..b3c4324e4ed20 --- /dev/null +++ b/sql/core/src/test/scala/org/apache/spark/sql/AsOfJoinSQLSuite.scala @@ -0,0 +1,74 @@ +/* + * 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 + +import org.apache.spark.sql.catalyst.parser.ParseException +import org.apache.spark.sql.internal.SQLConf +import org.apache.spark.sql.test.SharedSparkSession + +/** + * SQL ASOF JOIN surface tests (parser and feature gating). + * Execution semantics and complex MATCH_CONDITION types are covered by + * [[AsOfJoinSortMergeSQLSuite]], which requires sort-merge ASOF join. + */ +class AsOfJoinSQLSuite extends QueryTest with SharedSparkSession { + + override def beforeAll(): Unit = { + super.beforeAll() + spark.conf.set(SQLConf.SQL_ASOF_JOIN_ENABLED.key, "true") + spark.conf.set(SQLConf.SORT_MERGE_AS_OF_JOIN_ENABLED.key, "true") + } + + override def afterAll(): Unit = { + spark.conf.unset(SQLConf.SORT_MERGE_AS_OF_JOIN_ENABLED.key) + spark.conf.unset(SQLConf.SQL_ASOF_JOIN_ENABLED.key) + super.afterAll() + } + + test("equality operator is rejected in MATCH_CONDITION") { + sql( + """ + |CREATE OR REPLACE TEMP VIEW trades(trade_time, symbol) AS + |VALUES (TIMESTAMP '2026-06-29 10:00:00', 'AAPL') + |""".stripMargin) + sql( + """ + |CREATE OR REPLACE TEMP VIEW quotes(quote_time, symbol) AS + |VALUES (TIMESTAMP '2026-06-29 09:00:00', 'AAPL') + |""".stripMargin) + val sqlText = + """ + |SELECT * FROM trades t + |ASOF JOIN quotes q + | MATCH_CONDITION (t.trade_time = q.quote_time) + | ON t.symbol = q.symbol + |""".stripMargin + checkError( + exception = intercept[ParseException](sql(sqlText)), + condition = "ASOF_JOIN_MATCH_CONDITION_INVALID_OPERATOR", + sqlState = Some("42K0E"), + parameters = Map("operator" -> "="), + queryContext = Array( + ExpectedContext( + fragment = """ASOF JOIN quotes q + | MATCH_CONDITION (t.trade_time = q.quote_time) + | ON t.symbol = q.symbol""".stripMargin, + start = 24, + stop = 114))) + } +} diff --git a/sql/core/src/test/scala/org/apache/spark/sql/AsOfJoinSortMergeSQLSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/AsOfJoinSortMergeSQLSuite.scala new file mode 100644 index 0000000000000..e5eaffb3a72d1 --- /dev/null +++ b/sql/core/src/test/scala/org/apache/spark/sql/AsOfJoinSortMergeSQLSuite.scala @@ -0,0 +1,612 @@ +/* + * 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 + +import java.sql.{Date, Timestamp} + +import org.apache.spark.sql.AnalysisException +import org.apache.spark.sql.execution.adaptive.AdaptiveSparkPlanHelper +import org.apache.spark.sql.execution.joins.SortMergeAsOfJoinExec +import org.apache.spark.sql.internal.SQLConf +import org.apache.spark.sql.test.SharedSparkSession + +/** + * SQL ASOF JOIN tests that require the sort-merge physical operator + * (`spark.sql.join.sortMergeAsOfJoin.enabled=true`). + */ +class AsOfJoinSortMergeSQLSuite extends QueryTest + with SharedSparkSession + with AdaptiveSparkPlanHelper { + + override def beforeAll(): Unit = { + super.beforeAll() + spark.conf.set(SQLConf.SQL_ASOF_JOIN_ENABLED.key, "true") + spark.conf.set(SQLConf.SORT_MERGE_AS_OF_JOIN_ENABLED.key, "true") + } + + override def afterAll(): Unit = { + spark.conf.unset(SQLConf.SORT_MERGE_AS_OF_JOIN_ENABLED.key) + spark.conf.unset(SQLConf.SQL_ASOF_JOIN_ENABLED.key) + super.afterAll() + } + + private def assertUsesSortMergeAsOfJoin(query: DataFrame): Unit = { + val plan = query.queryExecution.executedPlan + assert(collectWithSubqueries(plan) { + case _: SortMergeAsOfJoinExec => true + }.nonEmpty, s"Expected SortMergeAsOfJoinExec in plan:\n$plan") + } + + private def checkSortMergeAsOf(query: => DataFrame, expected: Seq[Row]): Unit = { + val df = query + assertUsesSortMergeAsOfJoin(df) + checkAnswer(df, expected) + } + + private def setupTradeQuoteViews(): Unit = { + sql( + """ + |CREATE OR REPLACE TEMP VIEW trades(trade_time, symbol, quantity) AS + |VALUES (TIMESTAMP '2026-06-29 10:00:05', 'AAPL', 100), + | (TIMESTAMP '2026-06-29 10:00:11', 'AAPL', 200), + | (TIMESTAMP '2026-06-29 10:00:12', 'MSFT', 50), + | (TIMESTAMP '2026-06-29 09:59:59', 'GOOG', 30) + |""".stripMargin) + sql( + """ + |CREATE OR REPLACE TEMP VIEW quotes(quote_time, symbol, bid_price) AS + |VALUES (TIMESTAMP '2026-06-29 10:00:00', 'AAPL', 180.10), + | (TIMESTAMP '2026-06-29 10:00:07', 'AAPL', 180.15), + | (TIMESTAMP '2026-06-29 10:00:10', 'AAPL', 180.20), + | (TIMESTAMP '2026-06-29 10:00:08', 'MSFT', 420.50) + |""".stripMargin) + } + + test("SQL ASOF JOIN requires sort-merge conf") { + setupTradeQuoteViews() + val sqlText = + """ + |SELECT t.trade_time, q.bid_price + |FROM trades t ASOF JOIN quotes q + | MATCH_CONDITION (t.trade_time >= q.quote_time) + | ON t.symbol = q.symbol + |""".stripMargin + withSQLConf(SQLConf.SORT_MERGE_AS_OF_JOIN_ENABLED.key -> "false") { + checkError( + exception = intercept[AnalysisException](sql(sqlText)), + condition = "AS_OF_JOIN.SORT_MERGE_REQUIRED", + parameters = Map("config" -> SQLConf.SORT_MERGE_AS_OF_JOIN_ENABLED.key), + queryContext = Array( + ExpectedContext( + fragment = """ASOF JOIN quotes q + | MATCH_CONDITION (t.trade_time >= q.quote_time) + | ON t.symbol = q.symbol""".stripMargin, + start = 48, + stop = 139))) + } + } + + test("INNER ASOF JOIN with TIMESTAMP MATCH_CONDITION") { + setupTradeQuoteViews() + checkSortMergeAsOf( + sql( + """ + |SELECT t.trade_time, t.symbol, t.quantity, q.bid_price + |FROM trades t + |ASOF JOIN quotes q + | MATCH_CONDITION (t.trade_time >= q.quote_time) + | ON t.symbol = q.symbol + |""".stripMargin), + Seq( + Row(Timestamp.valueOf("2026-06-29 10:00:05"), "AAPL", 100, 180.10), + Row(Timestamp.valueOf("2026-06-29 10:00:11"), "AAPL", 200, 180.20), + Row(Timestamp.valueOf("2026-06-29 10:00:12"), "MSFT", 50, 420.50))) + } + + test("LEFT ASOF JOIN preserves unmatched left rows") { + setupTradeQuoteViews() + checkSortMergeAsOf( + sql( + """ + |SELECT t.trade_time, t.symbol, q.bid_price + |FROM trades t + |LEFT ASOF JOIN quotes q + | MATCH_CONDITION (t.trade_time >= q.quote_time) + | ON t.symbol = q.symbol + |ORDER BY t.trade_time + |""".stripMargin), + Seq( + Row(Timestamp.valueOf("2026-06-29 09:59:59"), "GOOG", null), + Row(Timestamp.valueOf("2026-06-29 10:00:05"), "AAPL", 180.10), + Row(Timestamp.valueOf("2026-06-29 10:00:11"), "AAPL", 180.20), + Row(Timestamp.valueOf("2026-06-29 10:00:12"), "MSFT", 420.50))) + } + + test("USING is equivalent to ON symbol equality") { + setupTradeQuoteViews() + checkSortMergeAsOf( + sql( + """ + |SELECT trade_time, symbol, bid_price + |FROM trades + |ASOF JOIN quotes + | MATCH_CONDITION (trades.trade_time >= quotes.quote_time) + | USING (symbol) + |ORDER BY trade_time + |""".stripMargin), + Seq( + Row(Timestamp.valueOf("2026-06-29 10:00:05"), "AAPL", 180.10), + Row(Timestamp.valueOf("2026-06-29 10:00:11"), "AAPL", 180.20), + Row(Timestamp.valueOf("2026-06-29 10:00:12"), "MSFT", 420.50))) + } + + test("forward match with <=") { + sql( + """ + |CREATE OR REPLACE TEMP VIEW alerts(alert_time, host) AS + |VALUES (TIMESTAMP '2026-06-29 10:00:00', 'db-01') + |""".stripMargin) + sql( + """ + |CREATE OR REPLACE TEMP VIEW maintenance(window_start, host) AS + |VALUES (TIMESTAMP '2026-06-29 08:00:00', 'db-01'), + | (TIMESTAMP '2026-06-29 12:00:00', 'db-01') + |""".stripMargin) + checkSortMergeAsOf( + sql( + """ + |SELECT a.alert_time, a.host, m.window_start + |FROM alerts a + |ASOF JOIN maintenance m + | MATCH_CONDITION (a.alert_time <= m.window_start) + | ON a.host = m.host + |""".stripMargin), + Row( + Timestamp.valueOf("2026-06-29 10:00:00"), + "db-01", + Timestamp.valueOf("2026-06-29 12:00:00")) :: Nil) + } + + test("DATE scalar MATCH_CONDITION") { + checkSortMergeAsOf( + sql( + """ + |SELECT t.d, r.d AS matched_d + |FROM VALUES (DATE '2026-06-29') AS t(d) ASOF JOIN + | VALUES (DATE '2026-06-28'), (DATE '2026-06-29') AS r(d) + | MATCH_CONDITION (t.d >= r.d) + |""".stripMargin), + Row(Date.valueOf("2026-06-29"), Date.valueOf("2026-06-29")) :: Nil) + } + + test("INT scalar MATCH_CONDITION") { + checkSortMergeAsOf( + sql( + """ + |SELECT t.k, r.k AS matched_k + |FROM VALUES (10) AS t(k) ASOF JOIN VALUES (5), (7) AS r(k) + | MATCH_CONDITION (t.k >= r.k) + |""".stripMargin), + Row(10, 7) :: Nil) + } + + test("STRING scalar MATCH_CONDITION") { + checkSortMergeAsOf( + sql( + """ + |SELECT t.k, r.k AS matched_k + |FROM VALUES ('c') AS t(k) ASOF JOIN VALUES ('a'), ('b') AS r(k) + | MATCH_CONDITION (t.k >= r.k) + |""".stripMargin), + Row("c", "b") :: Nil) + } + + test("STRING scalar forward MATCH_CONDITION with <=") { + checkSortMergeAsOf( + sql( + """ + |SELECT t.k, r.k AS matched_k + |FROM VALUES ('b') AS t(k) ASOF JOIN VALUES ('c'), ('d'), ('z') AS r(k) + | MATCH_CONDITION (t.k <= r.k) + |""".stripMargin), + Row("b", "c") :: Nil) + } + + test("DECIMAL scalar MATCH_CONDITION") { + checkSortMergeAsOf( + sql( + """ + |SELECT t.k, r.k AS matched_k + |FROM VALUES (CAST(10.50 AS DECIMAL(10, 2))) AS t(k) ASOF JOIN + | VALUES (CAST(5.00 AS DECIMAL(10, 2))), + | (CAST(7.25 AS DECIMAL(10, 2))), + | (CAST(9.99 AS DECIMAL(10, 2))) AS r(k) + | MATCH_CONDITION (t.k >= r.k) + |""".stripMargin), + Row( + new java.math.BigDecimal("10.50"), + new java.math.BigDecimal("9.99")) :: Nil) + } + + test("DECIMAL scalar forward MATCH_CONDITION with <=") { + checkSortMergeAsOf( + sql( + """ + |SELECT t.k, r.k AS matched_k + |FROM VALUES (CAST(10.50 AS DECIMAL(10, 2))) AS t(k) ASOF JOIN + | VALUES (CAST(11.00 AS DECIMAL(10, 2))), + | (CAST(12.50 AS DECIMAL(10, 2))), + | (CAST(15.00 AS DECIMAL(10, 2))) AS r(k) + | MATCH_CONDITION (t.k <= r.k) + |""".stripMargin), + Row( + new java.math.BigDecimal("10.50"), + new java.math.BigDecimal("11.00")) :: Nil) + } + + test("DOUBLE scalar MATCH_CONDITION") { + checkSortMergeAsOf( + sql( + """ + |SELECT t.k, r.k AS matched_k + |FROM VALUES (CAST(10.5 AS DOUBLE)) AS t(k) ASOF JOIN + | VALUES (CAST(5.0 AS DOUBLE)), + | (CAST(7.25 AS DOUBLE)), + | (CAST(9.99 AS DOUBLE)) AS r(k) + | MATCH_CONDITION (t.k >= r.k) + |""".stripMargin), + Row(10.5, 9.99) :: Nil) + } + + test("DOUBLE scalar forward MATCH_CONDITION with <=") { + checkSortMergeAsOf( + sql( + """ + |SELECT t.k, r.k AS matched_k + |FROM VALUES (CAST(10.5 AS DOUBLE)) AS t(k) ASOF JOIN + | VALUES (CAST(11.0 AS DOUBLE)), + | (CAST(12.5 AS DOUBLE)), + | (CAST(15.0 AS DOUBLE)) AS r(k) + | MATCH_CONDITION (t.k <= r.k) + |""".stripMargin), + Row(10.5, 11.0) :: Nil) + } + + test("STRUCT tuple MATCH_CONDITION") { + sql( + """ + |CREATE OR REPLACE TEMP VIEW requests(req_ts, seq, service) AS + |VALUES (TIMESTAMP '2026-06-29 10:00:00', 5, 'api'), + | (TIMESTAMP '2026-06-29 10:03:00', 1, 'api') + |""".stripMargin) + 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'), + | (TIMESTAMP '2026-06-29 10:00:00', 3, 'api', 'v1.1') + |""".stripMargin) + checkSortMergeAsOf( + sql( + """ + |SELECT r.req_ts, r.seq, 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 + |ORDER BY r.req_ts, r.seq + |""".stripMargin), + Seq( + Row(Timestamp.valueOf("2026-06-29 10:00:00"), 5, "v1.1"), + Row(Timestamp.valueOf("2026-06-29 10:03:00"), 1, "v1.1"))) + } + + test("scalar leaf STRUCT tuple MATCH_CONDITION") { + sql( + """ + |CREATE OR REPLACE TEMP VIEW left_nested(inner_ts, inner_seq, tag) AS + |VALUES (TIMESTAMP '2026-06-29 10:00:00', 2, 'a') + |""".stripMargin) + sql( + """ + |CREATE OR REPLACE TEMP VIEW right_nested(inner_ts, inner_seq, tag) AS + |VALUES (TIMESTAMP '2026-06-29 09:00:00', 1, 'a'), + | (TIMESTAMP '2026-06-29 10:00:00', 1, 'a') + |""".stripMargin) + checkSortMergeAsOf( + sql( + """ + |SELECT r.tag, r.inner_seq + |FROM left_nested t + |ASOF JOIN right_nested r + | MATCH_CONDITION ((t.inner_ts, t.inner_seq, t.tag) >= (r.inner_ts, r.inner_seq, r.tag)) + |""".stripMargin), + Row("a", 1) :: Nil) + } + + test("whole STRUCT column MATCH_CONDITION") { + checkSortMergeAsOf( + sql( + """ + |SELECT r.k.tag, r.k.inner_seq + |FROM VALUES (named_struct( + | 'inner_ts', TIMESTAMP '2026-06-29 10:00:00', + | 'inner_seq', 2, + | 'tag', 'a')) AS t(k) + |ASOF JOIN ( + | SELECT * FROM VALUES + | (named_struct( + | 'inner_ts', TIMESTAMP '2026-06-29 09:00:00', + | 'inner_seq', 1, + | 'tag', 'a')), + | (named_struct( + | 'inner_ts', TIMESTAMP '2026-06-29 10:00:00', + | 'inner_seq', 1, + | 'tag', 'a')) AS r(k) + |) r + | MATCH_CONDITION (t.k >= r.k) + |""".stripMargin), + Row("a", 1) :: Nil) + } + + test("nested whole STRUCT column MATCH_CONDITION") { + sql( + """ + |CREATE OR REPLACE TEMP VIEW requests(req_ts, seq, service) AS + |VALUES (TIMESTAMP '2026-06-29 10:00:00', 5, 'api'), + | (TIMESTAMP '2026-06-29 10:03:00', 1, 'api') + |""".stripMargin) + 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'), + | (TIMESTAMP '2026-06-29 10:00:00', 3, 'api', 'v1.1') + |""".stripMargin) + checkSortMergeAsOf( + sql( + """ + |SELECT r.req_ts, r.seq, d.version + |FROM ( + | SELECT named_struct('ts', req_ts, 'seq', seq) AS k, service, req_ts, seq + | FROM requests + |) r + |ASOF JOIN ( + | SELECT named_struct('ts', deploy_ts, 'seq', seq) AS k, service, version, deploy_ts, seq + | FROM deploys + |) d + | MATCH_CONDITION (r.k >= d.k) + | ON r.service = d.service + |ORDER BY r.req_ts, r.seq + |""".stripMargin), + Seq( + Row(Timestamp.valueOf("2026-06-29 10:00:00"), 5, "v1.1"), + Row(Timestamp.valueOf("2026-06-29 10:03:00"), 1, "v1.1"))) + } + + test("ARRAY MATCH_CONDITION") { + checkSortMergeAsOf( + sql( + """ + |SELECT r.a + |FROM VALUES (ARRAY(1, 3)) AS t(a) + |ASOF JOIN VALUES (ARRAY(1, 2)), (ARRAY(1, 4)) AS r(a) + |MATCH_CONDITION (t.a >= r.a) + |""".stripMargin), + Row(Seq(1, 2)) :: Nil) + } + + test("ARRAY whole column MATCH_CONDITION") { + checkSortMergeAsOf( + sql( + """ + |SELECT r.a + |FROM VALUES (ARRAY(named_struct('seq', 1, 'val', 3))) AS t(a) + |ASOF JOIN ( + | SELECT * FROM VALUES + | (ARRAY(named_struct('seq', 1, 'val', 2))), + | (ARRAY(named_struct('seq', 1, 'val', 4))) AS r(a) + |) r + | MATCH_CONDITION (t.a >= r.a) + |""".stripMargin), + Row(Seq(Row(1, 2))) :: Nil) + } + + test("STRUCT tuple from scalar columns MATCH_CONDITION") { + checkSortMergeAsOf( + sql( + """ + |SELECT r.c1, r.c2 + |FROM VALUES (1, 2) AS t(c1, c2) + |ASOF JOIN VALUES (1, 1) AS r(c1, c2) + |MATCH_CONDITION ((t.c1, t.c2) >= (r.c1, r.c2)) + |""".stripMargin), + Row(1, 1) :: Nil) + } + + test("MATCH_CONDITION operand arithmetic with ON") { + setupTradeQuoteViews() + checkSortMergeAsOf( + sql( + """ + |SELECT count(*) AS cnt + |FROM trades t ASOF JOIN quotes q + | MATCH_CONDITION (t.trade_time >= date_trunc('hour', q.quote_time)) + | ON t.symbol = q.symbol + |""".stripMargin), + Row(3L) :: Nil) + } + + test("MATCH_CONDITION rejects cross-side operand references") { + setupTradeQuoteViews() + val sqlText = + """ + |SELECT count(*) + |FROM trades t ASOF JOIN quotes q + | MATCH_CONDITION (t.trade_time + q.quote_time >= q.quote_time) + | ON t.symbol = q.symbol + |""".stripMargin + checkError( + exception = intercept[AnalysisException](sql(sqlText)), + condition = "ASOF_JOIN_MATCH_CONDITION_TABLE_REFERENCE", + sqlState = Some("42K0E"), + parameters = Map( + "refs1" -> "\"(trade_time + quote_time)\"", + "refs2" -> "\"quote_time\""), + queryContext = Array( + ExpectedContext( + fragment = """ASOF JOIN quotes q + | MATCH_CONDITION (t.trade_time + q.quote_time >= q.quote_time) + | ON t.symbol = q.symbol""".stripMargin, + start = 31, + stop = 137))) + } + + test("MATCH_CONDITION rejects invalid table references") { + setupTradeQuoteViews() + val sqlText = + """ + |SELECT count(*) + |FROM trades t ASOF JOIN quotes q + | MATCH_CONDITION (t.symbol >= t.symbol) + | ON t.symbol = q.symbol + |""".stripMargin + checkError( + exception = intercept[AnalysisException](sql(sqlText)), + condition = "ASOF_JOIN_MATCH_CONDITION_TABLE_REFERENCE", + sqlState = Some("42K0E"), + parameters = Map( + "refs1" -> "\"symbol\"", + "refs2" -> "\"symbol\"")) + } + + test("MATCH_CONDITION rejects non-deterministic expressions") { + setupTradeQuoteViews() + val sqlText = + """ + |SELECT t.trade_time + |FROM trades t ASOF JOIN quotes q + | MATCH_CONDITION (rand() >= q.quote_time) + | 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" -> "\"rand()\""), + queryContext = Array( + ExpectedContext( + fragment = """ASOF JOIN quotes q + | MATCH_CONDITION (rand() >= q.quote_time) + | ON t.symbol = q.symbol""".stripMargin, + start = 35, + stop = 120))) + } + + test("MATCH_CONDITION rejects incompatible operand types") { + setupTradeQuoteViews() + val sqlText = + """ + |SELECT t.trade_time + |FROM trades t ASOF JOIN quotes q + | MATCH_CONDITION (t.trade_time >= q.symbol) + | ON t.symbol = q.symbol + |""".stripMargin + checkError( + exception = intercept[AnalysisException](sql(sqlText)), + condition = "ASOF_JOIN_MATCH_CONDITION_INVALID_TYPE", + sqlState = Some("42K09"), + parameters = Map( + "type1" -> "\"TIMESTAMP\"", + "type2" -> "\"STRING\""), + queryContext = Array( + ExpectedContext( + fragment = """ASOF JOIN quotes q + | MATCH_CONDITION (t.trade_time >= q.symbol) + | ON t.symbol = q.symbol""".stripMargin, + start = 35, + stop = 122))) + } + + test("LEFT ASOF JOIN null-pads NOT NULL right catalog columns") { + withTable("right_nn", "left_nn") { + sql("CREATE TABLE right_nn (k INT NOT NULL, v STRING NOT NULL) USING parquet") + sql("INSERT INTO right_nn VALUES (3, 'a')") + sql("CREATE TABLE left_nn (k INT, label STRING) USING parquet") + sql("INSERT INTO left_nn VALUES (2, 'unmatched')") + checkSortMergeAsOf( + sql( + """ + |SELECT l.k, l.label, r.v + |FROM left_nn l LEFT ASOF JOIN right_nn r + | MATCH_CONDITION (l.k >= r.k) + |""".stripMargin), + Row(2, "unmatched", null) :: Nil) + } + } + + test("strict > MATCH_CONDITION excludes equal right rows") { + checkSortMergeAsOf( + sql( + """ + |SELECT q.quote_time + |FROM VALUES (TIMESTAMP '2026-06-29 10:00:00', 'X') AS t(trade_time, symbol) + |ASOF JOIN VALUES + | (TIMESTAMP '2026-06-29 10:00:00', 'X'), + | (TIMESTAMP '2026-06-29 09:00:00', 'X') AS q(quote_time, symbol) + | MATCH_CONDITION (t.trade_time > q.quote_time) + | ON t.symbol = q.symbol + |""".stripMargin), + Row(Timestamp.valueOf("2026-06-29 09:00:00")) :: Nil) + } + + test("strict < forward MATCH_CONDITION") { + checkSortMergeAsOf( + sql( + """ + |SELECT r.ts + |FROM VALUES (TIMESTAMP '2026-06-29 10:00:00') AS t(ts) + |ASOF JOIN VALUES + | (TIMESTAMP '2026-06-29 10:00:00'), + | (TIMESTAMP '2026-06-29 11:00:00') AS r(ts) + | MATCH_CONDITION (t.ts < r.ts) + |""".stripMargin), + Row(Timestamp.valueOf("2026-06-29 11:00:00")) :: Nil) + } + + test("forward ARRAY <= MATCH_CONDITION") { + checkSortMergeAsOf( + sql( + """ + |SELECT r.a + |FROM VALUES (ARRAY(1, 5)) AS t(a) + |ASOF JOIN VALUES (ARRAY(1, 6)), (ARRAY(1, 8)) AS r(a) + | MATCH_CONDITION (t.a <= r.a) + |""".stripMargin), + Row(Seq(1, 6)) :: Nil) + } + + test("ARRAY operands with different field names") { + checkSortMergeAsOf( + sql( + """ + |SELECT r.a + |FROM VALUES (ARRAY(named_struct('x', 1, 'y', 3))) AS t(a) + |ASOF JOIN VALUES (ARRAY(named_struct('p', 1, 'q', 2))) AS r(a) + | MATCH_CONDITION (t.a >= r.a) + |""".stripMargin), + Row(Seq(Row(1, 2))) :: Nil) + } +} diff --git a/sql/core/src/test/scala/org/apache/spark/sql/errors/QueryParsingErrorsSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/errors/QueryParsingErrorsSuite.scala index f834db203b90c..aab49da38ddcc 100644 --- a/sql/core/src/test/scala/org/apache/spark/sql/errors/QueryParsingErrorsSuite.scala +++ b/sql/core/src/test/scala/org/apache/spark/sql/errors/QueryParsingErrorsSuite.scala @@ -472,7 +472,7 @@ class QueryParsingErrorsSuite extends SharedSparkSession { exception = parseException("select * from a left join_ b on a.id = b.id"), condition = "PARSE_SYNTAX_ERROR", sqlState = "42601", - parameters = Map("error" -> "'join_'", "hint" -> ": missing 'JOIN'")) + parameters = Map("error" -> "'join_'", "hint" -> "")) checkError( exception = parseException("select * from test where test.t is like 'test'"), diff --git a/sql/hive-thriftserver/src/test/scala/org/apache/spark/sql/hive/thriftserver/ThriftServerWithSparkContextSuite.scala b/sql/hive-thriftserver/src/test/scala/org/apache/spark/sql/hive/thriftserver/ThriftServerWithSparkContextSuite.scala index 4e8f117dc8a5c..ff9e203f5915d 100644 --- a/sql/hive-thriftserver/src/test/scala/org/apache/spark/sql/hive/thriftserver/ThriftServerWithSparkContextSuite.scala +++ b/sql/hive-thriftserver/src/test/scala/org/apache/spark/sql/hive/thriftserver/ThriftServerWithSparkContextSuite.scala @@ -214,7 +214,7 @@ trait ThriftServerWithSparkContextSuite extends SharedThriftServer { val sessionHandle = client.openSession(user, "") val infoValue = client.getInfo(sessionHandle, GetInfoType.CLI_ODBC_KEYWORDS) // scalastyle:off line.size.limit - assert(infoValue.getStringValue == "ADD,AFTER,AGGREGATE,ALIGN,ALL,ALTER,ALWAYS,ANALYZE,AND,ANTI,ANY,ANY_VALUE,APPLY,APPROX,ARCHIVE,ARRAY,AS,ASC,ASENSITIVE,AT,ATOMIC,AUTHORIZATION,AUTO,BEGIN,BERNOULLI,BETWEEN,BIGINT,BIN,BINARY,BINDING,BIN_DISTRIBUTE_RATIO,BIN_END,BIN_START,BOOLEAN,BOTH,BUCKET,BUCKETS,BY,BYTE,CACHE,CALL,CALLED,CASCADE,CASE,CAST,CATALOG,CATALOGS,CDC,CHANGE,CHANGES,CHAR,CHARACTER,CHECK,CLEAR,CLOSE,CLUSTER,CLUSTERED,CODEGEN,COLLATE,COLLATION,COLLATIONS,COLLECTION,COLUMN,COLUMNS,COMMENT,COMMIT,COMPACT,COMPACTIONS,COMPENSATION,COMPUTE,CONCATENATE,CONDITION,CONSTRAINT,CONTAINS,CONTINUE,COST,CREATE,CROSS,CUBE,CURRENT,CURRENT_DATABASE,CURRENT_DATE,CURRENT_PATH,CURRENT_SCHEMA,CURRENT_TIME,CURRENT_TIMESTAMP,CURRENT_USER,CURSOR,DATA,DATABASE,DATABASES,DATE,DATEADD,DATEDIFF,DATE_ADD,DATE_DIFF,DAY,DAYOFYEAR,DAYS,DBPROPERTIES,DEC,DECIMAL,DECLARE,DEFAULT,DEFAULT_PATH,DEFINED,DEFINER,DELAY,DELETE,DELIMITED,DESC,DESCRIBE,DETERMINISTIC,DFS,DIRECTORIES,DIRECTORY,DISTANCE,DISTINCT,DISTRIBUTE,DIV,DO,DOUBLE,DROP,ELSE,ELSEIF,END,ENFORCED,ESCAPE,ESCAPED,EVOLUTION,EXACT,EXCEPT,EXCHANGE,EXCLUDE,EXCLUSIVE,EXECUTE,EXISTS,EXIT,EXPLAIN,EXPORT,EXTEND,EXTENDED,EXTERNAL,EXTRACT,FALSE,FETCH,FIELDS,FILEFORMAT,FILTER,FIRST,FLOAT,FLOW,FOLLOWING,FOR,FOREIGN,FORMAT,FORMATTED,FOUND,FROM,FULL,FUNCTION,FUNCTIONS,GENERATED,GEOGRAPHY,GEOMETRY,GLOBAL,GRANT,GROUP,GROUPING,HANDLER,HAVING,HOUR,HOURS,IDENTIFIED,IDENTIFIER,IDENTITY,IF,IGNORE,ILIKE,IMMEDIATE,IMPORT,IN,INCLUDE,INCLUSIVE,INCREMENT,INDEX,INDEXES,INNER,INPATH,INPUT,INPUTFORMAT,INSENSITIVE,INSERT,INT,INTEGER,INTERSECT,INTERVAL,INTO,INVOKER,IS,ITEMS,ITERATE,JOIN,JSON,KEY,KEYS,LANGUAGE,LAST,LATERAL,LAZY,LEADING,LEAVE,LEFT,LEVEL,LIKE,LIMIT,LINES,LIST,LOAD,LOCAL,LOCALTIME,LOCATION,LOCK,LOCKS,LOGICAL,LONG,LOOP,MACRO,MAP,MATCHED,MATERIALIZED,MAX,MEASURE,MERGE,METRICS,MICROSECOND,MICROSECONDS,MILLISECOND,MILLISECONDS,MINUS,MINUTE,MINUTES,MODIFIES,MONTH,MONTHS,MSCK,NAME,NAMESPACE,NAMESPACES,NANOSECOND,NANOSECONDS,NATURAL,NEAREST,NEXT,NO,NONE,NORELY,NOT,NULL,NULLS,NUMERIC,OF,OFFSET,ON,ONLY,OPEN,OPTION,OPTIONS,OR,ORDER,OUT,OUTER,OUTPUTFORMAT,OVER,OVERLAPS,OVERLAY,OVERWRITE,PARTITION,PARTITIONED,PARTITIONS,PATH,PERCENT,PIVOT,PLACING,POSITION,PRECEDING,PRIMARY,PRINCIPALS,PROCEDURE,PROCEDURES,PROPERTIES,PURGE,QUALIFY,QUARTER,QUERY,RANGE,READ,READS,REAL,RECORDREADER,RECORDWRITER,RECOVER,RECURSION,RECURSIVE,REDUCE,REFERENCES,REFRESH,RELY,RENAME,REPAIR,REPEAT,REPEATABLE,REPLACE,RESET,RESPECT,RESTRICT,RETURN,RETURNS,REVOKE,RIGHT,ROLE,ROLES,ROLLBACK,ROLLUP,ROW,ROWS,SCHEMA,SCHEMAS,SECOND,SECONDS,SECURITY,SELECT,SEMI,SEPARATED,SEQUENCE,SERDE,SERDEPROPERTIES,SESSION_USER,SET,SETS,SHORT,SHOW,SIMILARITY,SINGLE,SKEWED,SMALLINT,SOME,SORT,SORTED,SOURCE,SPECIFIC,SQL,SQLEXCEPTION,SQLSTATE,START,STATISTICS,STORED,STRATIFY,STREAM,STREAMING,STRING,STRUCT,SUBSTR,SUBSTRING,SYNC,SYSTEM,SYSTEM_PATH,SYSTEM_TIME,SYSTEM_VERSION,TABLE,TABLES,TABLESAMPLE,TARGET,TBLPROPERTIES,TERMINATED,THEN,TIME,TIMEDIFF,TIMESTAMP,TIMESTAMPADD,TIMESTAMPDIFF,TIMESTAMP_LTZ,TIMESTAMP_NTZ,TINYINT,TO,TOUCH,TRAILING,TRANSACTION,TRANSACTIONS,TRANSFORM,TRIM,TRUE,TRUNCATE,TRY_CAST,TYPE,UNARCHIVE,UNBOUNDED,UNCACHE,UNIFORM,UNION,UNIQUE,UNKNOWN,UNLOCK,UNPIVOT,UNSET,UNTIL,UPDATE,USE,USER,USING,VALUE,VALUES,VAR,VARCHAR,VARIABLE,VARIANT,VERSION,VIEW,VIEWS,VOID,WATERMARK,WEEK,WEEKS,WHEN,WHERE,WHILE,WIDTH,WINDOW,WITH,WITHIN,WITHOUT,X,YEAR,YEARS,ZONE") + assert(infoValue.getStringValue == "ADD,AFTER,AGGREGATE,ALIGN,ALL,ALTER,ALWAYS,ANALYZE,AND,ANTI,ANY,ANY_VALUE,APPLY,APPROX,ARCHIVE,ARRAY,AS,ASC,ASENSITIVE,ASOF,AT,ATOMIC,AUTHORIZATION,AUTO,BEGIN,BERNOULLI,BETWEEN,BIGINT,BIN,BINARY,BINDING,BIN_DISTRIBUTE_RATIO,BIN_END,BIN_START,BOOLEAN,BOTH,BUCKET,BUCKETS,BY,BYTE,CACHE,CALL,CALLED,CASCADE,CASE,CAST,CATALOG,CATALOGS,CDC,CHANGE,CHANGES,CHAR,CHARACTER,CHECK,CLEAR,CLOSE,CLUSTER,CLUSTERED,CODEGEN,COLLATE,COLLATION,COLLATIONS,COLLECTION,COLUMN,COLUMNS,COMMENT,COMMIT,COMPACT,COMPACTIONS,COMPENSATION,COMPUTE,CONCATENATE,CONDITION,CONSTRAINT,CONTAINS,CONTINUE,COST,CREATE,CROSS,CUBE,CURRENT,CURRENT_DATABASE,CURRENT_DATE,CURRENT_PATH,CURRENT_SCHEMA,CURRENT_TIME,CURRENT_TIMESTAMP,CURRENT_USER,CURSOR,DATA,DATABASE,DATABASES,DATE,DATEADD,DATEDIFF,DATE_ADD,DATE_DIFF,DAY,DAYOFYEAR,DAYS,DBPROPERTIES,DEC,DECIMAL,DECLARE,DEFAULT,DEFAULT_PATH,DEFINED,DEFINER,DELAY,DELETE,DELIMITED,DESC,DESCRIBE,DETERMINISTIC,DFS,DIRECTORIES,DIRECTORY,DISTANCE,DISTINCT,DISTRIBUTE,DIV,DO,DOUBLE,DROP,ELSE,ELSEIF,END,ENFORCED,ESCAPE,ESCAPED,EVOLUTION,EXACT,EXCEPT,EXCHANGE,EXCLUDE,EXCLUSIVE,EXECUTE,EXISTS,EXIT,EXPLAIN,EXPORT,EXTEND,EXTENDED,EXTERNAL,EXTRACT,FALSE,FETCH,FIELDS,FILEFORMAT,FILTER,FIRST,FLOAT,FLOW,FOLLOWING,FOR,FOREIGN,FORMAT,FORMATTED,FOUND,FROM,FULL,FUNCTION,FUNCTIONS,GENERATED,GEOGRAPHY,GEOMETRY,GLOBAL,GRANT,GROUP,GROUPING,HANDLER,HAVING,HOUR,HOURS,IDENTIFIED,IDENTIFIER,IDENTITY,IF,IGNORE,ILIKE,IMMEDIATE,IMPORT,IN,INCLUDE,INCLUSIVE,INCREMENT,INDEX,INDEXES,INNER,INPATH,INPUT,INPUTFORMAT,INSENSITIVE,INSERT,INT,INTEGER,INTERSECT,INTERVAL,INTO,INVOKER,IS,ITEMS,ITERATE,JOIN,JSON,KEY,KEYS,LANGUAGE,LAST,LATERAL,LAZY,LEADING,LEAVE,LEFT,LEVEL,LIKE,LIMIT,LINES,LIST,LOAD,LOCAL,LOCALTIME,LOCATION,LOCK,LOCKS,LOGICAL,LONG,LOOP,MACRO,MAP,MATCHED,MATCH_CONDITION,MATERIALIZED,MAX,MEASURE,MERGE,METRICS,MICROSECOND,MICROSECONDS,MILLISECOND,MILLISECONDS,MINUS,MINUTE,MINUTES,MODIFIES,MONTH,MONTHS,MSCK,NAME,NAMESPACE,NAMESPACES,NANOSECOND,NANOSECONDS,NATURAL,NEAREST,NEXT,NO,NONE,NORELY,NOT,NULL,NULLS,NUMERIC,OF,OFFSET,ON,ONLY,OPEN,OPTION,OPTIONS,OR,ORDER,OUT,OUTER,OUTPUTFORMAT,OVER,OVERLAPS,OVERLAY,OVERWRITE,PARTITION,PARTITIONED,PARTITIONS,PATH,PERCENT,PIVOT,PLACING,POSITION,PRECEDING,PRIMARY,PRINCIPALS,PROCEDURE,PROCEDURES,PROPERTIES,PURGE,QUALIFY,QUARTER,QUERY,RANGE,READ,READS,REAL,RECORDREADER,RECORDWRITER,RECOVER,RECURSION,RECURSIVE,REDUCE,REFERENCES,REFRESH,RELY,RENAME,REPAIR,REPEAT,REPEATABLE,REPLACE,RESET,RESPECT,RESTRICT,RETURN,RETURNS,REVOKE,RIGHT,ROLE,ROLES,ROLLBACK,ROLLUP,ROW,ROWS,SCHEMA,SCHEMAS,SECOND,SECONDS,SECURITY,SELECT,SEMI,SEPARATED,SEQUENCE,SERDE,SERDEPROPERTIES,SESSION_USER,SET,SETS,SHORT,SHOW,SIMILARITY,SINGLE,SKEWED,SMALLINT,SOME,SORT,SORTED,SOURCE,SPECIFIC,SQL,SQLEXCEPTION,SQLSTATE,START,STATISTICS,STORED,STRATIFY,STREAM,STREAMING,STRING,STRUCT,SUBSTR,SUBSTRING,SYNC,SYSTEM,SYSTEM_PATH,SYSTEM_TIME,SYSTEM_VERSION,TABLE,TABLES,TABLESAMPLE,TARGET,TBLPROPERTIES,TERMINATED,THEN,TIME,TIMEDIFF,TIMESTAMP,TIMESTAMPADD,TIMESTAMPDIFF,TIMESTAMP_LTZ,TIMESTAMP_NTZ,TINYINT,TO,TOUCH,TRAILING,TRANSACTION,TRANSACTIONS,TRANSFORM,TRIM,TRUE,TRUNCATE,TRY_CAST,TYPE,UNARCHIVE,UNBOUNDED,UNCACHE,UNIFORM,UNION,UNIQUE,UNKNOWN,UNLOCK,UNPIVOT,UNSET,UNTIL,UPDATE,USE,USER,USING,VALUE,VALUES,VAR,VARCHAR,VARIABLE,VARIANT,VERSION,VIEW,VIEWS,VOID,WATERMARK,WEEK,WEEKS,WHEN,WHERE,WHILE,WIDTH,WINDOW,WITH,WITHIN,WITHOUT,X,YEAR,YEARS,ZONE") // scalastyle:on line.size.limit } }