[SPARK-49287][SPARK-49432][CONNECT][SQL] Move streaming classes to sql/api

### What changes were proposed in this pull request?
This PR moves a number of streaming classes into sql/api.  This is the initial step in create a shared interface for streaming operations.

### Why are the changes needed?

### Does this PR introduce _any_ user-facing change?
No.

### How was this patch tested?
Existing tests.

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

Closes #48017 from hvanhovell/SPARK-49287.

Authored-by: Herman van Hovell <[email protected]>
Signed-off-by: Herman van Hovell <[email protected]>

Author: hvanhovell

common/utils/src/main/scala/org/apache/spark/scheduler/SparkListenerEvent.scala

@@ -0,0 +1,28 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.spark.scheduler
+
+import com.fasterxml.jackson.annotation.JsonTypeInfo
+
+import org.apache.spark.annotation.DeveloperApi
+
+@DeveloperApi
+@JsonTypeInfo(use = JsonTypeInfo.Id.CLASS, include = JsonTypeInfo.As.PROPERTY, property = "Event")
+trait SparkListenerEvent {
+  /* Whether output this event to the event log */
+  protected[spark] def logEvent: Boolean = true
+}

connector/connect/client/jvm/src/main/scala/org/apache/spark/sql/streaming/StreamingQuery.scala

@@ -18,166 +18,21 @@
 package org.apache.spark.sql.streaming
 
 import java.util.UUID
-import java.util.concurrent.TimeoutException
 
 import scala.jdk.CollectionConverters._
 
-import org.apache.spark.annotation.Evolving
 import org.apache.spark.connect.proto.Command
 import org.apache.spark.connect.proto.ExecutePlanResponse
 import org.apache.spark.connect.proto.StreamingQueryCommand
 import org.apache.spark.connect.proto.StreamingQueryCommandResult
 import org.apache.spark.connect.proto.StreamingQueryManagerCommandResult.StreamingQueryInstance
-import org.apache.spark.sql.SparkSession
+import org.apache.spark.sql.{api, Dataset, SparkSession}
 
-/**
- * A handle to a query that is executing continuously in the background as new data arrives. All
- * these methods are thread-safe.
- * @since 3.5.0
- */
-@Evolving
-trait StreamingQuery {
-  // This is a copy of StreamingQuery in sql/core/.../streaming/StreamingQuery.scala
-
-  /**
-   * Returns the user-specified name of the query, or null if not specified. This name can be
-   * specified in the `org.apache.spark.sql.streaming.DataStreamWriter` as
-   * `dataframe.writeStream.queryName("query").start()`. This name, if set, must be unique across
-   * all active queries.
-   *
-   * @since 3.5.0
-   */
-  def name: String
-
-  /**
-   * Returns the unique id of this query that persists across restarts from checkpoint data. That
-   * is, this id is generated when a query is started for the first time, and will be the same
-   * every time it is restarted from checkpoint data. Also see [[runId]].
-   *
-   * @since 3.5.0
-   */
-  def id: UUID
-
-  /**
-   * Returns the unique id of this run of the query. That is, every start/restart of a query will
-   * generate a unique runId. Therefore, every time a query is restarted from checkpoint, it will
-   * have the same [[id]] but different [[runId]]s.
-   */
-  def runId: UUID
-
-  /**
-   * Returns the `SparkSession` associated with `this`.
-   *
-   * @since 3.5.0
-   */
-  def sparkSession: SparkSession
-
-  /**
-   * Returns `true` if this query is actively running.
-   *
-   * @since 3.5.0
-   */
-  def isActive: Boolean
-
-  /**
-   * Returns the [[StreamingQueryException]] if the query was terminated by an exception.
-   * @since 3.5.0
-   */
-  def exception: Option[StreamingQueryException]
-
-  /**
-   * Returns the current status of the query.
-   *
-   * @since 3.5.0
-   */
-  def status: StreamingQueryStatus
-
-  /**
-   * Returns an array of the most recent [[StreamingQueryProgress]] updates for this query. The
-   * number of progress updates retained for each stream is configured by Spark session
-   * configuration `spark.sql.streaming.numRecentProgressUpdates`.
-   *
-   * @since 3.5.0
-   */
-  def recentProgress: Array[StreamingQueryProgress]
-
-  /**
-   * Returns the most recent [[StreamingQueryProgress]] update of this streaming query.
-   *
-   * @since 3.5.0
-   */
-  def lastProgress: StreamingQueryProgress
-
-  /**
-   * Waits for the termination of `this` query, either by `query.stop()` or by an exception. If
-   * the query has terminated with an exception, then the exception will be thrown.
-   *
-   * If the query has terminated, then all subsequent calls to this method will either return
-   * immediately (if the query was terminated by `stop()`), or throw the exception immediately (if
-   * the query has terminated with exception).
-   *
-   * @throws StreamingQueryException
-   *   if the query has terminated with an exception.
-   * @since 3.5.0
-   */
-  @throws[StreamingQueryException]
-  def awaitTermination(): Unit
-
-  /**
-   * Waits for the termination of `this` query, either by `query.stop()` or by an exception. If
-   * the query has terminated with an exception, then the exception will be thrown. Otherwise, it
-   * returns whether the query has terminated or not within the `timeoutMs` milliseconds.
-   *
-   * If the query has terminated, then all subsequent calls to this method will either return
-   * `true` immediately (if the query was terminated by `stop()`), or throw the exception
-   * immediately (if the query has terminated with exception).
-   *
-   * @throws StreamingQueryException
-   *   if the query has terminated with an exception
-   * @since 3.5.0
-   */
-  @throws[StreamingQueryException]
-  def awaitTermination(timeoutMs: Long): Boolean
-
-  /**
-   * Blocks until all available data in the source has been processed and committed to the sink.
-   * This method is intended for testing. Note that in the case of continually arriving data, this
-   * method may block forever. Additionally, this method is only guaranteed to block until data
-   * that has been synchronously appended data to a
-   * `org.apache.spark.sql.execution.streaming.Source` prior to invocation. (i.e. `getOffset` must
-   * immediately reflect the addition).
-   * @since 3.5.0
-   */
-  def processAllAvailable(): Unit
-
-  /**
-   * Stops the execution of this query if it is running. This waits until the termination of the
-   * query execution threads or until a timeout is hit.
-   *
-   * By default stop will block indefinitely. You can configure a timeout by the configuration
-   * `spark.sql.streaming.stopTimeout`. A timeout of 0 (or negative) milliseconds will block
-   * indefinitely. If a `TimeoutException` is thrown, users can retry stopping the stream. If the
-   * issue persists, it is advisable to kill the Spark application.
-   *
-   * @since 3.5.0
-   */
-  @throws[TimeoutException]
-  def stop(): Unit
-
-  /**
-   * Prints the physical plan to the console for debugging purposes.
-   * @since 3.5.0
-   */
-  def explain(): Unit
-
-  /**
-   * Prints the physical plan to the console for debugging purposes.
-   *
-   * @param extended
-   *   whether to do extended explain or not
-   * @since 3.5.0
-   */
-  def explain(extended: Boolean): Unit
+/** @inheritdoc */
+trait StreamingQuery extends api.StreamingQuery[Dataset] {
+
+  /** @inheritdoc */
+  override def sparkSession: SparkSession
 }
 
 class RemoteStreamingQuery(

connector/connect/client/jvm/src/test/scala/org/apache/spark/sql/connect/client/CheckConnectJvmClientCompatibility.scala

@@ -145,21 +145,6 @@ object CheckConnectJvmClientCompatibility {
     checkMiMaCompatibility(clientJar, protobufJar, includedRules, excludeRules)
   }
 
-  private lazy val mergeIntoWriterExcludeRules: Seq[ProblemFilter] = {
-    // Exclude some auto-generated methods in [[MergeIntoWriter]] classes.
-    // The incompatible changes are due to the uses of [[proto.Expression]] instead
-    // of [[catalyst.Expression]] in the method signature.
-    val classNames = Seq("WhenMatched", "WhenNotMatched", "WhenNotMatchedBySource")
-    val methodNames = Seq("apply", "condition", "copy", "copy$*", "unapply")
-
-    classNames.flatMap { className =>
-      methodNames.map { methodName =>
-        ProblemFilters.exclude[IncompatibleSignatureProblem](
-          s"org.apache.spark.sql.$className.$methodName")
-      }
-    }
-  }
-
   private def checkMiMaCompatibilityWithSqlModule(
       clientJar: File,
       sqlJar: File): List[Problem] = {
@@ -269,12 +254,6 @@ object CheckConnectJvmClientCompatibility {
         "org.apache.spark.sql.streaming.TestGroupState"),
       ProblemFilters.exclude[MissingClassProblem](
         "org.apache.spark.sql.streaming.TestGroupState$"),
-      ProblemFilters.exclude[MissingClassProblem](
-        "org.apache.spark.sql.streaming.PythonStreamingQueryListener"),
-      ProblemFilters.exclude[MissingClassProblem](
-        "org.apache.spark.sql.streaming.PythonStreamingQueryListenerWrapper"),
-      ProblemFilters.exclude[MissingTypesProblem](
-        "org.apache.spark.sql.streaming.StreamingQueryListener$Event"),
 
       // SQLImplicits
       ProblemFilters.exclude[Problem]("org.apache.spark.sql.SQLImplicits.rddToDatasetHolder"),
@@ -286,10 +265,6 @@ object CheckConnectJvmClientCompatibility {
       ProblemFilters.exclude[MissingClassProblem](
         "org.apache.spark.sql.artifact.ArtifactManager$"),
 
-      // UDFRegistration
-      ProblemFilters.exclude[DirectMissingMethodProblem](
-        "org.apache.spark.sql.UDFRegistration.register"),
-
       // ColumnNode conversions
       ProblemFilters.exclude[DirectMissingMethodProblem](
         "org.apache.spark.sql.SparkSession.Converter"),
@@ -304,6 +279,8 @@ object CheckConnectJvmClientCompatibility {
       ProblemFilters.exclude[MissingClassProblem]("org.apache.spark.sql.expressions.scalalang.*"),
 
       // UDFRegistration
+      ProblemFilters.exclude[DirectMissingMethodProblem](
+        "org.apache.spark.sql.UDFRegistration.register"),
       ProblemFilters.exclude[MissingTypesProblem]("org.apache.spark.sql.UDFRegistration"),
       ProblemFilters.exclude[DirectMissingMethodProblem](
         "org.apache.spark.sql.UDFRegistration.log*"),
@@ -326,14 +303,7 @@ object CheckConnectJvmClientCompatibility {
       ProblemFilters.exclude[DirectMissingMethodProblem](
         "org.apache.spark.sql.DataFrameReader.validateJsonSchema"),
       ProblemFilters.exclude[DirectMissingMethodProblem](
-        "org.apache.spark.sql.DataFrameReader.validateXmlSchema"),
-
-      // Datasource V2 partition transforms
-      ProblemFilters.exclude[MissingClassProblem]("org.apache.spark.sql.PartitionTransform"),
-      ProblemFilters.exclude[MissingClassProblem]("org.apache.spark.sql.PartitionTransform$"),
-      ProblemFilters.exclude[MissingClassProblem](
-        "org.apache.spark.sql.PartitionTransform$ExtractTransform")) ++
-      mergeIntoWriterExcludeRules
+        "org.apache.spark.sql.DataFrameReader.validateXmlSchema"))
 
     checkMiMaCompatibility(clientJar, sqlJar, includedRules, excludeRules)
   }
@@ -443,8 +413,7 @@ object CheckConnectJvmClientCompatibility {
 
       // Encoders are in the wrong JAR
       ProblemFilters.exclude[MissingClassProblem]("org.apache.spark.sql.Encoders"),
-      ProblemFilters.exclude[MissingClassProblem]("org.apache.spark.sql.Encoders$")) ++
-      mergeIntoWriterExcludeRules
+      ProblemFilters.exclude[MissingClassProblem]("org.apache.spark.sql.Encoders$"))
 
     checkMiMaCompatibility(sqlJar, clientJar, includedRules, excludeRules)
   }

core/src/main/scala/org/apache/spark/scheduler/SparkListener.scala

@@ -22,22 +22,13 @@ import javax.annotation.Nullable
 
 import scala.collection.Map
 
-import com.fasterxml.jackson.annotation.JsonTypeInfo
-
 import org.apache.spark.TaskEndReason
 import org.apache.spark.annotation.{DeveloperApi, Since}
 import org.apache.spark.executor.{ExecutorMetrics, TaskMetrics}
 import org.apache.spark.resource.ResourceProfile
 import org.apache.spark.scheduler.cluster.ExecutorInfo
 import org.apache.spark.storage.{BlockManagerId, BlockUpdatedInfo}
 
-@DeveloperApi
-@JsonTypeInfo(use = JsonTypeInfo.Id.CLASS, include = JsonTypeInfo.As.PROPERTY, property = "Event")
-trait SparkListenerEvent {
-  /* Whether output this event to the event log */
-  protected[spark] def logEvent: Boolean = true
-}
-
 @DeveloperApi
 case class SparkListenerStageSubmitted(stageInfo: StageInfo, properties: Properties = null)
   extends SparkListenerEvent

project/MimaExcludes.scala

@@ -159,6 +159,22 @@ object MimaExcludes {
     ProblemFilters.exclude[MissingClassProblem]("org.apache.spark.sql.CreateTableWriter"),
     ProblemFilters.exclude[MissingClassProblem]("org.apache.spark.sql.DataFrameWriterV2"),
     ProblemFilters.exclude[MissingClassProblem]("org.apache.spark.sql.WriteConfigMethods"),
+
+    // SPARK-49287: Shared Streaming interfaces
+    ProblemFilters.exclude[MissingClassProblem]("org.apache.spark.scheduler.SparkListenerEvent"),
+    ProblemFilters.exclude[MissingClassProblem]("org.apache.spark.sql.ForeachWriter"),
+    ProblemFilters.exclude[MissingClassProblem]("org.apache.spark.sql.streaming.SourceProgress"),
+    ProblemFilters.exclude[MissingClassProblem]("org.apache.spark.sql.streaming.SourceProgress$"),
+    ProblemFilters.exclude[MissingClassProblem]("org.apache.spark.sql.streaming.StateOperatorProgress"),
+    ProblemFilters.exclude[MissingClassProblem]("org.apache.spark.sql.streaming.StateOperatorProgress$"),
+    ProblemFilters.exclude[MissingClassProblem]("org.apache.spark.sql.streaming.StreamingQueryListener"),
+    ProblemFilters.exclude[MissingClassProblem]("org.apache.spark.sql.streaming.StreamingQueryListener$"),
+    ProblemFilters.exclude[MissingClassProblem]("org.apache.spark.sql.streaming.StreamingQueryListener$Event"),
+    ProblemFilters.exclude[MissingClassProblem]("org.apache.spark.sql.streaming.StreamingQueryListener$QueryIdleEvent"),
+    ProblemFilters.exclude[MissingClassProblem]("org.apache.spark.sql.streaming.StreamingQueryListener$QueryProgressEvent"),
+    ProblemFilters.exclude[MissingClassProblem]("org.apache.spark.sql.streaming.StreamingQueryListener$QueryStartedEvent"),
+    ProblemFilters.exclude[MissingClassProblem]("org.apache.spark.sql.streaming.StreamingQueryListener$QueryTerminatedEvent"),
+    ProblemFilters.exclude[MissingClassProblem]("org.apache.spark.sql.streaming.StreamingQueryStatus"),
   )
 
   // Default exclude rules

connector/connect/client/jvm/src/main/scala/org/apache/spark/sql/ForeachWriter.scala renamed to sql/api/src/main/scala/org/apache/spark/sql/ForeachWriter.scala

@@ -89,7 +89,7 @@ package org.apache.spark.sql
  *  });
  * }}}
  *
- * @since 3.5.0
+ * @since 2.0.0
  */
 abstract class ForeachWriter[T] extends Serializable {