From cf446cec1e6e44c0e28ed27b35bfe477c7733167 Mon Sep 17 00:00:00 2001 From: Tianle Zhang Date: Wed, 20 May 2026 13:20:43 -0700 Subject: [PATCH 1/6] OA-983: Scaffold broker config and join hint for MSE runtime filter Add the configuration surface for the upcoming MSE runtime filter (dynamic filter pushdown) feature without changing any runtime behavior: * CONFIG_OF_MSE_RUNTIME_FILTER_ENABLED (pinot.broker.mse.runtime.filter.enabled, default false) - broker-level kill switch. * CONFIG_OF_MSE_RUNTIME_FILTER_BUILD_CARDINALITY_THRESHOLD (pinot.broker.mse.runtime.filter.build.cardinality.threshold, default 1_000_000) - upper bound on build-side row count for the optimizer rule to trigger. * JoinHintOptions.ENABLE_RUNTIME_FILTER (enable_runtime_filter) - per-join override that takes precedence over the broker config and the threshold. * JoinHintOptions.isRuntimeFilterEnabled(Join) accessor. Subsequent PRs in this initiative wire the Bloom filter predicate primitive, the planner rule, and the runtime operators behind these settings. The feature stays disabled by default until rollout. Tests: * PinotHintOptionsTest covers absent hint, present-but-unrelated hint, true/false values, case-insensitive parsing, and the documented Boolean.parseBoolean fallback on unparseable values. Co-Authored-By: Claude Opus 4.7 (1M context) --- .../calcite/rel/hint/PinotHintOptions.java | 18 ++++ .../rel/hint/PinotHintOptionsTest.java | 85 +++++++++++++++++++ .../pinot/spi/utils/CommonConstants.java | 11 +++ 3 files changed, 114 insertions(+) create mode 100644 pinot-query-planner/src/test/java/org/apache/pinot/calcite/rel/hint/PinotHintOptionsTest.java diff --git a/pinot-query-planner/src/main/java/org/apache/pinot/calcite/rel/hint/PinotHintOptions.java b/pinot-query-planner/src/main/java/org/apache/pinot/calcite/rel/hint/PinotHintOptions.java index 1966ed3dd0..34ae82d182 100644 --- a/pinot-query-planner/src/main/java/org/apache/pinot/calcite/rel/hint/PinotHintOptions.java +++ b/pinot-query-planner/src/main/java/org/apache/pinot/calcite/rel/hint/PinotHintOptions.java @@ -113,6 +113,14 @@ 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: "true" / "false" + * (case-insensitive). When set, takes precedence over the broker-level + * {@code pinot.broker.mse.runtime.filter.enabled} config and any cardinality threshold for this join. + * When unset, the broker-level config governs behavior. See OA-983. + */ + public static final String ENABLE_RUNTIME_FILTER = "enable_runtime_filter"; + @Nullable public static Map getJoinHintOptions(Join join) { return PinotHintStrategyTable.getHintOptions(join.getHints(), JOIN_HINT_OPTIONS); @@ -143,6 +151,16 @@ 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, or {@code null} if the hint is absent. + * Callers should fall back to the broker-level config when this returns {@code null}. + */ + @Nullable + public static Boolean isRuntimeFilterEnabled(Join join) { + String hint = PinotHintStrategyTable.getHintOption(join.getHints(), JOIN_HINT_OPTIONS, ENABLE_RUNTIME_FILTER); + return hint != null ? Boolean.parseBoolean(hint) : null; + } } /** diff --git a/pinot-query-planner/src/test/java/org/apache/pinot/calcite/rel/hint/PinotHintOptionsTest.java b/pinot-query-planner/src/test/java/org/apache/pinot/calcite/rel/hint/PinotHintOptionsTest.java new file mode 100644 index 0000000000..df58ff7ee6 --- /dev/null +++ b/pinot-query-planner/src/test/java/org/apache/pinot/calcite/rel/hint/PinotHintOptionsTest.java @@ -0,0 +1,85 @@ +/** + * 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 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 isRuntimeFilterEnabledIsCaseInsensitive() { + Join join = joinWithHints(ImmutableList.of( + joinHint(PinotHintOptions.JoinHintOptions.ENABLE_RUNTIME_FILTER, "TRUE"))); + Assert.assertEquals(PinotHintOptions.JoinHintOptions.isRuntimeFilterEnabled(join), Boolean.TRUE); + } + + @Test + public void isRuntimeFilterEnabledTreatsUnparseableValueAsFalse() { + // Boolean.parseBoolean returns false for any non-"true" value rather than throwing. + // Document and lock in this behavior so unexpected hint values do not silently flip semantics. + Join join = joinWithHints(ImmutableList.of( + joinHint(PinotHintOptions.JoinHintOptions.ENABLE_RUNTIME_FILTER, "maybe"))); + Assert.assertEquals(PinotHintOptions.JoinHintOptions.isRuntimeFilterEnabled(join), Boolean.FALSE); + } +} diff --git a/pinot-spi/src/main/java/org/apache/pinot/spi/utils/CommonConstants.java b/pinot-spi/src/main/java/org/apache/pinot/spi/utils/CommonConstants.java index c84d03560a..bd7b941054 100644 --- a/pinot-spi/src/main/java/org/apache/pinot/spi/utils/CommonConstants.java +++ b/pinot-spi/src/main/java/org/apache/pinot/spi/utils/CommonConstants.java @@ -423,6 +423,17 @@ 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; + // Runtime filter (dynamic filter pushdown) for MSE hash joins. When enabled and the build-side + // estimated cardinality is below CONFIG_OF_MSE_RUNTIME_FILTER_BUILD_CARDINALITY_THRESHOLD, a + // Bloom filter over the join keys is constructed during build and applied on the probe side + // (ideally pushed into the leaf-stage segment scan) so non-matching rows are pruned early. + // Disabled by default while the feature is rolled out. + public static final String CONFIG_OF_MSE_RUNTIME_FILTER_ENABLED = "pinot.broker.mse.runtime.filter.enabled"; + public static final boolean DEFAULT_MSE_RUNTIME_FILTER_ENABLED = false; + public static final String CONFIG_OF_MSE_RUNTIME_FILTER_BUILD_CARDINALITY_THRESHOLD = + "pinot.broker.mse.runtime.filter.build.cardinality.threshold"; + public static final long DEFAULT_MSE_RUNTIME_FILTER_BUILD_CARDINALITY_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"; From 3410d5b37e8713f393245086ab7e2e8fa05b0527 Mon Sep 17 00:00:00 2001 From: Tianle Zhang Date: Wed, 20 May 2026 15:39:46 -0700 Subject: [PATCH 2/6] Address Copilot review comments on PR #195 - CommonConstants: reword the runtime-filter section to make clear that this PR ships configuration scaffolding only; the optimizer rule, build / probe operators, and leaf-stage predicate are added in follow-ups. - PinotHintOptions.isRuntimeFilterEnabled: switch from Boolean.parseBoolean to strict parsing. Recognized values are "true" / "false" (case-insensitive); anything else returns null so callers fall back to the broker-level config instead of silently treating typos as false. - Tests: replace the "treats unparseable value as false" case with explicit coverage for unrecognized and empty values returning null, and add a dedicated case-insensitive-for-false case. Co-Authored-By: Claude Opus 4.7 (1M context) --- .../calcite/rel/hint/PinotHintOptions.java | 26 ++++++++++++++----- .../rel/hint/PinotHintOptionsTest.java | 24 +++++++++++++---- .../pinot/spi/utils/CommonConstants.java | 12 +++++---- 3 files changed, 45 insertions(+), 17 deletions(-) diff --git a/pinot-query-planner/src/main/java/org/apache/pinot/calcite/rel/hint/PinotHintOptions.java b/pinot-query-planner/src/main/java/org/apache/pinot/calcite/rel/hint/PinotHintOptions.java index 34ae82d182..a125d67340 100644 --- a/pinot-query-planner/src/main/java/org/apache/pinot/calcite/rel/hint/PinotHintOptions.java +++ b/pinot-query-planner/src/main/java/org/apache/pinot/calcite/rel/hint/PinotHintOptions.java @@ -114,10 +114,11 @@ 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: "true" / "false" - * (case-insensitive). When set, takes precedence over the broker-level - * {@code pinot.broker.mse.runtime.filter.enabled} config and any cardinality threshold for this join. - * When unset, the broker-level config governs behavior. See OA-983. + * 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.enabled} config and any cardinality threshold + * for this join. When unset -- or set to an unrecognized value -- the broker-level config governs + * behavior. See OA-983. */ public static final String ENABLE_RUNTIME_FILTER = "enable_runtime_filter"; @@ -153,13 +154,24 @@ public static Boolean isColocatedByJoinKeys(Join join) { } /** - * Returns the per-join {@link #ENABLE_RUNTIME_FILTER} hint, or {@code null} if the hint is absent. - * Callers should fall back to the broker-level config when this returns {@code 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); - return hint != null ? Boolean.parseBoolean(hint) : null; + if (hint == null) { + return null; + } + if ("true".equalsIgnoreCase(hint)) { + return Boolean.TRUE; + } + if ("false".equalsIgnoreCase(hint)) { + return Boolean.FALSE; + } + return null; } } diff --git a/pinot-query-planner/src/test/java/org/apache/pinot/calcite/rel/hint/PinotHintOptionsTest.java b/pinot-query-planner/src/test/java/org/apache/pinot/calcite/rel/hint/PinotHintOptionsTest.java index df58ff7ee6..5aaa92479a 100644 --- a/pinot-query-planner/src/test/java/org/apache/pinot/calcite/rel/hint/PinotHintOptionsTest.java +++ b/pinot-query-planner/src/test/java/org/apache/pinot/calcite/rel/hint/PinotHintOptionsTest.java @@ -68,18 +68,32 @@ public void isRuntimeFilterEnabledReturnsFalseWhenHintIsFalse() { } @Test - public void isRuntimeFilterEnabledIsCaseInsensitive() { + 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 isRuntimeFilterEnabledTreatsUnparseableValueAsFalse() { - // Boolean.parseBoolean returns false for any non-"true" value rather than throwing. - // Document and lock in this behavior so unexpected hint values do not silently flip semantics. + public void isRuntimeFilterEnabledIsCaseInsensitiveForFalse() { Join join = joinWithHints(ImmutableList.of( - joinHint(PinotHintOptions.JoinHintOptions.ENABLE_RUNTIME_FILTER, "maybe"))); + 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)); + } } diff --git a/pinot-spi/src/main/java/org/apache/pinot/spi/utils/CommonConstants.java b/pinot-spi/src/main/java/org/apache/pinot/spi/utils/CommonConstants.java index bd7b941054..dce7426f6d 100644 --- a/pinot-spi/src/main/java/org/apache/pinot/spi/utils/CommonConstants.java +++ b/pinot-spi/src/main/java/org/apache/pinot/spi/utils/CommonConstants.java @@ -423,11 +423,13 @@ 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; - // Runtime filter (dynamic filter pushdown) for MSE hash joins. When enabled and the build-side - // estimated cardinality is below CONFIG_OF_MSE_RUNTIME_FILTER_BUILD_CARDINALITY_THRESHOLD, a - // Bloom filter over the join keys is constructed during build and applied on the probe side - // (ideally pushed into the leaf-stage segment scan) so non-matching rows are pruned early. - // Disabled by default while the feature is rolled out. + // Configuration surface for the planned MSE hash-join runtime filter (dynamic filter pushdown, + // OA-983). These keys are scaffolding only -- no runtime behavior is wired up in this PR. + // Follow-up PRs will plumb them into the planner rule, the build-side Bloom construction in + // HashJoinOperator, the probe-side RuntimeFilterOperator, and the leaf-stage segment-scan + // predicate. Once the feature is implemented, ENABLED will gate the optimizer rule and + // BUILD_CARDINALITY_THRESHOLD will be the upper bound on the build-side estimated row count + // for the rule to trigger. ENABLED stays false by default through the rollout. public static final String CONFIG_OF_MSE_RUNTIME_FILTER_ENABLED = "pinot.broker.mse.runtime.filter.enabled"; public static final boolean DEFAULT_MSE_RUNTIME_FILTER_ENABLED = false; public static final String CONFIG_OF_MSE_RUNTIME_FILTER_BUILD_CARDINALITY_THRESHOLD = From 242e2c031d395d08cd8700a326656ef071764d82 Mon Sep 17 00:00:00 2001 From: Tianle Zhang Date: Wed, 20 May 2026 15:47:56 -0700 Subject: [PATCH 3/6] Simplify MSE runtime filter config comment --- .../org/apache/pinot/spi/utils/CommonConstants.java | 10 +++------- 1 file changed, 3 insertions(+), 7 deletions(-) diff --git a/pinot-spi/src/main/java/org/apache/pinot/spi/utils/CommonConstants.java b/pinot-spi/src/main/java/org/apache/pinot/spi/utils/CommonConstants.java index dce7426f6d..f222f32f96 100644 --- a/pinot-spi/src/main/java/org/apache/pinot/spi/utils/CommonConstants.java +++ b/pinot-spi/src/main/java/org/apache/pinot/spi/utils/CommonConstants.java @@ -423,13 +423,9 @@ 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; - // Configuration surface for the planned MSE hash-join runtime filter (dynamic filter pushdown, - // OA-983). These keys are scaffolding only -- no runtime behavior is wired up in this PR. - // Follow-up PRs will plumb them into the planner rule, the build-side Bloom construction in - // HashJoinOperator, the probe-side RuntimeFilterOperator, and the leaf-stage segment-scan - // predicate. Once the feature is implemented, ENABLED will gate the optimizer rule and - // BUILD_CARDINALITY_THRESHOLD will be the upper bound on the build-side estimated row count - // for the rule to trigger. ENABLED stays false by default through the rollout. + // MSE hash-join runtime filter (dynamic filter pushdown). Scaffolding only -- not yet wired up. + // ENABLED gates the optimizer rule. BUILD_CARDINALITY_THRESHOLD caps the build-side estimated + // row count above which the rule will not trigger. public static final String CONFIG_OF_MSE_RUNTIME_FILTER_ENABLED = "pinot.broker.mse.runtime.filter.enabled"; public static final boolean DEFAULT_MSE_RUNTIME_FILTER_ENABLED = false; public static final String CONFIG_OF_MSE_RUNTIME_FILTER_BUILD_CARDINALITY_THRESHOLD = From 7f3446dbca200d397a25e491a7710886c34d9d18 Mon Sep 17 00:00:00 2001 From: Tianle Zhang Date: Wed, 20 May 2026 16:10:11 -0700 Subject: [PATCH 4/6] Drop ticket reference from ENABLE_RUNTIME_FILTER javadoc --- .../org/apache/pinot/calcite/rel/hint/PinotHintOptions.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/pinot-query-planner/src/main/java/org/apache/pinot/calcite/rel/hint/PinotHintOptions.java b/pinot-query-planner/src/main/java/org/apache/pinot/calcite/rel/hint/PinotHintOptions.java index a125d67340..b7ed37ea23 100644 --- a/pinot-query-planner/src/main/java/org/apache/pinot/calcite/rel/hint/PinotHintOptions.java +++ b/pinot-query-planner/src/main/java/org/apache/pinot/calcite/rel/hint/PinotHintOptions.java @@ -118,7 +118,7 @@ public static class JoinHintOptions { * {@code "false"} (case-insensitive). When set to a recognized value, takes precedence over the * broker-level {@code pinot.broker.mse.runtime.filter.enabled} config and any cardinality threshold * for this join. When unset -- or set to an unrecognized value -- the broker-level config governs - * behavior. See OA-983. + * behavior. */ public static final String ENABLE_RUNTIME_FILTER = "enable_runtime_filter"; From 846570f9620b0cfbb0a871c029560b9a2156615b Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Fri, 22 May 2026 00:13:43 +0000 Subject: [PATCH 5/6] Fix flaky QueryQuotaClusterIntegrationTest: use 500ms window for shouldFail tests MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Guava SmoothBursty rate limiter with tryAcquire() allows all queries when they arrive at exactly stableInterval/2 intervals (2x the configured rate). This is because canAcquire() uses <= so nextFreeTicketMicros==nowMicros always passes, while fractional stored permits (0.5 per step) keep nextFreeTicketMicros in sync. For a 25 QPS limit: stableInterval=40ms, stableInterval/2=20ms. Running 50 queries in 1000ms creates exactly 20ms spacing — the exact 2x boundary where no throttling occurs. All 50 queries pass, making assertTrue(failCount != 0) fail. Fix: use a 500ms window for shouldFail=true tests. This sends the same 50 queries in 500ms (100 QPS, 4x limit), reducing spacing to 10ms which is well below stableInterval/2=20ms, guaranteeing throttling at Q2+. Agent-Logs-Url: https://github.com/linkedin/pinot/sessions/e31c9274-85f0-4ecc-a185-f9b3215c7cb0 Co-authored-by: UOETianleZhang <27496716+UOETianleZhang@users.noreply.github.com> --- .../integration/tests/QueryQuotaClusterIntegrationTest.java | 6 +++++- 1 file changed, 5 insertions(+), 1 deletion(-) diff --git a/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/QueryQuotaClusterIntegrationTest.java b/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/QueryQuotaClusterIntegrationTest.java index f91a4a9148..320f13dbdb 100644 --- a/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/QueryQuotaClusterIntegrationTest.java +++ b/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); From 51e949cf26c528e8357bb06fd362810026979fa0 Mon Sep 17 00:00:00 2001 From: Tianle Zhang Date: Fri, 29 May 2026 12:47:37 -0700 Subject: [PATCH 6/6] Address review: rename runtime-filter configs, clarify scaffolding comment - CONFIG_OF_MSE_RUNTIME_FILTER_ENABLED -> ..._INJECTION_ENABLED (pinot.broker.mse.runtime.filter.injection.enabled), per review. - CONFIG_OF_MSE_RUNTIME_FILTER_BUILD_CARDINALITY_THRESHOLD -> ..._BUILD_ROWCOUNT_THRESHOLD (...build.rowcount.threshold): the gate is fed by RelMetadataQuery#getRowCount(), so the name now matches the source. - Reword the comment to make clear these are planned semantics; no code path reads them yet (consuming optimizer rule lands in a follow-up). Co-Authored-By: Claude Opus 4.8 (1M context) --- .../calcite/rel/hint/PinotHintOptions.java | 6 +++--- .../pinot/spi/utils/CommonConstants.java | 19 +++++++++++-------- 2 files changed, 14 insertions(+), 11 deletions(-) diff --git a/pinot-query-planner/src/main/java/org/apache/pinot/calcite/rel/hint/PinotHintOptions.java b/pinot-query-planner/src/main/java/org/apache/pinot/calcite/rel/hint/PinotHintOptions.java index b7ed37ea23..7ccf49fdaf 100644 --- a/pinot-query-planner/src/main/java/org/apache/pinot/calcite/rel/hint/PinotHintOptions.java +++ b/pinot-query-planner/src/main/java/org/apache/pinot/calcite/rel/hint/PinotHintOptions.java @@ -116,9 +116,9 @@ public static class JoinHintOptions { /** * 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.enabled} config and any cardinality threshold - * for this join. When unset -- or set to an unrecognized value -- the broker-level config governs - * behavior. + * 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"; diff --git a/pinot-spi/src/main/java/org/apache/pinot/spi/utils/CommonConstants.java b/pinot-spi/src/main/java/org/apache/pinot/spi/utils/CommonConstants.java index f222f32f96..b7d4e00bc0 100644 --- a/pinot-spi/src/main/java/org/apache/pinot/spi/utils/CommonConstants.java +++ b/pinot-spi/src/main/java/org/apache/pinot/spi/utils/CommonConstants.java @@ -423,14 +423,17 @@ 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 -- not yet wired up. - // ENABLED gates the optimizer rule. BUILD_CARDINALITY_THRESHOLD caps the build-side estimated - // row count above which the rule will not trigger. - public static final String CONFIG_OF_MSE_RUNTIME_FILTER_ENABLED = "pinot.broker.mse.runtime.filter.enabled"; - public static final boolean DEFAULT_MSE_RUNTIME_FILTER_ENABLED = false; - public static final String CONFIG_OF_MSE_RUNTIME_FILTER_BUILD_CARDINALITY_THRESHOLD = - "pinot.broker.mse.runtime.filter.build.cardinality.threshold"; - public static final long DEFAULT_MSE_RUNTIME_FILTER_BUILD_CARDINALITY_THRESHOLD = 1_000_000L; + // 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.