[GitHub] spark pull request #13335: [SPARK-15580][SQL]Add ContinuousQueryInfo to make...
Github user asfgit closed the pull request at: https://github.com/apache/spark/pull/13335 --- If your project is set up for it, you can reply to this email and have your reply appear on GitHub as well. If your project does not have this feature enabled and wishes so, or if the feature is enabled but not working, please contact infrastructure at infrastruct...@apache.org or file a JIRA ticket with INFRA. --- - To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] spark pull request #13335: [SPARK-15580][SQL]Add ContinuousQueryInfo to make...
Github user tdas commented on a diff in the pull request: https://github.com/apache/spark/pull/13335#discussion_r65970018 --- Diff: sql/core/src/main/scala/org/apache/spark/sql/streaming/ContinuousQueryInfo.scala --- @@ -0,0 +1,34 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + *http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.spark.sql.streaming + +import org.apache.spark.annotation.Experimental + +/** + * :: Experimental :: + * A class used to report information about the progress of a [[ContinuousQuery]]. + * + * @param name The [[ContinuousQuery]] name + * @param sourceStatuses The current statuses of the [[ContinuousQuery]]'s sources. + * @param sinkStatus The current status of the [[ContinuousQuery]]'s sink. + */ +@Experimental +class ContinuousQueryInfo private[sql]( + val name: String, + val sourceStatuses: Seq[SourceStatus], --- End diff -- 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. --- If your project is set up for it, you can reply to this email and have your reply appear on GitHub as well. If your project does not have this feature enabled and wishes so, or if the feature is enabled but not working, please contact infrastructure at infrastruct...@apache.org or file a JIRA ticket with INFRA. --- - To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] spark pull request #13335: [SPARK-15580][SQL]Add ContinuousQueryInfo to make...
Github user zsxwing commented on a diff in the pull request:
https://github.com/apache/spark/pull/13335#discussion_r65967892
--- Diff:
sql/core/src/main/scala/org/apache/spark/sql/streaming/ContinuousQueryListener.scala
---
@@ -70,26 +71,34 @@ abstract class ContinuousQueryListener {
object ContinuousQueryListener {
/**
- * Base type of [[ContinuousQueryListener]] events.
+ * Base type of [[ContinuousQueryListener]] events
* @since 2.0.0
*/
- trait Event
+ trait Event extends SparkListenerEvent
/**
- * Event representing the start of a query.
+ * Event representing the start of a query
* @since 2.0.0
*/
- class QueryStarted private[sql](val query: ContinuousQuery) extends Event
+ class QueryStarted private[sql](val queryInfo: ContinuousQueryInfo)
extends Event
/**
- * Event representing any progress updates in a query.
+ * Event representing any progress updates in a query
* @since 2.0.0
*/
- class QueryProgress private[sql](val query: ContinuousQuery) extends
Event
+ class QueryProgress private[sql](val queryInfo: ContinuousQueryInfo)
extends Event
/**
- * Event representing that termination of a query.
+ * Event representing that termination of a query
+ *
+ * @param queryInfo The query info.
+ * @param exception The exception message of the [[ContinuousQuery]] if
any. Otherwise, it
+ * will be `None`.
+ * @param stackTrace The stack trace of the exception if any.
* @since 2.0.0
*/
- class QueryTerminated private[sql](val query: ContinuousQuery) extends
Event
+ class QueryTerminated private[sql](
+ val queryInfo: ContinuousQueryInfo,
+ val exception: Option[String],
+ val stackTrace: Seq[StackTraceElement]) extends Event
--- End diff --
E.g., the user can change the content of an array.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] spark pull request #13335: [SPARK-15580][SQL]Add ContinuousQueryInfo to make...
Github user tdas commented on a diff in the pull request:
https://github.com/apache/spark/pull/13335#discussion_r65966655
--- Diff:
sql/core/src/main/scala/org/apache/spark/sql/streaming/ContinuousQueryListener.scala
---
@@ -70,26 +71,34 @@ abstract class ContinuousQueryListener {
object ContinuousQueryListener {
/**
- * Base type of [[ContinuousQueryListener]] events.
+ * Base type of [[ContinuousQueryListener]] events
* @since 2.0.0
*/
- trait Event
+ trait Event extends SparkListenerEvent
/**
- * Event representing the start of a query.
+ * Event representing the start of a query
* @since 2.0.0
*/
- class QueryStarted private[sql](val query: ContinuousQuery) extends Event
+ class QueryStarted private[sql](val queryInfo: ContinuousQueryInfo)
extends Event
/**
- * Event representing any progress updates in a query.
+ * Event representing any progress updates in a query
* @since 2.0.0
*/
- class QueryProgress private[sql](val query: ContinuousQuery) extends
Event
+ class QueryProgress private[sql](val queryInfo: ContinuousQueryInfo)
extends Event
/**
- * Event representing that termination of a query.
+ * Event representing that termination of a query
+ *
+ * @param queryInfo The query info.
+ * @param exception The exception message of the [[ContinuousQuery]] if
any. Otherwise, it
+ * will be `None`.
+ * @param stackTrace The stack trace of the exception if any.
* @since 2.0.0
*/
- class QueryTerminated private[sql](val query: ContinuousQuery) extends
Event
+ class QueryTerminated private[sql](
+ val queryInfo: ContinuousQueryInfo,
+ val exception: Option[String],
+ val stackTrace: Seq[StackTraceElement]) extends Event
--- End diff --
What mutable about this Array? And for Options, there is no workaround. And
Option is pretty simple to deal with in Java. Seq is more annoying, need to
look up Scala doc etc.?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] spark pull request #13335: [SPARK-15580][SQL]Add ContinuousQueryInfo to make...
Github user tdas commented on a diff in the pull request: https://github.com/apache/spark/pull/13335#discussion_r65966303 --- Diff: sql/core/src/main/scala/org/apache/spark/sql/streaming/ContinuousQueryInfo.scala --- @@ -0,0 +1,34 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + *http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.spark.sql.streaming + +import org.apache.spark.annotation.Experimental + +/** + * :: Experimental :: + * A class used to report information about the progress of a [[ContinuousQuery]]. + * + * @param name The [[ContinuousQuery]] name --- End diff -- Super Nit: missing period after name. --- If your project is set up for it, you can reply to this email and have your reply appear on GitHub as well. If your project does not have this feature enabled and wishes so, or if the feature is enabled but not working, please contact infrastructure at infrastruct...@apache.org or file a JIRA ticket with INFRA. --- - To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] spark pull request #13335: [SPARK-15580][SQL]Add ContinuousQueryInfo to make...
Github user tdas commented on a diff in the pull request: https://github.com/apache/spark/pull/13335#discussion_r65965736 --- Diff: sql/core/src/main/scala/org/apache/spark/sql/streaming/ContinuousQueryInfo.scala --- @@ -0,0 +1,34 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + *http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.spark.sql.streaming + +import org.apache.spark.annotation.Experimental + +/** + * :: Experimental :: + * A class used to report information about the progress of a [[ContinuousQuery]]. + * + * @param name The [[ContinuousQuery]] name + * @param sourceStatuses The current statuses of the [[ContinuousQuery]]'s sources. + * @param sinkStatus The current status of the [[ContinuousQuery]]'s sink. + */ +@Experimental +class ContinuousQueryInfo private[sql]( + val name: String, + val sourceStatuses: Seq[SourceStatus], --- End diff -- This should also probably be an Array. No need to be a Seq. --- If your project is set up for it, you can reply to this email and have your reply appear on GitHub as well. If your project does not have this feature enabled and wishes so, or if the feature is enabled but not working, please contact infrastructure at infrastruct...@apache.org or file a JIRA ticket with INFRA. --- - To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] spark pull request #13335: [SPARK-15580][SQL]Add ContinuousQueryInfo to make...
Github user zsxwing commented on a diff in the pull request:
https://github.com/apache/spark/pull/13335#discussion_r65964733
--- Diff:
sql/core/src/main/scala/org/apache/spark/sql/streaming/ContinuousQueryListener.scala
---
@@ -70,26 +71,34 @@ abstract class ContinuousQueryListener {
object ContinuousQueryListener {
/**
- * Base type of [[ContinuousQueryListener]] events.
+ * Base type of [[ContinuousQueryListener]] events
* @since 2.0.0
*/
- trait Event
+ trait Event extends SparkListenerEvent
/**
- * Event representing the start of a query.
+ * Event representing the start of a query
* @since 2.0.0
*/
- class QueryStarted private[sql](val query: ContinuousQuery) extends Event
+ class QueryStarted private[sql](val queryInfo: ContinuousQueryInfo)
extends Event
--- End diff --
Done
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] spark pull request #13335: [SPARK-15580][SQL]Add ContinuousQueryInfo to make...
Github user zsxwing commented on a diff in the pull request:
https://github.com/apache/spark/pull/13335#discussion_r65963896
--- Diff:
sql/core/src/main/scala/org/apache/spark/sql/streaming/ContinuousQueryListener.scala
---
@@ -70,26 +71,34 @@ abstract class ContinuousQueryListener {
object ContinuousQueryListener {
/**
- * Base type of [[ContinuousQueryListener]] events.
+ * Base type of [[ContinuousQueryListener]] events
* @since 2.0.0
*/
- trait Event
+ trait Event extends SparkListenerEvent
/**
- * Event representing the start of a query.
+ * Event representing the start of a query
* @since 2.0.0
*/
- class QueryStarted private[sql](val query: ContinuousQuery) extends Event
+ class QueryStarted private[sql](val queryInfo: ContinuousQueryInfo)
extends Event
/**
- * Event representing any progress updates in a query.
+ * Event representing any progress updates in a query
* @since 2.0.0
*/
- class QueryProgress private[sql](val query: ContinuousQuery) extends
Event
+ class QueryProgress private[sql](val queryInfo: ContinuousQueryInfo)
extends Event
/**
- * Event representing that termination of a query.
+ * Event representing that termination of a query
+ *
+ * @param queryInfo The query info.
+ * @param exception The exception message of the [[ContinuousQuery]] if
any. Otherwise, it
+ * will be `None`.
+ * @param stackTrace The stack trace of the exception if any.
* @since 2.0.0
*/
- class QueryTerminated private[sql](val query: ContinuousQuery) extends
Event
+ class QueryTerminated private[sql](
+ val queryInfo: ContinuousQueryInfo,
+ val exception: Option[String],
+ val stackTrace: Seq[StackTraceElement]) extends Event
--- End diff --
I don't want to expose a mutable Array to the user. Secondly, this one is
not Java friendly already since it uses `Option`.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] spark pull request #13335: [SPARK-15580][SQL]Add ContinuousQueryInfo to make...
Github user tdas commented on a diff in the pull request:
https://github.com/apache/spark/pull/13335#discussion_r65963272
--- Diff:
sql/core/src/test/scala/org/apache/spark/sql/streaming/ContinuousQueryListenerSuite.scala
---
@@ -141,6 +139,92 @@ class ContinuousQueryListenerSuite extends StreamTest
with BeforeAndAfter {
}
}
+ test("exception should be reported in QueryTerminated") {
+val listener = new QueryStatusCollector
+withListenerAdded(listener) {
+ val input = MemoryStream[Int]
+ testStream(input.toDS.map(_ / 0))(
+StartStream(),
+AddData(input, 1),
+ExpectFailure[SparkException](),
+Assert {
+ spark.sparkContext.listenerBus.waitUntilEmpty(1)
+ assert(listener.terminationStatus !== null)
+ assert(listener.terminationException.isDefined &&
--- End diff --
nit: make these conditions separate assert so its easy to debug if
something fails.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] spark pull request #13335: [SPARK-15580][SQL]Add ContinuousQueryInfo to make...
Github user tdas commented on a diff in the pull request:
https://github.com/apache/spark/pull/13335#discussion_r65963035
--- Diff:
sql/core/src/main/scala/org/apache/spark/sql/streaming/ContinuousQueryListener.scala
---
@@ -70,26 +71,34 @@ abstract class ContinuousQueryListener {
object ContinuousQueryListener {
/**
- * Base type of [[ContinuousQueryListener]] events.
+ * Base type of [[ContinuousQueryListener]] events
* @since 2.0.0
*/
- trait Event
+ trait Event extends SparkListenerEvent
/**
- * Event representing the start of a query.
+ * Event representing the start of a query
* @since 2.0.0
*/
- class QueryStarted private[sql](val query: ContinuousQuery) extends Event
+ class QueryStarted private[sql](val queryInfo: ContinuousQueryInfo)
extends Event
/**
- * Event representing any progress updates in a query.
+ * Event representing any progress updates in a query
* @since 2.0.0
*/
- class QueryProgress private[sql](val query: ContinuousQuery) extends
Event
+ class QueryProgress private[sql](val queryInfo: ContinuousQueryInfo)
extends Event
/**
- * Event representing that termination of a query.
+ * Event representing that termination of a query
+ *
+ * @param queryInfo The query info.
+ * @param exception The exception message of the [[ContinuousQuery]] if
any. Otherwise, it
+ * will be `None`.
+ * @param stackTrace The stack trace of the exception if any.
* @since 2.0.0
*/
- class QueryTerminated private[sql](val query: ContinuousQuery) extends
Event
+ class QueryTerminated private[sql](
+ val queryInfo: ContinuousQueryInfo,
+ val exception: Option[String],
+ val stackTrace: Seq[StackTraceElement]) extends Event
--- End diff --
Isnt it more Java friendly to make this Array instead of Seq? It does not
need to be Seq.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] spark pull request #13335: [SPARK-15580][SQL]Add ContinuousQueryInfo to make...
Github user tdas commented on a diff in the pull request:
https://github.com/apache/spark/pull/13335#discussion_r65962876
--- Diff:
sql/core/src/main/scala/org/apache/spark/sql/streaming/ContinuousQueryListener.scala
---
@@ -70,26 +71,34 @@ abstract class ContinuousQueryListener {
object ContinuousQueryListener {
/**
- * Base type of [[ContinuousQueryListener]] events.
+ * Base type of [[ContinuousQueryListener]] events
* @since 2.0.0
*/
- trait Event
+ trait Event extends SparkListenerEvent
/**
- * Event representing the start of a query.
+ * Event representing the start of a query
* @since 2.0.0
*/
- class QueryStarted private[sql](val query: ContinuousQuery) extends Event
+ class QueryStarted private[sql](val queryInfo: ContinuousQueryInfo)
extends Event
--- End diff --
Shouldnt these also be marked with @Experimental?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] spark pull request #13335: [SPARK-15580][SQL]Add ContinuousQueryInfo to make...
Github user tdas commented on a diff in the pull request:
https://github.com/apache/spark/pull/13335#discussion_r65962780
--- Diff:
sql/core/src/main/scala/org/apache/spark/sql/streaming/ContinuousQueryListener.scala
---
@@ -70,26 +71,34 @@ abstract class ContinuousQueryListener {
object ContinuousQueryListener {
/**
- * Base type of [[ContinuousQueryListener]] events.
+ * Base type of [[ContinuousQueryListener]] events
* @since 2.0.0
*/
- trait Event
+ trait Event extends SparkListenerEvent
/**
- * Event representing the start of a query.
+ * Event representing the start of a query
* @since 2.0.0
*/
- class QueryStarted private[sql](val query: ContinuousQuery) extends Event
+ class QueryStarted private[sql](val queryInfo: ContinuousQueryInfo)
extends Event
/**
- * Event representing any progress updates in a query.
+ * Event representing any progress updates in a query
* @since 2.0.0
*/
- class QueryProgress private[sql](val query: ContinuousQuery) extends
Event
+ class QueryProgress private[sql](val queryInfo: ContinuousQueryInfo)
extends Event
/**
- * Event representing that termination of a query.
+ * Event representing that termination of a query
+ *
+ * @param queryInfo The query info.
+ * @param exception The exception message of the [[ContinuousQuery]] if
any. Otherwise, it
+ * will be `None`.
+ * @param stackTrace The stack trace of the exception if any.
--- End diff --
nit: similar change as above. Also document that it will be empty if there
was no error.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] spark pull request #13335: [SPARK-15580][SQL]Add ContinuousQueryInfo to make...
Github user tdas commented on a diff in the pull request:
https://github.com/apache/spark/pull/13335#discussion_r65962584
--- Diff:
sql/core/src/main/scala/org/apache/spark/sql/streaming/ContinuousQueryListener.scala
---
@@ -70,26 +71,34 @@ abstract class ContinuousQueryListener {
object ContinuousQueryListener {
/**
- * Base type of [[ContinuousQueryListener]] events.
+ * Base type of [[ContinuousQueryListener]] events
* @since 2.0.0
*/
- trait Event
+ trait Event extends SparkListenerEvent
/**
- * Event representing the start of a query.
+ * Event representing the start of a query
* @since 2.0.0
*/
- class QueryStarted private[sql](val query: ContinuousQuery) extends Event
+ class QueryStarted private[sql](val queryInfo: ContinuousQueryInfo)
extends Event
/**
- * Event representing any progress updates in a query.
+ * Event representing any progress updates in a query
* @since 2.0.0
*/
- class QueryProgress private[sql](val query: ContinuousQuery) extends
Event
+ class QueryProgress private[sql](val queryInfo: ContinuousQueryInfo)
extends Event
/**
- * Event representing that termination of a query.
+ * Event representing that termination of a query
+ *
+ * @param queryInfo The query info.
+ * @param exception The exception message of the [[ContinuousQuery]] if
any. Otherwise, it
--- End diff --
nit: if any --> if the query was terminated with an exception.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] spark pull request #13335: [SPARK-15580][SQL]Add ContinuousQueryInfo to make...
Github user tdas commented on a diff in the pull request:
https://github.com/apache/spark/pull/13335#discussion_r65962502
--- Diff:
sql/core/src/main/scala/org/apache/spark/sql/streaming/ContinuousQueryListener.scala
---
@@ -70,26 +71,34 @@ abstract class ContinuousQueryListener {
object ContinuousQueryListener {
/**
- * Base type of [[ContinuousQueryListener]] events.
+ * Base type of [[ContinuousQueryListener]] events
* @since 2.0.0
*/
- trait Event
+ trait Event extends SparkListenerEvent
/**
- * Event representing the start of a query.
+ * Event representing the start of a query
* @since 2.0.0
*/
- class QueryStarted private[sql](val query: ContinuousQuery) extends Event
+ class QueryStarted private[sql](val queryInfo: ContinuousQueryInfo)
extends Event
/**
- * Event representing any progress updates in a query.
+ * Event representing any progress updates in a query
* @since 2.0.0
*/
- class QueryProgress private[sql](val query: ContinuousQuery) extends
Event
+ class QueryProgress private[sql](val queryInfo: ContinuousQueryInfo)
extends Event
/**
- * Event representing that termination of a query.
+ * Event representing that termination of a query
+ *
+ * @param queryInfo The query info.
--- End diff --
nit: "The query info" ---> "Information about the status of the query"
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] spark pull request #13335: [SPARK-15580][SQL]Add ContinuousQueryInfo to make...
Github user tdas commented on a diff in the pull request:
https://github.com/apache/spark/pull/13335#discussion_r65961978
--- Diff:
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])
--- End diff --
offset --> offsetDesc, so that it allows us in future to expose actual
offsets without the naming being weird.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] spark pull request #13335: [SPARK-15580][SQL]Add ContinuousQueryInfo to make...
Github user zsxwing commented on a diff in the pull request:
https://github.com/apache/spark/pull/13335#discussion_r65417137
--- Diff:
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/ContinuousQueryListenerBus.scala
---
@@ -69,14 +69,15 @@ class ContinuousQueryListenerBus(sparkListenerBus:
LiveListenerBus)
}
}
- /**
- * Wrapper for StreamingListenerEvent as SparkListenerEvent so that it
can be posted to Spark
- * listener bus.
- */
- private case class WrappedContinuousQueryListenerEvent(
- streamingListenerEvent: ContinuousQueryListener.Event) extends
SparkListenerEvent {
+}
-// Do not log streaming events in event log as history server does not
support these events.
-protected[spark] override def logEvent: Boolean = false
- }
+/**
+ * Wrapper for StreamingListenerEvent as SparkListenerEvent so that it can
be posted to Spark
+ * listener bus.
+ */
+case class WrappedContinuousQueryListenerEvent(
--- End diff --
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.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] spark pull request #13335: [SPARK-15580][SQL]Add ContinuousQueryInfo to make...
Github user vanzin commented on a diff in the pull request: https://github.com/apache/spark/pull/13335#discussion_r65412480 --- Diff: sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/Offset.scala --- @@ -17,11 +17,14 @@ package org.apache.spark.sql.execution.streaming +import com.fasterxml.jackson.annotation.JsonTypeInfo + /** * An offset is a monotonically increasing metric used to track progress in the computation of a * stream. An [[Offset]] must be comparable, and the result of `compareTo` must be consistent * with `equals` and `hashcode`. */ +@JsonTypeInfo(use = JsonTypeInfo.Id.CLASS, include = JsonTypeInfo.As.PROPERTY, property = "@class") --- End diff -- I've had way more problems with manual serialization than with jackson, so well, YMMV I guess. --- If your project is set up for it, you can reply to this email and have your reply appear on GitHub as well. If your project does not have this feature enabled and wishes so, or if the feature is enabled but not working, please contact infrastructure at infrastruct...@apache.org or file a JIRA ticket with INFRA. --- - To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] spark pull request #13335: [SPARK-15580][SQL]Add ContinuousQueryInfo to make...
Github user marmbrus commented on a diff in the pull request: https://github.com/apache/spark/pull/13335#discussion_r65411829 --- Diff: sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/Offset.scala --- @@ -17,11 +17,14 @@ package org.apache.spark.sql.execution.streaming +import com.fasterxml.jackson.annotation.JsonTypeInfo + /** * An offset is a monotonically increasing metric used to track progress in the computation of a * stream. An [[Offset]] must be comparable, and the result of `compareTo` must be consistent * with `equals` and `hashcode`. */ +@JsonTypeInfo(use = JsonTypeInfo.Id.CLASS, include = JsonTypeInfo.As.PROPERTY, property = "@class") --- End diff -- In my experience (with jackson specifically and people who have read the docs) you end up with surprises down the road due to magic. @zsxwing In this particular case, maybe we should remove the offsets from the status messages, or turn the to strings or something. These are going to be implemented by specific sinks, so I'm not sure we want to expose the details of arbitrary classes to the listener bus. --- If your project is set up for it, you can reply to this email and have your reply appear on GitHub as well. If your project does not have this feature enabled and wishes so, or if the feature is enabled but not working, please contact infrastructure at infrastruct...@apache.org or file a JIRA ticket with INFRA. --- - To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] spark pull request #13335: [SPARK-15580][SQL]Add ContinuousQueryInfo to make...
Github user zsxwing commented on a diff in the pull request: https://github.com/apache/spark/pull/13335#discussion_r65411387 --- Diff: sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/Offset.scala --- @@ -17,11 +17,14 @@ package org.apache.spark.sql.execution.streaming +import com.fasterxml.jackson.annotation.JsonTypeInfo + /** * An offset is a monotonically increasing metric used to track progress in the computation of a * stream. An [[Offset]] must be comparable, and the result of `compareTo` must be consistent * with `equals` and `hashcode`. */ +@JsonTypeInfo(use = JsonTypeInfo.Id.CLASS, include = JsonTypeInfo.As.PROPERTY, property = "@class") --- End diff -- > What does this mean? What JSON does this produce? This is for polymorphic serialization. It will write the class name to the field `@class` (set by property = "@class"). When deserializing an object, it needs the class name to know the concrete class since the reflection APIs will only provide the interface name. --- If your project is set up for it, you can reply to this email and have your reply appear on GitHub as well. If your project does not have this feature enabled and wishes so, or if the feature is enabled but not working, please contact infrastructure at infrastruct...@apache.org or file a JIRA ticket with INFRA. --- - To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] spark pull request #13335: [SPARK-15580][SQL]Add ContinuousQueryInfo to make...
Github user vanzin commented on a diff in the pull request: https://github.com/apache/spark/pull/13335#discussion_r65411215 --- Diff: sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/Offset.scala --- @@ -17,11 +17,14 @@ package org.apache.spark.sql.execution.streaming +import com.fasterxml.jackson.annotation.JsonTypeInfo + /** * An offset is a monotonically increasing metric used to track progress in the computation of a * stream. An [[Offset]] must be comparable, and the result of `compareTo` must be consistent * with `equals` and `hashcode`. */ +@JsonTypeInfo(use = JsonTypeInfo.Id.CLASS, include = JsonTypeInfo.As.PROPERTY, property = "@class") --- End diff -- That's why Jackson has javadocs. :-) You read it once instead of having to understand how every single type ends up in json by having to read the specific manual serialization code for that type... Anyway, to avoid copy & paste of this kind of code everywhere, it would be probably better to have a base trait for "traits that will be serialized in events and might have multiple implementations", and have that trait have this annotation (like `SparkListenerEvent` does). --- If your project is set up for it, you can reply to this email and have your reply appear on GitHub as well. If your project does not have this feature enabled and wishes so, or if the feature is enabled but not working, please contact infrastructure at infrastruct...@apache.org or file a JIRA ticket with INFRA. --- - To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] spark pull request #13335: [SPARK-15580][SQL]Add ContinuousQueryInfo to make...
Github user zsxwing commented on a diff in the pull request:
https://github.com/apache/spark/pull/13335#discussion_r65410405
--- Diff:
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/ContinuousQueryListenerBus.scala
---
@@ -71,15 +70,15 @@ class ContinuousQueryListenerBus(sparkListenerBus:
LiveListenerBus)
}
}
- /**
- * Wrapper for StreamingListenerEvent as SparkListenerEvent so that it
can be posted to Spark
- * listener bus.
- */
- private case class WrappedContinuousQueryListenerEvent(
- streamingListenerEvent: ContinuousQueryListener.Event)
-extends SparkListenerEvent {
+}
-// Do not log streaming events in event log as history server does not
support these events.
-protected[spark] override def logEvent: Boolean = false
- }
+/**
+ * Wrapper for StreamingListenerEvent as SparkListenerEvent so that it can
be posted to Spark
+ * listener bus.
+ */
+case class WrappedContinuousQueryListenerEvent(
+ streamingListenerEvent: ContinuousQueryListener.Event) extends
SparkListenerEvent {
+
+ // Do not log streaming events in event log as history server does not
support these events.
--- End diff --
It's robust. Just don't log unnecessary events since we don't have any UI
for them.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] spark pull request #13335: [SPARK-15580][SQL]Add ContinuousQueryInfo to make...
Github user marmbrus commented on a diff in the pull request: https://github.com/apache/spark/pull/13335#discussion_r65410185 --- Diff: sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/Offset.scala --- @@ -17,11 +17,14 @@ package org.apache.spark.sql.execution.streaming +import com.fasterxml.jackson.annotation.JsonTypeInfo + /** * An offset is a monotonically increasing metric used to track progress in the computation of a * stream. An [[Offset]] must be comparable, and the result of `compareTo` must be consistent * with `equals` and `hashcode`. */ +@JsonTypeInfo(use = JsonTypeInfo.Id.CLASS, include = JsonTypeInfo.As.PROPERTY, property = "@class") --- End diff -- What does this mean? What JSON does this produce? @vanzin, I find this very hard to reason about relative to constructing the JSON by hand. --- If your project is set up for it, you can reply to this email and have your reply appear on GitHub as well. If your project does not have this feature enabled and wishes so, or if the feature is enabled but not working, please contact infrastructure at infrastruct...@apache.org or file a JIRA ticket with INFRA. --- - To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] spark pull request #13335: [SPARK-15580][SQL]Add ContinuousQueryInfo to make...
Github user marmbrus commented on a diff in the pull request:
https://github.com/apache/spark/pull/13335#discussion_r65409940
--- Diff:
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/ContinuousQueryListenerBus.scala
---
@@ -71,15 +70,15 @@ class ContinuousQueryListenerBus(sparkListenerBus:
LiveListenerBus)
}
}
- /**
- * Wrapper for StreamingListenerEvent as SparkListenerEvent so that it
can be posted to Spark
- * listener bus.
- */
- private case class WrappedContinuousQueryListenerEvent(
- streamingListenerEvent: ContinuousQueryListener.Event)
-extends SparkListenerEvent {
+}
-// Do not log streaming events in event log as history server does not
support these events.
-protected[spark] override def logEvent: Boolean = false
- }
+/**
+ * Wrapper for StreamingListenerEvent as SparkListenerEvent so that it can
be posted to Spark
+ * listener bus.
+ */
+case class WrappedContinuousQueryListenerEvent(
+ streamingListenerEvent: ContinuousQueryListener.Event) extends
SparkListenerEvent {
+
+ // Do not log streaming events in event log as history server does not
support these events.
--- End diff --
The history server isn't robust to unknown messages? Seems bad that we
always drop them. What happens when we do add support?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] spark pull request #13335: [SPARK-15580][SQL]Add ContinuousQueryInfo to make...
Github user marmbrus commented on a diff in the pull request:
https://github.com/apache/spark/pull/13335#discussion_r65409815
--- Diff:
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/ContinuousQueryListenerBus.scala
---
@@ -69,14 +69,15 @@ class ContinuousQueryListenerBus(sparkListenerBus:
LiveListenerBus)
}
}
- /**
- * Wrapper for StreamingListenerEvent as SparkListenerEvent so that it
can be posted to Spark
- * listener bus.
- */
- private case class WrappedContinuousQueryListenerEvent(
- streamingListenerEvent: ContinuousQueryListener.Event) extends
SparkListenerEvent {
+}
-// Do not log streaming events in event log as history server does not
support these events.
-protected[spark] override def logEvent: Boolean = false
- }
+/**
+ * Wrapper for StreamingListenerEvent as SparkListenerEvent so that it can
be posted to Spark
+ * listener bus.
+ */
+case class WrappedContinuousQueryListenerEvent(
--- End diff --
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`?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] spark pull request #13335: [SPARK-15580][SQL]Add ContinuousQueryInfo to make...
Github user marmbrus commented on a diff in the pull request:
https://github.com/apache/spark/pull/13335#discussion_r65409474
--- Diff:
sql/core/src/main/scala/org/apache/spark/sql/streaming/ContinuousQueryListener.scala
---
@@ -69,27 +72,27 @@ abstract class ContinuousQueryListener {
@Experimental
object ContinuousQueryListener {
- /**
- * Base type of [[ContinuousQueryListener]] events.
- * @since 2.0.0
- */
+ /** Base type of [[ContinuousQueryListener]] events */
+ @JsonTypeInfo(
+use = JsonTypeInfo.Id.CLASS,
+include = JsonTypeInfo.As.PROPERTY,
+property = "@class")
trait Event
- /**
- * Event representing the start of a query.
- * @since 2.0.0
- */
- class QueryStarted private[sql](val query: ContinuousQuery) extends Event
+ /** Event representing the start of a query */
+ class QueryStarted private[sql](val queryInfo: ContinuousQueryInfo)
extends Event
- /**
- * Event representing any progress updates in a query.
- * @since 2.0.0
- */
- class QueryProgress private[sql](val query: ContinuousQuery) extends
Event
+ /** Event representing any progress updates in a query */
+ class QueryProgress private[sql](val queryInfo: ContinuousQueryInfo)
extends Event
/**
- * Event representing that termination of a query.
- * @since 2.0.0
+ * Event representing that termination of a query
+ *
+ * @param queryInfo
+ * @param exception The exception information of the [[ContinuousQuery]]
if any. Otherwise, it
+ * will be `None`.
*/
- class QueryTerminated private[sql](val query: ContinuousQuery) extends
Event
+ class QueryTerminated private[sql](
+ val queryInfo: ContinuousQueryInfo,
+ val exception: Option[String]) extends Event
--- End diff --
I don't think a string is enough here. I think you need the stack trace
also for it to be useful.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
