[SPARK-11848][SQL] Support EXPLAIN in DataSet APIs

When debugging DataSet API, I always need to print the logical and physical plans.

I am wondering if we should provide a simple API for EXPLAIN?

Author: gatorsmile <gatorsmile@gmail.com>

Closes #9832 from gatorsmile/explainDS.

sql/core/src/main/scala/org/apache/spark/sql/DataFrame.scala

@@ -37,7 +37,7 @@ import org.apache.spark.sql.catalyst.expressions.aggregate._
 import org.apache.spark.sql.catalyst.plans.logical._
 import org.apache.spark.sql.catalyst.plans.{Inner, JoinType}
 import org.apache.spark.sql.catalyst.{CatalystTypeConverters, ScalaReflection, SqlParser}
-import org.apache.spark.sql.execution.{EvaluatePython, ExplainCommand, FileRelation, LogicalRDD, QueryExecution, Queryable, SQLExecution}
+import org.apache.spark.sql.execution.{EvaluatePython, FileRelation, LogicalRDD, QueryExecution, Queryable, SQLExecution}
 import org.apache.spark.sql.execution.datasources.{CreateTableUsingAsSelect, LogicalRelation}
 import org.apache.spark.sql.execution.datasources.json.JacksonGenerator
 import org.apache.spark.sql.sources.HadoopFsRelation
@@ -308,27 +308,6 @@ class DataFrame private[sql](
   def printSchema(): Unit = println(schema.treeString)
   // scalastyle:on println
 
-  /**
-   * Prints the plans (logical and physical) to the console for debugging purposes.
-   * @group basic
-   * @since 1.3.0
-   */
-  def explain(extended: Boolean): Unit = {
-    val explain = ExplainCommand(queryExecution.logical, extended = extended)
-    withPlan(explain).queryExecution.executedPlan.executeCollect().foreach {
-      // scalastyle:off println
-      r => println(r.getString(0))
-      // scalastyle:on println
-    }
-  }
-
-  /**
-   * Only prints the physical plan to the console for debugging purposes.
-   * @group basic
-   * @since 1.3.0
-   */
-  def explain(): Unit = explain(extended = false)
-
   /**
    * Returns true if the `collect` and `take` methods can be run locally
    * (without any Spark executors).

sql/core/src/main/scala/org/apache/spark/sql/execution/Queryable.scala

@@ -17,6 +17,7 @@
 
 package org.apache.spark.sql.execution
 
+import org.apache.spark.sql.SQLContext
 import org.apache.spark.sql.types.StructType
 
 import scala.util.control.NonFatal
@@ -25,6 +26,7 @@ import scala.util.control.NonFatal
 private[sql] trait Queryable {
   def schema: StructType
   def queryExecution: QueryExecution
+  def sqlContext: SQLContext
 
   override def toString: String = {
     try {
@@ -34,4 +36,23 @@ private[sql] trait Queryable {
         s"Invalid tree; ${e.getMessage}:\n$queryExecution"
     }
   }
+
+  /**
+   * Prints the plans (logical and physical) to the console for debugging purposes.
+   * @since 1.3.0
+   */
+  def explain(extended: Boolean): Unit = {
+    val explain = ExplainCommand(queryExecution.logical, extended = extended)
+    sqlContext.executePlan(explain).executedPlan.executeCollect().foreach {
+      // scalastyle:off println
+      r => println(r.getString(0))
+      // scalastyle:on println
+    }
+  }
+
+  /**
+   * Only prints the physical plan to the console for debugging purposes.
+   * @since 1.3.0
+   */
+  def explain(): Unit = explain(extended = false)
 }