refactor: renamed query trigger start event to query execution start event

Author: JevonCowell

python/pyspark/sql/connect/streaming/query.py

@@ -33,7 +33,7 @@
     QueryProgressEvent,
     QueryIdleEvent,
     QueryTerminatedEvent,
-    QueryTriggerStartEvent,
+    QueryExecutionStartEvent,
     StreamingQueryProgress,
 )
 from pyspark.sql.streaming.query import (
@@ -394,15 +394,15 @@ def _query_event_handler(self, iter: Iterator[Dict[str, Any]]) -> None:
     @staticmethod
     def deserialize(
         event: pb2.StreamingQueryListenerEvent,
-    ) -> Union["QueryProgressEvent", "QueryIdleEvent", "QueryTerminatedEvent", "QueryTriggerStartEvent"]:
+    ) -> Union["QueryProgressEvent", "QueryIdleEvent", "QueryTerminatedEvent", "QueryExecutionStartEvent"]:
         if event.event_type == proto.StreamingQueryEventType.QUERY_PROGRESS_EVENT:
             return QueryProgressEvent.fromJson(json.loads(event.event_json))
         elif event.event_type == proto.StreamingQueryEventType.QUERY_TERMINATED_EVENT:
             return QueryTerminatedEvent.fromJson(json.loads(event.event_json))
         elif event.event_type == proto.StreamingQueryEventType.QUERY_IDLE_EVENT:
             return QueryIdleEvent.fromJson(json.loads(event.event_json))
-        elif event.event_type == proto.StreamingQueryEventType.QUERY_TRIGGER_START_EVENT:
-            return QueryTriggerStartEvent.fromJson(json.loads(event.event_json))
+        elif event.event_type == proto.StreamingQueryEventType.QUERY_EXECUTION_START_EVENT:
+            return QueryExecutionStartEvent.fromJson(json.loads(event.event_json))
         else:
             raise PySparkValueError(
                 errorClass="UNKNOWN_VALUE_FOR",
@@ -412,7 +412,7 @@ def deserialize(
     def post_to_all(
         self,
         event: Union[
-            "QueryStartedEvent", "QueryProgressEvent", "QueryIdleEvent", "QueryTerminatedEvent", "QueryTriggerStartEvent"
+            "QueryStartedEvent", "QueryProgressEvent", "QueryIdleEvent", "QueryTerminatedEvent", "QueryExecutionStartEvent"
         ],
     ) -> None:
         """
@@ -430,8 +430,8 @@ def post_to_all(
                         listener.onQueryIdle(event)
                     elif isinstance(event, QueryTerminatedEvent):
                         listener.onQueryTerminated(event)
-                    elif isinstance(event, QueryTriggerStartEvent):
-                        listener.onQueryTriggerStart(event)
+                    elif isinstance(event, QueryExecutionStartEvent):
+                        listener.onQueryExecutionStart(event)
                     else:
                         warnings.warn(f"Unknown StreamingQueryListener event: {event}")
                 except Exception as e:

python/pyspark/sql/connect/streaming/worker/listener_worker.py

@@ -39,7 +39,7 @@
     QueryProgressEvent,
     QueryTerminatedEvent,
     QueryIdleEvent,
-    QueryTriggerStartEvent,
+    QueryExecutionStartEvent,
 )
 from pyspark.worker_util import check_python_version
 
@@ -91,7 +91,7 @@ def process(listener_event_str, listener_event_type):  # type: ignore[no-untyped
         elif listener_event_type == 3:
             listener.onQueryTerminated(QueryTerminatedEvent.fromJson(listener_event))
         elif listener_event_type == 4:
-            listener.onQueryTriggerStart(QueryTriggerStartEvent.fromJson(listener_event))
+            listener.onQueryExecutionStart(QueryExecutionStartEvent.fromJson(listener_event))
 
     while True:
         event = utf8_deserializer.loads(infile)

python/pyspark/sql/dataframe.py

@@ -4373,7 +4373,7 @@ def observe(
         ...    def onQueryTerminated(self, event):
         ...        pass
         ...
-        ...    def onQueryTriggerStart(self, event):
+        ...    def onQueryExecutionStart(self, event):
         ...        pass
         ...
         >>> error_listener = MyErrorListener()

python/pyspark/sql/streaming/listener.py

@@ -60,7 +60,7 @@ class StreamingQueryListener(ABC):
     ...        # Do something with event.
     ...        pass
     ...
-    ...    def onQueryTriggerStart(self, event: QueryTriggerStartEvent) -> None:
+    ...    def onQueryExecutionStart(self, event: QueryExecutionStartEvent) -> None:
     ...        # Do something with event.
     ...        pass
     ...
@@ -133,7 +133,7 @@ def onQueryTerminated(self, event: "QueryTerminatedEvent") -> None:
 
     # NOTE: Do not mark this as abstract method, we are following the same pattern as
     # onQueryIdle to avoid breaking existing implementations.
-    def onQueryTriggerStart(self, event: "QueryTriggerStartEvent") -> None:
+    def onQueryExecutionStart(self, event: "QueryExecutionStartEvent") -> None:
         """
         Called when a query trigger is started.
         """
@@ -174,8 +174,8 @@ def onQueryIdle(self, jevent: "JavaObject") -> None:
     def onQueryTerminated(self, jevent: "JavaObject") -> None:
         self.pylistener.onQueryTerminated(QueryTerminatedEvent.fromJObject(jevent))
 
-    def onQueryTriggerStart(self, jevent: "JavaObject") -> None:
-        self.pylistener.onQueryTriggerStart(QueryTriggerStartEvent.fromJObject(jevent))
+    def onQueryExecutionStart(self, jevent: "JavaObject") -> None:
+        self.pylistener.onQueryExecutionStart(QueryExecutionStartEvent.fromJObject(jevent))
 
     class Java:
         implements = ["org.apache.spark.sql.streaming.PythonStreamingQueryListener"]
@@ -411,7 +411,7 @@ def errorClassOnException(self) -> Optional[str]:
         """
         return self._errorClassOnException
 
-class QueryTriggerStartEvent:
+class QueryExecutionStartEvent:
     """
     Event representing the start of a query trigger.
 
@@ -434,7 +434,7 @@ def __init__(
         self._timestamp: str = timestamp
 
     @classmethod
-    def fromJObject(cls, jevent: "JavaObject") -> "QueryTriggerStartEvent":
+    def fromJObject(cls, jevent: "JavaObject") -> "QueryExecutionStartEvent":
         return cls(
             id=uuid.UUID(jevent.id().toString()),
             runId=uuid.UUID(jevent.runId().toString()),
@@ -443,7 +443,7 @@ def fromJObject(cls, jevent: "JavaObject") -> "QueryTriggerStartEvent":
         )
 
     @classmethod
-    def fromJson(cls, j: Dict[str, Any]) -> "QueryTriggerStartEvent":
+    def fromJson(cls, j: Dict[str, Any]) -> "QueryExecutionStartEvent":
         return cls(
             id=uuid.UUID(j["id"]),
             runId=uuid.UUID(j["runId"]),

python/pyspark/sql/streaming/query.py

@@ -656,7 +656,7 @@ def addListener(self, listener: StreamingQueryListener) -> None:
         ...     def onQueryTerminated(self, event):
         ...         pass
         ...
-        ...     def onQueryTriggerStart(self, event):
+        ...     def onQueryExecutionStart(self, event):
         ...         pass
         ...
         >>> test_listener = TestListener()
@@ -708,7 +708,7 @@ def removeListener(self, listener: StreamingQueryListener) -> None:
         ...     def onQueryTerminated(self, event):
         ...         pass
         ...
-        ...     def onQueryTriggerStart(self, event):
+        ...     def onQueryExecutionStart(self, event):
         ...         pass
         ...
         >>> test_listener = TestListener()

python/pyspark/sql/tests/connect/streaming/test_parity_listener.py

@@ -46,7 +46,7 @@ def onQueryTerminated(self, event):
         df = self.spark.createDataFrame(data=[(e,)])
         df.write.mode("append").saveAsTable("listener_terminated_events")
 
-    def onQueryTriggerStart(self, event) -> None:
+    def onQueryExecutionStart(self, event) -> None:
         pass
 
 
@@ -86,7 +86,7 @@ def onQueryIdle(self, event):
     def onQueryTerminated(self, event):
         self.terminated.append(event)
 
-    def onQueryTriggerStart(self, event) -> None:
+    def onQueryExecutionStart(self, event) -> None:
         pass
 
 

python/pyspark/sql/tests/streaming/test_streaming_listener.py

@@ -216,7 +216,7 @@ def onQueryIdle(self, event):
             def onQueryTerminated(self, event):
                 pass
 
-            def onQueryTriggerStart(self, event):
+            def onQueryExecutionStart(self, event):
                 pass
 
         try:
@@ -361,7 +361,7 @@ def onQueryTerminated(self, event):
                 nonlocal terminated_event
                 terminated_event = event
 
-        # V3: The interface after the method `onQueryTriggerStart` is added. It is Spark 4.1+.
+        # V3: The interface after the method `onQueryExecutionStart` is added. It is Spark 4.1+.
         class TestListenerV3(StreamingQueryListener):
             def onQueryStarted(self, event):
                 nonlocal start_event
@@ -378,7 +378,7 @@ def onQueryTerminated(self, event):
                 nonlocal terminated_event
                 terminated_event = event
 
-            def onQueryTriggerStart(self, event):
+            def onQueryExecutionStart(self, event):
                 pass
 
         def verify(test_listener):
@@ -464,7 +464,7 @@ def onQueryIdle(self, event):
             def onQueryTerminated(self, event):
                 pass
 
-        # V3: The interface after the method `onQueryTriggerStart` is added. It is Spark 4.1+.
+        # V3: The interface after the method `onQueryExecutionStart` is added. It is Spark 4.1+.
         class TestListenerV3(StreamingQueryListener):
             def onQueryStarted(self, event):
                 pass
@@ -478,7 +478,7 @@ def onQueryIdle(self, event):
             def onQueryTerminated(self, event):
                 pass
 
-            def onQueryTriggerStart(self, event):
+            def onQueryExecutionStart(self, event):
                 pass
 
         def verify(test_listener):

python/pyspark/sql/tests/test_observation.py

@@ -137,7 +137,7 @@ def onQueryIdle(self, event):
             def onQueryTerminated(self, event):
                 pass
 
-            def onQueryTriggerStart(self, event):
+            def onQueryExecutionStart(self, event):
                 pass
 
         self.spark.streams.addListener(TestListener())

sql/api/src/main/scala/org/apache/spark/sql/streaming/StreamingQueryListener.scala

@@ -81,7 +81,7 @@ abstract class StreamingQueryListener extends Serializable {
    * Called when a query's microbatch trigger is started.
    * @since 4.2.0
    */
-  def onQueryTriggerStart(event: QueryTriggerStartEvent): Unit = {}
+  def onQueryExecutionStart(event: QueryExecutionStartEvent): Unit = {}
 }
 
 /**
@@ -98,7 +98,7 @@ private[spark] trait PythonStreamingQueryListener {
 
   def onQueryTerminated(event: QueryTerminatedEvent): Unit
 
-  def onQueryTriggerStart(event: QueryTriggerStartEvent): Unit
+  def onQueryExecutionStart(event: QueryExecutionStartEvent): Unit
 }
 
 private[spark] class PythonStreamingQueryListenerWrapper(listener: PythonStreamingQueryListener)
@@ -113,8 +113,8 @@ private[spark] class PythonStreamingQueryListenerWrapper(listener: PythonStreami
 
   def onQueryTerminated(event: QueryTerminatedEvent): Unit = listener.onQueryTerminated(event)
 
-  override def onQueryTriggerStart(event: QueryTriggerStartEvent): Unit =
-    listener.onQueryTriggerStart(event)
+  override def onQueryExecutionStart(event: QueryExecutionStartEvent): Unit =
+    listener.onQueryExecutionStart(event)
 }
 
 /**
@@ -313,7 +313,7 @@ object StreamingQueryListener extends Serializable {
    *  The timestamp start of a query trigger
    */
   @Evolving
-  class QueryTriggerStartEvent private[sql] (
+  class QueryExecutionStartEvent private[sql] (
       val id: UUID,
       val runId: UUID,
       val name: String,
@@ -330,10 +330,10 @@ object StreamingQueryListener extends Serializable {
     }
   }
 
-  private[spark] object QueryTriggerStartEvent {
-    private[spark] def fromJson(json: String): QueryTriggerStartEvent = {
+  private[spark] object QueryExecutionStartEvent {
+    private[spark] def fromJson(json: String): QueryExecutionStartEvent = {
       val parser = EventParser(json)
-      new QueryTriggerStartEvent(
+      new QueryExecutionStartEvent(
         parser.getUUID("id"),
         parser.getUUID("runId"),
         parser.getString("name"),

sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala

@@ -5683,7 +5683,7 @@ object SQLConf {
       .booleanConf
       .createWithDefault(true)
 
-  val STREAMING_QUERY_TRIGGER_START_EVENT_ENABLED =
+  val STREAMING_QUERY_EXECUTION_START_EVENT_ENABLED =
     buildConf("spark.sql.streaming.query.trigger.start.event.enabled")
       .internal()
       .doc("When set to true, spark will emit events for when streaming queries micro-batches " +
@@ -5692,10 +5692,10 @@ object SQLConf {
       .booleanConf
       .createWithDefault(false)
 
-  val STREAMING_QUERY_TRIGGER_START_EVENT_MIN_INTERVAL =
+  val STREAMING_QUERY_EXECUTION_START_EVENT_MIN_INTERVAL =
     buildConf("spark.sql.streaming.query.trigger.start.event.minInterval")
       .internal()
-      .doc("The minimum interval in milliseconds between QueryTriggerStart events emissions for " +
+      .doc("The minimum interval in milliseconds between QueryExecutionStart events emissions for " +
         "streaming queries. This is to avoid flooding the event queue with events for " +
         "high-frequency queries.")
       .version("4.2.0")
@@ -6496,11 +6496,11 @@ class SQLConf extends Serializable with Logging with SqlApiConf {
   def disabledV2StreamingMicroBatchReaders: String =
     getConf(DISABLED_V2_STREAMING_MICROBATCH_READERS)
 
-  def streamingQueryTriggerStartEventEnabled: Boolean =
-    getConf(STREAMING_QUERY_TRIGGER_START_EVENT_ENABLED)
+  def streamingQueryExecutionStartEventEnabled: Boolean =
+    getConf(STREAMING_QUERY_EXECUTION_START_EVENT_ENABLED)
 
-  def streamingQueryTriggerStartEventMinInterval: Long =
-    getConf(STREAMING_QUERY_TRIGGER_START_EVENT_MIN_INTERVAL)
+  def streamingQueryExecutionStartEventMinInterval: Long =
+    getConf(STREAMING_QUERY_EXECUTION_START_EVENT_MIN_INTERVAL)
 
   def fastFailFileFormatOutput: Boolean = getConf(FASTFAIL_ON_FILEFORMAT_OUTPUT)