diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/Expression.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/Expression.scala
index 3912f5f4375fdb3bfb7d3f983f895a1af5d36236..0411ce3aefda17660c36b055195bc3b8c9c744ab 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/Expression.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/Expression.scala
@@ -33,14 +33,11 @@ abstract class Expression extends TreeNode[Expression] {
* executed.
*
* The following conditions are used to determine suitability for constant folding:
- * - A [[expressions.Coalesce Coalesce]] is foldable if all of its children are foldable
- * - A [[expressions.BinaryExpression BinaryExpression]] is foldable if its both left and right
- * child are foldable
- * - A [[expressions.Not Not]], [[expressions.IsNull IsNull]], or
- * [[expressions.IsNotNull IsNotNull]] is foldable if its child is foldable.
- * - A [[expressions.Literal]] is foldable.
- * - A [[expressions.Cast Cast]] or [[expressions.UnaryMinus UnaryMinus]] is foldable if its
- * child is foldable.
+ * - A [[Coalesce]] is foldable if all of its children are foldable
+ * - A [[BinaryExpression]] is foldable if its both left and right child are foldable
+ * - A [[Not]], [[IsNull]], or [[IsNotNull]] is foldable if its child is foldable
+ * - A [[Literal]] is foldable
+ * - A [[Cast]] or [[UnaryMinus]] is foldable if its child is foldable
*/
def foldable: Boolean = false
def nullable: Boolean
@@ -58,7 +55,7 @@ abstract class Expression extends TreeNode[Expression] {
lazy val resolved: Boolean = childrenResolved
/**
- * Returns the [[types.DataType DataType]] of the result of evaluating this expression. It is
+ * Returns the [[DataType]] of the result of evaluating this expression. It is
* invalid to query the dataType of an unresolved expression (i.e., when `resolved` == false).
*/
def dataType: DataType
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/QueryPlan.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/QueryPlan.scala
index 00e2d3bc24be9ddc364096fccf2a9cae2aab1425..7b82e19b2e714002a7ccec112d0dcb51c68f3b9e 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/QueryPlan.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/QueryPlan.scala
@@ -18,7 +18,6 @@
package org.apache.spark.sql.catalyst.plans
import org.apache.spark.sql.catalyst.expressions.{Attribute, Expression}
-import org.apache.spark.sql.catalyst.plans
import org.apache.spark.sql.catalyst.trees.TreeNode
import org.apache.spark.sql.catalyst.types.{ArrayType, DataType, StructField, StructType}
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/LogicalPlan.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/LogicalPlan.scala
index 0933a31c362d8d462fd70a05150b9ffaf835c858..edc37e3877c0ed0bb4d7eb845b226e1d57c063a7 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/LogicalPlan.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/LogicalPlan.scala
@@ -41,19 +41,19 @@ abstract class LogicalPlan extends QueryPlan[LogicalPlan] {
/**
* Returns true if this expression and all its children have been resolved to a specific schema
* and false if it is still contains any unresolved placeholders. Implementations of LogicalPlan
- * can override this (e.g. [[catalyst.analysis.UnresolvedRelation UnresolvedRelation]] should
- * return `false`).
+ * can override this (e.g.
+ * [[org.apache.spark.sql.catalyst.analysis.UnresolvedRelation UnresolvedRelation]]
+ * should return `false`).
*/
lazy val resolved: Boolean = !expressions.exists(!_.resolved) && childrenResolved
/**
* Returns true if all its children of this query plan have been resolved.
*/
- def childrenResolved = !children.exists(!_.resolved)
+ def childrenResolved: Boolean = !children.exists(!_.resolved)
/**
- * Optionally resolves the given string to a
- * [[catalyst.expressions.NamedExpression NamedExpression]]. The attribute is expressed as
+ * Optionally resolves the given string to a [[NamedExpression]]. The attribute is expressed as
* as string in the following form: `[scope].AttributeName.[nested].[fields]...`.
*/
def resolve(name: String): Option[NamedExpression] = {
@@ -93,7 +93,7 @@ abstract class LeafNode extends LogicalPlan with trees.LeafNode[LogicalPlan] {
self: Product =>
// Leaf nodes by definition cannot reference any input attributes.
- def references = Set.empty
+ override def references = Set.empty
}
/**
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/basicOperators.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/basicOperators.scala
index b777cf4249196a283abcc686144fa3ac0727ef3f..3e0639867b278cd40e0d10b84ce55997c200fcf7 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/basicOperators.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/basicOperators.scala
@@ -27,7 +27,7 @@ case class Project(projectList: Seq[NamedExpression], child: LogicalPlan) extend
}
/**
- * Applies a [[catalyst.expressions.Generator Generator]] to a stream of input rows, combining the
+ * Applies a [[Generator]] to a stream of input rows, combining the
* output of each into a new stream of rows. This operation is similar to a `flatMap` in functional
* programming with one important additional feature, which allows the input rows to be joined with
* their output.
@@ -46,32 +46,32 @@ case class Generate(
child: LogicalPlan)
extends UnaryNode {
- protected def generatorOutput =
+ protected def generatorOutput: Seq[Attribute] =
alias
.map(a => generator.output.map(_.withQualifiers(a :: Nil)))
.getOrElse(generator.output)
- def output =
+ override def output =
if (join) child.output ++ generatorOutput else generatorOutput
- def references =
+ override def references =
if (join) child.outputSet else generator.references
}
case class Filter(condition: Expression, child: LogicalPlan) extends UnaryNode {
- def output = child.output
- def references = condition.references
+ override def output = child.output
+ override def references = condition.references
}
case class Union(left: LogicalPlan, right: LogicalPlan) extends BinaryNode {
// TODO: These aren't really the same attributes as nullability etc might change.
- def output = left.output
+ override def output = left.output
override lazy val resolved =
childrenResolved &&
!left.output.zip(right.output).exists { case (l,r) => l.dataType != r.dataType }
- def references = Set.empty
+ override def references = Set.empty
}
case class Join(
@@ -80,8 +80,8 @@ case class Join(
joinType: JoinType,
condition: Option[Expression]) extends BinaryNode {
- def references = condition.map(_.references).getOrElse(Set.empty)
- def output = joinType match {
+ override def references = condition.map(_.references).getOrElse(Set.empty)
+ override def output = joinType match {
case LeftSemi =>
left.output
case _ =>
@@ -96,9 +96,9 @@ case class InsertIntoTable(
overwrite: Boolean)
extends LogicalPlan {
// The table being inserted into is a child for the purposes of transformations.
- def children = table :: child :: Nil
- def references = Set.empty
- def output = child.output
+ override def children = table :: child :: Nil
+ override def references = Set.empty
+ override def output = child.output
override lazy val resolved = childrenResolved && child.output.zip(table.output).forall {
case (childAttr, tableAttr) => childAttr.dataType == tableAttr.dataType
@@ -109,20 +109,20 @@ case class InsertIntoCreatedTable(
databaseName: Option[String],
tableName: String,
child: LogicalPlan) extends UnaryNode {
- def references = Set.empty
- def output = child.output
+ override def references = Set.empty
+ override def output = child.output
}
case class WriteToFile(
path: String,
child: LogicalPlan) extends UnaryNode {
- def references = Set.empty
- def output = child.output
+ override def references = Set.empty
+ override def output = child.output
}
case class Sort(order: Seq[SortOrder], child: LogicalPlan) extends UnaryNode {
- def output = child.output
- def references = order.flatMap(_.references).toSet
+ override def output = child.output
+ override def references = order.flatMap(_.references).toSet
}
case class Aggregate(
@@ -131,18 +131,19 @@ case class Aggregate(
child: LogicalPlan)
extends UnaryNode {
- def output = aggregateExpressions.map(_.toAttribute)
- def references = (groupingExpressions ++ aggregateExpressions).flatMap(_.references).toSet
+ override def output = aggregateExpressions.map(_.toAttribute)
+ override def references =
+ (groupingExpressions ++ aggregateExpressions).flatMap(_.references).toSet
}
case class Limit(limitExpr: Expression, child: LogicalPlan) extends UnaryNode {
- def output = child.output
- def references = limitExpr.references
+ override def output = child.output
+ override def references = limitExpr.references
}
case class Subquery(alias: String, child: LogicalPlan) extends UnaryNode {
- def output = child.output.map(_.withQualifiers(alias :: Nil))
- def references = Set.empty
+ override def output = child.output.map(_.withQualifiers(alias :: Nil))
+ override def references = Set.empty
}
/**
@@ -159,7 +160,7 @@ case class LowerCaseSchema(child: LogicalPlan) extends UnaryNode {
case otherType => otherType
}
- val output = child.output.map {
+ override val output = child.output.map {
case a: AttributeReference =>
AttributeReference(
a.name.toLowerCase,
@@ -170,21 +171,21 @@ case class LowerCaseSchema(child: LogicalPlan) extends UnaryNode {
case other => other
}
- def references = Set.empty
+ override def references = Set.empty
}
case class Sample(fraction: Double, withReplacement: Boolean, seed: Long, child: LogicalPlan)
extends UnaryNode {
- def output = child.output
- def references = Set.empty
+ override def output = child.output
+ override def references = Set.empty
}
case class Distinct(child: LogicalPlan) extends UnaryNode {
- def output = child.output
- def references = child.outputSet
+ override def output = child.output
+ override def references = child.outputSet
}
case object NoRelation extends LeafNode {
- def output = Nil
+ override def output = Nil
}
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/physical/partitioning.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/physical/partitioning.scala
index ffb3a92f8f3402f18bc87d2fa08076bb86557046..4bb022cf238af5190c28a9c76815bcb29c8b5e90 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/physical/partitioning.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/physical/partitioning.scala
@@ -46,7 +46,7 @@ case object AllTuples extends Distribution
/**
* Represents data where tuples that share the same values for the `clustering`
- * [[catalyst.expressions.Expression Expressions]] will be co-located. Based on the context, this
+ * [[Expression Expressions]] will be co-located. Based on the context, this
* can mean such tuples are either co-located in the same partition or they will be contiguous
* within a single partition.
*/
@@ -60,7 +60,7 @@ case class ClusteredDistribution(clustering: Seq[Expression]) extends Distributi
/**
* Represents data where tuples have been ordered according to the `ordering`
- * [[catalyst.expressions.Expression Expressions]]. This is a strictly stronger guarantee than
+ * [[Expression Expressions]]. This is a strictly stronger guarantee than
* [[ClusteredDistribution]] as an ordering will ensure that tuples that share the same value for
* the ordering expressions are contiguous and will never be split across partitions.
*/
@@ -79,19 +79,17 @@ sealed trait Partitioning {
val numPartitions: Int
/**
- * Returns true iff the guarantees made by this
- * [[catalyst.plans.physical.Partitioning Partitioning]] are sufficient to satisfy
- * the partitioning scheme mandated by the `required`
- * [[catalyst.plans.physical.Distribution Distribution]], i.e. the current dataset does not
- * need to be re-partitioned for the `required` Distribution (it is possible that tuples within
- * a partition need to be reorganized).
+ * Returns true iff the guarantees made by this [[Partitioning]] are sufficient
+ * to satisfy the partitioning scheme mandated by the `required` [[Distribution]],
+ * i.e. the current dataset does not need to be re-partitioned for the `required`
+ * Distribution (it is possible that tuples within a partition need to be reorganized).
*/
def satisfies(required: Distribution): Boolean
/**
* Returns true iff all distribution guarantees made by this partitioning can also be made
* for the `other` specified partitioning.
- * For example, two [[catalyst.plans.physical.HashPartitioning HashPartitioning]]s are
+ * For example, two [[HashPartitioning HashPartitioning]]s are
* only compatible if the `numPartitions` of them is the same.
*/
def compatibleWith(other: Partitioning): Boolean
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/Exchange.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/Exchange.scala
index 05dfb85b38b02750ac3b07c7d1e32dab62413e04..f46fa0516566f018313d7a70241988cb35e7496b 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/Exchange.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/Exchange.scala
@@ -20,7 +20,7 @@ package org.apache.spark.sql.execution
import org.apache.spark.annotation.DeveloperApi
import org.apache.spark.{HashPartitioner, RangePartitioner, SparkConf}
import org.apache.spark.rdd.ShuffledRDD
-import org.apache.spark.sql.{SQLConf, SQLContext, Row}
+import org.apache.spark.sql.{SQLContext, Row}
import org.apache.spark.sql.catalyst.errors.attachTree
import org.apache.spark.sql.catalyst.expressions.{NoBind, MutableProjection, RowOrdering}
import org.apache.spark.sql.catalyst.plans.physical._
@@ -82,9 +82,10 @@ case class Exchange(newPartitioning: Partitioning, child: SparkPlan) extends Una
}
/**
- * Ensures that the [[catalyst.plans.physical.Partitioning Partitioning]] of input data meets the
- * [[catalyst.plans.physical.Distribution Distribution]] requirements for each operator by inserting
- * [[Exchange]] Operators where required.
+ * Ensures that the [[org.apache.spark.sql.catalyst.plans.physical.Partitioning Partitioning]]
+ * of input data meets the
+ * [[org.apache.spark.sql.catalyst.plans.physical.Distribution Distribution]] requirements for
+ * each operator by inserting [[Exchange]] Operators where required.
*/
private[sql] case class AddExchange(sqlContext: SQLContext) extends Rule[SparkPlan] {
// TODO: Determine the number of partitions.