Factored execute() method of physical commands to parent class Command

Author: liancheng

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

@@ -40,6 +40,8 @@ trait Command {
   protected[sql] lazy val sideEffectResult: Seq[Row] = Seq.empty[Row]
 
   override def executeCollect(): Array[Row] = sideEffectResult.toArray
+
+  override def execute(): RDD[Row] = sqlContext.sparkContext.parallelize(sideEffectResult, 1)
 }
 
 /**
@@ -93,8 +95,6 @@ case class SetCommand(
       throw new IllegalArgumentException()
   }
 
-  def execute(): RDD[Row] = context.sparkContext.parallelize(sideEffectResult, 1)
-
   override def otherCopyArgs = context :: Nil
 }
 
@@ -123,8 +123,6 @@ case class ExplainCommand(
     ("Error occurred during query planning: \n" + cause.getMessage).split("\n").map(Row(_))
   }
 
-  def execute(): RDD[Row] = context.sparkContext.parallelize(sideEffectResult, 1)
-
   override def otherCopyArgs = context :: Nil
 }
 
@@ -144,11 +142,6 @@ case class CacheCommand(tableName: String, doCache: Boolean)(@transient context:
     Seq.empty[Row]
   }
 
-  override def execute(): RDD[Row] = {
-    sideEffectResult
-    context.emptyResult
-  }
-
   override def output: Seq[Attribute] = Seq.empty
 }
 
@@ -164,6 +157,4 @@ case class DescribeCommand(child: SparkPlan, output: Seq[Attribute])(
     Row("# Registered as a temporary table", null, null) +:
       child.output.map(field => Row(field.name, field.dataType.toString, null))
   }
-
-  override def execute(): RDD[Row] = context.sparkContext.parallelize(sideEffectResult, 1)
 }

sql/hive/src/main/scala/org/apache/spark/sql/hive/execution/DescribeHiveTableCommand.scala

@@ -74,7 +74,5 @@ case class DescribeHiveTableCommand(
     }
   }
 
-  override def execute(): RDD[Row] = context.sparkContext.parallelize(sideEffectResult, 1)
-
   override def otherCopyArgs = context :: Nil
 }

sql/hive/src/main/scala/org/apache/spark/sql/hive/execution/NativeCommand.scala

@@ -34,7 +34,5 @@ case class NativeCommand(
 
   override protected[sql] lazy val sideEffectResult: Seq[Row] = context.runSqlHive(sql).map(Row(_))
 
-  override def execute(): RDD[Row] = context.sparkContext.parallelize(sideEffectResult, 1)
-
   override def otherCopyArgs = context :: Nil
 }

sql/hive/src/main/scala/org/apache/spark/sql/hive/execution/commands.scala

@@ -41,11 +41,6 @@ case class AnalyzeTable(tableName: String) extends LeafNode with Command {
     hiveContext.analyze(tableName)
     Seq.empty[Row]
   }
-
-  override def execute(): RDD[Row] = {
-    sideEffectResult
-    sparkContext.emptyRDD[Row]
-  }
 }
 
 /**
@@ -54,7 +49,6 @@ case class AnalyzeTable(tableName: String) extends LeafNode with Command {
  */
 @DeveloperApi
 case class DropTable(tableName: String, ifExists: Boolean) extends LeafNode with Command {
-
   def hiveContext = sqlContext.asInstanceOf[HiveContext]
 
   def output = Seq.empty
@@ -65,9 +59,4 @@ case class DropTable(tableName: String, ifExists: Boolean) extends LeafNode with
     hiveContext.catalog.unregisterTable(None, tableName)
     Seq.empty[Row]
   }
-
-  override def execute(): RDD[Row] = {
-    sideEffectResult
-    sparkContext.emptyRDD[Row]
-  }
 }