Add RUNTIME_FILTER predicate primitive (Bloom) for MSE runtime filter

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

@@ -0,0 +1,142 @@
+/**
+ * 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;
+
+
+/**
+ * Bloom-filter implementation of {@link RuntimeFilter}, used by {@link RuntimeFilterPredicate} for
+ * the v1 runtime-filter feature.
+ *
+ * <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 BloomRuntimeFilter implements RuntimeFilter {
+
+  private final DataType _dataType;
+  private final BloomFilter<Object> _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;
+    BloomFilter bf;
+    switch (dataType) {
+      case INT:
+      case FLOAT:
+        bf = BloomFilter.create(Funnels.integerFunnel(), expectedInsertions, fpp);
+        break;
+      case LONG:
+      case DOUBLE:
+        bf = BloomFilter.create(Funnels.longFunnel(), expectedInsertions, fpp);
+        break;
+      case STRING:
+        bf = BloomFilter.create(Funnels.unencodedCharsFunnel(), expectedInsertions, fpp);
+        break;
+      case BYTES:
+        bf = BloomFilter.create(Funnels.byteArrayFunnel(), expectedInsertions, fpp);
+        break;
+      default:
+        throw new IllegalArgumentException("BloomRuntimeFilter does not support data type: " + dataType);
+    }
+    _bloomFilter = (BloomFilter<Object>) bf;
+  }
+
+  @Override
+  public Kind getKind() {
+    return Kind.BLOOM;
+  }
+
+  @Override
+  public DataType getDataType() {
+    return _dataType;
+  }
+
+  public void add(int value) {
+    _bloomFilter.put(value);
+  }
+
+  public void add(long value) {
+    _bloomFilter.put(value);
+  }
+
+  public void add(float value) {
+    // Match BloomFilterIdSet: encode by raw IEEE-754 int bits so the INT funnel hashes values
+    // with the same bit pattern to the same slot. Values that differ only by bit pattern
+    // (-0.0f vs 0.0f, distinct NaN payloads) are NOT collapsed by this encoding.
+    _bloomFilter.put(Float.floatToRawIntBits(value));
+  }
+
+  public void add(double value) {
+    _bloomFilter.put(Double.doubleToRawLongBits(value));
+  }
+
+  public void add(String value) {
+    _bloomFilter.put(value);
+  }
+
+  public void add(byte[] value) {
+    _bloomFilter.put(value);
+  }
+
+  public boolean mightContain(int value) {
+    return _bloomFilter.mightContain(value);
+  }
+
+  public boolean mightContain(long value) {
+    return _bloomFilter.mightContain(value);
+  }
+
+  public boolean mightContain(float value) {
+    return _bloomFilter.mightContain(Float.floatToRawIntBits(value));
+  }
+
+  public boolean mightContain(double value) {
+    return _bloomFilter.mightContain(Double.doubleToRawLongBits(value));
+  }
+
+  public boolean mightContain(String value) {
+    return _bloomFilter.mightContain(value);
+  }
+
+  public boolean mightContain(byte[] value) {
+    return _bloomFilter.mightContain(value);
+  }
+}

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

@@ -39,7 +39,16 @@ enum Type {
     JSON_MATCH,
     IS_NULL,
     IS_NOT_NULL(true),
-    VECTOR_SIMILARITY;
+    VECTOR_SIMILARITY,
+    // Umbrella predicate for runtime filters (dynamic filter pushdown) injected by the MSE runtime
+    // filter feature. Carries a RuntimeFilter built from the hash-join build side; rows whose
+    // value the filter reports as "definitely not present" can be skipped at the leaf-stage scan.
+    // v1 only carries a Bloom filter (Kind.BLOOM); min/max and IN-list variants slot in as
+    // additional Kind values without a new Predicate.Type or new dispatch chain. Inclusive
+    // semantics: a row matches when the filter says "maybe present". False positives are allowed
+    // (downstream join still performs the exact match); false negatives must not occur.
+    // NULL handling is out of scope for v1; see RuntimeFilter for the NULL contract and TODO.
+    RUNTIME_FILTER;
 
     private final boolean _exclusive;
 

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

@@ -0,0 +1,72 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pinot.common.request.context.predicate;
+
+import org.apache.pinot.spi.data.FieldSpec.DataType;
+
+
+/**
+ * In-memory filter built at query time by the build side of a hash join and consumed by the probe
+ * side (typically at segment-scan time via {@link RuntimeFilterPredicate}). Carries enough
+ * type information for the evaluator factory to dispatch to a {@link Kind}-specific implementation
+ * without the rest of the engine knowing about the concrete filter form.
+ *
+ * <p>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.</p>
+ *
+ * <h3>NULL handling (v1 contract)</h3>
+ * <p>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 <em>not</em> 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).
+ *
+ * <p>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.</p>
+ */
+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();
+}

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