Fix StreamingQuery explain command

sql/core/src/main/scala/org/apache/spark/sql/execution/command/commands.scala

@@ -89,19 +89,21 @@ case class ExecutedCommandExec(cmd: RunnableCommand) extends SparkPlan {
  * @param output output schema
  * @param extended whether to do extended explain or not
  * @param codegen whether to output generated code from whole-stage codegen or not
+ * @param streaming whether it's a streaming plan
  */
 case class ExplainCommand(
     logicalPlan: LogicalPlan,
     override val output: Seq[Attribute] =
       Seq(AttributeReference("plan", StringType, nullable = true)()),
     extended: Boolean = false,
-    codegen: Boolean = false)
+    codegen: Boolean = false,
+    streaming: Boolean = false)
   extends RunnableCommand {
 
   // Run through the optimizer to generate the physical plan.
   override def run(sparkSession: SparkSession): Seq[Row] = try {
     val queryExecution =
-      if (logicalPlan.isStreaming) {
+      if (streaming) {
         // This is used only by explaining `Dataset/DataFrame` created by `spark.readStream`, so the
         // output mode does not matter since there is no `Sink`.
         new IncrementalExecution(sparkSession, logicalPlan, OutputMode.Append(), "<unknown>", 0, 0)

sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/StreamExecution.scala

@@ -673,7 +673,7 @@ class StreamExecution(
     if (lastExecution == null) {
       "No physical plan. Waiting for data."
     } else {
-      val explain = ExplainCommand(lastExecution.logical, extended = extended)
+      val explain = ExplainCommand(lastExecution.logical, extended = extended, streaming = true)
       sparkSession.sessionState.executePlan(explain).executedPlan.executeCollect()
         .map(_.getString(0)).mkString("\n")
     }

sql/core/src/test/scala/org/apache/spark/sql/streaming/StreamSuite.scala

@@ -23,6 +23,7 @@ import scala.util.control.ControlThrowable
 import org.apache.spark.sql._
 import org.apache.spark.sql.catalyst.streaming.InternalOutputModes
 import org.apache.spark.sql.execution.streaming._
+import org.apache.spark.sql.functions._
 import org.apache.spark.sql.sources.StreamSourceProvider
 import org.apache.spark.sql.types.{IntegerType, StructField, StructType}
 
@@ -277,10 +278,11 @@ class StreamSuite extends StreamTest {
 
   test("explain") {
     val inputData = MemoryStream[String]
-    val df = inputData.toDS().map(_ + "foo")
+    val df = inputData.toDS().map(_ + "foo").groupBy("value").agg(count("*"))
     // Test `explain` not throwing errors
     df.explain()
-    val q = df.writeStream.queryName("memory_explain").format("memory").start()
+    val q = df.writeStream.queryName("memory_explain").outputMode("complete").format("memory")
+      .start()
       .asInstanceOf[StreamingQueryWrapper]
       .streamingQuery
     try {
@@ -294,12 +296,16 @@ class StreamSuite extends StreamTest {
       // `extended = false` only displays the physical plan.
       assert("LocalRelation".r.findAllMatchIn(explainWithoutExtended).size === 0)
       assert("LocalTableScan".r.findAllMatchIn(explainWithoutExtended).size === 1)
+      // Use "StateStoreRestore" to verify that it does output a streaming physical plan
+      assert(explainWithoutExtended.contains("StateStoreRestore"))
 
       val explainWithExtended = q.explainInternal(true)
       // `extended = true` displays 3 logical plans (Parsed/Optimized/Optimized) and 1 physical
       // plan.
       assert("LocalRelation".r.findAllMatchIn(explainWithExtended).size === 3)
       assert("LocalTableScan".r.findAllMatchIn(explainWithExtended).size === 1)
+      // Use "StateStoreRestore" to verify that it does output a streaming physical plan
+      assert(explainWithExtended.contains("StateStoreRestore"))
     } finally {
       q.stop()
     }