[SPARK-20233] Move star-filter conf as a method param and other changes.

Author: ioana-delaney

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/CostBasedJoinReorder.scala

@@ -149,7 +149,7 @@ object JoinReorderDP extends PredicateHelper with Logging {
     }.toMap)
 
     // Build filters from the join graph to be used by the search algorithm.
-    val filters = JoinReorderDPFilters(conf).buildJoinGraphInfo(items, conditions, itemIndex)
+    val filters = JoinReorderDPFilters.buildJoinGraphInfo(conf, items, conditions, itemIndex)
 
     // Build plans for next levels until the last level has only one plan. This plan contains
     // all items that can be joined, so there's no need to continue.
@@ -249,15 +249,15 @@ object JoinReorderDP extends PredicateHelper with Logging {
       return None
     }
 
-    if (conf.joinReorderDPStarFilter && filters.isDefined) {
+    if (filters.isDefined) {
       // Apply star-join filter, which ensures that tables in a star schema relationship
       // are planned together. The star-filter will eliminate joins among star and non-star
       // tables until the star joins are built. The following combinations are allowed:
       // 1. (oneJoinPlan U otherJoinPlan) is a subset of star-join
       // 2. star-join is a subset of (oneJoinPlan U otherJoinPlan)
       // 3. (oneJoinPlan U otherJoinPlan) is a subset of non star-join
       val isValidJoinCombination =
-        JoinReorderDPFilters(conf).starJoinFilter(oneJoinPlan.itemIds, otherJoinPlan.itemIds,
+        JoinReorderDPFilters.starJoinFilter(oneJoinPlan.itemIds, otherJoinPlan.itemIds,
           filters.get)
       if (!isValidJoinCombination) return None
     }
@@ -362,34 +362,33 @@ case class Cost(card: BigInt, size: BigInt) {
  *
  * Filters (2) and (3) are not implemented.
  */
-case class JoinReorderDPFilters(conf: SQLConf) extends PredicateHelper {
+object JoinReorderDPFilters extends PredicateHelper {
   /**
    * Builds join graph information to be used by the filtering strategies.
    * Currently, it builds the sets of star/non-star joins.
    * It can be extended with the sets of connected/unconnected joins, which
    * can be used to filter Cartesian products.
    */
   def buildJoinGraphInfo(
+      conf: SQLConf,
       items: Seq[LogicalPlan],
       conditions: Set[Expression],
-      planIndex: Seq[(LogicalPlan, Int)]): Option[JoinGraphInfo] = {
-
-    // Compute the tables in a star-schema relationship.
-    val starJoin = StarSchemaDetection(conf).findStarJoins(items, conditions.toSeq)
-    val nonStarJoin = items.filterNot(starJoin.contains(_))
-
-    if (starJoin.nonEmpty && nonStarJoin.nonEmpty) {
-      val (starInt, nonStarInt) = planIndex.collect {
-        case (p, i) if starJoin.contains(p) =>
-          (Some(i), None)
-        case (p, i) if nonStarJoin.contains(p) =>
-          (None, Some(i))
-        case _ =>
-          (None, None)
-      }.unzip
-      Some(JoinGraphInfo(starInt.flatten.toSet, nonStarInt.flatten.toSet))
+      itemIndex: Seq[(LogicalPlan, Int)]): Option[JoinGraphInfo] = {
+
+    if (conf.joinReorderDPStarFilter) {
+      // Compute the tables in a star-schema relationship.
+      val starJoin = StarSchemaDetection(conf).findStarJoins(items, conditions.toSeq)
+      val nonStarJoin = items.filterNot(starJoin.contains(_))
+
+      if (starJoin.nonEmpty && nonStarJoin.nonEmpty) {
+        val itemMap = itemIndex.toMap
+        Some(JoinGraphInfo(starJoin.map(itemMap).toSet, nonStarJoin.map(itemMap).toSet))
+      } else {
+        // Nothing interesting to return.
+        None
+      }
     } else {
-      // Nothing interesting to return.
+      // Star schema filter is not enabled.
       None
     }
   }

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/StarSchemaDetection.scala

@@ -76,7 +76,7 @@ case class StarSchemaDetection(conf: SQLConf) extends PredicateHelper {
 
     val emptyStarJoinPlan = Seq.empty[LogicalPlan]
 
-    if (!conf.starSchemaDetection || input.size < 2) {
+    if (input.size < 2) {
       emptyStarJoinPlan
     } else {
       // Find if the input plans are eligible for star join detection.

sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/optimizer/StarJoinCostBasedReorderSuite.scala

@@ -31,10 +31,8 @@ import org.apache.spark.sql.internal.SQLConf._
 class StarJoinCostBasedReorderSuite extends PlanTest with StatsEstimationTestBase {
 
   override val conf = new SQLConf().copy(
-    CASE_SENSITIVE -> true,
     CBO_ENABLED -> true,
     JOIN_REORDER_ENABLED -> true,
-    STARSCHEMA_DETECTION -> true,
     JOIN_REORDER_DP_STAR_FILTER -> true)
 
   object Optimize extends RuleExecutor[LogicalPlan] {