[GLUTEN-11708][VL] Translate might_contain to a subfield filter

backends-velox/src/main/java/org/apache/gluten/metrics/Metrics.java

@@ -49,6 +49,8 @@ public class Metrics implements IMetrics {
   public long[] processedSplits;
   public long[] skippedStrides;
   public long[] processedStrides;
+  public long[] skippedPages;
+  public long[] processedPages;
   public long[] remainingFilterTime;
   public long[] ioWaitTime;
   public long[] storageReadBytes;
@@ -103,6 +105,8 @@ public Metrics(
       long[] processedSplits,
       long[] skippedStrides,
       long[] processedStrides,
+      long[] skippedPages,
+      long[] processedPages,
       long[] remainingFilterTime,
       long[] ioWaitTime,
       long[] storageReadBytes,
@@ -149,6 +153,8 @@ public Metrics(
     this.processedSplits = processedSplits;
     this.skippedStrides = skippedStrides;
     this.processedStrides = processedStrides;
+    this.skippedPages = skippedPages;
+    this.processedPages = processedPages;
     this.remainingFilterTime = remainingFilterTime;
     this.ioWaitTime = ioWaitTime;
     this.storageReadBytes = storageReadBytes;
@@ -203,6 +209,8 @@ public OperatorMetrics getOperatorMetrics(int index) {
         processedSplits[index],
         skippedStrides[index],
         processedStrides[index],
+        skippedPages[index],
+        processedPages[index],
         remainingFilterTime[index],
         ioWaitTime[index],
         storageReadBytes[index],

backends-velox/src/main/java/org/apache/gluten/metrics/OperatorMetrics.java

@@ -47,6 +47,8 @@ public class OperatorMetrics implements IOperatorMetrics {
   public long processedSplits;
   public long skippedStrides;
   public long processedStrides;
+  public long skippedPages;
+  public long processedPages;
   public long remainingFilterTime;
   public long ioWaitTime;
   public long storageReadBytes;
@@ -96,6 +98,8 @@ public OperatorMetrics(
       long processedSplits,
       long skippedStrides,
       long processedStrides,
+      long skippedPages,
+      long processedPages,
       long remainingFilterTime,
       long ioWaitTime,
       long storageReadBytes,
@@ -140,6 +144,8 @@ public OperatorMetrics(
     this.processedSplits = processedSplits;
     this.skippedStrides = skippedStrides;
     this.processedStrides = processedStrides;
+    this.skippedPages = skippedPages;
+    this.processedPages = processedPages;
     this.remainingFilterTime = remainingFilterTime;
     this.ioWaitTime = ioWaitTime;
     this.storageReadBytes = storageReadBytes;

backends-velox/src/main/scala/org/apache/gluten/backendsapi/velox/VeloxMetricsApi.scala

@@ -136,6 +136,8 @@ class VeloxMetricsApi extends MetricsApi with Logging {
         "data source read time"),
       "skippedStrides" -> SQLMetrics.createMetric(sparkContext, "number of skipped row groups"),
       "processedStrides" -> SQLMetrics.createMetric(sparkContext, "number of processed row groups"),
+      "skippedPages" -> SQLMetrics.createMetric(sparkContext, "number of skipped pages"),
+      "processedPages" -> SQLMetrics.createMetric(sparkContext, "number of processed pages"),
       "remainingFilterTime" -> SQLMetrics.createNanoTimingMetric(
         sparkContext,
         "remaining filter time"),
@@ -194,6 +196,8 @@ class VeloxMetricsApi extends MetricsApi with Logging {
         "data source read time"),
       "skippedStrides" -> SQLMetrics.createMetric(sparkContext, "number of skipped row groups"),
       "processedStrides" -> SQLMetrics.createMetric(sparkContext, "number of processed row groups"),
+      "skippedPages" -> SQLMetrics.createMetric(sparkContext, "number of skipped pages"),
+      "processedPages" -> SQLMetrics.createMetric(sparkContext, "number of processed pages"),
       "remainingFilterTime" -> SQLMetrics.createNanoTimingMetric(
         sparkContext,
         "remaining filter time"),
@@ -252,6 +256,8 @@ class VeloxMetricsApi extends MetricsApi with Logging {
         "data source read time"),
       "skippedStrides" -> SQLMetrics.createMetric(sparkContext, "number of skipped row groups"),
       "processedStrides" -> SQLMetrics.createMetric(sparkContext, "number of processed row groups"),
+      "skippedPages" -> SQLMetrics.createMetric(sparkContext, "number of skipped pages"),
+      "processedPages" -> SQLMetrics.createMetric(sparkContext, "number of processed pages"),
       "remainingFilterTime" -> SQLMetrics.createNanoTimingMetric(
         sparkContext,
         "remaining filter time"),

backends-velox/src/main/scala/org/apache/gluten/metrics/BatchScanMetricsUpdater.scala

@@ -46,6 +46,8 @@ class BatchScanMetricsUpdater(@transient val metrics: Map[String, SQLMetric])
   private val processedSplits: Option[SQLMetric] = metric("processedSplits")
   private val skippedStrides: Option[SQLMetric] = metric("skippedStrides")
   private val processedStrides: Option[SQLMetric] = metric("processedStrides")
+  private val skippedPages: Option[SQLMetric] = metric("skippedPages")
+  private val processedPages: Option[SQLMetric] = metric("processedPages")
   private val remainingFilterTime: Option[SQLMetric] = metric("remainingFilterTime")
   private val ioWaitTime: Option[SQLMetric] = metric("ioWaitTime")
   private val storageReadBytes: Option[SQLMetric] = metric("storageReadBytes")
@@ -84,6 +86,8 @@ class BatchScanMetricsUpdater(@transient val metrics: Map[String, SQLMetric])
       ScanMetricsUtil.inc(processedSplits, operatorMetrics.processedSplits)
       ScanMetricsUtil.inc(skippedStrides, operatorMetrics.skippedStrides)
       ScanMetricsUtil.inc(processedStrides, operatorMetrics.processedStrides)
+      ScanMetricsUtil.inc(skippedPages, operatorMetrics.skippedPages)
+      ScanMetricsUtil.inc(processedPages, operatorMetrics.processedPages)
       ScanMetricsUtil.inc(remainingFilterTime, operatorMetrics.remainingFilterTime)
       ScanMetricsUtil.inc(ioWaitTime, operatorMetrics.ioWaitTime)
       ScanMetricsUtil.inc(storageReadBytes, operatorMetrics.storageReadBytes)

backends-velox/src/main/scala/org/apache/gluten/metrics/FileSourceScanMetricsUpdater.scala

@@ -44,6 +44,8 @@ class FileSourceScanMetricsUpdater(@transient val metrics: Map[String, SQLMetric
   private val processedSplits: Option[SQLMetric] = metric("processedSplits")
   private val skippedStrides: Option[SQLMetric] = metric("skippedStrides")
   private val processedStrides: Option[SQLMetric] = metric("processedStrides")
+  private val skippedPages: Option[SQLMetric] = metric("skippedPages")
+  private val processedPages: Option[SQLMetric] = metric("processedPages")
   private val remainingFilterTime: Option[SQLMetric] = metric("remainingFilterTime")
   private val ioWaitTime: Option[SQLMetric] = metric("ioWaitTime")
   private val storageReadBytes: Option[SQLMetric] = metric("storageReadBytes")
@@ -79,6 +81,8 @@ class FileSourceScanMetricsUpdater(@transient val metrics: Map[String, SQLMetric
       ScanMetricsUtil.inc(processedSplits, operatorMetrics.processedSplits)
       ScanMetricsUtil.inc(skippedStrides, operatorMetrics.skippedStrides)
       ScanMetricsUtil.inc(processedStrides, operatorMetrics.processedStrides)
+      ScanMetricsUtil.inc(skippedPages, operatorMetrics.skippedPages)
+      ScanMetricsUtil.inc(processedPages, operatorMetrics.processedPages)
       ScanMetricsUtil.inc(remainingFilterTime, operatorMetrics.remainingFilterTime)
       ScanMetricsUtil.inc(ioWaitTime, operatorMetrics.ioWaitTime)
       ScanMetricsUtil.inc(storageReadBytes, operatorMetrics.storageReadBytes)

backends-velox/src/main/scala/org/apache/gluten/metrics/HiveTableScanMetricsUpdater.scala

@@ -40,6 +40,8 @@ class HiveTableScanMetricsUpdater(@transient val metrics: Map[String, SQLMetric]
   private val processedSplits: Option[SQLMetric] = metric("processedSplits")
   private val skippedStrides: Option[SQLMetric] = metric("skippedStrides")
   private val processedStrides: Option[SQLMetric] = metric("processedStrides")
+  private val skippedPages: Option[SQLMetric] = metric("skippedPages")
+  private val processedPages: Option[SQLMetric] = metric("processedPages")
   private val remainingFilterTime: Option[SQLMetric] = metric("remainingFilterTime")
   private val ioWaitTime: Option[SQLMetric] = metric("ioWaitTime")
   private val storageReadBytes: Option[SQLMetric] = metric("storageReadBytes")
@@ -75,6 +77,8 @@ class HiveTableScanMetricsUpdater(@transient val metrics: Map[String, SQLMetric]
       ScanMetricsUtil.inc(processedSplits, operatorMetrics.processedSplits)
       ScanMetricsUtil.inc(skippedStrides, operatorMetrics.skippedStrides)
       ScanMetricsUtil.inc(processedStrides, operatorMetrics.processedStrides)
+      ScanMetricsUtil.inc(skippedPages, operatorMetrics.skippedPages)
+      ScanMetricsUtil.inc(processedPages, operatorMetrics.processedPages)
       ScanMetricsUtil.inc(remainingFilterTime, operatorMetrics.remainingFilterTime)
       ScanMetricsUtil.inc(ioWaitTime, operatorMetrics.ioWaitTime)
       ScanMetricsUtil.inc(storageReadBytes, operatorMetrics.storageReadBytes)

backends-velox/src/main/scala/org/apache/gluten/metrics/MetricsUtil.scala

@@ -129,6 +129,8 @@ object MetricsUtil extends Logging {
     var skippedSplits: Long = 0
     var processedSplits: Long = 0
     var skippedStrides: Long = 0
+    var skippedPages: Long = 0
+    var processedPages: Long = 0
     var processedStrides: Long = 0
     var remainingFilterTime: Long = 0
     var ioWaitTime: Long = 0
@@ -168,6 +170,8 @@ object MetricsUtil extends Logging {
       processedSplits += metrics.processedSplits
       skippedStrides += metrics.skippedStrides
       processedStrides += metrics.processedStrides
+      skippedPages += metrics.skippedPages
+      processedPages += metrics.processedPages
       remainingFilterTime += metrics.remainingFilterTime
       ioWaitTime += metrics.ioWaitTime
       storageReadBytes += metrics.storageReadBytes
@@ -213,6 +217,8 @@ object MetricsUtil extends Logging {
       processedSplits,
       skippedStrides,
       processedStrides,
+      skippedPages,
+      processedPages,
       remainingFilterTime,
       ioWaitTime,
       storageReadBytes,

cpp/core/jni/JniWrapper.cc

@@ -313,7 +313,7 @@ jint JNI_OnLoad(JavaVM* vm, void* reserved) {
       env,
       metricsBuilderClass,
       "<init>",
-      "([J[J[J[J[J[J[J[J[J[JJ[J[J[J[J[J[J[J[J[J[J[J[J[J[J[J[J[J[J[J[J[J[J[J[J[J[J[J[J[J[J[J[J[J[JLjava/lang/String;)V");
+      "([J[J[J[J[J[J[J[J[J[JJ[J[J[J[J[J[J[J[J[J[J[J[J[J[J[J[J[J[J[J[J[J[J[J[J[J[J[J[J[J[J[J[J[J[J[J[JLjava/lang/String;)V");
 
   nativeColumnarToRowInfoClass =
       createGlobalClassReferenceOrError(env, "Lorg/apache/gluten/vectorized/NativeColumnarToRowInfo;");
@@ -668,6 +668,8 @@ JNIEXPORT jobject JNICALL Java_org_apache_gluten_metrics_IteratorMetricsJniWrapp
       longArray[Metrics::kProcessedSplits],
       longArray[Metrics::kSkippedStrides],
       longArray[Metrics::kProcessedStrides],
+      longArray[Metrics::kSkippedPages],
+      longArray[Metrics::kProcessedPages],
       longArray[Metrics::kRemainingFilterTime],
       longArray[Metrics::kIoWaitTime],
       longArray[Metrics::kStorageReadBytes],

cpp/core/utils/Metrics.h

@@ -77,6 +77,8 @@ struct Metrics {
     kProcessedSplits,
     kSkippedStrides,
     kProcessedStrides,
+    kSkippedPages,
+    kProcessedPages,
     kRemainingFilterTime,
     kIoWaitTime,
     kStorageReadBytes,

cpp/velox/compute/WholeStageResultIterator.cc

@@ -61,6 +61,8 @@ const std::string kSkippedSplits = "skippedSplits";
 const std::string kProcessedSplits = "processedSplits";
 const std::string kSkippedStrides = "skippedStrides";
 const std::string kProcessedStrides = "processedStrides";
+const std::string kSkippedPages = "skippedPages";
+const std::string kProcessedPages = "processedPages";
 const std::string kRemainingFilterTime = "totalRemainingFilterWallNanos";
 const std::string kIoWaitTime = "ioWaitWallNanos";
 const std::string kStorageReadBytes = "storageReadBytes";
@@ -569,6 +571,10 @@ void WholeStageResultIterator::collectMetrics() {
       metrics_->get(Metrics::kSkippedStrides)[metricIndex] = runtimeMetric("sum", second->customStats, kSkippedStrides);
       metrics_->get(Metrics::kProcessedStrides)[metricIndex] =
           runtimeMetric("sum", second->customStats, kProcessedStrides);
+      metrics_->get(Metrics::kSkippedPages)[metricIndex] =
+          runtimeMetric("sum", second->customStats, kSkippedPages);
+      metrics_->get(Metrics::kProcessedPages)[metricIndex] =
+          runtimeMetric("sum", second->customStats, kProcessedPages);
       metrics_->get(Metrics::kRemainingFilterTime)[metricIndex] =
           runtimeMetric("sum", second->customStats, kRemainingFilterTime);
       metrics_->get(Metrics::kIoWaitTime)[metricIndex] = runtimeMetric("sum", second->customStats, kIoWaitTime);

cpp/velox/operators/functions/SparkExprToSubfieldFilterParser.cc

@@ -16,11 +16,104 @@
  */
 #include "operators/functions/SparkExprToSubfieldFilterParser.h"
 
+#include "utils/Exception.h"
+#include "velox/common/base/BloomFilter.h"
+#include "velox/functions/sparksql/XxHash64.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.
+/// Values are hashed with Spark-compatible XXH64 using the seed extracted from
+/// xxhash64_with_seed, then re-hashed with folly hasher for bloom filter bucket
+/// selection, matching bloom_filter_agg's insertion path.
+class SparkMightContain final : public common::BigintValuesUsingBloomFilter {
+ public:
+  SparkMightContain(VectorPtr constantVector, bool nullAllowed, int64_t seed)
+      : common::BigintValuesUsingBloomFilter(0, nullAllowed),
+        constantVector_(std::move(constantVector)),
+        seed_(seed) {
+    auto sv = constantVector_->as<SimpleVector<StringView>>()->valueAt(0);
+    view_ = std::make_unique<BloomFilterView>(sv.data());
+  }
+
+  bool testInt64(int64_t value) const override {
+    return view_->mayContain(folly::hasher<int64_t>()(
+        facebook::velox::functions::sparksql::XxHash64::hashInt64(value, seed_)));
+  }
+
+  bool testDouble(double value) const override {
+    return view_->mayContain(folly::hasher<int64_t>()(
+        facebook::velox::functions::sparksql::XxHash64::hashDouble(value, seed_)));
+  }
+
+  bool testFloat(float value) const override {
+    return view_->mayContain(folly::hasher<int64_t>()(
+        facebook::velox::functions::sparksql::XxHash64::hashFloat(value, seed_)));
+  }
+
+  bool testBytes(const char* data, int32_t len) const override {
+    return view_->mayContain(folly::hasher<int64_t>()(
+        facebook::velox::functions::sparksql::XxHash64::hashBytes(StringView(data, len), seed_)));
+  }
+
+  bool testTimestamp(const Timestamp& value) const override {
+    return view_->mayContain(folly::hasher<int64_t>()(
+        facebook::velox::functions::sparksql::XxHash64::hashTimestamp(value, seed_)));
+  }
+
+  bool testInt128(const int128_t& value) const override {
+    return view_->mayContain(folly::hasher<int64_t>()(
+        facebook::velox::functions::sparksql::XxHash64::hashLongDecimal(value, seed_)));
+  }
+
+  bool testInt64Range(int64_t /*min*/, int64_t /*max*/, bool /*hasNull*/) const override {
+    return true;
+  }
+
+  std::unique_ptr<Filter> clone(std::optional<bool> nullAllowed) const override {
+    return std::make_unique<SparkMightContain>(
+        constantVector_, nullAllowed.value_or(nullAllowed_), seed_);
+  }
+
+  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:
+  VectorPtr constantVector_;
+  std::unique_ptr<BloomFilterView> view_;
+  int64_t seed_;
+};
+
 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 +123,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 +187,34 @@ SparkExprToSubfieldFilterParser::leafCallToSubfieldFilter(
       }
       return std::make_pair(std::move(subfield), facebook::velox::exec::isNotNull());
     }
+  } else if (call.name() == "might_contain" && !negated) {
+    // Matches: might_contain(bloomFilter, xxhash64_with_seed(seed, field)).
+    GLUTEN_CHECK(call.inputs().size() == 2,
+        "might_contain expects 2 arguments: bloomFilter and xxhash64_with_seed(seed, field)");
+    const auto* hashCall =
+        dynamic_cast<const core::CallTypedExpr*>(call.inputs()[1].get());
+    if (hashCall && hashCall->name() == "xxhash64_with_seed") {
+      GLUTEN_CHECK(hashCall->inputs().size() == 2, "xxhash64_with_seed expects 2 arguments");
+      auto seedValue = toConstant(hashCall->inputs()[0], evaluator);
+      if (!seedValue || seedValue->isNullAt(0)) {
+        LOG(WARNING) << "might_contain: seed value is null or not constant, "
+                     << "cannot push down to subfield filter";
+        return std::nullopt;
+      }
+      auto seed = seedValue->as<SimpleVector<int64_t>>()->valueAt(0);
+      if (!toSubfield(hashCall->inputs()[1].get(), subfield)) {
+        LOG(WARNING) << "might_contain: second argument to xxhash64_with_seed "
+                     << "is not a subfield, cannot push down to subfield filter";
+        return std::nullopt;
+      }
+      auto bloomFilterValue = toConstant(call.inputs()[0], evaluator);
+      if (bloomFilterValue && !bloomFilterValue->isNullAt(0)) {
+        std::unique_ptr<common::Filter> filter =
+            std::make_unique<SparkMightContain>(bloomFilterValue, false /*nullAllowed*/, seed);
+        return combine(subfield, filter);
+      }
+    }
+    LOG(WARNING) << "might_contain could not be converted to a subfield filter";
   }
   return std::nullopt;
 }