From 4dde7caeda868e2682e245a940bf9f7c2d2a23f9 Mon Sep 17 00:00:00 2001 From: Shaojie Li Date: Wed, 3 Jun 2026 06:30:13 +0000 Subject: [PATCH] perf(parquet): Batch DELTA decode + SIMD filter pushdown MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Layered optimizations on top of #17633 to close the remaining DELTA-vs-PLAIN/DICT scan gap on TPC-H Q12 (DELTA-encoded lineitem). 1. Batched fast path in DeltaBpDecoder::readWithVisitor — for dense integer reads with deterministic filters and no hook, decode a chunk into the visitor's output buffer and dispatch one visitor.processRun() per chunk. 2. Inline readLong() state hoist in decodeLongs — keeps bufferStart_, the two remaining counters, lastValue_, minDelta_ and deltaBitWidth_ in registers across the inner loop; advances a running bitOffset instead of recomputing valuesConsumed * deltaBitWidth_ per row. 3. New base ColumnVisitor::processRun with two SIMD subpaths via dwio::common::processFixedFilter. Specialized visitors keep their own overrides; this base entry is reachable only from the new DELTA batched path. 4. Fix scalar-tail / SIMD-counter sync in processRun. The dense batched caller commits the SIMD loop's local counter via setNumValues(numValues), but the scalar tail's process() advances the reader's own numValues_ counter, so any rows produced by the tail when numInput < kWidth were silently dropped. Sync the local into the reader before the tail and pull the tail's increments back into the local after, so the final commit covers both phases. 5. Seven boundary tests in ParquetTableScanTest covering bit_width=0 constant deltas, narrow widths, the 32/33-bit boundary, mixed miniblock widths, a wide-and-negative pattern, and the scalar-tail regression above (deltaBinaryPackedFilterScalarTail). --- velox/dwio/common/ColumnVisitors.h | 74 +++++++ velox/dwio/parquet/reader/CMakeLists.txt | 1 + velox/dwio/parquet/reader/DeltaBpDecoder.cpp | 103 ++++++++++ velox/dwio/parquet/reader/DeltaBpDecoder.h | 181 +++++++++++------- .../tests/reader/ParquetTableScanTest.cpp | 117 +++++++++++ 5 files changed, 403 insertions(+), 73 deletions(-) create mode 100644 velox/dwio/parquet/reader/DeltaBpDecoder.cpp diff --git a/velox/dwio/common/ColumnVisitors.h b/velox/dwio/common/ColumnVisitors.h index 33d7a2d6e32..e9c39612ff0 100644 --- a/velox/dwio/common/ColumnVisitors.h +++ b/velox/dwio/common/ColumnVisitors.h @@ -408,6 +408,80 @@ class ColumnVisitor { inline void addNull(); inline void addOutputRow(vector_size_t row); + /// Bulk variant of process() with two SIMD fast paths: AlwaysTrue + /// (counter-only) and deterministic integral filter (batch-filter via + /// processFixedFilter). All other cases fall back to a per-row loop. + template + FOLLY_ALWAYS_INLINE void processRun( + const T* input, + int32_t numInput, + const int32_t* scatterRows, + int32_t* filterHits, + T* values, + int32_t& numValues) { + DCHECK_EQ(input, values + numValues); + if constexpr ( + !hasFilter && !hasHook && !scatter && isDense && + std::is_same_v) { + rowIndex_ += numInput; + numValues += numInput; + return; + } + if constexpr ( + hasFilter && !hasHook && !scatter && TFilter::deterministic && + (std::is_same_v || std::is_same_v || + std::is_same_v)) { + const int32_t firstRow = currentRow(); + constexpr int32_t kWidth = xsimd::batch::size; + const int32_t numValuesAtEntry = numValues; + int32_t i = 0; + while (i + kWidth <= numInput) { + auto batch = xsimd::load_unaligned(input + i); + ::facebook::velox::dwio::common::processFixedFilter< + T, + /*filterOnly=*/false, + /*scatter=*/false, + /*dense=*/true>( + batch, + kWidth, + firstRow + i, + filter_, + [&](int32_t /*offset*/) { + return ::facebook::velox::simd::loadGatherIndices( + rows_ + rowIndex_ + i); + }, + values, + filterHits, + numValues); + i += kWidth; + } + // Scalar tail uses process(), which appends through the reader's + // own numValues_ counter (not the local one). Sync the local SIMD + // count into the reader before the tail so process() lands at the + // next free slot, then sync the tail's appends back into the local + // counter so the surrounding setNumValues(numValues) commit covers + // them. + addNumValues(numValues - numValuesAtEntry); + rowIndex_ += i; + bool atEnd = false; + for (; i < numInput; ++i) { + process(input[i], atEnd); + if (atEnd) { + break; + } + } + numValues = reader_->numValues() - numValuesBias_; + return; + } + bool atEnd = false; + for (int32_t i = 0; i < numInput; ++i) { + process(input[i], atEnd); + if (atEnd) { + return; + } + } + } + const TFilter& filter() { return filter_; } diff --git a/velox/dwio/parquet/reader/CMakeLists.txt b/velox/dwio/parquet/reader/CMakeLists.txt index d73c8a31c67..5193a144970 100644 --- a/velox/dwio/parquet/reader/CMakeLists.txt +++ b/velox/dwio/parquet/reader/CMakeLists.txt @@ -14,6 +14,7 @@ velox_add_library( velox_dwio_native_parquet_reader + DeltaBpDecoder.cpp Metadata.cpp NestedStructureDecoder.cpp ParquetReader.cpp diff --git a/velox/dwio/parquet/reader/DeltaBpDecoder.cpp b/velox/dwio/parquet/reader/DeltaBpDecoder.cpp new file mode 100644 index 00000000000..7dd4552d3e4 --- /dev/null +++ b/velox/dwio/parquet/reader/DeltaBpDecoder.cpp @@ -0,0 +1,103 @@ +/* + * Copyright (c) Facebook, Inc. and its affiliates. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +#include "velox/dwio/parquet/reader/DeltaBpDecoder.h" + +#include + +namespace facebook::velox::parquet { + +bool DeltaBpDecoder::getVlqInt(uint64_t& v) { + uint64_t tmp = 0; + for (int i = 0; i < folly::kMaxVarintLength64; i++) { + uint8_t byte = *(bufferStart_++); + tmp |= static_cast(byte & 0x7F) << (7 * i); + if ((byte & 0x80) == 0) { + v = tmp; + return true; + } + } + return false; +} + +bool DeltaBpDecoder::getZigZagVlqInt(int64_t& v) { + uint64_t u; + if (!getVlqInt(u)) { + return false; + } + v = (u >> 1) ^ (~(u & 1) + 1); + return true; +} + +void DeltaBpDecoder::initHeader() { + if (!getVlqInt(valuesPerBlock_) || !getVlqInt(miniBlocksPerBlock_) || + !getVlqInt(totalValueCount_) || !getZigZagVlqInt(lastValue_)) { + VELOX_FAIL("initHeader EOF"); + } + + VELOX_CHECK_GT(valuesPerBlock_, 0, "cannot have zero value per block"); + VELOX_CHECK_EQ( + valuesPerBlock_ % 128, + 0, + "the number of values in a block must be multiple of 128, but it's {}", + valuesPerBlock_); + VELOX_CHECK_GT( + miniBlocksPerBlock_, 0, "cannot have zero miniblock per block"); + valuesPerMiniBlock_ = valuesPerBlock_ / miniBlocksPerBlock_; + VELOX_CHECK_GT( + valuesPerMiniBlock_, 0, "cannot have zero value per miniblock"); + VELOX_CHECK_EQ( + valuesPerMiniBlock_ % 32, + 0, + "the number of values in a miniblock must be multiple of 32, but it's {}", + valuesPerMiniBlock_); + + totalValuesRemaining_ = totalValueCount_; + deltaBitWidths_.resize(miniBlocksPerBlock_); + firstBlockInitialized_ = false; + valuesRemainingCurrentMiniBlock_ = 0; +} + +void DeltaBpDecoder::initBlock() { + VELOX_DCHECK_GT(totalValuesRemaining_, 0, "initBlock called at EOF"); + + if (!getZigZagVlqInt(minDelta_)) { + VELOX_FAIL("initBlock EOF"); + } + + // read the bitwidth of each miniblock + for (uint32_t i = 0; i < miniBlocksPerBlock_; ++i) { + deltaBitWidths_[i] = *(bufferStart_++); + // Note that non-conformant bitwidth entries are allowed by the Parquet + // spec for extraneous miniblocks in the last block (GH-14923), so we + // check the bitwidths when actually using them (see initMiniBlock()). + } + + miniBlockIdx_ = 0; + firstBlockInitialized_ = true; + initMiniBlock(deltaBitWidths_[0]); +} + +void DeltaBpDecoder::initMiniBlock(int32_t bitWidth) { + VELOX_DCHECK_LE( + bitWidth, + kMaxDeltaBitWidth, + "delta bit width larger than integer bit width"); + deltaBitWidth_ = bitWidth; + valuesRemainingCurrentMiniBlock_ = valuesPerMiniBlock_; +} + +} // namespace facebook::velox::parquet diff --git a/velox/dwio/parquet/reader/DeltaBpDecoder.h b/velox/dwio/parquet/reader/DeltaBpDecoder.h index 75e846de705..7f5dfdf1ced 100644 --- a/velox/dwio/parquet/reader/DeltaBpDecoder.h +++ b/velox/dwio/parquet/reader/DeltaBpDecoder.h @@ -16,7 +16,6 @@ #pragma once -#include #include "velox/common/base/BitUtil.h" #include "velox/common/base/Exceptions.h" #include "velox/common/base/Nulls.h" @@ -36,7 +35,8 @@ class DeltaBpDecoder { } 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); } @@ -49,6 +49,13 @@ class DeltaBpDecoder { void readWithVisitor(const uint64_t* nulls, Visitor visitor) { int32_t current = visitor.start(); skip(current, 0, nulls); + if constexpr ( + Visitor::dense && !hasNulls && Visitor::FilterType::deterministic && + std::is_same_v && + std::is_integral_v) { + readWithVisitorDenseBatched(visitor); + return; + } int32_t toSkip; bool atEnd = false; const bool allowNulls = hasNulls && visitor.allowNulls(); @@ -89,95 +96,123 @@ class DeltaBpDecoder { } template - void readValues(T* values, int32_t numValues) { + FOLLY_ALWAYS_INLINE void readValues(T* values, int32_t numValues) { VELOX_DCHECK_LE(numValues, totalValuesRemaining_); - for (auto i = 0; i < numValues; i++) { - values[i] = T(readLong()); + if constexpr (std::is_integral_v) { + decodeLongs(values, numValues); + } else { + for (auto i = 0; i < numValues; i++) { + values[i] = T(readLong()); + } } } private: - bool getVlqInt(uint64_t& v) { - uint64_t tmp = 0; - for (int i = 0; i < folly::kMaxVarintLength64; i++) { - uint8_t byte = *(bufferStart_++); - tmp |= static_cast(byte & 0x7F) << (7 * i); - if ((byte & 0x80) == 0) { - v = tmp; - return true; - } + // Dense + integral + NoHook fast path: decode each chunk directly into + // visitor's output buffer, then dispatch one processRun() per chunk. + template + void readWithVisitorDenseBatched(Visitor& visitor) { + using DataType = typename Visitor::DataType; + constexpr bool kHasFilter = + !std:: + is_same_v; + constexpr int32_t kBatch = 256; + const int32_t total = visitor.numRows(); + DataType* output = visitor.rawValues(total); + int32_t* filterHits = kHasFilter ? visitor.outputRows(total) : nullptr; + int32_t numValues = 0; + int32_t consumed = 0; + while (consumed < total) { + const int32_t n = std::min(kBatch, total - consumed); + DataType* dst = output + numValues; + decodeLongs(dst, n); + visitor.template processRun< + kHasFilter, + /*hasHook=*/false, + /*scatter=*/false>( + dst, + n, + /*scatterRows=*/nullptr, + filterHits, + output, + numValues); + consumed += n; } - return false; + visitor.setNumValues(numValues); } - bool getZigZagVlqInt(int64_t& v) { - uint64_t u; - if (!getVlqInt(u)) { - return false; - } - v = (u >> 1) ^ (~(u & 1) + 1); - return true; - } + // Inlined readLong() with state hoisted to locals; advances a running + // bitOffset to avoid a multiply per row. DataType narrowing covers int32. + template + void decodeLongs(DataType* out, int32_t n) { + const char* bufStart = bufferStart_; + uint64_t valsPerMiniBlk = valuesPerMiniBlock_; + uint64_t miniBlockRemaining = valuesRemainingCurrentMiniBlock_; + uint64_t totalRemaining = totalValuesRemaining_; + int64_t lastValue = lastValue_; + int64_t minDelta = minDelta_; + uint64_t deltaBitWidth = deltaBitWidth_; + uint64_t bitOffset = (valsPerMiniBlk - miniBlockRemaining) * deltaBitWidth; - void initHeader() { - if (!getVlqInt(valuesPerBlock_) || !getVlqInt(miniBlocksPerBlock_) || - !getVlqInt(totalValueCount_) || !getZigZagVlqInt(lastValue_)) { - VELOX_FAIL("initHeader EOF"); - } + int32_t i = 0; + while (i < n) { + if (miniBlockRemaining == 0) { + bufferStart_ = bufStart; + valuesRemainingCurrentMiniBlock_ = 0; + totalValuesRemaining_ = totalRemaining; + lastValue_ = lastValue; + int64_t v = readLong(); + out[i] = static_cast(v); + bufStart = bufferStart_; + miniBlockRemaining = valuesRemainingCurrentMiniBlock_; + totalRemaining = totalValuesRemaining_; + lastValue = lastValue_; + minDelta = minDelta_; + deltaBitWidth = deltaBitWidth_; + bitOffset = (valsPerMiniBlk - miniBlockRemaining) * deltaBitWidth; + ++i; + continue; + } - VELOX_CHECK_GT(valuesPerBlock_, 0, "cannot have zero value per block"); - VELOX_CHECK_EQ( - valuesPerBlock_ % 128, - 0, - "the number of values in a block must be multiple of 128, but it's {}", - valuesPerBlock_); - VELOX_CHECK_GT( - miniBlocksPerBlock_, 0, "cannot have zero miniblock per block"); - valuesPerMiniBlock_ = valuesPerBlock_ / miniBlocksPerBlock_; - VELOX_CHECK_GT( - valuesPerMiniBlock_, 0, "cannot have zero value per miniblock"); - VELOX_CHECK_EQ( - valuesPerMiniBlock_ % 32, - 0, - "the number of values in a miniblock must be multiple of 32, but it's {}", - valuesPerMiniBlock_); + uint64_t value = 0; + if (deltaBitWidth) { + value = bits::detail::loadBits( + reinterpret_cast(bufStart), + bitOffset, + deltaBitWidth); + value &= (~0ULL >> (64 - deltaBitWidth)); + } + uint64_t result = static_cast(minDelta) + value + + static_cast(lastValue); + lastValue = static_cast(result); + out[i] = static_cast(result); + bitOffset += deltaBitWidth; + --miniBlockRemaining; + --totalRemaining; + if (miniBlockRemaining == 0 || totalRemaining == 0) { + bufStart += bits::nbytes(deltaBitWidth * valsPerMiniBlk); + bitOffset = 0; + } + ++i; + } - totalValuesRemaining_ = totalValueCount_; - deltaBitWidths_.resize(miniBlocksPerBlock_); - firstBlockInitialized_ = false; - valuesRemainingCurrentMiniBlock_ = 0; + bufferStart_ = bufStart; + valuesRemainingCurrentMiniBlock_ = miniBlockRemaining; + totalValuesRemaining_ = totalRemaining; + lastValue_ = lastValue; } - void initBlock() { - VELOX_DCHECK_GT(totalValuesRemaining_, 0, "initBlock called at EOF"); + bool getVlqInt(uint64_t& v); - if (!getZigZagVlqInt(minDelta_)) { - VELOX_FAIL("initBlock EOF"); - } + bool getZigZagVlqInt(int64_t& v); - // read the bitwidth of each miniblock - for (uint32_t i = 0; i < miniBlocksPerBlock_; ++i) { - deltaBitWidths_[i] = *(bufferStart_++); - // Note that non-conformant bitwidth entries are allowed by the Parquet - // spec for extraneous miniblocks in the last block (GH-14923), so we - // check the bitwidths when actually using them (see initMiniBlock()). - } + void initHeader(); - miniBlockIdx_ = 0; - firstBlockInitialized_ = true; - initMiniBlock(deltaBitWidths_[0]); - } + void initBlock(); - void initMiniBlock(int32_t bitWidth) { - VELOX_DCHECK_LE( - bitWidth, - kMaxDeltaBitWidth, - "delta bit width larger than integer bit width"); - deltaBitWidth_ = bitWidth; - valuesRemainingCurrentMiniBlock_ = valuesPerMiniBlock_; - } + void initMiniBlock(int32_t bitWidth); - int64_t readLong() { + FOLLY_ALWAYS_INLINE int64_t readLong() { int64_t value = 0; if (valuesRemainingCurrentMiniBlock_ == 0) { if (!firstBlockInitialized_) { diff --git a/velox/dwio/parquet/tests/reader/ParquetTableScanTest.cpp b/velox/dwio/parquet/tests/reader/ParquetTableScanTest.cpp index e50311ccce2..b39a4bcec28 100644 --- a/velox/dwio/parquet/tests/reader/ParquetTableScanTest.cpp +++ b/velox/dwio/parquet/tests/reader/ParquetTableScanTest.cpp @@ -1620,6 +1620,123 @@ TEST_F(ParquetTableScanTest, deltaBinaryPackedNarrowBitWidth) { run.template operator()(); } +TEST_F(ParquetTableScanTest, deltaBinaryPackedBitWidth32) { + // Deltas alternating between 0 and 2^32-1 force packed bit_width = 32, + // the upper boundary of dwio::common::unpack. + WriterOptions options; + options.enableDictionary = false; + options.encoding = + facebook::velox::parquet::arrow::Encoding::kDeltaBinaryPacked; + + constexpr vector_size_t kSize = 1024; + constexpr int64_t kStep = (1LL << 32) - 1; + auto vector = makeRowVector( + {"c"}, {makeFlatVector(kSize, [](auto row) { + return static_cast((row / 2) * kStep + (row % 2 ? kStep : 0)); + })}); + auto file = TempFilePath::create(); + writeToParquetFile(file->getPath(), {vector}, options); + loadData(vector->rowType(), vector); + + assertSelect({makeSplit(file->getPath())}, {"c"}, "SELECT c FROM tmp"); +} + +TEST_F(ParquetTableScanTest, deltaBinaryPackedBitWidth33) { + // Deltas alternating between 0 and 2^32 force packed bit_width = 33, + // just past the 32-bit boundary; readLong must fall back to the + // per-row loadBits path. + WriterOptions options; + options.enableDictionary = false; + options.encoding = + facebook::velox::parquet::arrow::Encoding::kDeltaBinaryPacked; + + constexpr vector_size_t kSize = 1024; + constexpr int64_t kStep = 1LL << 32; + auto vector = makeRowVector( + {"c"}, {makeFlatVector(kSize, [](auto row) { + return static_cast((row / 2) * kStep + (row % 2 ? kStep : 0)); + })}); + auto file = TempFilePath::create(); + writeToParquetFile(file->getPath(), {vector}, options); + loadData(vector->rowType(), vector); + + assertSelect({makeSplit(file->getPath())}, {"c"}, "SELECT c FROM tmp"); +} + +TEST_F(ParquetTableScanTest, deltaBinaryPackedMixedMiniblockWidths) { + // Within one block, the four miniblocks each pick a different bit_width + // (1, 8, 16, 24). Verifies that scratch is invalidated and refilled when + // the decoder transitions across miniblocks with differing widths. + WriterOptions options; + options.enableDictionary = false; + options.encoding = + facebook::velox::parquet::arrow::Encoding::kDeltaBinaryPacked; + + constexpr vector_size_t kSize = 128; + auto vector = + makeRowVector({"c"}, {makeFlatVector(kSize, [](auto row) { + auto deltaForRow = [](vector_size_t r) -> int64_t { + const int mb = (r / 32) % 4; + const int parity = r % 2; + if (mb == 0) + return parity; + if (mb == 1) + return parity ? 200LL : 0LL; + if (mb == 2) + return parity ? 50'000LL : 0LL; + return parity ? 12'000'000LL : 0LL; + }; + int64_t v = 0; + for (vector_size_t i = 0; i < row; ++i) { + v += deltaForRow(i); + } + return v; + })}); + auto file = TempFilePath::create(); + writeToParquetFile(file->getPath(), {vector}, options); + loadData(vector->rowType(), vector); + + assertSelect({makeSplit(file->getPath())}, {"c"}, "SELECT c FROM tmp"); +} + +TEST_F(ParquetTableScanTest, deltaBinaryPackedFilterScalarTail) { + // Regression: when readWithVisitorDenseBatched calls processRun with + // numInput < kWidth (SIMD batch size), the SIMD loop never runs and the + // scalar tail handles all rows. The tail had been calling visitor.process() + // which advances reader.numValues_ directly; the surrounding caller then + // calls setNumValues(local) at the end with a local counter that the tail + // never updated, silently dropping every row the tail produced. Reproduces + // by chaining filters so that one column's reader receives a single-value + // batch (total < kWidth) whose row would otherwise pass. + WriterOptions options; + options.enableDictionary = false; + options.encoding = + facebook::velox::parquet::arrow::Encoding::kDeltaBinaryPacked; + options.dataPageSize = 8 * 1024; + options.batchSize = 1024; + + constexpr vector_size_t kSize = 128 * 1024; + auto vector = makeRowVector( + {"a", "b"}, + {makeFlatVector( + kSize, [](auto row) { return static_cast(row); }), + makeFlatVector( + kSize, [](auto row) { return 100'000LL + row * 31LL; })}); + auto file = TempFilePath::create(); + writeToParquetFile(file->getPath(), {vector}, options); + loadData(vector->rowType(), vector); + + // a BETWEEN 10000 AND 70000 keeps exactly 1 row in some 10K-row scan + // window (row 70000), forcing b's reader to processRun a 1-row batch + // that lands entirely in the scalar tail. + assertSelectWithFilter( + {makeSplit(file->getPath())}, + {"a", "b"}, + {"a BETWEEN 10000 AND 70000", "b > 200000"}, + "", + "SELECT a, b FROM tmp WHERE a BETWEEN 10000 AND 70000 AND b > 200000"); +} + TEST_F(ParquetTableScanTest, deltaBinaryPackedWideAndNegative) { WriterOptions options; options.enableDictionary = false;