Add RUNTIME_FILTER predicate primitive (Bloom) for MSE runtime filter

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.
+ *
+ * <p>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.</p>
+ *
+ * <p>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.</p>
+ */
+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() + ")";
+  }
+}

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. 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;
 

pinot-common/src/main/java/org/apache/pinot/common/request/context/predicate/RuntimeBloomFilter.java

@@ -0,0 +1,149 @@
+/**
+ * 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}.
+ *
+ * <p>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.</p>
+ *
+ * <p>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 changes will add a wire format alongside the runtime side-channel.</p>
+ *
+ * <p>The caller is responsible for invoking the {@code mightContain} overload that matches the
+ * 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.</p>
+ */
+@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: 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<Object> _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<Object>) 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 by raw IEEE-754 int bits so the INT funnel hashes values
+    // with the same bit pattern to the same slot. Values that differ only by bit pattern
+    // (-0.0f vs 0.0f, distinct NaN payloads) are NOT collapsed by this encoding.
+    _bloomFilter.put(Float.floatToRawIntBits(value));
+  }
+
+  public void add(double value) {
+    _bloomFilter.put(Double.doubleToRawLongBits(value));
+  }
+
+  public void add(String value) {
+    _bloomFilter.put(value);
+  }
+
+  public void add(byte[] value) {
+    _bloomFilter.put(value);
+  }
+
+  public boolean mightContain(int value) {
+    return _bloomFilter.mightContain(value);
+  }
+
+  public boolean mightContain(long value) {
+    return _bloomFilter.mightContain(value);
+  }
+
+  public boolean mightContain(float value) {
+    return _bloomFilter.mightContain(Float.floatToRawIntBits(value));
+  }
+
+  public boolean mightContain(double value) {
+    return _bloomFilter.mightContain(Double.doubleToRawLongBits(value));
+  }
+
+  public boolean mightContain(String value) {
+    return _bloomFilter.mightContain(value);
+  }
+
+  public boolean mightContain(byte[] value) {
+    return _bloomFilter.mightContain(value);
+  }
+
+  /** Returns the funnel family used internally; visible for tests. */
+  FunnelType funnelType() {
+    return _funnelType;
+  }
+}