Scaffold broker config and join hint for MSE runtime filter

pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/QueryQuotaClusterIntegrationTest.java

@@ -440,7 +440,11 @@ private void runQueriesOnBroker(float qps, boolean shouldFail) {
 
   private void runQueriesOnBroker(float qps, boolean shouldFail, String tableName) {
     int failCount = 0;
-    long deadline = System.currentTimeMillis() + 1000;
+    // For "should fail" tests use a 500ms window so queries arrive at ~4x the rate limit.
+    // Guava SmoothBursty allows all queries when spacing == stableInterval/2 (exactly 2x rate);
+    // a shorter window produces spacing well below stableInterval/2, guaranteeing throttling.
+    long windowMs = shouldFail ? 500L : 1000L;
+    long deadline = System.currentTimeMillis() + windowMs;
 
     for (int i = 0; i < qps; i++) {
       sleep(deadline, qps - i);

pinot-query-planner/src/main/java/org/apache/pinot/calcite/rel/hint/PinotHintOptions.java

@@ -113,6 +113,15 @@ public static class JoinHintOptions {
      */
     public static final String APPEND_DISTINCT_TO_SEMI_JOIN_PROJECT = "append_distinct_to_semi_join_project";
 
+    /**
+     * Per-join override for runtime filter (dynamic filter pushdown). Accepted values: {@code "true"} or
+     * {@code "false"} (case-insensitive). When set to a recognized value, takes precedence over the
+     * broker-level {@code pinot.broker.mse.runtime.filter.injection.enabled} config and the build-side
+     * row-count threshold for this join. When unset -- or set to an unrecognized value -- the broker-level
+     * config governs behavior.
+     */
+    public static final String ENABLE_RUNTIME_FILTER = "enable_runtime_filter";
+
     @Nullable
     public static Map<String, String> getJoinHintOptions(Join join) {
       return PinotHintStrategyTable.getHintOptions(join.getHints(), JOIN_HINT_OPTIONS);
@@ -143,6 +152,27 @@ public static Boolean isColocatedByJoinKeys(Join join) {
       String hint = PinotHintStrategyTable.getHintOption(join.getHints(), JOIN_HINT_OPTIONS, IS_COLOCATED_BY_JOIN_KEYS);
       return hint != null ? Boolean.parseBoolean(hint) : null;
     }
+
+    /**
+     * Returns the per-join {@link #ENABLE_RUNTIME_FILTER} hint as a {@link Boolean}, or {@code null} if
+     * the hint is absent or its value is not a recognized boolean literal ({@code "true"} / {@code "false"},
+     * case-insensitive). Callers should fall back to the broker-level config when this returns
+     * {@code null}; this keeps typos and stray values from silently flipping the override.
+     */
+    @Nullable
+    public static Boolean isRuntimeFilterEnabled(Join join) {
+      String hint = PinotHintStrategyTable.getHintOption(join.getHints(), JOIN_HINT_OPTIONS, ENABLE_RUNTIME_FILTER);
+      if (hint == null) {
+        return null;
+      }
+      if ("true".equalsIgnoreCase(hint)) {
+        return Boolean.TRUE;
+      }
+      if ("false".equalsIgnoreCase(hint)) {
+        return Boolean.FALSE;
+      }
+      return null;
+    }
   }
 
   /**

pinot-query-planner/src/test/java/org/apache/pinot/calcite/rel/hint/PinotHintOptionsTest.java

@@ -0,0 +1,99 @@
+/**
+ * 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.pinot.calcite.rel.hint;
+
+import com.google.common.collect.ImmutableList;
+import org.apache.calcite.rel.core.Join;
+import org.apache.calcite.rel.hint.RelHint;
+import org.mockito.Mockito;
+import org.testng.Assert;
+import org.testng.annotations.Test;
+
+
+public class PinotHintOptionsTest {
+
+  private static Join joinWithHints(ImmutableList<RelHint> hints) {
+    Join join = Mockito.mock(Join.class);
+    Mockito.when(join.getHints()).thenReturn(hints);
+    return join;
+  }
+
+  private static RelHint joinHint(String optionKey, String optionValue) {
+    return RelHint.builder(PinotHintOptions.JOIN_HINT_OPTIONS)
+        .hintOption(optionKey, optionValue)
+        .build();
+  }
+
+  @Test
+  public void isRuntimeFilterEnabledReturnsNullWhenNoHintsPresent() {
+    Join join = joinWithHints(ImmutableList.of());
+    Assert.assertNull(PinotHintOptions.JoinHintOptions.isRuntimeFilterEnabled(join));
+  }
+
+  @Test
+  public void isRuntimeFilterEnabledReturnsNullWhenJoinHintsExistButRuntimeFilterAbsent() {
+    Join join = joinWithHints(ImmutableList.of(
+        joinHint(PinotHintOptions.JoinHintOptions.IS_COLOCATED_BY_JOIN_KEYS, "true")));
+    Assert.assertNull(PinotHintOptions.JoinHintOptions.isRuntimeFilterEnabled(join));
+  }
+
+  @Test
+  public void isRuntimeFilterEnabledReturnsTrueWhenHintIsTrue() {
+    Join join = joinWithHints(ImmutableList.of(
+        joinHint(PinotHintOptions.JoinHintOptions.ENABLE_RUNTIME_FILTER, "true")));
+    Assert.assertEquals(PinotHintOptions.JoinHintOptions.isRuntimeFilterEnabled(join), Boolean.TRUE);
+  }
+
+  @Test
+  public void isRuntimeFilterEnabledReturnsFalseWhenHintIsFalse() {
+    Join join = joinWithHints(ImmutableList.of(
+        joinHint(PinotHintOptions.JoinHintOptions.ENABLE_RUNTIME_FILTER, "false")));
+    Assert.assertEquals(PinotHintOptions.JoinHintOptions.isRuntimeFilterEnabled(join), Boolean.FALSE);
+  }
+
+  @Test
+  public void isRuntimeFilterEnabledIsCaseInsensitiveForTrue() {
+    Join join = joinWithHints(ImmutableList.of(
+        joinHint(PinotHintOptions.JoinHintOptions.ENABLE_RUNTIME_FILTER, "TRUE")));
+    Assert.assertEquals(PinotHintOptions.JoinHintOptions.isRuntimeFilterEnabled(join), Boolean.TRUE);
+  }
+
+  @Test
+  public void isRuntimeFilterEnabledIsCaseInsensitiveForFalse() {
+    Join join = joinWithHints(ImmutableList.of(
+        joinHint(PinotHintOptions.JoinHintOptions.ENABLE_RUNTIME_FILTER, "False")));
+    Assert.assertEquals(PinotHintOptions.JoinHintOptions.isRuntimeFilterEnabled(join), Boolean.FALSE);
+  }
+
+  @Test
+  public void isRuntimeFilterEnabledReturnsNullForUnrecognizedValue() {
+    // Stray or misspelled values must NOT silently override the broker-level config. Returning null
+    // lets the caller fall back to the broker setting instead of treating "maybe" or "yes" as false.
+    Join join = joinWithHints(ImmutableList.of(
+        joinHint(PinotHintOptions.JoinHintOptions.ENABLE_RUNTIME_FILTER, "maybe")));
+    Assert.assertNull(PinotHintOptions.JoinHintOptions.isRuntimeFilterEnabled(join));
+  }
+
+  @Test
+  public void isRuntimeFilterEnabledReturnsNullForEmptyValue() {
+    Join join = joinWithHints(ImmutableList.of(
+        joinHint(PinotHintOptions.JoinHintOptions.ENABLE_RUNTIME_FILTER, "")));
+    Assert.assertNull(PinotHintOptions.JoinHintOptions.isRuntimeFilterEnabled(join));
+  }
+}

pinot-spi/src/main/java/org/apache/pinot/spi/utils/CommonConstants.java

@@ -423,6 +423,18 @@ public static class Broker {
     public static final String CONFIG_OF_MSE_MAX_SERVER_QUERY_THREADS = "pinot.broker.mse.max.server.query.threads";
     public static final int DEFAULT_MSE_MAX_SERVER_QUERY_THREADS = -1;
 
+    // MSE hash-join runtime filter (dynamic filter pushdown). Scaffolding only -- no code path reads
+    // these yet; the consuming optimizer rule lands in a follow-up. The semantics described below are
+    // the planned/intended behavior, not what happens today.
+    // INJECTION_ENABLED will gate the optimizer rule. BUILD_ROWCOUNT_THRESHOLD will cap the build-side
+    // estimated row count (RelMetadataQuery#getRowCount) above which the rule will not trigger.
+    public static final String CONFIG_OF_MSE_RUNTIME_FILTER_INJECTION_ENABLED =
+        "pinot.broker.mse.runtime.filter.injection.enabled";
+    public static final boolean DEFAULT_MSE_RUNTIME_FILTER_INJECTION_ENABLED = false;
+    public static final String CONFIG_OF_MSE_RUNTIME_FILTER_BUILD_ROWCOUNT_THRESHOLD =
+        "pinot.broker.mse.runtime.filter.build.rowcount.threshold";
+    public static final long DEFAULT_MSE_RUNTIME_FILTER_BUILD_ROWCOUNT_THRESHOLD = 1_000_000L;
+
     // Configure the request handler type used by broker to handler inbound query request.
     // NOTE: the request handler type refers to the communication between Broker and Server.
     public static final String BROKER_REQUEST_HANDLER_TYPE = "pinot.broker.request.handler.type";