From 3339e6f674264e387ab6c4f9c356bd3bc7167221 Mon Sep 17 00:00:00 2001 From: Andrew Or Date: Fri, 4 Sep 2015 15:20:20 -0700 Subject: [SPARK-10450] [SQL] Minor improvements to readability / style / typos etc. Author: Andrew Or Closes #8603 from andrewor14/minor-sql-changes. --- .../spark/sql/catalyst/expressions/predicates.scala | 2 +- .../spark/sql/catalyst/planning/QueryPlanner.scala | 2 +- .../apache/spark/sql/catalyst/planning/patterns.scala | 8 ++++---- .../org/apache/spark/sql/execution/SparkStrategies.scala | 16 ++++++++-------- .../apache/spark/sql/execution/joins/SortMergeJoin.scala | 2 +- 5 files changed, 15 insertions(+), 15 deletions(-) (limited to 'sql') diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/predicates.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/predicates.scala index 65706dba7d..daefc016bc 100644 --- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/predicates.scala +++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/predicates.scala @@ -67,7 +67,7 @@ trait PredicateHelper { /** * Returns true if `expr` can be evaluated using only the output of `plan`. This method - * can be used to determine when is is acceptable to move expression evaluation within a query + * can be used to determine when it is acceptable to move expression evaluation within a query * plan. * * For example consider a join between two relations R(a, b) and S(c, d). diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/planning/QueryPlanner.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/planning/QueryPlanner.scala index 73a21884a4..56a3dd02f9 100644 --- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/planning/QueryPlanner.scala +++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/planning/QueryPlanner.scala @@ -51,7 +51,7 @@ abstract class QueryPlanner[PhysicalPlan <: TreeNode[PhysicalPlan]] { * filled in automatically by the QueryPlanner using the other execution strategies that are * available. */ - protected def planLater(plan: LogicalPlan) = this.plan(plan).next() + protected def planLater(plan: LogicalPlan): PhysicalPlan = this.plan(plan).next() def plan(plan: LogicalPlan): Iterator[PhysicalPlan] = { // Obviously a lot to do here still... diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/planning/patterns.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/planning/patterns.scala index b9ca712c1e..e8abcd63f7 100644 --- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/planning/patterns.scala +++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/planning/patterns.scala @@ -195,8 +195,9 @@ object ExtractEquiJoinKeys extends Logging with PredicateHelper { // as join keys. val (joinPredicates, otherPredicates) = condition.map(splitConjunctivePredicates).getOrElse(Nil).partition { - case EqualTo(l, r) if (canEvaluate(l, left) && canEvaluate(r, right)) || - (canEvaluate(l, right) && canEvaluate(r, left)) => true + case EqualTo(l, r) => + (canEvaluate(l, left) && canEvaluate(r, right)) || + (canEvaluate(l, right) && canEvaluate(r, left)) case _ => false } @@ -204,10 +205,9 @@ object ExtractEquiJoinKeys extends Logging with PredicateHelper { case EqualTo(l, r) if canEvaluate(l, left) && canEvaluate(r, right) => (l, r) case EqualTo(l, r) if canEvaluate(l, right) && canEvaluate(r, left) => (r, l) } - val leftKeys = joinKeys.map(_._1) - val rightKeys = joinKeys.map(_._2) if (joinKeys.nonEmpty) { + val (leftKeys, rightKeys) = joinKeys.unzip logDebug(s"leftKeys:$leftKeys | rightKeys:$rightKeys") Some((joinType, leftKeys, rightKeys, otherPredicates.reduceOption(And), left, right)) } else { diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/SparkStrategies.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/SparkStrategies.scala index 4df53687a0..2170bc73a0 100644 --- a/sql/core/src/main/scala/org/apache/spark/sql/execution/SparkStrategies.scala +++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/SparkStrategies.scala @@ -87,7 +87,7 @@ private[sql] abstract class SparkStrategies extends QueryPlanner[SparkPlan] { left: LogicalPlan, right: LogicalPlan, condition: Option[Expression], - side: joins.BuildSide) = { + side: joins.BuildSide): Seq[SparkPlan] = { val broadcastHashJoin = execution.joins.BroadcastHashJoin( leftKeys, rightKeys, side, planLater(left), planLater(right)) condition.map(Filter(_, broadcastHashJoin)).getOrElse(broadcastHashJoin) :: Nil @@ -123,12 +123,12 @@ private[sql] abstract class SparkStrategies extends QueryPlanner[SparkPlan] { // --- Outer joins -------------------------------------------------------------------------- case ExtractEquiJoinKeys( - LeftOuter, leftKeys, rightKeys, condition, left, CanBroadcast(right)) => + LeftOuter, leftKeys, rightKeys, condition, left, CanBroadcast(right)) => joins.BroadcastHashOuterJoin( leftKeys, rightKeys, LeftOuter, condition, planLater(left), planLater(right)) :: Nil case ExtractEquiJoinKeys( - RightOuter, leftKeys, rightKeys, condition, CanBroadcast(left), right) => + RightOuter, leftKeys, rightKeys, condition, CanBroadcast(left), right) => joins.BroadcastHashOuterJoin( leftKeys, rightKeys, RightOuter, condition, planLater(left), planLater(right)) :: Nil @@ -156,11 +156,11 @@ private[sql] abstract class SparkStrategies extends QueryPlanner[SparkPlan] { def apply(plan: LogicalPlan): Seq[SparkPlan] = plan match { // Aggregations that can be performed in two phases, before and after the shuffle. case PartialAggregation( - namedGroupingAttributes, - rewrittenAggregateExpressions, - groupingExpressions, - partialComputation, - child) if !canBeConvertedToNewAggregation(plan) => + namedGroupingAttributes, + rewrittenAggregateExpressions, + groupingExpressions, + partialComputation, + child) if !canBeConvertedToNewAggregation(plan) => execution.Aggregate( partial = false, namedGroupingAttributes, diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/joins/SortMergeJoin.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/joins/SortMergeJoin.scala index 6b7322671d..906f20d2a7 100644 --- a/sql/core/src/main/scala/org/apache/spark/sql/execution/joins/SortMergeJoin.scala +++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/joins/SortMergeJoin.scala @@ -238,7 +238,7 @@ private[joins] class SortMergeJoinScanner( * Advances the streamed input iterator and buffers all rows from the buffered input that * have matching keys. * @return true if the streamed iterator returned a row, false otherwise. If this returns true, - * then [getStreamedRow and [[getBufferedMatches]] can be called to produce the outer + * then [[getStreamedRow]] and [[getBufferedMatches]] can be called to produce the outer * join results. */ final def findNextOuterJoinRows(): Boolean = { -- cgit v1.2.3