You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by do...@apache.org on 2019/01/27 18:05:27 UTC
[spark] branch branch-2.4 updated: [SPARK-26379][SS][FOLLOWUP] Use
dummy TimeZoneId to avoid UnresolvedException in CurrentBatchTimestamp
This is an automated email from the ASF dual-hosted git repository.
dongjoon pushed a commit to branch branch-2.4
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/branch-2.4 by this push:
new 448a063 [SPARK-26379][SS][FOLLOWUP] Use dummy TimeZoneId to avoid UnresolvedException in CurrentBatchTimestamp
448a063 is described below
commit 448a06375fcdae04a17006bdb2d02a2b5fccaca7
Author: Dongjoon Hyun <do...@apache.org>
AuthorDate: Sun Jan 27 10:04:51 2019 -0800
[SPARK-26379][SS][FOLLOWUP] Use dummy TimeZoneId to avoid UnresolvedException in CurrentBatchTimestamp
## What changes were proposed in this pull request?
Spark replaces `CurrentTimestamp` with `CurrentBatchTimestamp`.
However, `CurrentBatchTimestamp` is `TimeZoneAwareExpression` while `CurrentTimestamp` isn't.
Without TimeZoneId, `CurrentBatchTimestamp` becomes unresolved and raises `UnresolvedException`.
Since `CurrentDate` is `TimeZoneAwareExpression`, there is no problem with `CurrentDate`.
This PR reverts the [previous patch](https://github.com/apache/spark/pull/23609) on `MicroBatchExecution` and fixes the root cause.
## How was this patch tested?
Pass the Jenkins with the updated test cases.
Closes #23660 from dongjoon-hyun/SPARK-26379.
Authored-by: Dongjoon Hyun <do...@apache.org>
Signed-off-by: Dongjoon Hyun <do...@apache.org>
(cherry picked from commit 1ca6b8bc3df19503c00414e62161227725a99520)
Signed-off-by: Dongjoon Hyun <do...@apache.org>
---
.../sql/execution/streaming/MicroBatchExecution.scala | 15 +++++----------
.../org/apache/spark/sql/streaming/StreamSuite.scala | 10 +++-------
2 files changed, 8 insertions(+), 17 deletions(-)
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/MicroBatchExecution.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/MicroBatchExecution.scala
index e58182c..3bcc26b 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/MicroBatchExecution.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/MicroBatchExecution.scala
@@ -489,27 +489,22 @@ class MicroBatchExecution(
// Rewire the plan to use the new attributes that were returned by the source.
val newAttributePlan = newBatchesPlan transformAllExpressions {
case ct: CurrentTimestamp =>
+ // CurrentTimestamp is not TimeZoneAwareExpression while CurrentBatchTimestamp is.
+ // Without TimeZoneId, CurrentBatchTimestamp is unresolved. Here, we use an explicit
+ // dummy string to prevent UnresolvedException and to prevent to be used in the future.
CurrentBatchTimestamp(offsetSeqMetadata.batchTimestampMs,
- ct.dataType)
+ ct.dataType, Some("Dummy TimeZoneId"))
case cd: CurrentDate =>
CurrentBatchTimestamp(offsetSeqMetadata.batchTimestampMs,
cd.dataType, cd.timeZoneId)
}
- // Pre-resolve new attributes to ensure all attributes are resolved before
- // accessing schema of logical plan. Note that it only leverages the information
- // of attributes, so we don't need to concern about the value of literals.
-
- val newAttrPlanPreResolvedForSchema = newAttributePlan transformAllExpressions {
- case cbt: CurrentBatchTimestamp => cbt.toLiteral
- }
-
val triggerLogicalPlan = sink match {
case _: Sink => newAttributePlan
case s: StreamWriteSupport =>
val writer = s.createStreamWriter(
s"$runId",
- newAttrPlanPreResolvedForSchema.schema,
+ newAttributePlan.schema,
outputMode,
new DataSourceOptions(extraOptions.asJava))
WriteToDataSourceV2(new MicroBatchWriter(currentBatchId, writer), newAttributePlan)
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/streaming/StreamSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/streaming/StreamSuite.scala
index 766bee6..ae3689a 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/streaming/StreamSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/streaming/StreamSuite.scala
@@ -1084,21 +1084,19 @@ class StreamSuite extends StreamTest {
}
}
- test("SPARK-26379 Structured Streaming - Exception on adding current_timestamp / current_date" +
+ test("SPARK-26379 Structured Streaming - Exception on adding current_timestamp " +
" to Dataset - use v2 sink") {
testCurrentTimestampOnStreamingQuery(useV2Sink = true)
}
- test("SPARK-26379 Structured Streaming - Exception on adding current_timestamp / current_date" +
+ test("SPARK-26379 Structured Streaming - Exception on adding current_timestamp " +
" to Dataset - use v1 sink") {
testCurrentTimestampOnStreamingQuery(useV2Sink = false)
}
private def testCurrentTimestampOnStreamingQuery(useV2Sink: Boolean): Unit = {
val input = MemoryStream[Int]
- val df = input.toDS()
- .withColumn("cur_timestamp", lit(current_timestamp()))
- .withColumn("cur_date", lit(current_date()))
+ val df = input.toDS().withColumn("cur_timestamp", lit(current_timestamp()))
def assertBatchOutputAndUpdateLastTimestamp(
rows: Seq[Row],
@@ -1109,8 +1107,6 @@ class StreamSuite extends StreamTest {
val row = rows.head
assert(row.getInt(0) === expectedValue)
assert(row.getTimestamp(1).getTime >= curTimestamp)
- val days = DateTimeUtils.millisToDays(row.getDate(2).getTime)
- assert(days == curDate || days == curDate + 1)
row.getTimestamp(1).getTime
}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org