GH-51268: [C++][Parquet] Scan DELTA_BINARY_PACKED deltas a vector at a time

cpp/src/parquet/decoder.cc

@@ -30,6 +30,10 @@
 #include <utility>
 #include <vector>
 
+#if defined(ARROW_HAVE_NEON) || defined(ARROW_HAVE_SSE4_2)
+#  include <xsimd/xsimd.hpp>
+#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<ByteArrayType> {
 // ----------------------------------------------------------------------
 // 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 <std::size_t kShift, typename Batch>
+Batch InclusiveScanSteps(Batch v) {
+  if constexpr (kShift < Batch::size) {
+    v += xsimd::slide_left<kShift * sizeof(typename Batch::value_type)>(v);
+    return InclusiveScanSteps<kShift * 2>(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 <typename T>
+std::make_unsigned_t<T> PrefixSumDeltas(T* values, int num_values,
+                                        std::make_unsigned_t<T> min_delta,
+                                        std::make_unsigned_t<T> last) {
+  using UT = std::make_unsigned_t<T>;
+  int i = 0;
+
+#if defined(ARROW_HAVE_NEON) || defined(ARROW_HAVE_SSE4_2)
+  using Batch = xsimd::batch<UT>;
+  constexpr int kLanes = static_cast<int>(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<UT, LastLane, xsimd::default_arch>();
+    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<UT>(xsimd::batch<T>::load_unaligned(values + i));
+      v = InclusiveScanSteps<1>(v + min_delta_v) + carry;
+      xsimd::bitwise_cast<T>(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<UT>(values[i]);
+    values[i] = static_cast<T>(last);
+  }
+  return last;
+}
+
+}  // namespace
+
 template <typename DType>
 class DeltaBitPackDecoder : public TypedDecoderImpl<DType> {
  public:
@@ -1601,6 +1670,28 @@ class DeltaBitPackDecoder : public TypedDecoderImpl<DType> {
     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<int>(std::min<int64_t>(max_values, total_values_remaining_));
     if (max_values == 0) {
@@ -1642,8 +1733,20 @@ class DeltaBitPackDecoder : public TypedDecoderImpl<DType> {
         }
       }
 
-      int values_decode = std::min(values_remaining_current_mini_block_,
-                                   static_cast<uint32_t>(max_values - i));
+      const uint32_t values_available = static_cast<uint32_t>(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<int>(
+          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<DType> {
             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<UT>(min_delta_) + static_cast<UT>(buffer[i + j]) +
-                          static_cast<UT>(last_value_);
-          last_value_ = buffer[i + j];
-        }
+        last_value_ = static_cast<T>(PrefixSumDeltas(buffer + i, values_decode,
+                                                     static_cast<UT>(min_delta_),
+                                                     static_cast<UT>(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;

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 <typename DType>
+static auto MakeDeltaBitPackingInputNarrowSorted(size_t length) {
+  using T = typename DType::c_type;
+  auto numbers = std::vector<T>(length);
+  ::arrow::randint<T, T>(length, 0, 1000, &numbers);
+  T value = 0;
+  for (auto& number : numbers) {
+    value = static_cast<T>(value + number);
+    number = value;
+  }
+  return numbers;
+}
+
 template <typename DType>
 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<Int64Type>(state, MakeDeltaBitPackingInputNarrow<Int64Type>);
 }
 
+static void BM_DeltaBitPackingEncode_Int32_NarrowSorted(benchmark::State& state) {
+  BM_DeltaBitPackingEncode<Int32Type>(state,
+                                      MakeDeltaBitPackingInputNarrowSorted<Int32Type>);
+}
+
+static void BM_DeltaBitPackingEncode_Int64_NarrowSorted(benchmark::State& state) {
+  BM_DeltaBitPackingEncode<Int64Type>(state,
+                                      MakeDeltaBitPackingInputNarrowSorted<Int64Type>);
+}
+
 static void BM_DeltaBitPackingEncode_Int32_Wide(benchmark::State& state) {
   BM_DeltaBitPackingEncode<Int32Type>(state, MakeDeltaBitPackingInputWide<Int32Type>);
 }
@@ -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<Int64Type>(state, MakeDeltaBitPackingInputNarrow<Int64Type>);
 }
 
+static void BM_DeltaBitPackingDecode_Int32_NarrowSorted(benchmark::State& state) {
+  BM_DeltaBitPackingDecode<Int32Type>(state,
+                                      MakeDeltaBitPackingInputNarrowSorted<Int32Type>);
+}
+
+static void BM_DeltaBitPackingDecode_Int64_NarrowSorted(benchmark::State& state) {
+  BM_DeltaBitPackingDecode<Int64Type>(state,
+                                      MakeDeltaBitPackingInputNarrowSorted<Int64Type>);
+}
+
 static void BM_DeltaBitPackingDecode_Int32_Wide(benchmark::State& state) {
   BM_DeltaBitPackingDecode<Int32Type>(state, MakeDeltaBitPackingInputWide<Int32Type>);
 }
@@ -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);
 

cpp/src/parquet/encoding_test.cc

@@ -23,6 +23,7 @@
 #include <functional>
 #include <limits>
 #include <span>
+#include <type_traits>
 #include <utility>
 #include <vector>
 
@@ -1752,7 +1753,8 @@ class TestDeltaBitPackEncoding : public TestEncodingBase<Type> {
   using c_type = typename Type::c_type;
   static constexpr int TYPE = Type::type_num;
   static constexpr size_t kNumRoundTrips = 3;
-  const std::vector<int> kReadBatchSizes = {1, 11};
+  // 100 spans several miniblocks but still ends inside one.
+  const std::vector<int> 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<int32_t, typename TypeParam::c_type> ? 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<int>& widths, T frame, int trailing_values) {
+    std::vector<T> 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>(T{1} << (width - 1));
+      for (int i = 0; i < kValuesPerMiniBlock; ++i) {
+        current = static_cast<T>(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<T>(current + frame);
+      values.push_back(current);
+    }
+    return values;
+  };
+
+  struct Case {
+    const char* name;
+    std::vector<int> widths;
+    int trailing_values;
+  };
+  const std::vector<Case> 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<T>(-5)}) {
+      ARROW_SCOPED_TRACE("case = ", c.name, ", frame = ", static_cast<int64_t>(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<T>;
+  constexpr int kBits = static_cast<int>(sizeof(T) * 8);
+
+  auto make_values = [](int width, T frame, int num_deltas) {
+    std::vector<T> values;
+    values.reserve(num_deltas + 1);
+    const UT spread = width == kBits ? ~UT{0} : static_cast<UT>((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<T>(current));
+    for (int i = 0; i < num_deltas; ++i) {
+      current = static_cast<UT>(current + static_cast<UT>(frame) +
+                                (i % 3 == 0 ? spread : UT{0}));
+      values.push_back(static_cast<T>(current));
+    }
+    return values;
+  };
+
+  for (int width = 0; width <= kBits; ++width) {
+    for (const T frame : {T{0}, static_cast<T>(-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<int64_t>(frame),
+                           ", num_deltas = ", num_deltas);
+        this->CheckRoundtripWithValues(make_values(width, frame, num_deltas));
+      }
+    }
+  }
+}
+
 // ----------------------------------------------------------------------
 // Rle for Boolean encode/decode tests.