You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@spark.apache.org by "srielau (via GitHub)" <gi...@apache.org> on 2026/07/15 01:39:52 UTC
[PR] [SPARK-58123][SQL] Add ResolveAsOfJoin analysis for SQL ASOF JOIN [spark]
srielau opened a new pull request, #57264:
URL: https://github.com/apache/spark/pull/57264
### What changes were proposed in this pull request?
Second stacked PR for [SPARK-58092](https://issues.apache.org/jira/browse/SPARK-58092) / [SPARK-58123](https://issues.apache.org/jira/browse/SPARK-58123).
**Depends on #57251** (parser). Until that lands, the diff vs `master` includes parser commits; the analysis-only delta is `SPARK-58122..SPARK-58123` (1 commit). After #57251 merges, this branch will be rebased onto `master` for a clean diff.
**Analysis rule**
- Add `ResolveAsOfJoin` to materialize `MATCH_CONDITION` into `asOfCondition`, `orderExpression`, and per-side sort expressions
- Expand `USING (...)` into equi-join predicates via `NaturalAndUsingJoinResolution`
- Validate MATCH_CONDITION operands: cross-side references, determinism, and type compatibility (including struct/array)
**Logical plan**
- Extend `AsOfJoin` with `leftSortExprs`, `rightSortExprs`, and `requiresSortMergeAsOfJoin`
- Add `resolveMatchComparison` / `matchSortExpressions` helpers shared with the DataFrame API path
- Teach `Analyzer` to resolve `AsOfMatchCondition` operands before materialization
**Gating**
- `CheckAnalysis` rejects SQL ASOF plans when `spark.sql.join.sortMergeAsOfJoin.enabled=false` (`AS_OF_JOIN.SORT_MERGE_REQUIRED`)
**Out of scope (follow-up PRs)**
- Sort-merge physical operator wiring for multi-column sort keys
- `AsOfJoinSortMergeSQLSuite` execution / success-path integration tests
### Why are the changes needed?
Parser-only (#57251) produces unresolved `AsOfJoin` nodes with a parsed `AsOfMatchCondition`. Analysis must resolve operands, validate semantics, and produce the same logical shape the existing DataFrame `joinAsOf` path uses before physical planning can run.
### Does this PR introduce _any_ user-facing change?
Yes (behind `spark.sql.join.asofJoin.enabled`, default off). With the parser conf enabled, SQL ASOF JOIN now progresses through analysis: valid queries become analyzable logical plans; invalid MATCH_CONDITION expressions fail with structured analysis errors instead of reaching execution unresolved.
### How was this patch tested?
Added analysis negative tests to `AsOfJoinSQLSuite`:
- `AS_OF_JOIN.SORT_MERGE_REQUIRED`
- `ASOF_JOIN_MATCH_CONDITION_TABLE_REFERENCE` (cross-side and invalid refs)
- `ASOF_JOIN_MATCH_CONDITION_INVALID_EXPRESSION`
- `ASOF_JOIN_MATCH_CONDITION_INVALID_TYPE`
Locally:
```
build/sbt "sql/testOnly org.apache.spark.sql.AsOfJoinSQLSuite" \
"catalyst/testOnly org.apache.spark.sql.catalyst.optimizer.RewriteAsOfJoinSuite"
```
### Was this patch authored or co-authored using generative AI tooling?
No.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-58123][SQL] Add ResolveAsOfJoin analysis for SQL ASOF JOIN [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on code in PR #57264:
URL: https://github.com/apache/spark/pull/57264#discussion_r3591859448
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/basicLogicalOperators.scala:
##########
@@ -2621,7 +2624,287 @@ object AsOfJoin {
usingColumns = usingColumns,
matchLeftOperand = Some(leftExpr),
matchOperator = Some(operator),
- matchRightOperand = Some(rightExpr))
+ matchRightOperand = Some(rightExpr),
+ requiresSortMergeAsOfJoin = true)
+ }
+
+ def resolveMatchComparison(
Review Comment:
`resolveMatchComparison` has no callers now — `ResolveAsOfJoin` calls `normalizeMatchOperands` and `materializeMatchComparison` directly, and there's no other caller in the tree. Drop it, unless it's meant as the DataFrame-path entry point for the follow-up, in which case wire that up so it isn't dead.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolveAsOfJoin.scala:
##########
@@ -0,0 +1,253 @@
+/*
+ * 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.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,
+ CalendarIntervalType,
+ DataType,
+ DatetimeType,
+ DateType,
+ DayTimeIntervalType,
+ NumericType,
+ StringType,
+ StructType,
+ TimestampNTZType,
+ TimestampType,
+ YearMonthIntervalType
+}
+
+/**
+ * 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,
+ matchLeft,
+ matchOp,
+ matchRight,
+ _,
+ _,
+ _)
+ 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 = (matchLeft, matchOp, matchRight) match {
+ case (Some(leftExpr), Some(operator), Some(rightExpr)) =>
+ AsOfJoinValidation.validateMatchConditionTableReferences(
+ joinBase, left, right, leftExpr, rightExpr)
+ if (leftExpr.resolved && rightExpr.resolved) {
+ AsOfJoinValidation.validateMatchConditionOperands(joinBase, leftExpr, rightExpr)
+ val (leftOperand, rightOperand, normalizedOp) =
+ AsOfJoin.normalizeMatchOperands(left, right, leftExpr, operator, rightExpr)
+ AsOfJoinValidation.validateMatchConditionPlannerSupport(
+ joinBase, leftOperand, rightOperand)
+ val (asOfCondition, orderExpression, leftSortExprs, rightSortExprs) =
+ AsOfJoin.materializeMatchComparison(leftOperand, rightOperand, normalizedOp)
+ joinBase.copy(
+ asOfCondition = asOfCondition,
+ orderExpression = orderExpression,
+ leftSortExprs = leftSortExprs,
+ rightSortExprs = rightSortExprs,
+ matchLeftOperand = None,
+ matchOperator = None,
+ matchRightOperand = None)
+ } else {
+ joinBase
+ }
+ case (None, None, None) => 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)))
+ }
+ }
+
+ def validateMatchConditionPlannerSupport(
+ join: AsOfJoin,
+ leftOperand: Expression,
+ rightOperand: Expression): Unit = {
+ if (!areScalarSubtractBasedOperands(leftOperand, rightOperand)) {
+ join.failAnalysis(
+ errorClass = "AS_OF_JOIN.UNSUPPORTED_MATCH_CONDITION_OPERAND",
+ messageParameters = Map(
+ "type1" -> toSQLType(leftOperand.dataType),
+ "type2" -> toSQLType(rightOperand.dataType)))
+ }
+ }
+
+ /**
+ * Until multi-column sort-merge ASOF execution lands (SPARK-58124), the planner can only
+ * consume MATCH_CONDITION plans whose `orderExpression` is a scalar `Subtract`. STRING and
+ * composite operands use other distance expressions that `findFromOrder` cannot parse.
+ */
+ private def areScalarSubtractBasedOperands(
+ leftExpr: Expression,
+ rightExpr: Expression): Boolean = {
+ supportsSubtractForPlanner(leftExpr.dataType) && supportsSubtractForPlanner(rightExpr.dataType)
+ }
+
+ private def supportsSubtractForPlanner(dataType: DataType): Boolean = {
Review Comment:
This is a verbatim copy of `AsOfJoin.supportsSubtract` (same type list). The two are coupled by construction — the planner-support gate must accept exactly the types `buildLeafOrderExpression` routes down the `Subtract` path — so keeping two copies invites silent drift if one side later gains a type. Reuse the existing predicate rather than duplicating it (e.g. expose `AsOfJoin.supportsSubtract` and call it here).
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-58123][SQL] Add ResolveAsOfJoin analysis for SQL ASOF JOIN [spark]
Posted by "srielau (via GitHub)" <gi...@apache.org>.
srielau commented on code in PR #57264:
URL: https://github.com/apache/spark/pull/57264#discussion_r3592681792
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/basicLogicalOperators.scala:
##########
@@ -2621,7 +2624,287 @@ object AsOfJoin {
usingColumns = usingColumns,
matchLeftOperand = Some(leftExpr),
matchOperator = Some(operator),
- matchRightOperand = Some(rightExpr))
+ matchRightOperand = Some(rightExpr),
+ requiresSortMergeAsOfJoin = true)
+ }
+
+ def resolveMatchComparison(
Review Comment:
Dropped in `e44becab19e` — `ResolveAsOfJoin` already calls `normalizeMatchOperands` and `materializeMatchComparison` directly, and there are no other callers. Not reserving it for the DataFrame path; that path still goes through `AsOfJoin.apply` + `RewriteAsOfJoin`.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolveAsOfJoin.scala:
##########
@@ -0,0 +1,253 @@
+/*
+ * 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.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,
+ CalendarIntervalType,
+ DataType,
+ DatetimeType,
+ DateType,
+ DayTimeIntervalType,
+ NumericType,
+ StringType,
+ StructType,
+ TimestampNTZType,
+ TimestampType,
+ YearMonthIntervalType
+}
+
+/**
+ * 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,
+ matchLeft,
+ matchOp,
+ matchRight,
+ _,
+ _,
+ _)
+ 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 = (matchLeft, matchOp, matchRight) match {
+ case (Some(leftExpr), Some(operator), Some(rightExpr)) =>
+ AsOfJoinValidation.validateMatchConditionTableReferences(
+ joinBase, left, right, leftExpr, rightExpr)
+ if (leftExpr.resolved && rightExpr.resolved) {
+ AsOfJoinValidation.validateMatchConditionOperands(joinBase, leftExpr, rightExpr)
+ val (leftOperand, rightOperand, normalizedOp) =
+ AsOfJoin.normalizeMatchOperands(left, right, leftExpr, operator, rightExpr)
+ AsOfJoinValidation.validateMatchConditionPlannerSupport(
+ joinBase, leftOperand, rightOperand)
+ val (asOfCondition, orderExpression, leftSortExprs, rightSortExprs) =
+ AsOfJoin.materializeMatchComparison(leftOperand, rightOperand, normalizedOp)
+ joinBase.copy(
+ asOfCondition = asOfCondition,
+ orderExpression = orderExpression,
+ leftSortExprs = leftSortExprs,
+ rightSortExprs = rightSortExprs,
+ matchLeftOperand = None,
+ matchOperator = None,
+ matchRightOperand = None)
+ } else {
+ joinBase
+ }
+ case (None, None, None) => 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)))
+ }
+ }
+
+ def validateMatchConditionPlannerSupport(
+ join: AsOfJoin,
+ leftOperand: Expression,
+ rightOperand: Expression): Unit = {
+ if (!areScalarSubtractBasedOperands(leftOperand, rightOperand)) {
+ join.failAnalysis(
+ errorClass = "AS_OF_JOIN.UNSUPPORTED_MATCH_CONDITION_OPERAND",
+ messageParameters = Map(
+ "type1" -> toSQLType(leftOperand.dataType),
+ "type2" -> toSQLType(rightOperand.dataType)))
+ }
+ }
+
+ /**
+ * Until multi-column sort-merge ASOF execution lands (SPARK-58124), the planner can only
+ * consume MATCH_CONDITION plans whose `orderExpression` is a scalar `Subtract`. STRING and
+ * composite operands use other distance expressions that `findFromOrder` cannot parse.
+ */
+ private def areScalarSubtractBasedOperands(
+ leftExpr: Expression,
+ rightExpr: Expression): Boolean = {
+ supportsSubtractForPlanner(leftExpr.dataType) && supportsSubtractForPlanner(rightExpr.dataType)
+ }
+
+ private def supportsSubtractForPlanner(dataType: DataType): Boolean = {
Review Comment:
Agreed — duplicated predicate removed in `e44becab19e`. `AsOfJoin.supportsSubtract` is now `private[catalyst]` and `validateMatchConditionPlannerSupport` calls it for both operands, so the analysis gate stays coupled to `buildLeafOrderExpression`.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-58123][SQL] Add ResolveAsOfJoin analysis for SQL ASOF JOIN [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan closed pull request #57264: [SPARK-58123][SQL] Add ResolveAsOfJoin analysis for SQL ASOF JOIN
URL: https://github.com/apache/spark/pull/57264
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-58123][SQL] Add ResolveAsOfJoin analysis for SQL ASOF JOIN [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on code in PR #57264:
URL: https://github.com/apache/spark/pull/57264#discussion_r3589525981
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolveAsOfJoin.scala:
##########
@@ -0,0 +1,202 @@
+/*
+ * 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.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,
+ matchLeft,
+ matchOp,
+ matchRight,
+ _,
+ _,
+ _)
+ 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 = (matchLeft, matchOp, matchRight) match {
+ case (Some(leftExpr), Some(operator), Some(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,
+ matchLeftOperand = None,
+ matchOperator = None,
+ matchRightOperand = None)
+ } else {
+ joinBase
+ }
+ case (None, None, None) => 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) ||
Review Comment:
This type check accepts any orderable, type-compatible operands, but the only planner consumer can't handle all of them. When `sortMergeAsOfJoin.enabled` is on, a SQL AsOfJoin isn't rewritten and reaches `AsOfJoinSelection.findFromOrder` (SparkStrategies.scala:514-528), which only recognizes `Subtract` / `If(_, Subtract, _)`. But a plain STRING operand produces `If(EqualTo, 0, If(GreaterThan, 1, -1))` (via `buildSignedComparisonDistance`), and struct/array operands produce `CreateStruct`/`ZipWith` — none of which `findFromOrder` matches. So e.g. `MATCH_CONDITION (t.symbol >= q.symbol)` (both confs on) passes analysis, skips the rewrite, then throws `IllegalStateException("Cannot extract as-of keys from order expression: ...")` at planning.
I traced this from source but didn't run a build — can you confirm the planning-time throw? If it reproduces, gating the operand types this analysis supports down to what the physical path can consume (with a clear analysis error for the rest) until the composite/multi-column sort-merge operator lands would keep supported-syntax queries from failing with an internal error.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/basicLogicalOperators.scala:
##########
@@ -2563,6 +2566,8 @@ case class AsOfJoin(
matchOperator.isEmpty &&
matchRightOperand.isEmpty &&
expressions.forall(_.resolved) &&
+ leftSortExprs.forall(_.resolved) &&
Review Comment:
These two clauses are redundant: `expressions.forall(_.resolved)` on the line above already covers `leftSortExprs`/`rightSortExprs`, since `QueryPlan._expressions` recurses into `Seq[Expression]` case-class fields. Harmless, but can be dropped.
##########
sql/core/src/test/scala/org/apache/spark/sql/AsOfJoinSQLSuite.scala:
##########
@@ -38,6 +38,19 @@ class AsOfJoinSQLSuite extends QueryTest with SharedSparkSession {
super.afterAll()
}
+ private def setupTradeQuoteViews(): Unit = {
Review Comment:
All the added tests are negative; the one test with a valid `MATCH_CONDITION` ("requires sort-merge conf") asserts failure with the conf off. A positive analyze-only test with `sortMergeAsOfJoin.enabled=true` — asserting a valid condition resolves to the expected plan, no execution backend needed — would cover the resolution path directly, and a case with a STRING (or struct) operand would have surfaced the planner-capability gap flagged in `ResolveAsOfJoin`.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-58123][SQL] Add ResolveAsOfJoin analysis for SQL ASOF JOIN [spark]
Posted by "srielau (via GitHub)" <gi...@apache.org>.
srielau commented on code in PR #57264:
URL: https://github.com/apache/spark/pull/57264#discussion_r3590151148
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolveAsOfJoin.scala:
##########
@@ -0,0 +1,202 @@
+/*
+ * 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.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,
+ matchLeft,
+ matchOp,
+ matchRight,
+ _,
+ _,
+ _)
+ 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 = (matchLeft, matchOp, matchRight) match {
+ case (Some(leftExpr), Some(operator), Some(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,
+ matchLeftOperand = None,
+ matchOperator = None,
+ matchRightOperand = None)
+ } else {
+ joinBase
+ }
+ case (None, None, None) => 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) ||
Review Comment:
Confirmed — with both confs on, STRING/struct/array operands passed analysis but failed at planning with `IllegalStateException` from `findFromOrder`.
Fixed in `7cf7621fa12`: after operand normalization, analysis now rejects non-scalar-subtract operands with `AS_OF_JOIN.UNSUPPORTED_MATCH_CONDITION_OPERAND` until multi-column sort-merge execution lands. #57277 lifts that gate once `SortMergeAsOfJoinExec` consumes `leftSortExprs`/`rightSortExprs` directly.
##########
sql/core/src/test/scala/org/apache/spark/sql/AsOfJoinSQLSuite.scala:
##########
@@ -38,6 +38,19 @@ class AsOfJoinSQLSuite extends QueryTest with SharedSparkSession {
super.afterAll()
}
+ private def setupTradeQuoteViews(): Unit = {
Review Comment:
Added in `7cf7621fa12`:
- **Positive analyze-only test**: valid TIMESTAMP `MATCH_CONDITION` with `sortMergeAsOfJoin.enabled=true` — asserts resolved `AsOfJoin` with populated `leftSortExprs`/`rightSortExprs`.
- **STRING operand test**: asserts `AS_OF_JOIN.UNSUPPORTED_MATCH_CONDITION_OPERAND` at analysis (surfaces the planner-capability gap you flagged).
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/basicLogicalOperators.scala:
##########
@@ -2563,6 +2566,8 @@ case class AsOfJoin(
matchOperator.isEmpty &&
matchRightOperand.isEmpty &&
expressions.forall(_.resolved) &&
+ leftSortExprs.forall(_.resolved) &&
Review Comment:
Dropped — `leftSortExprs.forall(_.resolved)` / `rightSortExprs.forall(_.resolved)` removed in `7cf7621fa12`.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-58123][SQL] Add ResolveAsOfJoin analysis for SQL ASOF JOIN [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on PR #57264:
URL: https://github.com/apache/spark/pull/57264#issuecomment-4993611867
**Merge Summary:**
- merged into master https://github.com/apache/spark/commit/ea9d6a7c9180e4fb342643b5f5c5aa2df03c385b
- merged into branch-4.x https://github.com/apache/spark/commit/081b4e04612d2df911b1cf0d11e8ac21d28fcc28
*Posted by `merge_spark_pr.py`*
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org