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.
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
[SPARK-15580][SQL]Add ContinuousQueryInfo to make ContinuousQueryListener events serializable #13335
[SPARK-15580][SQL]Add ContinuousQueryInfo to make ContinuousQueryListener events serializable #13335
Changes from 2 commits
4f7ee7f
e6bb76c
621a867
3669df2
9b539cd
7df7e9c
e5e405f
e665c26
00ccfdf
be4b8c3
8dfc105
3230a12
1f00c60
File filter
Filter by extension
Conversations
Jump to
Diff view
Diff view
There are no files selected for viewing
Original file line number | Diff line number | Diff line change |
---|---|---|
|
@@ -18,7 +18,7 @@ | |
package org.apache.spark.sql.util | ||
|
||
import org.apache.spark.annotation.Experimental | ||
import org.apache.spark.sql.ContinuousQuery | ||
import org.apache.spark.sql._ | ||
import org.apache.spark.sql.util.ContinuousQueryListener._ | ||
|
||
/** | ||
|
@@ -65,11 +65,23 @@ object ContinuousQueryListener { | |
trait Event | ||
|
||
/** Event representing the start of a query */ | ||
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 */ | ||
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 */ | ||
class QueryTerminated private[sql](val query: ContinuousQuery) extends Event | ||
class QueryTerminated private[sql](val queryInfo: ContinuousQueryInfo) extends Event | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Should the data just be in the different events? Its a little odd that There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Removed |
||
} | ||
|
||
/** | ||
* :: Experimental :: | ||
* A class that contains information about [[ContinuousQuery]]. | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. A class used to report information about the progress of a [[ContinuousQuery]]. |
||
*/ | ||
@Experimental | ||
case class ContinuousQueryInfo( | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. We should scaladoc these parameters. There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. We also probably don't want this to be a There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Removed |
||
name: String, | ||
isActive: Boolean, | ||
sourceStatuses: Seq[SourceStatus], | ||
sinkStatus: SinkStatus, | ||
exception: Option[ContinuousQueryException]) | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Is There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Good catch. We should use There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Though the stacktrace is also very useful. That might be safe to serialize. There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Replaced it with String |
Original file line number | Diff line number | Diff line change |
---|---|---|
|
@@ -52,7 +52,7 @@ class ContinuousQueryListenerSuite extends StreamTest with SharedSQLContext with | |
Assert("Incorrect query status in onQueryStarted") { | ||
val status = listener.startStatus | ||
assert(status != null) | ||
assert(status.active == true) | ||
assert(status.isActive == true) | ||
assert(status.sourceStatuses.size === 1) | ||
assert(status.sourceStatuses(0).description.contains("Memory")) | ||
|
||
|
@@ -74,7 +74,7 @@ class ContinuousQueryListenerSuite extends StreamTest with SharedSQLContext with | |
assert(listener.progressStatuses.size === 1) | ||
val status = listener.progressStatuses.peek() | ||
assert(status != null) | ||
assert(status.active == true) | ||
assert(status.isActive == true) | ||
assert(status.sourceStatuses(0).offset === Some(LongOffset(0))) | ||
assert(status.sinkStatus.offset === CompositeOffset.fill(LongOffset(0))) | ||
|
||
|
@@ -88,7 +88,7 @@ class ContinuousQueryListenerSuite extends StreamTest with SharedSQLContext with | |
val status = listener.terminationStatus | ||
assert(status != null) | ||
|
||
assert(status.active === false) // must be inactive by the time onQueryTerm is called | ||
assert(status.isActive === false) // must be inactive by the time onQueryTerm is called | ||
assert(status.sourceStatuses(0).offset === Some(LongOffset(0))) | ||
assert(status.sinkStatus.offset === CompositeOffset.fill(LongOffset(0))) | ||
} | ||
|
@@ -165,9 +165,9 @@ class ContinuousQueryListenerSuite extends StreamTest with SharedSQLContext with | |
// to catch errors in the async listener events | ||
@volatile private var asyncTestWaiter = new Waiter | ||
|
||
@volatile var startStatus: QueryStatus = null | ||
@volatile var terminationStatus: QueryStatus = null | ||
val progressStatuses = new ConcurrentLinkedQueue[QueryStatus] | ||
@volatile var startStatus: ContinuousQueryInfo = null | ||
@volatile var terminationStatus: ContinuousQueryInfo = null | ||
val progressStatuses = new ConcurrentLinkedQueue[ContinuousQueryInfo] | ||
|
||
def reset(): Unit = { | ||
startStatus = null | ||
|
@@ -183,35 +183,23 @@ class ContinuousQueryListenerSuite extends StreamTest with SharedSQLContext with | |
|
||
override def onQueryStarted(queryStarted: QueryStarted): Unit = { | ||
asyncTestWaiter { | ||
startStatus = QueryStatus(queryStarted.query) | ||
startStatus = queryStarted.queryInfo | ||
} | ||
} | ||
|
||
override def onQueryProgress(queryProgress: QueryProgress): Unit = { | ||
asyncTestWaiter { | ||
assert(startStatus != null, "onQueryProgress called before onQueryStarted") | ||
progressStatuses.add(QueryStatus(queryProgress.query)) | ||
progressStatuses.add(queryProgress.queryInfo) | ||
} | ||
} | ||
|
||
override def onQueryTerminated(queryTerminated: QueryTerminated): Unit = { | ||
asyncTestWaiter { | ||
assert(startStatus != null, "onQueryTerminated called before onQueryStarted") | ||
terminationStatus = QueryStatus(queryTerminated.query) | ||
terminationStatus = queryTerminated.queryInfo | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. The exception is not tested. There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Added a test for the exception |
||
} | ||
asyncTestWaiter.dismiss() | ||
} | ||
} | ||
|
||
case class QueryStatus( | ||
active: Boolean, | ||
exception: Option[Exception], | ||
sourceStatuses: Array[SourceStatus], | ||
sinkStatus: SinkStatus) | ||
|
||
object QueryStatus { | ||
def apply(query: ContinuousQuery): QueryStatus = { | ||
QueryStatus(query.isActive, query.exception, query.sourceStatuses, query.sinkStatus) | ||
} | ||
} | ||
} |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Do we need to make this a deep copy, especially for
sourceStatuses
andsinkStatus
?There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
No. They are immutable.