You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@spark.apache.org by "alyssa-db (via GitHub)" <gi...@apache.org> on 2026/10/06 00:56:04 UTC
[PR] [SPARK-59986][SS] Expose trigger metadata in streaming query progress [spark]
alyssa-db opened a new pull request, #59247:
URL: https://github.com/apache/spark/pull/59247
### What changes were proposed in this pull request?
Add structured metadata for the configured trigger to `StreamingQueryProgress.json` and `prettyJson`, and preserve it through `QueryProgressEvent` reconstruction, event-log serialization, and the status store.
An internal serializer handles the built-in trigger types. The existing `StreamingQueryProgress` constructor and public `Trigger` API are preserved. Legacy records and unknown custom trigger types omit the field. The status-store protobuf adds optional string field 13 to retain the JSON object.
### Why are the changes needed?
Monitoring consumers need to distinguish trigger modes and inspect their configured intervals directly from streaming progress records.
[SPARK-59986](https://issues.apache.org/jira/browse/SPARK-59986)
### Does this PR introduce _any_ user-facing change?
Yes. Progress JSON gains an optional top-level `trigger` object:
| Trigger | Example |
|---|---|
| ProcessingTime | `"trigger":{"type":"ProcessingTime","intervalMs":1000}` |
| Continuous | `"trigger":{"type":"Continuous","intervalMs":2000}` |
| RealTime | `"trigger":{"type":"RealTime","batchDurationMs":3000}` |
| AvailableNow | `"trigger":{"type":"AvailableNow"}` |
| OneTime | `"trigger":{"type":"OneTime"}` |
Existing progress fields are preserved. Continuous execution currently records epochs without publishing progress records; its metadata is covered through progress construction and serialization. This PR preserves that publication behavior.
### How was this patch tested?
Added 34 cases covering all built-in triggers, zero and default intervals, an interval exceeding 32-bit range, compact/pretty/Jackson JSON, event reconstruction, status-store persistence, legacy absence, custom trigger omission, live micro-batch and RealTime queries, and idle progress.
Verified with Java 17 on upstream master `5926a10389e8eace9f655dc138f4bb30254f7bed`:
```sh
build/sbt -Dsbt.override.build.repos=true -batch \
'sql/testOnly *StreamingQueryStatusAndProgressSuite *KVStoreProtobufSerializerSuite *ContinuousQueryStatusAndProgressSuite' \
'sql/testOnly *StreamingQueryListenerSuite -- -z serialization -z SPARK-59986' \
'sql/testOnly *StreamRealTimeModeSuite -- -z SPARK-59986'
```
Results: **43/43**, **12/12**, and **1/1** passed, respectively; **56 tests passed, 0 failed**.
```sh
build/sbt -Dsbt.override.build.repos=true -batch \
'sql-api/scalaStyleOnCompile' \
'sql/scalaStyleOnCompile' \
'sql/scalaStyleOnTest'
```
All three checks reported **0 errors and 0 warnings**.
```sh
build/mvn -B -pl sql/api scalafmt:format \
-Dscalafmt.skip=false -Dscalafmt.validateOnly=true -Dscalafmt.changedOnly=false
```
SQL API formatting validation passed. Dependency resolution used local Maven mirror settings. Protobuf schema compilation and `git diff --check` also passed.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: OpenAI Codex CLI 0.154.0
This contribution is submitted under the project's Apache License 2.0.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-59986][SS] Expose trigger metadata in streaming query progress [spark]
Posted by "eason-yuchen-liu (via GitHub)" <gi...@apache.org>.
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-59986][SS] Expose trigger metadata in streaming query progress [spark]
Posted by "eason-yuchen-liu (via GitHub)" <gi...@apache.org>.
eason-yuchen-liu commented on code in PR #59247:
URL: https://github.com/apache/spark/pull/59247#discussion_r4274451042
##########
sql/api/src/main/scala/org/apache/spark/sql/streaming/progress.scala:
##########
@@ -216,9 +219,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[ObjectToStringDeserializer])
+ triggerJson: String = null)
Review Comment:
You can pass in the original Trigger object in the constructor, and toJson at L266.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-59986][SS] Expose trigger metadata in streaming query progress [spark]
Posted by "alyssa-db (via GitHub)" <gi...@apache.org>.
alyssa-db commented on code in PR #59247:
URL: https://github.com/apache/spark/pull/59247#discussion_r4289496736
##########
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:
yes, we can remove the default value, just pass null.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-59986][SS] Expose trigger metadata in streaming query progress [spark]
Posted by "alyssa-db (via GitHub)" <gi...@apache.org>.
alyssa-db commented on PR #59247:
URL: https://github.com/apache/spark/pull/59247#issuecomment-6352568763
@HeartSaVioR Could you kindly take a look?
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-59986][SS] Expose trigger metadata in streaming query progress [spark]
Posted by "eason-yuchen-liu (via GitHub)" <gi...@apache.org>.
eason-yuchen-liu commented on PR #59247:
URL: https://github.com/apache/spark/pull/59247#issuecomment-6279635112
Left a comment: https://github.com/apache/spark/pull/59247#discussion_r4319896906.
The rest looks good to me.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-59986][SS] Expose trigger metadata in streaming query progress [spark]
Posted by "eason-yuchen-liu (via GitHub)" <gi...@apache.org>.
eason-yuchen-liu commented on code in PR #59247:
URL: https://github.com/apache/spark/pull/59247#discussion_r4319896906
##########
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:
Throwing in this method is very rare. We should not be paying runtime of each query to validate this case. IMO we should remove this/
##########
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:
Throwing in this method is very rare. We should not be paying runtime of each query to validate this case. IMO we should remove 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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-59986][SS] Expose trigger metadata in streaming query progress [spark]
Posted by "HyukjinKwon (via GitHub)" <gi...@apache.org>.
HyukjinKwon commented on code in PR #59247:
URL: https://github.com/apache/spark/pull/59247#discussion_r4265171782
##########
sql/api/src/main/scala/org/apache/spark/sql/execution/streaming/Triggers.scala:
##########
@@ -27,10 +27,23 @@ import org.apache.spark.SparkIllegalArgumentException
import org.apache.spark.sql.catalyst.util.DateTimeConstants.MICROS_PER_DAY
import org.apache.spark.sql.catalyst.util.SparkDateTimeUtils.microsToMillis
import org.apache.spark.sql.catalyst.util.SparkIntervalUtils
+import org.apache.spark.sql.errors.ExecutionErrors
import org.apache.spark.sql.streaming.Trigger
import org.apache.spark.unsafe.types.UTF8String
-private object Triggers {
+private[spark] object Triggers {
+ def toJson(trigger: Trigger): String = trigger match {
+ case ProcessingTimeTrigger(intervalMs) =>
+ s"""{"type":"ProcessingTime","intervalMs":$intervalMs}"""
+ case ContinuousTrigger(intervalMs) =>
+ s"""{"type":"Continuous","intervalMs":$intervalMs}"""
+ case RealTimeTrigger(batchDurationMs) =>
+ s"""{"type":"RealTime","batchDurationMs":$batchDurationMs}"""
+ case AvailableNowTrigger => """{"type":"AvailableNow"}"""
+ case OneTimeTrigger => """{"type":"OneTime"}"""
+ case _ => throw ExecutionErrors.unreachableError()
Review Comment:
**Nit (P3):** `Trigger` is a public, non-final class, so this branch is reachable. `df.writeStream.trigger(new Trigger {}).start()`, the input the new test builds, now fails inside `new MicroBatchExecution(...)`, because `ProgressReporter` calls `toJson` from its constructor. The user sees `[_LEGACY_ERROR_TEMP_2028] This line should be unreachable.`, which names neither the trigger nor the problem and reads like a Spark bug. Before this PR, the same query failed with `Unknown type of trigger: <trigger>` from `MicroBatchExecution.getTrigger`.
Failing fast here is fine; the error just needs to describe the input. For example:
```scala
case other => throw new IllegalStateException(s"Unknown type of trigger: $other")
```
An error condition that takes the trigger class as a parameter would also work. Only Scala/Java callers that subclass `Trigger` can reach this; PySpark and Spark Connect build only the built-in triggers.
**Verification:**
- **Regression:** Serializing a custom Trigger subclass, and starting a streaming query with one, both fail with an error whose message identifies the unsupported trigger rather than _LEGACY_ERROR_TEMP_2028.
- **Behavior:** Trigger JSON for every built-in trigger is unchanged.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-59986][SS] Expose trigger metadata in streaming query progress [spark]
Posted by "alyssa-db (via GitHub)" <gi...@apache.org>.
alyssa-db commented on code in PR #59247:
URL: https://github.com/apache/spark/pull/59247#discussion_r4289478678
##########
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:
For deserialization. @JsonProperty maps the public JSON field trigger to triggerConfig. The custom deserializer @JsonDeserialize() converts that JSON object into a Trigger, e.g Trigger.ProcessingTime(1000).
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-59986][SS] Expose trigger metadata in streaming query progress [spark]
Posted by "HyukjinKwon (via GitHub)" <gi...@apache.org>.
HyukjinKwon commented on code in PR #59247:
URL: https://github.com/apache/spark/pull/59247#discussion_r4277360367
##########
sql/core/src/test/scala/org/apache/spark/sql/streaming/StreamingQueryStatusAndProgressSuite.scala:
##########
@@ -43,6 +46,111 @@ import org.apache.spark.sql.types.StructType
import org.apache.spark.util.ArrayImplicits._
class StreamingQueryStatusAndProgressSuite extends StreamTest with Eventually with Matchers {
+ gridTest("SPARK-59986: trigger metadata")(Seq[(Trigger, String)](
+ Trigger.ProcessingTime(0) -> """{"type":"ProcessingTime","intervalMs":0}""",
+ Trigger.ProcessingTime(3000000000L) ->
+ """{"type":"ProcessingTime","intervalMs":3000000000}""",
+ Trigger.Continuous(2000) -> """{"type":"Continuous","intervalMs":2000}""",
+ Trigger.RealTime(3000) -> """{"type":"RealTime","batchDurationMs":3000}""",
+ Trigger.AvailableNow() -> """{"type":"AvailableNow"}""",
+ Trigger.Once() -> """{"type":"Once"}""")) {
+ case (trigger, expected) =>
+ assert(Triggers.toJson(trigger) === expected)
+ assert(Triggers.fromJson(expected) === trigger)
+ }
+
+ test("SPARK-59986: unknown trigger metadata identifies the unsupported trigger") {
+ val trigger = new Trigger {}
+ val exception = intercept[IllegalStateException] {
+ Triggers.toJson(trigger)
+ }
+ assert(exception.getMessage === s"Unknown type of trigger: $trigger")
+ }
+
+ test("SPARK-59986: starting a query with an unknown trigger identifies the input") {
+ import testImplicits._
+
+ val input = MemoryStream[Int]
+ val trigger = new Trigger {}
+ val exception = intercept[IllegalStateException] {
Review Comment:
**Non-blocking (P2):** This intercept no longer sees an exception. Since 96b71ac, `ProgressReporter` keeps `triggerConfig` instead of eagerly calling `Triggers.toJson` in its constructor, so nothing on the `start()` path validates the trigger. The only check left is `MicroBatchExecution.getTrigger`, reached through the `logicalPlan` lazy val on the query thread after `startLatch` has let `start()` return. `stop()` only joins the thread and doesn't rethrow. So `start().stop()` completes normally, and the test fails with "Expected exception java.lang.IllegalStateException to be thrown, but no exception was thrown".
The query still terminates asynchronously with `Unknown type of trigger: ...`, the same as on master. But this test and the PR description's "fails query construction" claim no longer match the code.
To keep the documented behavior, restore a construction-time check. For example, call `Triggers.toJson(triggerConfig)` when `ProgressReporter` is constructed, which happens synchronously in `StreamExecution`'s constructor inside `createQuery`. That makes `start()` throw as described.
**Verification:**
- **Regression:** Starting a streaming query with a custom Trigger subclass fails synchronously with an IllegalStateException whose message identifies the trigger.
- **Behavior:** Micro-batch, real-time, and continuous progress for built-in triggers still reports the configured trigger metadata.
##########
sql/api/src/main/scala/org/apache/spark/sql/execution/streaming/Triggers.scala:
##########
@@ -30,7 +31,33 @@ import org.apache.spark.sql.catalyst.util.SparkIntervalUtils
import org.apache.spark.sql.streaming.Trigger
import org.apache.spark.unsafe.types.UTF8String
-private object Triggers {
+private[spark] object Triggers {
+ private implicit val formats: DefaultFormats = DefaultFormats
+
+ def toJson(trigger: Trigger): String = trigger match {
+ case ProcessingTimeTrigger(intervalMs) =>
+ s"""{"type":"ProcessingTime","intervalMs":$intervalMs}"""
+ case ContinuousTrigger(intervalMs) =>
+ s"""{"type":"Continuous","intervalMs":$intervalMs}"""
+ case RealTimeTrigger(batchDurationMs) =>
+ s"""{"type":"RealTime","batchDurationMs":$batchDurationMs}"""
+ case AvailableNowTrigger => """{"type":"AvailableNow"}"""
+ case OneTimeTrigger => """{"type":"Once"}"""
+ case other => throw new IllegalStateException(s"Unknown type of trigger: $other")
+ }
+
+ def fromJson(json: String): Trigger = {
+ val value = parse(json)
+ (value \ "type").extract[String] match {
+ case "ProcessingTime" => ProcessingTimeTrigger((value \ "intervalMs").extract[Long])
+ case "Continuous" => ContinuousTrigger((value \ "intervalMs").extract[Long])
+ case "RealTime" => RealTimeTrigger((value \ "batchDurationMs").extract[Long])
+ case "AvailableNow" => AvailableNowTrigger
+ case "Once" => OneTimeTrigger
+ case other => throw new IllegalStateException(s"Unknown type of trigger: $other")
Review Comment:
**Non-blocking (P2):** This decoder is the one part of progress decoding that isn't forward compatible. `fromJson` throws for any `type` outside the five built-ins, and the `extract[Long]` calls throw when an interval key is missing. Both JVM decoders call it with no fallback: `ObjectToTriggerDeserializer`, used by `StreamingQueryProgress.fromJson` and `QueryProgressEvent.fromJson`, and `StreamingQueryProgressSerializer.deserialize` for the status store.
Suppose a Scala Spark Connect client with this change talks to a later server that adds a trigger kind (`RealTime` shows the set does grow) or changes a key. `RemoteStreamingQuery.lastProgress` and `recentProgress` then throw. In `StreamingQueryListenerBus.queryEventHandler`, the same exception reaches the loop-level catch, which removes every client-side listener. An older history server replaying a newer event log also stops at that event, because `ReplayListenerBus` only skips `ClassNotFoundException` and `UnrecognizedPropertyException`.
Everything around this degrades instead: both mappers set `FAIL_ON_UNKNOWN_PROPERTIES=false`, offsets go through `ObjectToStringDeserializer`, and the listener bus only warns on unknown events. The PySpark `fromJson` in this PR already keeps the dict as-is. Classic in-process `lastProgress` is unaffected because it never decodes.
Released clients can't be patched, so the tolerance needs to ship with the field's first release. I'd make this helper total: return `null`, the documented "metadata unavailable" value, for an unknown `type` or a missing or malformed interval key. Then use it from both deserializers.
**Verification:**
- **Regression:** Decoding progress JSON with an unknown trigger type, and with a missing interval key, returns a progress record whose trigger is null instead of throwing, on both the Jackson and status-store protobuf paths.
- **Compatibility:** Every built-in trigger type still serializes to the same JSON and decodes back to the same Trigger value.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-59986][SS] Expose trigger metadata in streaming query progress [spark]
Posted by "HyukjinKwon (via GitHub)" <gi...@apache.org>.
HyukjinKwon commented on code in PR #59247:
URL: https://github.com/apache/spark/pull/59247#discussion_r4266869432
##########
sql/api/src/main/scala/org/apache/spark/sql/streaming/progress.scala:
##########
@@ -216,9 +219,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[ObjectToStringDeserializer])
+ triggerJson: String = null)
extends Serializable {
+ /**
+ * JSON metadata for the configured trigger, or null when metadata is unavailable.
+ *
+ * The {{{type}}} field is {{{ProcessingTime}}}, {{{Continuous}}}, {{{RealTime}}},
Review Comment:
**Nit (P3):** Scaladoc treats `{{{ ... }}}` as a block-level code delimiter, not inline code. Mid-line the braces render literally (`The {{{type}}} field is ...`), and lines 232 and 233 start with `{{{`, so each of those opens a separate code block and splits the sentence. In the generated Scala and Java API docs, the only JVM-side description of the trigger vocabulary comes out garbled. The rest of this class doc already uses backticks for inline code (for example `StreamingQuery.id()`).
Switching to backticks keeps the wording unchanged:
```scala
* The `type` field is `ProcessingTime`, `Continuous`, `RealTime`,
* `AvailableNow`, or `Once`. `ProcessingTime` and `Continuous` include
* `intervalMs`; `RealTime` includes `batchDurationMs`. Both intervals are in
* milliseconds. `AvailableNow` and `Once` contain only the `type` field.
```
**Verification:**
- **Inspection:** The trigger accessor Scaladoc uses backtick inline code for every identifier and contains no triple-brace delimiters.
- **Compile:** The sql/api module still compiles and passes its Scala style checks after the documentation edit.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-59986][SS] Expose trigger metadata in streaming query progress [spark]
Posted by "alyssa-db (via GitHub)" <gi...@apache.org>.
alyssa-db commented on PR #59247:
URL: https://github.com/apache/spark/pull/59247#issuecomment-6151895402
@eason-yuchen-liu
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-59986][SS] Expose trigger metadata in streaming query progress [spark]
Posted by "eason-yuchen-liu (via GitHub)" <gi...@apache.org>.
eason-yuchen-liu commented on code in PR #59247:
URL: https://github.com/apache/spark/pull/59247#discussion_r4264028855
##########
sql/api/src/main/scala/org/apache/spark/sql/streaming/progress.scala:
##########
@@ -219,6 +220,12 @@ class StreamingQueryProgress private[spark] (
val observedMetrics: ju.Map[String, Row])
extends Serializable {
+ // JSON for the configured trigger, omitted for legacy progress records.
+ @JsonRawValue
+ @JsonInclude(JsonInclude.Include.NON_NULL)
+ @JsonDeserialize(using = classOf[ObjectToStringDeserializer])
+ private[spark] var trigger: String = _
Review Comment:
Can we make it a constructor variable of StreamingQueryProgress similar to other fields?
##########
sql/api/src/main/scala/org/apache/spark/sql/execution/streaming/Triggers.scala:
##########
@@ -30,7 +30,19 @@ import org.apache.spark.sql.catalyst.util.SparkIntervalUtils
import org.apache.spark.sql.streaming.Trigger
import org.apache.spark.unsafe.types.UTF8String
-private object Triggers {
+private[spark] object Triggers {
+ def toJson(trigger: Trigger): String = trigger match {
+ case ProcessingTimeTrigger(intervalMs) =>
+ s"""{"type":"ProcessingTime","intervalMs":$intervalMs}"""
+ case ContinuousTrigger(intervalMs) =>
+ s"""{"type":"Continuous","intervalMs":$intervalMs}"""
+ case RealTimeTrigger(batchDurationMs) =>
+ s"""{"type":"RealTime","batchDurationMs":$batchDurationMs}"""
+ case AvailableNowTrigger => """{"type":"AvailableNow"}"""
+ case OneTimeTrigger => """{"type":"OneTime"}"""
+ case _ => null
Review Comment:
better to throw unreachable error
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-59986][SS] Expose trigger metadata in streaming query progress [spark]
Posted by "alyssa-db (via GitHub)" <gi...@apache.org>.
alyssa-db commented on code in PR #59247:
URL: https://github.com/apache/spark/pull/59247#discussion_r4289384388
##########
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:
This is validate / throw an error during query.start(), to check if triggerConfig doesn't map to one of the options. Otherwise it throws during query.processAllAvailable() or awaitTermination().
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-59986][SS] Expose trigger metadata in streaming query progress [spark]
Posted by "alyssa-db (via GitHub)" <gi...@apache.org>.
alyssa-db commented on code in PR #59247:
URL: https://github.com/apache/spark/pull/59247#discussion_r4289485595
##########
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:
I see deserialization used with StreamingQueryProgress.fromJson, such as [here](https://github.com/apache/spark/blob/ba9a922fbd14da1b5186a91184cafa5b7c50b5db/sql/connect/common/src/main/scala/org/apache/spark/sql/connect/StreamingQuery.scala#L64-L75) and a good number of examples.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-59986][SS] Expose trigger metadata in streaming query progress [spark]
Posted by "HyukjinKwon (via GitHub)" <gi...@apache.org>.
HyukjinKwon commented on code in PR #59247:
URL: https://github.com/apache/spark/pull/59247#discussion_r4263088983
##########
python/pyspark/sql/streaming/listener.py:
##########
@@ -583,6 +588,15 @@ def batchDuration(self) -> int:
"""
return self["batchDuration"]
+ @property
+ def trigger(self) -> Optional[Dict[str, Any]]:
+ """
+ The configured trigger type and options, or ``None`` when unavailable.
+
+ .. versionadded:: 5.0.0
Review Comment:
**Nit (P3):** This is master's version. Under the repo's versioning policy (AGENTS.md, "Versioning and Branch Policy"), an additive change opened against master ships first in `branch-4.x` (currently 4.4.0-SNAPSHOT). As written, the PySpark 4.4 docs would say the property first appeared in 5.0.0. Other new PySpark APIs on this base already use 4.4.0.
```suggestion
.. versionadded:: 4.4.0
```
**Verification:**
- **Inspection:** The version annotation on the new trigger property matches the branch-4.x release version (4.4.0), and the changed module has no `versionadded:: 5.0.0`.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-59986][SS] Expose trigger metadata in streaming query progress [spark]
Posted by "alyssa-db (via GitHub)" <gi...@apache.org>.
alyssa-db commented on PR #59247:
URL: https://github.com/apache/spark/pull/59247#issuecomment-6375829131
Rather than logging a trigger externally, we will use internal logging approaches.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-59986][SS] Expose trigger metadata in streaming query progress [spark]
Posted by "alyssa-db (via GitHub)" <gi...@apache.org>.
alyssa-db closed pull request #59247: [SPARK-59986][SS] Expose trigger metadata in streaming query progress
URL: https://github.com/apache/spark/pull/59247
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org