[SPARK-59345][SQL] Add RUNTIME_FILTER join hint

common/utils/src/main/resources/error/error-conditions.json

@@ -10217,7 +10217,7 @@
   },
   "_LEGACY_ERROR_TEMP_1046" : {
     "message" : [
-      "Join strategy hint parameter should be an identifier or string but was <unsupported> (<class>)."
+      "Join hint parameter should be an identifier or string but was <unsupported> (<class>)."
     ]
   },
   "_LEGACY_ERROR_TEMP_1047" : {

docs/sql-ref-syntax-qry-select-hints.md

@@ -172,6 +172,55 @@ SELECT /*+ SHUFFLE_REPLICATE_NL(t1) */ * FROM t1 INNER JOIN t2 ON t1.key = t2.ke
 SELECT /*+ BROADCAST(t1), MERGE(t1, t2) */ * FROM t1 INNER JOIN t2 ON t1.key = t2.key;
 ```
 
+### Runtime Filter Hints
+
+A runtime filter prunes one side of a join using the join key values found on the other side, so
+rows that cannot match are discarded early. Spark decides on its own whether such a filter is worth
+building, based on estimates of how much data it would save. Runtime filter hints let users make
+that decision instead, for the cases where the estimates are unavailable or wrong.
+
+#### Runtime Filter Hints Types
+
+* **RUNTIME_FILTER**
+
+    Suggests that Spark build a runtime filter from the hinted relation and use it to prune the
+    other side of the join. Use it when the hinted side is known to match only a small fraction
+    of the other side, but Spark does not choose a runtime filter on its own, typically because
+    table statistics are missing or misleading. The hinted side may be any relation or subquery,
+    and is never itself pruned. The hint does not choose how the pruning is done; Spark picks the
+    mechanism. `RUNTIME_FILTER` can be combined with a join strategy hint.
+
+The hint overrides Spark's cost estimates, but not the requirements that make a runtime filter
+correct, so Spark is not guaranteed to follow it. A side that join semantics forbid pruning is
+never pruned, e.g. the left side of a `LEFT OUTER` join, whose rows must all appear in the output.
+The hinted side must produce the same rows each time it is evaluated, since building the filter
+may evaluate it separately from the join; a side whose rows or join keys depend on evaluation
+order, such as a `LIMIT` without a unique ordering, a `TABLESAMPLE` without `REPEATABLE`, or a
+key computed by `first` or `last`, does not qualify. Building the filter may evaluate the hinted
+side once more, which is the cost the hint asks Spark to spend. A hint that cannot be applied does
+not make Spark build a filter in the opposite direction instead.
+
+Spark issues a warning with the reason when it cannot apply the hint. Hinting both sides of a join
+is ambiguous, since each side would then have to be built from the other; Spark warns and ignores
+the hint.
+
+#### Examples
+
+```sql
+-- Build a runtime filter from t2 and use it to prune t1.
+SELECT /*+ RUNTIME_FILTER(t2) */ * FROM t1 INNER JOIN t2 ON t1.key = t2.key;
+
+-- The hinted side may be any relation or subquery, not only a table.
+SELECT /*+ RUNTIME_FILTER(t2) */ *
+FROM t1 INNER JOIN (SELECT DISTINCT key FROM t3) t2 ON t1.key = t2.key;
+
+-- A runtime filter hint can be combined with a join strategy hint.
+SELECT /*+ MERGE(t1, t2), RUNTIME_FILTER(t2) */ * FROM t1 INNER JOIN t2 ON t1.key = t2.key;
+
+-- Hinting both sides is ambiguous, so Spark issues a warning and ignores the hint.
+SELECT /*+ RUNTIME_FILTER(t1, t2) */ * FROM t1 INNER JOIN t2 ON t1.key = t2.key;
+```
+
 ### Related Statements
 
 * [JOIN](sql-ref-syntax-qry-select-join.html)

python/pyspark/sql/connect/proto/relations_pb2.pyi

@@ -3448,7 +3448,8 @@ class Hint(google.protobuf.message.Message):
     name: builtins.str
     """(Required) Hint name.
 
-    Supported Join hints include BROADCAST, MERGE, SHUFFLE_HASH, SHUFFLE_REPLICATE_NL.
+    Supported Join hints include BROADCAST, MERGE, SHUFFLE_HASH, SHUFFLE_REPLICATE_NL,
+    RUNTIME_FILTER.
 
     Supported partitioning hints include COALESCE, REPARTITION, REPARTITION_BY_RANGE.
     """

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

@@ -586,7 +586,7 @@ class Analyzer(
     Batch("Disable Hints", Once,
       new ResolveHints.DisableHints),
     Batch("Hints", fixedPoint,
-      Seq(ResolveHints.ResolveJoinStrategyHints,
+      Seq(ResolveHints.ResolveJoinHints,
         ResolveHints.ResolveCoalesceHints) ++
         hintResolutionRules: _*),
     Batch("Simple Sanity Check", Once,

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

@@ -31,7 +31,8 @@ import org.apache.spark.sql.internal.SQLConf
 
 
 /**
- * Collection of rules related to hints. The only hint currently available is join strategy hint.
+ * Collection of rules related to hints: join strategy hints, the runtime filter hint, and
+ * partitioning hints.
  *
  * Note that this is separately into two rules because in the future we might introduce new hint
  * rules that have different ordering requirements from join strategies.
@@ -41,7 +42,7 @@ object ResolveHints {
   /**
    * Checks if the given multi-part identifiers are matched with each other.
    *
-   * The [[ResolveJoinStrategyHints]] rule is applied before the resolution batch in the analyzer
+   * The [[ResolveJoinHints]] rule is applied before the resolution batch in the analyzer
    * and we cannot semantically compare them at this stage. Therefore, we follow a simple rule;
    * they match if an identifier in a hint is a tail of an identifier in a relation. This process
    * is independent of a session catalog (`currentDb` in [[SessionCatalog]]) and it just compares
@@ -77,22 +78,28 @@ object ResolveHints {
    * is not aliased differently), subquery, or common table expression that match the specified
    * name.
    *
+   * [[RuntimeFilterHint]] is resolved here too. It takes the same per-relation form, e.g.
+   * "RUNTIME_FILTER(a)", and applies to a join side, so it shares this rule's relation matching
+   * and lands on the same [[HintInfo]] -- which is what lets it accompany a join strategy hint on
+   * a relation instead of displacing one.
+   *
    * The hint resolution works by recursively traversing down the query plan to find a relation or
    * subquery that matches one of the specified relation aliases. The traversal does not go past
    * beyond any view reference, with clause or subquery alias.
    *
    * This rule must happen before common table expressions.
    */
-  object ResolveJoinStrategyHints extends Rule[LogicalPlan] {
+  object ResolveJoinHints extends Rule[LogicalPlan] {
     private def hintErrorHandler = conf.hintErrorHandler
 
     def resolver: Resolver = conf.resolver
 
     private def createHintInfo(hintName: String): HintInfo = {
-      HintInfo(strategy =
-        JoinStrategyHint.strategies.find(
+      HintInfo(
+        strategy = JoinStrategyHint.strategies.find(
           _.hintAliases.map(
-            _.toUpperCase(Locale.ROOT)).contains(hintName.toUpperCase(Locale.ROOT))))
+            _.toUpperCase(Locale.ROOT)).contains(hintName.toUpperCase(Locale.ROOT))),
+        runtimeFilterSource = RuntimeFilterHint.isRuntimeFilterHintName(hintName))
     }
 
     private def matchedIdentifier(identInHint: Seq[String], identInQuery: Seq[String]): Boolean =
@@ -157,7 +164,8 @@ object ResolveHints {
 
     def apply(plan: LogicalPlan): LogicalPlan = plan.resolveOperatorsUpWithPruning(
       _.containsPattern(UNRESOLVED_HINT), ruleId) {
-      case h: UnresolvedHint if JoinStrategyHint.isJoinStrategyHintName(h.name) =>
+      case h: UnresolvedHint if JoinStrategyHint.isJoinStrategyHintName(h.name) ||
+          RuntimeFilterHint.isRuntimeFilterHintName(h.name) =>
         if (h.parameters.isEmpty) {
           // If there is no table alias specified, apply the hint on the entire subtree.
           ResolvedHint(h.child, createHintInfo(h.name))
@@ -167,7 +175,7 @@ object ResolveHints {
             case StringLiteral(tableName) => UnresolvedAttribute.parseAttributeName(tableName)
             case tableId: UnresolvedAttribute => tableId.nameParts
             case unsupported =>
-              throw QueryCompilationErrors.joinStrategyHintParameterNotSupportedError(unsupported)
+              throw QueryCompilationErrors.joinHintParameterNotSupportedError(unsupported)
           }.toSet
           val relationsInHintWithMatch = new mutable.HashSet[Seq[String]]
           val applied = applyJoinStrategyHint(