Add the DELTA_LENGTH_BYTE_ARRAY nz map consumer

cpp/benchmarks/io/parquet/parquet_reader_encoding.cpp

@@ -121,11 +121,17 @@ void bench_read_encoding(nvbench::state& state,
   auto const nesting     = use_nullable_page_size_matrix
                              ? cudf::size_type{0}
                              : static_cast<cudf::size_type>(state.get_int64("nesting"));
-  auto const validity =
-    use_nullable_page_size_matrix ? state.get_string("validity") : "no_validity";
-  auto const page_rows = use_nullable_page_size_matrix
-                           ? static_cast<cudf::size_type>(state.get_int64("page_rows"))
-                           : cudf::size_type{0};
+  auto const validity    = state.get_string("validity");
+  auto const page_rows   = use_nullable_page_size_matrix
+                             ? static_cast<cudf::size_type>(state.get_int64("page_rows"))
+                             : cudf::size_type{0};
+  // `create_nested_table` builds its own profile with the default 1% null probability, so run the
+  // nested configuration once, under the `validity` value that describes it.
+  if (nesting > 0 and validity != "nullable_1") {
+    state.skip("validity does not apply to the nested table builder");
+    return;
+  }
+
   cuio_source_sink_pair source_sink(source_type);
 
   auto const num_rows_written = [&]() {
@@ -185,6 +191,7 @@ void BM_parquet_read_delta_string_nullable_page_sizes(nvbench::state& state)
 
 NVBENCH_BENCH(BM_parquet_read_delta_binary)
   .set_name("parquet_read_delta_binary")
+  .add_string_axis("validity", {"no_validity", "nullable_1", "nullable_50"})
   .add_string_axis("encoding", {"PLAIN", "DELTA_BINARY_PACKED"})
   .add_string_axis("io_type", {"DEVICE_BUFFER"})
   .set_min_samples(4)
@@ -195,6 +202,7 @@ NVBENCH_BENCH(BM_parquet_read_delta_binary)
 
 NVBENCH_BENCH(BM_parquet_read_delta_string)
   .set_name("parquet_read_delta_string")
+  .add_string_axis("validity", {"no_validity", "nullable_1", "nullable_50"})
   .add_string_axis("encoding", {"PLAIN", "DELTA_LENGTH_BYTE_ARRAY", "DELTA_BYTE_ARRAY"})
   .add_string_axis("io_type", {"DEVICE_BUFFER"})
   .set_min_samples(4)

cpp/src/io/parquet/delta_binary.cuh

@@ -9,6 +9,7 @@
 
 #include <cooperative_groups/reduce.h>
 #include <cooperative_groups/scan.h>
+#include <cuda/std/limits>
 
 #include <climits>
 
@@ -176,7 +177,14 @@ struct delta_binary_decoder {
     // Validate the header against the DELTA_BINARY_PACKED spec: the mini-block count must evenly
     // divide the block size, and each mini-block must hold a multiple of 32 values. The decoders
     // rely on the latter to advance from one mini-block to the next.
-    if (mini_block_count == 0 or block_size == 0 or (block_size % mini_block_count) != 0 or
+    //
+    // `block_size` and `value_count` are checked against `max_count` because the decoder steps
+    // `current_value_idx` (32 bits) by `values_per_mb` until it reaches `value_count`. With both at
+    // most INT32_MAX that sum cannot wrap, so `find_end_of_block` always finishes. Parquet stores a
+    // page's value count as int32, so a valid page never exceeds it.
+    constexpr uleb128_t max_count = cuda::std::numeric_limits<int32_t>::max();
+    if (mini_block_count == 0 or block_size == 0 or block_size > max_count or
+        value_count > max_count or (block_size % mini_block_count) != 0 or
         ((block_size / mini_block_count) % delta_mini_block_size_multiple) != 0) {
       error         = true;
       value_count   = 0;
@@ -197,6 +205,13 @@ struct delta_binary_decoder {
     if (value_count > 1) { init_mini_block(false); }
   }
 
+  // Bytes taken by `num_values` values packed at `bits` each. Computed in 64 bits because `bits`
+  // comes from the page unchecked, so the product can overflow 32.
+  [[nodiscard]] static inline __device__ uint64_t packed_bytes(uint32_t bits, uint32_t num_values)
+  {
+    return static_cast<uint64_t>(bits) * num_values / CHAR_BIT;
+  }
+
   // skip to the start of the next mini-block. should only be called on thread 0.
   // calls init_binary_block if currently on the last mini-block in a block.
   // is_decode indicates whether this is being called from initialization code (false) or
@@ -209,12 +224,12 @@ struct delta_binary_decoder {
 
     // just set pointer to start of next mini_block
     if (cur_mb < mini_block_count - 1) {
-      cur_mb_start += cur_bitwidths[cur_mb] * values_per_mb / CHAR_BIT;
+      cur_mb_start += packed_bytes(cur_bitwidths[cur_mb], values_per_mb);
       cur_mb++;
     }
     // out of mini-blocks, start a new block
     else {
-      block_start = cur_mb_start + cur_bitwidths[cur_mb] * values_per_mb / CHAR_BIT;
+      block_start = cur_mb_start + packed_bytes(cur_bitwidths[cur_mb], values_per_mb);
       init_mini_block(is_decode);
     }
   }
@@ -271,7 +286,7 @@ struct delta_binary_decoder {
     uint32_t const mb_bits = cur_bitwidths[cur_mb];
 
     // position at the end of this pass's values since the following calculates negative indexes
-    auto const d_start = cur_mb_start + (pass + 1) * (warp_size * mb_bits / CHAR_BIT);
+    auto const d_start = cur_mb_start + packed_bytes(mb_bits, (pass + 1) * warp_size);
 
     // unpack deltas. modified from version in decode_dictionary_indices(), but
     // that one only unpacks up to bitwidths of 24. simplified some since this