Github user brkyvz commented on a diff in the pull request:
https://github.com/apache/spark/pull/20941#discussion_r178208005
--- Diff:
sql/core/src/test/scala/org/apache/spark/sql/streaming/StreamTest.scala ---
@@ -444,6 +445,26 @@ trait StreamTest extends QueryTest with
SharedSQLContext with TimeLimits with Be
}
}
+ if (currentStream.isInstanceOf[MicroBatchExecution]) {
+ // Verify if stateful operators have correct metadata and
distribution
+ // This can often catch hard to debug errors when developing
stateful operators
+ val executedPlan = currentStream.lastExecution.executedPlan
+ executedPlan.collect { case s: StatefulOperator => s }.foreach { s
=>
+ assert(s.stateInfo.isDefined)
+ assert(s.stateInfo.get.numPartitions >= 1)
+
+ s.requiredChildDistribution.foreach { d =>
+ withClue(s"$s specifies incorrect # partitions in
requiredChildDistribution $d") {
+ assert(d.requiredNumPartitions.isDefined)
+ assert(d.requiredNumPartitions.get >= 1)
+ if (d != AllTuples) {
+ assert(d.requiredNumPartitions.get ==
s.stateInfo.get.numPartitions)
--- End diff --
can you also verify that this is equal to the number of partitions in the
metadata?
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
