From a9359a79798913f38a68d1a7fb56db0b36324ef4 Mon Sep 17 00:00:00 2001 From: Shaojie Li Date: Sat, 30 May 2026 06:56:11 +0000 Subject: [PATCH] perf(parquet): Reuse DeltaByteArrayDecoder via reset() + devirtualize Three layered optimizations on the DELTA_BYTE_ARRAY hot path: 1. Drop the DeltaByteArrayDecoderBase virtual base. Both DeltaByteArrayDecoder and DeltaLengthByteArrayDecoder are concrete classes now; PageReader holds the right pointer type for each encoding. Removes vtable dispatch from per-row readString(). 2. Lazy-init child decoders via std::optional + reset(pageData_). Previously every page boundary called make_unique on three RleBpDecoders + reset their state; now the decoders are constructed once per column-chunk and re-pointed at the next page via reset(). Saves ~3 heap allocations per page. 3. Replace `std::string lastValue_` with `std::vector lastValueBuf_` plus an int32 length. perf showed ~6.5% in std::string::_M_replace on every readString() because the prefix-share encoding repeatedly resizes the previous-value buffer. The vector path uses raw resize+memcpy with no allocator interaction once the buffer reaches max suffix size. Bench (TPC-H Q12 SF10, on top of #pr-A): wall scan_cpu PR A only 0.85 s 2.13 s + reset() reuse 0.82 s 2.03 s + raw char buffer 0.78 s 1.91 s + devirtualize readString 0.74 s 1.74 s End-to-end: PR A + PR B brings Q12 SF10 from PR #17633's 1.02 s wall down to 0.74 s wall (-27%). Q12 SF100 16.3 s -> 13.1 s (-20%). Tracking: jaylisde/velox#2 --- .../parquet/reader/DeltaByteArrayDecoder.h | 176 ++++++++++++------ velox/dwio/parquet/reader/PageReader.cpp | 13 +- 2 files changed, 123 insertions(+), 66 deletions(-) diff --git a/velox/dwio/parquet/reader/DeltaByteArrayDecoder.h b/velox/dwio/parquet/reader/DeltaByteArrayDecoder.h index b8ad2d18d48..11d3aa42854 100644 --- a/velox/dwio/parquet/reader/DeltaByteArrayDecoder.h +++ b/velox/dwio/parquet/reader/DeltaByteArrayDecoder.h @@ -16,24 +16,47 @@ #pragma once +#include + #include "velox/common/base/BitUtil.h" #include "velox/common/base/Nulls.h" #include "velox/dwio/parquet/reader/DeltaBpDecoder.h" namespace facebook::velox::parquet { -class DeltaByteArrayDecoderBase { +// DeltaLengthByteArrayDecoder is adapted from Apache Arrow: +// https://github.com/apache/arrow/blob/apache-arrow-15.0.0/cpp/src/parquet/encoding.cc#L2758-L2889 +class DeltaLengthByteArrayDecoder { public: - virtual ~DeltaByteArrayDecoderBase() = default; + DeltaLengthByteArrayDecoder() = default; + + explicit DeltaLengthByteArrayDecoder(const char* start) { + reset(start); + } - virtual std::string_view readString() = 0; + // Re-initialize the decoder for a new page. Reuses the existing + // bufferedLength_ vector capacity; only allocates if numLength + // exceeds it. + void reset(const char* start) { + lengthDecoder_.emplace(start); + decodeLengths(); + bufferStart_ = lengthDecoder_->bufferStart(); + } + + FOLLY_ALWAYS_INLINE std::string_view readString() { + const int64_t length = bufferedLength_[lengthIdx_++]; + VELOX_CHECK_GE(length, 0, "negative string delta length"); + bufferStart_ += length; + return std::string_view(bufferStart_ - length, length); + } void skip(uint64_t numValues) { skip(numValues, 0, nullptr); } template - inline void skip(int32_t numValues, int32_t current, const uint64_t* nulls) { + FOLLY_ALWAYS_INLINE void + skip(int32_t numValues, int32_t current, const uint64_t* nulls) { if (hasNulls) { numValues = bits::countNonNulls(nulls, current, current + numValues); } @@ -76,24 +99,6 @@ class DeltaByteArrayDecoderBase { } } } -}; - -// DeltaByteArrayDecoder is adapted from Apache Arrow: -// https://github.com/apache/arrow/blob/apache-arrow-15.0.0/cpp/src/parquet/encoding.cc#L2758-L2889 -class DeltaLengthByteArrayDecoder : public DeltaByteArrayDecoderBase { - public: - explicit DeltaLengthByteArrayDecoder(const char* start) { - lengthDecoder_ = std::make_unique(start); - decodeLengths(); - bufferStart_ = lengthDecoder_->bufferStart(); - } - - std::string_view readString() override { - const int64_t length = bufferedLength_[lengthIdx_++]; - VELOX_CHECK_GE(length, 0, "negative string delta length"); - bufferStart_ += length; - return std::string_view(bufferStart_ - length, length); - } private: void decodeLengths() { @@ -107,7 +112,7 @@ class DeltaLengthByteArrayDecoder : public DeltaByteArrayDecoderBase { } const char* bufferStart_; - std::unique_ptr lengthDecoder_; + std::optional lengthDecoder_; int32_t numValidValues_{0}; uint32_t lengthIdx_{0}; std::vector bufferedLength_; @@ -115,74 +120,121 @@ class DeltaLengthByteArrayDecoder : public DeltaByteArrayDecoderBase { // DeltaByteArrayDecoder is adapted from Apache Arrow: // https://github.com/apache/arrow/blob/apache-arrow-15.0.0/cpp/src/parquet/encoding.cc#L3301-L3545 -class DeltaByteArrayDecoder : public DeltaByteArrayDecoderBase { +class DeltaByteArrayDecoder { public: + DeltaByteArrayDecoder() = default; + explicit DeltaByteArrayDecoder(const char* start) { - prefixLenDecoder_ = std::make_unique(start); + reset(start); + } + + // Re-initialize the decoder for a new page. Reuses the existing + // bufferedPrefixLength_, lastValueBuf_, and the embedded + // suffixDecoder_'s buffers; only allocates if their capacities are + // insufficient. + void reset(const char* start) { + prefixLenDecoder_.emplace(start); int64_t numPrefix = prefixLenDecoder_->validValuesCount(); bufferedPrefixLength_.resize(numPrefix); prefixLenDecoder_->readValues( bufferedPrefixLength_.data(), static_cast(numPrefix)); prefixLenOffset_ = 0; numValidValues_ = static_cast(numPrefix); + lastValueLen_ = 0; - suffixDecoder_ = std::make_unique( - prefixLenDecoder_->bufferStart()); + if (!suffixDecoder_.has_value()) { + suffixDecoder_.emplace(prefixLenDecoder_->bufferStart()); + } else { + suffixDecoder_->reset(prefixLenDecoder_->bufferStart()); + } } - std::string_view readString() override { + FOLLY_ALWAYS_INLINE std::string_view readString() { auto suffix = suffixDecoder_->readString(); - bool isFirstRun = (prefixLenOffset_ == 0); const int64_t prefixLength = bufferedPrefixLength_[prefixLenOffset_++]; VELOX_CHECK_GE( prefixLength, 0, "negative prefix length in DELTA_BYTE_ARRAY"); + VELOX_CHECK_LE( + prefixLength, + lastValueLen_, + "prefix length too large in DELTA_BYTE_ARRAY"); - buildReadValue(isFirstRun, prefixLength, suffix); + // Append suffix bytes after the kept prefix in lastValueBuf_, growing + // the buffer if needed. lastValueBuf_ persists across calls and + // amortizes allocation. + const size_t need = prefixLength + suffix.size(); + if (need > lastValueBuf_.size()) { + lastValueBuf_.resize(need); + } + if (!suffix.empty()) { + memcpy(lastValueBuf_.data() + prefixLength, suffix.data(), suffix.size()); + } + lastValueLen_ = static_cast(need); numValidValues_--; - return {lastValue_}; + return {lastValueBuf_.data(), static_cast(lastValueLen_)}; } - private: - void buildReadValue( - bool isFirstRun, - const int64_t prefixLength, - std::string_view suffix) { - VELOX_CHECK_LE( - prefixLength, - lastValue_.size(), - "prefix length too large in DELTA_BYTE_ARRAY"); + void skip(uint64_t numValues) { + skip(numValues, 0, nullptr); + } - if (prefixLength == 0) { - // prefix is empty. - lastValue_ = std::string{suffix}; - return; + template + FOLLY_ALWAYS_INLINE void + skip(int32_t numValues, int32_t current, const uint64_t* nulls) { + if (hasNulls) { + numValues = bits::countNonNulls(nulls, current, current + numValues); } + for (int32_t i = 0; i < numValues; ++i) { + readString(); + } + } - if (!isFirstRun) { - if (suffix.empty()) { - // suffix is empty: read value can simply point to the prefix - // of the lastValue_. This is not possible for the first run since - // the prefix would point to the mutable `lastValue_`. - lastValue_ = lastValue_.substr(0, prefixLength); + template + void readWithVisitor(const uint64_t* nulls, Visitor visitor) { + int32_t current = visitor.start(); + skip(current, 0, nulls); + int32_t toSkip; + bool atEnd = false; + const bool allowNulls = hasNulls && visitor.allowNulls(); + for (;;) { + if (hasNulls && allowNulls && bits::isBitNull(nulls, current)) { + toSkip = visitor.processNull(atEnd); + } else { + if (hasNulls && !allowNulls) { + toSkip = visitor.checkAndSkipNulls(nulls, current, atEnd); + if (!Visitor::dense) { + skip(toSkip, current, nullptr); + } + if (atEnd) { + return; + } + } + + // We are at a non-null value on a row to visit. + toSkip = visitor.process(readString(), atEnd); + } + ++current; + if (toSkip) { + skip(toSkip, current, nulls); + current += toSkip; + } + if (atEnd) { return; } } - - lastValue_.resize(prefixLength + suffix.size()); - - // Both prefix and suffix are non-empty, so we need to decode the string - // into read value. - // Just keep the prefix in lastValue_, and copy the suffix. - memcpy(lastValue_.data() + prefixLength, suffix.data(), suffix.size()); } - std::unique_ptr prefixLenDecoder_; - std::unique_ptr suffixLenDecoder_; - std::unique_ptr suffixDecoder_; - - std::string lastValue_; + private: + std::optional prefixLenDecoder_; + std::optional suffixDecoder_; + + // Reconstructed value of the most recently returned string. The + // returned string_view aliases this buffer; the next call to + // readString() overwrites it. + std::vector lastValueBuf_; + int32_t lastValueLen_{0}; int32_t numValidValues_{0}; uint32_t prefixLenOffset_{0}; std::vector bufferedPrefixLength_; diff --git a/velox/dwio/parquet/reader/PageReader.cpp b/velox/dwio/parquet/reader/PageReader.cpp index 8010c477a06..2b21cfe8369 100644 --- a/velox/dwio/parquet/reader/PageReader.cpp +++ b/velox/dwio/parquet/reader/PageReader.cpp @@ -846,15 +846,20 @@ void PageReader::makeDecoder() { break; case Encoding::DELTA_BYTE_ARRAY: if (parquetType == thrift::Type::BYTE_ARRAY) { - deltaByteArrDecoder_ = - std::make_unique(pageData_); + if (!deltaByteArrDecoder_) { + deltaByteArrDecoder_ = std::make_unique(); + } + deltaByteArrDecoder_->reset(pageData_); break; } [[fallthrough]]; case Encoding::DELTA_LENGTH_BYTE_ARRAY: if (parquetType == thrift::Type::BYTE_ARRAY) { - deltaLengthByteArrDecoder_ = - std::make_unique(pageData_); + if (!deltaLengthByteArrDecoder_) { + deltaLengthByteArrDecoder_ = + std::make_unique(); + } + deltaLengthByteArrDecoder_->reset(pageData_); break; } [[fallthrough]];