apacheGH-43891: [C++][Parquet] Faster reading of FIXED_LEN_BYTE_ARRAY data

Improve the performance of reading FIXED_LEN_BYTE_ARRAY columns to Arrow, by avoiding an intermediate read to FLBA structures.

This especially helps improve the speed of reading FLOAT16 columns and makes it faster than FLOAT.

Author: pitrou

cpp/src/arrow/array/builder_binary.h

@@ -855,11 +855,41 @@ class ARROW_EXPORT FixedSizeBinaryBuilder : public ArrayBuilder {
   /// This pointer becomes invalid on the next modifying operation.
   const uint8_t* GetValue(int64_t i) const;
 
-  /// Temporary access to a value.
+  /// Temporary mutable access to a value.
+  ///
+  /// This pointer becomes invalid on the next modifying operation.
+  uint8_t* GetMutableValue(int64_t i) {
+    uint8_t* data_ptr = byte_builder_.mutable_data();
+    return data_ptr + i * byte_width_;
+  }
+
+  /// Temporary mutable access to a value.
   ///
   /// This view becomes invalid on the next modifying operation.
   std::string_view GetView(int64_t i) const;
 
+  /// Advance builder without allocating nor writing any values
+  ///
+  /// The internal pointer is advanced by `length` values and the same number
+  /// of non-null entries are appended to the validity bitmap.
+  /// This method assumes that the `length` values were populated directly,
+  /// for example using `GetMutableValue`.
+  void UnsafeAdvance(int64_t length) {
+    byte_builder_.UnsafeAdvance(length * byte_width_);
+    UnsafeAppendToBitmap(length, true);
+  }
+
+  /// Advance builder without allocating nor writing any values
+  ///
+  /// The internal pointer is advanced by `length` values and the same number
+  /// of validity bits are appended to the validity bitmap.
+  /// This method assumes that the `length` values were populated directly,
+  /// for example using `GetMutableValue`.
+  void UnsafeAdvance(int64_t length, const uint8_t* validity, int64_t valid_bits_offset) {
+    byte_builder_.UnsafeAdvance(length * byte_width_);
+    UnsafeAppendToBitmap(validity, valid_bits_offset, length);
+  }
+
   static constexpr int64_t memory_limit() {
     return std::numeric_limits<int64_t>::max() - 1;
   }
@@ -872,14 +902,6 @@ class ARROW_EXPORT FixedSizeBinaryBuilder : public ArrayBuilder {
   int32_t byte_width_;
   BufferBuilder byte_builder_;
 
-  /// Temporary access to a value.
-  ///
-  /// This pointer becomes invalid on the next modifying operation.
-  uint8_t* GetMutableValue(int64_t i) {
-    uint8_t* data_ptr = byte_builder_.mutable_data();
-    return data_ptr + i * byte_width_;
-  }
-
   void CheckValueSize(int64_t size);
 };
 

cpp/src/arrow/array/builder_primitive.h

@@ -141,6 +141,10 @@ class NumericBuilder
 
   value_type GetValue(int64_t index) const { return data_builder_.data()[index]; }
 
+  value_type* GetMutableValue(int64_t index) {
+    return &data_builder_.mutable_data()[index];
+  }
+
   void Reset() override {
     data_builder_.Reset();
     ArrayBuilder::Reset();
@@ -319,6 +323,28 @@ class NumericBuilder
     data_builder_.UnsafeAppend(value_type{});  // zero
   }
 
+  /// Advance builder without allocating nor writing any values
+  ///
+  /// The internal pointer is advanced by `length` values and the same number
+  /// of non-null entries are appended to the validity bitmap.
+  /// This method assumes that the `length` values were populated directly,
+  /// for example using `GetMutableValue`.
+  void UnsafeAdvance(int64_t length) {
+    data_builder_.UnsafeAdvance(length);
+    UnsafeAppendToBitmap(length, true);
+  }
+
+  /// Advance builder without allocating nor writing any values
+  ///
+  /// The internal pointer is advanced by `length` values and the same number
+  /// of validity bits are appended to the validity bitmap.
+  /// This method assumes that the `length` values were populated directly,
+  /// for example using `GetMutableValue`.
+  void UnsafeAdvance(int64_t length, const uint8_t* validity, int64_t valid_bits_offset) {
+    data_builder_.UnsafeAdvance(length);
+    UnsafeAppendToBitmap(validity, valid_bits_offset, length);
+  }
+
   std::shared_ptr<DataType> type() const override { return type_; }
 
  protected:

cpp/src/arrow/buffer_builder.h

@@ -295,6 +295,10 @@ class TypedBufferBuilder<
     return bytes_builder_.Advance(length * sizeof(T));
   }
 
+  void UnsafeAdvance(const int64_t length) {
+    bytes_builder_.UnsafeAdvance(length * sizeof(T));
+  }
+
   Status Finish(std::shared_ptr<Buffer>* out, bool shrink_to_fit = true) {
     return bytes_builder_.Finish(out, shrink_to_fit);
   }

cpp/src/arrow/util/spaced_internal.h

@@ -23,9 +23,7 @@
 
 #include "arrow/util/bit_run_reader.h"
 
-namespace arrow {
-namespace util {
-namespace internal {
+namespace arrow::util::internal {
 
 /// \brief Compress the buffer to spaced, excluding the null entries.
 ///
@@ -53,27 +51,29 @@ inline int SpacedCompress(const T* src, int num_values, const uint8_t* valid_bit
   return num_valid_values;
 }
 
-/// \brief Relocate values in buffer into positions of non-null values as indicated by
-/// a validity bitmap.
+/// \brief Relocate values according to a validity bitmap, to the right
+///
+/// Non-null values should initially be densely packed at the left of the buffer.
+/// This method spreads the values out according to the given validity bitmap.
+/// Null entries are zero-initialized.
 ///
 /// \param[in, out] buffer the in-place buffer
 /// \param[in] num_values total size of buffer including null slots
 /// \param[in] null_count number of null slots
 /// \param[in] valid_bits bitmap data indicating position of valid slots
 /// \param[in] valid_bits_offset offset into valid_bits
-/// \return The number of values expanded, including nulls.
 template <typename T>
-inline int SpacedExpand(T* buffer, int num_values, int null_count,
-                        const uint8_t* valid_bits, int64_t valid_bits_offset) {
+inline void SpacedExpandRightward(T* buffer, int num_values, int null_count,
+                                  const uint8_t* valid_bits, int64_t valid_bits_offset) {
   // Point to end as we add the spacing from the back.
   int idx_decode = num_values - null_count;
 
   // Depending on the number of nulls, some of the value slots in buffer may
   // be uninitialized, and this will cause valgrind warnings / potentially UB
-  std::memset(static_cast<void*>(buffer + idx_decode), 0, null_count * sizeof(T));
+  memset(static_cast<void*>(buffer + idx_decode), 0, null_count * sizeof(T));
   if (idx_decode == 0) {
     // All nulls, nothing more to do
-    return num_values;
+    return;
   }
 
   arrow::internal::ReverseSetBitRunReader reader(valid_bits, valid_bits_offset,
@@ -85,14 +85,60 @@ inline int SpacedExpand(T* buffer, int num_values, int null_count,
     }
     idx_decode -= static_cast<int32_t>(run.length);
     assert(idx_decode >= 0);
-    std::memmove(buffer + run.position, buffer + idx_decode, run.length * sizeof(T));
+    if (idx_decode == run.position) {
+      // We have come to the point where no more expansion is required: the remaining
+      // values are already in their final position.
+      return;
+    }
+    // Source and destination may overlap if run.length > 1
+    memmove(buffer + run.position, buffer + idx_decode, run.length * sizeof(T));
   }
 
   // Otherwise caller gave an incorrect null_count
   assert(idx_decode == 0);
-  return num_values;
 }
 
-}  // namespace internal
-}  // namespace util
-}  // namespace arrow
+/// \brief Relocate values according to a validity bitmap, to the left
+///
+/// Non-null values should initially be densely packed at the right of the buffer.
+/// This method spreads the values out according to the given validity bitmap.
+/// Null entries are zero-initialized.
+///
+/// \param[in, out] buffer the in-place buffer
+/// \param[in] byte_width the byte width of values
+/// \param[in] length total length of buffer including null slots
+/// \param[in] null_count number of null slots
+/// \param[in] valid_bits bitmap data indicating position of valid slots
+/// \param[in] valid_bits_offset offset into valid_bits
+inline void SpacedExpandLeftward(uint8_t* buffer, int byte_width, int64_t length,
+                                 int64_t null_count, const uint8_t* valid_bits,
+                                 int64_t valid_bits_offset) {
+  // Point to start of values.
+  int64_t idx_decode = byte_width * null_count;
+
+  // Depending on the number of nulls, some of the value slots in buffer may
+  // be uninitialized, and this will cause valgrind warnings / potentially UB
+  memset(buffer, 0, idx_decode);
+
+  arrow::internal::SetBitRunReader reader(valid_bits, valid_bits_offset, length);
+  while (true) {
+    const auto run = reader.NextRun();
+    if (run.length == 0) {
+      break;
+    }
+    if (idx_decode == run.position * byte_width) {
+      // We have come to the point where no more expansion is required: the remaining
+      // values are already in their final position.
+      return;
+    }
+    // Source and destination may overlap if run.length > 1
+    memmove(buffer + run.position * byte_width, buffer + idx_decode,
+            run.length * byte_width);
+    idx_decode += run.length * byte_width;
+  }
+
+  // Otherwise caller gave an incorrect null_count
+  assert(idx_decode == length * byte_width);
+}
+
+}  // namespace arrow::util::internal

cpp/src/parquet/arrow/reader_writer_benchmark.cc

@@ -417,22 +417,34 @@ BENCHMARK_TEMPLATE2(BM_ReadColumn, true, BooleanType)
     ->Args({5, 10});
 
 //
-// Benchmark reading a PLAIN-encoded primitive column
+// Benchmark reading a non-dict-encoded primitive column
 //
 
-template <bool nullable, typename ParquetType>
-static void BM_ReadColumnPlain(::benchmark::State& state) {
+template <typename ParquetType>
+static void BenchmarkReadNonDictColumn(::benchmark::State& state, bool nullable,
+                                       Encoding::type encoding) {
   using c_type = typename ArrowType<ParquetType>::c_type;
 
   const std::vector<c_type> values(BENCHMARK_SIZE, static_cast<c_type>(42));
   std::shared_ptr<Table> table =
       TableFromVector<ParquetType>(values, /*nullable=*/nullable, state.range(0));
 
-  auto properties = WriterProperties::Builder().disable_dictionary()->build();
+  auto properties =
+      WriterProperties::Builder().disable_dictionary()->encoding(encoding)->build();
   BenchmarkReadTable(state, *table, properties, table->num_rows(),
                      BytesForItems<ParquetType>(table->num_rows()));
 }
 
+template <bool nullable, typename ParquetType>
+static void BM_ReadColumnPlain(::benchmark::State& state) {
+  BenchmarkReadNonDictColumn<ParquetType>(state, nullable, Encoding::PLAIN);
+}
+
+template <bool nullable, typename ParquetType>
+static void BM_ReadColumnByteStreamSplit(::benchmark::State& state) {
+  BenchmarkReadNonDictColumn<ParquetType>(state, nullable, Encoding::BYTE_STREAM_SPLIT);
+}
+
 BENCHMARK_TEMPLATE2(BM_ReadColumnPlain, false, Int32Type)
     ->ArgNames({"null_probability"})
     ->Args({kAlternatingOrNa});
@@ -455,12 +467,24 @@ BENCHMARK_TEMPLATE2(BM_ReadColumnPlain, true, Float16LogicalType)
     ->Args({99})
     ->Args({100});
 
+BENCHMARK_TEMPLATE2(BM_ReadColumnByteStreamSplit, false, Float16LogicalType)
+    ->ArgNames({"null_probability"})
+    ->Args({kAlternatingOrNa});
+BENCHMARK_TEMPLATE2(BM_ReadColumnByteStreamSplit, true, Float16LogicalType)
+    ->ArgNames({"null_probability"})
+    ->Args({0})
+    ->Args({1})
+    ->Args({50})
+    ->Args({99})
+    ->Args({100});
+
 //
 // Benchmark reading binary column
 //
 
 static void BenchmarkReadBinaryColumn(::benchmark::State& state,
-                                      const std::shared_ptr<::arrow::DataType>& type) {
+                                      const std::shared_ptr<::arrow::DataType>& type,
+                                      Encoding::type encoding) {
   std::shared_ptr<Table> table =
       RandomStringTable(type, BENCHMARK_SIZE, state.range(1), state.range(0));
 
@@ -470,7 +494,9 @@ static void BenchmarkReadBinaryColumn(::benchmark::State& state,
   for (size_t i = 1; i < column.buffers.size(); ++i) {
     total_bytes += column.buffers[i]->size();
   }
-  BenchmarkReadTable(state, *table, table->num_rows(), total_bytes);
+
+  auto properties = WriterProperties::Builder().encoding(encoding)->build();
+  BenchmarkReadTable(state, *table, properties, table->num_rows(), total_bytes);
 }
 
 static void SetReadBinaryColumnArgs(benchmark::internal::Benchmark* b) {
@@ -487,17 +513,41 @@ static void SetReadBinaryColumnArgs(benchmark::internal::Benchmark* b) {
       ->Args({99, kInfiniteUniqueValues});
 }
 
+static void SetReadBinaryColumnArgsWithoutDictEncoding(
+    benchmark::internal::Benchmark* b) {
+  b->ArgNames({"null_probability", "unique_values"})
+      // Dict-encoding is already tested in the PLAIN benchmarks, so only exercise
+      // non-dict-encoding using high cardinality.
+      ->Args({0, kInfiniteUniqueValues})
+      ->Args({1, kInfiniteUniqueValues})
+      ->Args({50, kInfiniteUniqueValues})
+      ->Args({99, kInfiniteUniqueValues});
+}
+
 static void BM_ReadBinaryColumn(::benchmark::State& state) {
-  BenchmarkReadBinaryColumn(state, ::arrow::utf8());
+  BenchmarkReadBinaryColumn(state, ::arrow::utf8(), Encoding::PLAIN);
 }
 
 static void BM_ReadBinaryViewColumn(::benchmark::State& state) {
-  BenchmarkReadBinaryColumn(state, ::arrow::large_utf8());
+  BenchmarkReadBinaryColumn(state, ::arrow::large_utf8(), Encoding::PLAIN);
+}
+
+static void BM_ReadBinaryColumnDeltaByteArray(::benchmark::State& state) {
+  BenchmarkReadBinaryColumn(state, ::arrow::utf8(), Encoding::DELTA_BYTE_ARRAY);
+}
+
+static void BM_ReadBinaryViewColumnDeltaByteArray(::benchmark::State& state) {
+  BenchmarkReadBinaryColumn(state, ::arrow::large_utf8(), Encoding::DELTA_BYTE_ARRAY);
 }
 
 BENCHMARK(BM_ReadBinaryColumn)->Apply(SetReadBinaryColumnArgs);
 BENCHMARK(BM_ReadBinaryViewColumn)->Apply(SetReadBinaryColumnArgs);
 
+BENCHMARK(BM_ReadBinaryColumnDeltaByteArray)
+    ->Apply(SetReadBinaryColumnArgsWithoutDictEncoding);
+BENCHMARK(BM_ReadBinaryViewColumnDeltaByteArray)
+    ->Apply(SetReadBinaryColumnArgsWithoutDictEncoding);
+
 //
 // Benchmark reading a nested column
 //