This is an automated email from the ASF dual-hosted git repository. tdas 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 e4237bb [SPARK-32794][SS] Fixed rare corner case error in micro-batch engine with some stateful queries + no-data-batches + V1 sources e4237bb is described below commit e4237bbda68c23a3367fab56fd8cdb521f8a1ae2 Author: Tathagata Das <tathagata.das1...@gmail.com> AuthorDate: Wed Sep 9 13:35:51 2020 -0400 [SPARK-32794][SS] Fixed rare corner case error in micro-batch engine with some stateful queries + no-data-batches + V1 sources ### What changes were proposed in this pull request? Make MicroBatchExecution explicitly call `getBatch` when the start and end offsets are the same. ### Why are the changes needed? Structured Streaming micro-batch engine has the contract with V1 data sources that, after a restart, it will call `source.getBatch()` on the last batch attempted before the restart. However, a very rare combination of sequences violates this contract. It occurs only when - The streaming query has specific types of stateful operations with watermarks (e.g., aggregation in append, mapGroupsWithState with timeouts). - These queries can execute a batch even without new data when the previous updates the watermark and the stateful ops are such that the new watermark can cause new output/cleanup. Such batches are called no-data-batches. - The last batch before termination was an incomplete no-data-batch. Upon restart, the micro-batch engine fails to call `source.getBatch` when attempting to re-execute the incomplete no-data-batch. This occurs because no-data-batches has the same and end offsets, and when a batch is executed, if the start and end offset is same then calling `source.getBatch` is skipped as it is assumed the generated plan will be empty. This only affects V1 data sources like Delta and Autoloader which rely on this invariant to detect in the source whether the query is being started from scratch or restarted. ### Does this PR introduce _any_ user-facing change? No ### How was this patch tested? New unit test with a mock v1 source that fails without the fix. Closes #29651 from tdas/SPARK-32794. Authored-by: Tathagata Das <tathagata.das1...@gmail.com> Signed-off-by: Tathagata Das <tathagata.das1...@gmail.com> --- .../execution/streaming/MicroBatchExecution.scala | 11 ++ .../streaming/MicroBatchExecutionSuite.scala | 123 +++++++++++++++++++++ .../apache/spark/sql/streaming/StreamTest.scala | 8 ++ 3 files changed, 142 insertions(+) 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 abbfea8..468a8c9 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 @@ -321,6 +321,17 @@ class MicroBatchExecution( committedOffsets ++= availableOffsets watermarkTracker.setWatermark( math.max(watermarkTracker.currentWatermark, commitMetadata.nextBatchWatermarkMs)) + } else if (latestCommittedBatchId == latestBatchId - 1) { + availableOffsets.foreach { + case (source: Source, end: Offset) => + val start = committedOffsets.get(source).map(_.asInstanceOf[Offset]) + if (start.map(_ == end).getOrElse(true)) { + source.getBatch(start, end) + } + case nonV1Tuple => + // The V2 API does not have the same edge case requiring getBatch to be called + // here, so we do nothing here. + } } else if (latestCommittedBatchId < latestBatchId - 1) { logWarning(s"Batch completion log latest batch id is " + s"${latestCommittedBatchId}, which is not trailing " + diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/streaming/MicroBatchExecutionSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/streaming/MicroBatchExecutionSuite.scala index c0f25e3..a508f92 100644 --- a/sql/core/src/test/scala/org/apache/spark/sql/execution/streaming/MicroBatchExecutionSuite.scala +++ b/sql/core/src/test/scala/org/apache/spark/sql/execution/streaming/MicroBatchExecutionSuite.scala @@ -19,8 +19,13 @@ package org.apache.spark.sql.execution.streaming import org.scalatest.BeforeAndAfter +import org.apache.spark.sql.{DataFrame, Dataset, SparkSession} +import org.apache.spark.sql.catalyst.plans.logical.Range +import org.apache.spark.sql.connector.read.streaming +import org.apache.spark.sql.connector.read.streaming.SparkDataStream import org.apache.spark.sql.functions.{count, timestamp_seconds, window} import org.apache.spark.sql.streaming.StreamTest +import org.apache.spark.sql.types.{LongType, StructType} class MicroBatchExecutionSuite extends StreamTest with BeforeAndAfter { @@ -68,4 +73,122 @@ class MicroBatchExecutionSuite extends StreamTest with BeforeAndAfter { CheckNewAnswer((25, 1), (30, 1)) // This should not throw the error reported in SPARK-24156 ) } + + test("no-data-batch re-executed after restart should call V1 source.getBatch()") { + val testSource = ReExecutedBatchTestSource(spark) + val df = testSource.toDF() + .withColumn("eventTime", timestamp_seconds($"value")) + .withWatermark("eventTime", "10 seconds") + .groupBy(window($"eventTime", "5 seconds") as 'window) + .agg(count("*") as 'count) + .select($"window".getField("start").cast("long").as[Long]) + + /** Reset this test source so that it appears to be a new source requiring initialization */ + def resetSource(): StreamAction = Execute("reset source") { _ => + testSource.reset() // Make it look like a new source that needs to be re-initialized + require(testSource.currentOffset === 0) + require(testSource.getBatchCallCount === 0) + } + + /** Add data to this test source by incrementing its available offset */ + def addData(numNewRows: Int): StreamAction = new AddData { + override def addData(query: Option[StreamExecution]): (SparkDataStream, streaming.Offset) = { + testSource.incrementAvailableOffset(numNewRows) + (testSource, testSource.getOffset.get) + } + } + + testStream(df)( + addData(numNewRows = 10), // generate values 1...10, sets watermark to 0 + CheckAnswer(), + addData(numNewRows = 10), // generate values 11...20, sets watermark to 10 + ProcessAllAvailable(), // let no-data-batch be executed + CheckAnswer(0, 5), // start time of windows closed and outputted + Execute("verify source internal state before stop") { q => + // Last batch should be a no-data batch + require(q.lastProgress.numInputRows === 0) + // Source should have expected internal state + require(testSource.currentOffset === 20) + // getBatch should be called only for 2 batches with data, not for no-data-batches + assert(testSource.getBatchCallCount === 2) + }, + StopStream, + + /* Verify that if the last no-data-batch was incomplete, getBatch() is called only once */ + Execute("mark last batch as incomplete") { q => + // Delete the last committed batch from the commit log to signify that the last batch + // (a no-data batch) did not complete and has to be re-executed on restart. + val commit = q.commitLog.getLatest().map(_._1).getOrElse(-1L) + q.commitLog.purgeAfter(commit - 1) + }, + resetSource(), + StartStream(), + ProcessAllAvailable(), // allow initialization and re-execution + Execute("verify source.getBatch() called after re-executed no-data-batch") { q => + // After restart, getBatch() should be called once even for no-data batch + assert(testSource.getBatchCallCount === 1) + assert(testSource.currentOffset === 20) + }, + addData(numNewRows = 10), // generate values 21...30, sets watermark to 20 + ProcessAllAvailable(), // let no-data-batch be executed + CheckAnswer(0, 5, 10, 15), + StopStream, + + /* Verify that if the last no-data-batch was complete, getBatch() is still called only once */ + Execute("verify last batch was complete") { q => + // Verify that the commit log records the last batch as completed + require(q.commitLog.getLatest().map(_._1).get === q.offsetLog.getLatest().map(_._1).get) + }, + resetSource(), + StartStream(), + ProcessAllAvailable(), // allow initialization to completed + Execute("verify source.getBatch() called even if no-data-batch was not re-executed") { q => + // After restart, getBatch() should be called even for no-data batch, but only once + assert(testSource.getBatchCallCount === 1) + assert(testSource.currentOffset === 30) + }, + addData(numNewRows = 10), // generate values 31...40, sets watermark to 30 + ProcessAllAvailable(), // let no-data-batch be executed + CheckAnswer(0, 5, 10, 15, 20, 25) + ) + } + + + case class ReExecutedBatchTestSource(spark: SparkSession) extends Source { + @volatile var currentOffset = 0L + @volatile var getBatchCallCount = 0 + + override def getOffset: Option[Offset] = { + if (currentOffset <= 0) None else Some(LongOffset(currentOffset)) + } + + override def getBatch(start: Option[Offset], end: Offset): DataFrame = { + getBatchCallCount = getBatchCallCount + 1 + if (currentOffset == 0) currentOffset = getOffsetValue(end) + val plan = Range( + start.map(getOffsetValue).getOrElse(0L) + 1L, getOffsetValue(end) + 1L, 1, None, + isStreaming = true) + Dataset.ofRows(spark, plan) + } + + def incrementAvailableOffset(numNewRows: Int): Unit = { + currentOffset = currentOffset + numNewRows + } + + def reset(): Unit = { + currentOffset = 0L + getBatchCallCount = 0 + } + def toDF(): DataFrame = Dataset.ofRows(spark, StreamingExecutionRelation(this, spark)) + override def schema: StructType = new StructType().add("value", LongType) + override def stop(): Unit = {} + private def getOffsetValue(offset: Offset): Long = { + offset match { + case s: SerializedOffset => LongOffset(s).offset + case l: LongOffset => l.offset + case _ => throw new IllegalArgumentException("incorrect offset type: " + offset) + } + } + } } + diff --git a/sql/core/src/test/scala/org/apache/spark/sql/streaming/StreamTest.scala b/sql/core/src/test/scala/org/apache/spark/sql/streaming/StreamTest.scala index ceea1e4..7a2e29f 100644 --- a/sql/core/src/test/scala/org/apache/spark/sql/streaming/StreamTest.scala +++ b/sql/core/src/test/scala/org/apache/spark/sql/streaming/StreamTest.scala @@ -306,6 +306,14 @@ trait StreamTest extends QueryTest with SharedSparkSession with TimeLimits with def apply(func: StreamExecution => Any): AssertOnQuery = apply("Execute")(func) } + /** Call `StreamingQuery.processAllAvailable()` to wait. */ + object ProcessAllAvailable { + def apply(): AssertOnQuery = AssertOnQuery { query => + query.processAllAvailable() + true + } + } + object AwaitEpoch { def apply(epoch: Long): AssertOnQuery = Execute { --------------------------------------------------------------------- To unsubscribe, e-mail: commits-unsubscr...@spark.apache.org For additional commands, e-mail: commits-h...@spark.apache.org