feat(parquet): add the DELTA_LENGTH_BYTE_ARRAY nz map consumer

Third of the nz map series. Still opt-in behind `LIBCUDF_PARQUET_NZ_MAP=1`.

`decode_delta_length_byte_array_kernel_with_nz_map` decodes flat
DELTA_LENGTH_BYTE_ARRAY pages using the nz map instead of decoding definition
levels. Its decode loop is the producer/writer pipeline of the DELTA_BINARY
consumer from #24501: warp 0 decodes a batch of lengths into the decoder's
rolling buffer while warp 1 writes the batch before it, placing each length at
the row the map gives. The rest of the kernel follows
`decode_delta_length_byte_array_kernel`: page setup, the skip-resume path that
finds a slice's starting byte offset, the null zero-fill, the offsets scan and
the character copy.

The level-decoding kernel computed three things as side effects of
`gpuDecodeLevels`, which the consumer now computes directly: the null count,
which is `num_rows` minus the map's valid values at or after `first_row`; the
row count the offsets scan covers; and the zero-fill's range, which is
`num_rows`.

`classify_nz_map_kind` now also selects DELTA_LENGTH_BYTE_ARRAY pages. Routing
matches the DELTA_BINARY pair: the level-decoding kernel's skip is gated on the
same flag that decides whether the consumer is launched. The host now records
which encodings still have a page without a map, in `unmapped_kernel_mask`, so
each level-decoding kernel is launched only when its encoding has work for it
rather than on one bool scoped to DELTA_BINARY.

Each consumer now runs on its own stream, forked from the producer's with
`fork_streams` and joined back with `join_streams`. That orders it after the
producer while the two consumers overlap with each other and with the
level-decoding kernels. Sharing the producer's stream, as the DELTA_BINARY
consumer did on its own, would run the two consumers one after the other.

Both nz map consumers now reject a page whose map asks for more values than its
delta header declares, before writing anything, and report
`DATA_STREAM_OVERRUN`. The decoder's cursor advances a mini-block at a time and
can pass the end of a short stream, so the writer read rolling-buffer slots the
page never wrote and the read succeeded with values that were never decoded.
That was already true of the DELTA_BINARY_PACKED consumer from #24501, which is
fixed here too. `nz_count` counts from the start of the page, so it is the
number of values a read needs, and a bounded read that uses fewer values than
the page declares is unaffected. The DLBA consumer also returns before the
offsets scan if its loop stops short, rather than scanning unwritten lengths.

`init_binary_block` now rejects a block size or value count above INT32_MAX.
Both are uleb128s, so 64 bits wide, while the decoder steps its 32-bit
`current_value_idx` by `values_per_mb` until it reaches `value_count`. A block
size of 2^32 with one mini-block narrows `values_per_mb` to zero, and a block
size of 2^31 with a value count above it wraps the index back to zero; either
way `find_end_of_block` never finishes. The DELTA_LENGTH_BYTE_ARRAY and
DELTA_BYTE_ARRAY decoders call it, as does the string-size pass that runs for
those pages whether or not the nz map is on. Parquet stores a page's value count
as int32, so no valid page exceeds either bound. The DELTA_BINARY_PACKED
decoders never call it and decoded garbage rather than hanging; they now report
the header as unsupported.

A mini-block's byte length is now computed in 64 bits. The bitwidth comes from
the page unvalidated, so up to 255, and `values_per_mb` is bounded only by what
fits in 32, so `bitwidth * values_per_mb / 8` could wrap and place the next
mini-block's start below the block, where the `block_end` checks still pass and
the unpack reads the wrong bytes.

Tests: `NzMapMatchesLevelDecoder` gains a nullable DELTA_LENGTH_BYTE_ARRAY
column, without which its consumer is never selected. `NzMapNarrowDeltaBlock`
reads DELTA blocks of 32 and 64 values, narrower than the spec allows but
accepted by `init_binary_block`, and a 256-value mini-block that takes more than
one pass per batch, with the nz map off and on.
`NzMapRejectsShortDeltaStream` builds pages that ask for more values than their
stream declares, for both consumers, plus controls that read a stream holding
exactly enough and a bounded read holding more than it needs.
`DeltaHeaderCountsTooLarge` reads three oversized headers through both encodings
with the nz map off and on.

The two 512 MiB delta benchmarks gain a `validity` axis. The existing nullable
sweep is 8 MiB, which is L2-resident on an H100, so the nz map's extra traffic of
4 bytes per value could not show up there -- and for a string column that cost is
carried against a much larger payload than the fixed-width case. The nested path
builds its own data profile and ignores `validity`, so those cells are skipped
rather than run three times over on identical data.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

Author: vyasr

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, so the nested path ignores `validity` and every
+  // value of it yields the same table. Run that configuration once.
+  if (nesting > 0 and validity != "no_validity") {
+    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);
     }
   }