Skip to content

[SPARK-26379][SS] Fix issue on adding current_timestamp/current_date to streaming query #23609

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
HeartSaVioR wants to merge 4 commits into from
Closed

[SPARK-26379][SS] Fix issue on adding current_timestamp/current_date to streaming query #23609

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
afb0dc4
SPARK-26379 Fix issue on adding current_timestamp/current_date to str…
HeartSaVioR Jan 22, 2019
8f30a75
Fix build issue which local maven wasn't caught as error strangely
HeartSaVioR Jan 22, 2019
d09845b
Address review comments from @gaborgsomogyi
HeartSaVioR Jan 22, 2019
687c3e4
Reduce test time
HeartSaVioR Jan 22, 2019
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
10 changes: 9 additions & 1 deletion sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/MicroBatchExecution.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -508,12 +508,20 @@ class MicroBatchExecution(
cd.dataType, cd.timeZoneId)
}
Copy link
Copy Markdown
Member

@dongjoon-hyun dongjoon-hyun Jan 26, 2019

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The root cause is CurrentBatchTimestamp is TimeZoneAwareExpression which is unresolved without 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 {
Copy link
Copy Markdown
Contributor Author

@HeartSaVioR HeartSaVioR Jan 22, 2019

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

We can even leverage newBatchesPlan and remove this: this effort is to ensure going through the same path with further transformation when extracting schema.

Copy link
Copy Markdown
Contributor

@gaborgsomogyi gaborgsomogyi Jan 22, 2019

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I'm fine with this.

Copy link
Copy Markdown
Member

@dongjoon-hyun dongjoon-hyun Jan 23, 2019

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@HeartSaVioR . I prefer to leverage the existing lines (501 ~ 509) as you said like the following. I assume that the following is what you mean. Eventually, it's two line changes (excluding comments). WDYT?

    // Rewire the plan to use the new attributes that were returned by the source.
    val newAttributePlan = newBatchesPlan transformAllExpressions {
      case ct: CurrentTimestamp =>
        CurrentBatchTimestamp(offsetSeqMetadata.batchTimestampMs,
-          ct.dataType)
+          ct.dataType).toLiteral
      case cd: CurrentDate =>
        CurrentBatchTimestamp(offsetSeqMetadata.batchTimestampMs,
-          cd.dataType, cd.timeZoneId)
+          cd.dataType, cd.timeZoneId).toLiteral
    }

Copy link
Copy Markdown
Contributor Author

@HeartSaVioR HeartSaVioR Jan 23, 2019

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@dongjoon-hyun
Actually I meant we pass newBatchesPlan instead of newAttributePlan only for extracting schema since we actually guess the name and data type will not be changed from IncrementalExecution. But IMHO I feel safer pre-transforming it.

Transforming CurrentBatchTimestamp to Literal not in IncrementalExecution breaks the intention what CurrentBatchTimestamp javadoc describes. Please note that I used transformed plan only for extracting schema and the actual plan is not changed.

Copy link
Copy Markdown
Member

@dongjoon-hyun dongjoon-hyun Jan 25, 2019

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

+1

case cbt: CurrentBatchTimestamp => cbt.toLiteral
}

val triggerLogicalPlan = sink match {
case _: Sink => newAttributePlan
case s: StreamingWriteSupportProvider =>
val writer = s.createStreamingWriteSupport(
s"$runId",
newAttributePlan.schema,
newAttrPlanPreResolvedForSchema.schema,
outputMode,
new DataSourceOptions(extraOptions.asJava))
WriteToDataSourceV2(new MicroBatchWrite(currentBatchId, writer), newAttributePlan)
Expand Down
46 changes: 46 additions & 0 deletions sql/core/src/test/scala/org/apache/spark/sql/streaming/StreamSuite.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -34,6 +34,7 @@ import org.apache.spark.scheduler.{SparkListener, SparkListenerJobStart}
import org.apache.spark.sql._
import org.apache.spark.sql.catalyst.plans.logical.Range
import org.apache.spark.sql.catalyst.streaming.InternalOutputModes
import org.apache.spark.sql.catalyst.util.DateTimeUtils
import org.apache.spark.sql.execution.command.ExplainCommand
import org.apache.spark.sql.execution.streaming._
import org.apache.spark.sql.execution.streaming.sources.ContinuousMemoryStream
Expand Down Expand Up @@ -1079,6 +1080,51 @@ class StreamSuite extends StreamTest {
assert(query.exception.isEmpty)
}
}

test("SPARK-26379 Structured Streaming - Exception on adding current_timestamp / current_date" +
" to Dataset - use v2 sink") {
testCurrentTimestampOnStreamingQuery(useV2Sink = true)
}

test("SPARK-26379 Structured Streaming - Exception on adding current_timestamp / current_date" +
" 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()))
Copy link
Copy Markdown
Member

@dongjoon-hyun dongjoon-hyun Jan 26, 2019

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Currently, this hides current_timestamp and makes this test case succeeds always.


def assertBatchOutputAndUpdateLastTimestamp(
rows: Seq[Row],
curTimestamp: Long,
curDate: Int,
expectedValue: Int): Long = {
assert(rows.size === 1)
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
}

var lastTimestamp = System.currentTimeMillis()
val currentDate = DateTimeUtils.millisToDays(lastTimestamp)
testStream(df, useV2Sink = useV2Sink) (
AddData(input, 1),
CheckLastBatch { rows: Seq[Row] =>
lastTimestamp = assertBatchOutputAndUpdateLastTimestamp(rows, lastTimestamp, currentDate, 1)
},
Execute { _ => Thread.sleep(1000) },
AddData(input, 2),
CheckLastBatch { rows: Seq[Row] =>
lastTimestamp = assertBatchOutputAndUpdateLastTimestamp(rows, lastTimestamp, currentDate, 2)
}
)
}
}

abstract class FakeSource extends StreamSourceProvider {
Expand Down