Skip to content

perf(parquet): Reuse DeltaByteArrayDecoder via reset() + devirtualize #4

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
jaylisde wants to merge 1 commit into from
Closed

perf(parquet): Reuse DeltaByteArrayDecoder via reset() + devirtualize #4

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
176 changes: 114 additions & 62 deletions velox/dwio/parquet/reader/DeltaByteArrayDecoder.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -16,24 +16,47 @@

#pragma once

#include <optional>

#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<false>(numValues, 0, nullptr);
}

template <bool hasNulls>
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);
}
Expand Down Expand Up @@ -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<DeltaBpDecoder>(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() {
Expand All @@ -107,82 +112,129 @@ class DeltaLengthByteArrayDecoder : public DeltaByteArrayDecoderBase {
}

const char* bufferStart_;
std::unique_ptr<DeltaBpDecoder> lengthDecoder_;
std::optional<DeltaBpDecoder> lengthDecoder_;
int32_t numValidValues_{0};
uint32_t lengthIdx_{0};
std::vector<uint32_t> bufferedLength_;
};

// 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<DeltaBpDecoder>(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<uint32_t>(
bufferedPrefixLength_.data(), static_cast<int32_t>(numPrefix));
prefixLenOffset_ = 0;
numValidValues_ = static_cast<int32_t>(numPrefix);
lastValueLen_ = 0;

suffixDecoder_ = std::make_unique<DeltaLengthByteArrayDecoder>(
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<int32_t>(need);

numValidValues_--;
return {lastValue_};
return {lastValueBuf_.data(), static_cast<size_t>(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<false>(numValues, 0, nullptr);
}

if (prefixLength == 0) {
// prefix is empty.
lastValue_ = std::string{suffix};
return;
template <bool hasNulls>
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 <bool hasNulls, typename Visitor>
void readWithVisitor(const uint64_t* nulls, Visitor visitor) {
int32_t current = visitor.start();
skip<hasNulls>(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<false>(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<hasNulls>(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<DeltaBpDecoder> prefixLenDecoder_;
std::unique_ptr<DeltaBpDecoder> suffixLenDecoder_;
std::unique_ptr<DeltaLengthByteArrayDecoder> suffixDecoder_;

std::string lastValue_;
private:
std::optional<DeltaBpDecoder> prefixLenDecoder_;
std::optional<DeltaLengthByteArrayDecoder> 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<char> lastValueBuf_;
int32_t lastValueLen_{0};
int32_t numValidValues_{0};
uint32_t prefixLenOffset_{0};
std::vector<uint32_t> bufferedPrefixLength_;
Expand Down
13 changes: 9 additions & 4 deletions velox/dwio/parquet/reader/PageReader.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -846,15 +846,20 @@ void PageReader::makeDecoder() {
break;
case Encoding::DELTA_BYTE_ARRAY:
if (parquetType == thrift::Type::BYTE_ARRAY) {
deltaByteArrDecoder_ =
std::make_unique<DeltaByteArrayDecoder>(pageData_);
if (!deltaByteArrDecoder_) {
deltaByteArrDecoder_ = std::make_unique<DeltaByteArrayDecoder>();
}
deltaByteArrDecoder_->reset(pageData_);
break;
}
[[fallthrough]];
case Encoding::DELTA_LENGTH_BYTE_ARRAY:
if (parquetType == thrift::Type::BYTE_ARRAY) {
deltaLengthByteArrDecoder_ =
std::make_unique<DeltaLengthByteArrayDecoder>(pageData_);
if (!deltaLengthByteArrDecoder_) {
deltaLengthByteArrDecoder_ =
std::make_unique<DeltaLengthByteArrayDecoder>();
}
deltaLengthByteArrDecoder_->reset(pageData_);
break;
}
[[fallthrough]];
Expand Down