[GLUTEN-11708][VL] Translate might_contain as a subfield filter for scan-level bloom filter pushdown

cpp/velox/operators/functions/SparkExprToSubfieldFilterParser.cc

@@ -16,11 +16,69 @@
  */
 #include "operators/functions/SparkExprToSubfieldFilterParser.h"
 
+#include "velox/common/base/BloomFilter.h"
+#include "velox/expression/Expr.h"
+#include "velox/vector/ComplexVector.h"
+
 namespace gluten {
 
 using namespace facebook::velox;
 
 namespace {
+
+// Evaluates an expression as a constant. Returns nullptr if the expression is
+// not constant or evaluation fails. Errors are intentionally swallowed because
+// a non-evaluable expression simply means the filter cannot be pushed down.
+VectorPtr toConstant(const core::TypedExprPtr& expr, core::ExpressionEvaluator* evaluator) {
+  auto exprSet = evaluator->compile(expr);
+  if (!exprSet->exprs()[0]->isConstantExpr()) {
+    return nullptr;
+  }
+  RowVector input(evaluator->pool(), ROW({}, {}), nullptr, 1, std::vector<VectorPtr>{});
+  SelectivityVector rows(1);
+  VectorPtr result;
+  try {
+    evaluator->evaluate(exprSet.get(), rows, input, result);
+  } catch (const VeloxUserError&) {
+    return nullptr;
+  }
+  return result;
+}
+
+/// Subfield filter backed by Velox's BloomFilter from bloom_filter_agg / might_contain.
+class SparkMightContain final : public common::Filter {
+ public:
+  SparkMightContain(const char* serializedData, bool nullAllowed)
+      : Filter(true, nullAllowed, common::FilterKind::kBigintValuesUsingBloomFilter) {
+    bloomFilter_.merge(serializedData);
+  }
+
+  bool testInt64(int64_t value) const final {
+    return bloomFilter_.mayContain(folly::hasher<int64_t>()(value));
+  }
+
+  bool testInt64Range(int64_t /*min*/, int64_t /*max*/, bool /*hasNull*/) const final {
+    return true;
+  }
+
+  std::unique_ptr<Filter> clone(std::optional<bool> nullAllowed) const override {
+    std::vector<char> data(bloomFilter_.serializedSize());
+    bloomFilter_.serialize(data.data());
+    return std::make_unique<SparkMightContain>(data.data(), nullAllowed.value_or(nullAllowed_));
+  }
+
+  bool testingEquals(const Filter& other) const override {
+    return dynamic_cast<const SparkMightContain*>(&other) != nullptr;
+  }
+
+  folly::dynamic serialize() const override {
+    VELOX_UNSUPPORTED("Serialization is not supported for SparkMightContain");
+  }
+
+ private:
+  BloomFilter<> bloomFilter_;
+};
+
 std::optional<std::pair<facebook::velox::common::Subfield, std::unique_ptr<facebook::velox::common::Filter>>> combine(
     facebook::velox::common::Subfield& subfield,
     std::unique_ptr<facebook::velox::common::Filter>& filter) {
@@ -30,6 +88,7 @@ std::optional<std::pair<facebook::velox::common::Subfield, std::unique_ptr<faceb
 
   return std::nullopt;
 }
+
 } // namespace
 
 std::optional<std::pair<facebook::velox::common::Subfield, std::unique_ptr<facebook::velox::common::Filter>>>
@@ -93,6 +152,20 @@ SparkExprToSubfieldFilterParser::leafCallToSubfieldFilter(
       }
       return std::make_pair(std::move(subfield), facebook::velox::exec::isNotNull());
     }
+  } else if (call.name() == "might_contain" && !negated) {
+    // might_contain(bloomFilter, value) — the column to filter is input[1].
+    if (call.inputs().size() >= 2) {
+      const auto* valueSide = call.inputs()[1].get();
+      if (toSubfield(valueSide, subfield)) {
+        auto bloomFilterValue = toConstant(call.inputs()[0], evaluator);
+        if (bloomFilterValue && !bloomFilterValue->isNullAt(0)) {
+          auto sv = bloomFilterValue->as<SimpleVector<StringView>>()->valueAt(0);
+          std::unique_ptr<common::Filter> filter =
+              std::make_unique<SparkMightContain>(sv.data(), false /*nullAllowed*/);
+          return combine(subfield, filter);
+        }
+      }
+    }
   }
   return std::nullopt;
 }

cpp/velox/tests/CMakeLists.txt

@@ -128,6 +128,8 @@ add_velox_test(
   VeloxToSubstraitTypeTest.cc)
 add_velox_test(spark_functions_test SOURCES SparkFunctionTest.cc
                FunctionTest.cc)
+add_velox_test(spark_expr_to_subfield_filter_parser_test SOURCES
+               SparkExprToSubfieldFilterParserTest.cc)
 add_velox_test(runtime_test SOURCES RuntimeTest.cc)
 add_velox_test(velox_memory_test SOURCES MemoryManagerTest.cc)
 add_velox_test(buffer_outputstream_test SOURCES BufferOutputStreamTest.cc)

cpp/velox/tests/SparkExprToSubfieldFilterParserTest.cc

@@ -0,0 +1,192 @@
+/*
+ * 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.
+ */
+
+#include <gtest/gtest.h>
+
+#include "operators/functions/SparkExprToSubfieldFilterParser.h"
+#include "velox/common/base/BloomFilter.h"
+#include "velox/core/QueryCtx.h"
+#include "velox/expression/Expr.h"
+#include "velox/vector/tests/utils/VectorTestBase.h"
+
+using namespace facebook::velox;
+using namespace facebook::velox::common;
+
+namespace gluten {
+namespace {
+
+class SparkExprToSubfieldFilterParserTest : public ::testing::Test,
+                                            public test::VectorTestBase {
+ protected:
+  static void SetUpTestCase() {
+    memory::MemoryManager::testingSetInstance(memory::MemoryManager::Options{});
+  }
+
+  /// Builds a serialized Velox BloomFilter containing the given int64 values.
+  std::vector<char> makeSerializedBloomFilter(
+      const std::vector<int64_t>& values) {
+    BloomFilter<> bf;
+    bf.reset(std::max<int32_t>(100, values.size() * 4));
+    for (auto v : values) {
+      bf.insert(folly::hasher<int64_t>()(v));
+    }
+    std::vector<char> data(bf.serializedSize());
+    bf.serialize(data.data());
+    return data;
+  }
+
+  /// Creates a ConstantTypedExpr wrapping serialized bloom filter bytes.
+  core::TypedExprPtr makeVarbinaryConstant(const std::vector<char>& data) {
+    auto vector = makeFlatVector<StringView>(
+        std::vector<StringView>{StringView(data.data(), data.size())},
+        VARBINARY());
+    return std::make_shared<const core::ConstantTypedExpr>(vector);
+  }
+
+  /// Creates a null VARBINARY constant expression.
+  core::TypedExprPtr makeNullVarbinaryConstant() {
+    auto vector = BaseVector::createNullConstant(VARBINARY(), 1, pool());
+    return std::make_shared<const core::ConstantTypedExpr>(vector);
+  }
+
+  /// Constructs a might_contain(bloomFilter, value) CallTypedExpr.
+  core::CallTypedExprPtr makeMightContainCall(
+      const core::TypedExprPtr& bloomFilterExpr,
+      const core::TypedExprPtr& valueExpr) {
+    return std::make_shared<core::CallTypedExpr>(
+        BOOLEAN(), "might_contain", bloomFilterExpr, valueExpr);
+  }
+
+  /// Calls leafCallToSubfieldFilter on the parser. Returns (subfield, nullptr)
+  /// when the parser cannot translate the expression.
+  std::pair<Subfield, std::unique_ptr<Filter>> parse(
+      const core::CallTypedExprPtr& call,
+      bool negated = false) {
+    if (auto result =
+            parser_.leafCallToSubfieldFilter(*call, &evaluator_, negated)) {
+      return std::move(result.value());
+    }
+    return std::make_pair(Subfield(), nullptr);
+  }
+
+ private:
+  std::shared_ptr<core::QueryCtx> queryCtx_{core::QueryCtx::create()};
+  exec::SimpleExpressionEvaluator evaluator_{queryCtx_.get(), pool()};
+  SparkExprToSubfieldFilterParser parser_;
+};
+
+TEST_F(SparkExprToSubfieldFilterParserTest, mightContainBasic) {
+  std::vector<int64_t> inserted = {42, 100, 200, 300};
+  auto serialized = makeSerializedBloomFilter(inserted);
+
+  auto bloomExpr = makeVarbinaryConstant(serialized);
+  auto columnExpr =
+      std::make_shared<core::FieldAccessTypedExpr>(BIGINT(), "a");
+  auto call = makeMightContainCall(bloomExpr, columnExpr);
+
+  auto [subfield, filter] = parse(call);
+
+  ASSERT_TRUE(filter);
+
+  // Verify the subfield points to column "a".
+  ASSERT_EQ(subfield.path().size(), 1);
+  EXPECT_EQ(*subfield.path()[0], Subfield::NestedField("a"));
+
+  // All inserted values must pass the bloom filter.
+  for (auto v : inserted) {
+    EXPECT_TRUE(filter->testInt64(v)) << "Value " << v << " should pass";
+  }
+
+  // Most non-inserted values should be rejected.
+  int falsePositives = 0;
+  for (int64_t v = 1000; v < 2000; v++) {
+    if (filter->testInt64(v)) {
+      ++falsePositives;
+    }
+  }
+  EXPECT_LT(falsePositives, 100) << "Too many false positives";
+}
+
+TEST_F(SparkExprToSubfieldFilterParserTest, mightContainNullBloomFilter) {
+  auto nullExpr = makeNullVarbinaryConstant();
+  auto columnExpr =
+      std::make_shared<core::FieldAccessTypedExpr>(BIGINT(), "a");
+  auto call = makeMightContainCall(nullExpr, columnExpr);
+
+  auto [subfield, filter] = parse(call);
+  EXPECT_FALSE(filter);
+}
+
+TEST_F(SparkExprToSubfieldFilterParserTest, mightContainNegated) {
+  auto serialized = makeSerializedBloomFilter({42});
+  auto bloomExpr = makeVarbinaryConstant(serialized);
+  auto columnExpr =
+      std::make_shared<core::FieldAccessTypedExpr>(BIGINT(), "a");
+  auto call = makeMightContainCall(bloomExpr, columnExpr);
+
+  auto [subfield, filter] = parse(call, /*negated=*/true);
+  EXPECT_FALSE(filter);
+}
+
+TEST_F(SparkExprToSubfieldFilterParserTest, mightContainNonColumnValue) {
+  auto serialized = makeSerializedBloomFilter({42});
+  auto bloomExpr = makeVarbinaryConstant(serialized);
+  // Use a constant (not a column reference) as the value argument.
+  auto constValue = makeVarbinaryConstant(serialized); // type doesn't matter
+  auto call = makeMightContainCall(bloomExpr, constValue);
+
+  auto [subfield, filter] = parse(call);
+  EXPECT_FALSE(filter);
+}
+
+TEST_F(SparkExprToSubfieldFilterParserTest, mightContainInt64Range) {
+  auto serialized = makeSerializedBloomFilter({42});
+  auto bloomExpr = makeVarbinaryConstant(serialized);
+  auto columnExpr =
+      std::make_shared<core::FieldAccessTypedExpr>(BIGINT(), "a");
+  auto call = makeMightContainCall(bloomExpr, columnExpr);
+
+  auto [subfield, filter] = parse(call);
+  ASSERT_TRUE(filter);
+
+  // Bloom filters cannot efficiently prune integer ranges.
+  EXPECT_TRUE(filter->testInt64Range(0, 1000, false));
+  EXPECT_TRUE(filter->testInt64Range(0, 1000, true));
+}
+
+TEST_F(SparkExprToSubfieldFilterParserTest, mightContainClone) {
+  std::vector<int64_t> inserted = {42, 100};
+  auto serialized = makeSerializedBloomFilter(inserted);
+  auto bloomExpr = makeVarbinaryConstant(serialized);
+  auto columnExpr =
+      std::make_shared<core::FieldAccessTypedExpr>(BIGINT(), "a");
+  auto call = makeMightContainCall(bloomExpr, columnExpr);
+
+  auto [subfield, filter] = parse(call);
+  ASSERT_TRUE(filter);
+
+  auto cloned = filter->clone(std::nullopt);
+  ASSERT_TRUE(cloned);
+  EXPECT_TRUE(filter->testingEquals(*cloned));
+
+  for (auto v : inserted) {
+    EXPECT_TRUE(cloned->testInt64(v));
+  }
+}
+
+} // namespace
+} // namespace gluten