This is an automated email from the ASF dual-hosted git repository.
zhztheplayer pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/gluten.git
The following commit(s) were added to refs/heads/main by this push:
new 533fe85c0a [GLUTEN-11708][VL] Enable Bloom filter might_contain to
subfield filter translation (#12793)
533fe85c0a is described below
commit 533fe85c0a9fdf70226b18400ce382c28eb8bced
Author: Hongze Zhang <[email protected]>
AuthorDate: Mon Aug 17 19:25:09 2026 +0100
[GLUTEN-11708][VL] Enable Bloom filter might_contain to subfield filter
translation (#12793)
---
.../org/apache/gluten/config/VeloxConfig.scala | 8 ++
cpp/velox/compute/VeloxBackend.cc | 3 +-
cpp/velox/config/VeloxConfig.h | 3 +
.../functions/SparkExprToSubfieldFilterParser.cc | 109 +++++++++++++++++++++
.../functions/SparkExprToSubfieldFilterParser.h | 6 ++
docs/velox-configuration.md | 1 +
6 files changed, 129 insertions(+), 1 deletion(-)
diff --git
a/backends-velox/src/main/scala/org/apache/gluten/config/VeloxConfig.scala
b/backends-velox/src/main/scala/org/apache/gluten/config/VeloxConfig.scala
index d0e55c2238..802a47cadd 100644
--- a/backends-velox/src/main/scala/org/apache/gluten/config/VeloxConfig.scala
+++ b/backends-velox/src/main/scala/org/apache/gluten/config/VeloxConfig.scala
@@ -113,6 +113,8 @@ class VeloxConfig(conf: SQLConf) extends GlutenConfig(conf)
{
def hashProbeBloomFilterBypassMinPct: Int =
getConf(HASH_PROBE_BLOOM_FILTER_BYPASS_MIN_PCT)
+ def scanBloomFilterPushdownEnabled: Boolean =
getConf(SCAN_BLOOM_FILTER_PUSHDOWN_ENABLED)
+
def enableTimestampNtzValidation: Boolean =
getConf(ENABLE_TIMESTAMP_NTZ_VALIDATION)
def enableDriverSideBroadcastHashTableBuild: Boolean =
@@ -550,6 +552,12 @@ object VeloxConfig extends ConfigRegistry {
.checkValue(value => value >= 0 && value <= 100, "The percentage must be
in [0, 100]")
.createWithDefault(85)
+ val SCAN_BLOOM_FILTER_PUSHDOWN_ENABLED =
+
buildStaticConf("spark.gluten.sql.columnar.backend.velox.scan.bloomFilterPushdown.enabled")
+ .doc("Whether to push Bloom filters into Velox scans.")
+ .booleanConf
+ .createWithDefault(false)
+
val COLUMNAR_VELOX_FILE_HANDLE_CACHE_ENABLED =
buildStaticConf("spark.gluten.sql.columnar.backend.velox.fileHandleCacheEnabled")
.doc(
diff --git a/cpp/velox/compute/VeloxBackend.cc
b/cpp/velox/compute/VeloxBackend.cc
index 3b735601a1..074aec9df4 100644
--- a/cpp/velox/compute/VeloxBackend.cc
+++ b/cpp/velox/compute/VeloxBackend.cc
@@ -260,7 +260,8 @@ void VeloxBackend::init(
velox::parquet::registerParquetReaderFactory();
velox::parquet::registerParquetWriterFactory();
velox::orc::registerOrcReaderFactory();
-
velox::exec::ExprToSubfieldFilterParser::registerParser(std::make_unique<SparkExprToSubfieldFilterParser>());
+
velox::exec::ExprToSubfieldFilterParser::registerParser(std::make_unique<SparkExprToSubfieldFilterParser>(
+ backendConf_->get<bool>(kScanBloomFilterPushdownEnabled,
kScanBloomFilterPushdownEnabledDefault)));
velox::connector::hive::BufferedInputBuilder::registerBuilder(std::make_shared<GlutenBufferedInputBuilder>());
// Register Velox functions
diff --git a/cpp/velox/config/VeloxConfig.h b/cpp/velox/config/VeloxConfig.h
index 7dde032d43..c0ae4b3ff8 100644
--- a/cpp/velox/config/VeloxConfig.h
+++ b/cpp/velox/config/VeloxConfig.h
@@ -84,6 +84,9 @@ const std::string kSparkBloomFilterExpectedNumItems =
"spark.sql.optimizer.runti
const std::string kSparkBloomFilterNumBits =
"spark.sql.optimizer.runtime.bloomFilter.numBits";
const std::string kSparkBloomFilterMaxNumBits =
"spark.sql.optimizer.runtime.bloomFilter.maxNumBits";
const std::string kSparkBloomFilterMaxNumItems =
"spark.sql.optimizer.runtime.bloomFilter.maxNumItems";
+const std::string kScanBloomFilterPushdownEnabled =
+ "spark.gluten.sql.columnar.backend.velox.scan.bloomFilterPushdown.enabled";
+const bool kScanBloomFilterPushdownEnabledDefault = false;
const std::string kVeloxSplitPreloadPerDriver =
"spark.gluten.sql.columnar.backend.velox.SplitPreloadPerDriver";
const std::string kHashProbeDynamicFilterPushdownEnabled =
diff --git a/cpp/velox/operators/functions/SparkExprToSubfieldFilterParser.cc
b/cpp/velox/operators/functions/SparkExprToSubfieldFilterParser.cc
index 20baaa413e..31d44211cd 100644
--- a/cpp/velox/operators/functions/SparkExprToSubfieldFilterParser.cc
+++ b/cpp/velox/operators/functions/SparkExprToSubfieldFilterParser.cc
@@ -16,11 +16,83 @@
*/
#include "operators/functions/SparkExprToSubfieldFilterParser.h"
+#include "utils/Exception.h"
+#include "velox/common/base/BloomFilter.h"
+#include "velox/expression/Expr.h"
+#include "velox/functions/sparksql/XxHash64.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& error) {
+ VLOG(1) << "Failed to evaluate constant expression for scan filter
pushdown: " << error.what();
+ 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.
+template <bool kIsInt32>
+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 {
+ uint64_t hash;
+ if constexpr (kIsInt32) {
+ hash =
functions::sparksql::XxHash64::hashInt32(static_cast<int32_t>(value), seed_);
+ } else {
+ hash = functions::sparksql::XxHash64::hashInt64(value, seed_);
+ }
+ return view_->mayContain(folly::hasher<int64_t>()(hash));
+ }
+
+ 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<kIsInt32>>(constantVector_,
nullAllowed.value_or(nullAllowed_), seed_);
+ }
+
+ bool testingEquals(const Filter& other) const override {
+ return dynamic_cast<const SparkMightContain<kIsInt32>*>(&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 +102,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 +166,42 @@ SparkExprToSubfieldFilterParser::leafCallToSubfieldFilter(
}
return std::make_pair(std::move(subfield),
facebook::velox::exec::isNotNull());
}
+ } else if (scanBloomFilterPushdownEnabled_ && 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");
+ const auto inputTypeKind = hashCall->inputs()[1]->type()->kind();
+ if (inputTypeKind != TypeKind::INTEGER && inputTypeKind !=
TypeKind::BIGINT) {
+ return std::nullopt;
+ }
+ 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;
+ if (inputTypeKind == TypeKind::INTEGER) {
+ filter = std::make_unique<SparkMightContain<true>>(bloomFilterValue,
false /*nullAllowed*/, seed);
+ } else {
+ filter =
std::make_unique<SparkMightContain<false>>(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/cpp/velox/operators/functions/SparkExprToSubfieldFilterParser.h
b/cpp/velox/operators/functions/SparkExprToSubfieldFilterParser.h
index 0f7d826446..0f28d30f6a 100644
--- a/cpp/velox/operators/functions/SparkExprToSubfieldFilterParser.h
+++ b/cpp/velox/operators/functions/SparkExprToSubfieldFilterParser.h
@@ -23,11 +23,17 @@ namespace gluten {
/// 2) The supported functions vary.
class SparkExprToSubfieldFilterParser : public
facebook::velox::exec::ExprToSubfieldFilterParser {
public:
+ explicit SparkExprToSubfieldFilterParser(bool scanBloomFilterPushdownEnabled)
+ : scanBloomFilterPushdownEnabled_(scanBloomFilterPushdownEnabled) {}
+
std::optional<std::pair<facebook::velox::common::Subfield,
std::unique_ptr<facebook::velox::common::Filter>>>
leafCallToSubfieldFilter(
const facebook::velox::core::CallTypedExpr& call,
facebook::velox::core::ExpressionEvaluator* evaluator,
bool negated) override;
+
+ private:
+ const bool scanBloomFilterPushdownEnabled_;
};
} // namespace gluten
diff --git a/docs/velox-configuration.md b/docs/velox-configuration.md
index 5a0a9bf498..712260f984 100644
--- a/docs/velox-configuration.md
+++ b/docs/velox-configuration.md
@@ -75,6 +75,7 @@ nav_order: 16
| spark.gluten.sql.columnar.backend.velox.resizeBatches.shuffleInput.minSize
| 🔄 Dynamic | <undefined> | The minimum batch size for shuffle. If
size of an input batch is smaller than the value, it will be combined with
other batches before sending to shuffle. Only functions when
spark.gluten.sql.columnar.backend.velox.resizeBatches.shuffleInput is set to
true. Default value: 0.25 * <max batch size>
[...]
|
spark.gluten.sql.columnar.backend.velox.resizeBatches.shuffleInputOutput.minSize
| 🔄 Dynamic | <undefined> | The minimum batch size for shuffle input
and output. If size of an input batch is smaller than the value, it will be
combined with other batches before sending to shuffle. The same applies for
batches output by shuffle read. Only functions when
spark.gluten.sql.columnar.backend.velox.resizeBatches.shuffleInput or
spark.gluten.sql.columnar.backend.velox.resizeBatches.shu [...]
| spark.gluten.sql.columnar.backend.velox.resizeBatches.shuffleOutput
| 🔄 Dynamic | false | If true, combine small columnar
batches together right after shuffle read. The default minimum output batch
size is equal to 0.25 * spark.gluten.sql.columnar.maxBatchSize
[...]
+| spark.gluten.sql.columnar.backend.velox.scan.bloomFilterPushdown.enabled
| âš“ Static | false | Whether to push Bloom filters into
Velox scans.
[...]
| spark.gluten.sql.columnar.backend.velox.showTaskMetricsWhenFinished
| 🔄 Dynamic | false | Show velox full task metrics when
finished.
[...]
| spark.gluten.sql.columnar.backend.velox.spillFileSystem
| 🔄 Dynamic | local | The filesystem used to store spill
data. local: The local file system. heap-over-local: Write file to JVM heap if
having extra heap space. Otherwise write to local file system.
[...]
| spark.gluten.sql.columnar.backend.velox.spillStrategy
| 🔄 Dynamic | auto | none: Disable spill on Velox backend;
auto: Let Spark memory manager manage Velox's spilling
[...]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
