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 67c7192d95..468db95325 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 @@ -115,6 +115,9 @@ class VeloxConfig(conf: SQLConf) extends GlutenConfig(conf) { def scanBloomFilterPushdownEnabled: Boolean = getConf(SCAN_BLOOM_FILTER_PUSHDOWN_ENABLED) + def scanBloomFilterBufferCacheEnabled: Boolean = + getConf(SCAN_BLOOM_FILTER_BUFFER_CACHE_ENABLED) + def enableTimestampNtzValidation: Boolean = getConf(ENABLE_TIMESTAMP_NTZ_VALIDATION) def enableDriverSideBroadcastHashTableBuild: Boolean = @@ -567,6 +570,12 @@ object VeloxConfig extends ConfigRegistry { .booleanConf .createWithDefault(false) + val SCAN_BLOOM_FILTER_BUFFER_CACHE_ENABLED = + buildConf("spark.gluten.sql.columnar.backend.velox.scan.bloomFilterBufferCache.enabled") + .doc("Whether to share scan Bloom filter buffers across Velox tasks in an executor.") + .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 074aec9df4..1f7e360577 100644 --- a/cpp/velox/compute/VeloxBackend.cc +++ b/cpp/velox/compute/VeloxBackend.cc @@ -260,8 +260,8 @@ void VeloxBackend::init( velox::parquet::registerParquetReaderFactory(); velox::parquet::registerParquetWriterFactory(); velox::orc::registerOrcReaderFactory(); - velox::exec::ExprToSubfieldFilterParser::registerParser(std::make_unique( - backendConf_->get(kScanBloomFilterPushdownEnabled, kScanBloomFilterPushdownEnabledDefault))); + velox::exec::ExprToSubfieldFilterParser::registerParser( + std::make_unique(backendConf_)); velox::connector::hive::BufferedInputBuilder::registerBuilder(std::make_shared()); // Register Velox functions diff --git a/cpp/velox/config/VeloxConfig.h b/cpp/velox/config/VeloxConfig.h index f9351560cf..46fcf790f8 100644 --- a/cpp/velox/config/VeloxConfig.h +++ b/cpp/velox/config/VeloxConfig.h @@ -88,6 +88,9 @@ const std::string kSparkBloomFilterMaxNumItems = "spark.sql.optimizer.runtime.bl const std::string kScanBloomFilterPushdownEnabled = "spark.gluten.sql.columnar.backend.velox.scan.bloomFilterPushdown.enabled"; const bool kScanBloomFilterPushdownEnabledDefault = false; +const std::string kScanBloomFilterBufferCacheEnabled = + "spark.gluten.sql.columnar.backend.velox.scan.bloomFilterBufferCache.enabled"; +const bool kScanBloomFilterBufferCacheEnabledDefault = 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 31d44211cd..5313f89493 100644 --- a/cpp/velox/operators/functions/SparkExprToSubfieldFilterParser.cc +++ b/cpp/velox/operators/functions/SparkExprToSubfieldFilterParser.cc @@ -16,6 +16,14 @@ */ #include "operators/functions/SparkExprToSubfieldFilterParser.h" +#include +#include +#include +#include +#include +#include + +#include "config/VeloxConfig.h" #include "utils/Exception.h" #include "velox/common/base/BloomFilter.h" #include "velox/expression/Expr.h" @@ -28,6 +36,60 @@ using namespace facebook::velox; namespace { +// Shares immutable Bloom filter bytes across tasks in this executor process. +// Weak references let the buffer expire after the last filter releases it. +class BloomFilterBufferCache { + public: + static BloomFilterBufferCache& instance() { + static BloomFilterBufferCache cache; + return cache; + } + + std::shared_ptr intern(StringView value) { + const std::string_view bytes(value.data(), value.size()); + const auto hash = std::hash{}(bytes); + std::lock_guard lock(mutex_); + auto mapIt = entries_.find(hash); + if (mapIt != entries_.end()) { + for (const auto& entry : mapIt->second) { + if (auto buffer = entry.lock()) { + // Hash collisions must not cause different filters to share bytes. + if (buffer->size() == bytes.size() && std::equal(buffer->begin(), buffer->end(), bytes.begin())) { + return buffer; + } + } + } + } + auto buffer = std::make_shared(bytes); + entries_[hash].emplace_back(buffer); + if (++entriesSinceSweep_ >= kSweepInterval) { + sweep(); + entriesSinceSweep_ = 0; + } + return buffer; + } + + private: + void sweep() { + for (auto mapIt = entries_.begin(); mapIt != entries_.end();) { + auto& entries = mapIt->second; + entries.erase( + std::remove_if(entries.begin(), entries.end(), [](const auto& entry) { return entry.expired(); }), + entries.end()); + if (entries.empty()) { + mapIt = entries_.erase(mapIt); + } else { + ++mapIt; + } + } + } + + static constexpr size_t kSweepInterval = 128; + std::mutex mutex_; + std::unordered_map>> entries_; + size_t entriesSinceSweep_{0}; +}; + // 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. @@ -55,10 +117,10 @@ VectorPtr toConstant(const core::TypedExprPtr& expr, core::ExpressionEvaluator* template 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()); + SparkMightContain(std::shared_ptr buffer, bool nullAllowed, int64_t seed) + : common::BigintValuesUsingBloomFilter(0, nullAllowed), buffer_(std::move(buffer)), seed_(seed) { + // BloomFilterView is non-owning; buffer_ keeps its bytes alive. + view_ = std::make_unique(buffer_->data()); } bool testInt64(int64_t value) const override { @@ -76,7 +138,7 @@ class SparkMightContain final : public common::BigintValuesUsingBloomFilter { } std::unique_ptr clone(std::optional nullAllowed) const override { - return std::make_unique>(constantVector_, nullAllowed.value_or(nullAllowed_), seed_); + return std::make_unique>(buffer_, nullAllowed.value_or(nullAllowed_), seed_); } bool testingEquals(const Filter& other) const override { @@ -88,7 +150,7 @@ class SparkMightContain final : public common::BigintValuesUsingBloomFilter { } private: - VectorPtr constantVector_; + std::shared_ptr buffer_; std::unique_ptr view_; int64_t seed_; }; @@ -166,7 +228,9 @@ SparkExprToSubfieldFilterParser::leafCallToSubfieldFilter( } return std::make_pair(std::move(subfield), facebook::velox::exec::isNotNull()); } - } else if (scanBloomFilterPushdownEnabled_ && call.name() == "might_contain" && !negated) { + } else if ( + backendConf_->get(kScanBloomFilterPushdownEnabled, kScanBloomFilterPushdownEnabledDefault) && + call.name() == "might_contain" && !negated) { // Matches: might_contain(bloomFilter, xxhash64_with_seed(seed, field)). GLUTEN_CHECK( call.inputs().size() == 2, @@ -192,11 +256,18 @@ SparkExprToSubfieldFilterParser::leafCallToSubfieldFilter( } auto bloomFilterValue = toConstant(call.inputs()[0], evaluator); if (bloomFilterValue && !bloomFilterValue->isNullAt(0)) { + const auto bloomFilterBytes = bloomFilterValue->as>()->valueAt(0); + std::shared_ptr bloomFilterBuffer; + if (backendConf_->get(kScanBloomFilterBufferCacheEnabled, kScanBloomFilterBufferCacheEnabledDefault)) { + bloomFilterBuffer = BloomFilterBufferCache::instance().intern(bloomFilterBytes); + } else { + bloomFilterBuffer = std::make_shared(bloomFilterBytes.data(), bloomFilterBytes.size()); + } std::unique_ptr filter; if (inputTypeKind == TypeKind::INTEGER) { - filter = std::make_unique>(bloomFilterValue, false /*nullAllowed*/, seed); + filter = std::make_unique>(bloomFilterBuffer, false /*nullAllowed*/, seed); } else { - filter = std::make_unique>(bloomFilterValue, false /*nullAllowed*/, seed); + filter = std::make_unique>(bloomFilterBuffer, false /*nullAllowed*/, seed); } return combine(subfield, filter); } diff --git a/cpp/velox/operators/functions/SparkExprToSubfieldFilterParser.h b/cpp/velox/operators/functions/SparkExprToSubfieldFilterParser.h index 0f28d30f6a..aa5a8d6b94 100644 --- a/cpp/velox/operators/functions/SparkExprToSubfieldFilterParser.h +++ b/cpp/velox/operators/functions/SparkExprToSubfieldFilterParser.h @@ -14,6 +14,10 @@ * See the License for the specific language governing permissions and * limitations under the License. */ + +#pragma once + +#include "velox/common/config/Config.h" #include "velox/expression/ExprToSubfieldFilter.h" namespace gluten { @@ -23,8 +27,8 @@ namespace gluten { /// 2) The supported functions vary. class SparkExprToSubfieldFilterParser : public facebook::velox::exec::ExprToSubfieldFilterParser { public: - explicit SparkExprToSubfieldFilterParser(bool scanBloomFilterPushdownEnabled) - : scanBloomFilterPushdownEnabled_(scanBloomFilterPushdownEnabled) {} + explicit SparkExprToSubfieldFilterParser(std::shared_ptr backendConf) + : backendConf_(std::move(backendConf)) {} std::optional>> leafCallToSubfieldFilter( @@ -33,7 +37,7 @@ class SparkExprToSubfieldFilterParser : public facebook::velox::exec::ExprToSubf bool negated) override; private: - const bool scanBloomFilterPushdownEnabled_; + const std::shared_ptr backendConf_; }; } // namespace gluten diff --git a/docs/velox-configuration.md b/docs/velox-configuration.md index a21373fe4a..c49579aef4 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 * | | 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.shuffleOutput is set to true. Default value: 0.25 * | | 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.bloomFilterBufferCache.enabled | 🔄 Dynamic | false | Whether to share scan Bloom filter buffers across Velox tasks in an executor. | | 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. |