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: [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]
