This is an automated email from the ASF dual-hosted git repository.
gengliangwang pushed a commit to branch branch-4.x
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/branch-4.x by this push:
new 4159c905e229 [SPARK-57421][SQL] Support @-syntax version time travel
on table names
4159c905e229 is described below
commit 4159c905e22999609f0032f749c481db6a443360
Author: sotikoug83 <[email protected]>
AuthorDate: Mon Jul 6 09:06:06 2026 -0700
[SPARK-57421][SQL] Support @-syntax version time travel on table names
### What changes were proposed in this pull request?
Add the `` version time travel shorthand on table names so that
```
SELECT * FROM namevN
```
and
```
spark.read.format("format").table("namevN")
```
parse into the same plan as the existing `VERSION AS OF` and
`option("versionAsOf")` clauses. Support for SQL, the DataFrame reader, and
Spark Connect.
### Why are the changes needed?
Spark has SQL `VERSION AS OF` and the `versionAsOf` reader options but not
the compact `` suffix. Resolving it at parse time also simplifies the time
travel entry points pipeline.
### Does this PR introduce _any_ user-facing change?
Yes, new syntax on table names gated by a `conf`
(`spark.sql.timeTravel.atSyntax.enabled`, default **true**) which allows time
travel. No longer gives a parse error when enabled and the syntax is used.
### How was this patch tested?
New tests in `PlanParserSuite`, `SparkConnectPlannerSuite`,
`DataSourceV2SQLSuite`, and `DataStreamTableAPISuite`.
### Was this patch authored or co-authored using generative AI tooling?
With help from Claude :)
Closes #56481 from sotikoug83/SPARK-57421-at-syntax-time-travel.
Authored-by: sotikoug83 <[email protected]>
Signed-off-by: Gengliang Wang <[email protected]>
(cherry picked from commit 3e247620d058da9c421ea51e032ad7ed39952a8f)
Signed-off-by: Gengliang Wang <[email protected]>
---
.../src/main/resources/error/error-conditions.json | 7 +-
.../spark/sql/catalyst/parser/SqlBaseLexer.g4 | 1 +
.../spark/sql/catalyst/parser/SqlBaseParser.g4 | 16 +++-
.../spark/sql/errors/QueryParsingErrors.scala | 14 ++++
.../sql/catalyst/parser/AbstractSqlParser.scala | 10 +++
.../spark/sql/catalyst/parser/AstBuilder.scala | 54 +++++++++++++-
.../sql/catalyst/parser/ParserInterface.scala | 10 +++
.../sql/catalyst/parser/TemporalIdentifier.scala | 36 +++++++++
.../org/apache/spark/sql/internal/SQLConf.scala | 10 +++
.../sql/catalyst/parser/PlanParserSuite.scala | 56 ++++++++++++++
.../sql/connect/planner/SparkConnectPlanner.scala | 11 ++-
.../connect/planner/SparkConnectPlannerSuite.scala | 42 ++++++++++-
.../apache/spark/sql/classic/DataFrameReader.scala | 9 ++-
.../spark/sql/classic/DataStreamReader.scala | 9 ++-
.../analyzer-results/pipe-operators.sql.out | 4 +-
.../sql-tests/results/pipe-operators.sql.out | 4 +-
.../spark/sql/connector/DataSourceV2SQLSuite.scala | 86 ++++++++++++++++++++++
.../streaming/test/DataStreamTableAPISuite.scala | 9 +++
18 files changed, 370 insertions(+), 18 deletions(-)
diff --git a/common/utils/src/main/resources/error/error-conditions.json
b/common/utils/src/main/resources/error/error-conditions.json
index 8b9292286aab..03bffb930e73 100644
--- a/common/utils/src/main/resources/error/error-conditions.json
+++ b/common/utils/src/main/resources/error/error-conditions.json
@@ -5647,7 +5647,7 @@
},
"MULTIPLE_TIME_TRAVEL_SPEC" : {
"message" : [
- "Cannot specify time travel in both the time travel clause and options."
+ "Cannot specify time travel in more than one of: the '@' suffix in the
table name, the time travel clause, and the read options."
],
"sqlState" : "42K0E"
},
@@ -8574,6 +8574,11 @@
"Time travel on the relation: <relationId>."
]
},
+ "TIME_TRAVEL_AT_SYNTAX" : {
+ "message" : [
+ "Time travel using the '@' suffix in table names. Set <config> to
true to enable it."
+ ]
+ },
"TOO_MANY_TYPE_ARGUMENTS_FOR_UDF_CLASS" : {
"message" : [
"UDF class with <num> type arguments."
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 0e940ee5b4b0..f4a9aa1d5471 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
@@ -591,6 +591,7 @@ OPERATOR_PIPE: '|>';
HAT: '^';
COLON: ':';
DOUBLE_COLON: '::';
+AT_VERSION: '@V';
ARROW: '->';
FAT_ARROW : '=>';
HENT_START: '/*+';
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 2466cf62272a..7d75aaebd546 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
@@ -201,6 +201,10 @@ singleTableIdentifier
: tableIdentifier EOF
;
+singleTemporalTableIdentifier
+ : temporalTableIdentifier EOF
+ ;
+
singleMultipartIdentifier
: multipartIdentifier EOF
;
@@ -1133,7 +1137,7 @@ relationPrimary
: streamRelationPrimary #streamRelation
| identifierReference changesClause
optionsClause? tableAlias #changelogTableName
- | identifierReference temporalClause?
+ | temporalTableIdentifierReference temporalClause?
optionsClause? sample? watermarkClause? tableAlias #tableName
| LEFT_PAREN query RIGHT_PAREN sample? watermarkClause?
tableAlias #aliasedQuery
@@ -1239,6 +1243,16 @@ tableIdentifier
: (db=errorCapturingIdentifier DOT)? table=errorCapturingIdentifier
;
+temporalTableIdentifier
+ : id=multipartIdentifier AT_VERSION version
+ | id=multipartIdentifier
+ ;
+
+temporalTableIdentifierReference
+ : identifierReference AT_VERSION version
+ | identifierReference
+ ;
+
functionIdentifier
: (db=errorCapturingIdentifier DOT)? function=errorCapturingIdentifier
;
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 558bda49a302..01ba078be729 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
@@ -878,6 +878,20 @@ private[sql] object QueryParsingErrors extends
DataTypeErrorsBase {
ctx)
}
+ def multipleTimeTravelSpec(ctx: ParserRuleContext): Throwable = {
+ new ParseException(
+ errorClass = "MULTIPLE_TIME_TRAVEL_SPEC",
+ messageParameters = Map.empty,
+ ctx)
+ }
+
+ def timeTravelAtSyntaxDisabled(configKey: String, ctx: ParserRuleContext):
Throwable = {
+ new ParseException(
+ errorClass = "UNSUPPORTED_FEATURE.TIME_TRAVEL_AT_SYNTAX",
+ messageParameters = Map("config" -> toSQLConf(configKey)),
+ 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/parser/AbstractSqlParser.scala
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/parser/AbstractSqlParser.scala
index 29bf924f244e..e6d53ecc1e25 100644
---
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/parser/AbstractSqlParser.scala
+++
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/parser/AbstractSqlParser.scala
@@ -72,6 +72,16 @@ abstract class AbstractSqlParser extends AbstractParser with
ParserInterface {
}
}
+ /** Creates a TemporalIdentifier for a given SQL string */
+ override def parseTemporalTableIdentifier(sqlText: String):
TemporalIdentifier = {
+ parse(sqlText) { parser =>
+ val ctx = parser.singleTemporalTableIdentifier()
+ withErrorHandling(ctx, Some(sqlText)) {
+ astBuilder.visitSingleTemporalTableIdentifier(ctx)
+ }
+ }
+ }
+
/** Creates LogicalPlan for a given SQL string of query. */
override def parseQuery(sqlText: String): LogicalPlan =
parse(sqlText) { parser =>
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 aa3d68198e2d..996d72c10da9 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
@@ -708,6 +708,11 @@ class AstBuilder extends DataTypeAstBuilder
visitMultipartIdentifier(ctx.multipartIdentifier)
}
+ override def visitSingleTemporalTableIdentifier(
+ ctx: SingleTemporalTableIdentifierContext): TemporalIdentifier =
withOrigin(ctx) {
+ visitTemporalTableIdentifier(ctx.temporalTableIdentifier)
+ }
+
override def visitSinglePathElementList(
ctx: SinglePathElementListContext): Seq[PathElement] = withOrigin(ctx) {
ctx.pathElement().asScala.map(visitPathElement).toSeq
@@ -2620,13 +2625,56 @@ class AstBuilder extends DataTypeAstBuilder
* Create an aliased table reference. This is typically used in FROM clauses.
*/
override def visitTableName(ctx: TableNameContext): LogicalPlan =
withOrigin(ctx) {
- val relation = createUnresolvedRelation(ctx.identifierReference,
Option(ctx.optionsClause))
- val table = mayApplyAliasPlan(
- ctx.tableAlias, relation.optionalMap(ctx.temporalClause)(withTimeTravel))
+ val ttCtx = ctx.temporalTableIdentifierReference
+ val relation = createUnresolvedRelation(ttCtx.identifierReference,
Option(ctx.optionsClause))
+ val withTimeTravelSpec = withTableTimeTravel(relation, ttCtx,
ctx.temporalClause)
+ val table = mayApplyAliasPlan(ctx.tableAlias, withTimeTravelSpec)
val sample = table.optionalMap(ctx.sample)(withSample)
sample.optionalMap(ctx.watermarkClause)(withWatermark)
}
+ /**
+ * Applies the table-name '@' time travel suffix and/or the `AS OF` clause
to `relation`.
+ */
+ private def withTableTimeTravel(
+ relation: LogicalPlan,
+ ttCtx: TemporalTableIdentifierReferenceContext,
+ clause: TemporalClauseContext): LogicalPlan = {
+ val (atTimestamp, atVersion) = temporalSpec(ttCtx, ttCtx.version)
+ val hasAtSpec = atTimestamp.isDefined || atVersion.isDefined
+ if (hasAtSpec && clause != null) {
+ withOrigin(clause) {
+ throw QueryParsingErrors.multipleTimeTravelSpec(clause)
+ }
+ }
+ val withAtSpec =
+ if (hasAtSpec) RelationTimeTravel(relation, atTimestamp, atVersion) else
relation
+ withAtSpec.optionalMap(clause)(withTimeTravel)
+ }
+
+ override def visitTemporalTableIdentifier(
+ ctx: TemporalTableIdentifierContext): TemporalIdentifier =
withOrigin(ctx) {
+ val (_, version) = temporalSpec(ctx, ctx.version)
+ TemporalIdentifier(visitMultipartIdentifier(ctx.id), version)
+ }
+
+ /**
+ * Extract the optional '@v<version>' time travel suffix of a table
identifier.
+ */
+ private def temporalSpec(
+ ctx: ParserRuleContext,
+ versionCtx: VersionContext): (Option[Expression], Option[String]) = {
+ if (versionCtx == null) {
+ (None, None)
+ } else {
+ if (!conf.getConf(SQLConf.TIME_TRAVEL_AT_SYNTAX_ENABLED)) {
+ throw QueryParsingErrors.timeTravelAtSyntaxDisabled(
+ SQLConf.TIME_TRAVEL_AT_SYNTAX_ENABLED.key, ctx)
+ }
+ (None, visitVersion(versionCtx))
+ }
+ }
+
override def visitVersion(ctx: VersionContext): Option[String] = {
if (ctx != null) {
if (ctx.INTEGER_VALUE() != null) {
diff --git
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/parser/ParserInterface.scala
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/parser/ParserInterface.scala
index bcbc7039f53c..193d68e9a444 100644
---
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/parser/ParserInterface.scala
+++
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/parser/ParserInterface.scala
@@ -69,6 +69,16 @@ trait ParserInterface extends DataTypeParserInterface {
@throws[ParseException]("Text cannot be parsed to a multi-part identifier")
def parseMultipartIdentifier(sqlText: String): Seq[String]
+ /**
+ * Parse a string to a [[TemporalIdentifier]].
+ */
+ @throws[ParseException]("Text cannot be parsed to a temporal table
identifier")
+ def parseTemporalTableIdentifier(sqlText: String): TemporalIdentifier = {
+ // Default implementation does not recognize the time travel suffix.
Concrete
+ // implementations can override this to support it.
+ TemporalIdentifier(parseMultipartIdentifier(sqlText), None)
+ }
+
/**
* Parse a query string to a [[LogicalPlan]].
*/
diff --git
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/parser/TemporalIdentifier.scala
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/parser/TemporalIdentifier.scala
new file mode 100644
index 000000000000..9be5c4016a7c
--- /dev/null
+++
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/parser/TemporalIdentifier.scala
@@ -0,0 +1,36 @@
+/*
+ * 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.parser
+
+import org.apache.spark.sql.catalyst.analysis.RelationTimeTravel
+import org.apache.spark.sql.catalyst.plans.logical.LogicalPlan
+
+/**
+ * Result of parsing a table name that may carry an '@v<version>' time travel
suffix.
+ */
+case class TemporalIdentifier(
+ nameParts: Seq[String],
+ version: Option[String]) {
+
+ def isTemporal: Boolean = version.isDefined
+
+ /** Wraps `plan` in a [[RelationTimeTravel]] if a time travel suffix was
specified. */
+ def wrapTimeTravel(plan: LogicalPlan): LogicalPlan = {
+ if (isTemporal) RelationTimeTravel(plan, None, version) else plan
+ }
+}
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 1ca873d7da1a..86a19e45a045 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
@@ -7136,6 +7136,16 @@ object SQLConf {
.stringConf
.createWithDefault("versionAsOf")
+ val TIME_TRAVEL_AT_SYNTAX_ENABLED =
+ buildConf("spark.sql.timeTravel.atSyntax.enabled")
+ .doc("When true, a table name in a query or in table-reading APIs can
carry a time " +
+ "travel suffix: 'name@v123' reads version 123 of the table. When
false, '@' in " +
+ "table names fails at parse time.")
+ .version("4.3.0")
+ .withBindingPolicy(ConfigBindingPolicy.NOT_APPLICABLE)
+ .booleanConf
+ .createWithDefault(true)
+
val OPERATOR_PIPE_SYNTAX_ENABLED =
buildConf("spark.sql.operatorPipeSyntaxEnabled")
.doc("If true, enable operator pipe syntax for Apache Spark SQL. This
uses the operator " +
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 e20071b3de62..712da9a9508a 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
@@ -2193,6 +2193,62 @@ class PlanParserSuite extends AnalysisTest {
stop = 38))
}
+ test("at syntax time travel") {
+ def versionPlan(version: String): LogicalPlan = {
+ Project(Seq(UnresolvedStar(None)),
+ RelationTimeTravel(UnresolvedRelation(Seq("a", "b", "c")), None,
Some(version)))
+ }
+ assertEqual("SELECT * FROM a.b.c@v123456789", versionPlan("123456789"))
+ assertEqual("SELECT * FROM a.b.c@V123456789", versionPlan("123456789"))
+ assertEqual("SELECT * FROM a.b.c @v123456789", versionPlan("123456789"))
+ assertEqual("SELECT * FROM a.b.c@v'Snapshot123456789'",
versionPlan("Snapshot123456789"))
+
+ assertEqual("SELECT * FROM `t@v1`",
+ Project(Seq(UnresolvedStar(None)), UnresolvedRelation(Seq("t@v1"))))
+
+ // A non-time-travel '@' suffix is always a parse error.
+ Seq("SELECT * FROM a@foo", "SELECT * FROM a@", "SELECT * FROM
a@v").foreach { q =>
+ assert(intercept[ParseException](parsePlan(q)).getCondition ==
"PARSE_SYNTAX_ERROR",
+ s"expected PARSE_SYNTAX_ERROR for: $q")
+ }
+
+ // The '@' suffix conflicts with an AS OF clause.
+ checkError(
+ exception = parseException("SELECT * FROM t@v1 VERSION AS OF 2"),
+ condition = "MULTIPLE_TIME_TRAVEL_SPEC",
+ parameters = Map.empty,
+ context = ExpectedContext(fragment = "VERSION AS OF 2", start = 19, stop
= 33))
+ checkError(
+ exception = parseException("SELECT * FROM t@v1 TIMESTAMP AS OF
'2019-01-29'"),
+ condition = "MULTIPLE_TIME_TRAVEL_SPEC",
+ parameters = Map.empty,
+ context = ExpectedContext(fragment = "TIMESTAMP AS OF '2019-01-29'",
start = 19, stop = 46))
+
+ assert(intercept[ParseException] {
+ parsePlan("INSERT INTO t@v1 VALUES (1)")
+ }.getCondition == "PARSE_SYNTAX_ERROR")
+
+ withSQLConf(SQLConf.TIME_TRAVEL_AT_SYNTAX_ENABLED.key -> "false") {
+ checkError(
+ exception = parseException("SELECT * FROM t@v1"),
+ condition = "UNSUPPORTED_FEATURE.TIME_TRAVEL_AT_SYNTAX",
+ parameters = Map("config" ->
"\"spark.sql.timeTravel.atSyntax.enabled\""),
+ context = ExpectedContext(fragment = "t@v1", start = 14, stop = 17))
+ }
+ }
+
+ test("parseTemporalTableIdentifier") {
+ assert(parseTemporalTableIdentifier("a.b") ===
+ TemporalIdentifier(Seq("a", "b"), None))
+ assert(parseTemporalTableIdentifier("a.b@v5") ===
+ TemporalIdentifier(Seq("a", "b"), Some("5")))
+ assert(parseTemporalTableIdentifier("`t@v1`") ===
+ TemporalIdentifier(Seq("t@v1"), None))
+ Seq("a.b@x", "a@foo", "a@", "a@v").foreach { s =>
+ intercept[ParseException](parseTemporalTableIdentifier(s))
+ }
+ }
+
test("CHANGES clause - version range") {
def changesFromVersion(
startVersion: String,
diff --git
a/sql/connect/server/src/main/scala/org/apache/spark/sql/connect/planner/SparkConnectPlanner.scala
b/sql/connect/server/src/main/scala/org/apache/spark/sql/connect/planner/SparkConnectPlanner.scala
index 94e7938b7c72..1922eb14bda7 100644
---
a/sql/connect/server/src/main/scala/org/apache/spark/sql/connect/planner/SparkConnectPlanner.scala
+++
b/sql/connect/server/src/main/scala/org/apache/spark/sql/connect/planner/SparkConnectPlanner.scala
@@ -1660,10 +1660,17 @@ class SparkConnectPlanner(
rel.getReadTypeCase match {
case proto.Read.ReadTypeCase.NAMED_TABLE =>
- UnresolvedRelation(
-
parser.parseMultipartIdentifier(rel.getNamedTable.getUnparsedIdentifier),
+ val temporalIdent =
+
parser.parseTemporalTableIdentifier(rel.getNamedTable.getUnparsedIdentifier)
+ if (temporalIdent.isTemporal && rel.getIsStreaming) {
+ throw QueryCompilationErrors.timeTravelUnsupportedError(
+ QueryCompilationErrors.toSQLId(temporalIdent.nameParts))
+ }
+ val relation = UnresolvedRelation(
+ temporalIdent.nameParts,
new CaseInsensitiveStringMap(rel.getNamedTable.getOptionsMap),
isStreaming = rel.getIsStreaming)
+ temporalIdent.wrapTimeTravel(relation)
case proto.Read.ReadTypeCase.DATA_SOURCE if !rel.getIsStreaming =>
val reader = session.read
diff --git
a/sql/connect/server/src/test/scala/org/apache/spark/sql/connect/planner/SparkConnectPlannerSuite.scala
b/sql/connect/server/src/test/scala/org/apache/spark/sql/connect/planner/SparkConnectPlannerSuite.scala
index 89ccaea93a04..22f7cabc8c18 100644
---
a/sql/connect/server/src/test/scala/org/apache/spark/sql/connect/planner/SparkConnectPlannerSuite.scala
+++
b/sql/connect/server/src/test/scala/org/apache/spark/sql/connect/planner/SparkConnectPlannerSuite.scala
@@ -26,7 +26,7 @@ import org.apache.spark.connect.proto
import org.apache.spark.connect.proto.Expression.{Alias, ExpressionString,
UnresolvedStar}
import org.apache.spark.sql.{AnalysisException, Row}
import org.apache.spark.sql.catalyst.InternalRow
-import org.apache.spark.sql.catalyst.analysis.{UnresolvedAlias,
UnresolvedFunction, UnresolvedRelation}
+import org.apache.spark.sql.catalyst.analysis.{RelationTimeTravel,
UnresolvedAlias, UnresolvedFunction, UnresolvedRelation}
import org.apache.spark.sql.catalyst.expressions.{AttributeReference,
UnsafeProjection}
import org.apache.spark.sql.catalyst.plans.logical
import org.apache.spark.sql.catalyst.plans.logical.{Aggregate, LogicalPlan}
@@ -159,6 +159,46 @@ class SparkConnectPlannerSuite extends SparkFunSuite with
SparkConnectPlanTest {
assert(res.nodeName == "UnresolvedRelation")
}
+ test("Read with at syntax time travel") {
+ val read = proto.Read
+ .newBuilder()
+
.setNamedTable(proto.Read.NamedTable.newBuilder.setUnparsedIdentifier("name@v1").build())
+ .build()
+ val res = transform(proto.Relation.newBuilder.setRead(read).build())
+ res match {
+ case RelationTimeTravel(relation: UnresolvedRelation, timestamp,
version) =>
+ assert(relation.multipartIdentifier === Seq("name"))
+ assert(timestamp.isEmpty)
+ assert(version === Some("1"))
+ case other => fail(s"Expected RelationTimeTravel but got: $other")
+ }
+
+ // Streaming reads do not support time travel.
+ val streamingRead = read.toBuilder.setIsStreaming(true).build()
+ val e = intercept[AnalysisException] {
+ transform(proto.Relation.newBuilder.setRead(streamingRead).build())
+ }
+ assert(e.getCondition === "UNSUPPORTED_FEATURE.TIME_TRAVEL")
+
+ // A non-time-travel '@' suffix is a parse error.
+ val badRead = read.toBuilder
+
.setNamedTable(proto.Read.NamedTable.newBuilder.setUnparsedIdentifier("name@foo").build())
+ .build()
+ val pe = intercept[AnalysisException] {
+ transform(proto.Relation.newBuilder.setRead(badRead).build())
+ }
+ assert(pe.getCondition === "PARSE_SYNTAX_ERROR")
+
+ // A backticked '@' name stays a literal table name.
+ val quotedRead = read.toBuilder
+
.setNamedTable(proto.Read.NamedTable.newBuilder.setUnparsedIdentifier("`name@v1`").build())
+ .build()
+ transform(proto.Relation.newBuilder.setRead(quotedRead).build()) match {
+ case u: UnresolvedRelation => assert(u.multipartIdentifier ===
Seq("name@v1"))
+ case other => fail(s"Expected a literal UnresolvedRelation but got:
$other")
+ }
+ }
+
test("Simple Table with options") {
val read = proto.Read.newBuilder().build()
// Invalid read without Table name.
diff --git
a/sql/core/src/main/scala/org/apache/spark/sql/classic/DataFrameReader.scala
b/sql/core/src/main/scala/org/apache/spark/sql/classic/DataFrameReader.scala
index 5ba6cb204075..f791a932b3a8 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/classic/DataFrameReader.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/classic/DataFrameReader.scala
@@ -322,10 +322,11 @@ class DataFrameReader private[sql](sparkSession:
SparkSession)
/** @inheritdoc */
def table(tableName: String): DataFrame = {
assertNoSpecifiedSchema("table")
- val multipartIdentifier =
- sparkSession.sessionState.sqlParser.parseMultipartIdentifier(tableName)
- Dataset.ofRows(sparkSession, UnresolvedRelation(multipartIdentifier,
- new CaseInsensitiveStringMap(extraOptions.toMap.asJava)))
+ val temporalIdent =
+
sparkSession.sessionState.sqlParser.parseTemporalTableIdentifier(tableName)
+ val relation = UnresolvedRelation(temporalIdent.nameParts,
+ new CaseInsensitiveStringMap(extraOptions.toMap.asJava))
+ Dataset.ofRows(sparkSession, temporalIdent.wrapTimeTravel(relation))
}
/** @inheritdoc */
diff --git
a/sql/core/src/main/scala/org/apache/spark/sql/classic/DataStreamReader.scala
b/sql/core/src/main/scala/org/apache/spark/sql/classic/DataStreamReader.scala
index eb3120cac05a..b359554e527c 100644
---
a/sql/core/src/main/scala/org/apache/spark/sql/classic/DataStreamReader.scala
+++
b/sql/core/src/main/scala/org/apache/spark/sql/classic/DataStreamReader.scala
@@ -102,9 +102,14 @@ final class DataStreamReader private[sql](sparkSession:
SparkSession)
/** @inheritdoc */
def table(tableName: String): DataFrame = {
require(tableName != null, "The table name can't be null")
- val identifier =
sparkSession.sessionState.sqlParser.parseMultipartIdentifier(tableName)
+ val temporalIdent =
+
sparkSession.sessionState.sqlParser.parseTemporalTableIdentifier(tableName)
+ if (temporalIdent.isTemporal) {
+ throw QueryCompilationErrors.timeTravelUnsupportedError(
+ QueryCompilationErrors.toSQLId(temporalIdent.nameParts))
+ }
val unresolved = UnresolvedRelation(
- identifier,
+ temporalIdent.nameParts,
new CaseInsensitiveStringMap(extraOptions.toMap.asJava),
isStreaming = true)
val plan = NamedStreamingRelation.withUserProvidedName(unresolved,
userProvidedSourceName)
diff --git
a/sql/core/src/test/resources/sql-tests/analyzer-results/pipe-operators.sql.out
b/sql/core/src/test/resources/sql-tests/analyzer-results/pipe-operators.sql.out
index e9d5f79315a6..d136341fa32b 100644
---
a/sql/core/src/test/resources/sql-tests/analyzer-results/pipe-operators.sql.out
+++
b/sql/core/src/test/resources/sql-tests/analyzer-results/pipe-operators.sql.out
@@ -1346,8 +1346,8 @@ org.apache.spark.sql.catalyst.parser.ParseException
"errorClass" : "PARSE_SYNTAX_ERROR",
"sqlState" : "42601",
"messageParameters" : {
- "error" : "'@'",
- "hint" : ""
+ "error" : "'@v'",
+ "hint" : ": extra input '@v'"
}
}
diff --git
a/sql/core/src/test/resources/sql-tests/results/pipe-operators.sql.out
b/sql/core/src/test/resources/sql-tests/results/pipe-operators.sql.out
index aa0a377a95a4..fabe74b49158 100644
--- a/sql/core/src/test/resources/sql-tests/results/pipe-operators.sql.out
+++ b/sql/core/src/test/resources/sql-tests/results/pipe-operators.sql.out
@@ -1279,8 +1279,8 @@ org.apache.spark.sql.catalyst.parser.ParseException
"errorClass" : "PARSE_SYNTAX_ERROR",
"sqlState" : "42601",
"messageParameters" : {
- "error" : "'@'",
- "hint" : ""
+ "error" : "'@v'",
+ "hint" : ": extra input '@v'"
}
}
diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/connector/DataSourceV2SQLSuite.scala
b/sql/core/src/test/scala/org/apache/spark/sql/connector/DataSourceV2SQLSuite.scala
index 3c5d1626ca1f..7ce2945898c3 100644
---
a/sql/core/src/test/scala/org/apache/spark/sql/connector/DataSourceV2SQLSuite.scala
+++
b/sql/core/src/test/scala/org/apache/spark/sql/connector/DataSourceV2SQLSuite.scala
@@ -3941,6 +3941,92 @@ class DataSourceV2SQLSuiteV1Filter
}
}
+ test("time travel with at syntax") {
+ sql("use testcat")
+ val t1 = "testcat.tSnapshot123456789"
+ val t2 = "testcat.t2345678910"
+ withTable(t1, t2) {
+ sql(s"CREATE TABLE $t1 (id int) USING foo")
+ sql(s"CREATE TABLE $t2 (id int) USING foo")
+ sql(s"INSERT INTO $t1 VALUES (1), (2)")
+ sql(s"INSERT INTO $t2 VALUES (3), (4)")
+
+ checkAnswer(sql("SELECT * FROM t@v2345678910"), Seq(Row(3), Row(4)))
+ checkAnswer(sql("SELECT * FROM t@V2345678910"), Seq(Row(3), Row(4)))
+ checkAnswer(sql("SELECT * FROM t@v'Snapshot123456789'"), Seq(Row(1),
Row(2)))
+ checkAnswer(spark.read.table("t@v2345678910"), Seq(Row(3), Row(4)))
+ checkAnswer(spark.table("t@v2345678910"), Seq(Row(3), Row(4)))
+
+ checkError(
+ exception = intercept[ParseException] {
+ sql("SELECT * FROM t@v2345678910 VERSION AS OF 1")
+ },
+ condition = "MULTIPLE_TIME_TRAVEL_SPEC",
+ parameters = Map.empty,
+ context = ExpectedContext(fragment = "VERSION AS OF 1", start = 28,
stop = 42))
+ checkError(
+ exception = intercept[ParseException] {
+ sql("SELECT * FROM t@v2345678910 TIMESTAMP AS OF '2019-01-29'")
+ },
+ condition = "MULTIPLE_TIME_TRAVEL_SPEC",
+ parameters = Map.empty,
+ context = ExpectedContext(
+ fragment = "TIMESTAMP AS OF '2019-01-29'", start = 28, stop = 55))
+ checkError(
+ exception = intercept[AnalysisException] {
+ spark.read.option("versionAsOf",
"2345678910").table("t@v2345678910").collect()
+ },
+ condition = "MULTIPLE_TIME_TRAVEL_SPEC",
+ parameters = Map.empty)
+ checkError(
+ exception = intercept[AnalysisException] {
+ spark.read.option("timestampAsOf",
"2019-01-29").table("t@v2345678910").collect()
+ },
+ condition = "MULTIPLE_TIME_TRAVEL_SPEC",
+ parameters = Map.empty)
+ checkError(
+ exception = intercept[AnalysisException] {
+ sql("SELECT * FROM t@v2345678910 WITH ('timestampAsOf' =
'2019-01-29')")
+ },
+ condition = "MULTIPLE_TIME_TRAVEL_SPEC",
+ parameters = Map.empty)
+
+ withSQLConf(SQLConf.TIME_TRAVEL_AT_SYNTAX_ENABLED.key -> "false") {
+ checkError(
+ exception = intercept[ParseException] {
+ sql("SELECT * FROM t@v2345678910")
+ },
+ condition = "UNSUPPORTED_FEATURE.TIME_TRAVEL_AT_SYNTAX",
+ parameters = Map("config" ->
"\"spark.sql.timeTravel.atSyntax.enabled\""),
+ context = ExpectedContext(fragment = "t@v2345678910", start = 14,
stop = 26))
+ intercept[ParseException](spark.read.table("t@v2345678910"))
+ }
+ }
+
+ intercept[ParseException](sql("SELECT * FROM t@foo"))
+ intercept[ParseException](spark.read.table("t@foo"))
+ withTable("testcat.`weird@v1`") {
+ sql("CREATE TABLE testcat.`weird@v1` (id int) USING foo")
+ sql("INSERT INTO testcat.`weird@v1` VALUES (42)")
+ checkAnswer(sql("SELECT * FROM `weird@v1`"), Row(42))
+ checkAnswer(spark.read.table("`weird@v1`"), Row(42))
+ }
+
+ withTempView("v") {
+ spark.range(1).createOrReplaceTempView("v")
+ checkError(
+ exception = analysisException("SELECT * FROM v@v1"),
+ condition = "UNSUPPORTED_FEATURE.TIME_TRAVEL",
+ sqlState = None,
+ parameters = Map("relationId" -> "`v`"))
+ }
+ checkError(
+ exception = analysisException("WITH x AS (SELECT 1) SELECT * FROM x@v1"),
+ condition = "UNSUPPORTED_FEATURE.TIME_TRAVEL",
+ sqlState = None,
+ parameters = Map("relationId" -> "`x`"))
+ }
+
test("SPARK-37827: put build-in properties into V1Table.properties to adapt
v2 command") {
val t = "tbl"
withTable(t) {
diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/streaming/test/DataStreamTableAPISuite.scala
b/sql/core/src/test/scala/org/apache/spark/sql/streaming/test/DataStreamTableAPISuite.scala
index e2c74533e7f3..2ba374e3d674 100644
---
a/sql/core/src/test/scala/org/apache/spark/sql/streaming/test/DataStreamTableAPISuite.scala
+++
b/sql/core/src/test/scala/org/apache/spark/sql/streaming/test/DataStreamTableAPISuite.scala
@@ -84,6 +84,15 @@ class DataStreamTableAPISuite extends StreamTest with
BeforeAndAfter {
checkErrorTableNotFound(e, "`non_exist_table`")
}
+ test("read: time travel @-syntax is unsupported for streaming") {
+ checkError(
+ exception = intercept[AnalysisException] {
+ spark.readStream.table("t@v1")
+ },
+ condition = "UNSUPPORTED_FEATURE.TIME_TRAVEL",
+ parameters = Map("relationId" -> "`t`"))
+ }
+
test("read: stream table API with temp view") {
val tblName = "my_table"
val stream = MemoryStream[Int]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
