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); }
   }
 
+  // Byte length of one mini-block at `bits` per value. Computed in 64 bits because `bits` comes
+  // from the page unchecked, so `bits * values_per_mb` can overflow 32.
+  [[nodiscard]] inline __device__ uint64_t mini_block_bytes(uint32_t bits) const
+  {
+    return static_cast<uint64_t>(bits) * values_per_mb / 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 += mini_block_bytes(cur_bitwidths[cur_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 + mini_block_bytes(cur_bitwidths[cur_mb]);
       init_mini_block(is_decode);
     }
   }