From 9b02b0a76e5e673095e9a620fe70f8439a41faa2 Mon Sep 17 00:00:00 2001 From: Tianle Zhang Date: Wed, 20 May 2026 16:13:37 -0700 Subject: [PATCH 1/7] OA-983: Add BLOOM_MEMBERSHIP predicate primitive for MSE runtime filter Lay the leaf-stage primitive used by the upcoming MSE hash-join runtime filter (dynamic filter pushdown). Independent of the broker configuration scaffolding in #195 -- this PR only adds the predicate plumbing and an in-process bloom filter wrapper; no rule registration, no operator wiring, and the predicate has no producer yet. Added: * Predicate.Type.BLOOM_MEMBERSHIP -- new enum value, inclusive semantics. * BloomMembershipPredicate (pinot-common) -- carries a RuntimeBloomFilter and an ExpressionContext. Reference equality on the filter; two distinct instances are conceptually different runtime filters even if their bit arrays match. * RuntimeBloomFilter (pinot-common) -- in-memory bloom wrapper mirroring the funnel choices in BloomFilterIdSet (INT/FLOAT share an int funnel, LONG/DOUBLE share a long funnel, plus STRING and BYTES funnels). Lives in pinot-common so predicate classes can reference it; serialization for cross-worker transport is intentionally deferred. * BloomMembershipPredicateEvaluatorFactory (pinot-core) -- per-DataType inner evaluators (INT/LONG/FLOAT/DOUBLE/STRING/BYTES), each delegating applySV to the matching RuntimeBloomFilter.mightContain overload. The factory fails fast on dataType mismatch so misconfigured wiring shows up at construction rather than producing garbage at query time. * PredicateEvaluatorProvider dispatch -- raw-value branch routes BLOOM_MEMBERSHIP to the new factory. Dictionary-based branch throws UnsupportedOperationException with a clear message; a dict-id pre-materialization variant is deferred until we see workloads needing it. Tests (TestNG): * RuntimeBloomFilterTest (12 cases) -- constructor rejects unsupported data types; no false negatives for INT/LONG/FLOAT/DOUBLE/STRING/BYTES; observed false-positive rate stays within 5x of configured fpp; byte[] content equality hashes identically; raw-bits float encoding documented. * BloomMembershipPredicateEvaluatorFactoryTest (13 cases) -- factory rejects type mismatch and unsupported types; evaluator metadata (predicate-type, data-type, dictionary-based, exclusive) is correct; no false negatives across all six SV types; INT false-positive rate within tolerance; base-class default applySV overloads throw for the wrong type so callers cannot accidentally call applySV(long) on an INT-typed bloom; MV inherits the correct semantics via the base loop; PredicateEvaluatorProvider routes raw-value BLOOM_MEMBERSHIP to the factory; provider rejects the dictionary-based path. Co-Authored-By: Claude Opus 4.7 (1M context) --- .../predicate/BloomMembershipPredicate.java | 79 ++++++ .../request/context/predicate/Predicate.java | 8 +- .../context/predicate/RuntimeBloomFilter.java | 143 +++++++++++ .../predicate/RuntimeBloomFilterTest.java | 172 +++++++++++++ ...omMembershipPredicateEvaluatorFactory.java | 179 ++++++++++++++ .../predicate/PredicateEvaluatorProvider.java | 11 + ...mbershipPredicateEvaluatorFactoryTest.java | 233 ++++++++++++++++++ 7 files changed, 824 insertions(+), 1 deletion(-) create mode 100644 pinot-common/src/main/java/org/apache/pinot/common/request/context/predicate/BloomMembershipPredicate.java create mode 100644 pinot-common/src/main/java/org/apache/pinot/common/request/context/predicate/RuntimeBloomFilter.java create mode 100644 pinot-common/src/test/java/org/apache/pinot/common/request/context/predicate/RuntimeBloomFilterTest.java create mode 100644 pinot-core/src/main/java/org/apache/pinot/core/operator/filter/predicate/BloomMembershipPredicateEvaluatorFactory.java create mode 100644 pinot-core/src/test/java/org/apache/pinot/core/operator/filter/predicate/BloomMembershipPredicateEvaluatorFactoryTest.java diff --git a/pinot-common/src/main/java/org/apache/pinot/common/request/context/predicate/BloomMembershipPredicate.java b/pinot-common/src/main/java/org/apache/pinot/common/request/context/predicate/BloomMembershipPredicate.java new file mode 100644 index 0000000000..b4b40e163d --- /dev/null +++ b/pinot-common/src/main/java/org/apache/pinot/common/request/context/predicate/BloomMembershipPredicate.java @@ -0,0 +1,79 @@ +/** + * 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 RuntimeBloomFilter} reports as possibly present. + * + *

Injected by the MSE runtime filter feature (OA-983) at the leaf-stage filter boundary. + * Inclusive predicate: a row matches when {@code bloomFilter.mightContain(value)} returns true. + * False positives are permitted; downstream the actual hash join still performs exact matching.

+ * + *

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 phase). Cross-worker transport is + * out of scope for this PR.

+ */ +public class BloomMembershipPredicate extends BasePredicate { + + private final RuntimeBloomFilter _bloomFilter; + + public BloomMembershipPredicate(ExpressionContext lhs, RuntimeBloomFilter bloomFilter) { + super(lhs); + _bloomFilter = Objects.requireNonNull(bloomFilter, "bloomFilter"); + } + + @Override + public Type getType() { + return Type.BLOOM_MEMBERSHIP; + } + + public RuntimeBloomFilter getBloomFilter() { + return _bloomFilter; + } + + @Override + public boolean equals(Object o) { + if (this == o) { + return true; + } + if (!(o instanceof BloomMembershipPredicate)) { + return false; + } + BloomMembershipPredicate that = (BloomMembershipPredicate) o; + // Reference equality on the bloom filter is intentional: two distinct RuntimeBloomFilter + // instances are conceptually different runtime filters even if they happen to contain the + // same keys. There is no cheap structural equality on Guava BloomFilter that we want to lean on. + return Objects.equals(_lhs, that._lhs) && _bloomFilter == that._bloomFilter; + } + + @Override + public int hashCode() { + return Objects.hash(_lhs, System.identityHashCode(_bloomFilter)); + } + + @Override + public String toString() { + return _lhs + " BLOOM_MEMBERSHIP(" + _bloomFilter.getDataType() + ")"; + } +} 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..4da7848c60 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,13 @@ enum Type { JSON_MATCH, IS_NULL, IS_NOT_NULL(true), - VECTOR_SIMILARITY; + VECTOR_SIMILARITY, + // Runtime Bloom-filter membership test, injected by the MSE runtime filter feature (OA-983). + // Carries a RuntimeBloomFilter built from the join build side; rows whose value is "not in" + // the filter can be skipped at the leaf-stage scan. Inclusive: a row matches when the filter + // mightContain(value). False positives are allowed (downstream join still does exact match); + // false negatives must not occur. + BLOOM_MEMBERSHIP; private final boolean _exclusive; diff --git a/pinot-common/src/main/java/org/apache/pinot/common/request/context/predicate/RuntimeBloomFilter.java b/pinot-common/src/main/java/org/apache/pinot/common/request/context/predicate/RuntimeBloomFilter.java new file mode 100644 index 0000000000..d423db5dc0 --- /dev/null +++ b/pinot-common/src/main/java/org/apache/pinot/common/request/context/predicate/RuntimeBloomFilter.java @@ -0,0 +1,143 @@ +/** + * 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 com.google.common.hash.BloomFilter; +import com.google.common.hash.Funnels; +import org.apache.pinot.spi.data.FieldSpec.DataType; + + +/** + * In-memory Bloom filter built at query time and consumed by {@link BloomMembershipPredicate}. + * + *

This is the in-process companion to the runtime filter feature (OA-983). The MSE runtime + * builds one of these from the hash-join build side, hands it to the probe side, and the leaf + * stage uses it to prune non-matching rows during the segment scan.

+ * + *

The implementation mirrors {@code BloomFilterIdSet} (pinot-core) - same funnel choices, same + * raw-bits encoding for FLOAT/DOUBLE - but lives in pinot-common so it can be referenced by + * predicate classes. Serialization for cross-worker transport is intentionally out of scope here; + * follow-up PRs will add a wire format alongside the runtime side-channel.

+ * + *

The caller is responsible for invoking the {@code mightContain} overload that matches the + * column's data type. Calling the wrong overload does not throw but produces undefined membership + * results.

+ */ +@SuppressWarnings("UnstableApiUsage") +public final class RuntimeBloomFilter { + + /** + * Funnel-family selected at construction time based on the source {@link DataType}. FLOAT/DOUBLE + * reuse the INT/LONG funnels via raw-bits encoding so equal float values hash identically. + */ + private enum FunnelType { + INT, LONG, STRING, BYTES + } + + private final DataType _dataType; + private final FunnelType _funnelType; + private final BloomFilter _bloomFilter; + + @SuppressWarnings({"unchecked", "rawtypes"}) + public RuntimeBloomFilter(DataType dataType, int expectedInsertions, double fpp) { + _dataType = dataType; + BloomFilter bf; + switch (dataType) { + case INT: + case FLOAT: + _funnelType = FunnelType.INT; + bf = BloomFilter.create(Funnels.integerFunnel(), expectedInsertions, fpp); + break; + case LONG: + case DOUBLE: + _funnelType = FunnelType.LONG; + bf = BloomFilter.create(Funnels.longFunnel(), expectedInsertions, fpp); + break; + case STRING: + _funnelType = FunnelType.STRING; + bf = BloomFilter.create(Funnels.unencodedCharsFunnel(), expectedInsertions, fpp); + break; + case BYTES: + _funnelType = FunnelType.BYTES; + bf = BloomFilter.create(Funnels.byteArrayFunnel(), expectedInsertions, fpp); + break; + default: + throw new IllegalArgumentException("RuntimeBloomFilter does not support data type: " + dataType); + } + _bloomFilter = (BloomFilter) bf; + } + + 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 as raw int bits so the INT funnel hashes equal floats identically. + _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); + } + + /** Returns the funnel family used internally; visible for tests. */ + FunnelType funnelType() { + return _funnelType; + } +} diff --git a/pinot-common/src/test/java/org/apache/pinot/common/request/context/predicate/RuntimeBloomFilterTest.java b/pinot-common/src/test/java/org/apache/pinot/common/request/context/predicate/RuntimeBloomFilterTest.java new file mode 100644 index 0000000000..3d8b0f8730 --- /dev/null +++ b/pinot-common/src/test/java/org/apache/pinot/common/request/context/predicate/RuntimeBloomFilterTest.java @@ -0,0 +1,172 @@ +/** + * 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 RuntimeBloomFilterTest { + + private static final double FPP = 0.01; + private static final int EXPECTED_INSERTIONS = 1000; + + @Test + public void constructorRejectsUnsupportedDataType() { + Assert.assertThrows(IllegalArgumentException.class, + () -> new RuntimeBloomFilter(DataType.BIG_DECIMAL, EXPECTED_INSERTIONS, FPP)); + Assert.assertThrows(IllegalArgumentException.class, + () -> new RuntimeBloomFilter(DataType.TIMESTAMP, EXPECTED_INSERTIONS, FPP)); + Assert.assertThrows(IllegalArgumentException.class, + () -> new RuntimeBloomFilter(DataType.BOOLEAN, EXPECTED_INSERTIONS, FPP)); + } + + @Test + public void getDataTypeReportsConstructorArgument() { + RuntimeBloomFilter bf = new RuntimeBloomFilter(DataType.LONG, 16, FPP); + Assert.assertEquals(bf.getDataType(), DataType.LONG); + } + + // ---------- INT ---------- + + @Test + public void intAddedValuesAlwaysMightContain() { + RuntimeBloomFilter bf = new RuntimeBloomFilter(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() { + RuntimeBloomFilter bf = new RuntimeBloomFilter(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() { + RuntimeBloomFilter bf = new RuntimeBloomFilter(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() { + RuntimeBloomFilter bf = new RuntimeBloomFilter(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 floatDifferentBitPatternsAreDistinct() { + // 0.0f and -0.0f have the same value semantically but different raw bits. With raw-bits hashing + // we expect them to be treated as distinct. Lock in that behavior. + RuntimeBloomFilter bf = new RuntimeBloomFilter(DataType.FLOAT, EXPECTED_INSERTIONS, FPP); + bf.add(0.0f); + Assert.assertTrue(bf.mightContain(0.0f)); + // -0.0f may or may not collide with 0.0f under Bloom (false-positive territory). We do NOT + // assert false here -- we just verify the API doesn't throw. + boolean negZeroContained = bf.mightContain(-0.0f); + Assert.assertTrue(negZeroContained || !negZeroContained); // tautology; doc the contract + } + + // ---------- DOUBLE ---------- + + @Test + public void doubleAddedValuesAlwaysMightContain() { + RuntimeBloomFilter bf = new RuntimeBloomFilter(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() { + RuntimeBloomFilter bf = new RuntimeBloomFilter(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() { + RuntimeBloomFilter bf = new RuntimeBloomFilter(DataType.STRING, EXPECTED_INSERTIONS, FPP); + bf.add(""); + Assert.assertTrue(bf.mightContain("")); + } + + // ---------- BYTES ---------- + + @Test + public void bytesAddedValuesAlwaysMightContain() { + RuntimeBloomFilter bf = new RuntimeBloomFilter(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). + RuntimeBloomFilter bf = new RuntimeBloomFilter(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/BloomMembershipPredicateEvaluatorFactory.java b/pinot-core/src/main/java/org/apache/pinot/core/operator/filter/predicate/BloomMembershipPredicateEvaluatorFactory.java new file mode 100644 index 0000000000..8d0f9b563c --- /dev/null +++ b/pinot-core/src/main/java/org/apache/pinot/core/operator/filter/predicate/BloomMembershipPredicateEvaluatorFactory.java @@ -0,0 +1,179 @@ +/** + * 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.BloomMembershipPredicate; +import org.apache.pinot.common.request.context.predicate.RuntimeBloomFilter; +import org.apache.pinot.spi.data.FieldSpec.DataType; + + +/** + * Factory for {@link PredicateEvaluator}s that test single-value column entries against a + * {@link RuntimeBloomFilter}. Raw-value evaluation only - dictionary-based evaluation for runtime + * Bloom filters is intentionally deferred (see {@code newDictionaryBasedEvaluator}). + * + *

Per OA-983, false positives are allowed (downstream join still performs exact matching) but + * false negatives must not occur. Each per-type evaluator delegates directly to the matching + * {@code RuntimeBloomFilter.mightContain(...)} overload so the filter's data-type-specific + * funnel selects the correct hashing.

+ */ +public class BloomMembershipPredicateEvaluatorFactory { + private BloomMembershipPredicateEvaluatorFactory() { + } + + /** + * Builds a raw-value evaluator for the given {@code dataType}. The evaluator dispatches to the + * {@code RuntimeBloomFilter} via the type-appropriate {@code mightContain} call. Supported + * types: INT, LONG, FLOAT, DOUBLE, STRING, BYTES (matches {@link RuntimeBloomFilter}). Other + * data types throw {@link IllegalArgumentException} at construction; callers must not request + * BLOOM_MEMBERSHIP for unsupported column types. + */ + public static PredicateEvaluator newRawValueBasedEvaluator(BloomMembershipPredicate predicate, DataType dataType) { + RuntimeBloomFilter bloomFilter = predicate.getBloomFilter(); + if (bloomFilter.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( + "RuntimeBloomFilter dataType " + bloomFilter.getDataType() + + " does not match evaluator dataType " + dataType); + } + switch (dataType) { + case INT: + return new IntBloomMembershipPredicateEvaluator(predicate); + case LONG: + return new LongBloomMembershipPredicateEvaluator(predicate); + case FLOAT: + return new FloatBloomMembershipPredicateEvaluator(predicate); + case DOUBLE: + return new DoubleBloomMembershipPredicateEvaluator(predicate); + case STRING: + return new StringBloomMembershipPredicateEvaluator(predicate); + case BYTES: + return new BytesBloomMembershipPredicateEvaluator(predicate); + default: + throw new IllegalArgumentException("BLOOM_MEMBERSHIP does not support data type: " + dataType); + } + } + + private abstract static class BaseBloomMembershipPredicateEvaluator extends BaseRawValueBasedPredicateEvaluator { + final RuntimeBloomFilter _bloomFilter; + + BaseBloomMembershipPredicateEvaluator(BloomMembershipPredicate predicate) { + super(predicate); + _bloomFilter = predicate.getBloomFilter(); + } + } + + private static final class IntBloomMembershipPredicateEvaluator extends BaseBloomMembershipPredicateEvaluator { + IntBloomMembershipPredicateEvaluator(BloomMembershipPredicate predicate) { + super(predicate); + } + + @Override + public DataType getDataType() { + return DataType.INT; + } + + @Override + public boolean applySV(int value) { + return _bloomFilter.mightContain(value); + } + } + + private static final class LongBloomMembershipPredicateEvaluator extends BaseBloomMembershipPredicateEvaluator { + LongBloomMembershipPredicateEvaluator(BloomMembershipPredicate predicate) { + super(predicate); + } + + @Override + public DataType getDataType() { + return DataType.LONG; + } + + @Override + public boolean applySV(long value) { + return _bloomFilter.mightContain(value); + } + } + + private static final class FloatBloomMembershipPredicateEvaluator extends BaseBloomMembershipPredicateEvaluator { + FloatBloomMembershipPredicateEvaluator(BloomMembershipPredicate predicate) { + super(predicate); + } + + @Override + public DataType getDataType() { + return DataType.FLOAT; + } + + @Override + public boolean applySV(float value) { + return _bloomFilter.mightContain(value); + } + } + + private static final class DoubleBloomMembershipPredicateEvaluator extends BaseBloomMembershipPredicateEvaluator { + DoubleBloomMembershipPredicateEvaluator(BloomMembershipPredicate predicate) { + super(predicate); + } + + @Override + public DataType getDataType() { + return DataType.DOUBLE; + } + + @Override + public boolean applySV(double value) { + return _bloomFilter.mightContain(value); + } + } + + private static final class StringBloomMembershipPredicateEvaluator extends BaseBloomMembershipPredicateEvaluator { + StringBloomMembershipPredicateEvaluator(BloomMembershipPredicate predicate) { + super(predicate); + } + + @Override + public DataType getDataType() { + return DataType.STRING; + } + + @Override + public boolean applySV(String value) { + return _bloomFilter.mightContain(value); + } + } + + private static final class BytesBloomMembershipPredicateEvaluator extends BaseBloomMembershipPredicateEvaluator { + BytesBloomMembershipPredicateEvaluator(BloomMembershipPredicate predicate) { + super(predicate); + } + + @Override + public DataType getDataType() { + return DataType.BYTES; + } + + @Override + public boolean applySV(byte[] value) { + return _bloomFilter.mightContain(value); + } + } +} 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..d2ca42e4fa 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 @@ -19,6 +19,7 @@ package org.apache.pinot.core.operator.filter.predicate; import javax.annotation.Nullable; +import org.apache.pinot.common.request.context.predicate.BloomMembershipPredicate; import org.apache.pinot.common.request.context.predicate.EqPredicate; import org.apache.pinot.common.request.context.predicate.InPredicate; import org.apache.pinot.common.request.context.predicate.NotEqPredicate; @@ -66,6 +67,13 @@ public static PredicateEvaluator getPredicateEvaluator(Predicate predicate, @Nul case REGEXP_LIKE: return RegexpLikePredicateEvaluatorFactory .newDictionaryBasedEvaluator((RegexpLikePredicate) predicate, dictionary, dataType); + case BLOOM_MEMBERSHIP: + // Dictionary-based BLOOM_MEMBERSHIP is intentionally not wired in this PR. Per OA-983 + // the predicate is initially injected only at the raw-value scan path; a future PR can + // add a dict-id pre-materialization variant once we have measured workloads that need + // it. Throw clearly so accidental wiring during planner work is caught early. + throw new UnsupportedOperationException( + "Dictionary-based BLOOM_MEMBERSHIP predicate is not yet supported"); default: throw new UnsupportedOperationException("Unsupported predicate type: " + predicate.getType()); } @@ -85,6 +93,9 @@ public static PredicateEvaluator getPredicateEvaluator(Predicate predicate, @Nul case REGEXP_LIKE: return RegexpLikePredicateEvaluatorFactory .newRawValueBasedEvaluator((RegexpLikePredicate) predicate, dataType); + case BLOOM_MEMBERSHIP: + return BloomMembershipPredicateEvaluatorFactory + .newRawValueBasedEvaluator((BloomMembershipPredicate) predicate, dataType); default: throw new UnsupportedOperationException("Unsupported predicate type: " + predicate.getType()); } diff --git a/pinot-core/src/test/java/org/apache/pinot/core/operator/filter/predicate/BloomMembershipPredicateEvaluatorFactoryTest.java b/pinot-core/src/test/java/org/apache/pinot/core/operator/filter/predicate/BloomMembershipPredicateEvaluatorFactoryTest.java new file mode 100644 index 0000000000..9903a252bd --- /dev/null +++ b/pinot-core/src/test/java/org/apache/pinot/core/operator/filter/predicate/BloomMembershipPredicateEvaluatorFactoryTest.java @@ -0,0 +1,233 @@ +/** + * 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.BloomMembershipPredicate; +import org.apache.pinot.common.request.context.predicate.Predicate; +import org.apache.pinot.common.request.context.predicate.RuntimeBloomFilter; +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 BloomMembershipPredicateEvaluatorFactoryTest { + + 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 RuntimeBloomFilterTest tolerance + + // ---------- Factory wiring ---------- + + @Test + public void factoryRejectsDataTypeMismatch() { + RuntimeBloomFilter bloom = new RuntimeBloomFilter(DataType.INT, EXPECTED_INSERTIONS, FPP); + BloomMembershipPredicate predicate = new BloomMembershipPredicate(COLUMN, bloom); + Assert.assertThrows(IllegalArgumentException.class, + () -> BloomMembershipPredicateEvaluatorFactory.newRawValueBasedEvaluator(predicate, DataType.LONG)); + } + + @Test + public void factoryRejectsUnsupportedDataType() { + RuntimeBloomFilter bloom = new RuntimeBloomFilter(DataType.STRING, EXPECTED_INSERTIONS, FPP); + BloomMembershipPredicate predicate = new BloomMembershipPredicate(COLUMN, bloom); + // STRING bloom but BIG_DECIMAL evaluator -- the dataType-mismatch guard fires first. + Assert.assertThrows(IllegalArgumentException.class, + () -> BloomMembershipPredicateEvaluatorFactory.newRawValueBasedEvaluator(predicate, DataType.BIG_DECIMAL)); + } + + @Test + public void evaluatorReportsExpectedMetadata() { + RuntimeBloomFilter bloom = new RuntimeBloomFilter(DataType.LONG, EXPECTED_INSERTIONS, FPP); + BloomMembershipPredicate predicate = new BloomMembershipPredicate(COLUMN, bloom); + PredicateEvaluator evaluator = + BloomMembershipPredicateEvaluatorFactory.newRawValueBasedEvaluator(predicate, DataType.LONG); + + Assert.assertSame(evaluator.getPredicate(), predicate); + Assert.assertEquals(evaluator.getPredicateType(), Predicate.Type.BLOOM_MEMBERSHIP); + Assert.assertEquals(evaluator.getDataType(), DataType.LONG); + Assert.assertFalse(evaluator.isDictionaryBased(), "raw-value evaluator must not be dictionary-based"); + Assert.assertFalse(evaluator.isExclusive(), "BLOOM_MEMBERSHIP is an inclusive predicate"); + Assert.assertFalse(evaluator.isAlwaysTrue()); + Assert.assertFalse(evaluator.isAlwaysFalse()); + } + + // ---------- Per-type membership ---------- + + @Test + public void intEvaluatorHasNoFalseNegativesAndLowFalsePositiveRate() { + RuntimeBloomFilter bloom = new RuntimeBloomFilter(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() { + RuntimeBloomFilter bloom = new RuntimeBloomFilter(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() { + RuntimeBloomFilter bloom = new RuntimeBloomFilter(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() { + RuntimeBloomFilter bloom = new RuntimeBloomFilter(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() { + RuntimeBloomFilter bloom = new RuntimeBloomFilter(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() { + RuntimeBloomFilter bloom = new RuntimeBloomFilter(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. + RuntimeBloomFilter bloom = new RuntimeBloomFilter(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() { + RuntimeBloomFilter bloom = new RuntimeBloomFilter(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 providerRoutesRawValueBasedBloomMembershipToFactory() { + RuntimeBloomFilter bloom = new RuntimeBloomFilter(DataType.LONG, EXPECTED_INSERTIONS, FPP); + BloomMembershipPredicate predicate = new BloomMembershipPredicate(COLUMN, bloom); + PredicateEvaluator evaluator = + PredicateEvaluatorProvider.getPredicateEvaluator(predicate, null, DataType.LONG); + Assert.assertEquals(evaluator.getPredicateType(), Predicate.Type.BLOOM_MEMBERSHIP); + Assert.assertEquals(evaluator.getDataType(), DataType.LONG); + } + + @Test + public void providerRejectsDictionaryBasedBloomMembership() { + RuntimeBloomFilter bloom = new RuntimeBloomFilter(DataType.INT, EXPECTED_INSERTIONS, FPP); + BloomMembershipPredicate predicate = new BloomMembershipPredicate(COLUMN, bloom); + // Passing a non-null dictionary -- which routes us into the dictionary-based switch -- must + // throw because dict-based BLOOM_MEMBERSHIP is deferred. PredicateEvaluatorProvider wraps the + // UnsupportedOperationException in a BadQueryRequestException; assert any throwable so the + // contract holds even if the wrapping changes. + Dictionary mockDictionary = Mockito.mock(Dictionary.class); + Assert.assertThrows(Exception.class, + () -> PredicateEvaluatorProvider.getPredicateEvaluator(predicate, mockDictionary, DataType.INT)); + } + + // ---------- helpers ---------- + + private static PredicateEvaluator newEvaluator(RuntimeBloomFilter bloom, DataType dataType) { + BloomMembershipPredicate predicate = new BloomMembershipPredicate(COLUMN, bloom); + return BloomMembershipPredicateEvaluatorFactory.newRawValueBasedEvaluator(predicate, dataType); + } +} From 70fc63cf1f37286b510c786590cbb64ab753754e Mon Sep 17 00:00:00 2001 From: Tianle Zhang Date: Wed, 20 May 2026 16:30:38 -0700 Subject: [PATCH 2/7] Address review on BLOOM_MEMBERSHIP predicate * RuntimeBloomFilter class Javadoc: drop the inaccurate "calling the wrong overload does not throw" claim. The underlying Guava BloomFilter is constructed with a single Funnel matched to the DataType, so a wrong-typed call may throw ClassCastException when the boxed value cannot be passed to that funnel, or return a meaningless answer. Callers must not rely on either outcome. * FunnelType Javadoc: replace "equal float values hash identically" with the accurate raw-bits contract. Values with the same IEEE-754 bit pattern hash to the same slot; -0.0f vs 0.0f and distinct NaN payloads hash differently. * add(float) inline comment: same rewording -- raw-bits encoding, not semantic equality. * RuntimeBloomFilterTest.floatDifferentBitPatternsAreDistinct: replace the tautological assertTrue(x || !x) with a meaningful pin-down of the "same bits -> same hash" guarantee (including a non-canonical NaN). The negative-zero call is now a smoke-only API exercise. * Drop stale internal references from Javadoc/comments across the predicate, factory, and provider files; keep only the technical description that survives code review on its own. Co-Authored-By: Claude Opus 4.7 (1M context) --- .../predicate/BloomMembershipPredicate.java | 10 ++++----- .../request/context/predicate/Predicate.java | 6 ++--- .../context/predicate/RuntimeBloomFilter.java | 22 ++++++++++++------- .../predicate/RuntimeBloomFilterTest.java | 21 ++++++++++-------- ...omMembershipPredicateEvaluatorFactory.java | 4 ++-- .../predicate/PredicateEvaluatorProvider.java | 8 +++---- 6 files changed, 40 insertions(+), 31 deletions(-) diff --git a/pinot-common/src/main/java/org/apache/pinot/common/request/context/predicate/BloomMembershipPredicate.java b/pinot-common/src/main/java/org/apache/pinot/common/request/context/predicate/BloomMembershipPredicate.java index b4b40e163d..cc153e8def 100644 --- a/pinot-common/src/main/java/org/apache/pinot/common/request/context/predicate/BloomMembershipPredicate.java +++ b/pinot-common/src/main/java/org/apache/pinot/common/request/context/predicate/BloomMembershipPredicate.java @@ -25,14 +25,14 @@ /** * Predicate matching values that the carried {@link RuntimeBloomFilter} reports as possibly present. * - *

Injected by the MSE runtime filter feature (OA-983) at the leaf-stage filter boundary. - * Inclusive predicate: a row matches when {@code bloomFilter.mightContain(value)} returns true. - * False positives are permitted; downstream the actual hash join still performs exact matching.

+ *

Injected by the MSE runtime filter feature at the leaf-stage filter boundary. Inclusive + * predicate: a row matches when {@code bloomFilter.mightContain(value)} returns true. False + * positives are permitted; downstream the actual hash join still performs exact matching.

* *

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 phase). Cross-worker transport is - * out of scope for this PR.

+ * filter tree via {@code QueryContext} (wired up in a later change). Cross-worker transport is + * out of scope here.

*/ public class BloomMembershipPredicate extends BasePredicate { 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 4da7848c60..06a179e843 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 @@ -40,9 +40,9 @@ enum Type { IS_NULL, IS_NOT_NULL(true), VECTOR_SIMILARITY, - // Runtime Bloom-filter membership test, injected by the MSE runtime filter feature (OA-983). - // Carries a RuntimeBloomFilter built from the join build side; rows whose value is "not in" - // the filter can be skipped at the leaf-stage scan. Inclusive: a row matches when the filter + // Runtime Bloom-filter membership test, injected by the MSE runtime filter feature. Carries a + // RuntimeBloomFilter built from the join build side; rows whose value is "not in" the filter + // can be skipped at the leaf-stage scan. Inclusive: a row matches when the filter // mightContain(value). False positives are allowed (downstream join still does exact match); // false negatives must not occur. BLOOM_MEMBERSHIP; diff --git a/pinot-common/src/main/java/org/apache/pinot/common/request/context/predicate/RuntimeBloomFilter.java b/pinot-common/src/main/java/org/apache/pinot/common/request/context/predicate/RuntimeBloomFilter.java index d423db5dc0..2ebdaa0b9c 100644 --- a/pinot-common/src/main/java/org/apache/pinot/common/request/context/predicate/RuntimeBloomFilter.java +++ b/pinot-common/src/main/java/org/apache/pinot/common/request/context/predicate/RuntimeBloomFilter.java @@ -26,25 +26,29 @@ /** * In-memory Bloom filter built at query time and consumed by {@link BloomMembershipPredicate}. * - *

This is the in-process companion to the runtime filter feature (OA-983). The MSE runtime - * builds one of these from the hash-join build side, hands it to the probe side, and the leaf - * stage uses it to prune non-matching rows during the segment scan.

+ *

The MSE runtime builds one of these from the hash-join build side, hands it to the probe + * side, and the leaf stage uses it to prune non-matching rows during the segment scan.

* *

The implementation mirrors {@code BloomFilterIdSet} (pinot-core) - same funnel choices, same * raw-bits encoding for FLOAT/DOUBLE - but lives in pinot-common so it can be referenced by * predicate classes. Serialization for cross-worker transport is intentionally out of scope here; - * follow-up PRs will add a wire format alongside the runtime side-channel.

+ * follow-up changes will add a wire format alongside the runtime side-channel.

* *

The caller is responsible for invoking the {@code mightContain} overload that matches the - * column's data type. Calling the wrong overload does not throw but produces undefined membership - * results.

+ * column's data type. Calling the wrong overload is undefined: the underlying Guava + * {@link BloomFilter} is constructed with a single {@link com.google.common.hash.Funnel} matched + * to the {@link DataType}, so a wrong-typed call may throw {@link ClassCastException} (when the + * boxed value cannot be passed to that funnel) or return a meaningless membership answer. Callers + * must not rely on either outcome.

*/ @SuppressWarnings("UnstableApiUsage") public final class RuntimeBloomFilter { /** * Funnel-family selected at construction time based on the source {@link DataType}. FLOAT/DOUBLE - * reuse the INT/LONG funnels via raw-bits encoding so equal float values hash identically. + * reuse the INT/LONG funnels via raw-bits encoding: two float values with the same raw IEEE-754 + * bit pattern hash identically; values that differ only by bit pattern (e.g. {@code 0.0f} vs + * {@code -0.0f}, or distinct NaN payloads) hash to different slots. */ private enum FunnelType { INT, LONG, STRING, BYTES @@ -96,7 +100,9 @@ public void add(long value) { } public void add(float value) { - // Match BloomFilterIdSet: encode as raw int bits so the INT funnel hashes equal floats identically. + // 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)); } diff --git a/pinot-common/src/test/java/org/apache/pinot/common/request/context/predicate/RuntimeBloomFilterTest.java b/pinot-common/src/test/java/org/apache/pinot/common/request/context/predicate/RuntimeBloomFilterTest.java index 3d8b0f8730..0f6c8357f9 100644 --- a/pinot-common/src/test/java/org/apache/pinot/common/request/context/predicate/RuntimeBloomFilterTest.java +++ b/pinot-common/src/test/java/org/apache/pinot/common/request/context/predicate/RuntimeBloomFilterTest.java @@ -103,16 +103,19 @@ public void floatAddedValuesAlwaysMightContain() { } @Test - public void floatDifferentBitPatternsAreDistinct() { - // 0.0f and -0.0f have the same value semantically but different raw bits. With raw-bits hashing - // we expect them to be treated as distinct. Lock in that behavior. + 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. RuntimeBloomFilter bf = new RuntimeBloomFilter(DataType.FLOAT, EXPECTED_INSERTIONS, FPP); - bf.add(0.0f); - Assert.assertTrue(bf.mightContain(0.0f)); - // -0.0f may or may not collide with 0.0f under Bloom (false-positive territory). We do NOT - // assert false here -- we just verify the API doesn't throw. - boolean negZeroContained = bf.mightContain(-0.0f); - Assert.assertTrue(negZeroContained || !negZeroContained); // tautology; doc the contract + 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 ---------- diff --git a/pinot-core/src/main/java/org/apache/pinot/core/operator/filter/predicate/BloomMembershipPredicateEvaluatorFactory.java b/pinot-core/src/main/java/org/apache/pinot/core/operator/filter/predicate/BloomMembershipPredicateEvaluatorFactory.java index 8d0f9b563c..e28bfa3857 100644 --- a/pinot-core/src/main/java/org/apache/pinot/core/operator/filter/predicate/BloomMembershipPredicateEvaluatorFactory.java +++ b/pinot-core/src/main/java/org/apache/pinot/core/operator/filter/predicate/BloomMembershipPredicateEvaluatorFactory.java @@ -28,8 +28,8 @@ * {@link RuntimeBloomFilter}. Raw-value evaluation only - dictionary-based evaluation for runtime * Bloom filters is intentionally deferred (see {@code newDictionaryBasedEvaluator}). * - *

Per OA-983, false positives are allowed (downstream join still performs exact matching) but - * false negatives must not occur. Each per-type evaluator delegates directly to the matching + *

False positives are allowed (downstream join still performs exact matching) but false + * negatives must not occur. Each per-type evaluator delegates directly to the matching * {@code RuntimeBloomFilter.mightContain(...)} overload so the filter's data-type-specific * funnel selects the correct hashing.

*/ 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 d2ca42e4fa..6499ce9086 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 @@ -68,10 +68,10 @@ public static PredicateEvaluator getPredicateEvaluator(Predicate predicate, @Nul return RegexpLikePredicateEvaluatorFactory .newDictionaryBasedEvaluator((RegexpLikePredicate) predicate, dictionary, dataType); case BLOOM_MEMBERSHIP: - // Dictionary-based BLOOM_MEMBERSHIP is intentionally not wired in this PR. Per OA-983 - // the predicate is initially injected only at the raw-value scan path; a future PR can - // add a dict-id pre-materialization variant once we have measured workloads that need - // it. Throw clearly so accidental wiring during planner work is caught early. + // Dictionary-based BLOOM_MEMBERSHIP is intentionally not wired up yet. The predicate is + // currently injected only at the raw-value scan path; a dict-id pre-materialization + // variant can be added once we have measured workloads that need it. Throw clearly so + // accidental wiring during planner work is caught early. throw new UnsupportedOperationException( "Dictionary-based BLOOM_MEMBERSHIP predicate is not yet supported"); default: From 39826e9c402e66f6e56ace489dd9d02aaf26e872 Mon Sep 17 00:00:00 2001 From: Tianle Zhang Date: Thu, 28 May 2026 17:12:53 -0700 Subject: [PATCH 3/7] Generalize predicate type: BLOOM_MEMBERSHIP -> RUNTIME_FILTER Refactor the v1 predicate primitive so the predicate type, the predicate class, and the leaf-stage dispatch path are agnostic to the concrete filter form (Bloom today; min-max range and IN-list later). Adding a future filter form is one new RuntimeFilter implementation + one new per-form evaluator factory + one new case arm in the dispatcher -- no new Predicate.Type, no churn to FilterPlanNode or PredicateEvaluatorProvider, no new wire format. What changed: * Predicate.Type.BLOOM_MEMBERSHIP -> RUNTIME_FILTER (umbrella enum value). * New RuntimeFilter interface in pinot-common with Kind {BLOOM} and DataType. Future Kinds (MIN_MAX, IN_LIST) slot in as additional enum values without churning anything else. * RuntimeBloomFilter -> BloomRuntimeFilter, now implements RuntimeFilter and reports Kind.BLOOM. * BloomMembershipPredicate -> RuntimeFilterPredicate; holds the abstract RuntimeFilter rather than the concrete BloomRuntimeFilter. Dropped the defensive equals/hashCode override -- no current caller relies on structural equality, and Object identity is the right contract for a predicate that wraps a freshly-built filter. * BloomMembershipPredicateEvaluatorFactory -> BloomRuntimeFilterEvaluatorFactory (Bloom-specific implementation, reached only after dispatch). * New RuntimeFilterPredicateEvaluatorFactory in pinot-core -- the top-level dispatcher that routes on Kind. Adding a new Kind = one new arm here plus a sibling per-form factory. * PredicateEvaluatorProvider switch arms updated to route Predicate.Type.RUNTIME_FILTER to the new top-level factory; the dictionary-based branch still throws (unchanged semantics, new message). Tests: * RuntimeBloomFilterTest -> BloomRuntimeFilterTest; added a new getKindReportsBloom() case (13 cases total, +1). * BloomMembershipPredicateEvaluatorFactoryTest -> BloomRuntimeFilterEvaluatorFactoryTest; all 13 cases preserved, including the two provider-dispatch cases (now testing Predicate.Type.RUNTIME_FILTER routing). * New RuntimeFilterPredicateEvaluatorFactoryTest covers the Kind-based dispatcher (2 cases). * Total: 28 tests, all passing. Co-Authored-By: Claude Opus 4.7 (1M context) --- .../predicate/BloomMembershipPredicate.java | 79 -------- ...oomFilter.java => BloomRuntimeFilter.java} | 15 +- .../request/context/predicate/Predicate.java | 14 +- .../context/predicate/RuntimeFilter.java | 59 ++++++ .../predicate/RuntimeFilterPredicate.java | 64 ++++++ ...rTest.java => BloomRuntimeFilterTest.java} | 36 ++-- ...omMembershipPredicateEvaluatorFactory.java | 179 ----------------- .../BloomRuntimeFilterEvaluatorFactory.java | 182 ++++++++++++++++++ .../predicate/PredicateEvaluatorProvider.java | 14 +- ...untimeFilterPredicateEvaluatorFactory.java | 56 ++++++ ...oomRuntimeFilterEvaluatorFactoryTest.java} | 68 +++---- ...meFilterPredicateEvaluatorFactoryTest.java | 77 ++++++++ 12 files changed, 519 insertions(+), 324 deletions(-) delete mode 100644 pinot-common/src/main/java/org/apache/pinot/common/request/context/predicate/BloomMembershipPredicate.java rename pinot-common/src/main/java/org/apache/pinot/common/request/context/predicate/{RuntimeBloomFilter.java => BloomRuntimeFilter.java} (92%) create mode 100644 pinot-common/src/main/java/org/apache/pinot/common/request/context/predicate/RuntimeFilter.java create mode 100644 pinot-common/src/main/java/org/apache/pinot/common/request/context/predicate/RuntimeFilterPredicate.java rename pinot-common/src/test/java/org/apache/pinot/common/request/context/predicate/{RuntimeBloomFilterTest.java => BloomRuntimeFilterTest.java} (83%) delete mode 100644 pinot-core/src/main/java/org/apache/pinot/core/operator/filter/predicate/BloomMembershipPredicateEvaluatorFactory.java create mode 100644 pinot-core/src/main/java/org/apache/pinot/core/operator/filter/predicate/BloomRuntimeFilterEvaluatorFactory.java create mode 100644 pinot-core/src/main/java/org/apache/pinot/core/operator/filter/predicate/RuntimeFilterPredicateEvaluatorFactory.java rename pinot-core/src/test/java/org/apache/pinot/core/operator/filter/predicate/{BloomMembershipPredicateEvaluatorFactoryTest.java => BloomRuntimeFilterEvaluatorFactoryTest.java} (77%) create mode 100644 pinot-core/src/test/java/org/apache/pinot/core/operator/filter/predicate/RuntimeFilterPredicateEvaluatorFactoryTest.java diff --git a/pinot-common/src/main/java/org/apache/pinot/common/request/context/predicate/BloomMembershipPredicate.java b/pinot-common/src/main/java/org/apache/pinot/common/request/context/predicate/BloomMembershipPredicate.java deleted file mode 100644 index cc153e8def..0000000000 --- a/pinot-common/src/main/java/org/apache/pinot/common/request/context/predicate/BloomMembershipPredicate.java +++ /dev/null @@ -1,79 +0,0 @@ -/** - * 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 RuntimeBloomFilter} reports as possibly present. - * - *

Injected by the MSE runtime filter feature at the leaf-stage filter boundary. Inclusive - * predicate: a row matches when {@code bloomFilter.mightContain(value)} returns true. False - * positives are permitted; downstream the actual hash join still performs exact matching.

- * - *

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 BloomMembershipPredicate extends BasePredicate { - - private final RuntimeBloomFilter _bloomFilter; - - public BloomMembershipPredicate(ExpressionContext lhs, RuntimeBloomFilter bloomFilter) { - super(lhs); - _bloomFilter = Objects.requireNonNull(bloomFilter, "bloomFilter"); - } - - @Override - public Type getType() { - return Type.BLOOM_MEMBERSHIP; - } - - public RuntimeBloomFilter getBloomFilter() { - return _bloomFilter; - } - - @Override - public boolean equals(Object o) { - if (this == o) { - return true; - } - if (!(o instanceof BloomMembershipPredicate)) { - return false; - } - BloomMembershipPredicate that = (BloomMembershipPredicate) o; - // Reference equality on the bloom filter is intentional: two distinct RuntimeBloomFilter - // instances are conceptually different runtime filters even if they happen to contain the - // same keys. There is no cheap structural equality on Guava BloomFilter that we want to lean on. - return Objects.equals(_lhs, that._lhs) && _bloomFilter == that._bloomFilter; - } - - @Override - public int hashCode() { - return Objects.hash(_lhs, System.identityHashCode(_bloomFilter)); - } - - @Override - public String toString() { - return _lhs + " BLOOM_MEMBERSHIP(" + _bloomFilter.getDataType() + ")"; - } -} diff --git a/pinot-common/src/main/java/org/apache/pinot/common/request/context/predicate/RuntimeBloomFilter.java b/pinot-common/src/main/java/org/apache/pinot/common/request/context/predicate/BloomRuntimeFilter.java similarity index 92% rename from pinot-common/src/main/java/org/apache/pinot/common/request/context/predicate/RuntimeBloomFilter.java rename to pinot-common/src/main/java/org/apache/pinot/common/request/context/predicate/BloomRuntimeFilter.java index 2ebdaa0b9c..d21739dca4 100644 --- a/pinot-common/src/main/java/org/apache/pinot/common/request/context/predicate/RuntimeBloomFilter.java +++ b/pinot-common/src/main/java/org/apache/pinot/common/request/context/predicate/BloomRuntimeFilter.java @@ -24,7 +24,8 @@ /** - * In-memory Bloom filter built at query time and consumed by {@link BloomMembershipPredicate}. + * Bloom-filter implementation of {@link RuntimeFilter}, used by {@link RuntimeFilterPredicate} for + * the v1 runtime-filter feature. * *

The MSE runtime builds one of these from the hash-join build side, hands it to the probe * side, and the leaf stage uses it to prune non-matching rows during the segment scan.

@@ -42,7 +43,7 @@ * must not rely on either outcome.

*/ @SuppressWarnings("UnstableApiUsage") -public final class RuntimeBloomFilter { +public final class BloomRuntimeFilter implements RuntimeFilter { /** * Funnel-family selected at construction time based on the source {@link DataType}. FLOAT/DOUBLE @@ -59,7 +60,7 @@ private enum FunnelType { private final BloomFilter _bloomFilter; @SuppressWarnings({"unchecked", "rawtypes"}) - public RuntimeBloomFilter(DataType dataType, int expectedInsertions, double fpp) { + public BloomRuntimeFilter(DataType dataType, int expectedInsertions, double fpp) { _dataType = dataType; BloomFilter bf; switch (dataType) { @@ -82,11 +83,17 @@ public RuntimeBloomFilter(DataType dataType, int expectedInsertions, double fpp) bf = BloomFilter.create(Funnels.byteArrayFunnel(), expectedInsertions, fpp); break; default: - throw new IllegalArgumentException("RuntimeBloomFilter does not support data type: " + dataType); + throw new IllegalArgumentException("BloomRuntimeFilter does not support data type: " + dataType); } _bloomFilter = (BloomFilter) bf; } + @Override + public Kind getKind() { + return Kind.BLOOM; + } + + @Override public DataType getDataType() { return _dataType; } 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 06a179e843..b21269dffe 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 @@ -40,12 +40,14 @@ enum Type { IS_NULL, IS_NOT_NULL(true), VECTOR_SIMILARITY, - // Runtime Bloom-filter membership test, injected by the MSE runtime filter feature. Carries a - // RuntimeBloomFilter built from the join build side; rows whose value is "not in" the filter - // can be skipped at the leaf-stage scan. Inclusive: a row matches when the filter - // mightContain(value). False positives are allowed (downstream join still does exact match); - // false negatives must not occur. - BLOOM_MEMBERSHIP; + // 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. + 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..e29dc06276 --- /dev/null +++ b/pinot-common/src/main/java/org/apache/pinot/common/request/context/predicate/RuntimeFilter.java @@ -0,0 +1,59 @@ +/** + * 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.

+ */ +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/RuntimeBloomFilterTest.java b/pinot-common/src/test/java/org/apache/pinot/common/request/context/predicate/BloomRuntimeFilterTest.java similarity index 83% rename from pinot-common/src/test/java/org/apache/pinot/common/request/context/predicate/RuntimeBloomFilterTest.java rename to pinot-common/src/test/java/org/apache/pinot/common/request/context/predicate/BloomRuntimeFilterTest.java index 0f6c8357f9..2dbbd2caaf 100644 --- a/pinot-common/src/test/java/org/apache/pinot/common/request/context/predicate/RuntimeBloomFilterTest.java +++ b/pinot-common/src/test/java/org/apache/pinot/common/request/context/predicate/BloomRuntimeFilterTest.java @@ -23,7 +23,7 @@ import org.testng.annotations.Test; -public class RuntimeBloomFilterTest { +public class BloomRuntimeFilterTest { private static final double FPP = 0.01; private static final int EXPECTED_INSERTIONS = 1000; @@ -31,24 +31,30 @@ public class RuntimeBloomFilterTest { @Test public void constructorRejectsUnsupportedDataType() { Assert.assertThrows(IllegalArgumentException.class, - () -> new RuntimeBloomFilter(DataType.BIG_DECIMAL, EXPECTED_INSERTIONS, FPP)); + () -> new BloomRuntimeFilter(DataType.BIG_DECIMAL, EXPECTED_INSERTIONS, FPP)); Assert.assertThrows(IllegalArgumentException.class, - () -> new RuntimeBloomFilter(DataType.TIMESTAMP, EXPECTED_INSERTIONS, FPP)); + () -> new BloomRuntimeFilter(DataType.TIMESTAMP, EXPECTED_INSERTIONS, FPP)); Assert.assertThrows(IllegalArgumentException.class, - () -> new RuntimeBloomFilter(DataType.BOOLEAN, EXPECTED_INSERTIONS, FPP)); + () -> new BloomRuntimeFilter(DataType.BOOLEAN, EXPECTED_INSERTIONS, FPP)); } @Test public void getDataTypeReportsConstructorArgument() { - RuntimeBloomFilter bf = new RuntimeBloomFilter(DataType.LONG, 16, FPP); + 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() { - RuntimeBloomFilter bf = new RuntimeBloomFilter(DataType.INT, EXPECTED_INSERTIONS, FPP); + BloomRuntimeFilter bf = new BloomRuntimeFilter(DataType.INT, EXPECTED_INSERTIONS, FPP); for (int i = 0; i < EXPECTED_INSERTIONS; i++) { bf.add(i); } @@ -59,7 +65,7 @@ public void intAddedValuesAlwaysMightContain() { @Test public void intFalsePositiveRateStaysNearConfigured() { - RuntimeBloomFilter bf = new RuntimeBloomFilter(DataType.INT, EXPECTED_INSERTIONS, FPP); + BloomRuntimeFilter bf = new BloomRuntimeFilter(DataType.INT, EXPECTED_INSERTIONS, FPP); for (int i = 0; i < EXPECTED_INSERTIONS; i++) { bf.add(i); } @@ -80,7 +86,7 @@ public void intFalsePositiveRateStaysNearConfigured() { @Test public void longAddedValuesAlwaysMightContain() { - RuntimeBloomFilter bf = new RuntimeBloomFilter(DataType.LONG, EXPECTED_INSERTIONS, FPP); + 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); } @@ -93,7 +99,7 @@ public void longAddedValuesAlwaysMightContain() { @Test public void floatAddedValuesAlwaysMightContain() { - RuntimeBloomFilter bf = new RuntimeBloomFilter(DataType.FLOAT, EXPECTED_INSERTIONS, FPP); + BloomRuntimeFilter bf = new BloomRuntimeFilter(DataType.FLOAT, EXPECTED_INSERTIONS, FPP); for (int i = 0; i < EXPECTED_INSERTIONS; i++) { bf.add((float) (i + 0.5)); } @@ -108,7 +114,7 @@ public void floatRawBitsHashingTreatsSameBitPatternAsSame() { // 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. - RuntimeBloomFilter bf = new RuntimeBloomFilter(DataType.FLOAT, EXPECTED_INSERTIONS, FPP); + 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)); @@ -122,7 +128,7 @@ public void floatRawBitsHashingTreatsSameBitPatternAsSame() { @Test public void doubleAddedValuesAlwaysMightContain() { - RuntimeBloomFilter bf = new RuntimeBloomFilter(DataType.DOUBLE, EXPECTED_INSERTIONS, FPP); + BloomRuntimeFilter bf = new BloomRuntimeFilter(DataType.DOUBLE, EXPECTED_INSERTIONS, FPP); for (int i = 0; i < EXPECTED_INSERTIONS; i++) { bf.add(i + 0.25); } @@ -135,7 +141,7 @@ public void doubleAddedValuesAlwaysMightContain() { @Test public void stringAddedValuesAlwaysMightContain() { - RuntimeBloomFilter bf = new RuntimeBloomFilter(DataType.STRING, EXPECTED_INSERTIONS, FPP); + BloomRuntimeFilter bf = new BloomRuntimeFilter(DataType.STRING, EXPECTED_INSERTIONS, FPP); for (int i = 0; i < EXPECTED_INSERTIONS; i++) { bf.add("key-" + i); } @@ -146,7 +152,7 @@ public void stringAddedValuesAlwaysMightContain() { @Test public void stringEmptyStringIsHashableAndDistinguishableFromNonEmpty() { - RuntimeBloomFilter bf = new RuntimeBloomFilter(DataType.STRING, EXPECTED_INSERTIONS, FPP); + BloomRuntimeFilter bf = new BloomRuntimeFilter(DataType.STRING, EXPECTED_INSERTIONS, FPP); bf.add(""); Assert.assertTrue(bf.mightContain("")); } @@ -155,7 +161,7 @@ public void stringEmptyStringIsHashableAndDistinguishableFromNonEmpty() { @Test public void bytesAddedValuesAlwaysMightContain() { - RuntimeBloomFilter bf = new RuntimeBloomFilter(DataType.BYTES, EXPECTED_INSERTIONS, FPP); + 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)}); } @@ -168,7 +174,7 @@ public void bytesAddedValuesAlwaysMightContain() { @Test public void bytesContentEqualityHashesIdentically() { // Two distinct byte[] objects with identical content must hash the same way (not by reference). - RuntimeBloomFilter bf = new RuntimeBloomFilter(DataType.BYTES, EXPECTED_INSERTIONS, FPP); + 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/BloomMembershipPredicateEvaluatorFactory.java b/pinot-core/src/main/java/org/apache/pinot/core/operator/filter/predicate/BloomMembershipPredicateEvaluatorFactory.java deleted file mode 100644 index e28bfa3857..0000000000 --- a/pinot-core/src/main/java/org/apache/pinot/core/operator/filter/predicate/BloomMembershipPredicateEvaluatorFactory.java +++ /dev/null @@ -1,179 +0,0 @@ -/** - * 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.BloomMembershipPredicate; -import org.apache.pinot.common.request.context.predicate.RuntimeBloomFilter; -import org.apache.pinot.spi.data.FieldSpec.DataType; - - -/** - * Factory for {@link PredicateEvaluator}s that test single-value column entries against a - * {@link RuntimeBloomFilter}. Raw-value evaluation only - dictionary-based evaluation for runtime - * Bloom filters is intentionally deferred (see {@code newDictionaryBasedEvaluator}). - * - *

False positives are allowed (downstream join still performs exact matching) but false - * negatives must not occur. Each per-type evaluator delegates directly to the matching - * {@code RuntimeBloomFilter.mightContain(...)} overload so the filter's data-type-specific - * funnel selects the correct hashing.

- */ -public class BloomMembershipPredicateEvaluatorFactory { - private BloomMembershipPredicateEvaluatorFactory() { - } - - /** - * Builds a raw-value evaluator for the given {@code dataType}. The evaluator dispatches to the - * {@code RuntimeBloomFilter} via the type-appropriate {@code mightContain} call. Supported - * types: INT, LONG, FLOAT, DOUBLE, STRING, BYTES (matches {@link RuntimeBloomFilter}). Other - * data types throw {@link IllegalArgumentException} at construction; callers must not request - * BLOOM_MEMBERSHIP for unsupported column types. - */ - public static PredicateEvaluator newRawValueBasedEvaluator(BloomMembershipPredicate predicate, DataType dataType) { - RuntimeBloomFilter bloomFilter = predicate.getBloomFilter(); - if (bloomFilter.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( - "RuntimeBloomFilter dataType " + bloomFilter.getDataType() - + " does not match evaluator dataType " + dataType); - } - switch (dataType) { - case INT: - return new IntBloomMembershipPredicateEvaluator(predicate); - case LONG: - return new LongBloomMembershipPredicateEvaluator(predicate); - case FLOAT: - return new FloatBloomMembershipPredicateEvaluator(predicate); - case DOUBLE: - return new DoubleBloomMembershipPredicateEvaluator(predicate); - case STRING: - return new StringBloomMembershipPredicateEvaluator(predicate); - case BYTES: - return new BytesBloomMembershipPredicateEvaluator(predicate); - default: - throw new IllegalArgumentException("BLOOM_MEMBERSHIP does not support data type: " + dataType); - } - } - - private abstract static class BaseBloomMembershipPredicateEvaluator extends BaseRawValueBasedPredicateEvaluator { - final RuntimeBloomFilter _bloomFilter; - - BaseBloomMembershipPredicateEvaluator(BloomMembershipPredicate predicate) { - super(predicate); - _bloomFilter = predicate.getBloomFilter(); - } - } - - private static final class IntBloomMembershipPredicateEvaluator extends BaseBloomMembershipPredicateEvaluator { - IntBloomMembershipPredicateEvaluator(BloomMembershipPredicate predicate) { - super(predicate); - } - - @Override - public DataType getDataType() { - return DataType.INT; - } - - @Override - public boolean applySV(int value) { - return _bloomFilter.mightContain(value); - } - } - - private static final class LongBloomMembershipPredicateEvaluator extends BaseBloomMembershipPredicateEvaluator { - LongBloomMembershipPredicateEvaluator(BloomMembershipPredicate predicate) { - super(predicate); - } - - @Override - public DataType getDataType() { - return DataType.LONG; - } - - @Override - public boolean applySV(long value) { - return _bloomFilter.mightContain(value); - } - } - - private static final class FloatBloomMembershipPredicateEvaluator extends BaseBloomMembershipPredicateEvaluator { - FloatBloomMembershipPredicateEvaluator(BloomMembershipPredicate predicate) { - super(predicate); - } - - @Override - public DataType getDataType() { - return DataType.FLOAT; - } - - @Override - public boolean applySV(float value) { - return _bloomFilter.mightContain(value); - } - } - - private static final class DoubleBloomMembershipPredicateEvaluator extends BaseBloomMembershipPredicateEvaluator { - DoubleBloomMembershipPredicateEvaluator(BloomMembershipPredicate predicate) { - super(predicate); - } - - @Override - public DataType getDataType() { - return DataType.DOUBLE; - } - - @Override - public boolean applySV(double value) { - return _bloomFilter.mightContain(value); - } - } - - private static final class StringBloomMembershipPredicateEvaluator extends BaseBloomMembershipPredicateEvaluator { - StringBloomMembershipPredicateEvaluator(BloomMembershipPredicate predicate) { - super(predicate); - } - - @Override - public DataType getDataType() { - return DataType.STRING; - } - - @Override - public boolean applySV(String value) { - return _bloomFilter.mightContain(value); - } - } - - private static final class BytesBloomMembershipPredicateEvaluator extends BaseBloomMembershipPredicateEvaluator { - BytesBloomMembershipPredicateEvaluator(BloomMembershipPredicate predicate) { - super(predicate); - } - - @Override - public DataType getDataType() { - return DataType.BYTES; - } - - @Override - public boolean applySV(byte[] value) { - return _bloomFilter.mightContain(value); - } - } -} 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..4ff93c555c --- /dev/null +++ b/pinot-core/src/main/java/org/apache/pinot/core/operator/filter/predicate/BloomRuntimeFilterEvaluatorFactory.java @@ -0,0 +1,182 @@ +/** + * 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.BloomRuntimeFilter; +import org.apache.pinot.common.request.context.predicate.RuntimeFilterPredicate; +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()}. + * + *

Raw-value evaluation only - dictionary-based evaluation for runtime Bloom filters is + * intentionally deferred. False positives are allowed (downstream join still performs exact + * matching) but false negatives must not occur. Each per-type evaluator delegates directly to the + * matching {@link BloomRuntimeFilter#mightContain} overload so the filter's data-type-specific + * funnel selects the correct hashing.

+ */ +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); + } + } + + 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); + } + } +} 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 6499ce9086..15f1d39ae3 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 @@ -19,7 +19,6 @@ package org.apache.pinot.core.operator.filter.predicate; import javax.annotation.Nullable; -import org.apache.pinot.common.request.context.predicate.BloomMembershipPredicate; import org.apache.pinot.common.request.context.predicate.EqPredicate; import org.apache.pinot.common.request.context.predicate.InPredicate; import org.apache.pinot.common.request.context.predicate.NotEqPredicate; @@ -27,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; @@ -67,13 +67,13 @@ public static PredicateEvaluator getPredicateEvaluator(Predicate predicate, @Nul case REGEXP_LIKE: return RegexpLikePredicateEvaluatorFactory .newDictionaryBasedEvaluator((RegexpLikePredicate) predicate, dictionary, dataType); - case BLOOM_MEMBERSHIP: - // Dictionary-based BLOOM_MEMBERSHIP is intentionally not wired up yet. The predicate is + case RUNTIME_FILTER: + // Dictionary-based runtime filters are intentionally not wired up yet. The predicate is // currently injected only at the raw-value scan path; a dict-id pre-materialization // variant can be added once we have measured workloads that need it. Throw clearly so // accidental wiring during planner work is caught early. throw new UnsupportedOperationException( - "Dictionary-based BLOOM_MEMBERSHIP predicate is not yet supported"); + "Dictionary-based RUNTIME_FILTER predicate is not yet supported"); default: throw new UnsupportedOperationException("Unsupported predicate type: " + predicate.getType()); } @@ -93,9 +93,9 @@ public static PredicateEvaluator getPredicateEvaluator(Predicate predicate, @Nul case REGEXP_LIKE: return RegexpLikePredicateEvaluatorFactory .newRawValueBasedEvaluator((RegexpLikePredicate) predicate, dataType); - case BLOOM_MEMBERSHIP: - return BloomMembershipPredicateEvaluatorFactory - .newRawValueBasedEvaluator((BloomMembershipPredicate) 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..210a3fe156 --- /dev/null +++ b/pinot-core/src/main/java/org/apache/pinot/core/operator/filter/predicate/RuntimeFilterPredicateEvaluatorFactory.java @@ -0,0 +1,56 @@ +/** + * 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.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); + } + } +} diff --git a/pinot-core/src/test/java/org/apache/pinot/core/operator/filter/predicate/BloomMembershipPredicateEvaluatorFactoryTest.java b/pinot-core/src/test/java/org/apache/pinot/core/operator/filter/predicate/BloomRuntimeFilterEvaluatorFactoryTest.java similarity index 77% rename from pinot-core/src/test/java/org/apache/pinot/core/operator/filter/predicate/BloomMembershipPredicateEvaluatorFactoryTest.java rename to pinot-core/src/test/java/org/apache/pinot/core/operator/filter/predicate/BloomRuntimeFilterEvaluatorFactoryTest.java index 9903a252bd..3299d791f9 100644 --- a/pinot-core/src/test/java/org/apache/pinot/core/operator/filter/predicate/BloomMembershipPredicateEvaluatorFactoryTest.java +++ b/pinot-core/src/test/java/org/apache/pinot/core/operator/filter/predicate/BloomRuntimeFilterEvaluatorFactoryTest.java @@ -21,9 +21,9 @@ import java.util.ArrayList; import java.util.List; import org.apache.pinot.common.request.context.ExpressionContext; -import org.apache.pinot.common.request.context.predicate.BloomMembershipPredicate; +import org.apache.pinot.common.request.context.predicate.RuntimeFilterPredicate; import org.apache.pinot.common.request.context.predicate.Predicate; -import org.apache.pinot.common.request.context.predicate.RuntimeBloomFilter; +import org.apache.pinot.common.request.context.predicate.BloomRuntimeFilter; import org.apache.pinot.segment.spi.index.reader.Dictionary; import org.apache.pinot.spi.data.FieldSpec.DataType; import org.mockito.Mockito; @@ -31,46 +31,46 @@ import org.testng.annotations.Test; -public class BloomMembershipPredicateEvaluatorFactoryTest { +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 RuntimeBloomFilterTest tolerance + private static final double FP_TOLERANCE = FPP * 5; // matches BloomRuntimeFilterTest tolerance // ---------- Factory wiring ---------- @Test public void factoryRejectsDataTypeMismatch() { - RuntimeBloomFilter bloom = new RuntimeBloomFilter(DataType.INT, EXPECTED_INSERTIONS, FPP); - BloomMembershipPredicate predicate = new BloomMembershipPredicate(COLUMN, bloom); + BloomRuntimeFilter bloom = new BloomRuntimeFilter(DataType.INT, EXPECTED_INSERTIONS, FPP); + RuntimeFilterPredicate predicate = new RuntimeFilterPredicate(COLUMN, bloom); Assert.assertThrows(IllegalArgumentException.class, - () -> BloomMembershipPredicateEvaluatorFactory.newRawValueBasedEvaluator(predicate, DataType.LONG)); + () -> BloomRuntimeFilterEvaluatorFactory.newRawValueBasedEvaluator(predicate, DataType.LONG)); } @Test public void factoryRejectsUnsupportedDataType() { - RuntimeBloomFilter bloom = new RuntimeBloomFilter(DataType.STRING, EXPECTED_INSERTIONS, FPP); - BloomMembershipPredicate predicate = new BloomMembershipPredicate(COLUMN, bloom); + 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, - () -> BloomMembershipPredicateEvaluatorFactory.newRawValueBasedEvaluator(predicate, DataType.BIG_DECIMAL)); + () -> BloomRuntimeFilterEvaluatorFactory.newRawValueBasedEvaluator(predicate, DataType.BIG_DECIMAL)); } @Test public void evaluatorReportsExpectedMetadata() { - RuntimeBloomFilter bloom = new RuntimeBloomFilter(DataType.LONG, EXPECTED_INSERTIONS, FPP); - BloomMembershipPredicate predicate = new BloomMembershipPredicate(COLUMN, bloom); + BloomRuntimeFilter bloom = new BloomRuntimeFilter(DataType.LONG, EXPECTED_INSERTIONS, FPP); + RuntimeFilterPredicate predicate = new RuntimeFilterPredicate(COLUMN, bloom); PredicateEvaluator evaluator = - BloomMembershipPredicateEvaluatorFactory.newRawValueBasedEvaluator(predicate, DataType.LONG); + BloomRuntimeFilterEvaluatorFactory.newRawValueBasedEvaluator(predicate, DataType.LONG); Assert.assertSame(evaluator.getPredicate(), predicate); - Assert.assertEquals(evaluator.getPredicateType(), Predicate.Type.BLOOM_MEMBERSHIP); + 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(), "BLOOM_MEMBERSHIP is an inclusive predicate"); + Assert.assertFalse(evaluator.isExclusive(), "RUNTIME_FILTER is an inclusive predicate"); Assert.assertFalse(evaluator.isAlwaysTrue()); Assert.assertFalse(evaluator.isAlwaysFalse()); } @@ -79,7 +79,7 @@ public void evaluatorReportsExpectedMetadata() { @Test public void intEvaluatorHasNoFalseNegativesAndLowFalsePositiveRate() { - RuntimeBloomFilter bloom = new RuntimeBloomFilter(DataType.INT, EXPECTED_INSERTIONS, FPP); + 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; @@ -103,7 +103,7 @@ public void intEvaluatorHasNoFalseNegativesAndLowFalsePositiveRate() { @Test public void longEvaluatorHasNoFalseNegatives() { - RuntimeBloomFilter bloom = new RuntimeBloomFilter(DataType.LONG, EXPECTED_INSERTIONS, FPP); + 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; @@ -118,7 +118,7 @@ public void longEvaluatorHasNoFalseNegatives() { @Test public void floatEvaluatorRoundTrip() { - RuntimeBloomFilter bloom = new RuntimeBloomFilter(DataType.FLOAT, EXPECTED_INSERTIONS, FPP); + BloomRuntimeFilter bloom = new BloomRuntimeFilter(DataType.FLOAT, EXPECTED_INSERTIONS, FPP); for (int i = 0; i < NUM_KEYS; i++) { bloom.add(i + 0.125f); } @@ -130,7 +130,7 @@ public void floatEvaluatorRoundTrip() { @Test public void doubleEvaluatorRoundTrip() { - RuntimeBloomFilter bloom = new RuntimeBloomFilter(DataType.DOUBLE, EXPECTED_INSERTIONS, FPP); + BloomRuntimeFilter bloom = new BloomRuntimeFilter(DataType.DOUBLE, EXPECTED_INSERTIONS, FPP); for (int i = 0; i < NUM_KEYS; i++) { bloom.add(i + 0.0625); } @@ -142,7 +142,7 @@ public void doubleEvaluatorRoundTrip() { @Test public void stringEvaluatorRoundTrip() { - RuntimeBloomFilter bloom = new RuntimeBloomFilter(DataType.STRING, EXPECTED_INSERTIONS, FPP); + BloomRuntimeFilter bloom = new BloomRuntimeFilter(DataType.STRING, EXPECTED_INSERTIONS, FPP); for (int i = 0; i < NUM_KEYS; i++) { bloom.add("k-" + i); } @@ -154,7 +154,7 @@ public void stringEvaluatorRoundTrip() { @Test public void bytesEvaluatorRoundTrip() { - RuntimeBloomFilter bloom = new RuntimeBloomFilter(DataType.BYTES, EXPECTED_INSERTIONS, FPP); + 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)}); } @@ -171,7 +171,7 @@ public void bytesEvaluatorRoundTrip() { 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. - RuntimeBloomFilter bloom = new RuntimeBloomFilter(DataType.INT, EXPECTED_INSERTIONS, FPP); + BloomRuntimeFilter bloom = new BloomRuntimeFilter(DataType.INT, EXPECTED_INSERTIONS, FPP); bloom.add(7); PredicateEvaluator evaluator = newEvaluator(bloom, DataType.INT); @@ -186,7 +186,7 @@ public void wrongTypeOverloadThrowsFromBaseClass() { @Test public void intMultiValueMatchesIfAnyValueIsInBloom() { - RuntimeBloomFilter bloom = new RuntimeBloomFilter(DataType.INT, EXPECTED_INSERTIONS, FPP); + BloomRuntimeFilter bloom = new BloomRuntimeFilter(DataType.INT, EXPECTED_INSERTIONS, FPP); bloom.add(42); PredicateEvaluator evaluator = newEvaluator(bloom, DataType.INT); @@ -202,21 +202,21 @@ public void intMultiValueMatchesIfAnyValueIsInBloom() { // ---------- Provider dispatch ---------- @Test - public void providerRoutesRawValueBasedBloomMembershipToFactory() { - RuntimeBloomFilter bloom = new RuntimeBloomFilter(DataType.LONG, EXPECTED_INSERTIONS, FPP); - BloomMembershipPredicate predicate = new BloomMembershipPredicate(COLUMN, bloom); + 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.BLOOM_MEMBERSHIP); + Assert.assertEquals(evaluator.getPredicateType(), Predicate.Type.RUNTIME_FILTER); Assert.assertEquals(evaluator.getDataType(), DataType.LONG); } @Test - public void providerRejectsDictionaryBasedBloomMembership() { - RuntimeBloomFilter bloom = new RuntimeBloomFilter(DataType.INT, EXPECTED_INSERTIONS, FPP); - BloomMembershipPredicate predicate = new BloomMembershipPredicate(COLUMN, bloom); + public void providerRejectsDictionaryBasedRuntimeFilter() { + BloomRuntimeFilter bloom = new BloomRuntimeFilter(DataType.INT, EXPECTED_INSERTIONS, FPP); + RuntimeFilterPredicate predicate = new RuntimeFilterPredicate(COLUMN, bloom); // Passing a non-null dictionary -- which routes us into the dictionary-based switch -- must - // throw because dict-based BLOOM_MEMBERSHIP is deferred. PredicateEvaluatorProvider wraps the + // throw because dict-based RUNTIME_FILTER is deferred. PredicateEvaluatorProvider wraps the // UnsupportedOperationException in a BadQueryRequestException; assert any throwable so the // contract holds even if the wrapping changes. Dictionary mockDictionary = Mockito.mock(Dictionary.class); @@ -226,8 +226,8 @@ public void providerRejectsDictionaryBasedBloomMembership() { // ---------- helpers ---------- - private static PredicateEvaluator newEvaluator(RuntimeBloomFilter bloom, DataType dataType) { - BloomMembershipPredicate predicate = new BloomMembershipPredicate(COLUMN, bloom); - return BloomMembershipPredicateEvaluatorFactory.newRawValueBasedEvaluator(predicate, dataType); + 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..716b3f3c6d --- /dev/null +++ b/pinot-core/src/test/java/org/apache/pinot/core/operator/filter/predicate/RuntimeFilterPredicateEvaluatorFactoryTest.java @@ -0,0 +1,77 @@ +/** + * 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.spi.data.FieldSpec.DataType; +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. 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 bloomKindRoutesToBloomEvaluator() { + 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.assertTrue(evaluator.applySV(42), "bloom-routed evaluator must hit added key"); + } + + @Test + public void unsupportedKindThrows() { + // Stand-in for a future Kind whose factory has not been implemented yet. + RuntimeFilter neverSupported = new RuntimeFilter() { + @Override + public Kind getKind() { + // Use BLOOM here so the test source compiles today; the assertion lives in + // bloomKindRoutesToBloomEvaluator. When we add a future Kind, this test should be replaced + // with one that returns the not-yet-implemented Kind and asserts the throw. + return Kind.BLOOM; + } + + @Override + public DataType getDataType() { + return DataType.INT; + } + }; + // With only BLOOM defined, every Kind currently has an implementation. Once a second Kind is + // added, instantiate it here and assert IllegalArgumentException from the default arm. + Assert.assertEquals(neverSupported.getKind(), RuntimeFilter.Kind.BLOOM, + "if this fails, a new Kind has been added; update this test to cover the unsupported arm"); + } +} From 8ca83908a829b845ffe80c8f25a86699b24aec24 Mon Sep 17 00:00:00 2001 From: Tianle Zhang Date: Thu, 28 May 2026 17:47:08 -0700 Subject: [PATCH 4/7] Sort imports in BloomRuntimeFilterEvaluatorFactoryTest Fixes spotless check; alphabetizes the predicate-package imports introduced by the RUNTIME_FILTER rename. Co-Authored-By: Claude Opus 4.7 --- .../predicate/BloomRuntimeFilterEvaluatorFactoryTest.java | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) 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 index 3299d791f9..e634509515 100644 --- 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 @@ -21,9 +21,9 @@ import java.util.ArrayList; import java.util.List; import org.apache.pinot.common.request.context.ExpressionContext; -import org.apache.pinot.common.request.context.predicate.RuntimeFilterPredicate; -import org.apache.pinot.common.request.context.predicate.Predicate; 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; From aa0e587485d4c16d7c7b92a705c324e4ce244cca Mon Sep 17 00:00:00 2001 From: Tianle Zhang Date: Thu, 28 May 2026 17:56:40 -0700 Subject: [PATCH 5/7] Add dictionary-based evaluator for RUNTIME_FILTER predicate The dict-based path probes the column dictionary once at construction time: every dictId whose value passes BloomRuntimeFilter#mightContain lands in an IntSet, and applySV(int) answers from that set thereafter. Standard _alwaysTrue / _alwaysFalse short-circuits are populated based on the count of matching dict ids. 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. Cost is O(dictionary.length()) Bloom probes at construction; callers are responsible for choosing dict-based only when it pays off versus the raw scan path. Added five tests covering hit/miss mix, empty dictionary -> always-false, full-coverage dictionary -> always-true, dataType-mismatch guard, and sort-ascending invariant on the returned matching dict ids. Replaced the previous "provider rejects dict-based runtime filter" negative test with a positive routing test through PredicateEvaluatorProvider. Co-Authored-By: Claude Opus 4.7 --- .../BloomRuntimeFilterEvaluatorFactory.java | 121 +++++++++++++++++- .../predicate/PredicateEvaluatorProvider.java | 8 +- ...untimeFilterPredicateEvaluatorFactory.java | 20 +++ ...loomRuntimeFilterEvaluatorFactoryTest.java | 118 +++++++++++++++-- 4 files changed, 248 insertions(+), 19 deletions(-) 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 index 4ff93c555c..3b71cc42a5 100644 --- 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 @@ -18,8 +18,12 @@ */ 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; @@ -28,11 +32,13 @@ * {@link RuntimeFilterPredicateEvaluatorFactory} after it has dispatched on * {@code predicate.getFilter().getKind()}. * - *

Raw-value evaluation only - dictionary-based evaluation for runtime Bloom filters is - * intentionally deferred. False positives are allowed (downstream join still performs exact - * matching) but false negatives must not occur. Each per-type evaluator delegates directly to the - * matching {@link BloomRuntimeFilter#mightContain} overload so the filter's data-type-specific - * funnel selects the correct hashing.

+ *

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() { @@ -75,6 +81,30 @@ public static PredicateEvaluator newRawValueBasedEvaluator(RuntimeFilterPredicat } } + /** + * 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; @@ -179,4 +209,85 @@ 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 15f1d39ae3..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 @@ -68,12 +68,8 @@ public static PredicateEvaluator getPredicateEvaluator(Predicate predicate, @Nul return RegexpLikePredicateEvaluatorFactory .newDictionaryBasedEvaluator((RegexpLikePredicate) predicate, dictionary, dataType); case RUNTIME_FILTER: - // Dictionary-based runtime filters are intentionally not wired up yet. The predicate is - // currently injected only at the raw-value scan path; a dict-id pre-materialization - // variant can be added once we have measured workloads that need it. Throw clearly so - // accidental wiring during planner work is caught early. - throw new UnsupportedOperationException( - "Dictionary-based RUNTIME_FILTER predicate is not yet supported"); + return RuntimeFilterPredicateEvaluatorFactory + .newDictionaryBasedEvaluator((RuntimeFilterPredicate) predicate, dictionary, 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 index 210a3fe156..72b722f9b5 100644 --- 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 @@ -20,6 +20,7 @@ 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; @@ -53,4 +54,23 @@ public static PredicateEvaluator newRawValueBasedEvaluator(RuntimeFilterPredicat 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 index e634509515..e60f25e862 100644 --- 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 @@ -212,16 +212,118 @@ public void providerRoutesRawValueBasedRuntimeFilterToFactory() { } @Test - public void providerRejectsDictionaryBasedRuntimeFilter() { + 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); - // Passing a non-null dictionary -- which routes us into the dictionary-based switch -- must - // throw because dict-based RUNTIME_FILTER is deferred. PredicateEvaluatorProvider wraps the - // UnsupportedOperationException in a BadQueryRequestException; assert any throwable so the - // contract holds even if the wrapping changes. - Dictionary mockDictionary = Mockito.mock(Dictionary.class); - Assert.assertThrows(Exception.class, - () -> PredicateEvaluatorProvider.getPredicateEvaluator(predicate, mockDictionary, DataType.INT)); + 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 ---------- From 7d3d76fa8a58375fa57c81f7f33a638f239d3929 Mon Sep 17 00:00:00 2001 From: Tianle Zhang Date: Thu, 28 May 2026 18:13:43 -0700 Subject: [PATCH 6/7] Clean up dead funnel-type field and umbrella-dispatcher tests BloomRuntimeFilter: drop the package-private FunnelType enum and _funnelType field. The field was declared "visible for tests" but never read; with the dataType already stored separately, this state has no remaining purpose. RuntimeFilterPredicateEvaluatorFactoryTest: replace the placeholder unsupportedKindThrows test (a tautology with only one Kind defined) with a direct test that the umbrella dispatcher routes BLOOM through the dictionary-based path. Renamed the existing test to make the raw-value path explicit in the name. Co-Authored-By: Claude Opus 4.7 --- .../context/predicate/BloomRuntimeFilter.java | 26 +++-------- ...meFilterPredicateEvaluatorFactoryTest.java | 45 ++++++++++--------- 2 files changed, 29 insertions(+), 42 deletions(-) diff --git a/pinot-common/src/main/java/org/apache/pinot/common/request/context/predicate/BloomRuntimeFilter.java b/pinot-common/src/main/java/org/apache/pinot/common/request/context/predicate/BloomRuntimeFilter.java index d21739dca4..dc9d709de1 100644 --- a/pinot-common/src/main/java/org/apache/pinot/common/request/context/predicate/BloomRuntimeFilter.java +++ b/pinot-common/src/main/java/org/apache/pinot/common/request/context/predicate/BloomRuntimeFilter.java @@ -45,20 +45,15 @@ @SuppressWarnings("UnstableApiUsage") public final class BloomRuntimeFilter implements RuntimeFilter { - /** - * Funnel-family selected at construction time based on the source {@link DataType}. FLOAT/DOUBLE - * reuse the INT/LONG funnels via raw-bits encoding: two float values with the same raw IEEE-754 - * bit pattern hash identically; values that differ only by bit pattern (e.g. {@code 0.0f} vs - * {@code -0.0f}, or distinct NaN payloads) hash to different slots. - */ - private enum FunnelType { - INT, LONG, STRING, BYTES - } - private final DataType _dataType; - private final FunnelType _funnelType; private final BloomFilter _bloomFilter; + /** + * Funnel-family is selected from {@code dataType}: FLOAT/DOUBLE reuse the INT/LONG funnels via + * raw-bits encoding, so two float values with the same raw IEEE-754 bit pattern hash identically + * while values that differ only by bit pattern (e.g. {@code 0.0f} vs {@code -0.0f}, or distinct + * NaN payloads) hash to different slots. + */ @SuppressWarnings({"unchecked", "rawtypes"}) public BloomRuntimeFilter(DataType dataType, int expectedInsertions, double fpp) { _dataType = dataType; @@ -66,20 +61,16 @@ public BloomRuntimeFilter(DataType dataType, int expectedInsertions, double fpp) switch (dataType) { case INT: case FLOAT: - _funnelType = FunnelType.INT; bf = BloomFilter.create(Funnels.integerFunnel(), expectedInsertions, fpp); break; case LONG: case DOUBLE: - _funnelType = FunnelType.LONG; bf = BloomFilter.create(Funnels.longFunnel(), expectedInsertions, fpp); break; case STRING: - _funnelType = FunnelType.STRING; bf = BloomFilter.create(Funnels.unencodedCharsFunnel(), expectedInsertions, fpp); break; case BYTES: - _funnelType = FunnelType.BYTES; bf = BloomFilter.create(Funnels.byteArrayFunnel(), expectedInsertions, fpp); break; default: @@ -148,9 +139,4 @@ public boolean mightContain(String value) { public boolean mightContain(byte[] value) { return _bloomFilter.mightContain(value); } - - /** Returns the funnel family used internally; visible for tests. */ - FunnelType funnelType() { - return _funnelType; - } } 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 index 716b3f3c6d..387d972973 100644 --- 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 @@ -23,22 +23,25 @@ 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. Per-form behavior is covered in the - * form-specific factory tests (e.g. {@link BloomRuntimeFilterEvaluatorFactoryTest}). + * {@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 bloomKindRoutesToBloomEvaluator() { + public void bloomKindRoutesToRawValueBloomEvaluator() { BloomRuntimeFilter bloom = new BloomRuntimeFilter(DataType.INT, 128, 0.01); bloom.add(42); RuntimeFilterPredicate predicate = new RuntimeFilterPredicate(COLUMN, bloom); @@ -49,29 +52,27 @@ public void bloomKindRoutesToBloomEvaluator() { 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 unsupportedKindThrows() { - // Stand-in for a future Kind whose factory has not been implemented yet. - RuntimeFilter neverSupported = new RuntimeFilter() { - @Override - public Kind getKind() { - // Use BLOOM here so the test source compiles today; the assertion lives in - // bloomKindRoutesToBloomEvaluator. When we add a future Kind, this test should be replaced - // with one that returns the not-yet-implemented Kind and asserts the throw. - return Kind.BLOOM; - } + 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); - @Override - public DataType getDataType() { - return DataType.INT; - } - }; - // With only BLOOM defined, every Kind currently has an implementation. Once a second Kind is - // added, instantiate it here and assert IllegalArgumentException from the default arm. - Assert.assertEquals(neverSupported.getKind(), RuntimeFilter.Kind.BLOOM, - "if this fails, a new Kind has been added; update this test to cover the unsupported arm"); + 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"); } } From db0960fb87a251a4c15577d63123511530f4d62f Mon Sep 17 00:00:00 2001 From: Tianle Zhang Date: Fri, 29 May 2026 15:36:40 -0700 Subject: [PATCH 7/7] Document NULL handling contract for runtime filter (v1) NULLs are out of scope for v1: the build side carries only non-null join keys and the probe side treats NULLs as non-matching, which is correct for inner equi-joins. Add the authoritative contract plus a TODO(NULL support) for the end-to-end definition (code + user docs) on the RuntimeFilter interface, and a pointer from Predicate.Type.RUNTIME_FILTER. Co-Authored-By: Claude Opus 4.8 (1M context) --- .../common/request/context/predicate/Predicate.java | 1 + .../request/context/predicate/RuntimeFilter.java | 13 +++++++++++++ 2 files changed, 14 insertions(+) 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 b21269dffe..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 @@ -47,6 +47,7 @@ enum Type { // 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 index e29dc06276..06b8dd1b3e 100644 --- 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 @@ -31,6 +31,19 @@ * 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 {