From e686ed5408472a18b0214886d9b71916d9bc0ecc Mon Sep 17 00:00:00 2001 From: Prateek Gaur Date: Fri, 18 Sep 2026 19:43:47 +0000 Subject: [PATCH 1/2] [C++][Parquet] Unpack equal-width DELTA_BINARY_PACKED miniblocks in one call Consecutive miniblocks sharing a bit width are packed back to back with no padding and hold a multiple of 32 values, so a run of them is bit-identical to one longer run at that width and can be unpacked in a single call. A miniblock joins the run only when its stored width equals the current one, which InitMiniBlock has already validated, so an unchecked width is never used. The run stops at the end of the block and at what the caller has room for. The prefix-sum loop also keeps the running value and the frame of reference in locals. Both have the same type as the caller's output buffer, so the compiler cannot prove the store does not alias them and otherwise reloads both for every value. Arithmetic is unchanged: every term stays in the unsigned type, so the wrapping the format specifies is preserved and no decoded value changes. --- cpp/src/parquet/decoder.cc | 52 +++++++++++++++++++++--- cpp/src/parquet/encoding_test.cc | 69 +++++++++++++++++++++++++++++++- 2 files changed, 114 insertions(+), 7 deletions(-) diff --git a/cpp/src/parquet/decoder.cc b/cpp/src/parquet/decoder.cc index c4d3fe5a8a5a..f7b2d655da80 100644 --- a/cpp/src/parquet/decoder.cc +++ b/cpp/src/parquet/decoder.cc @@ -1601,6 +1601,28 @@ class DeltaBitPackDecoder : public TypedDecoderImpl { values_remaining_current_mini_block_ = values_per_mini_block_; } + // Returns how many whole miniblocks after the current one can be unpacked in the + // same call. Miniblocks are packed back to back with no padding and hold a multiple + // of 32 values, so at a fixed bit width a run of them is bit-identical to one longer + // run. A miniblock only joins the run when its width equals delta_bit_width_, which + // InitMiniBlock has validated, so an unchecked width is never used. + uint32_t CoalescibleMiniBlocks(uint32_t values_available) const { + // A run can only start once there is room for the rest of the current miniblock. + if (values_available < values_remaining_current_mini_block_) { + return 0; + } + const uint8_t* bit_widths = delta_bit_widths_->data(); + uint32_t values_needed = values_remaining_current_mini_block_; + uint32_t count = 0; + while (mini_block_idx_ + count + 1 < mini_blocks_per_block_ && + bit_widths[mini_block_idx_ + count + 1] == delta_bit_width_ && + values_available - values_needed >= values_per_mini_block_) { + values_needed += values_per_mini_block_; + ++count; + } + return count; + } + int GetInternal(T* buffer, int max_values) { max_values = static_cast(std::min(max_values, total_values_remaining_)); if (max_values == 0) { @@ -1642,8 +1664,20 @@ class DeltaBitPackDecoder : public TypedDecoderImpl { } } - int values_decode = std::min(values_remaining_current_mini_block_, - static_cast(max_values - i)); + const uint32_t values_available = static_cast(max_values - i); + const uint32_t values_this_mini_block = + std::min(values_remaining_current_mini_block_, values_available); + // The zero bit width path below decodes without the unpacker, so there is no + // call to fold miniblocks into. Zero-width neighbours would otherwise join a run + // like any other equal width, and decode correctly, but that measured slower. + const uint32_t mini_blocks_coalesced = + delta_bit_width_ == 0 ? 0 : CoalescibleMiniBlocks(values_available); + // A non-empty run implies the current miniblock is fully drained, which the + // cursor update below relies on. + DCHECK(mini_blocks_coalesced == 0 || + values_this_mini_block == values_remaining_current_mini_block_); + const int values_decode = static_cast( + values_this_mini_block + mini_blocks_coalesced * values_per_mini_block_); if (delta_bit_width_ == 0) { // Fast path that avoids a back-to-back dependency between two consecutive // computations: we know all deltas decode to zero. We actually don't @@ -1658,15 +1692,21 @@ class DeltaBitPackDecoder : public TypedDecoderImpl { values_decode) { ParquetException::EofException(); } + // Held in locals because a store to `buffer` may alias these members, which + // would force a reload of each per value. + UT last = static_cast(last_value_); + const UT min_delta = static_cast(min_delta_); for (int j = 0; j < values_decode; ++j) { // Addition between min_delta, packed int and last_value should be treated as // unsigned addition. Overflow is as expected. - buffer[i + j] = static_cast(min_delta_) + static_cast(buffer[i + j]) + - static_cast(last_value_); - last_value_ = buffer[i + j]; + last += min_delta + static_cast(buffer[i + j]); + buffer[i + j] = last; } + last_value_ = static_cast(last); } - values_remaining_current_mini_block_ -= values_decode; + // The last miniblock of a coalesced run becomes the current one, fully drained. + mini_block_idx_ += mini_blocks_coalesced; + values_remaining_current_mini_block_ -= values_this_mini_block; i += values_decode; } total_values_remaining_ -= max_values; diff --git a/cpp/src/parquet/encoding_test.cc b/cpp/src/parquet/encoding_test.cc index 831829e4a210..878b23c93072 100644 --- a/cpp/src/parquet/encoding_test.cc +++ b/cpp/src/parquet/encoding_test.cc @@ -1752,7 +1752,8 @@ class TestDeltaBitPackEncoding : public TestEncodingBase { using c_type = typename Type::c_type; static constexpr int TYPE = Type::type_num; static constexpr size_t kNumRoundTrips = 3; - const std::vector kReadBatchSizes = {1, 11}; + // 100 spans several miniblocks but still ends inside one. + const std::vector kReadBatchSizes = {1, 11, 100}; void InitBoundData(int nvalues, int repeats, c_type half_range) { num_values_ = nvalues * repeats; @@ -2034,6 +2035,72 @@ TYPED_TEST(TestDeltaBitPackEncoding, ZeroDeltaBitWidth) { this->CheckRoundtripWithValues(int_values); } +TYPED_TEST(TestDeltaBitPackEncoding, MiniblockBitWidthRuns) { + // A decoder may unpack a run of miniblocks sharing a bit width in one call. Cover + // the width patterns that decide where such a run starts and stops. + using T = typename TypeParam::c_type; + + // Same values as in DeltaBitPackEncoder + constexpr int kValuesPerBlock = + std::is_same_v ? 128 : 256; + constexpr int kMiniBlocksPerBlock = 4; + constexpr int kValuesPerMiniBlock = kValuesPerBlock / kMiniBlocksPerBlock; + + // Gives miniblock i the bit width widths[i]: alternating deltas of `frame` and + // `frame + 2^(w-1)` make w the smallest width holding the residual, and `frame` the + // smallest delta, so it is the frame the encoder stores. + auto make_values = [](const std::vector& widths, T frame, int trailing_values) { + std::vector values; + values.reserve(widths.size() * kValuesPerMiniBlock + trailing_values + 1); + // The first value travels in the header and contributes no delta. + T current = 0; + values.push_back(current); + for (const int width : widths) { + const T spread = width == 0 ? T{0} : static_cast(T{1} << (width - 1)); + for (int i = 0; i < kValuesPerMiniBlock; ++i) { + current = static_cast(current + frame + (i % 2 == 0 ? T{0} : spread)); + values.push_back(current); + } + } + // A tail shorter than a miniblock makes a run stop at the end of the values. + for (int i = 0; i < trailing_values; ++i) { + current = static_cast(current + frame); + values.push_back(current); + } + return values; + }; + + struct Case { + const char* name; + std::vector widths; + int trailing_values; + }; + const std::vector cases = { + // One run covering every miniblock of the block. + {"uniform widths", {4, 4, 4, 4}, 0}, + // No two neighbours share a width, so no run forms. + {"no repeated width", {1, 8, 3, 16}, 0}, + // Runs that end partway through the block. + {"two runs of two", {1, 1, 8, 8}, 0}, + {"run then a change", {4, 4, 4, 16}, 0}, + // Zero-width miniblocks beside a run. + {"zero widths first", {0, 0, 3, 3}, 0}, + {"zero widths last", {3, 3, 0, 0}, 0}, + {"zero width inside a run", {3, 0, 3, 3}, 0}, + // Widths match across a block boundary, where a run must still stop. + {"across a block boundary", {4, 4, 4, 4, 4, 4, 4, 4}, 0}, + // A final block that ends in the middle of a miniblock. + {"partial last block", {4, 4, 4, 4}, 5}, + }; + + for (const auto& c : cases) { + for (const T frame : {T{0}, static_cast(-5)}) { + ARROW_SCOPED_TRACE("case = ", c.name, ", frame = ", static_cast(frame)); + this->CheckRoundtripWithValues(make_values(c.widths, frame, c.trailing_values)); + } + } +} + // ---------------------------------------------------------------------- // Rle for Boolean encode/decode tests. From 5f91b9b3b14155158196e69aaea6784407a4a3ba Mon Sep 17 00:00:00 2001 From: Prateek Gaur Date: Fri, 18 Sep 2026 19:54:32 +0000 Subject: [PATCH 2/2] [C++][Parquet] Scan DELTA_BINARY_PACKED deltas a vector at a time The value-at-a-time prefix sum becomes a helper that scans whole registers with a log-step inclusive scan and finishes the remainder one value at a time. Adding the frame of reference before the scan turns its running multiple into a term the scan produces rather than a multiply per lane, and the running total is carried between registers in a vector register rather than through a general-purpose one. The xsimd include and the vector loop are guarded on ARROW_HAVE_NEON or ARROW_HAVE_SSE4_2, so a build with neither compiles the scalar loop alone. The vector loop is compiled only where a register holds four or more values, which at the 128-bit baseline vectorizes 32-bit values and leaves 64-bit ones on the value-at-a-time loop. Also adds a decode benchmark on non-decreasing values, the shape this encoding is usually chosen for. Arithmetic is unchanged: every term stays in the unsigned type, so the wrapping the format specifies is preserved and no decoded value changes. --- cpp/src/parquet/decoder.cc | 83 +++++++++++++++++++++++---- cpp/src/parquet/encoding_benchmark.cc | 40 +++++++++++++ cpp/src/parquet/encoding_test.cc | 40 +++++++++++++ 3 files changed, 152 insertions(+), 11 deletions(-) diff --git a/cpp/src/parquet/decoder.cc b/cpp/src/parquet/decoder.cc index f7b2d655da80..9f0ca6474915 100644 --- a/cpp/src/parquet/decoder.cc +++ b/cpp/src/parquet/decoder.cc @@ -30,6 +30,10 @@ #include #include +#if defined(ARROW_HAVE_NEON) || defined(ARROW_HAVE_SSE4_2) +# include +#endif + #include "arrow/array.h" #include "arrow/array/builder_binary.h" #include "arrow/array/builder_dict.h" @@ -1434,6 +1438,71 @@ class DictByteArrayDecoderImpl : public DictDecoderImpl { // ---------------------------------------------------------------------- // DELTA_BINARY_PACKED decoder +namespace { + +#if defined(ARROW_HAVE_NEON) || defined(ARROW_HAVE_SSE4_2) +// One step of an inclusive scan, recursed at compile time over powers of two. Each step +// adds the vector to a copy of itself slid up by kShift lanes, zero-filling the lanes +// it vacates, so after the last step lane k holds the sum of lanes 0 through k. +template +Batch InclusiveScanSteps(Batch v) { + if constexpr (kShift < Batch::size) { + v += xsimd::slide_left(v); + return InclusiveScanSteps(v); + } else { + return v; + } +} +#endif + +// Turns a run of deltas into the values they encode, in place: on return element k holds +// `last + (k + 1) * min_delta + sum of deltas 0..k`, and the return value is the last +// element. Every term is unsigned, so the wrapping matches the loop this replaces. +template +std::make_unsigned_t PrefixSumDeltas(T* values, int num_values, + std::make_unsigned_t min_delta, + std::make_unsigned_t last) { + using UT = std::make_unsigned_t; + int i = 0; + +#if defined(ARROW_HAVE_NEON) || defined(ARROW_HAVE_SSE4_2) + using Batch = xsimd::batch; + constexpr int kLanes = static_cast(Batch::size); + // At two lanes the scan loses to the additions it replaces, so it is compiled only + // where a register holds four or more values; narrower ones use the loop below. + if constexpr (kLanes >= 4) { + // Broadcast pattern for the last lane, which carries the running value into the + // next vector without a round trip through a general-purpose register. + struct LastLane { + static constexpr unsigned get(unsigned /*index*/, unsigned size) { + return size - 1; + } + }; + const auto last_lane = + xsimd::make_batch_constant(); + const Batch min_delta_v(min_delta); + Batch carry(last); + for (; i + kLanes <= num_values; i += kLanes) { + // Adding the frame before the scan turns its running multiple into a term the + // scan produces, rather than a multiply per lane. + Batch v = xsimd::bitwise_cast(xsimd::batch::load_unaligned(values + i)); + v = InclusiveScanSteps<1>(v + min_delta_v) + carry; + xsimd::bitwise_cast(v).store_unaligned(values + i); + carry = xsimd::swizzle(v, last_lane); + } + last = carry.get(0); + } +#endif + + for (; i < num_values; ++i) { + last += min_delta + static_cast(values[i]); + values[i] = static_cast(last); + } + return last; +} + +} // namespace + template class DeltaBitPackDecoder : public TypedDecoderImpl { public: @@ -1692,17 +1761,9 @@ class DeltaBitPackDecoder : public TypedDecoderImpl { values_decode) { ParquetException::EofException(); } - // Held in locals because a store to `buffer` may alias these members, which - // would force a reload of each per value. - UT last = static_cast(last_value_); - const UT min_delta = static_cast(min_delta_); - for (int j = 0; j < values_decode; ++j) { - // Addition between min_delta, packed int and last_value should be treated as - // unsigned addition. Overflow is as expected. - last += min_delta + static_cast(buffer[i + j]); - buffer[i + j] = last; - } - last_value_ = static_cast(last); + last_value_ = static_cast(PrefixSumDeltas(buffer + i, values_decode, + static_cast(min_delta_), + static_cast(last_value_))); } // The last miniblock of a coalesced run becomes the current one, fully drained. mini_block_idx_ += mini_blocks_coalesced; diff --git a/cpp/src/parquet/encoding_benchmark.cc b/cpp/src/parquet/encoding_benchmark.cc index bea1a5807a2a..6bcce1a2c6e6 100644 --- a/cpp/src/parquet/encoding_benchmark.cc +++ b/cpp/src/parquet/encoding_benchmark.cc @@ -675,6 +675,22 @@ static auto MakeDeltaBitPackingInputNarrow(size_t length) { return numbers; } +// Non-decreasing values, the shape this encoding is usually chosen for. The deltas come +// from the same 1000-wide range as Narrow, so the two inputs differ in the order of the +// values and not in the width a delta is packed into. +template +static auto MakeDeltaBitPackingInputNarrowSorted(size_t length) { + using T = typename DType::c_type; + auto numbers = std::vector(length); + ::arrow::randint(length, 0, 1000, &numbers); + T value = 0; + for (auto& number : numbers) { + value = static_cast(value + number); + number = value; + } + return numbers; +} + template static auto MakeDeltaBitPackingInputWide(size_t length) { using T = typename DType::c_type; @@ -713,6 +729,16 @@ static void BM_DeltaBitPackingEncode_Int64_Narrow(benchmark::State& state) { BM_DeltaBitPackingEncode(state, MakeDeltaBitPackingInputNarrow); } +static void BM_DeltaBitPackingEncode_Int32_NarrowSorted(benchmark::State& state) { + BM_DeltaBitPackingEncode(state, + MakeDeltaBitPackingInputNarrowSorted); +} + +static void BM_DeltaBitPackingEncode_Int64_NarrowSorted(benchmark::State& state) { + BM_DeltaBitPackingEncode(state, + MakeDeltaBitPackingInputNarrowSorted); +} + static void BM_DeltaBitPackingEncode_Int32_Wide(benchmark::State& state) { BM_DeltaBitPackingEncode(state, MakeDeltaBitPackingInputWide); } @@ -725,6 +751,8 @@ BENCHMARK(BM_DeltaBitPackingEncode_Int32_Fixed)->Range(MIN_RANGE, MAX_RANGE); BENCHMARK(BM_DeltaBitPackingEncode_Int64_Fixed)->Range(MIN_RANGE, MAX_RANGE); BENCHMARK(BM_DeltaBitPackingEncode_Int32_Narrow)->Range(MIN_RANGE, MAX_RANGE); BENCHMARK(BM_DeltaBitPackingEncode_Int64_Narrow)->Range(MIN_RANGE, MAX_RANGE); +BENCHMARK(BM_DeltaBitPackingEncode_Int32_NarrowSorted)->Range(MIN_RANGE, MAX_RANGE); +BENCHMARK(BM_DeltaBitPackingEncode_Int64_NarrowSorted)->Range(MIN_RANGE, MAX_RANGE); BENCHMARK(BM_DeltaBitPackingEncode_Int32_Wide)->Range(MIN_RANGE, MAX_RANGE); BENCHMARK(BM_DeltaBitPackingEncode_Int64_Wide)->Range(MIN_RANGE, MAX_RANGE); @@ -762,6 +790,16 @@ static void BM_DeltaBitPackingDecode_Int64_Narrow(benchmark::State& state) { BM_DeltaBitPackingDecode(state, MakeDeltaBitPackingInputNarrow); } +static void BM_DeltaBitPackingDecode_Int32_NarrowSorted(benchmark::State& state) { + BM_DeltaBitPackingDecode(state, + MakeDeltaBitPackingInputNarrowSorted); +} + +static void BM_DeltaBitPackingDecode_Int64_NarrowSorted(benchmark::State& state) { + BM_DeltaBitPackingDecode(state, + MakeDeltaBitPackingInputNarrowSorted); +} + static void BM_DeltaBitPackingDecode_Int32_Wide(benchmark::State& state) { BM_DeltaBitPackingDecode(state, MakeDeltaBitPackingInputWide); } @@ -774,6 +812,8 @@ BENCHMARK(BM_DeltaBitPackingDecode_Int32_Fixed)->Range(MIN_RANGE, MAX_RANGE); BENCHMARK(BM_DeltaBitPackingDecode_Int64_Fixed)->Range(MIN_RANGE, MAX_RANGE); BENCHMARK(BM_DeltaBitPackingDecode_Int32_Narrow)->Range(MIN_RANGE, MAX_RANGE); BENCHMARK(BM_DeltaBitPackingDecode_Int64_Narrow)->Range(MIN_RANGE, MAX_RANGE); +BENCHMARK(BM_DeltaBitPackingDecode_Int32_NarrowSorted)->Range(MIN_RANGE, MAX_RANGE); +BENCHMARK(BM_DeltaBitPackingDecode_Int64_NarrowSorted)->Range(MIN_RANGE, MAX_RANGE); BENCHMARK(BM_DeltaBitPackingDecode_Int32_Wide)->Range(MIN_RANGE, MAX_RANGE); BENCHMARK(BM_DeltaBitPackingDecode_Int64_Wide)->Range(MIN_RANGE, MAX_RANGE); diff --git a/cpp/src/parquet/encoding_test.cc b/cpp/src/parquet/encoding_test.cc index 878b23c93072..752a2f2a90a6 100644 --- a/cpp/src/parquet/encoding_test.cc +++ b/cpp/src/parquet/encoding_test.cc @@ -23,6 +23,7 @@ #include #include #include +#include #include #include @@ -2101,6 +2102,45 @@ TYPED_TEST(TestDeltaBitPackEncoding, MiniblockBitWidthRuns) { } } +TYPED_TEST(TestDeltaBitPackEncoding, PrefixSumVectorAndTail) { + // A decoder may accumulate the running total several deltas at a time and finish the + // remainder one at a time. Walk every residual bit width at enough lengths to leave + // every remainder such a group can leave, so each width is decoded through the + // grouped path, through the remainder, and across the hand-off between them. A + // non-zero frame checks the running multiple, and wrapping the total checks that no + // term is signed. + using T = typename TypeParam::c_type; + using UT = std::make_unsigned_t; + constexpr int kBits = static_cast(sizeof(T) * 8); + + auto make_values = [](int width, T frame, int num_deltas) { + std::vector values; + values.reserve(num_deltas + 1); + const UT spread = width == kBits ? ~UT{0} : static_cast((UT{1} << width) - 1); + // Two deltas in three sit at the frame, so it is the smallest in every miniblock. + UT current = 0; + values.push_back(static_cast(current)); + for (int i = 0; i < num_deltas; ++i) { + current = static_cast(current + static_cast(frame) + + (i % 3 == 0 ? spread : UT{0})); + values.push_back(static_cast(current)); + } + return values; + }; + + for (int width = 0; width <= kBits; ++width) { + for (const T frame : {T{0}, static_cast(-5), T{7}}) { + // 16-23 leaves every remainder for group sizes up to eight; 201 crosses a block + // boundary with a remainder left. + for (const int num_deltas : {16, 17, 18, 19, 20, 21, 22, 23, 201}) { + ARROW_SCOPED_TRACE("width = ", width, ", frame = ", static_cast(frame), + ", num_deltas = ", num_deltas); + this->CheckRoundtripWithValues(make_values(width, frame, num_deltas)); + } + } + } +} + // ---------------------------------------------------------------------- // Rle for Boolean encode/decode tests.