diff --git a/common/utils/src/main/resources/error/error-conditions.json b/common/utils/src/main/resources/error/error-conditions.json index 5c82f2f7a536c..f8db8e14d88ed 100644 --- a/common/utils/src/main/resources/error/error-conditions.json +++ b/common/utils/src/main/resources/error/error-conditions.json @@ -176,7 +176,7 @@ }, "ASOF_JOIN_MATCH_CONDITION_TABLE_REFERENCE" : { "message" : [ - "Each operand of the MATCH_CONDITION comparison must reference only columns of one join input. Got: and ." + "Each operand of the MATCH_CONDITION comparison must reference exactly one join input, and the two operands must reference different inputs. Got: and ." ], "sqlState" : "42K0E" }, 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 1223ee6ea541e..1487a2252d41b 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 @@ -2812,8 +2812,8 @@ object AsOfJoin { expr1: Expression, operator: MatchComparisonOperator, expr2: Expression): (Expression, Expression, MatchComparisonOperator) = { - val expr1Side = operandJoinSide(expr1, leftSet, rightSet, syntacticIsLeft = true) - val expr2Side = operandJoinSide(expr2, leftSet, rightSet, syntacticIsLeft = false) + 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) @@ -2825,13 +2825,12 @@ object AsOfJoin { private def operandJoinSide( expr: Expression, leftSet: AttributeSet, - rightSet: AttributeSet, - syntacticIsLeft: Boolean): Option[Boolean] = { + rightSet: AttributeSet): Option[Boolean] = { val refs = expr.references if (refs.isEmpty) { - // Literals, CURRENT_TIMESTAMP(), session variables, etc. have no column refs; - // use MATCH_CONDITION syntactic position (expr1/expr2) for join-side assignment. - Some(syntacticIsLeft) + // Constant operand (literal, current_timestamp(), session variable): references no join + // input, so it binds to no side, and MATCH_CONDITION needs one operand bound to each side. + None } else if (refs.subsetOf(leftSet)) { Some(true) } else if (refs.subsetOf(rightSet)) { diff --git a/sql/core/src/test/resources/sql-tests/analyzer-results/join-asof-errors.sql.out b/sql/core/src/test/resources/sql-tests/analyzer-results/join-asof-errors.sql.out index e594fb7833af6..596b3db5ce838 100644 --- a/sql/core/src/test/resources/sql-tests/analyzer-results/join-asof-errors.sql.out +++ b/sql/core/src/test/resources/sql-tests/analyzer-results/join-asof-errors.sql.out @@ -515,3 +515,67 @@ org.apache.spark.sql.AnalysisException "refs2" : "\"quote_time - INTERVAL '01' HOUR\"" } } + + +-- !query +SELECT * FROM trades t ASOF JOIN quotes q + MATCH_CONDITION (t.trade_time >= TIMESTAMP '2026-06-29 10:00:00') + ON t.symbol = q.symbol +-- !query analysis +org.apache.spark.sql.AnalysisException +{ + "errorClass" : "ASOF_JOIN_MATCH_CONDITION_TABLE_REFERENCE", + "sqlState" : "42K0E", + "messageParameters" : { + "refs1" : "\"trade_time\"", + "refs2" : "\"TIMESTAMP '2026-06-29 10:00:00'\"" + } +} + + +-- !query +SELECT * FROM trades t ASOF JOIN quotes q + MATCH_CONDITION (TIMESTAMP '2026-06-29 10:00:00' >= q.quote_time) + ON t.symbol = q.symbol +-- !query analysis +org.apache.spark.sql.AnalysisException +{ + "errorClass" : "ASOF_JOIN_MATCH_CONDITION_TABLE_REFERENCE", + "sqlState" : "42K0E", + "messageParameters" : { + "refs1" : "\"TIMESTAMP '2026-06-29 10:00:00'\"", + "refs2" : "\"quote_time\"" + } +} + + +-- !query +SELECT * FROM trades t ASOF JOIN quotes q + MATCH_CONDITION (current_timestamp() >= q.quote_time) + ON t.symbol = q.symbol +-- !query analysis +org.apache.spark.sql.AnalysisException +{ + "errorClass" : "ASOF_JOIN_MATCH_CONDITION_TABLE_REFERENCE", + "sqlState" : "42K0E", + "messageParameters" : { + "refs1" : "\"current_timestamp()\"", + "refs2" : "\"quote_time\"" + } +} + + +-- !query +SELECT * FROM trades t ASOF JOIN quotes q + MATCH_CONDITION (TIMESTAMP '2026-06-29 10:00:01' >= TIMESTAMP '2026-06-29 10:00:00') + ON t.symbol = q.symbol +-- !query analysis +org.apache.spark.sql.AnalysisException +{ + "errorClass" : "ASOF_JOIN_MATCH_CONDITION_TABLE_REFERENCE", + "sqlState" : "42K0E", + "messageParameters" : { + "refs1" : "\"TIMESTAMP '2026-06-29 10:00:01'\"", + "refs2" : "\"TIMESTAMP '2026-06-29 10:00:00'\"" + } +} diff --git a/sql/core/src/test/resources/sql-tests/analyzer-results/join-asof-expressions.sql.out b/sql/core/src/test/resources/sql-tests/analyzer-results/join-asof-expressions.sql.out index 3fed752c46025..d08c8aacd123e 100644 --- a/sql/core/src/test/resources/sql-tests/analyzer-results/join-asof-expressions.sql.out +++ b/sql/core/src/test/resources/sql-tests/analyzer-results/join-asof-expressions.sql.out @@ -195,33 +195,6 @@ org.apache.spark.sql.AnalysisException } --- !query -SELECT count(*) AS cnt -FROM trades t ASOF JOIN quotes q - MATCH_CONDITION (current_timestamp() >= q.quote_time) - ON t.symbol = q.symbol --- !query analysis -[Analyzer test output redacted due to nondeterminism] - - --- !query -SELECT count(*) AS cnt -FROM trades t ASOF JOIN quotes q - MATCH_CONDITION (t.trade_time >= TIMESTAMP '2026-06-29 10:00:00') - ON t.symbol = q.symbol --- !query analysis -[Analyzer test output redacted due to nondeterminism] - - --- !query -SELECT count(*) AS cnt -FROM trades t ASOF JOIN quotes q - MATCH_CONDITION (TIMESTAMP '2026-06-29 10:00:00' >= q.quote_time) - ON t.symbol = q.symbol --- !query analysis -[Analyzer test output redacted due to nondeterminism] - - -- !query SELECT count(*) AS cnt FROM trades t ASOF JOIN quotes q diff --git a/sql/core/src/test/resources/sql-tests/inputs/join-asof-errors.sql b/sql/core/src/test/resources/sql-tests/inputs/join-asof-errors.sql index 980362cf4fd5a..ff32b0e927610 100644 --- a/sql/core/src/test/resources/sql-tests/inputs/join-asof-errors.sql +++ b/sql/core/src/test/resources/sql-tests/inputs/join-asof-errors.sql @@ -129,3 +129,23 @@ SELECT * FROM trades t ASOF JOIN quotes q SELECT * FROM trades t ASOF JOIN quotes q MATCH_CONDITION (q.quote_time >= q.quote_time - INTERVAL 1 HOUR) ON t.symbol = q.symbol; + +-- FVT-ASOF-3-024: literal constant operand (right) references no join input +SELECT * FROM trades t ASOF JOIN quotes q + MATCH_CONDITION (t.trade_time >= TIMESTAMP '2026-06-29 10:00:00') + ON t.symbol = q.symbol; + +-- FVT-ASOF-3-025: literal constant operand (left) references no join input +SELECT * FROM trades t ASOF JOIN quotes q + MATCH_CONDITION (TIMESTAMP '2026-06-29 10:00:00' >= q.quote_time) + ON t.symbol = q.symbol; + +-- FVT-ASOF-3-026: query-foldable constant operand (current_timestamp) references no join input +SELECT * FROM trades t ASOF JOIN quotes q + MATCH_CONDITION (current_timestamp() >= q.quote_time) + ON t.symbol = q.symbol; + +-- FVT-ASOF-3-027: both operands are constants that reference no join input +SELECT * FROM trades t ASOF JOIN quotes q + MATCH_CONDITION (TIMESTAMP '2026-06-29 10:00:01' >= TIMESTAMP '2026-06-29 10:00:00') + ON t.symbol = q.symbol; diff --git a/sql/core/src/test/resources/sql-tests/inputs/join-asof-expressions.sql b/sql/core/src/test/resources/sql-tests/inputs/join-asof-expressions.sql index 50b74c8b01f8b..116dfc7a284dc 100644 --- a/sql/core/src/test/resources/sql-tests/inputs/join-asof-expressions.sql +++ b/sql/core/src/test/resources/sql-tests/inputs/join-asof-expressions.sql @@ -68,23 +68,8 @@ SELECT * FROM trades t ASOF JOIN quotes q MATCH_CONDITION (jitter_ts(t.trade_time) >= q.quote_time) ON t.symbol = q.symbol; --- FVT-ASOF-8-008: CURRENT_TIMESTAMP in MATCH_CONDITION operand (query-foldable constant) -SELECT count(*) AS cnt -FROM trades t ASOF JOIN quotes q - MATCH_CONDITION (current_timestamp() >= q.quote_time) - ON t.symbol = q.symbol; - --- FVT-ASOF-8-008a: literal constant in MATCH_CONDITION operand (right) -SELECT count(*) AS cnt -FROM trades t ASOF JOIN quotes q - MATCH_CONDITION (t.trade_time >= TIMESTAMP '2026-06-29 10:00:00') - ON t.symbol = q.symbol; - --- FVT-ASOF-8-008b: literal constant in MATCH_CONDITION operand (left) -SELECT count(*) AS cnt -FROM trades t ASOF JOIN quotes q - MATCH_CONDITION (TIMESTAMP '2026-06-29 10:00:00' >= q.quote_time) - ON t.symbol = q.symbol; +-- FVT-ASOF-8-008: constant MATCH_CONDITION operand (literal or query-foldable) is rejected; +-- it references no join input. Covered by FVT-ASOF-3-024 to FVT-ASOF-3-026 in join-asof-errors.sql -- FVT-ASOF-8-009: QUALIFY on ASOF — covered by FVT-ASOF-6-007 diff --git a/sql/core/src/test/resources/sql-tests/results/join-asof-errors.sql.out b/sql/core/src/test/resources/sql-tests/results/join-asof-errors.sql.out index 1295ff9b15877..f2c4c94d782c6 100644 --- a/sql/core/src/test/resources/sql-tests/results/join-asof-errors.sql.out +++ b/sql/core/src/test/resources/sql-tests/results/join-asof-errors.sql.out @@ -557,3 +557,75 @@ org.apache.spark.sql.AnalysisException "refs2" : "\"quote_time - INTERVAL '01' HOUR\"" } } + + +-- !query +SELECT * FROM trades t ASOF JOIN quotes q + MATCH_CONDITION (t.trade_time >= TIMESTAMP '2026-06-29 10:00:00') + ON t.symbol = q.symbol +-- !query schema +struct<> +-- !query output +org.apache.spark.sql.AnalysisException +{ + "errorClass" : "ASOF_JOIN_MATCH_CONDITION_TABLE_REFERENCE", + "sqlState" : "42K0E", + "messageParameters" : { + "refs1" : "\"trade_time\"", + "refs2" : "\"TIMESTAMP '2026-06-29 10:00:00'\"" + } +} + + +-- !query +SELECT * FROM trades t ASOF JOIN quotes q + MATCH_CONDITION (TIMESTAMP '2026-06-29 10:00:00' >= q.quote_time) + ON t.symbol = q.symbol +-- !query schema +struct<> +-- !query output +org.apache.spark.sql.AnalysisException +{ + "errorClass" : "ASOF_JOIN_MATCH_CONDITION_TABLE_REFERENCE", + "sqlState" : "42K0E", + "messageParameters" : { + "refs1" : "\"TIMESTAMP '2026-06-29 10:00:00'\"", + "refs2" : "\"quote_time\"" + } +} + + +-- !query +SELECT * FROM trades t ASOF JOIN quotes q + MATCH_CONDITION (current_timestamp() >= q.quote_time) + ON t.symbol = q.symbol +-- !query schema +struct<> +-- !query output +org.apache.spark.sql.AnalysisException +{ + "errorClass" : "ASOF_JOIN_MATCH_CONDITION_TABLE_REFERENCE", + "sqlState" : "42K0E", + "messageParameters" : { + "refs1" : "\"current_timestamp()\"", + "refs2" : "\"quote_time\"" + } +} + + +-- !query +SELECT * FROM trades t ASOF JOIN quotes q + MATCH_CONDITION (TIMESTAMP '2026-06-29 10:00:01' >= TIMESTAMP '2026-06-29 10:00:00') + ON t.symbol = q.symbol +-- !query schema +struct<> +-- !query output +org.apache.spark.sql.AnalysisException +{ + "errorClass" : "ASOF_JOIN_MATCH_CONDITION_TABLE_REFERENCE", + "sqlState" : "42K0E", + "messageParameters" : { + "refs1" : "\"TIMESTAMP '2026-06-29 10:00:01'\"", + "refs2" : "\"TIMESTAMP '2026-06-29 10:00:00'\"" + } +} diff --git a/sql/core/src/test/resources/sql-tests/results/join-asof-expressions.sql.out b/sql/core/src/test/resources/sql-tests/results/join-asof-expressions.sql.out index fd9b7740daffb..536eb98f705a5 100644 --- a/sql/core/src/test/resources/sql-tests/results/join-asof-expressions.sql.out +++ b/sql/core/src/test/resources/sql-tests/results/join-asof-expressions.sql.out @@ -157,39 +157,6 @@ org.apache.spark.sql.AnalysisException } --- !query -SELECT count(*) AS cnt -FROM trades t ASOF JOIN quotes q - MATCH_CONDITION (current_timestamp() >= q.quote_time) - ON t.symbol = q.symbol --- !query schema -struct --- !query output -3 - - --- !query -SELECT count(*) AS cnt -FROM trades t ASOF JOIN quotes q - MATCH_CONDITION (t.trade_time >= TIMESTAMP '2026-06-29 10:00:00') - ON t.symbol = q.symbol --- !query schema -struct --- !query output -3 - - --- !query -SELECT count(*) AS cnt -FROM trades t ASOF JOIN quotes q - MATCH_CONDITION (TIMESTAMP '2026-06-29 10:00:00' >= q.quote_time) - ON t.symbol = q.symbol --- !query schema -struct --- !query output -2 - - -- !query SELECT count(*) AS cnt FROM trades t ASOF JOIN quotes q diff --git a/sql/core/src/test/scala/org/apache/spark/sql/AsOfJoinSQLSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/AsOfJoinSQLSuite.scala index 36962df5af455..eab4443eacb64 100644 --- a/sql/core/src/test/scala/org/apache/spark/sql/AsOfJoinSQLSuite.scala +++ b/sql/core/src/test/scala/org/apache/spark/sql/AsOfJoinSQLSuite.scala @@ -210,7 +210,7 @@ class AsOfJoinSQLSuite extends QueryTest with SharedSparkSession { stop = 122))) } - test("MATCH_CONDITION accepts CURRENT_TIMESTAMP as left operand") { + test("MATCH_CONDITION rejects a query-foldable constant operand (no join input reference)") { setupTradeQuoteViews() val sqlText = """ @@ -219,15 +219,16 @@ class AsOfJoinSQLSuite extends QueryTest with SharedSparkSession { | MATCH_CONDITION (current_timestamp() >= q.quote_time) | ON t.symbol = q.symbol |""".stripMargin - val asOfJoin = sql(sqlText).queryExecution.analyzed.collectFirst { - case j: AsOfJoin => j - }.get - assert(asOfJoin.asOfCondition.resolved) - assert(asOfJoin.leftSortExprs.nonEmpty) - assert(asOfJoin.rightSortExprs.nonEmpty) + checkError( + exception = intercept[AnalysisException](sql(sqlText)), + condition = "ASOF_JOIN_MATCH_CONDITION_TABLE_REFERENCE", + sqlState = Some("42K0E"), + parameters = Map( + "refs1" -> "\"current_timestamp()\"", + "refs2" -> "\"quote_time\"")) } - test("MATCH_CONDITION accepts literal constant as right operand") { + test("MATCH_CONDITION rejects a literal constant operand (no join input reference)") { setupTradeQuoteViews() val sqlText = """ @@ -236,10 +237,51 @@ class AsOfJoinSQLSuite extends QueryTest with SharedSparkSession { | MATCH_CONDITION (t.trade_time >= TIMESTAMP '2026-06-29 10:00:00') | ON t.symbol = q.symbol |""".stripMargin - val asOfJoin = sql(sqlText).queryExecution.analyzed.collectFirst { - case j: AsOfJoin => j - }.get - assert(asOfJoin.asOfCondition.resolved) + checkError( + exception = intercept[AnalysisException](sql(sqlText)), + condition = "ASOF_JOIN_MATCH_CONDITION_TABLE_REFERENCE", + sqlState = Some("42K0E"), + parameters = Map( + "refs1" -> "\"trade_time\"", + "refs2" -> "\"TIMESTAMP '2026-06-29 10:00:00'\"")) + } + + test("MATCH_CONDITION rejects two constant operands (neither references a join input)") { + setupTradeQuoteViews() + val sqlText = + """ + |SELECT count(*) + |FROM trades t ASOF JOIN quotes q + | MATCH_CONDITION (TIMESTAMP '2026-06-29 10:00:01' >= TIMESTAMP '2026-06-29 10:00:00') + | 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" -> "\"TIMESTAMP '2026-06-29 10:00:01'\"", + "refs2" -> "\"TIMESTAMP '2026-06-29 10:00:00'\"")) + } + + test("MATCH_CONDITION rejects a constant operand under the single-pass analyzer") { + setupTradeQuoteViews() + val sqlText = + """ + |SELECT count(*) + |FROM trades t ASOF JOIN quotes q + | MATCH_CONDITION (t.trade_time >= TIMESTAMP '2026-06-29 10:00:00') + | ON t.symbol = q.symbol + |""".stripMargin + withSQLConf(SQLConf.ANALYZER_SINGLE_PASS_RESOLVER_ENABLED.key -> "true") { + checkError( + exception = intercept[AnalysisException](sql(sqlText)), + condition = "ASOF_JOIN_MATCH_CONDITION_TABLE_REFERENCE", + sqlState = Some("42K0E"), + parameters = Map( + "refs1" -> "\"trade_time\"", + "refs2" -> "\"TIMESTAMP '2026-06-29 10:00:00'\"")) + } } test("MATCH_CONDITION rejects scalar subquery operand") {