[SPARK-15580][SQL]Add ContinuousQueryInfo to make ContinuousQueryListener events serializable #13335

zsxwing · 2016-05-26T20:37:27Z

What changes were proposed in this pull request?

This PR adds ContinuousQueryInfo to make ContinuousQueryListener events serializable in order to support writing events into the event log.

How was this patch tested?

Jenkins unit tests.

…izable

SparkQA · 2016-05-26T22:24:18Z

Test build #59415 has finished for PR 13335 at commit 4f7ee7f.

This patch passes all tests.
This patch merges cleanly.
This patch adds the following public classes (experimental):
- case class ContinuousQueryInfo(

jerryshao · 2016-05-27T05:23:35Z

sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/StreamExecution.scala

@@ -484,6 +484,15 @@ class StreamExecution(
     """.stripMargin
  }

+  private def toInfo: ContinuousQueryInfo = {


Do we need to make this a deep copy, especially for sourceStatuses and sinkStatus ?

No. They are immutable.

zsxwing · 2016-05-27T17:36:30Z

/cc @marmbrus @tdas

marmbrus · 2016-05-27T19:04:37Z

sql/core/src/main/scala/org/apache/spark/sql/util/ContinuousQueryListener.scala

+ * A class that contains information about [[ContinuousQuery]].
+ */
+@Experimental
+case class ContinuousQueryInfo(


We should scaladoc these parameters.

We also probably don't want this to be a case class. The issue is that if we add new parameters in the future, the unapply method will break binary compatibility. The constructor should also be private.

Removed case.

SparkQA · 2016-05-27T19:19:36Z

Test build #59502 has finished for PR 13335 at commit e6bb76c.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

tdas · 2016-05-27T19:51:49Z

sql/core/src/test/scala/org/apache/spark/sql/util/ContinuousQueryListenerSuite.scala

      }
    }

    override def onQueryTerminated(queryTerminated: QueryTerminated): Unit = {
      asyncTestWaiter {
        assert(startStatus != null, "onQueryTerminated called before onQueryStarted")
-        terminationStatus = QueryStatus(queryTerminated.query)
+        terminationStatus = queryTerminated.queryInfo


The exception is not tested.

Added a test for the exception

marmbrus · 2016-05-27T20:09:25Z

Can you include the json produced as a sanity check?

SparkQA · 2016-05-27T20:46:04Z

Test build #59514 has finished for PR 13335 at commit 621a867.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

zsxwing · 2016-05-30T05:38:28Z

I reverted c24b6b6 as there is a Option serialization bug in jackson-module-scala 2.7.3 (FasterXML/jackson-module-scala#240)

zsxwing · 2016-05-30T05:40:41Z

sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/CompositeOffset.scala

@@ -22,7 +22,7 @@ package org.apache.spark.sql.execution.streaming
 * [[Source]]s that are present in a streaming query. This is similar to simplified, single-instance
 * vector clock that must progress linearly forward.
 */
-case class CompositeOffset(offsets: Seq[Option[Offset]]) extends Offset {
+case class CompositeOffset(offsets: Array[Option[Offset]]) extends Offset {


Change the type to Array as jackson-module-scala doesn't support Seq[Option[X]]. Probably because of the type erasure.

SparkQA · 2016-05-30T07:28:28Z

Test build #59608 has finished for PR 13335 at commit d7c7b18.

This patch passes all tests.
This patch merges cleanly.
This patch adds the following public classes (experimental):
- * case class Person(name: String, age: Long)

zsxwing · 2016-05-31T20:02:06Z

I reverted c24b6b6 as there is a Option serialization bug in jackson-module-scala 2.7.3 (FasterXML/jackson-module-scala#240)

Removed this one and put it in #13417. Will update this one once #13417 gets merged

SparkQA · 2016-05-31T20:11:36Z

Test build #59674 has finished for PR 13335 at commit 3669df2.

This patch fails MiMa tests.
This patch merges cleanly.
This patch adds the following public classes (experimental):
- case class CompositeOffset(offsets: Array[Option[Offset]]) extends Offset
- case class WrappedContinuousQueryListenerEvent(

SparkQA · 2016-05-31T23:28:27Z

Test build #59685 has finished for PR 13335 at commit 9b539cd.

This patch passes all tests.
This patch merges cleanly.
This patch adds the following public classes (experimental):
- * case class Person(name: String, age: Long)

zsxwing · 2016-05-31T23:50:34Z

...ore/src/main/scala/org/apache/spark/sql/execution/streaming/ContinuousQueryListenerBus.scala

+ * Wrapper for StreamingListenerEvent as SparkListenerEvent so that it can be posted to Spark
+ * listener bus.
+ */
+case class WrappedContinuousQueryListenerEvent(


Move it out of ContinuousQueryListenerBus in order to eliminate the reference

I don't actually understand the structure of all the different even busses, but why do we need a different event type? Why don't all the events just extend SparkListenerEvent?

Good point. This pattern is from the Streaming events. But that's for binary compatibility. We don't need this pattern in the new APIs.

zsxwing · 2016-05-31T23:51:41Z

@marmbrus could you take another look, please?

zsxwing · 2016-06-01T18:42:15Z

@marmbrus addressed

SparkQA · 2016-06-01T19:31:05Z

Test build #59752 has finished for PR 13335 at commit 7df7e9c.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

SparkQA · 2016-06-01T20:16:47Z

Test build #59758 has finished for PR 13335 at commit 00ccfdf.

This patch fails Spark unit tests.
This patch merges cleanly.
This patch adds no public classes.

SparkQA · 2016-06-01T20:24:03Z

Test build #59760 has finished for PR 13335 at commit be4b8c3.

This patch fails Spark unit tests.
This patch merges cleanly.
This patch adds no public classes.

SparkQA · 2016-06-01T22:32:37Z

Test build #59771 has finished for PR 13335 at commit 8dfc105.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

tdas · 2016-06-06T20:25:39Z

sql/core/src/main/scala/org/apache/spark/sql/streaming/SourceStatus.scala

@@ -31,4 +31,4 @@ import org.apache.spark.sql.execution.streaming.{Offset, Source}
 @Experimental
 class SourceStatus private[sql] (
    val description: String,
-    val offset: Option[Offset])
+    val offset: Option[String])


offset --> offsetDesc, so that it allows us in future to expose actual offsets without the naming being weird.

zsxwing · 2016-06-06T20:43:20Z

@tdas addressed

tdas · 2016-06-06T20:47:28Z

sql/core/src/main/scala/org/apache/spark/sql/streaming/ContinuousQueryInfo.scala

+@Experimental
+class ContinuousQueryInfo private[sql](
+  val name: String,
+  val sourceStatuses: Seq[SourceStatus],


This should also probably be an Array. No need to be a Seq.

Offline discussion: this is okay. this is how it is for SparkListener, and avoid problems like the array being accidentally modified internally by the listener generator.

tdas · 2016-06-06T21:13:14Z

LGTM. Will merge after tests pass.

SparkQA · 2016-06-06T22:22:39Z

Test build #60069 has finished for PR 13335 at commit 3230a12.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

SparkQA · 2016-06-06T22:48:44Z

Test build #60072 has finished for PR 13335 at commit 1f00c60.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

tdas · 2016-06-07T23:39:03Z

Merging this to master and 2.0. Thanks @zsxwing

…tener events serializable ## What changes were proposed in this pull request? This PR adds ContinuousQueryInfo to make ContinuousQueryListener events serializable in order to support writing events into the event log. ## How was this patch tested? Jenkins unit tests. Author: Shixiong Zhu <shixiong@databricks.com> Closes #13335 from zsxwing/query-info. (cherry picked from commit 0cfd619) Signed-off-by: Tathagata Das <tathagata.das1565@gmail.com>

Add ContinuousQueryInfo to make ContinuousQueryListener events serial…

4f7ee7f

…izable

jerryshao reviewed May 27, 2016
View reviewed changes

Fix doc

e6bb76c

marmbrus reviewed May 27, 2016
View reviewed changes

Address

621a867

tdas reviewed May 27, 2016
View reviewed changes

zsxwing reviewed May 30, 2016
View reviewed changes

zsxwing mentioned this pull request May 31, 2016

[SPARK-11753][SQL][test-hadoop2.2] Make allowNonNumericNumbers option work #9759

Closed

Make ContinuousQueryListener events JSPN serializable and add tests

3669df2

Merge remote-tracking branch 'origin/master' into query-info

9b539cd

zsxwing reviewed May 31, 2016
View reviewed changes

Merge remote-tracking branch 'origin/master' into query-info

7df7e9c

zsxwing added 3 commits June 1, 2016 11:29

Turn offset to string

e665c26

Remove WrappedContinuousQueryListenerEvent

00ccfdf

Restore unnecessary changes

be4b8c3

Fix tests

8dfc105

tdas reviewed Jun 6, 2016
View reviewed changes

Address TD's comments

3230a12

tdas reviewed Jun 6, 2016
View reviewed changes

Address more

1f00c60

asfgit closed this in 0cfd619 Jun 7, 2016

zsxwing deleted the query-info branch June 8, 2016 18:23

[SPARK-15580][SQL]Add ContinuousQueryInfo to make ContinuousQueryListener events serializable #13335

[SPARK-15580][SQL]Add ContinuousQueryInfo to make ContinuousQueryListener events serializable #13335

Conversation

zsxwing commented May 26, 2016

What changes were proposed in this pull request?

How was this patch tested?

SparkQA commented May 26, 2016

Choose a reason for hiding this comment

Choose a reason for hiding this comment

zsxwing commented May 27, 2016

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

SparkQA commented May 27, 2016

Choose a reason for hiding this comment

Choose a reason for hiding this comment

marmbrus commented May 27, 2016

SparkQA commented May 27, 2016

zsxwing commented May 30, 2016

Choose a reason for hiding this comment

SparkQA commented May 30, 2016

zsxwing commented May 31, 2016

SparkQA commented May 31, 2016

SparkQA commented May 31, 2016

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

zsxwing commented May 31, 2016

zsxwing commented Jun 1, 2016

SparkQA commented Jun 1, 2016

SparkQA commented Jun 1, 2016

SparkQA commented Jun 1, 2016

SparkQA commented Jun 1, 2016

Choose a reason for hiding this comment

zsxwing commented Jun 6, 2016

Choose a reason for hiding this comment

Choose a reason for hiding this comment

tdas commented Jun 6, 2016

SparkQA commented Jun 6, 2016

SparkQA commented Jun 6, 2016

tdas commented Jun 7, 2016