This is an automated email from the ASF dual-hosted git repository.
gengliangwang pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/master by this push:
new e27dc9791d30 [SPARK-58012][SQL][CONNECT] Support @-syntax timestamp
time travel on table names
e27dc9791d30 is described below
commit e27dc9791d30190fdbe73376f57cbdc3969d7f31
Author: sotikoug83 <[email protected]>
AuthorDate: Thu Jul 9 20:12:03 2026 -0700
[SPARK-58012][SQL][CONNECT] Support @-syntax timestamp time travel on table
names
### What changes were proposed in this pull request?
Add the `` timestamp time travel shorthand on table names so that
```
SELECT * FROM name<TS>
```
and
```
spark.read.format("format").table("name<ts>")
```
parse into the same plan as the existing `TIMESTAMP AS OF` and
`option("timestampAsOf")` clauses. Support for SQL, the DataFrame reader, and
Spark Connect.
### Why are the changes needed?
Spark has SQL `AS OF` and the `timestampAsOf` 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.
Additionally, different from `TIMESTAMP AS OF <TS>` (which allows for arbitrary
expressions in `<TS>`), the `<ts>` timestamp only accepts the fixed 17-digit
`yyyyMMddHHmmssSSS` Delta convention.
### 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 #56847 from sotikoug83/SPARK-57421-at-syntax-timestamp.
Authored-by: sotikoug83 <[email protected]>
Signed-off-by: Gengliang Wang <[email protected]>
---
.../src/main/resources/error/error-conditions.json | 6 ++
.../spark/sql/catalyst/parser/SqlBaseLexer.g4 | 1 +
.../spark/sql/catalyst/parser/SqlBaseParser.g4 | 6 +-
.../spark/sql/errors/QueryParsingErrors.scala | 10 ++++
.../spark/sql/catalyst/parser/AstBuilder.scala | 48 ++++++++++++---
.../sql/catalyst/parser/ParserInterface.scala | 2 +-
.../sql/catalyst/parser/TemporalIdentifier.scala | 14 ++++-
.../org/apache/spark/sql/internal/SQLConf.scala | 7 ++-
.../sql/catalyst/parser/PlanParserSuite.scala | 70 ++++++++++++++++++++--
.../connect/planner/SparkConnectPlannerSuite.scala | 19 ++++++
.../spark/sql/connector/DataSourceV2SQLSuite.scala | 31 ++++++++++
.../streaming/test/DataStreamTableAPISuite.scala | 6 ++
12 files changed, 200 insertions(+), 20 deletions(-)
diff --git a/common/utils/src/main/resources/error/error-conditions.json
b/common/utils/src/main/resources/error/error-conditions.json
index b09909f8f9aa..4c5cf5fe5d69 100644
--- a/common/utils/src/main/resources/error/error-conditions.json
+++ b/common/utils/src/main/resources/error/error-conditions.json
@@ -5116,6 +5116,12 @@
},
"sqlState" : "42K0E"
},
+ "INVALID_TIME_TRAVEL_TIMESTAMP_FORMAT" : {
+ "message" : [
+ "The provided timestamp <timestamp> doesn't match the expected syntax
<format>."
+ ],
+ "sqlState" : "22000"
+ },
"INVALID_TYPED_LITERAL" : {
"message" : [
"The value of the typed literal <valueType> is invalid: <value>."
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 e15be3fb76ec..3e5c79613f19 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
@@ -595,6 +595,7 @@ OPERATOR_PIPE: '|>';
HAT: '^';
COLON: ':';
DOUBLE_COLON: '::';
+AT_SIGN: '@';
AT_VERSION: '@V';
ARROW: '->';
FAT_ARROW : '=>';
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 38d4460965fb..6b2237f15cc4 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
@@ -1276,12 +1276,14 @@ tableIdentifier
;
temporalTableIdentifier
- : id=multipartIdentifier AT_VERSION version
+ : id=multipartIdentifier AT_SIGN timestamp=INTEGER_VALUE
+ | id=multipartIdentifier AT_VERSION version
| id=multipartIdentifier
;
temporalTableIdentifierReference
- : identifierReference AT_VERSION version
+ : identifierReference AT_SIGN timestamp=INTEGER_VALUE
+ | identifierReference AT_VERSION version
| identifierReference
;
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 01ba078be729..f73b0457f87f 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,16 @@ private[sql] object QueryParsingErrors extends
DataTypeErrorsBase {
ctx)
}
+ def invalidAtSyntaxTimestamp(
+ timestamp: String,
+ format: String,
+ ctx: ParserRuleContext): Throwable = {
+ new ParseException(
+ errorClass = "INVALID_TIME_TRAVEL_TIMESTAMP_FORMAT",
+ messageParameters = Map("timestamp" -> timestamp, "format" -> format),
+ ctx)
+ }
+
def multipleTimeTravelSpec(ctx: ParserRuleContext): Throwable = {
new ParseException(
errorClass = "MULTIPLE_TIME_TRAVEL_SPEC",
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 8389e79331a9..032e032fa1a9 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
@@ -17,6 +17,7 @@
package org.apache.spark.sql.catalyst.parser
+import java.time.{DateTimeException, LocalDateTime}
import java.util.{List, Locale}
import java.util.concurrent.TimeUnit
@@ -45,7 +46,7 @@ import org.apache.spark.sql.catalyst.plans.logical._
import org.apache.spark.sql.catalyst.trees.{CurrentOrigin, Origin}
import org.apache.spark.sql.catalyst.trees.TreePattern.PARAMETER
import org.apache.spark.sql.catalyst.types.DataTypeUtils
-import org.apache.spark.sql.catalyst.util.{CharVarcharUtils, CollationFactory,
DateTimeUtils, EvaluateUnresolvedInlineTable, IntervalUtils}
+import org.apache.spark.sql.catalyst.util.{CharVarcharUtils, CollationFactory,
DateTimeConstants, DateTimeUtils, EvaluateUnresolvedInlineTable, IntervalUtils}
import org.apache.spark.sql.catalyst.util.DateTimeUtils.{convertSpecialDate,
convertSpecialTimestamp, convertSpecialTimestampNTZ, fractionalSecondsDigits,
getZoneId, stringToDate, stringToTime, stringToTimestamp,
stringToTimestampLTZNanos, stringToTimestampNTZNanos,
stringToTimestampWithoutTimeZone}
import org.apache.spark.sql.connector.catalog.{CatalogV2Util,
ChangelogContext, PathElement, SupportsNamespaces, TableCatalog,
TableWritePrivilege}
import org.apache.spark.sql.connector.catalog.ChangelogRange.{TimestampRange,
UnboundedRange, VersionRange}
@@ -2705,7 +2706,7 @@ class AstBuilder extends DataTypeAstBuilder
relation: LogicalPlan,
ttCtx: TemporalTableIdentifierReferenceContext,
clause: TemporalClauseContext): LogicalPlan = {
- val (atTimestamp, atVersion) = temporalSpec(ttCtx, ttCtx.version)
+ val (atTimestamp, atVersion) = temporalSpec(ttCtx, ttCtx.timestamp,
ttCtx.version)
val hasAtSpec = atTimestamp.isDefined || atVersion.isDefined
if (hasAtSpec && clause != null) {
withOrigin(clause) {
@@ -2719,24 +2720,57 @@ class AstBuilder extends DataTypeAstBuilder
override def visitTemporalTableIdentifier(
ctx: TemporalTableIdentifierContext): TemporalIdentifier =
withOrigin(ctx) {
- val (_, version) = temporalSpec(ctx, ctx.version)
- TemporalIdentifier(visitMultipartIdentifier(ctx.id), version)
+ val (timestamp, version) = temporalSpec(ctx, ctx.timestamp, ctx.version)
+ TemporalIdentifier(visitMultipartIdentifier(ctx.id), timestamp, version)
}
/**
- * Extract the optional '@v<version>' time travel suffix of a table
identifier.
+ * Parse the digits of an '@' time travel timestamp (format
yyyyMMddHHmmssSSS) to
+ * microseconds since epoch in the session time zone.
+ */
+ private def parseAtSyntaxTimestamp(text: String, ctx: ParserRuleContext):
Long = {
+ val format = TemporalIdentifier.TimestampFormat
+ if (text.length != format.length) {
+ throw QueryParsingErrors.invalidAtSyntaxTimestamp(text, format, ctx)
+ }
+ try {
+ val localDateTime = LocalDateTime.of(
+ text.substring(0, 4).toInt,
+ text.substring(4, 6).toInt,
+ text.substring(6, 8).toInt,
+ text.substring(8, 10).toInt,
+ text.substring(10, 12).toInt,
+ text.substring(12, 14).toInt,
+ text.substring(14, 17).toInt *
DateTimeConstants.NANOS_PER_MILLIS.toInt)
+ DateTimeUtils.instantToMicros(
+ localDateTime.atZone(getZoneId(conf.sessionLocalTimeZone)).toInstant)
+ } catch {
+ case _: DateTimeException =>
+ throw QueryParsingErrors.invalidAtSyntaxTimestamp(text, format, ctx)
+ }
+ }
+
+ /**
+ * Extract the optional '@' time travel suffix of a table identifier:
'@<timestamp>'
+ * (format yyyyMMddHHmmssSSS) or '@v<version>'.
*/
private def temporalSpec(
ctx: ParserRuleContext,
+ timestampToken: Token,
versionCtx: VersionContext): (Option[Expression], Option[String]) = {
- if (versionCtx == null) {
+ if (timestampToken == null && 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))
+ if (timestampToken != null) {
+ val micros = parseAtSyntaxTimestamp(timestampToken.getText, ctx)
+ (Some(Literal(micros, TimestampType)), None)
+ } else {
+ (None, visitVersion(versionCtx))
+ }
}
}
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 4456303ffef5..4ca6fe3cbd91 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
@@ -76,7 +76,7 @@ trait ParserInterface extends DataTypeParserInterface {
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)
+ TemporalIdentifier(parseMultipartIdentifier(sqlText), None, None)
}
/**
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
index 9be5c4016a7c..44bd7020155a 100644
---
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
@@ -18,19 +18,27 @@
package org.apache.spark.sql.catalyst.parser
import org.apache.spark.sql.catalyst.analysis.RelationTimeTravel
+import org.apache.spark.sql.catalyst.expressions.Expression
import org.apache.spark.sql.catalyst.plans.logical.LogicalPlan
/**
- * Result of parsing a table name that may carry an '@v<version>' time travel
suffix.
+ * Result of parsing a table name that may carry an '@' time travel suffix:
`name@v<version>`
+ * or `name@<yyyyMMddHHmmssSSS>`. At most one of `timestamp` and `version` is
set.
*/
case class TemporalIdentifier(
nameParts: Seq[String],
+ timestamp: Option[Expression],
version: Option[String]) {
- def isTemporal: Boolean = version.isDefined
+ def isTemporal: Boolean = timestamp.isDefined || 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
+ if (isTemporal) RelationTimeTravel(plan, timestamp, version) else plan
}
}
+
+object TemporalIdentifier {
+ /** The fixed timestamp format accepted in the suffix. */
+ val TimestampFormat: String = "yyyyMMddHHmmssSSS"
+}
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 3315cdc72332..a3f0fd8ab43e 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
@@ -7186,8 +7186,11 @@ object SQLConf {
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.")
+ "travel suffix: 'name@v123' reads version 123 of the table, and " +
+ "'name@20240101000000000' (format yyyyMMddHHmmssSSS, interpreted in
the session " +
+ "time zone) reads the table as of that timestamp, to millisecond
precision (use " +
+ "TIMESTAMP AS OF for finer granularity). When false, '@' in table
names fails at " +
+ "parse time.")
.version("4.3.0")
.withBindingPolicy(ConfigBindingPolicy.NOT_APPLICABLE)
.booleanConf
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 e44ba2194a8e..e9f1a94f7010 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
@@ -26,11 +26,12 @@ import
org.apache.spark.sql.catalyst.analysis.{AnalysisTest, RelationChanges, Re
import org.apache.spark.sql.catalyst.expressions._
import org.apache.spark.sql.catalyst.plans._
import org.apache.spark.sql.catalyst.plans.logical._
-import org.apache.spark.sql.catalyst.util.{EvaluateUnresolvedInlineTable,
IntervalUtils}
+import org.apache.spark.sql.catalyst.util.{DateTimeUtils,
EvaluateUnresolvedInlineTable, IntervalUtils}
import org.apache.spark.sql.connector.catalog.{ChangelogContext,
ChangelogRange}
import org.apache.spark.sql.internal.SQLConf
-import org.apache.spark.sql.types.{Decimal, DecimalType, IntegerType,
LongType, StringType}
+import org.apache.spark.sql.types.{Decimal, DecimalType, IntegerType,
LongType, StringType, TimestampType}
import org.apache.spark.sql.util.CaseInsensitiveStringMap
+import org.apache.spark.unsafe.types.UTF8String
/**
* Parser test cases for rules defined in [[CatalystSqlParser]] /
[[AstBuilder]].
@@ -2203,6 +2204,26 @@ class PlanParserSuite extends AnalysisTest {
assertEqual("SELECT * FROM a.b.c @v123456789", versionPlan("123456789"))
assertEqual("SELECT * FROM a.b.c@v'Snapshot123456789'",
versionPlan("Snapshot123456789"))
+ val micros = DateTimeUtils.stringToTimestampAnsi(
+ UTF8String.fromString("2019-01-29 00:37:58"),
+ DateTimeUtils.getZoneId(SQLConf.get.sessionLocalTimeZone))
+ assertEqual("SELECT * FROM a.b.c@20190129003758000",
+ Project(Seq(UnresolvedStar(None)),
+ RelationTimeTravel(
+ UnresolvedRelation(Seq("a", "b", "c")),
+ Some(Literal(micros, TimestampType)),
+ None)))
+
+ val microsWithMillis = DateTimeUtils.stringToTimestampAnsi(
+ UTF8String.fromString("2019-01-29 00:37:58.123"),
+ DateTimeUtils.getZoneId(SQLConf.get.sessionLocalTimeZone))
+ assertEqual("SELECT * FROM a.b.c@20190129003758123",
+ Project(Seq(UnresolvedStar(None)),
+ RelationTimeTravel(
+ UnresolvedRelation(Seq("a", "b", "c")),
+ Some(Literal(microsWithMillis, TimestampType)),
+ None)))
+
assertEqual("SELECT * FROM `t@v1`",
Project(Seq(UnresolvedStar(None)), UnresolvedRelation(Seq("t@v1"))))
@@ -2223,6 +2244,40 @@ class PlanParserSuite extends AnalysisTest {
condition = "MULTIPLE_TIME_TRAVEL_SPEC",
parameters = Map.empty,
context = ExpectedContext(fragment = "TIMESTAMP AS OF '2019-01-29'",
start = 19, stop = 46))
+ checkError(
+ exception = parseException("SELECT * FROM t@20190129003758000 VERSION AS
OF 2"),
+ condition = "MULTIPLE_TIME_TRAVEL_SPEC",
+ parameters = Map.empty,
+ context = ExpectedContext(fragment = "VERSION AS OF 2", start = 34, stop
= 48))
+ checkError(
+ exception = parseException("SELECT * FROM t@20190129003758000 TIMESTAMP
AS OF '2019-01-29'"),
+ condition = "MULTIPLE_TIME_TRAVEL_SPEC",
+ parameters = Map.empty,
+ context = ExpectedContext(fragment = "TIMESTAMP AS OF '2019-01-29'",
start = 34, stop = 61))
+
+ checkError(
+ exception = parseException("SELECT * FROM t@123"),
+ condition = "INVALID_TIME_TRAVEL_TIMESTAMP_FORMAT",
+ parameters = Map("timestamp" -> "123", "format" -> "yyyyMMddHHmmssSSS"),
+ context = ExpectedContext(fragment = "t@123", start = 14, stop = 18))
+
+ checkError(
+ exception = parseException("SELECT * FROM t@20191301000000000"),
+ condition = "INVALID_TIME_TRAVEL_TIMESTAMP_FORMAT",
+ parameters = Map("timestamp" -> "20191301000000000", "format" ->
"yyyyMMddHHmmssSSS"),
+ context = ExpectedContext(fragment = "t@20191301000000000", start = 14,
stop = 32))
+
+ // Wrong-length timestamps are rejected.
+ checkError(
+ exception = parseException("SELECT * FROM t@20190129003758"),
+ condition = "INVALID_TIME_TRAVEL_TIMESTAMP_FORMAT",
+ parameters = Map("timestamp" -> "20190129003758", "format" ->
"yyyyMMddHHmmssSSS"),
+ context = ExpectedContext(fragment = "t@20190129003758", start = 14,
stop = 29))
+ checkError(
+ exception = parseException("SELECT * FROM t@20190129"),
+ condition = "INVALID_TIME_TRAVEL_TIMESTAMP_FORMAT",
+ parameters = Map("timestamp" -> "20190129", "format" ->
"yyyyMMddHHmmssSSS"),
+ context = ExpectedContext(fragment = "t@20190129", start = 14, stop =
23))
assert(intercept[ParseException] {
parsePlan("INSERT INTO t@v1 VALUES (1)")
@@ -2239,11 +2294,16 @@ class PlanParserSuite extends AnalysisTest {
test("parseTemporalTableIdentifier") {
assert(parseTemporalTableIdentifier("a.b") ===
- TemporalIdentifier(Seq("a", "b"), None))
+ TemporalIdentifier(Seq("a", "b"), None, None))
assert(parseTemporalTableIdentifier("a.b@v5") ===
- TemporalIdentifier(Seq("a", "b"), Some("5")))
+ TemporalIdentifier(Seq("a", "b"), None, Some("5")))
assert(parseTemporalTableIdentifier("`t@v1`") ===
- TemporalIdentifier(Seq("t@v1"), None))
+ TemporalIdentifier(Seq("t@v1"), None, None))
+ val micros = DateTimeUtils.stringToTimestampAnsi(
+ UTF8String.fromString("2019-01-29 00:37:58"),
+ DateTimeUtils.getZoneId(SQLConf.get.sessionLocalTimeZone))
+ assert(parseTemporalTableIdentifier("t@20190129003758000") ===
+ TemporalIdentifier(Seq("t"), Some(Literal(micros, TimestampType)), None))
Seq("a.b@x", "a@foo", "a@", "a@v").foreach { s =>
intercept[ParseException](parseTemporalTableIdentifier(s))
}
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 22f7cabc8c18..991b3f2a6c05 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
@@ -160,6 +160,7 @@ class SparkConnectPlannerSuite extends SparkFunSuite with
SparkConnectPlanTest {
}
test("Read with at syntax time travel") {
+ // Version time travel
val read = proto.Read
.newBuilder()
.setNamedTable(proto.Read.NamedTable.newBuilder.setUnparsedIdentifier("name@v1").build())
@@ -173,12 +174,30 @@ class SparkConnectPlannerSuite extends SparkFunSuite with
SparkConnectPlanTest {
case other => fail(s"Expected RelationTimeTravel but got: $other")
}
+ // Timestamp time travel
+ val tsRead = read.toBuilder
+ .setNamedTable(
+
proto.Read.NamedTable.newBuilder.setUnparsedIdentifier("name@20190129003758000").build())
+ .build()
+ transform(proto.Relation.newBuilder.setRead(tsRead).build()) match {
+ case RelationTimeTravel(relation: UnresolvedRelation, timestamp,
version) =>
+ assert(relation.multipartIdentifier === Seq("name"))
+ assert(timestamp.isDefined)
+ assert(version.isEmpty)
+ 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")
+ val streamingTsRead = tsRead.toBuilder.setIsStreaming(true).build()
+ val tsErr = intercept[AnalysisException] {
+ transform(proto.Relation.newBuilder.setRead(streamingTsRead).build())
+ }
+ assert(tsErr.getCondition === "UNSUPPORTED_FEATURE.TIME_TRAVEL")
// A non-time-travel '@' suffix is a parse error.
val badRead = read.toBuilder
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 7ce2945898c3..4f855e7e453d 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
@@ -4003,6 +4003,37 @@ class DataSourceV2SQLSuiteV1Filter
}
}
+ val ts1 = DateTimeUtils.stringToTimestampAnsi(
+ UTF8String.fromString("2019-01-29 00:37:58"),
+ DateTimeUtils.getZoneId(SQLConf.get.sessionLocalTimeZone))
+ val t3 = s"testcat.t$ts1"
+ withTable(t3) {
+ sql(s"CREATE TABLE $t3 (id int) USING foo")
+ sql(s"INSERT INTO $t3 VALUES (5), (6)")
+
+ checkAnswer(sql("SELECT * FROM t@20190129003758000"), Seq(Row(5),
Row(6)))
+ checkAnswer(spark.read.table("t@20190129003758000"), Seq(Row(5), Row(6)))
+
+ checkError(
+ exception = intercept[AnalysisException] {
+ spark.read.option("versionAsOf",
"1").table("t@20190129003758000").collect()
+ },
+ condition = "MULTIPLE_TIME_TRAVEL_SPEC",
+ parameters = Map.empty)
+ checkError(
+ exception = intercept[AnalysisException] {
+ spark.read.option("timestampAsOf",
"2019-01-29").table("t@20190129003758000").collect()
+ },
+ condition = "MULTIPLE_TIME_TRAVEL_SPEC",
+ parameters = Map.empty)
+ checkError(
+ exception = intercept[AnalysisException] {
+ sql("SELECT * FROM t@20190129003758000 WITH ('timestampAsOf' =
'2019-01-29')")
+ },
+ condition = "MULTIPLE_TIME_TRAVEL_SPEC",
+ parameters = Map.empty)
+ }
+
intercept[ParseException](sql("SELECT * FROM t@foo"))
intercept[ParseException](spark.read.table("t@foo"))
withTable("testcat.`weird@v1`") {
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 2ba374e3d674..da41abe76bd9 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
@@ -91,6 +91,12 @@ class DataStreamTableAPISuite extends StreamTest with
BeforeAndAfter {
},
condition = "UNSUPPORTED_FEATURE.TIME_TRAVEL",
parameters = Map("relationId" -> "`t`"))
+ checkError(
+ exception = intercept[AnalysisException] {
+ spark.readStream.table("t@20190129003758000")
+ },
+ condition = "UNSUPPORTED_FEATURE.TIME_TRAVEL",
+ parameters = Map("relationId" -> "`t`"))
}
test("read: stream table API with temp view") {
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
