Fixed coding style issues in sql/core

Author: liancheng

sql/core/src/main/scala/org/apache/spark/rdd/PartitionLocalRDDFunctions.scala

@@ -23,7 +23,6 @@ import scala.reflect._
 import scala.collection.mutable.ArrayBuffer
 
 import org.apache.spark._
-import org.apache.spark.Aggregator
 import org.apache.spark.SparkContext._
 import org.apache.spark.util.collection.AppendOnlyMap
 

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

@@ -20,14 +20,13 @@ package org.apache.spark.sql
 import scala.language.implicitConversions
 import scala.reflect.runtime.universe.TypeTag
 
-import org.apache.spark.{SparkContext, SparkConf}
+import org.apache.spark.SparkContext
 import org.apache.spark.rdd.RDD
 import org.apache.spark.sql.catalyst.analysis._
 import org.apache.spark.sql.catalyst.dsl
 import org.apache.spark.sql.catalyst.expressions._
 import org.apache.spark.sql.catalyst.optimizer.Optimizer
-import org.apache.spark.sql.catalyst.planning.QueryPlanner
-import org.apache.spark.sql.catalyst.plans.logical.{LogicalPlan, NativeCommand, WriteToFile}
+import org.apache.spark.sql.catalyst.plans.logical.LogicalPlan
 import org.apache.spark.sql.catalyst.rules.RuleExecutor
 import org.apache.spark.sql.execution._
 

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

@@ -17,13 +17,13 @@
 
 package org.apache.spark.sql
 
-import org.apache.spark.{OneToOneDependency, Dependency, Partition, TaskContext}
 import org.apache.spark.rdd.RDD
 import org.apache.spark.sql.catalyst.analysis._
 import org.apache.spark.sql.catalyst.expressions._
-import org.apache.spark.sql.catalyst.plans.{Inner, JoinType}
 import org.apache.spark.sql.catalyst.plans.logical._
+import org.apache.spark.sql.catalyst.plans.{Inner, JoinType}
 import org.apache.spark.sql.catalyst.types.BooleanType
+import org.apache.spark.{OneToOneDependency, Dependency, Partition, TaskContext}
 
 /**
  * <span class="badge" style="float: right; background-color: darkblue;">ALPHA COMPONENT</span>

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

@@ -18,7 +18,6 @@
 package org.apache.spark.sql
 package execution
 
-import java.nio.ByteBuffer
 
 import com.esotericsoftware.kryo.{Kryo, Serializer}
 import com.esotericsoftware.kryo.io.{Output, Input}
@@ -28,16 +27,16 @@ import org.apache.spark.rdd.ShuffledRDD
 import org.apache.spark.serializer.KryoSerializer
 import org.apache.spark.util.MutablePair
 
-import catalyst.rules.Rule
-import catalyst.errors._
-import catalyst.expressions._
-import catalyst.plans.physical._
+import org.apache.spark.sql.catalyst.rules.Rule
+import org.apache.spark.sql.catalyst.errors._
+import org.apache.spark.sql.catalyst.expressions._
+import org.apache.spark.sql.catalyst.plans.physical._
 
 private class SparkSqlSerializer(conf: SparkConf) extends KryoSerializer(conf) {
   override def newKryo(): Kryo = {
     val kryo = new Kryo
     kryo.setRegistrationRequired(true)
-    kryo.register(classOf[MutablePair[_,_]])
+    kryo.register(classOf[MutablePair[_, _]])
     kryo.register(classOf[Array[Any]])
     kryo.register(classOf[org.apache.spark.sql.catalyst.expressions.GenericRow])
     kryo.register(classOf[org.apache.spark.sql.catalyst.expressions.GenericMutableRow])
@@ -52,7 +51,7 @@ private class SparkSqlSerializer(conf: SparkConf) extends KryoSerializer(conf) {
 private class BigDecimalSerializer extends Serializer[BigDecimal] {
   def write(kryo: Kryo, output: Output, bd: math.BigDecimal) {
     // TODO: There are probably more efficient representations than strings...
-    output.writeString(bd.toString)
+    output.writeString(bd.toString())
   }
 
   def read(kryo: Kryo, input: Input, tpe: Class[BigDecimal]): BigDecimal = {
@@ -68,7 +67,7 @@ case class Exchange(newPartitioning: Partitioning, child: SparkPlan) extends Una
 
   def execute() = attachTree(this , "execute") {
     newPartitioning match {
-      case HashPartitioning(expressions, numPartitions) => {
+      case HashPartitioning(expressions, numPartitions) =>
         // TODO: Eliminate redundant expressions in grouping key and value.
         val rdd = child.execute().mapPartitions { iter =>
           val hashExpressions = new MutableProjection(expressions)
@@ -79,8 +78,8 @@ case class Exchange(newPartitioning: Partitioning, child: SparkPlan) extends Una
         val shuffled = new ShuffledRDD[Row, Row, MutablePair[Row, Row]](rdd, part)
         shuffled.setSerializer(new SparkSqlSerializer(new SparkConf(false)))
         shuffled.map(_._2)
-      }
-      case RangePartitioning(sortingExpressions, numPartitions) => {
+
+      case RangePartitioning(sortingExpressions, numPartitions) =>
         // TODO: RangePartitioner should take an Ordering.
         implicit val ordering = new RowOrdering(sortingExpressions)
 
@@ -93,9 +92,8 @@ case class Exchange(newPartitioning: Partitioning, child: SparkPlan) extends Una
         shuffled.setSerializer(new SparkSqlSerializer(new SparkConf(false)))
 
         shuffled.map(_._1)
-      }
       case SinglePartition =>
-        child.execute().coalesce(1, true)
+        child.execute().coalesce(1, shuffle = true)
 
       case _ => sys.error(s"Exchange not implemented for $newPartitioning")
       // TODO: Handle BroadcastPartitioning.

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

@@ -18,8 +18,7 @@
 package org.apache.spark.sql
 package execution
 
-import catalyst.expressions._
-import catalyst.types._
+import org.apache.spark.sql.catalyst.expressions._
 
 /**
  * Applies a [[catalyst.expressions.Generator Generator]] to a stream of input rows, combining the

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

@@ -20,13 +20,13 @@ package execution
 
 import org.apache.spark.SparkContext
 
-import catalyst.expressions._
-import catalyst.planning._
-import catalyst.plans._
-import catalyst.plans.logical.LogicalPlan
-import catalyst.plans.physical._
-import parquet.ParquetRelation
-import parquet.InsertIntoParquetTable
+import org.apache.spark.sql.catalyst.expressions._
+import org.apache.spark.sql.catalyst.planning._
+import org.apache.spark.sql.catalyst.plans._
+import org.apache.spark.sql.catalyst.plans.logical.LogicalPlan
+import org.apache.spark.sql.catalyst.plans.physical._
+import org.apache.spark.sql.parquet.ParquetRelation
+import org.apache.spark.sql.parquet.InsertIntoParquetTable
 
 abstract class SparkStrategies extends QueryPlanner[SparkPlan] {
 
@@ -172,7 +172,7 @@ abstract class SparkStrategies extends QueryPlanner[SparkPlan] {
 
   // Can we automate these 'pass through' operations?
   object BasicOperators extends Strategy {
-    // TOOD: Set
+    // TODO: Set
     val numPartitions = 200
     def apply(plan: LogicalPlan): Seq[SparkPlan] = plan match {
       case logical.Distinct(child) =>

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

@@ -19,13 +19,10 @@ package org.apache.spark.sql
 package execution
 
 import org.apache.spark.SparkContext
-
-import catalyst.errors._
-import catalyst.expressions._
-import catalyst.plans.physical.{UnspecifiedDistribution, ClusteredDistribution, AllTuples}
-import catalyst.types._
-
 import org.apache.spark.rdd.PartitionLocalRDDFunctions._
+import org.apache.spark.sql.catalyst.errors._
+import org.apache.spark.sql.catalyst.expressions._
+import org.apache.spark.sql.catalyst.plans.physical._
 
 /**
  * Groups input data by `groupingExpressions` and computes the `aggregateExpressions` for each

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

@@ -23,18 +23,17 @@ import scala.reflect.runtime.universe.TypeTag
 import org.apache.spark.rdd.RDD
 import org.apache.spark.SparkContext
 
-import catalyst.errors._
-import catalyst.expressions._
-import catalyst.plans.physical.{UnspecifiedDistribution, OrderedDistribution}
-import catalyst.plans.logical.LogicalPlan
-import catalyst.ScalaReflection
+import org.apache.spark.sql.catalyst.errors._
+import org.apache.spark.sql.catalyst.expressions._
+import org.apache.spark.sql.catalyst.plans.physical.{UnspecifiedDistribution, OrderedDistribution}
+import org.apache.spark.sql.catalyst.ScalaReflection
 
 case class Project(projectList: Seq[NamedExpression], child: SparkPlan) extends UnaryNode {
   def output = projectList.map(_.toAttribute)
 
   def execute() = child.execute().mapPartitions { iter =>
-    @transient val resuableProjection = new MutableProjection(projectList)
-    iter.map(resuableProjection)
+    @transient val reusableProjection = new MutableProjection(projectList)
+    iter.map(reusableProjection)
   }
 }
 

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

@@ -23,10 +23,10 @@ import scala.collection.mutable
 import org.apache.spark.rdd.RDD
 import org.apache.spark.SparkContext
 
-import catalyst.errors._
-import catalyst.expressions._
-import catalyst.plans._
-import catalyst.plans.physical.{ClusteredDistribution, Partitioning}
+import org.apache.spark.sql.catalyst.errors._
+import org.apache.spark.sql.catalyst.expressions._
+import org.apache.spark.sql.catalyst.plans._
+import org.apache.spark.sql.catalyst.plans.physical.{ClusteredDistribution, Partitioning}
 
 import org.apache.spark.rdd.PartitionLocalRDDFunctions._
 

sql/core/src/main/scala/org/apache/spark/sql/parquet/ParquetRelation.scala

@@ -19,28 +19,28 @@ package org.apache.spark.sql.parquet
 
 import java.io.{IOException, FileNotFoundException}
 
-import org.apache.hadoop.fs.{Path, FileSystem}
+import scala.collection.JavaConversions._
+
 import org.apache.hadoop.conf.Configuration
-import org.apache.hadoop.mapreduce.Job
 import org.apache.hadoop.fs.permission.FsAction
+import org.apache.hadoop.fs.{Path, FileSystem}
+import org.apache.hadoop.mapreduce.Job
 
-import org.apache.spark.sql.catalyst.plans.logical.{LogicalPlan, BaseRelation}
-import org.apache.spark.sql.catalyst.types._
-import org.apache.spark.sql.catalyst.types.ArrayType
-import org.apache.spark.sql.catalyst.expressions.{Row, AttributeReference, Attribute}
-import org.apache.spark.sql.catalyst.analysis.UnresolvedException
-
-import parquet.schema.{MessageTypeParser, MessageType}
+import parquet.hadoop.metadata.{FileMetaData, ParquetMetadata}
+import parquet.hadoop.util.ContextUtil
+import parquet.hadoop.{Footer, ParquetFileWriter, ParquetFileReader}
+import parquet.io.api.{Binary, RecordConsumer}
 import parquet.schema.PrimitiveType.{PrimitiveTypeName => ParquetPrimitiveTypeName}
+import parquet.schema.Type.Repetition
+import parquet.schema.{MessageTypeParser, MessageType}
 import parquet.schema.{PrimitiveType => ParquetPrimitiveType}
 import parquet.schema.{Type => ParquetType}
-import parquet.schema.Type.Repetition
-import parquet.io.api.{Binary, RecordConsumer}
-import parquet.hadoop.{Footer, ParquetFileWriter, ParquetFileReader}
-import parquet.hadoop.metadata.{FileMetaData, ParquetMetadata}
-import parquet.hadoop.util.ContextUtil
 
-import scala.collection.JavaConversions._
+import org.apache.spark.sql.catalyst.analysis.UnresolvedException
+import org.apache.spark.sql.catalyst.expressions.{Row, AttributeReference, Attribute}
+import org.apache.spark.sql.catalyst.plans.logical.{LogicalPlan, BaseRelation}
+import org.apache.spark.sql.catalyst.types.ArrayType
+import org.apache.spark.sql.catalyst.types._
 
 /**
  * Relation that consists of data stored in a Parquet columnar format.
@@ -55,7 +55,7 @@ import scala.collection.JavaConversions._
  * @param tableName The name of the relation that can be used in queries.
  * @param path The path to the Parquet file.
  */
-case class ParquetRelation(val tableName: String, val path: String) extends BaseRelation {
+case class ParquetRelation(tableName: String, path: String) extends BaseRelation {
 
   /** Schema derived from ParquetFile **/
   def parquetSchema: MessageType =
@@ -145,11 +145,10 @@ object ParquetTypesConverter {
     case ParquetPrimitiveTypeName.FLOAT => FloatType
     case ParquetPrimitiveTypeName.INT32 => IntegerType
     case ParquetPrimitiveTypeName.INT64 => LongType
-    case ParquetPrimitiveTypeName.INT96 => {
+    case ParquetPrimitiveTypeName.INT96 =>
       // TODO: add BigInteger type? TODO(andre) use DecimalType instead????
       sys.error("Warning: potential loss of precision: converting INT96 to long")
       LongType
-    }
     case _ => sys.error(
       s"Unsupported parquet datatype $parquetType")
   }
@@ -186,11 +185,10 @@ object ParquetTypesConverter {
 
   def convertToAttributes(parquetSchema: MessageType) : Seq[Attribute] = {
     parquetSchema.getColumns.map {
-      case (desc) => {
+      case (desc) =>
         val ctype = toDataType(desc.getType)
         val name: String = desc.getPath.mkString(".")
         new AttributeReference(name, ctype, false)()
-      }
     }
   }
 
@@ -245,7 +243,7 @@ object ParquetTypesConverter {
    * Try to read Parquet metadata at the given Path. We first see if there is a summary file
    * in the parent directory. If so, this is used. Else we read the actual footer at the given
    * location.
-   * @param path The path at which we expect one (or more) Parquet files.
+   * @param origPath The path at which we expect one (or more) Parquet files.
    * @return The `ParquetMetadata` containing among other things the schema.
    */
   def readMetaData(origPath: Path): ParquetMetadata = {

sql/core/src/main/scala/org/apache/spark/sql/parquet/ParquetTableOperations.scala

@@ -17,24 +17,24 @@
 
 package org.apache.spark.sql.parquet
 
+import java.io.IOException
+import java.text.SimpleDateFormat
+import java.util.Date
+
+import org.apache.hadoop.conf.Configuration
+import org.apache.hadoop.fs.Path
+import org.apache.hadoop.mapreduce._
+import org.apache.hadoop.mapreduce.lib.output.{FileOutputFormat => NewFileOutputFormat}
+
+import parquet.hadoop.util.ContextUtil
+import parquet.hadoop.{ParquetOutputFormat, ParquetInputFormat}
 import parquet.io.InvalidRecordException
 import parquet.schema.MessageType
-import parquet.hadoop.{ParquetOutputFormat, ParquetInputFormat}
-import parquet.hadoop.util.ContextUtil
 
 import org.apache.spark.rdd.RDD
-import org.apache.spark.{TaskContext, SerializableWritable, SparkContext}
 import org.apache.spark.sql.catalyst.expressions.{Row, Attribute, Expression}
 import org.apache.spark.sql.execution.{SparkPlan, UnaryNode, LeafNode}
-
-import org.apache.hadoop.mapreduce.lib.output.{FileOutputFormat => NewFileOutputFormat}
-import org.apache.hadoop.mapreduce._
-import org.apache.hadoop.conf.Configuration
-import org.apache.hadoop.fs.Path
-
-import java.io.IOException
-import java.text.SimpleDateFormat
-import java.util.Date
+import org.apache.spark.{TaskContext, SerializableWritable, SparkContext}
 
 /**
  * Parquet table scan operator. Imports the file that backs the given

sql/core/src/main/scala/org/apache/spark/sql/parquet/ParquetTableSupport.scala

@@ -19,15 +19,14 @@ package org.apache.spark.sql.parquet
 
 import org.apache.hadoop.conf.Configuration
 
-import org.apache.spark.Logging
-
+import parquet.column.ParquetProperties
+import parquet.hadoop.ParquetOutputFormat
+import parquet.hadoop.api.ReadSupport.ReadContext
+import parquet.hadoop.api.{WriteSupport, ReadSupport}
 import parquet.io.api._
 import parquet.schema.{MessageTypeParser, MessageType}
-import parquet.hadoop.api.{WriteSupport, ReadSupport}
-import parquet.hadoop.api.ReadSupport.ReadContext
-import parquet.hadoop.ParquetOutputFormat
-import parquet.column.ParquetProperties
 
+import org.apache.spark.Logging
 import org.apache.spark.sql.catalyst.expressions.{Row, Attribute}
 import org.apache.spark.sql.catalyst.types._
 
@@ -95,8 +94,7 @@ class RowWriteSupport extends WriteSupport[Row] with Logging {
   }
 
   def getSchema(configuration: Configuration): MessageType = {
-    return MessageTypeParser.parseMessageType(
-      configuration.get(RowWriteSupport.PARQUET_ROW_SCHEMA))
+    MessageTypeParser.parseMessageType(configuration.get(RowWriteSupport.PARQUET_ROW_SCHEMA))
   }
 
   private var schema: MessageType = null
@@ -108,7 +106,7 @@ class RowWriteSupport extends WriteSupport[Row] with Logging {
     attributes = ParquetTypesConverter.convertToAttributes(schema)
     new WriteSupport.WriteContext(
       schema,
-      new java.util.HashMap[java.lang.String, java.lang.String]());
+      new java.util.HashMap[java.lang.String, java.lang.String]())
   }
 
   override def prepareForWrite(recordConsumer: RecordConsumer): Unit = {