parquet: batch consecutive null/empty rows in write_list

Restructure `write_list()` to accumulate consecutive null and empty rows
and flush them in a single `visit_leaves()` call using
`extend(repeat_n(...))`, instead of calling `visit_leaves()` per row.

With sparse data (99% nulls), a 4096-row batch previously triggered
~4000 individual tree traversals, each pushing a single value per leaf.
Now consecutive null/empty runs are collapsed into one traversal that
extends all leaf level buffers in bulk.

This follows the same pattern already used by `write_struct()`. The
`write_non_null_slice` path is unchanged since each non-null row has
different offsets and cannot be batched.

Benchmark results (list_primitive_sparse_99pct_null, 65536 rows, 99% nulls):

  benchmark                    baseline     optimized    change
  sparse_list/default          173.01 µs    147.14 µs    -15%
  sparse_list/parquet_2        178.83 µs    147.19 µs    -18%
  sparse_list/bloom_filter     241.27 µs    203.94 µs    -15%
  sparse_list/zstd             196.23 µs    168.50 µs    -14%
  sparse_list/zstd_parquet_2   155.58 µs    155.58 µs     ~0% (baseline noisy)

No measurable change on list benchmarks with 25% nulls.

Signed-off-by: Hippolyte Barraud <hippolyte.barraud@datadoghq.com>

Author: HippoBaro

parquet/benches/arrow_writer.rs

@@ -411,6 +411,10 @@ fn create_batches() -> Vec<(&'static str, RecordBatch)> {
     let batch = create_list_primitive_bench_batch_non_null(BATCH_SIZE, 0.25, 0.75).unwrap();
     batches.push(("list_primitive_non_null", batch));
 
+    // Sparse list: 99% nulls — exercises batched null handling in write_list
+    let batch = create_list_primitive_bench_batch(BATCH_SIZE, 0.99, 0.75).unwrap();
+    batches.push(("list_primitive_sparse_99pct_null", batch));
+
     batches
 }
 

parquet/src/arrow/arrow_writer/levels.rs

@@ -335,51 +335,81 @@ impl LevelInfoBuilder {
                 })
             };
 
-        let write_empty_slice = |child: &mut LevelInfoBuilder| {
-            child.visit_leaves(|leaf| {
-                let rep_levels = leaf.rep_levels.as_mut().unwrap();
-                rep_levels.push(ctx.rep_level - 1);
-                let def_levels = leaf.def_levels.as_mut().unwrap();
-                def_levels.push(ctx.def_level - 1);
-            })
+        let flush_nulls = |child: &mut LevelInfoBuilder, count: usize| {
+            if count > 0 {
+                child.visit_leaves(|leaf| {
+                    leaf.rep_levels
+                        .as_mut()
+                        .unwrap()
+                        .extend(std::iter::repeat_n(ctx.rep_level - 1, count));
+                    leaf.def_levels
+                        .as_mut()
+                        .unwrap()
+                        .extend(std::iter::repeat_n(ctx.def_level - 2, count));
+                });
+            }
         };
 
-        let write_null_slice = |child: &mut LevelInfoBuilder| {
-            child.visit_leaves(|leaf| {
-                let rep_levels = leaf.rep_levels.as_mut().unwrap();
-                rep_levels.push(ctx.rep_level - 1);
-                let def_levels = leaf.def_levels.as_mut().unwrap();
-                def_levels.push(ctx.def_level - 2);
-            })
+        let flush_empties = |child: &mut LevelInfoBuilder, count: usize| {
+            if count > 0 {
+                child.visit_leaves(|leaf| {
+                    leaf.rep_levels
+                        .as_mut()
+                        .unwrap()
+                        .extend(std::iter::repeat_n(ctx.rep_level - 1, count));
+                    leaf.def_levels
+                        .as_mut()
+                        .unwrap()
+                        .extend(std::iter::repeat_n(ctx.def_level - 1, count));
+                });
+            }
         };
 
         match nulls {
             Some(nulls) => {
                 let null_offset = range.start;
+                let mut pending_nulls: usize = 0;
+                let mut pending_empties: usize = 0;
+
                 // TODO: Faster bitmask iteration (#1757)
                 for (idx, w) in offsets.windows(2).enumerate() {
                     let is_valid = nulls.is_valid(idx + null_offset);
                     let start_idx = w[0].as_usize();
                     let end_idx = w[1].as_usize();
+
                     if !is_valid {
-                        write_null_slice(child)
+                        flush_empties(child, pending_empties);
+                        pending_empties = 0;
+                        pending_nulls += 1;
                     } else if start_idx == end_idx {
-                        write_empty_slice(child)
+                        flush_nulls(child, pending_nulls);
+                        pending_nulls = 0;
+                        pending_empties += 1;
                     } else {
-                        write_non_null_slice(child, start_idx, end_idx)
+                        flush_nulls(child, pending_nulls);
+                        pending_nulls = 0;
+                        flush_empties(child, pending_empties);
+                        pending_empties = 0;
+                        write_non_null_slice(child, start_idx, end_idx);
                     }
                 }
+                flush_nulls(child, pending_nulls);
+                flush_empties(child, pending_empties);
             }
             None => {
+                let mut pending_empties: usize = 0;
                 for w in offsets.windows(2) {
                     let start_idx = w[0].as_usize();
                     let end_idx = w[1].as_usize();
                     if start_idx == end_idx {
-                        write_empty_slice(child)
+                        pending_empties += 1;
                     } else {
-                        write_non_null_slice(child, start_idx, end_idx)
+                        flush_empties(child, pending_empties);
+                        pending_empties = 0;
+                        write_non_null_slice(child, start_idx, end_idx);
                     }
                 }
+                flush_empties(child, pending_empties);
             }
         }
     }