) bf;
+ }
+
+ @Override
+ public Kind getKind() {
+ return Kind.BLOOM;
+ }
+
+ @Override
+ public DataType getDataType() {
+ return _dataType;
+ }
+
+ public void add(int value) {
+ _bloomFilter.put(value);
+ }
+
+ public void add(long value) {
+ _bloomFilter.put(value);
+ }
+
+ public void add(float value) {
+ // Match BloomFilterIdSet: encode by raw IEEE-754 int bits so the INT funnel hashes values
+ // with the same bit pattern to the same slot. Values that differ only by bit pattern
+ // (-0.0f vs 0.0f, distinct NaN payloads) are NOT collapsed by this encoding.
+ _bloomFilter.put(Float.floatToRawIntBits(value));
+ }
+
+ public void add(double value) {
+ _bloomFilter.put(Double.doubleToRawLongBits(value));
+ }
+
+ public void add(String value) {
+ _bloomFilter.put(value);
+ }
+
+ public void add(byte[] value) {
+ _bloomFilter.put(value);
+ }
+
+ public boolean mightContain(int value) {
+ return _bloomFilter.mightContain(value);
+ }
+
+ public boolean mightContain(long value) {
+ return _bloomFilter.mightContain(value);
+ }
+
+ public boolean mightContain(float value) {
+ return _bloomFilter.mightContain(Float.floatToRawIntBits(value));
+ }
+
+ public boolean mightContain(double value) {
+ return _bloomFilter.mightContain(Double.doubleToRawLongBits(value));
+ }
+
+ public boolean mightContain(String value) {
+ return _bloomFilter.mightContain(value);
+ }
+
+ public boolean mightContain(byte[] value) {
+ return _bloomFilter.mightContain(value);
+ }
+}
diff --git a/pinot-common/src/main/java/org/apache/pinot/common/request/context/predicate/Predicate.java b/pinot-common/src/main/java/org/apache/pinot/common/request/context/predicate/Predicate.java
index db88ed5c11..293e4cf4fa 100644
--- a/pinot-common/src/main/java/org/apache/pinot/common/request/context/predicate/Predicate.java
+++ b/pinot-common/src/main/java/org/apache/pinot/common/request/context/predicate/Predicate.java
@@ -39,7 +39,16 @@ enum Type {
JSON_MATCH,
IS_NULL,
IS_NOT_NULL(true),
- VECTOR_SIMILARITY;
+ VECTOR_SIMILARITY,
+ // Umbrella predicate for runtime filters (dynamic filter pushdown) injected by the MSE runtime
+ // filter feature. Carries a RuntimeFilter built from the hash-join build side; rows whose
+ // value the filter reports as "definitely not present" can be skipped at the leaf-stage scan.
+ // v1 only carries a Bloom filter (Kind.BLOOM); min/max and IN-list variants slot in as
+ // additional Kind values without a new Predicate.Type or new dispatch chain. Inclusive
+ // semantics: a row matches when the filter says "maybe present". False positives are allowed
+ // (downstream join still performs the exact match); false negatives must not occur.
+ // NULL handling is out of scope for v1; see RuntimeFilter for the NULL contract and TODO.
+ RUNTIME_FILTER;
private final boolean _exclusive;
diff --git a/pinot-common/src/main/java/org/apache/pinot/common/request/context/predicate/RuntimeFilter.java b/pinot-common/src/main/java/org/apache/pinot/common/request/context/predicate/RuntimeFilter.java
new file mode 100644
index 0000000000..06b8dd1b3e
--- /dev/null
+++ b/pinot-common/src/main/java/org/apache/pinot/common/request/context/predicate/RuntimeFilter.java
@@ -0,0 +1,72 @@
+/**
+ * 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.common.request.context.predicate;
+
+import org.apache.pinot.spi.data.FieldSpec.DataType;
+
+
+/**
+ * In-memory filter built at query time by the build side of a hash join and consumed by the probe
+ * side (typically at segment-scan time via {@link RuntimeFilterPredicate}). Carries enough
+ * type information for the evaluator factory to dispatch to a {@link Kind}-specific implementation
+ * without the rest of the engine knowing about the concrete filter form.
+ *
+ * v1 implements only {@link Kind#BLOOM} (see {@link BloomRuntimeFilter}). Min/max-range and
+ * IN-list variants slot in as additional {@link Kind} values + corresponding {@link RuntimeFilter}
+ * implementations without churn to the predicate type, the leaf-stage injection logic, or the
+ * planner rule.
+ *
+ * NULL handling (v1 contract)
+ * v1 does not model NULLs. The build side adds only non-null join keys to the filter, and the
+ * probe-side evaluator is applied to raw column values without consulting the column's null-value
+ * vector. This is safe for an inner equi-join because a NULL join key never matches anything, so
+ * such rows are dropped by the join regardless of the runtime filter. It is not safe to
+ * push a runtime filter onto a column whose NULLs must survive the scan (e.g. outer-join or
+ * NULL-aware semantics); the planner rule is responsible for not generating a runtime filter in
+ * those cases (see the type/eligibility enforcement that lives in the rule).
+ *
+ *
TODO(NULL support): define the end-to-end NULL contract and document it in both the code and
+ * the user docs before relaxing the planner-rule restriction above. Until then NULLs are simply
+ * absent from the filter and the probe side treats them as non-matching.
+ */
+public interface RuntimeFilter {
+
+ /**
+ * The concrete filter form this instance carries. Used by the runtime-filter evaluator factory
+ * to dispatch to the form-specific implementation.
+ */
+ enum Kind {
+ BLOOM
+ // MIN_MAX, // future
+ // IN_LIST, // future
+ }
+
+ /**
+ * Returns the {@link Kind} of this filter. The runtime-filter evaluator factory uses this to
+ * route to the form-specific implementation.
+ */
+ Kind getKind();
+
+ /**
+ * Returns the column data type this filter was built for. The evaluator factory enforces that
+ * this matches the column the predicate is being applied to; mismatches fail fast at
+ * construction time rather than producing garbage membership answers at query time.
+ */
+ DataType getDataType();
+}
diff --git a/pinot-common/src/main/java/org/apache/pinot/common/request/context/predicate/RuntimeFilterPredicate.java b/pinot-common/src/main/java/org/apache/pinot/common/request/context/predicate/RuntimeFilterPredicate.java
new file mode 100644
index 0000000000..7ad5efb389
--- /dev/null
+++ b/pinot-common/src/main/java/org/apache/pinot/common/request/context/predicate/RuntimeFilterPredicate.java
@@ -0,0 +1,64 @@
+/**
+ * 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.common.request.context.predicate;
+
+import java.util.Objects;
+import org.apache.pinot.common.request.context.ExpressionContext;
+
+
+/**
+ * Predicate matching values that the carried {@link RuntimeFilter} reports as possibly present
+ * (umbrella predicate for the MSE runtime-filter feature).
+ *
+ * Inclusive: a row matches when the carried filter's membership check returns true. False
+ * positives are permitted; downstream the actual hash join still performs the exact match. False
+ * negatives must not occur.
+ *
+ * The concrete filter form (Bloom for v1; future: min/max range, IN-list) is encoded in
+ * {@link RuntimeFilter#getKind()}. The evaluator factory dispatches on this {@link RuntimeFilter.Kind}
+ * so adding a new filter form does not require a new {@link Predicate.Type}.
+ *
+ * This predicate carries the filter object directly rather than a serialized form. It is built
+ * inside the broker / server JVM that runs the join build side and handed to the leaf-stage filter
+ * tree via {@code QueryContext} (wired up in a later change). Cross-worker transport is out of
+ * scope here.
+ */
+public class RuntimeFilterPredicate extends BasePredicate {
+
+ private final RuntimeFilter _filter;
+
+ public RuntimeFilterPredicate(ExpressionContext lhs, RuntimeFilter filter) {
+ super(lhs);
+ _filter = Objects.requireNonNull(filter, "filter");
+ }
+
+ @Override
+ public Type getType() {
+ return Type.RUNTIME_FILTER;
+ }
+
+ public RuntimeFilter getFilter() {
+ return _filter;
+ }
+
+ @Override
+ public String toString() {
+ return _lhs + " RUNTIME_FILTER(" + _filter.getKind() + ", " + _filter.getDataType() + ")";
+ }
+}
diff --git a/pinot-common/src/test/java/org/apache/pinot/common/request/context/predicate/BloomRuntimeFilterTest.java b/pinot-common/src/test/java/org/apache/pinot/common/request/context/predicate/BloomRuntimeFilterTest.java
new file mode 100644
index 0000000000..2dbbd2caaf
--- /dev/null
+++ b/pinot-common/src/test/java/org/apache/pinot/common/request/context/predicate/BloomRuntimeFilterTest.java
@@ -0,0 +1,181 @@
+/**
+ * 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.common.request.context.predicate;
+
+import org.apache.pinot.spi.data.FieldSpec.DataType;
+import org.testng.Assert;
+import org.testng.annotations.Test;
+
+
+public class BloomRuntimeFilterTest {
+
+ private static final double FPP = 0.01;
+ private static final int EXPECTED_INSERTIONS = 1000;
+
+ @Test
+ public void constructorRejectsUnsupportedDataType() {
+ Assert.assertThrows(IllegalArgumentException.class,
+ () -> new BloomRuntimeFilter(DataType.BIG_DECIMAL, EXPECTED_INSERTIONS, FPP));
+ Assert.assertThrows(IllegalArgumentException.class,
+ () -> new BloomRuntimeFilter(DataType.TIMESTAMP, EXPECTED_INSERTIONS, FPP));
+ Assert.assertThrows(IllegalArgumentException.class,
+ () -> new BloomRuntimeFilter(DataType.BOOLEAN, EXPECTED_INSERTIONS, FPP));
+ }
+
+ @Test
+ public void getDataTypeReportsConstructorArgument() {
+ BloomRuntimeFilter bf = new BloomRuntimeFilter(DataType.LONG, 16, FPP);
+ Assert.assertEquals(bf.getDataType(), DataType.LONG);
+ }
+
+ @Test
+ public void getKindReportsBloom() {
+ BloomRuntimeFilter bf = new BloomRuntimeFilter(DataType.INT, 16, FPP);
+ Assert.assertEquals(bf.getKind(), RuntimeFilter.Kind.BLOOM);
+ }
+
+ // ---------- INT ----------
+
+ @Test
+ public void intAddedValuesAlwaysMightContain() {
+ BloomRuntimeFilter bf = new BloomRuntimeFilter(DataType.INT, EXPECTED_INSERTIONS, FPP);
+ for (int i = 0; i < EXPECTED_INSERTIONS; i++) {
+ bf.add(i);
+ }
+ for (int i = 0; i < EXPECTED_INSERTIONS; i++) {
+ Assert.assertTrue(bf.mightContain(i), "false negative for added int " + i);
+ }
+ }
+
+ @Test
+ public void intFalsePositiveRateStaysNearConfigured() {
+ BloomRuntimeFilter bf = new BloomRuntimeFilter(DataType.INT, EXPECTED_INSERTIONS, FPP);
+ for (int i = 0; i < EXPECTED_INSERTIONS; i++) {
+ bf.add(i);
+ }
+ int falsePositives = 0;
+ int trials = 10_000;
+ for (int i = EXPECTED_INSERTIONS; i < EXPECTED_INSERTIONS + trials; i++) {
+ if (bf.mightContain(i)) {
+ falsePositives++;
+ }
+ }
+ // Allow 5x the configured fpp to keep the test stable across Guava versions / JDK changes.
+ double observedFpp = (double) falsePositives / trials;
+ Assert.assertTrue(observedFpp < FPP * 5,
+ "observed fpp " + observedFpp + " is unexpectedly high vs configured " + FPP);
+ }
+
+ // ---------- LONG ----------
+
+ @Test
+ public void longAddedValuesAlwaysMightContain() {
+ BloomRuntimeFilter bf = new BloomRuntimeFilter(DataType.LONG, EXPECTED_INSERTIONS, FPP);
+ for (long v = Integer.MAX_VALUE; v < Integer.MAX_VALUE + EXPECTED_INSERTIONS; v++) {
+ bf.add(v);
+ }
+ for (long v = Integer.MAX_VALUE; v < Integer.MAX_VALUE + EXPECTED_INSERTIONS; v++) {
+ Assert.assertTrue(bf.mightContain(v), "false negative for added long " + v);
+ }
+ }
+
+ // ---------- FLOAT ----------
+
+ @Test
+ public void floatAddedValuesAlwaysMightContain() {
+ BloomRuntimeFilter bf = new BloomRuntimeFilter(DataType.FLOAT, EXPECTED_INSERTIONS, FPP);
+ for (int i = 0; i < EXPECTED_INSERTIONS; i++) {
+ bf.add((float) (i + 0.5));
+ }
+ for (int i = 0; i < EXPECTED_INSERTIONS; i++) {
+ Assert.assertTrue(bf.mightContain((float) (i + 0.5)), "false negative for added float " + i);
+ }
+ }
+
+ @Test
+ public void floatRawBitsHashingTreatsSameBitPatternAsSame() {
+ // Lock in the raw-bits hashing contract: a float with the same IEEE-754 bit pattern as an
+ // added value always reports mightContain == true. We do NOT assert anything about -0.0f vs
+ // 0.0f (they have different bit patterns; either outcome is permissible under Bloom semantics)
+ // -- the goal here is to pin down the "same bits -> same hash" guarantee.
+ BloomRuntimeFilter bf = new BloomRuntimeFilter(DataType.FLOAT, EXPECTED_INSERTIONS, FPP);
+ bf.add(1.5f);
+ bf.add(Float.intBitsToFloat(0x7fc00001)); // a non-canonical NaN with a specific payload
+ Assert.assertTrue(bf.mightContain(1.5f));
+ Assert.assertTrue(bf.mightContain(Float.intBitsToFloat(0x7fc00001)));
+ // Smoke-check the negative-zero API call: just exercise the method, do not assert its result
+ // (it is allowed to collide with 0.0f as a false positive but is not guaranteed to).
+ bf.mightContain(-0.0f);
+ }
+
+ // ---------- DOUBLE ----------
+
+ @Test
+ public void doubleAddedValuesAlwaysMightContain() {
+ BloomRuntimeFilter bf = new BloomRuntimeFilter(DataType.DOUBLE, EXPECTED_INSERTIONS, FPP);
+ for (int i = 0; i < EXPECTED_INSERTIONS; i++) {
+ bf.add(i + 0.25);
+ }
+ for (int i = 0; i < EXPECTED_INSERTIONS; i++) {
+ Assert.assertTrue(bf.mightContain(i + 0.25), "false negative for added double " + i);
+ }
+ }
+
+ // ---------- STRING ----------
+
+ @Test
+ public void stringAddedValuesAlwaysMightContain() {
+ BloomRuntimeFilter bf = new BloomRuntimeFilter(DataType.STRING, EXPECTED_INSERTIONS, FPP);
+ for (int i = 0; i < EXPECTED_INSERTIONS; i++) {
+ bf.add("key-" + i);
+ }
+ for (int i = 0; i < EXPECTED_INSERTIONS; i++) {
+ Assert.assertTrue(bf.mightContain("key-" + i), "false negative for added string key-" + i);
+ }
+ }
+
+ @Test
+ public void stringEmptyStringIsHashableAndDistinguishableFromNonEmpty() {
+ BloomRuntimeFilter bf = new BloomRuntimeFilter(DataType.STRING, EXPECTED_INSERTIONS, FPP);
+ bf.add("");
+ Assert.assertTrue(bf.mightContain(""));
+ }
+
+ // ---------- BYTES ----------
+
+ @Test
+ public void bytesAddedValuesAlwaysMightContain() {
+ BloomRuntimeFilter bf = new BloomRuntimeFilter(DataType.BYTES, EXPECTED_INSERTIONS, FPP);
+ for (int i = 0; i < EXPECTED_INSERTIONS; i++) {
+ bf.add(new byte[]{(byte) i, (byte) (i + 1)});
+ }
+ for (int i = 0; i < EXPECTED_INSERTIONS; i++) {
+ Assert.assertTrue(bf.mightContain(new byte[]{(byte) i, (byte) (i + 1)}),
+ "false negative for added byte[] index " + i);
+ }
+ }
+
+ @Test
+ public void bytesContentEqualityHashesIdentically() {
+ // Two distinct byte[] objects with identical content must hash the same way (not by reference).
+ BloomRuntimeFilter bf = new BloomRuntimeFilter(DataType.BYTES, EXPECTED_INSERTIONS, FPP);
+ bf.add(new byte[]{1, 2, 3});
+ Assert.assertTrue(bf.mightContain(new byte[]{1, 2, 3}));
+ }
+}
diff --git a/pinot-core/src/main/java/org/apache/pinot/core/operator/filter/predicate/BloomRuntimeFilterEvaluatorFactory.java b/pinot-core/src/main/java/org/apache/pinot/core/operator/filter/predicate/BloomRuntimeFilterEvaluatorFactory.java
new file mode 100644
index 0000000000..3b71cc42a5
--- /dev/null
+++ b/pinot-core/src/main/java/org/apache/pinot/core/operator/filter/predicate/BloomRuntimeFilterEvaluatorFactory.java
@@ -0,0 +1,293 @@
+/**
+ * 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.core.operator.filter.predicate;
+
+import it.unimi.dsi.fastutil.ints.IntOpenHashSet;
+import it.unimi.dsi.fastutil.ints.IntSet;
+import java.util.Arrays;
+import org.apache.pinot.common.request.context.predicate.BloomRuntimeFilter;
+import org.apache.pinot.common.request.context.predicate.RuntimeFilterPredicate;
+import org.apache.pinot.segment.spi.index.reader.Dictionary;
+import org.apache.pinot.spi.data.FieldSpec.DataType;
+
+
+/**
+ * Bloom-specific implementation of the runtime-filter evaluator. Reached via
+ * {@link RuntimeFilterPredicateEvaluatorFactory} after it has dispatched on
+ * {@code predicate.getFilter().getKind()}.
+ *
+ * Both raw-value and dictionary-based evaluation are supported. False positives are allowed
+ * (downstream join still performs exact matching) but false negatives must not occur. Each
+ * per-type raw-value evaluator delegates directly to the matching
+ * {@link BloomRuntimeFilter#mightContain} overload so the filter's data-type-specific funnel
+ * selects the correct hashing. The dictionary-based evaluator pre-materialises matching dict ids
+ * at construction time by probing the Bloom filter for every dictionary entry, then answers
+ * {@code applySV(int dictId)} from that set.
+ */
+public class BloomRuntimeFilterEvaluatorFactory {
+ private BloomRuntimeFilterEvaluatorFactory() {
+ }
+
+ /**
+ * Builds a raw-value evaluator for the given {@code dataType} against the
+ * {@link BloomRuntimeFilter} carried inside {@code predicate}. Supported column types: INT, LONG,
+ * FLOAT, DOUBLE, STRING, BYTES. Other data types throw {@link IllegalArgumentException} at
+ * construction; callers must not request the Bloom runtime filter for unsupported column types.
+ *
+ * Callers must ensure {@code predicate.getFilter()} is a {@link BloomRuntimeFilter} — this is
+ * the responsibility of {@link RuntimeFilterPredicateEvaluatorFactory}, which dispatches on
+ * {@code Kind} before reaching this factory.
+ */
+ public static PredicateEvaluator newRawValueBasedEvaluator(RuntimeFilterPredicate predicate, DataType dataType) {
+ BloomRuntimeFilter filter = (BloomRuntimeFilter) predicate.getFilter();
+ if (filter.getDataType() != dataType) {
+ // Hashing relies on the funnel matching the column type; a mismatch would silently produce
+ // garbage membership answers. Fail fast at evaluator construction so the bad wiring shows up
+ // during planning instead of at query time.
+ throw new IllegalArgumentException(
+ "BloomRuntimeFilter dataType " + filter.getDataType() + " does not match evaluator dataType " + dataType);
+ }
+ switch (dataType) {
+ case INT:
+ return new IntBloomRuntimeFilterEvaluator(predicate);
+ case LONG:
+ return new LongBloomRuntimeFilterEvaluator(predicate);
+ case FLOAT:
+ return new FloatBloomRuntimeFilterEvaluator(predicate);
+ case DOUBLE:
+ return new DoubleBloomRuntimeFilterEvaluator(predicate);
+ case STRING:
+ return new StringBloomRuntimeFilterEvaluator(predicate);
+ case BYTES:
+ return new BytesBloomRuntimeFilterEvaluator(predicate);
+ default:
+ throw new IllegalArgumentException("Bloom runtime filter does not support data type: " + dataType);
+ }
+ }
+
+ /**
+ * Builds a dictionary-based evaluator for the given {@code dataType} against the
+ * {@link BloomRuntimeFilter} carried inside {@code predicate}. The dictionary is probed once at
+ * construction time: every dictId whose value passes {@link BloomRuntimeFilter#mightContain}
+ * lands in the matching-dictId set, and {@code applySV(int)} answers from that set thereafter.
+ *
+ * Cost is O(dictionary.length()) Bloom probes at construction. For high-cardinality columns
+ * this is a real cost — the planner is responsible for choosing dictionary-based evaluation only
+ * when it pays off. Once built, downstream scan paths get the standard
+ * {@code _matchingDictIds} / {@code _alwaysTrue} / {@code _alwaysFalse} contract.
+ *
+ * False-positive semantics carry over: a dictId may end up in the matching set due to a Bloom
+ * false positive, but no true match will be excluded.
+ */
+ public static BaseDictionaryBasedPredicateEvaluator newDictionaryBasedEvaluator(
+ RuntimeFilterPredicate predicate, Dictionary dictionary, DataType dataType) {
+ BloomRuntimeFilter filter = (BloomRuntimeFilter) predicate.getFilter();
+ if (filter.getDataType() != dataType) {
+ throw new IllegalArgumentException(
+ "BloomRuntimeFilter dataType " + filter.getDataType() + " does not match evaluator dataType " + dataType);
+ }
+ return new DictionaryBasedBloomRuntimeFilterEvaluator(predicate, dictionary, dataType);
+ }
+
+ private abstract static class BaseBloomRuntimeFilterEvaluator extends BaseRawValueBasedPredicateEvaluator {
+ final BloomRuntimeFilter _filter;
+
+ BaseBloomRuntimeFilterEvaluator(RuntimeFilterPredicate predicate) {
+ super(predicate);
+ _filter = (BloomRuntimeFilter) predicate.getFilter();
+ }
+ }
+
+ private static final class IntBloomRuntimeFilterEvaluator extends BaseBloomRuntimeFilterEvaluator {
+ IntBloomRuntimeFilterEvaluator(RuntimeFilterPredicate predicate) {
+ super(predicate);
+ }
+
+ @Override
+ public DataType getDataType() {
+ return DataType.INT;
+ }
+
+ @Override
+ public boolean applySV(int value) {
+ return _filter.mightContain(value);
+ }
+ }
+
+ private static final class LongBloomRuntimeFilterEvaluator extends BaseBloomRuntimeFilterEvaluator {
+ LongBloomRuntimeFilterEvaluator(RuntimeFilterPredicate predicate) {
+ super(predicate);
+ }
+
+ @Override
+ public DataType getDataType() {
+ return DataType.LONG;
+ }
+
+ @Override
+ public boolean applySV(long value) {
+ return _filter.mightContain(value);
+ }
+ }
+
+ private static final class FloatBloomRuntimeFilterEvaluator extends BaseBloomRuntimeFilterEvaluator {
+ FloatBloomRuntimeFilterEvaluator(RuntimeFilterPredicate predicate) {
+ super(predicate);
+ }
+
+ @Override
+ public DataType getDataType() {
+ return DataType.FLOAT;
+ }
+
+ @Override
+ public boolean applySV(float value) {
+ return _filter.mightContain(value);
+ }
+ }
+
+ private static final class DoubleBloomRuntimeFilterEvaluator extends BaseBloomRuntimeFilterEvaluator {
+ DoubleBloomRuntimeFilterEvaluator(RuntimeFilterPredicate predicate) {
+ super(predicate);
+ }
+
+ @Override
+ public DataType getDataType() {
+ return DataType.DOUBLE;
+ }
+
+ @Override
+ public boolean applySV(double value) {
+ return _filter.mightContain(value);
+ }
+ }
+
+ private static final class StringBloomRuntimeFilterEvaluator extends BaseBloomRuntimeFilterEvaluator {
+ StringBloomRuntimeFilterEvaluator(RuntimeFilterPredicate predicate) {
+ super(predicate);
+ }
+
+ @Override
+ public DataType getDataType() {
+ return DataType.STRING;
+ }
+
+ @Override
+ public boolean applySV(String value) {
+ return _filter.mightContain(value);
+ }
+ }
+
+ private static final class BytesBloomRuntimeFilterEvaluator extends BaseBloomRuntimeFilterEvaluator {
+ BytesBloomRuntimeFilterEvaluator(RuntimeFilterPredicate predicate) {
+ super(predicate);
+ }
+
+ @Override
+ public DataType getDataType() {
+ return DataType.BYTES;
+ }
+
+ @Override
+ public boolean applySV(byte[] value) {
+ return _filter.mightContain(value);
+ }
+ }
+
+ private static final class DictionaryBasedBloomRuntimeFilterEvaluator extends BaseDictionaryBasedPredicateEvaluator {
+ private final IntSet _matchingDictIdSet;
+
+ DictionaryBasedBloomRuntimeFilterEvaluator(
+ RuntimeFilterPredicate predicate, Dictionary dictionary, DataType dataType) {
+ super(predicate, dictionary);
+ BloomRuntimeFilter filter = (BloomRuntimeFilter) predicate.getFilter();
+ int dictLen = dictionary.length();
+ _matchingDictIdSet = new IntOpenHashSet();
+ switch (dataType) {
+ case INT:
+ for (int dictId = 0; dictId < dictLen; dictId++) {
+ if (filter.mightContain(dictionary.getIntValue(dictId))) {
+ _matchingDictIdSet.add(dictId);
+ }
+ }
+ break;
+ case LONG:
+ for (int dictId = 0; dictId < dictLen; dictId++) {
+ if (filter.mightContain(dictionary.getLongValue(dictId))) {
+ _matchingDictIdSet.add(dictId);
+ }
+ }
+ break;
+ case FLOAT:
+ for (int dictId = 0; dictId < dictLen; dictId++) {
+ if (filter.mightContain(dictionary.getFloatValue(dictId))) {
+ _matchingDictIdSet.add(dictId);
+ }
+ }
+ break;
+ case DOUBLE:
+ for (int dictId = 0; dictId < dictLen; dictId++) {
+ if (filter.mightContain(dictionary.getDoubleValue(dictId))) {
+ _matchingDictIdSet.add(dictId);
+ }
+ }
+ break;
+ case STRING:
+ for (int dictId = 0; dictId < dictLen; dictId++) {
+ if (filter.mightContain(dictionary.getStringValue(dictId))) {
+ _matchingDictIdSet.add(dictId);
+ }
+ }
+ break;
+ case BYTES:
+ for (int dictId = 0; dictId < dictLen; dictId++) {
+ if (filter.mightContain(dictionary.getBytesValue(dictId))) {
+ _matchingDictIdSet.add(dictId);
+ }
+ }
+ break;
+ default:
+ throw new IllegalArgumentException("Bloom runtime filter does not support data type: " + dataType);
+ }
+ int numMatching = _matchingDictIdSet.size();
+ if (numMatching == 0) {
+ _alwaysFalse = true;
+ } else if (numMatching == dictLen) {
+ _alwaysTrue = true;
+ }
+ }
+
+ @Override
+ protected int[] calculateMatchingDictIds() {
+ int[] matchingDictIds = _matchingDictIdSet.toIntArray();
+ Arrays.sort(matchingDictIds);
+ return matchingDictIds;
+ }
+
+ @Override
+ public int getNumMatchingItems() {
+ return _matchingDictIdSet.size();
+ }
+
+ @Override
+ public boolean applySV(int dictId) {
+ return _matchingDictIdSet.contains(dictId);
+ }
+ }
+}
diff --git a/pinot-core/src/main/java/org/apache/pinot/core/operator/filter/predicate/PredicateEvaluatorProvider.java b/pinot-core/src/main/java/org/apache/pinot/core/operator/filter/predicate/PredicateEvaluatorProvider.java
index 519168818b..cbd863b7d8 100644
--- a/pinot-core/src/main/java/org/apache/pinot/core/operator/filter/predicate/PredicateEvaluatorProvider.java
+++ b/pinot-core/src/main/java/org/apache/pinot/core/operator/filter/predicate/PredicateEvaluatorProvider.java
@@ -26,6 +26,7 @@
import org.apache.pinot.common.request.context.predicate.Predicate;
import org.apache.pinot.common.request.context.predicate.RangePredicate;
import org.apache.pinot.common.request.context.predicate.RegexpLikePredicate;
+import org.apache.pinot.common.request.context.predicate.RuntimeFilterPredicate;
import org.apache.pinot.core.query.request.context.QueryContext;
import org.apache.pinot.segment.spi.datasource.DataSource;
import org.apache.pinot.segment.spi.index.reader.Dictionary;
@@ -66,6 +67,9 @@ public static PredicateEvaluator getPredicateEvaluator(Predicate predicate, @Nul
case REGEXP_LIKE:
return RegexpLikePredicateEvaluatorFactory
.newDictionaryBasedEvaluator((RegexpLikePredicate) predicate, dictionary, dataType);
+ case RUNTIME_FILTER:
+ return RuntimeFilterPredicateEvaluatorFactory
+ .newDictionaryBasedEvaluator((RuntimeFilterPredicate) predicate, dictionary, dataType);
default:
throw new UnsupportedOperationException("Unsupported predicate type: " + predicate.getType());
}
@@ -85,6 +89,9 @@ public static PredicateEvaluator getPredicateEvaluator(Predicate predicate, @Nul
case REGEXP_LIKE:
return RegexpLikePredicateEvaluatorFactory
.newRawValueBasedEvaluator((RegexpLikePredicate) predicate, dataType);
+ case RUNTIME_FILTER:
+ return RuntimeFilterPredicateEvaluatorFactory
+ .newRawValueBasedEvaluator((RuntimeFilterPredicate) predicate, dataType);
default:
throw new UnsupportedOperationException("Unsupported predicate type: " + predicate.getType());
}
diff --git a/pinot-core/src/main/java/org/apache/pinot/core/operator/filter/predicate/RuntimeFilterPredicateEvaluatorFactory.java b/pinot-core/src/main/java/org/apache/pinot/core/operator/filter/predicate/RuntimeFilterPredicateEvaluatorFactory.java
new file mode 100644
index 0000000000..72b722f9b5
--- /dev/null
+++ b/pinot-core/src/main/java/org/apache/pinot/core/operator/filter/predicate/RuntimeFilterPredicateEvaluatorFactory.java
@@ -0,0 +1,76 @@
+/**
+ * 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.core.operator.filter.predicate;
+
+import org.apache.pinot.common.request.context.predicate.RuntimeFilter;
+import org.apache.pinot.common.request.context.predicate.RuntimeFilterPredicate;
+import org.apache.pinot.segment.spi.index.reader.Dictionary;
+import org.apache.pinot.spi.data.FieldSpec.DataType;
+
+
+/**
+ * Top-level dispatcher for {@link RuntimeFilterPredicate} evaluation. Routes to a form-specific
+ * factory based on {@link RuntimeFilter#getKind()}.
+ *
+ * This indirection is what lets the rest of the engine (planner rule, leaf injection, provider
+ * dispatch) stay agnostic of the concrete filter form. Adding a new form means adding a
+ * {@link RuntimeFilter} implementation, a per-form evaluator factory, and one new case arm here.
+ * No new {@link org.apache.pinot.common.request.context.predicate.Predicate.Type}, no churn to
+ * {@code FilterPlanNode}, no churn to {@code PredicateEvaluatorProvider}.
+ */
+public class RuntimeFilterPredicateEvaluatorFactory {
+ private RuntimeFilterPredicateEvaluatorFactory() {
+ }
+
+ /**
+ * Builds a raw-value evaluator for the given {@code dataType} against the {@link RuntimeFilter}
+ * carried inside {@code predicate}, dispatching on {@link RuntimeFilter#getKind()}.
+ *
+ * @throws IllegalArgumentException if the filter kind is not supported (e.g. min-max or IN-list
+ * requested before its implementation lands)
+ */
+ public static PredicateEvaluator newRawValueBasedEvaluator(RuntimeFilterPredicate predicate, DataType dataType) {
+ RuntimeFilter.Kind kind = predicate.getFilter().getKind();
+ switch (kind) {
+ case BLOOM:
+ return BloomRuntimeFilterEvaluatorFactory.newRawValueBasedEvaluator(predicate, dataType);
+ default:
+ throw new IllegalArgumentException("Unsupported runtime filter kind: " + kind);
+ }
+ }
+
+ /**
+ * Builds a dictionary-based evaluator for the given {@code dataType} against the
+ * {@link RuntimeFilter} carried inside {@code predicate}, dispatching on
+ * {@link RuntimeFilter#getKind()}.
+ *
+ * @throws IllegalArgumentException if the filter kind is not supported (e.g. min-max or IN-list
+ * requested before its implementation lands)
+ */
+ public static BaseDictionaryBasedPredicateEvaluator newDictionaryBasedEvaluator(
+ RuntimeFilterPredicate predicate, Dictionary dictionary, DataType dataType) {
+ RuntimeFilter.Kind kind = predicate.getFilter().getKind();
+ switch (kind) {
+ case BLOOM:
+ return BloomRuntimeFilterEvaluatorFactory.newDictionaryBasedEvaluator(predicate, dictionary, dataType);
+ default:
+ throw new IllegalArgumentException("Unsupported runtime filter kind: " + kind);
+ }
+ }
+}
diff --git a/pinot-core/src/test/java/org/apache/pinot/core/operator/filter/predicate/BloomRuntimeFilterEvaluatorFactoryTest.java b/pinot-core/src/test/java/org/apache/pinot/core/operator/filter/predicate/BloomRuntimeFilterEvaluatorFactoryTest.java
new file mode 100644
index 0000000000..e60f25e862
--- /dev/null
+++ b/pinot-core/src/test/java/org/apache/pinot/core/operator/filter/predicate/BloomRuntimeFilterEvaluatorFactoryTest.java
@@ -0,0 +1,335 @@
+/**
+ * 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.core.operator.filter.predicate;
+
+import java.util.ArrayList;
+import java.util.List;
+import org.apache.pinot.common.request.context.ExpressionContext;
+import org.apache.pinot.common.request.context.predicate.BloomRuntimeFilter;
+import org.apache.pinot.common.request.context.predicate.Predicate;
+import org.apache.pinot.common.request.context.predicate.RuntimeFilterPredicate;
+import org.apache.pinot.segment.spi.index.reader.Dictionary;
+import org.apache.pinot.spi.data.FieldSpec.DataType;
+import org.mockito.Mockito;
+import org.testng.Assert;
+import org.testng.annotations.Test;
+
+
+public class BloomRuntimeFilterEvaluatorFactoryTest {
+
+ private static final ExpressionContext COLUMN = ExpressionContext.forIdentifier("column");
+ private static final int EXPECTED_INSERTIONS = 1000;
+ private static final double FPP = 0.01;
+ private static final int NUM_KEYS = 200;
+ private static final int NUM_NEGATIVE_PROBES = 5000;
+ private static final double FP_TOLERANCE = FPP * 5; // matches BloomRuntimeFilterTest tolerance
+
+ // ---------- Factory wiring ----------
+
+ @Test
+ public void factoryRejectsDataTypeMismatch() {
+ BloomRuntimeFilter bloom = new BloomRuntimeFilter(DataType.INT, EXPECTED_INSERTIONS, FPP);
+ RuntimeFilterPredicate predicate = new RuntimeFilterPredicate(COLUMN, bloom);
+ Assert.assertThrows(IllegalArgumentException.class,
+ () -> BloomRuntimeFilterEvaluatorFactory.newRawValueBasedEvaluator(predicate, DataType.LONG));
+ }
+
+ @Test
+ public void factoryRejectsUnsupportedDataType() {
+ BloomRuntimeFilter bloom = new BloomRuntimeFilter(DataType.STRING, EXPECTED_INSERTIONS, FPP);
+ RuntimeFilterPredicate predicate = new RuntimeFilterPredicate(COLUMN, bloom);
+ // STRING bloom but BIG_DECIMAL evaluator -- the dataType-mismatch guard fires first.
+ Assert.assertThrows(IllegalArgumentException.class,
+ () -> BloomRuntimeFilterEvaluatorFactory.newRawValueBasedEvaluator(predicate, DataType.BIG_DECIMAL));
+ }
+
+ @Test
+ public void evaluatorReportsExpectedMetadata() {
+ BloomRuntimeFilter bloom = new BloomRuntimeFilter(DataType.LONG, EXPECTED_INSERTIONS, FPP);
+ RuntimeFilterPredicate predicate = new RuntimeFilterPredicate(COLUMN, bloom);
+ PredicateEvaluator evaluator =
+ BloomRuntimeFilterEvaluatorFactory.newRawValueBasedEvaluator(predicate, DataType.LONG);
+
+ Assert.assertSame(evaluator.getPredicate(), predicate);
+ Assert.assertEquals(evaluator.getPredicateType(), Predicate.Type.RUNTIME_FILTER);
+ Assert.assertEquals(evaluator.getDataType(), DataType.LONG);
+ Assert.assertFalse(evaluator.isDictionaryBased(), "raw-value evaluator must not be dictionary-based");
+ Assert.assertFalse(evaluator.isExclusive(), "RUNTIME_FILTER is an inclusive predicate");
+ Assert.assertFalse(evaluator.isAlwaysTrue());
+ Assert.assertFalse(evaluator.isAlwaysFalse());
+ }
+
+ // ---------- Per-type membership ----------
+
+ @Test
+ public void intEvaluatorHasNoFalseNegativesAndLowFalsePositiveRate() {
+ BloomRuntimeFilter bloom = new BloomRuntimeFilter(DataType.INT, EXPECTED_INSERTIONS, FPP);
+ int[] added = new int[NUM_KEYS];
+ for (int i = 0; i < NUM_KEYS; i++) {
+ added[i] = i * 13;
+ bloom.add(added[i]);
+ }
+ PredicateEvaluator evaluator = newEvaluator(bloom, DataType.INT);
+
+ for (int v : added) {
+ Assert.assertTrue(evaluator.applySV(v), "false negative for added int " + v);
+ }
+ int falsePositives = 0;
+ for (int i = 1; i <= NUM_NEGATIVE_PROBES; i++) {
+ int probe = -i; // disjoint from added (which are non-negative multiples of 13)
+ if (evaluator.applySV(probe)) {
+ falsePositives++;
+ }
+ }
+ double fp = (double) falsePositives / NUM_NEGATIVE_PROBES;
+ Assert.assertTrue(fp < FP_TOLERANCE, "int fp " + fp + " unexpectedly high");
+ }
+
+ @Test
+ public void longEvaluatorHasNoFalseNegatives() {
+ BloomRuntimeFilter bloom = new BloomRuntimeFilter(DataType.LONG, EXPECTED_INSERTIONS, FPP);
+ List added = new ArrayList<>(NUM_KEYS);
+ for (int i = 0; i < NUM_KEYS; i++) {
+ long v = Long.MAX_VALUE - i;
+ added.add(v);
+ bloom.add(v);
+ }
+ PredicateEvaluator evaluator = newEvaluator(bloom, DataType.LONG);
+ for (long v : added) {
+ Assert.assertTrue(evaluator.applySV(v), "false negative for added long " + v);
+ }
+ }
+
+ @Test
+ public void floatEvaluatorRoundTrip() {
+ BloomRuntimeFilter bloom = new BloomRuntimeFilter(DataType.FLOAT, EXPECTED_INSERTIONS, FPP);
+ for (int i = 0; i < NUM_KEYS; i++) {
+ bloom.add(i + 0.125f);
+ }
+ PredicateEvaluator evaluator = newEvaluator(bloom, DataType.FLOAT);
+ for (int i = 0; i < NUM_KEYS; i++) {
+ Assert.assertTrue(evaluator.applySV(i + 0.125f), "false negative for added float index " + i);
+ }
+ }
+
+ @Test
+ public void doubleEvaluatorRoundTrip() {
+ BloomRuntimeFilter bloom = new BloomRuntimeFilter(DataType.DOUBLE, EXPECTED_INSERTIONS, FPP);
+ for (int i = 0; i < NUM_KEYS; i++) {
+ bloom.add(i + 0.0625);
+ }
+ PredicateEvaluator evaluator = newEvaluator(bloom, DataType.DOUBLE);
+ for (int i = 0; i < NUM_KEYS; i++) {
+ Assert.assertTrue(evaluator.applySV(i + 0.0625), "false negative for added double index " + i);
+ }
+ }
+
+ @Test
+ public void stringEvaluatorRoundTrip() {
+ BloomRuntimeFilter bloom = new BloomRuntimeFilter(DataType.STRING, EXPECTED_INSERTIONS, FPP);
+ for (int i = 0; i < NUM_KEYS; i++) {
+ bloom.add("k-" + i);
+ }
+ PredicateEvaluator evaluator = newEvaluator(bloom, DataType.STRING);
+ for (int i = 0; i < NUM_KEYS; i++) {
+ Assert.assertTrue(evaluator.applySV("k-" + i), "false negative for added string k-" + i);
+ }
+ }
+
+ @Test
+ public void bytesEvaluatorRoundTrip() {
+ BloomRuntimeFilter bloom = new BloomRuntimeFilter(DataType.BYTES, EXPECTED_INSERTIONS, FPP);
+ for (int i = 0; i < NUM_KEYS; i++) {
+ bloom.add(new byte[]{(byte) i, (byte) (i + 1), (byte) (i + 2)});
+ }
+ PredicateEvaluator evaluator = newEvaluator(bloom, DataType.BYTES);
+ for (int i = 0; i < NUM_KEYS; i++) {
+ Assert.assertTrue(evaluator.applySV(new byte[]{(byte) i, (byte) (i + 1), (byte) (i + 2)}),
+ "false negative for added bytes index " + i);
+ }
+ }
+
+ // ---------- Wrong-overload defaults from base class ----------
+
+ @Test
+ public void wrongTypeOverloadThrowsFromBaseClass() {
+ // INT evaluator only overrides applySV(int); other overloads inherit the base class's
+ // UnsupportedOperationException. Lock in that contract so callers can rely on it.
+ BloomRuntimeFilter bloom = new BloomRuntimeFilter(DataType.INT, EXPECTED_INSERTIONS, FPP);
+ bloom.add(7);
+ PredicateEvaluator evaluator = newEvaluator(bloom, DataType.INT);
+
+ Assert.assertThrows(UnsupportedOperationException.class, () -> evaluator.applySV(7L));
+ Assert.assertThrows(UnsupportedOperationException.class, () -> evaluator.applySV(7.0f));
+ Assert.assertThrows(UnsupportedOperationException.class, () -> evaluator.applySV(7.0));
+ Assert.assertThrows(UnsupportedOperationException.class, () -> evaluator.applySV("7"));
+ Assert.assertThrows(UnsupportedOperationException.class, () -> evaluator.applySV(new byte[]{7}));
+ }
+
+ // ---------- Multi-value default loops through SV ----------
+
+ @Test
+ public void intMultiValueMatchesIfAnyValueIsInBloom() {
+ BloomRuntimeFilter bloom = new BloomRuntimeFilter(DataType.INT, EXPECTED_INSERTIONS, FPP);
+ bloom.add(42);
+ PredicateEvaluator evaluator = newEvaluator(bloom, DataType.INT);
+
+ int[] mvWithMatch = new int[]{-1, -2, 42, -3};
+ Assert.assertTrue(evaluator.applyMV(mvWithMatch, mvWithMatch.length));
+
+ int[] mvNoMatch = new int[]{-1, -2, -3};
+ // Bloom may produce false positives in principle, but with one inserted key in a 1000-capacity
+ // filter the chance of a hit on three negative probes is well under the test tolerance.
+ Assert.assertFalse(evaluator.applyMV(mvNoMatch, mvNoMatch.length));
+ }
+
+ // ---------- Provider dispatch ----------
+
+ @Test
+ public void providerRoutesRawValueBasedRuntimeFilterToFactory() {
+ BloomRuntimeFilter bloom = new BloomRuntimeFilter(DataType.LONG, EXPECTED_INSERTIONS, FPP);
+ RuntimeFilterPredicate predicate = new RuntimeFilterPredicate(COLUMN, bloom);
+ PredicateEvaluator evaluator =
+ PredicateEvaluatorProvider.getPredicateEvaluator(predicate, null, DataType.LONG);
+ Assert.assertEquals(evaluator.getPredicateType(), Predicate.Type.RUNTIME_FILTER);
+ Assert.assertEquals(evaluator.getDataType(), DataType.LONG);
+ }
+
+ @Test
+ public void providerRoutesDictionaryBasedRuntimeFilterToFactory() {
+ BloomRuntimeFilter bloom = new BloomRuntimeFilter(DataType.INT, EXPECTED_INSERTIONS, FPP);
+ bloom.add(42);
+ RuntimeFilterPredicate predicate = new RuntimeFilterPredicate(COLUMN, bloom);
+ Dictionary dictionary = Mockito.mock(Dictionary.class);
+ Mockito.when(dictionary.length()).thenReturn(2);
+ Mockito.when(dictionary.getIntValue(0)).thenReturn(42);
+ Mockito.when(dictionary.getIntValue(1)).thenReturn(-1);
+ PredicateEvaluator evaluator =
+ PredicateEvaluatorProvider.getPredicateEvaluator(predicate, dictionary, DataType.INT);
+ Assert.assertTrue(evaluator.isDictionaryBased(), "dict-routed evaluator must be dictionary-based");
+ Assert.assertEquals(evaluator.getPredicateType(), Predicate.Type.RUNTIME_FILTER);
+ }
+
+ // ---------- Dictionary-based evaluator ----------
+
+ @Test
+ public void dictEvaluatorMatchesDictIdsWithBloomHits() {
+ BloomRuntimeFilter bloom = new BloomRuntimeFilter(DataType.INT, EXPECTED_INSERTIONS, FPP);
+ bloom.add(10);
+ bloom.add(20);
+ RuntimeFilterPredicate predicate = new RuntimeFilterPredicate(COLUMN, bloom);
+
+ Dictionary dictionary = Mockito.mock(Dictionary.class);
+ // Dictionary has 4 entries: 10 (match), 11 (miss), 20 (match), 99 (miss).
+ // 11 and 99 chosen to make false-positive probability vanishingly small for EXPECTED_INSERTIONS=1000.
+ Mockito.when(dictionary.length()).thenReturn(4);
+ Mockito.when(dictionary.getIntValue(0)).thenReturn(10);
+ Mockito.when(dictionary.getIntValue(1)).thenReturn(11);
+ Mockito.when(dictionary.getIntValue(2)).thenReturn(20);
+ Mockito.when(dictionary.getIntValue(3)).thenReturn(99);
+
+ BaseDictionaryBasedPredicateEvaluator evaluator =
+ BloomRuntimeFilterEvaluatorFactory.newDictionaryBasedEvaluator(predicate, dictionary, DataType.INT);
+
+ Assert.assertTrue(evaluator.applySV(0), "dictId 0 (value 10) is in bloom");
+ Assert.assertFalse(evaluator.applySV(1), "dictId 1 (value 11) is not in bloom");
+ Assert.assertTrue(evaluator.applySV(2), "dictId 2 (value 20) is in bloom");
+ Assert.assertFalse(evaluator.applySV(3), "dictId 3 (value 99) is not in bloom");
+ Assert.assertEquals(evaluator.getMatchingDictIds(), new int[]{0, 2});
+ }
+
+ @Test
+ public void dictEvaluatorEmptyDictionaryIsAlwaysFalse() {
+ BloomRuntimeFilter bloom = new BloomRuntimeFilter(DataType.LONG, EXPECTED_INSERTIONS, FPP);
+ bloom.add(7L);
+ RuntimeFilterPredicate predicate = new RuntimeFilterPredicate(COLUMN, bloom);
+
+ Dictionary dictionary = Mockito.mock(Dictionary.class);
+ Mockito.when(dictionary.length()).thenReturn(0);
+
+ BaseDictionaryBasedPredicateEvaluator evaluator =
+ BloomRuntimeFilterEvaluatorFactory.newDictionaryBasedEvaluator(predicate, dictionary, DataType.LONG);
+
+ Assert.assertTrue(evaluator.isAlwaysFalse(), "empty dictionary yields no matches and must be always-false");
+ Assert.assertFalse(evaluator.isAlwaysTrue());
+ Assert.assertEquals(evaluator.getMatchingDictIds().length, 0);
+ }
+
+ @Test
+ public void dictEvaluatorAllValuesMatchedIsAlwaysTrue() {
+ BloomRuntimeFilter bloom = new BloomRuntimeFilter(DataType.STRING, EXPECTED_INSERTIONS, FPP);
+ bloom.add("a");
+ bloom.add("b");
+ bloom.add("c");
+ RuntimeFilterPredicate predicate = new RuntimeFilterPredicate(COLUMN, bloom);
+
+ Dictionary dictionary = Mockito.mock(Dictionary.class);
+ Mockito.when(dictionary.length()).thenReturn(3);
+ Mockito.when(dictionary.getStringValue(0)).thenReturn("a");
+ Mockito.when(dictionary.getStringValue(1)).thenReturn("b");
+ Mockito.when(dictionary.getStringValue(2)).thenReturn("c");
+
+ BaseDictionaryBasedPredicateEvaluator evaluator =
+ BloomRuntimeFilterEvaluatorFactory.newDictionaryBasedEvaluator(predicate, dictionary, DataType.STRING);
+
+ Assert.assertTrue(evaluator.isAlwaysTrue(), "every dictId matched -- must be always-true");
+ Assert.assertFalse(evaluator.isAlwaysFalse());
+ Assert.assertEquals(evaluator.getNumMatchingItems(), 3);
+ }
+
+ @Test
+ public void dictEvaluatorRejectsDataTypeMismatch() {
+ BloomRuntimeFilter bloom = new BloomRuntimeFilter(DataType.INT, EXPECTED_INSERTIONS, FPP);
+ RuntimeFilterPredicate predicate = new RuntimeFilterPredicate(COLUMN, bloom);
+ Dictionary dictionary = Mockito.mock(Dictionary.class);
+ Assert.assertThrows(IllegalArgumentException.class,
+ () -> BloomRuntimeFilterEvaluatorFactory.newDictionaryBasedEvaluator(predicate, dictionary, DataType.LONG));
+ }
+
+ @Test
+ public void dictEvaluatorMatchingDictIdsAreSortedAscending() {
+ BloomRuntimeFilter bloom = new BloomRuntimeFilter(DataType.INT, EXPECTED_INSERTIONS, FPP);
+ bloom.add(5);
+ bloom.add(15);
+ bloom.add(25);
+ RuntimeFilterPredicate predicate = new RuntimeFilterPredicate(COLUMN, bloom);
+
+ Dictionary dictionary = Mockito.mock(Dictionary.class);
+ // Insert matches at dictIds 4, 1, 3 -- not in increasing order -- to verify the returned array is sorted.
+ Mockito.when(dictionary.length()).thenReturn(5);
+ Mockito.when(dictionary.getIntValue(0)).thenReturn(-1);
+ Mockito.when(dictionary.getIntValue(1)).thenReturn(15);
+ Mockito.when(dictionary.getIntValue(2)).thenReturn(-1);
+ Mockito.when(dictionary.getIntValue(3)).thenReturn(25);
+ Mockito.when(dictionary.getIntValue(4)).thenReturn(5);
+
+ BaseDictionaryBasedPredicateEvaluator evaluator =
+ BloomRuntimeFilterEvaluatorFactory.newDictionaryBasedEvaluator(predicate, dictionary, DataType.INT);
+
+ int[] matching = evaluator.getMatchingDictIds();
+ Assert.assertEquals(matching, new int[]{1, 3, 4}, "matching dict ids must be returned sorted ascending");
+ }
+
+ // ---------- helpers ----------
+
+ private static PredicateEvaluator newEvaluator(BloomRuntimeFilter bloom, DataType dataType) {
+ RuntimeFilterPredicate predicate = new RuntimeFilterPredicate(COLUMN, bloom);
+ return BloomRuntimeFilterEvaluatorFactory.newRawValueBasedEvaluator(predicate, dataType);
+ }
+}
diff --git a/pinot-core/src/test/java/org/apache/pinot/core/operator/filter/predicate/RuntimeFilterPredicateEvaluatorFactoryTest.java b/pinot-core/src/test/java/org/apache/pinot/core/operator/filter/predicate/RuntimeFilterPredicateEvaluatorFactoryTest.java
new file mode 100644
index 0000000000..387d972973
--- /dev/null
+++ b/pinot-core/src/test/java/org/apache/pinot/core/operator/filter/predicate/RuntimeFilterPredicateEvaluatorFactoryTest.java
@@ -0,0 +1,78 @@
+/**
+ * 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.core.operator.filter.predicate;
+
+import org.apache.pinot.common.request.context.ExpressionContext;
+import org.apache.pinot.common.request.context.predicate.BloomRuntimeFilter;
+import org.apache.pinot.common.request.context.predicate.Predicate;
+import org.apache.pinot.common.request.context.predicate.RuntimeFilter;
+import org.apache.pinot.common.request.context.predicate.RuntimeFilterPredicate;
+import org.apache.pinot.segment.spi.index.reader.Dictionary;
+import org.apache.pinot.spi.data.FieldSpec.DataType;
+import org.mockito.Mockito;
+import org.testng.Assert;
+import org.testng.annotations.Test;
+
+
+/**
+ * Verifies the top-level {@link RuntimeFilterPredicateEvaluatorFactory} dispatches on
+ * {@link RuntimeFilter.Kind} and produces a working evaluator for both raw-value and dictionary
+ * code paths. Per-form behavior is covered in the form-specific factory tests
+ * (e.g. {@link BloomRuntimeFilterEvaluatorFactoryTest}).
+ */
+public class RuntimeFilterPredicateEvaluatorFactoryTest {
+
+ private static final ExpressionContext COLUMN = ExpressionContext.forIdentifier("column");
+
+ @Test
+ public void bloomKindRoutesToRawValueBloomEvaluator() {
+ BloomRuntimeFilter bloom = new BloomRuntimeFilter(DataType.INT, 128, 0.01);
+ bloom.add(42);
+ RuntimeFilterPredicate predicate = new RuntimeFilterPredicate(COLUMN, bloom);
+
+ PredicateEvaluator evaluator =
+ RuntimeFilterPredicateEvaluatorFactory.newRawValueBasedEvaluator(predicate, DataType.INT);
+
+ Assert.assertSame(evaluator.getPredicate(), predicate);
+ Assert.assertEquals(evaluator.getPredicateType(), Predicate.Type.RUNTIME_FILTER);
+ Assert.assertEquals(evaluator.getDataType(), DataType.INT);
+ Assert.assertFalse(evaluator.isDictionaryBased());
+ Assert.assertTrue(evaluator.applySV(42), "bloom-routed evaluator must hit added key");
+ }
+
+ @Test
+ public void bloomKindRoutesToDictionaryBloomEvaluator() {
+ BloomRuntimeFilter bloom = new BloomRuntimeFilter(DataType.INT, 128, 0.01);
+ bloom.add(7);
+ RuntimeFilterPredicate predicate = new RuntimeFilterPredicate(COLUMN, bloom);
+ Dictionary dictionary = Mockito.mock(Dictionary.class);
+ Mockito.when(dictionary.length()).thenReturn(2);
+ Mockito.when(dictionary.getIntValue(0)).thenReturn(7);
+ Mockito.when(dictionary.getIntValue(1)).thenReturn(-99);
+
+ PredicateEvaluator evaluator =
+ RuntimeFilterPredicateEvaluatorFactory.newDictionaryBasedEvaluator(predicate, dictionary, DataType.INT);
+
+ Assert.assertSame(evaluator.getPredicate(), predicate);
+ Assert.assertEquals(evaluator.getPredicateType(), Predicate.Type.RUNTIME_FILTER);
+ Assert.assertTrue(evaluator.isDictionaryBased());
+ Assert.assertTrue(evaluator.applySV(0), "dictId 0 (value 7) matches the bloom");
+ Assert.assertFalse(evaluator.applySV(1), "dictId 1 (value -99) does not match the bloom");
+ }
+}