eason-yuchen-liu commented on code in PR #59247:
URL: https://github.com/apache/spark/pull/59247#discussion_r4288928184
##########
sql/core/src/test/scala/org/apache/spark/sql/streaming/StreamingQueryStatusAndProgressSuite.scala:
##########
@@ -776,6 +898,23 @@ object StreamingQueryStatusAndProgressSuite {
observedMetrics = null
)
+ def progressWithTrigger(trigger: Trigger): StreamingQueryProgress = {
Review Comment:
why not create an additional new progress(7)
##########
sql/api/src/main/scala/org/apache/spark/sql/streaming/progress.scala:
##########
@@ -243,6 +264,10 @@ class StreamingQueryProgress private[spark] (
("timestamp" -> JString(timestamp)) ~
("batchId" -> JInt(batchId)) ~
("batchDuration" -> JInt(batchDuration)) ~
+ ("trigger" ->
Review Comment:
reuse `trigger` method?
##########
sql/api/src/main/scala/org/apache/spark/sql/streaming/progress.scala:
##########
@@ -295,6 +320,13 @@ private class ObjectToStringDeserializer extends
JsonDeserializer[String] {
}
}
+private class ObjectToTriggerDeserializer extends JsonDeserializer[Trigger] {
Review Comment:
Is `ObjectToTriggerDeserializer` really necessary? We are passing the
Trigger object so is deser used?
##########
sql/api/src/main/scala/org/apache/spark/sql/streaming/progress.scala:
##########
@@ -216,9 +220,26 @@ class StreamingQueryProgress private[spark] (
val sources: Array[SourceProgress],
val sink: SinkProgress,
@JsonDeserialize(contentAs = classOf[GenericRowWithSchema])
- val observedMetrics: ju.Map[String, Row])
+ val observedMetrics: ju.Map[String, Row],
+ @JsonProperty("trigger")
+ @JsonDeserialize(using = classOf[ObjectToTriggerDeserializer])
+ triggerConfig: Trigger = null)
Review Comment:
can we remove the default value so that callsites do not forget about it or
is there some compatibility concerns?
##########
sql/api/src/main/scala/org/apache/spark/sql/streaming/progress.scala:
##########
@@ -216,9 +220,26 @@ class StreamingQueryProgress private[spark] (
val sources: Array[SourceProgress],
val sink: SinkProgress,
@JsonDeserialize(contentAs = classOf[GenericRowWithSchema])
- val observedMetrics: ju.Map[String, Row])
+ val observedMetrics: ju.Map[String, Row],
+ @JsonProperty("trigger")
Review Comment:
why is these tags necessary?
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/runtime/ProgressReporter.scala:
##########
@@ -55,9 +55,12 @@ import org.apache.spark.util.{Clock, Utils}
class ProgressReporter(
private val sparkSession: SparkSession,
private val triggerClock: Clock,
+ private[runtime] val triggerConfig: Trigger,
val logicalPlan: () => LogicalPlan)
extends Logging {
+ Triggers.toJson(triggerConfig)
Review Comment:
why do we need this?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
