anishshri-db commented on code in PR #42940:
URL: https://github.com/apache/spark/pull/42940#discussion_r1329417880
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/MicroBatchExecution.scala:
##########
@@ -52,11 +52,40 @@ class MicroBatchExecution(
@volatile protected var sources: Seq[SparkDataStream] = Seq.empty
- protected val triggerExecutor: TriggerExecutor = trigger match {
- case t: ProcessingTimeTrigger => ProcessingTimeExecutor(t, triggerClock)
- case OneTimeTrigger => SingleBatchExecutor()
- case AvailableNowTrigger => MultiBatchExecutor()
- case _ => throw new IllegalStateException(s"Unknown type of trigger:
$trigger")
+ @volatile protected[sql] var triggerExecutor: TriggerExecutor = _
+
+ protected def getTrigger(): TriggerExecutor = {
+ assert(sources.nonEmpty, "sources should have been retrieved from the
plan!")
+ trigger match {
+ case t: ProcessingTimeTrigger => ProcessingTimeExecutor(t, triggerClock)
+ case OneTimeTrigger => SingleBatchExecutor()
+ case AvailableNowTrigger =>
+ // See SPARK-45178 for more details.
+ if (sparkSession.sqlContext.conf.getConf(
Review Comment:
Not sure Im reading this correctly. If the flag is enabled, we use
multi-batch directly ? Shouldn't this be the case if the flag is disabled ?
--
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]
