diff --git a/backends-velox/src/main/java/org/apache/gluten/metrics/Metrics.java b/backends-velox/src/main/java/org/apache/gluten/metrics/Metrics.java index 12f38d84e5e..97d2a81b8e5 100644 --- a/backends-velox/src/main/java/org/apache/gluten/metrics/Metrics.java +++ b/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], diff --git a/backends-velox/src/main/java/org/apache/gluten/metrics/OperatorMetrics.java b/backends-velox/src/main/java/org/apache/gluten/metrics/OperatorMetrics.java index 18feed0b9e1..dcf1f136e82 100644 --- a/backends-velox/src/main/java/org/apache/gluten/metrics/OperatorMetrics.java +++ b/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; diff --git a/backends-velox/src/main/scala/org/apache/gluten/backendsapi/velox/VeloxMetricsApi.scala b/backends-velox/src/main/scala/org/apache/gluten/backendsapi/velox/VeloxMetricsApi.scala index 272574fe79b..a44c8662fe9 100644 --- a/backends-velox/src/main/scala/org/apache/gluten/backendsapi/velox/VeloxMetricsApi.scala +++ b/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"), diff --git a/backends-velox/src/main/scala/org/apache/gluten/metrics/BatchScanMetricsUpdater.scala b/backends-velox/src/main/scala/org/apache/gluten/metrics/BatchScanMetricsUpdater.scala index 45767a2c535..3b31c392237 100644 --- a/backends-velox/src/main/scala/org/apache/gluten/metrics/BatchScanMetricsUpdater.scala +++ b/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) diff --git a/backends-velox/src/main/scala/org/apache/gluten/metrics/FileSourceScanMetricsUpdater.scala b/backends-velox/src/main/scala/org/apache/gluten/metrics/FileSourceScanMetricsUpdater.scala index 4b62447aac6..380b52a0e55 100644 --- a/backends-velox/src/main/scala/org/apache/gluten/metrics/FileSourceScanMetricsUpdater.scala +++ b/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) diff --git a/backends-velox/src/main/scala/org/apache/gluten/metrics/HiveTableScanMetricsUpdater.scala b/backends-velox/src/main/scala/org/apache/gluten/metrics/HiveTableScanMetricsUpdater.scala index 7affd730aa0..5accb4db580 100644 --- a/backends-velox/src/main/scala/org/apache/gluten/metrics/HiveTableScanMetricsUpdater.scala +++ b/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) diff --git a/backends-velox/src/main/scala/org/apache/gluten/metrics/MetricsUtil.scala b/backends-velox/src/main/scala/org/apache/gluten/metrics/MetricsUtil.scala index 878eb2dcf1d..16ca7a691a2 100644 --- a/backends-velox/src/main/scala/org/apache/gluten/metrics/MetricsUtil.scala +++ b/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, diff --git a/cpp/core/jni/JniWrapper.cc b/cpp/core/jni/JniWrapper.cc index f69c57fb364..4d608e78191 100644 --- a/cpp/core/jni/JniWrapper.cc +++ b/cpp/core/jni/JniWrapper.cc @@ -313,7 +313,7 @@ jint JNI_OnLoad(JavaVM* vm, void* reserved) { env, metricsBuilderClass, "", - "([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], diff --git a/cpp/core/utils/Metrics.h b/cpp/core/utils/Metrics.h index 67c0b485c72..5c4bb3083c6 100644 --- a/cpp/core/utils/Metrics.h +++ b/cpp/core/utils/Metrics.h @@ -77,6 +77,8 @@ struct Metrics { kProcessedSplits, kSkippedStrides, kProcessedStrides, + kSkippedPages, + kProcessedPages, kRemainingFilterTime, kIoWaitTime, kStorageReadBytes, diff --git a/cpp/velox/compute/WholeStageResultIterator.cc b/cpp/velox/compute/WholeStageResultIterator.cc index 29213c2e0e2..7942aa96f9e 100644 --- a/cpp/velox/compute/WholeStageResultIterator.cc +++ b/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); diff --git a/cpp/velox/operators/functions/SparkExprToSubfieldFilterParser.cc b/cpp/velox/operators/functions/SparkExprToSubfieldFilterParser.cc index 20baaa413eb..df0dec32328 100644 --- a/cpp/velox/operators/functions/SparkExprToSubfieldFilterParser.cc +++ b/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{}); + 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>()->valueAt(0); + view_ = std::make_unique(sv.data()); + } + + bool testInt64(int64_t value) const override { + return view_->mayContain(folly::hasher()( + facebook::velox::functions::sparksql::XxHash64::hashInt64(value, seed_))); + } + + bool testDouble(double value) const override { + return view_->mayContain(folly::hasher()( + facebook::velox::functions::sparksql::XxHash64::hashDouble(value, seed_))); + } + + bool testFloat(float value) const override { + return view_->mayContain(folly::hasher()( + facebook::velox::functions::sparksql::XxHash64::hashFloat(value, seed_))); + } + + bool testBytes(const char* data, int32_t len) const override { + return view_->mayContain(folly::hasher()( + facebook::velox::functions::sparksql::XxHash64::hashBytes(StringView(data, len), seed_))); + } + + bool testTimestamp(const Timestamp& value) const override { + return view_->mayContain(folly::hasher()( + facebook::velox::functions::sparksql::XxHash64::hashTimestamp(value, seed_))); + } + + bool testInt128(const int128_t& value) const override { + return view_->mayContain(folly::hasher()( + 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 clone(std::optional nullAllowed) const override { + return std::make_unique( + constantVector_, nullAllowed.value_or(nullAllowed_), seed_); + } + + bool testingEquals(const Filter& other) const override { + return dynamic_cast(&other) != nullptr; + } + + folly::dynamic serialize() const override { + VELOX_UNSUPPORTED("Serialization is not supported for SparkMightContain"); + } + + private: + VectorPtr constantVector_; + std::unique_ptr view_; + int64_t seed_; +}; + std::optional>> combine( facebook::velox::common::Subfield& subfield, std::unique_ptr& filter) { @@ -30,6 +123,7 @@ std::optional>> @@ -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(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>()->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 filter = + std::make_unique(bloomFilterValue, false /*nullAllowed*/, seed); + return combine(subfield, filter); + } + } + LOG(WARNING) << "might_contain could not be converted to a subfield filter"; } return std::nullopt; } diff --git a/ep/build-velox/src/get-velox.sh b/ep/build-velox/src/get-velox.sh index 0ba44acfaa9..4828dcf9369 100755 --- a/ep/build-velox/src/get-velox.sh +++ b/ep/build-velox/src/get-velox.sh @@ -166,6 +166,28 @@ function apply_compilation_fixes { git add ${VELOX_HOME}/CMake/resolve_dependency_modules/arrow/modify_arrow.patch # to avoid the file from being deleted by git clean -dffx :/ } +function apply_bloomfilter_final_patch { + echo "Applying bloom filter final-removal patch..." + pushd $VELOX_HOME + git apply --check ${CURRENT_DIR}/remove-bloomfilter-final.patch && \ + git apply ${CURRENT_DIR}/remove-bloomfilter-final.patch || { + echo "Failed to apply bloom filter final-removal patch" + exit 1 + } + popd +} + +function apply_skipped_processed_pages_patch { + echo "Applying skipped/processed pages patch..." + pushd $VELOX_HOME + git apply --check ${CURRENT_DIR}/skipped-processed-pages.patch && \ + git apply ${CURRENT_DIR}/skipped-processed-pages.patch || { + echo "Failed to apply skipped/processed pages patch" + exit 1 + } + popd +} + function setup_linux { local LINUX_DISTRIBUTION=$(. /etc/os-release && echo ${ID}) local LINUX_VERSION_ID=$(. /etc/os-release && echo ${VERSION_ID}) @@ -248,4 +270,8 @@ apply_provided_velox_patch apply_compilation_fixes +apply_bloomfilter_final_patch + +apply_skipped_processed_pages_patch + echo "Finished getting Velox code" diff --git a/ep/build-velox/src/remove-bloomfilter-final.patch b/ep/build-velox/src/remove-bloomfilter-final.patch new file mode 100644 index 00000000000..cdc70d621be --- /dev/null +++ b/ep/build-velox/src/remove-bloomfilter-final.patch @@ -0,0 +1,31 @@ +diff --git a/velox/type/Filter.h b/velox/type/Filter.h +index 82ecd8eb2..cc8e6b31b 100644 +--- a/velox/type/Filter.h ++++ b/velox/type/Filter.h +@@ -1290,7 +1290,7 @@ class BigintValuesUsingBitmask final : public Filter { + const int64_t max_; + }; + +-class BigintValuesUsingBloomFilter final : public Filter { ++class BigintValuesUsingBloomFilter : public Filter { + public: + static int64_t numBlocks(int64_t capacity) { + return SplitBlockBloomFilter::numBlocks(capacity, 0.01); +@@ -1301,7 +1301,7 @@ class BigintValuesUsingBloomFilter final : public Filter { + blocks_(numBlocks(capacity)), + filter_(blocks_) {} + +- bool testInt64(int64_t value) const final { ++ bool testInt64(int64_t value) const override { + return filter_.mayContain(hash(value)); + } + +@@ -1318,7 +1318,7 @@ class BigintValuesUsingBloomFilter final : public Filter { + } + + bool testInt64Range(int64_t /*min*/, int64_t /*max*/, bool /*hasNull*/) +- const final { ++ const override { + return true; + } + diff --git a/ep/build-velox/src/skipped-processed-pages.patch b/ep/build-velox/src/skipped-processed-pages.patch new file mode 100644 index 00000000000..5aad32260ec --- /dev/null +++ b/ep/build-velox/src/skipped-processed-pages.patch @@ -0,0 +1,182 @@ +Index: velox/dwio/common/Statistics.h +IDEA additional info: +Subsystem: com.intellij.openapi.diff.impl.patch.CharsetEP +<+>UTF-8 +=================================================================== +diff --git a/velox/dwio/common/Statistics.h b/velox/dwio/common/Statistics.h +--- a/velox/dwio/common/Statistics.h (revision 3c20437c8aa13004d733bcf497347a7d8cae0c37) ++++ b/velox/dwio/common/Statistics.h (revision b12f0943c3cf7c4b72e1bcef759c1b85fbb7ef04) +@@ -682,6 +682,13 @@ + // Total time spent in loading pages, in nanoseconds. + io::IoCounter pageLoadTimeNs; + ++ // Number of pages skipped (only page header read, compressed data never ++ // fetched or decompressed). ++ int64_t skippedPages{0}; ++ ++ // Number of pages processed (compressed data fetched and decompressed). ++ int64_t processedPages{0}; ++ + // Per-column decoding statistics. Only populated when decoding stats + // collection is enabled. + std::optional decodingStatsSet; +@@ -702,6 +709,8 @@ + void mergeFrom(const ColumnReaderStatistics& other) { + flattenStringDictionaryValues += other.flattenStringDictionaryValues; + pageLoadTimeNs.merge(other.pageLoadTimeNs); ++ skippedPages += other.skippedPages; ++ processedPages += other.processedPages; + if (other.decodingStatsSet) { + if (!decodingStatsSet) { + decodingStatsSet.emplace(); +@@ -728,6 +737,12 @@ + pageLoadTimeNs.max(), + RuntimeCounter::Unit::kNanos)); + } ++ if (skippedPages > 0) { ++ result.emplace("skippedPages", RuntimeMetric(skippedPages)); ++ } ++ if (processedPages > 0) { ++ result.emplace("processedPages", RuntimeMetric(processedPages)); ++ } + if (decodingStatsSet) { + decodingStatsSet->toRuntimeMetrics(result); + } +Index: velox/dwio/parquet/reader/PageReader.cpp +IDEA additional info: +Subsystem: com.intellij.openapi.diff.impl.patch.CharsetEP +<+>UTF-8 +=================================================================== +diff --git a/velox/dwio/parquet/reader/PageReader.cpp b/velox/dwio/parquet/reader/PageReader.cpp +--- a/velox/dwio/parquet/reader/PageReader.cpp (revision 3c20437c8aa13004d733bcf497347a7d8cae0c37) ++++ b/velox/dwio/parquet/reader/PageReader.cpp (revision 0b73156d3a21fd8bc1e5506170311d227ad21963) +@@ -73,9 +73,11 @@ + inputStream_.get(), + bufferStart_, + bufferEnd_); ++ ++stats_.skippedPages; + continue; + } + prepareDictionary(pageHeader); ++ ++stats_.processedPages; + continue; + default: + break; // ignore INDEX page type and any other custom extensions +@@ -271,9 +273,11 @@ + bufferStart_, + bufferEnd_); + ++ ++stats_.skippedPages; + return; + } + pageData_ = readBytes(*pageHeader.compressed_page_size(), pageBuffer_); ++ ++stats_.processedPages; + pageData_ = decompressData( + pageData_, + *pageHeader.compressed_page_size(), +@@ -350,6 +354,7 @@ + inputStream_.get(), + bufferStart_, + bufferEnd_); ++ ++stats_.skippedPages; + return; + } + +@@ -367,6 +372,7 @@ + defineLength, + bytes); + pageData_ = readBytes(bytes, pageBuffer_); ++ ++stats_.processedPages; + + if (repeatLength) { + repeatDecoder_ = std::make_unique( +Index: velox/dwio/parquet/reader/ParquetReader.cpp +IDEA additional info: +Subsystem: com.intellij.openapi.diff.impl.patch.CharsetEP +<+>UTF-8 +=================================================================== +diff --git a/velox/dwio/parquet/reader/ParquetReader.cpp b/velox/dwio/parquet/reader/ParquetReader.cpp +--- a/velox/dwio/parquet/reader/ParquetReader.cpp (revision 3c20437c8aa13004d733bcf497347a7d8cae0c37) ++++ b/velox/dwio/parquet/reader/ParquetReader.cpp (revision b12f0943c3cf7c4b72e1bcef759c1b85fbb7ef04) +@@ -1596,8 +1596,7 @@ + stats.skippedStrides += skippedStrides_; + stats.processedStrides += rowGroupIds_.size(); + stats.parquetFooterEstimatedBytes += readerBase_->initialThriftSize(); +- stats.columnReaderStats.pageLoadTimeNs.merge( +- columnReaderStats_.pageLoadTimeNs); ++ stats.columnReaderStats.mergeFrom(columnReaderStats_); + } + + void resetFilterCaches() { +Index: velox/dwio/parquet/tests/reader/ParquetReaderTest.cpp +IDEA additional info: +Subsystem: com.intellij.openapi.diff.impl.patch.CharsetEP +<+>UTF-8 +=================================================================== +diff --git a/velox/dwio/parquet/tests/reader/ParquetReaderTest.cpp b/velox/dwio/parquet/tests/reader/ParquetReaderTest.cpp +--- a/velox/dwio/parquet/tests/reader/ParquetReaderTest.cpp (revision 3c20437c8aa13004d733bcf497347a7d8cae0c37) ++++ b/velox/dwio/parquet/tests/reader/ParquetReaderTest.cpp (revision 3d85c5cebfba7217c5738dea51993d61dfad72dc) +@@ -22,6 +22,7 @@ + #include "velox/dwio/parquet/tests/ParquetTestBase.h" + #include "velox/dwio/parquet/thrift/ParquetThrift.h" + #include "velox/expression/ExprToSubfieldFilter.h" ++#include "velox/type/Filter.h" + #include "velox/vector/tests/utils/VectorMaker.h" + + using namespace facebook::velox; +@@ -2035,3 +2036,55 @@ + + EXPECT_EQ(leafPool_->usedBytes(), initialUsage); + } ++ ++TEST_F(ParquetReaderTest, pageSkipStats) { ++ constexpr int kNumWrittenRows = 100'000; ++ constexpr int kNumResultRows = 200; ++ ++ auto rowType = ROW({"a", "b"}, {BIGINT(), BIGINT()}); ++ ++ auto a = makeFlatVector(kNumWrittenRows, [](auto i) { return i; }); ++ auto b = makeFlatVector( ++ kNumWrittenRows, [](auto i) { return (i * 7919) % kNumWrittenRows; }); ++ auto data = makeRowVector({"a", "b"}, {a, b}); ++ ++ auto run = [&](bool useDataPageV2) { ++ ParquetWriterOptions writerOptions; ++ writerOptions.dataPageSize = 256; ++ writerOptions.enableDictionary = false; ++ writerOptions.useParquetDataPageV2 = useDataPageV2; ++ ++ auto* sink = write(data, writerOptions); ++ auto reader = createReaderInMemory(*sink); ++ ++ FilterMap filters; ++ filters.emplace( ++ "a", ++ std::make_unique( ++ kNumWrittenRows - kNumResultRows, kNumWrittenRows - 1, false)); ++ ++ auto scanSpec = makeScanSpec(rowType); ++ for (auto& [col, f] : filters) { ++ scanSpec->getOrCreateChild(Subfield(col))->setFilter(std::move(f)); ++ } ++ auto rowReaderOpts = makeRowReaderOpts(rowType); ++ rowReaderOpts.setScanSpec(scanSpec); ++ auto rowReader = reader->createRowReader(rowReaderOpts); ++ ++ uint64_t totalRows = 0; ++ VectorPtr result = BaseVector::create(rowType, 0, leafPool_.get()); ++ while (rowReader->next(1000, result)) { ++ totalRows += result->size(); ++ LazyVector::ensureLoadedRows(result, SelectivityVector(result->size())); ++ } ++ EXPECT_EQ(totalRows, kNumResultRows); ++ RuntimeStatistics stats; ++ rowReader->updateRuntimeStats(stats); ++ EXPECT_GT(stats.columnReaderStats.processedPages, 0); ++ EXPECT_GT(stats.columnReaderStats.skippedPages, 0); ++ }; ++ ++ for (const auto useDataPageV2 : {false, true}) { ++ run(useDataPageV2); ++ } ++} diff --git a/gluten-ut/spark35/src/test/scala/org/apache/spark/sql/GlutenBloomFilterAggregateQuerySuite.scala b/gluten-ut/spark35/src/test/scala/org/apache/spark/sql/GlutenBloomFilterAggregateQuerySuite.scala index 3eb59d8feca..9b7c585aac9 100644 --- a/gluten-ut/spark35/src/test/scala/org/apache/spark/sql/GlutenBloomFilterAggregateQuerySuite.scala +++ b/gluten-ut/spark35/src/test/scala/org/apache/spark/sql/GlutenBloomFilterAggregateQuerySuite.scala @@ -18,7 +18,7 @@ package org.apache.spark.sql import org.apache.gluten.backendsapi.BackendsApiManager import org.apache.gluten.config.GlutenConfig -import org.apache.gluten.execution.HashAggregateExecBaseTransformer +import org.apache.gluten.execution.{FileSourceScanExecTransformer, HashAggregateExecBaseTransformer} import org.apache.spark.SparkConf import org.apache.spark.sql.execution.adaptive.AdaptiveSparkPlanHelper @@ -112,6 +112,44 @@ class GlutenBloomFilterAggregateQuerySuite } } + testGluten("Test might_contain constant bloom filter pushdown to scan") { + val table = "bloom_filter_test" + withTable(table) { + (1L to 200L).toDF("col").write.format("parquet").saveAsTable(table) + + // Build a bloom filter over xxhash64(col) so it matches the probe side. + val hexBf = spark.sql( + s"SELECT hex(bloom_filter_agg(xxhash64(col), " + + s"cast(200 as long), cast($veloxBloomFilterMaxNumBits as long))) " + + s"FROM $table") + .collect()(0).getString(0) + + val sqlString = + s"SELECT * FROM $table WHERE might_contain(X'$hexBf', xxhash64(col))" + + val df = spark.sql(sqlString) + val result = df.collect() + // All 200 rows should match since the bloom filter covers them. + assert(result.length == 200) + + // Verify the filter was pushed to the scan. + val scans = collect(df.queryExecution.executedPlan) { + case s: FileSourceScanExecTransformer => s + } + assert(scans.nonEmpty) + scans.foreach { scan => + val metrics = scan.metrics + assert(metrics.contains("rawInputRows")) + assert(metrics.contains("numOutputRows")) + assert(metrics.contains("remainingFilterTime")) + assert(metrics("rawInputRows").value > 0) + assert(metrics("numOutputRows").value > 0) + // The remaining filter time should be 0 since the filter was pushed down to the scan as subfield filter. + assert(metrics("remainingFilterTime").value == 0) + } + } + } + testGluten("Test bloom_filter_agg agg fallback") { val table = "bloom_filter_test" val numEstimatedItems = 5000000L