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;