[WIP] Introduce a new callback "onQueryExecutionStart" to StreamingQueryListener

[JevonCowell]

What changes were proposed in this pull request?

• Added a new streaming query event called onQueryExecutionStart. This query event is published every time a streaming query is triggered.
• Added the necessary methods to publish this new event to all StreamingQueryListeners with backwards compatibility.

Why are the changes needed?

Currently, StreamingQueryListeners does not notify users when a query is triggered; it only provides updates when a query starts, progresses, becomes idle, or terminates. By introducing an event that is emitted whenever a query is triggered, users will be informed of this occurrence. This new event allows StreamingQueryListeners to function as a "state machine."

In my use case, I plan to use this new event along with QueryProgress and QueryIdle to develop a Maintenance Job Orchestration System. This system will programmatically submit and execute Spark commands, determining the eligibility of maintenance job executions based on the activity status of streaming queries.

Bare bone example of Maintenance Job Orchestration System:

sequenceDiagram
    participant Scheduler as MaintenanceScheduler
    participant Orchestrator as MaintenanceJobOrchestrator
    participant Job as MaintenanceJob
    participant BackgroundExecutor as BackgroundTaskExecutor
    participant Listener as StreamingQueryListener
    Orchestrator->>Listener: Register consumer to update query status
    Scheduler->>Scheduler: scheduleJob(Job)
    alt MaintenanceJob is null
        Scheduler->>Scheduler: throw IllegalArgumentException
    else MaintenanceJob is not enabled
        Scheduler->>Scheduler: return
    else
        loop Submit to Orchestrator Queue on a schedule
            Scheduler->>Orchestrator: enqueue(MaintenanceJobRequest)
        end
    end
    loop Continuous Operation
        BackgroundExecutor->>Orchestrator: attempt to run jobs in queue
        Listener->>Orchestrator: isStreamingQueryActivelyRunning()
        Orchestrator->>Orchestrator: checkIfSafeToRunJob()
        alt A Spark job is currently executing
            Orchestrator->>Orchestrator: return
        else No Spark jobs are currently executing
            Orchestrator->>BackgroundExecutor: submit(job.execute(sparkSession))
            BackgroundExecutor->>Job: execute()
            Job->>Orchestrator: return result
            Orchestrator->>Orchestrator: log job
        end
    end
    Listener->>Listener: shutdown()
    BackgroundExecutor->>BackgroundExecutor: shutdown()
    Orchestrator->>Orchestrator: shutdown()
    Scheduler->>Scheduler: shutdown()

Loading

I envision several other valuable use cases, such as enhanced metrics, improved alerting capabilities, and potential optimizations within Spark itself.

Initially, I considered introducing a QueryAwaitEvent to complement the onQueryExecutionStart. This could provide a cleaner approach, avoiding the need to manage both onQueryProgress and onQueryIdle. However, I believe addressing the current gap with onQueryExecutionStart alone is more effective.

Does this PR introduce any user-facing change?

Yes/No. I tried my best to match how QueryIdle was added to maintain backwards compatibility & onQueryExecutionStart egress is disabled by default. Users will have to use spark conf to opt into this feature. Existing implementations should not break as a result of this.

Users should do the following to opt-in:

import org.apache.spark.sql.SparkSession

val spark = SparkSession.builder
  .appName("ExampleApp")
  .config("spark.sql.streaming.query.trigger.start.event.enabled", true)
  .config("spark.sql.streaming.query.trigger.start.event.minInterval", 2000)
  .getOrCreate()

To utilize the new method they just need to override it like so:

Scala:

/**
 * Called when a query's microbatch trigger is started.
 * @since 4.2.0
 */
def onQueryTriggerStart(event: onQueryExecutionStart): Unit = {
  println(s"Query trigger started: ${event.id}")
}

Java:

@Override
public void onQueryTriggerStart(onQueryExecutionStart event) {
    System.out.println("Query trigger started: " + event.getId());
}

Python:

def onQueryTriggerStart(self, event):
    print(f"Query trigger started: {event.id}")

How was this patch tested?

IN-PROGRESS

Was this patch authored or co-authored using generative AI tooling?

I utilize ChatGPT to get familiar with the Spark codebase since this is my first time diving in. Aside from that, all code is 100% authored by me.

• RFC on SPARK-51731
• Based on SPARK-51731 RFC add relevant test
• Add New PR pointing to Spark's master branch

[JevonCowell]

I'm not super familiar with spark code, but are these based on commands_pb2.pyi? If so shouldn't onQueryIdle be 3?

[JevonCowell]

Not sure if i should make this a dependency chain or just inherit from the first collection

[JevonCowell]

Will sleep on this but wondering if this is the best place to put this, both ContinousProcessing & MicroBatchExecution utilize this method. Based on documentation, ContinousProcessing might be executing this method in the milliseconds (need to dive into code to confirm, this is only my impression)

The

 if (now - lastNoExecutionProgressEventTime >= queryTriggerStartEventMinInterval) {
       postEvent(
         new QueryTriggerStartEvent(id, runId, name, formatTimestamp(currentTriggerStartTimestamp)))
       lastNoExecutionProgressEventTime = now
     }

should make it so that we are not emitting events on each "continuous batch" but still not 100% sure of the consequences of computing this every time and how it affects the "(~1 ms) end-to-end latency" SLA. I might need to move this over to just MicroBatchExecution but not sure about how apache folks will fewel about it.