diff --git a/cpp/src/parquet/decoder.cc b/cpp/src/parquet/decoder.cc index c4d3fe5a8a5a..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: @@ -1601,6 +1670,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 +1733,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 +1761,13 @@ 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_))); } - 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_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 831829e4a210..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 @@ -1752,7 +1753,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 +2036,111 @@ 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)); + } + } +} + +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.