[SPARK-16475][SQL] Broadcast Hint for SQL Queries

sql/catalyst/src/main/antlr4/org/apache/spark/sql/catalyst/parser/SqlBase.g4

@@ -338,7 +338,7 @@ querySpecification
        (RECORDREADER recordReader=STRING)?
        fromClause?
        (WHERE where=booleanExpression)?)
-    | ((kind=SELECT setQuantifier? namedExpressionSeq fromClause?
+    | ((kind=SELECT hint? setQuantifier? namedExpressionSeq fromClause?
        | fromClause (kind=SELECT setQuantifier? namedExpressionSeq)?)
        lateralView*
        (WHERE where=booleanExpression)?
@@ -347,6 +347,16 @@ querySpecification
        windows?)
     ;
 
+hint
+    : '/*+' hintStatement '*/'
+    ;
+
+hintStatement
+    : hintName=identifier
+    | hintName=identifier '(' parameters+=identifier parameters+=identifier ')'
+    | hintName=identifier '(' parameters+=identifier (',' parameters+=identifier)* ')'
+    ;
+
 fromClause
     : FROM relation (',' relation)* lateralView*
     ;
@@ -945,8 +955,12 @@ SIMPLE_COMMENT
     : '--' ~[\r\n]* '\r'? '\n'? -> channel(HIDDEN)
     ;
 
+BRACKETED_EMPTY_COMMENT
+    : '/**/' -> channel(HIDDEN)
+    ;
+
 BRACKETED_COMMENT
-    : '/*' .*? '*/' -> channel(HIDDEN)
+    : '/*' ~[+] .*? '*/' -> channel(HIDDEN)
     ;
 
 WS

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala

@@ -84,7 +84,8 @@ class Analyzer(
     Batch("Substitution", fixedPoint,
       CTESubstitution,
       WindowsSubstitution,
-      EliminateUnions),
+      EliminateUnions,
+      SubstituteHints),
     Batch("Resolution", fixedPoint,
       ResolveRelations ::
       ResolveReferences ::
@@ -1786,6 +1787,63 @@ class Analyzer(
     }
   }
 
+  /**
+   * Substitute Hints.
+   * - BROADCAST/BROADCASTJOIN/MAPJOIN match the closest table with the given name parameters.
+   *
+   * This rule substitutes `UnresolvedRelation`s in `Substitute` batch before `ResolveRelations`
+   * rule is applied. Here are two reasons.
+   * - To support `MetastoreRelation` in Hive module.
+   * - To reduce the effect of `Hint` on the other rules.
+   *
+   * After this rule, it is guaranteed that there exists no unknown `Hint` in the plan.
+   * All new `Hint`s should be transformed into concrete Hint classes `BroadcastHint` here.
+   */
+  object SubstituteHints extends Rule[LogicalPlan] {
+    val BROADCAST_HINT_NAMES = Set("BROADCAST", "BROADCASTJOIN", "MAPJOIN")
+
+    import scala.collection.mutable.Set
+    private def appendAllDescendant(set: Set[LogicalPlan], plan: LogicalPlan): Unit = {
+      set += plan
+      plan.children.foreach { child => appendAllDescendant(set, child) }
+    }
+
+    def apply(plan: LogicalPlan): LogicalPlan = plan transform {
+      case logical: LogicalPlan => logical transformDown {
+        case h @ Hint(name, parameters, child) if BROADCAST_HINT_NAMES.contains(name.toUpperCase) =>
+          var resolvedChild = child
+          for (table <- parameters) {
+            var stop = false
+            val skipNodeSet = scala.collection.mutable.Set.empty[LogicalPlan]
+            resolvedChild = resolvedChild.transformDown {
+              case n if skipNodeSet.contains(n) =>
+                skipNodeSet -= n
+                n
+              case p @ Project(_, _) if p != resolvedChild =>
+                appendAllDescendant(skipNodeSet, p)
+                skipNodeSet -= p
+                p
+              case r @ BroadcastHint(UnresolvedRelation(t, _))
+                  if !stop && resolver(t.table, table) =>
+                stop = true
+                r
+              case r @ UnresolvedRelation(t, alias) if !stop && resolver(t.table, table) =>
+                stop = true
+                if (alias.isDefined) {
+                  SubqueryAlias(alias.get, BroadcastHint(r.copy(alias = None)))
+                } else {
+                  BroadcastHint(r)
+                }
+            }
+          }
+          resolvedChild
+
+        // Remove unrecognized hints
+        case Hint(name, _, child) => child
+      }
+    }
+  }
+
   /**
    * Check and add proper window frames for all window functions.
    */

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/CheckAnalysis.scala

@@ -372,6 +372,10 @@ trait CheckAnalysis extends PredicateHelper {
                  |in operator ${operator.simpleString}
                """.stripMargin)
 
+          case Hint(_, _, _) =>
+            throw new IllegalStateException(
+              "logical hint operator should have been removed by analyzer")
+
           case _ => // Analysis successful!
         }
     }

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/parser/AstBuilder.scala

@@ -377,7 +377,10 @@ class AstBuilder extends SqlBaseBaseVisitor[AnyRef] with Logging {
         }
 
         // Window
-        withDistinct.optionalMap(windows)(withWindows)
+        val withWindow = withDistinct.optionalMap(windows)(withWindows)
+
+        // Hint
+        withWindow.optionalMap(ctx.hint)(withHints)
     }
   }
 
@@ -508,6 +511,16 @@ class AstBuilder extends SqlBaseBaseVisitor[AnyRef] with Logging {
     }
   }
 
+  /**
+   * Add a Hint to a logical plan.
+   */
+  private def withHints(
+      ctx: HintContext,
+      query: LogicalPlan): LogicalPlan = withOrigin(ctx) {
+    val stmt = ctx.hintStatement
+    Hint(stmt.hintName.getText, stmt.parameters.asScala.map(_.getText), query)
+  }
+
   /**
    * Add a [[Generate]] (Lateral View) to a logical plan.
    */

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/basicLogicalOperators.scala

@@ -354,6 +354,15 @@ case class BroadcastHint(child: LogicalPlan) extends UnaryNode {
   override lazy val statistics: Statistics = super.statistics.copy(isBroadcastable = true)
 }
 
+/**
+ * A general hint for the child.
+ * A pair of (name, parameters).
+ */
+case class Hint(name: String, parameters: Seq[String], child: LogicalPlan) extends UnaryNode {
+  override lazy val resolved: Boolean = false
+  override def output: Seq[Attribute] = child.output
+}
+
 case class InsertIntoTable(
     table: LogicalPlan,
     partition: Map[String, Option[String]],

sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/analysis/AnalysisTest.scala

@@ -32,6 +32,7 @@ trait AnalysisTest extends PlanTest {
     val conf = new SimpleCatalystConf(caseSensitive)
     val catalog = new SessionCatalog(new InMemoryCatalog, EmptyFunctionRegistry, conf)
     catalog.createTempView("TaBlE", TestRelations.testRelation, overrideIfExists = true)
+    catalog.createTempView("TaBlE2", TestRelations.testRelation2, overrideIfExists = true)
     new Analyzer(catalog, conf) {
       override val extendedResolutionRules = EliminateSubqueryAliases :: Nil
     }

sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/analysis/SubstituteHintsSuite.scala

@@ -0,0 +1,125 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.catalyst.analysis
+
+import org.apache.spark.sql.catalyst.dsl.expressions._
+import org.apache.spark.sql.catalyst.dsl.plans._
+import org.apache.spark.sql.catalyst.plans.logical._
+
+class SubstituteHintsSuite extends AnalysisTest {
+  import org.apache.spark.sql.catalyst.analysis.TestRelations._
+
+  val a = testRelation.output(0)
+  val b = testRelation2.output(0)
+
+  test("case-sensitive or insensitive parameters") {
+    checkAnalysis(
+      Hint("MAPJOIN", Seq("TaBlE"), table("TaBlE")),
+      BroadcastHint(testRelation),
+      caseSensitive = false)
+
+    checkAnalysis(
+      Hint("MAPJOIN", Seq("table"), table("TaBlE")),
+      BroadcastHint(testRelation),
+      caseSensitive = false)
+
+    checkAnalysis(
+      Hint("MAPJOIN", Seq("TaBlE"), table("TaBlE")),
+      BroadcastHint(testRelation))
+
+    checkAnalysis(
+      Hint("MAPJOIN", Seq("table"), table("TaBlE")),
+      testRelation)
+  }
+
+  test("single hint") {
+    checkAnalysis(
+      Hint("MAPJOIN", Seq("TaBlE"), table("TaBlE").select(a)),
+      BroadcastHint(testRelation).select(a))
+
+    checkAnalysis(
+      Hint("MAPJOIN", Seq("TaBlE"), table("TaBlE").as("t").join(table("TaBlE2").as("u")).select(a)),
+      BroadcastHint(testRelation).join(testRelation2).select(a))
+
+    checkAnalysis(
+      Hint("MAPJOIN", Seq("TaBlE2"),
+        table("TaBlE").as("t").join(table("TaBlE2").as("u")).select(a)),
+      testRelation.join(BroadcastHint(testRelation2)).select(a))
+  }
+
+  test("single hint with multiple parameters") {
+    checkAnalysis(
+      Hint("MAPJOIN", Seq("TaBlE", "TaBlE"),
+        table("TaBlE").as("t").join(table("TaBlE2").as("u")).select(a)),
+      BroadcastHint(testRelation).join(testRelation2).select(a))
+
+    checkAnalysis(
+      Hint("MAPJOIN", Seq("TaBlE", "TaBlE2"),
+        table("TaBlE").as("t").join(table("TaBlE2").as("u")).select(a)),
+      BroadcastHint(testRelation).join(BroadcastHint(testRelation2)).select(a))
+  }
+
+  test("duplicated nested hints are transformed into one") {
+    checkAnalysis(
+      Hint("MAPJOIN", Seq("TaBlE"),
+        Hint("MAPJOIN", Seq("TaBlE"), table("TaBlE").as("t").select('a))
+          .join(table("TaBlE2").as("u")).select(a)),
+      BroadcastHint(testRelation).select(a).join(testRelation2).select(a))
+
+    checkAnalysis(
+      Hint("MAPJOIN", Seq("TaBlE2"),
+        table("TaBlE").as("t").select(a)
+          .join(Hint("MAPJOIN", Seq("TaBlE2"), table("TaBlE2").as("u").select(b))).select(a)),
+      testRelation.select(a).join(BroadcastHint(testRelation2).select(b)).select(a))
+  }
+
+  test("distinct nested two hints are handled separately") {
+    checkAnalysis(
+      Hint("MAPJOIN", Seq("TaBlE2"),
+        Hint("MAPJOIN", Seq("TaBlE"), table("TaBlE").as("t").select(a))
+          .join(table("TaBlE2").as("u")).select(a)),
+      BroadcastHint(testRelation).select(a).join(BroadcastHint(testRelation2)).select(a))
+
+    checkAnalysis(
+      Hint("MAPJOIN", Seq("TaBlE"),
+        table("TaBlE").as("t")
+          .join(Hint("MAPJOIN", Seq("TaBlE2"), table("TaBlE2").as("u").select(b))).select(a)),
+      BroadcastHint(testRelation).join(BroadcastHint(testRelation2).select(b)).select(a))
+  }
+
+  test("deep self join") {
+    checkAnalysis(
+      Hint("MAPJOIN", Seq("TaBlE"),
+        table("TaBlE").join(table("TaBlE")).join(table("TaBlE")).join(table("TaBlE")).select(a)),
+      BroadcastHint(testRelation).join(testRelation).join(testRelation).join(testRelation)
+        .select(a))
+  }
+
+  test("subquery should be ignored") {
+    checkAnalysis(
+      Hint("MAPJOIN", Seq("TaBlE"),
+        table("TaBlE").select(a).as("x").join(table("TaBlE")).select(a)),
+      testRelation.select(a).join(BroadcastHint(testRelation)).select(a))
+
+    checkAnalysis(
+      Hint("MAPJOIN", Seq("TaBlE"),
+        table("TaBlE").as("t").select(a).as("x")
+          .join(table("TaBlE2").as("t2")).select(a)),
+      testRelation.select(a).join(testRelation2).select(a))
+  }
+}

sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/parser/PlanParserSuite.scala

@@ -447,4 +447,46 @@ class PlanParserSuite extends PlanTest {
     assertEqual("select a, b from db.c where x !> 1",
       table("db", "c").where('x <= 1).select('a, 'b))
   }
+
+  test("select hint syntax") {
+    // Hive compatibility: Missing parameter raises ParseException.
+    val m = intercept[ParseException] {
+      parsePlan("SELECT /*+ HINT() */ * FROM t")
+    }.getMessage
+    assert(m.contains("no viable alternative at input"))
+
+    // Hive compatibility: No database.
+    val m2 = intercept[ParseException] {
+      parsePlan("SELECT /*+ MAPJOIN(default.t) */ * from default.t")
+    }.getMessage
+    assert(m2.contains("no viable alternative at input"))
+
+    comparePlans(
+      parsePlan("SELECT /*+ HINT */ * FROM t"),
+      Hint("HINT", Seq.empty, table("t").select(star())))
+
+    comparePlans(
+      parsePlan("SELECT /*+ BROADCASTJOIN(u) */ * FROM t"),
+      Hint("BROADCASTJOIN", Seq("u"), table("t").select(star())))
+
+    comparePlans(
+      parsePlan("SELECT /*+ MAPJOIN(u) */ * FROM t"),
+      Hint("MAPJOIN", Seq("u"), table("t").select(star())))
+
+    comparePlans(
+      parsePlan("SELECT /*+ STREAMTABLE(a,b,c) */ * FROM t"),
+      Hint("STREAMTABLE", Seq("a", "b", "c"), table("t").select(star())))
+
+    comparePlans(
+      parsePlan("SELECT /*+ INDEX(t emp_job_ix) */ * FROM t"),
+      Hint("INDEX", Seq("t", "emp_job_ix"), table("t").select(star())))
+
+    comparePlans(
+      parsePlan("SELECT /*+ MAPJOIN(`default.t`) */ * from `default.t`"),
+      Hint("MAPJOIN", Seq("default.t"), table("default.t").select(star())))
+
+    comparePlans(
+      parsePlan("SELECT /*+ MAPJOIN(t) */ a from t where true group by a order by a"),
+      Hint("MAPJOIN", Seq("t"), table("t").where(Literal(true)).groupBy('a)('a)).orderBy('a.asc))
+  }
 }