From 91748975e55592fd30b7d89e853ea2a291ce97fc Mon Sep 17 00:00:00 2001 From: Prateek Gaur Date: Thu, 10 Sep 2026 17:45:56 +0000 Subject: [PATCH 1/3] [C++][Parquet] Unpack equal-width DELTA_BINARY_PACKED miniblocks in one call The miniblocks of a DELTA_BINARY_PACKED block are packed back to back with no padding between them, so a run of miniblocks that share a bit width is bit-identical to a single longer run at that width. GetInternal called the bit unpacker once per miniblock all the same, which with the default geometry is one call per 32 values - mostly per-call setup. Look ahead over the block's stored bit widths and extend the current call over each following miniblock that has the same width and that the caller has room for in full. A miniblock joins the run only when its width equals the current delta_bit_width_, which InitMiniBlock has already validated, and the run also stops at the end of the block. Add a test over the width patterns that decide where a run starts and stops, and read at a batch size that stops partway through a coalesced run. On the decode benchmarks already in the tree this is 1.17x to 1.33x on top of the previous commit wherever the unpacker's per-call cost is a meaningful share of the work. Decoded values are identical; no encoded byte changes. --- cpp/src/parquet/decoder.cc | 51 ++++++++++++++++++-- cpp/src/parquet/encoding_test.cc | 82 +++++++++++++++++++++++++++++++- 2 files changed, 129 insertions(+), 4 deletions(-) diff --git a/cpp/src/parquet/decoder.cc b/cpp/src/parquet/decoder.cc index c4d3fe5a8a5a..2fa25554273b 100644 --- a/cpp/src/parquet/decoder.cc +++ b/cpp/src/parquet/decoder.cc @@ -1601,6 +1601,33 @@ class DeltaBitPackDecoder : public TypedDecoderImpl { values_remaining_current_mini_block_ = values_per_mini_block_; } + // The miniblocks of a block are packed back to back with no padding between them, + // so a run of miniblocks that share a bit width is bit-identical to a single + // longer run at that width, and can be unpacked in one call. Returns how many + // whole miniblocks following the current one may be folded into it, given how many + // more values the caller has room for. + // + // A miniblock joins the run only when its bit width equals delta_bit_width_, which + // InitMiniBlock has already validated. Coalescing therefore never depends on a + // width that has not been checked, including the non-conformant widths InitBlock + // tolerates for extraneous miniblocks. + uint32_t CoalescibleMiniBlocks(uint32_t values_available) const { + // Folding in a whole miniblock first requires room for the current one in full. + 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 +1669,22 @@ 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 default geometry is 32 values per miniblock, and a call that small is + // mostly per-call setup for the unpacker; folding a run of four into one call + // asks it for 128 values instead. A zero bit width decodes without asking the + // unpacker at all, so there is no call to fold and nothing to gain. + const uint32_t mini_blocks_coalesced = + delta_bit_width_ == 0 ? 0 : CoalescibleMiniBlocks(values_available); + // A miniblock is only folded in when there is room for the current one in + // full, so a non-empty run means this call drains the current miniblock along + // with every miniblock folded into it. The accounting below relies on that. + 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 @@ -1666,7 +1707,11 @@ class DeltaBitPackDecoder : public TypedDecoderImpl { last_value_ = buffer[i + j]; } } - values_remaining_current_mini_block_ -= values_decode; + // A coalesced call drained the miniblocks it folded in, so advance the block's + // cursor past them: the last miniblock of the run becomes the current one, + // with nothing left in it. + 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..39a5b6ecac98 100644 --- a/cpp/src/parquet/encoding_test.cc +++ b/cpp/src/parquet/encoding_test.cc @@ -1752,7 +1752,10 @@ 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}; + // 1 and 11 stop partway through a miniblock; 100 spans several of them but still + // ends inside one, so a decoder that unpacks whole miniblocks at a time has to + // both use and give up that shortcut within a single read. + const std::vector kReadBatchSizes = {1, 11, 100}; void InitBoundData(int nvalues, int repeats, c_type half_range) { num_values_ = nvalues * repeats; @@ -2034,6 +2037,83 @@ TYPED_TEST(TestDeltaBitPackEncoding, ZeroDeltaBitWidth) { this->CheckRoundtripWithValues(int_values); } +TYPED_TEST(TestDeltaBitPackEncoding, MiniblockBitWidthRuns) { + // The miniblocks of a block are packed back to back with no padding between + // them, so a run of miniblocks sharing a bit width is bit-identical to one + // longer run at that width and a decoder may unpack the whole run in a single + // call. Cover the patterns that decide where such a run starts and stops: a + // block whose miniblocks all share a width, one where none of them do, runs + // that end partway through a block or at its boundary, and zero-width + // miniblocks, which the closed form decodes and which must not join a run. + 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; + + // Produce values whose deltas give miniblock i the bit width widths[i]. Each + // miniblock alternates a delta of `frame` with a delta of `frame + 2^(w-1)`, + // and 2^(w-1) is the smallest value needing w bits, so the encoder stores width + // w for that miniblock. Every delta in the block is at least `frame`, which + // therefore becomes the frame the encoder stores; passing a negative one + // exercises a frame the decoder has to sign-extend. + 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 leaves the last block partly filled, so a + // run has to stop at the end of the encoded values rather than at a change of + // width. + 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 either side of a block boundary, where a run must still stop + // because the bit widths belong to their own block. + {"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 6df0db34faf2a82fa92b287cc3fa3f4c3dabfb2e Mon Sep 17 00:00:00 2001 From: Prateek Gaur Date: Wed, 9 Sep 2026 20:18:37 +0000 Subject: [PATCH 2/3] [C++][Parquet] Benchmark DELTA_BINARY_PACKED on non-decreasing values The two non-degenerate DELTA_BINARY_PACKED decode arms both hold values in random order, so their deltas are random and the frame sits well below zero. The columns this encoding is chosen for are usually non-decreasing, where every delta is non-negative instead and the frame is at or near zero. Add an arm whose deltas come from the same 1000-wide range as the existing narrow one but accumulate, so the two differ in the order of the values at a nearly unchanged packed width. --- cpp/src/parquet/encoding_benchmark.cc | 43 +++++++++++++++++++++++++++ 1 file changed, 43 insertions(+) diff --git a/cpp/src/parquet/encoding_benchmark.cc b/cpp/src/parquet/encoding_benchmark.cc index bea1a5807a2a..6f08317fe01e 100644 --- a/cpp/src/parquet/encoding_benchmark.cc +++ b/cpp/src/parquet/encoding_benchmark.cc @@ -675,6 +675,25 @@ static auto MakeDeltaBitPackingInputNarrow(size_t length) { return numbers; } +// The Narrow input above is in random order, so its deltas are random too. The +// columns this encoding is chosen for -- identifiers, timestamps, offsets -- are +// usually non-decreasing instead, which makes every delta non-negative and puts the +// frame at or near zero. The deltas here are drawn from the same 1000-wide range as +// Narrow, so the two arms differ in the order of the values and not in the number of +// bits 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 +732,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 +754,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 +793,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 +815,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); From 4c8423dfa2a35cb8c42522959ca6fa601357d82a Mon Sep 17 00:00:00 2001 From: Prateek Gaur Date: Wed, 9 Sep 2026 19:43:05 +0000 Subject: [PATCH 3/3] [C++][Parquet] Scan DELTA_BINARY_PACKED deltas a vector at a time The prefix sum that turns deltas back into values does one addition per value, and each one waits on the value before it, so the loop is bound by that chain rather than by how much arithmetic the machine can retire. Replace it with an inclusive scan, which shifts and adds a vector to itself once per power of two and so turns a vector of deltas into a vector of running sums. That shortens the chain from one addition per value to one per vector. The frame is added before the scan, which makes its running multiple fall out of the scan itself, and the previous vector's last value is carried forward in a vector register -- reading it out into a general-purpose register instead costs several times what the scan saves. The scan runs only where a register holds at least four values, the narrowest width measured to win; below that, and on the tail, the value-at-a-time loop does the work. xsimd is already a dependency of this target, so nothing is added to the build. Decode throughput over the previous commit, on the arms whose values are 32-bit: 1.26x on non-decreasing values, 1.24x in random order, and 1.23x at 31 bits per delta. The 64-bit arms hold two values in a register here and are unchanged. Together with the two commits before it this is 2.17x, 2.11x and 1.89x over main, or 2.64 to 5.73 GB/s of decoded output on non-decreasing values. --- cpp/src/parquet/decoder.cc | 82 +++++++++++++++++++++++++++++--- cpp/src/parquet/encoding_test.cc | 47 ++++++++++++++++++ 2 files changed, 122 insertions(+), 7 deletions(-) diff --git a/cpp/src/parquet/decoder.cc b/cpp/src/parquet/decoder.cc index 2fa25554273b..a8f673111b5f 100644 --- a/cpp/src/parquet/decoder.cc +++ b/cpp/src/parquet/decoder.cc @@ -30,6 +30,8 @@ #include #include +#include + #include "arrow/array.h" #include "arrow/array/builder_binary.h" #include "arrow/array/builder_dict.h" @@ -1434,6 +1436,76 @@ class DictByteArrayDecoderImpl : public DictDecoderImpl { // ---------------------------------------------------------------------- // DELTA_BINARY_PACKED decoder +namespace { + +// One doubling step of an inclusive scan per power of two, unrolled at compile time. +// Each step shifts the vector up by kShift lanes, filling with zero, and adds it to +// itself, so after the last step lane k holds the sum of lanes 0 through k. That is +// log2(lanes) additions in place of one per lane, and more importantly it replaces a +// dependency chain as long as the run with one as long as the number of vectors. +template +Batch InclusiveScanSteps(Batch v) { + if constexpr (kShift < Batch::size) { + v += xsimd::slide_left(v); + return InclusiveScanSteps(v); + } else { + return v; + } +} + +// Turns a run of deltas in place into the values they encode: on return element k +// holds `last + (k + 1) * min_delta + sum of deltas 0..k`. Every term is unsigned, +// so wrapping is well defined and matches what a value-at-a-time loop gives. +// +// The frame is added before the scan, which makes its running multiple fall out of +// the scan itself, and the previous vector's last value is carried in after. +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; + using Batch = xsimd::batch; + constexpr int kLanes = static_cast(Batch::size); + + int i = 0; + // A vector scan only pays off once a register holds enough values to beat the + // chain of additions it replaces. At two lanes it loses: one doubling step plus + // carrying the running value across vectors costs more than the two additions it + // saves. Four is the narrowest width measured to win, so that is the threshold; + // where a register holds fewer, this loop is dropped and the one below does all + // the work. + if constexpr (kLanes >= 4) { + // Broadcasts the last lane, which carries the running value into the next + // vector without a round trip through a general-purpose register. Reading the + // lane out into one instead costs several times what the scan saves. + 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) { + // The output buffer's element type is signed. The arithmetic is the same + // either way, but has to be spelled unsigned for the overflow to be defined. + 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); + } + 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: @@ -1699,13 +1771,9 @@ class DeltaBitPackDecoder : public TypedDecoderImpl { values_decode) { ParquetException::EofException(); } - 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_value_ = static_cast(PrefixSumDeltas(buffer + i, values_decode, + static_cast(min_delta_), + static_cast(last_value_))); } // A coalesced call drained the miniblocks it folded in, so advance the block's // cursor past them: the last miniblock of the run becomes the current one, diff --git a/cpp/src/parquet/encoding_test.cc b/cpp/src/parquet/encoding_test.cc index 39a5b6ecac98..7424fa8d196c 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 @@ -2114,6 +2115,52 @@ TYPED_TEST(TestDeltaBitPackEncoding, MiniblockBitWidthRuns) { } } +TYPED_TEST(TestDeltaBitPackEncoding, PrefixSumVectorAndTail) { + // A decoder may accumulate the running total several deltas at a time, finishing + // whatever does not fill a whole group one delta at a time and carrying the total + // from each group into the next. Walk every residual bit width, and at each one + // 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. + // + // Deltas alternate between the frame and the widest value the width can hold above + // it. That pins the stored width, keeps a non-zero frame in play -- whose running + // multiple grows with the index, so a decoder that folds it in per group has to get + // that multiple right -- and wraps the running total repeatedly, which is where a + // grouped total and a value-at-a-time one part company if any 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 every three sit at the frame, so the frame is the smallest delta + // in every miniblock and the stored residuals are 0 and `spread`. + 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 through 23 leaves every remainder for any group size up to eight; 201 runs + // long enough to cross miniblock and block boundaries 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.