1. Move sql package object and package-info to sql-core.

2. Minor updates on APIs.
3. Update scala doc.

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/BoundAttribute.scala

@@ -17,11 +17,12 @@
 
 package org.apache.spark.sql.catalyst.expressions
 
+import com.typesafe.scalalogging.slf4j.Logging
+
 import org.apache.spark.sql.catalyst.trees
 import org.apache.spark.sql.catalyst.errors.attachTree
 import org.apache.spark.sql.catalyst.plans.QueryPlan
 import org.apache.spark.sql.catalyst.rules.Rule
-import org.apache.spark.sql.Logging
 
 /**
  * A bound reference points to a specific slot in the input tuple, allowing the actual value

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/Row.scala

@@ -34,7 +34,7 @@ object Row {
   def unapplySeq(row: Row): Some[Seq[Any]] = Some(row)
 
   /**
-   * Construct a [[Row]] with the given values.
+   * This method can be used to construct a [[Row]] with the given values.
    */
   def apply(values: Any*): Row = new GenericRow(values.toArray)
 }

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/planning/QueryPlanner.scala

@@ -17,7 +17,8 @@
 
 package org.apache.spark.sql.catalyst.planning
 
-import org.apache.spark.sql.Logging
+import com.typesafe.scalalogging.slf4j.Logging
+
 import org.apache.spark.sql.catalyst.plans.logical.LogicalPlan
 import org.apache.spark.sql.catalyst.trees.TreeNode
 

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/planning/patterns.scala

@@ -19,7 +19,7 @@ package org.apache.spark.sql.catalyst.planning
 
 import scala.annotation.tailrec
 
-import org.apache.spark.sql.Logging
+import com.typesafe.scalalogging.slf4j.Logging
 
 import org.apache.spark.sql.catalyst.expressions._
 import org.apache.spark.sql.catalyst.plans._

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/QueryPlan.scala

@@ -128,7 +128,7 @@ abstract class QueryPlan[PlanType <: TreeNode[PlanType]] extends TreeNode[PlanTy
   def schema: StructType = StructType.fromAttributes(output)
 
   /** Returns the output schema in the tree format. */
-  def schemaString: String = schema.schemaString
+  def schemaString: String = schema.structString
 
   /** Prints out the schema in the tree format */
   def printSchema(): Unit = println(schemaString)

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/rules/Rule.scala

@@ -17,7 +17,8 @@
 
 package org.apache.spark.sql.catalyst.rules
 
-import org.apache.spark.sql.Logging
+import com.typesafe.scalalogging.slf4j.Logging
+
 import org.apache.spark.sql.catalyst.trees.TreeNode
 
 abstract class Rule[TreeType <: TreeNode[_]] extends Logging {

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/rules/RuleExecutor.scala

@@ -15,9 +15,9 @@
  * limitations under the License.
  */
 
-package org.apache.spark.sql
-package catalyst
-package rules
+package org.apache.spark.sql.catalyst.rules
+
+import com.typesafe.scalalogging.slf4j.Logging
 
 import org.apache.spark.sql.catalyst.trees.TreeNode
 import org.apache.spark.sql.catalyst.util.sideBySide

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/trees/package.scala

@@ -17,8 +17,6 @@
 
 package org.apache.spark.sql.catalyst
 
-import org.apache.spark.sql.Logger
-
 /**
  * A library for easily manipulating trees of operators.  Operators that extend TreeNode are
  * granted the following interface:
@@ -35,5 +33,6 @@ import org.apache.spark.sql.Logger
  */
 package object trees {
   // Since we want tree nodes to be lightweight, we create one logger for all treenode instances.
-  protected val logger = Logger("catalyst.trees")
+  protected val logger =
+    com.typesafe.scalalogging.slf4j.Logger(org.slf4j.LoggerFactory.getLogger("catalyst.trees"))
 }

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/types/dataTypes.scala

@@ -30,7 +30,7 @@ import org.apache.spark.util.Utils
 /**
  *
  */
-object DataType extends RegexParsers {
+protected[sql] object DataType extends RegexParsers {
   protected lazy val primitiveType: Parser[DataType] =
     "StringType" ^^^ StringType |
     "FloatType" ^^^ FloatType |
@@ -84,6 +84,21 @@ object DataType extends RegexParsers {
     case Success(result, _) => result
     case failure: NoSuccess => sys.error(s"Unsupported dataType: $asString, $failure")
   }
+
+  protected[types] def buildFormattedString(
+      dataType: DataType,
+      prefix: String,
+      builder: StringBuilder): Unit = {
+    dataType match {
+      case array: ArrayType =>
+        array.buildFormattedString(prefix, builder)
+      case struct: StructType =>
+        struct.buildFormattedString(prefix, builder)
+      case map: MapType =>
+        map.buildFormattedString(prefix, builder)
+      case _ =>
+    }
+  }
 }
 
 abstract class DataType {
@@ -244,22 +259,15 @@ case object FloatType extends FractionalType {
 }
 
 object ArrayType {
+  /** Construct a [[ArrayType]] object with the given element type. The `containsNull` is false. */
   def apply(elementType: DataType): ArrayType = ArrayType(elementType, false)
 }
 
 case class ArrayType(elementType: DataType, containsNull: Boolean) extends DataType {
   private[sql] def buildFormattedString(prefix: String, builder: StringBuilder): Unit = {
     builder.append(
       s"${prefix}-- element: ${elementType.simpleString} (containsNull = ${containsNull})\n")
-    elementType match {
-      case array: ArrayType =>
-        array.buildFormattedString(s"$prefix    |", builder)
-      case struct: StructType =>
-        struct.buildFormattedString(s"$prefix    |", builder)
-      case map: MapType =>
-        map.buildFormattedString(s"$prefix    |", builder)
-      case _ =>
-    }
+    DataType.buildFormattedString(elementType, s"$prefix    |", builder)
   }
 
   def simpleString: String = "array"
@@ -269,48 +277,41 @@ case class StructField(name: String, dataType: DataType, nullable: Boolean) {
 
   private[sql] def buildFormattedString(prefix: String, builder: StringBuilder): Unit = {
     builder.append(s"${prefix}-- ${name}: ${dataType.simpleString} (nullable = ${nullable})\n")
-    dataType match {
-      case array: ArrayType =>
-        array.buildFormattedString(s"$prefix    |", builder)
-      case struct: StructType =>
-        struct.buildFormattedString(s"$prefix    |", builder)
-      case map: MapType =>
-        map.buildFormattedString(s"$prefix    |", builder)
-      case _ =>
-    }
+    DataType.buildFormattedString(dataType, s"$prefix    |", builder)
   }
 }
 
 object StructType {
-  def fromAttributes(attributes: Seq[Attribute]): StructType =
+  protected[sql] def fromAttributes(attributes: Seq[Attribute]): StructType =
     StructType(attributes.map(a => StructField(a.name, a.dataType, a.nullable)))
 
   private def validateFields(fields: Seq[StructField]): Boolean =
     fields.map(field => field.name).distinct.size == fields.size
-
-  def apply[A <: String: ClassTag, B <: DataType: ClassTag](fields: (A, B)*): StructType =
-    StructType(fields.map(field => StructField(field._1, field._2, true)))
-
-  def apply[A <: String: ClassTag, B <: DataType: ClassTag, C <: Boolean: ClassTag](
-      fields: (A, B, C)*): StructType =
-    StructType(fields.map(field => StructField(field._1, field._2, field._3)))
 }
 
 case class StructType(fields: Seq[StructField]) extends DataType {
   require(StructType.validateFields(fields), "Found fields with the same name.")
 
+  /**
+   * Extracts a [[StructField]] of the given name. If the [[StructType]] object does not
+   * have a name matching the given name, `null` will be returned.
+   */
   def apply(name: String): StructField = {
     fields.find(f => f.name == name).orNull
   }
 
-  def apply(names: String*): StructType = {
-    val nameSet = names.toSet
-    StructType(fields.filter(f => nameSet.contains(f.name)))
+  /**
+   * Returns a [[StructType]] containing [[StructField]]s of the given names.
+   * Those names which do not have matching fields will be ignored.
+   */
+  def apply(names: Set[String]): StructType = {
+    StructType(fields.filter(f => names.contains(f.name)))
   }
 
-  def toAttributes = fields.map(f => AttributeReference(f.name, f.dataType, f.nullable)())
+  protected[sql] def toAttributes =
+    fields.map(f => AttributeReference(f.name, f.dataType, f.nullable)())
 
-  def schemaString: String = {
+  def structString: String = {
     val builder = new StringBuilder
     builder.append("root\n")
     val prefix = " |"
@@ -319,7 +320,7 @@ case class StructType(fields: Seq[StructField]) extends DataType {
     builder.toString()
   }
 
-  def printSchema(): Unit = println(schemaString)
+  def printStruct(): Unit = println(structString)
 
   private[sql] def buildFormattedString(prefix: String, builder: StringBuilder): Unit = {
     fields.foreach(field => field.buildFormattedString(prefix, builder))
@@ -331,26 +332,8 @@ case class StructType(fields: Seq[StructField]) extends DataType {
 case class MapType(keyType: DataType, valueType: DataType) extends DataType {
   private[sql] def buildFormattedString(prefix: String, builder: StringBuilder): Unit = {
     builder.append(s"${prefix}-- key: ${keyType.simpleString}\n")
-    keyType match {
-      case array: ArrayType =>
-        array.buildFormattedString(s"$prefix    |", builder)
-      case struct: StructType =>
-        struct.buildFormattedString(s"$prefix    |", builder)
-      case map: MapType =>
-        map.buildFormattedString(s"$prefix    |", builder)
-      case _ =>
-    }
-
-    builder.append(s"${prefix}-- value: ${valueType.simpleString}\n")
-    valueType match {
-      case array: ArrayType =>
-        array.buildFormattedString(s"$prefix    |", builder)
-      case struct: StructType =>
-        struct.buildFormattedString(s"$prefix    |", builder)
-      case map: MapType =>
-        map.buildFormattedString(s"$prefix    |", builder)
-      case _ =>
-    }
+    DataType.buildFormattedString(keyType, s"$prefix    |", builder)
+    DataType.buildFormattedString(valueType, s"$prefix    |", builder)
   }
 
   def simpleString: String = "map"