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); 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..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 @@ -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 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; + } } /** 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..5aaa92479a --- /dev/null +++ b/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 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)); + } +} 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..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,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";