cloud-fan commented on code in PR #59834: URL: https://github.com/apache/spark/pull/59834#discussion_r4395611808
########## sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/AsOfJoinMatchConditionResolution.scala: ########## @@ -0,0 +1,210 @@ +/* + * 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.expressions._ +import org.apache.spark.sql.catalyst.plans._ +import org.apache.spark.sql.catalyst.plans.logical.AsOfJoin +import org.apache.spark.sql.internal.SQLConf +import org.apache.spark.sql.types.{DecimalType, DoubleType, IntegralType, NullType} + +/** The executable fields produced by resolving ASOF keys. */ +private[sql] case class MaterializedMatchCondition( + asOfCondition: Expression, + direction: AsOfJoinDirection, + leftSortExpressions: Seq[Expression], + rightSortExpressions: Seq[Expression], + orderExpression: Option[Expression] = None) { + + /** Replaces the pending operands with the resolved comparison and sort keys. */ + def toJoin(join: AsOfJoin): AsOfJoin = { + join.copy( + asOfCondition = asOfCondition, + direction = direction, + leftSortExprs = leftSortExpressions, + rightSortExprs = rightSortExpressions, + matchLeftOperand = None, + matchOperator = None, + matchRightOperand = None, + requiresSortMergeAsOfJoin = orderExpression.isEmpty, + orderExpression = orderExpression + ) + } +} + +/** Shared ASOF key normalization for the fixed-point and single-pass analyzers. */ +private[sql] object AsOfJoinMatchConditionResolution { + + /** Materializes a SQL MATCH_CONDITION once both operands are resolved. */ + def materialize( + join: AsOfJoin, + leftSet: AttributeSet, + rightSet: AttributeSet, + leftOperand: Expression, + operator: MatchComparisonOperator, + rightOperand: Expression): Option[MaterializedMatchCondition] = { + AsOfJoinValidation.validateMatchConditionTableReferences( + join = join, + leftSet = leftSet, + rightSet = rightSet, + leftExpr = leftOperand, + rightExpr = rightOperand + ) + + if (leftOperand.resolved && rightOperand.resolved) { + AsOfJoinValidation.validateMatchConditionOperands( + join = join, + leftExpr = leftOperand, + rightExpr = rightOperand + ) + val (normalizedLeft, normalizedRight, normalizedOperator) = + AsOfJoin.normalizeMatchOperands( + leftSet = leftSet, + rightSet = rightSet, + expr1 = leftOperand, + operator = operator, + expr2 = rightOperand + ) + val (asOfCondition, leftSort, rightSort) = AsOfJoin.materializeMatchComparison( + leftOperand = normalizedLeft, + rightOperand = normalizedRight, + normalizedOp = normalizedOperator + ) + Some(MaterializedMatchCondition( + asOfCondition = asOfCondition, + direction = normalizedOperator.direction, + leftSortExpressions = leftSort, + rightSortExpressions = rightSort + )) + } else { + None + } + } + + /** Resolves DataFrame keys and selects the execution strategy during analysis. */ + def materializeDataFrame( + join: AsOfJoin, + leftSet: AttributeSet, + rightSet: AttributeSet, + leftOperand: Expression, + rightOperand: Expression): Option[MaterializedMatchCondition] = { + if (!leftOperand.resolved || !rightOperand.resolved || !join.tolerance.forall(_.resolved)) { + return None + } + + AsOfJoinValidation.validateTolerance(join) + + val useSortMerge = SQLConf.get.sortMergeAsOfJoinEnabled + if (useSortMerge) { + AsOfJoinValidation.validateMatchConditionTableReferences( + join = join, + leftSet = leftSet, + rightSet = rightSet, + leftExpr = leftOperand, + rightExpr = rightOperand + ) + AsOfJoinValidation.validateMatchConditionOperands( + join = join, + leftExpr = leftOperand, + rightExpr = rightOperand + ) + } + AsOfJoinValidation.validateNearestStringKeys( Review Comment: **Non-blocking (P2):** This check covers only `nearest`, so on the default correlated-subquery path, backward/forward joins on two STRING keys keep the mismatch this PR fixes for sort-merge. The condition from `makeAsOfCond` compares the keys as text, but with ANSI off the `makeOrderingExpr` distance below (line 175) becomes a DOUBLE subtraction, and `RewriteAsOfJoin` picks `MinBy` over it. Example (flag off, ANSI off, backward): left `'50'`, right `'45'` and `'100'`. Both rows pass `'50' >= r` as text, the distances are 5.0 and -50.0, and `MinBy` returns `'100'`. That is neither the text-nearest nor the numeric-nearest row; the flag-on path returns `'45'`. Non-numeric strings get a NULL distance and no match at all. With ANSI on the subtraction already fails analysis, so this affects only ANSI-off sessions. See **Shared repair plan 1** in the review body. ########## sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/AsOfJoinMatchConditionResolution.scala: ########## @@ -0,0 +1,210 @@ +/* + * 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.expressions._ +import org.apache.spark.sql.catalyst.plans._ +import org.apache.spark.sql.catalyst.plans.logical.AsOfJoin +import org.apache.spark.sql.internal.SQLConf +import org.apache.spark.sql.types.{DecimalType, DoubleType, IntegralType, NullType} + +/** The executable fields produced by resolving ASOF keys. */ +private[sql] case class MaterializedMatchCondition( + asOfCondition: Expression, + direction: AsOfJoinDirection, + leftSortExpressions: Seq[Expression], + rightSortExpressions: Seq[Expression], + orderExpression: Option[Expression] = None) { + + /** Replaces the pending operands with the resolved comparison and sort keys. */ + def toJoin(join: AsOfJoin): AsOfJoin = { + join.copy( + asOfCondition = asOfCondition, + direction = direction, + leftSortExprs = leftSortExpressions, + rightSortExprs = rightSortExpressions, + matchLeftOperand = None, + matchOperator = None, + matchRightOperand = None, + requiresSortMergeAsOfJoin = orderExpression.isEmpty, + orderExpression = orderExpression + ) + } +} + +/** Shared ASOF key normalization for the fixed-point and single-pass analyzers. */ +private[sql] object AsOfJoinMatchConditionResolution { + + /** Materializes a SQL MATCH_CONDITION once both operands are resolved. */ + def materialize( + join: AsOfJoin, + leftSet: AttributeSet, + rightSet: AttributeSet, + leftOperand: Expression, + operator: MatchComparisonOperator, + rightOperand: Expression): Option[MaterializedMatchCondition] = { + AsOfJoinValidation.validateMatchConditionTableReferences( + join = join, + leftSet = leftSet, + rightSet = rightSet, + leftExpr = leftOperand, + rightExpr = rightOperand + ) + + if (leftOperand.resolved && rightOperand.resolved) { + AsOfJoinValidation.validateMatchConditionOperands( + join = join, + leftExpr = leftOperand, + rightExpr = rightOperand + ) + val (normalizedLeft, normalizedRight, normalizedOperator) = + AsOfJoin.normalizeMatchOperands( + leftSet = leftSet, + rightSet = rightSet, + expr1 = leftOperand, + operator = operator, + expr2 = rightOperand + ) + val (asOfCondition, leftSort, rightSort) = AsOfJoin.materializeMatchComparison( + leftOperand = normalizedLeft, + rightOperand = normalizedRight, + normalizedOp = normalizedOperator + ) + Some(MaterializedMatchCondition( + asOfCondition = asOfCondition, + direction = normalizedOperator.direction, + leftSortExpressions = leftSort, + rightSortExpressions = rightSort + )) + } else { + None + } + } + + /** Resolves DataFrame keys and selects the execution strategy during analysis. */ + def materializeDataFrame( + join: AsOfJoin, + leftSet: AttributeSet, + rightSet: AttributeSet, + leftOperand: Expression, + rightOperand: Expression): Option[MaterializedMatchCondition] = { + if (!leftOperand.resolved || !rightOperand.resolved || !join.tolerance.forall(_.resolved)) { + return None + } + + AsOfJoinValidation.validateTolerance(join) + + val useSortMerge = SQLConf.get.sortMergeAsOfJoinEnabled + if (useSortMerge) { + AsOfJoinValidation.validateMatchConditionTableReferences( + join = join, + leftSet = leftSet, + rightSet = rightSet, + leftExpr = leftOperand, + rightExpr = rightOperand + ) + AsOfJoinValidation.validateMatchConditionOperands( + join = join, + leftExpr = leftOperand, + rightExpr = rightOperand + ) + } + AsOfJoinValidation.validateNearestStringKeys( + join = join, + leftKey = leftOperand, + rightKey = rightOperand + ) + val asOfCondition = AsOfJoin.makeAsOfCond( Review Comment: **Non-blocking (P2):** With the sort-merge flag on, STRING keys are now sorted as text, but this `asOfCondition` still carries a numeric tolerance bound. `Subtract`/`Add` of the left key and the tolerance is coerced to DOUBLE (BIGINT with ANSI), the right key is cast to match, and nothing rejects a tolerance on two STRING keys. Once a row has matched, `findLastBackward` stops at the first failing row, which is sound only if the bound is monotone in the right sort order. Example: backward, tolerance `lit(10)`, left `'50'`, right `'41'`, `'5'` and `'50'`. The text-sorted buffer is `'41' < '5' < '50'`, and the condition evaluates to true, false, true, so the scan returns `'41'` and misses the exact match `'50'`. The merge base (numeric sort) and the fallback both return `'50'`. Forward joins return the first eligible row and go wrong only when a residual condition rejects it. With ANSI on, non-numeric keys fail at runtime in the bound's cast instead. See **Shared repair plan 1** in the review body. ########## sql/core/src/test/scala/org/apache/spark/sql/SortMergeAsOfJoinSuite.scala: ########## @@ -1244,52 +1347,254 @@ class SortMergeAsOfJoinSuite extends QueryTest } gridTest("nearest join rejects STRING keys during analysis")(Seq(false, true)) { singlePass => - // With ANSI on, analysis already rejects the subtraction of two strings. withAsOfJoinAnalyzer(singlePass) { - withSQLConf(SQLConf.ANSI_ENABLED.key -> "false") { - val left = spark.sql("SELECT '50' AS k") - val right = spark.sql("SELECT '10' AS k, 'ten' AS tag") + for (ansi <- Seq(false, true)) { + withSQLConf( + SQLConf.SORT_MERGE_AS_OF_JOIN_ENABLED.key -> "true", Review Comment: **Nit (P3):** This test, like every `UNSUPPORTED_NEAREST_KEY_TYPE` assertion in the suite, runs with the sort-merge flag on, and `DataFrameAsOfJoinSuite` has no STRING nearest case. The declared flag-off rejection is therefore untested: moving the `validateNearestStringKeys` call into the `useSortMerge` block would keep the suite green. The off-then-on direction of the analysis-time strategy binding has the same gap. The removed 'supported keys use sort-merge after enabling the config' and 'mixed timestamp joins use fallback after enabling sort-merge' were the only tests that analyzed with the flag off and executed with it on; the new tests cover only on-then-off. **Verification:** - **Regression:** With the sort-merge flag off, a nearest joinAsOf on two STRING keys fails analysis with the STRING-key error under both analyzers and both ANSI modes. - **Regression:** A DataFrame and a temp view built from it, analyzed with the sort-merge flag off and executed after the flag is enabled, follow the strategy-binding contract the author keeps. Under the current design, that means the correlated-subquery rewrite and no SortMergeAsOfJoinExec. ########## sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/basicLogicalOperators.scala: ########## @@ -2965,200 +2993,32 @@ object AsOfJoin { } } - /** - * Returns the `asOfCondition` and `orderExpression` for one MATCH_CONDITION comparison. The - * operands are compared as whole values, so a NULL struct never matches, while a struct whose - * fields are all NULL can. - */ - private def buildMatchExpressions( - leftOperand: Expression, - rightOperand: Expression, - operator: MatchComparisonOperator): (Expression, Expression) = { - val orderExpression = buildOrderExpression(leftOperand, rightOperand, operator) - operator match { - case GreaterThanOrEqualOp => - (GreaterThanOrEqual(leftOperand, rightOperand), orderExpression) - case GreaterThanOp => - (GreaterThan(leftOperand, rightOperand), orderExpression) - case LessThanOrEqualOp => - (LessThanOrEqual(leftOperand, rightOperand), orderExpression) - case LessThanOp => - (LessThan(leftOperand, rightOperand), orderExpression) - } - } - - private def buildOrderExpression( - leftOperand: Expression, - rightOperand: Expression, - operator: MatchComparisonOperator): Expression = { - (leftOperand.dataType, rightOperand.dataType) match { - case (_: ArrayType, _) - 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, operator) - case (leftType, rightType) - if MatchConditionTypes.usesStructDecomposition(leftType, rightType) => - buildFlattenedStructOrderExpression( - leftOperand, rightOperand, leftType.asInstanceOf[StructType], operator) - case _ => - buildLeafOrderExpression(leftOperand, rightOperand, operator) - } - } - - 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[catalyst] 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, - operator: MatchComparisonOperator): Expression = { - val leftElementType = leftOperand.dataType.asInstanceOf[ArrayType].elementType - val rightElementType = rightOperand.dataType.asInstanceOf[ArrayType].elementType - // Both array inputs and the ZipWith lambda variables must share the element type the binary - // comparison uses. comparisonCommonType is the single source the type check also reads, so - // the two cannot disagree. castArrayElementType widens a coercible side (e.g. INT to BIGINT) - // and leaves a structurally equal side untouched. - val elementType = - MatchConditionTypes.comparisonCommonType(leftElementType, rightElementType) match { - case Some(commonElementType) => commonElementType - case None => - // Unreachable: usesArrayOrderExpression already required a common element type here. - throw SparkException.internalError( - "MATCH_CONDITION array elements have no common type") - } - val leftArray = castArrayElementType(leftOperand, elementType) - val rightArray = castArrayElementType(rightOperand, elementType) - elementType match { - // An empty struct has no fields to split into, so it takes the whole-value arm below. - case struct: StructType if struct.nonEmpty => - 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) - ZipWith( - leftArray, - rightArray, - 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( - leftArray, - rightArray, - LambdaFunction(elementOrder, Seq(leftElement, rightElement))) - } - } - - /** Cast an array operand to the given element type, keeping its own `containsNull`. */ - private def castArrayElementType(operand: Expression, elementType: DataType): Expression = { - val arrayType = operand.dataType.asInstanceOf[ArrayType] - // Skip the cast when the element already matches structurally, the rule the type check uses, - // so a structurally equal side keeps its own field names instead of being renamed by a cast. - val alreadyMatches = - DataType.equalsStructurally(arrayType.elementType, elementType, ignoreNullability = true) - if (alreadyMatches) { - operand - } else { - Cast(operand, ArrayType(elementType, arrayType.containsNull)) - } - } - - 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) - } - - private def collectStructLeafPairs( - leftOperand: Expression, - rightOperand: Expression, - structType: StructType): Seq[(Expression, Expression)] = { - structFieldExprs(leftOperand, structType) - .zip(structFieldExprs(rightOperand, structType)) - .flatMap { - case (left, right) => - if (MatchConditionTypes.usesStructDecomposition(left.dataType, right.dataType)) { - collectStructLeafPairs( - left, right, left.dataType.asInstanceOf[StructType]) - } else { - Seq((left, right)) - } - } - } - - /** Groups multiple field distances in a struct because their types may differ. */ - private def wrapCompositeOrderExpression(diffs: Seq[Expression]): Expression = { - diffs match { - case Seq(single) => single - case _ => CreateStruct(diffs) + /** Builds the subtraction-based distance for a DataFrame fallback join. */ + private[catalyst] def makeOrderingExpr( + leftAsOf: Expression, + rightAsOf: Expression, + direction: AsOfJoinDirection, + evalContext: NumericEvalContext): Expression = { + def subtract(left: Expression, right: Expression): Expression = { + Subtract(left = left, right = right, evalContext = evalContext) } - } - 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))) + direction match { + case Backward => subtract(leftAsOf, rightAsOf) + case Forward => subtract(rightAsOf, leftAsOf) + case Nearest => + If(GreaterThan(leftAsOf, rightAsOf), + subtract(leftAsOf, rightAsOf), subtract(rightAsOf, leftAsOf)) } } - private def makeAsOfCond( + private[catalyst] def makeAsOfCond( leftAsOf: Expression, rightAsOf: Expression, tolerance: Option[Expression], allowExactMatches: Boolean, - direction: AsOfJoinDirection): Expression = { + direction: AsOfJoinDirection, + evalContext: NumericEvalContext = NumericEvalContext.fromSQLConf(SQLConf.get)): Expression = { Review Comment: **Nit (P3):** The only caller (`materializeDataFrame`) passes `join.evalContext`, so this default is never used. It also lets a future caller build the tolerance bounds from the analysis-time `SQLConf` instead of the context captured on the join. `makeOrderingExpr` takes the same parameter without a default, and dropping the default here keeps the two consistent: ```suggestion evalContext: NumericEvalContext): Expression = { ``` **Verification:** - **Compile:** Catalyst compiles with every as-of condition build passing an explicit evalContext. - **Inspection:** No call site builds tolerance bounds from an implicit, analysis-time arithmetic context. ########## sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolveAsOfJoin.scala: ########## @@ -18,279 +18,80 @@ package org.apache.spark.sql.catalyst.analysis import org.apache.spark.sql.catalyst.SQLConfHelper -import org.apache.spark.sql.catalyst.expressions.{ - Cast, - Expression, - GreaterThan, - If, - SubqueryExpression, - SubtractTimestamps, - 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.{AsOfJoinDirection, MatchComparisonOperator} import org.apache.spark.sql.catalyst.plans.logical.{AsOfJoin, LogicalPlan} -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.trees.TreePattern.AS_OF_JOIN import org.apache.spark.sql.catalyst.util._ -import org.apache.spark.sql.errors.QueryErrorsBase -import org.apache.spark.sql.types.{DataType, StringType, TimestampFamily} /** - * Resolves SQL [[AsOfJoin]] operators: materializes `MATCH_CONDITION` into the fields of + * Resolves [[AsOfJoin]] operators: materializes `MATCH_CONDITION` into the fields of Review Comment: **Nit (P3):** Now that this says `[[AsOfJoin]]` rather than SQL `[[AsOfJoin]]`, it should also cover the new `(Some, None, Some)` branch below: DataFrame `joinAsOf` keys are resolved through `materializeDataFrame`, which validates the tolerance and picks the execution strategy during analysis. Otherwise readers won't learn that the strategy decision is made in this rule. **Verification:** - **Inspection:** The rule's scaladoc describes every branch of its match, including the DataFrame key branch and its analysis-time strategy choice. ########## sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/AsOfJoinValidation.scala: ########## @@ -0,0 +1,142 @@ +/* + * 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.{AttributeSet, Expression, SubqueryExpression, WindowExpression} +import org.apache.spark.sql.catalyst.expressions.aggregate.AggregateExpression +import org.apache.spark.sql.catalyst.plans.Nearest +import org.apache.spark.sql.catalyst.plans.logical.AsOfJoin +import org.apache.spark.sql.catalyst.plans.logical.AsOfJoin.MatchConditionTypes +import org.apache.spark.sql.catalyst.trees.TreePattern.GENERATOR +import org.apache.spark.sql.errors.{QueryCompilationErrors, QueryErrorsBase} +import org.apache.spark.sql.types.StringType + +/** Shared validation for SQL and DataFrame ASOF joins. */ +private[sql] object AsOfJoinValidation extends QueryErrorsBase with SQLConfHelper { + + /** Rejects a MATCH_CONDITION operand that references both inputs. */ + def validateMatchConditionTableReferences( + join: AsOfJoin, + leftSet: AttributeSet, + rightSet: AttributeSet, + leftExpr: Expression, + rightExpr: Expression): Unit = { + 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))) + } + } + + /** Validates ASOF key expressions, types and input references. */ + 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 (!MatchConditionTypes.areOperandsCompatible(leftExpr.dataType, rightExpr.dataType)) { + join.failAnalysis( + errorClass = "ASOF_JOIN_MATCH_CONDITION_INVALID_TYPE", + messageParameters = Map( + "type1" -> toSQLType(leftExpr.dataType), + "type2" -> toSQLType(rightExpr.dataType))) + } + + if (leftExpr.references.isEmpty || rightExpr.references.isEmpty) { + throw QueryCompilationErrors.asOfJoinMatchConditionTableReferenceError(leftExpr, rightExpr) + } + } + + /** Validates the type, foldability and sign of a DataFrame ASOF tolerance. */ + def validateTolerance(join: AsOfJoin): Unit = { + join.toleranceAssertion.foreach { assertion => + if (assertion.checkInputDataTypes().isFailure) { + TypeCoercionValidation.failOnTypeCheckResult(assertion, Some(join)) + } + if (!assertion.foldable) { + join.failAnalysis( + errorClass = "AS_OF_JOIN.TOLERANCE_IS_UNFOLDABLE", + messageParameters = Map.empty + ) + } + if (!assertion.eval().asInstanceOf[Boolean]) { + join.failAnalysis( + errorClass = "AS_OF_JOIN.TOLERANCE_IS_NON_NEGATIVE", + messageParameters = Map.empty + ) + } + } + } + + /** Rejects nearest STRING matching for either DataFrame execution strategy. */ + def validateNearestStringKeys( + join: AsOfJoin, + leftKey: Expression, + rightKey: Expression): Unit = { + if (join.direction == Nearest && + leftKey.dataType.isInstanceOf[StringType] && + rightKey.dataType.isInstanceOf[StringType]) { + validateNearestKeys( Review Comment: **Nit (P3):** `validateNearestStringKeys` now runs for both strategies, but it reuses the sort-merge check here. A default-path nearest join on STRING keys therefore fails with `AS_OF_JOIN.UNSUPPORTED_NEAREST_KEY_TYPE`, whose message reads `Nearest sort-merge as-of joins do not support "STRING" keys`. With `spark.sql.join.sortMergeAsOfJoin.enabled` at its default, the plan never uses sort-merge, so the message points users at an internal flag they never set. The cast advice is fine; only the strategy wording is wrong. See **Shared repair plan 1** in the review body. ########## sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/AsOfJoinMatchConditionResolution.scala: ########## @@ -0,0 +1,210 @@ +/* + * 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.expressions._ +import org.apache.spark.sql.catalyst.plans._ +import org.apache.spark.sql.catalyst.plans.logical.AsOfJoin +import org.apache.spark.sql.internal.SQLConf +import org.apache.spark.sql.types.{DecimalType, DoubleType, IntegralType, NullType} + +/** The executable fields produced by resolving ASOF keys. */ +private[sql] case class MaterializedMatchCondition( + asOfCondition: Expression, + direction: AsOfJoinDirection, + leftSortExpressions: Seq[Expression], + rightSortExpressions: Seq[Expression], + orderExpression: Option[Expression] = None) { + + /** Replaces the pending operands with the resolved comparison and sort keys. */ + def toJoin(join: AsOfJoin): AsOfJoin = { + join.copy( + asOfCondition = asOfCondition, + direction = direction, + leftSortExprs = leftSortExpressions, + rightSortExprs = rightSortExpressions, + matchLeftOperand = None, + matchOperator = None, + matchRightOperand = None, + requiresSortMergeAsOfJoin = orderExpression.isEmpty, + orderExpression = orderExpression + ) + } +} + +/** Shared ASOF key normalization for the fixed-point and single-pass analyzers. */ +private[sql] object AsOfJoinMatchConditionResolution { + + /** Materializes a SQL MATCH_CONDITION once both operands are resolved. */ + def materialize( + join: AsOfJoin, + leftSet: AttributeSet, + rightSet: AttributeSet, + leftOperand: Expression, + operator: MatchComparisonOperator, + rightOperand: Expression): Option[MaterializedMatchCondition] = { + AsOfJoinValidation.validateMatchConditionTableReferences( + join = join, + leftSet = leftSet, + rightSet = rightSet, + leftExpr = leftOperand, + rightExpr = rightOperand + ) + + if (leftOperand.resolved && rightOperand.resolved) { + AsOfJoinValidation.validateMatchConditionOperands( + join = join, + leftExpr = leftOperand, + rightExpr = rightOperand + ) + val (normalizedLeft, normalizedRight, normalizedOperator) = + AsOfJoin.normalizeMatchOperands( + leftSet = leftSet, + rightSet = rightSet, + expr1 = leftOperand, + operator = operator, + expr2 = rightOperand + ) + val (asOfCondition, leftSort, rightSort) = AsOfJoin.materializeMatchComparison( + leftOperand = normalizedLeft, + rightOperand = normalizedRight, + normalizedOp = normalizedOperator + ) + Some(MaterializedMatchCondition( + asOfCondition = asOfCondition, + direction = normalizedOperator.direction, + leftSortExpressions = leftSort, + rightSortExpressions = rightSort + )) + } else { + None + } + } + + /** Resolves DataFrame keys and selects the execution strategy during analysis. */ + def materializeDataFrame( + join: AsOfJoin, + leftSet: AttributeSet, + rightSet: AttributeSet, + leftOperand: Expression, + rightOperand: Expression): Option[MaterializedMatchCondition] = { + if (!leftOperand.resolved || !rightOperand.resolved || !join.tolerance.forall(_.resolved)) { + return None + } + + AsOfJoinValidation.validateTolerance(join) + + val useSortMerge = SQLConf.get.sortMergeAsOfJoinEnabled + if (useSortMerge) { + AsOfJoinValidation.validateMatchConditionTableReferences( Review Comment: **Nit (P3):** Reusing the SQL validation here is fine, but DataFrame callers now get SQL-specific messages. For example, `left.joinAsOf(right, lit(0), right.col("rk"), ...)` fails with "Each operand of the MATCH_CONDITION comparison must reference exactly one join input...", and mixed collations fail with "The MATCH_CONDITION operands must be of an orderable, mutually comparable type". A `joinAsOf` caller has no MATCH_CONDITION clause. This shows up only behind the internal flag today, but it becomes the default experience once sort-merge is the default. Could the validators take the surface they are checking, so DataFrame keys get as-of-key wording while SQL errors stay unchanged? **Verification:** - **Behavior:** With the sort-merge flag on, joinAsOf calls with constant, cross-side, nondeterministic or mixed-collation keys fail in both analyzers with errors that describe as-of join keys, not a MATCH_CONDITION clause. - **Compatibility:** SQL ASOF JOIN queries with invalid MATCH_CONDITION operands keep their existing error conditions and messages. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
