bench(parquet): add ListArray benchmarks for runtime and peak memory (#9846)

# Which issue does this PR close?

<!--
We generally require a GitHub issue to be filed for all bug fixes and
enhancements and this helps us generate change logs for our releases.
You can link an issue to this PR using the GitHub syntax.
-->

- Contributes to #9731
- Dependency of #9848

# Rationale for this change

See #9848

Existing benchmarks have some gaps in the types of columns they
exercise. Additionally, I would like to improve the memory efficiency of
the read/decode path in terms of RSS requirements, especially for sparse
inputs and we currently do not have any infrastructure to measure that.

# What changes are included in this PR?

Extend the existing `arrow_reader` runtime benchmarks with `Int32` and
`FixedBinary32` list columns alongside the existing `StringList`, with
parameterized null density (0%, 50%, 90%, 99%). The prior benchmarks
only covered string lists, which didn't surface costs specific to
fixed-width and primitive element types.

Add a new `arrow_reader_peak_memory` benchmark that measures peak heap
usage during `ListArrayReader::consume_batch` using a thread-local
tracking allocator. It captures how RSS-efficient we are when
materializing a column into its final Arrow in-memory representation.

# Are these changes tested?

All tests passing.

# Are there any user-facing changes?

None.

Signed-off-by: Hippolyte Barraud <hippolyte.barraud@datadoghq.com>
Co-authored-by: Andrew Lamb <andrew@nerdnetworks.org>

Author: HippoBaro

parquet/Cargo.toml

@@ -289,5 +289,10 @@ required-features = ["arrow"]
 name = "bloom_filter"
 harness = false
 
+[[bench]]
+name = "arrow_reader_peak_memory"
+required-features = ["arrow", "test_common", "experimental"]
+harness = false
+
 [lib]
 bench = false

parquet/benches/arrow_reader.rs

@@ -95,6 +95,16 @@ fn build_test_schema() -> SchemaDescPtr {
             OPTIONAL GROUP optional_struct_optional_int32_leaf {
                 OPTIONAL INT32 element;
             }
+            OPTIONAL GROUP int32_list (LIST) {
+                repeated group list {
+                    optional INT32 element;
+                }
+            }
+            OPTIONAL GROUP fixed32_list (LIST) {
+                repeated group list {
+                    optional FIXED_LEN_BYTE_ARRAY(32) element;
+                }
+            }
         }
         ";
     parse_message_type(message_type)
@@ -668,6 +678,150 @@ fn build_string_list_page_iterator(
     InMemoryPageIterator::new(pages)
 }
 
+fn build_int32_list_page_iterator(
+    column_desc: ColumnDescPtr,
+    null_density: f32,
+) -> impl PageIterator + Clone {
+    let max_def_level = column_desc.max_def_level();
+    let max_rep_level = column_desc.max_rep_level();
+    assert_eq!(max_def_level, 3);
+    assert_eq!(max_rep_level, 1);
+
+    let mut rng = seedable_rng();
+    let mut pages: Vec<Vec<parquet::column::page::Page>> = Vec::new();
+    for _i in 0..NUM_ROW_GROUPS {
+        let mut column_chunk_pages = Vec::new();
+        for _j in 0..PAGES_PER_GROUP {
+            let mut values: Vec<i32> = Vec::with_capacity(VALUES_PER_PAGE * MAX_LIST_LEN);
+            let mut def_levels = Vec::with_capacity(VALUES_PER_PAGE * MAX_LIST_LEN);
+            let mut rep_levels = Vec::with_capacity(VALUES_PER_PAGE * MAX_LIST_LEN);
+            for _k in 0..VALUES_PER_PAGE {
+                rep_levels.push(0);
+                if rng.random::<f32>() < null_density {
+                    def_levels.push(0);
+                    continue;
+                }
+                let len = rng.random_range(0..MAX_LIST_LEN);
+                if len == 0 {
+                    def_levels.push(1);
+                    continue;
+                }
+
+                (1..len).for_each(|_| rep_levels.push(1));
+
+                for _l in 0..len {
+                    if rng.random::<f32>() < null_density {
+                        def_levels.push(2);
+                    } else {
+                        def_levels.push(3);
+                        values.push(rng.random());
+                    }
+                }
+            }
+            let mut page_builder =
+                DataPageBuilderImpl::new(column_desc.clone(), values.len() as u32, true);
+            page_builder.add_rep_levels(max_rep_level, &rep_levels);
+            page_builder.add_def_levels(max_def_level, &def_levels);
+            page_builder.add_values::<Int32Type>(Encoding::PLAIN, &values);
+            column_chunk_pages.push(page_builder.consume());
+        }
+        pages.push(column_chunk_pages);
+    }
+
+    InMemoryPageIterator::new(pages)
+}
+
+fn create_int32_list_reader(
+    page_iterator: impl PageIterator + 'static,
+    column_desc: ColumnDescPtr,
+) -> Box<dyn ArrayReader> {
+    use parquet::arrow::array_reader::PrimitiveArrayReader;
+    let items = Box::new(
+        PrimitiveArrayReader::<Int32Type>::new(
+            Box::new(page_iterator),
+            column_desc,
+            None,
+            DEFAULT_BATCH_SIZE,
+        )
+        .unwrap(),
+    ) as Box<dyn ArrayReader>;
+    let field = Field::new_list_field(DataType::Int32, true);
+    let data_type = DataType::List(Arc::new(field));
+    Box::new(ListArrayReader::<i32>::new(items, data_type, 2, 1, true))
+}
+
+const FIXED_BYTE_LEN: usize = 32;
+
+fn build_fixed32_list_page_iterator(
+    column_desc: ColumnDescPtr,
+    null_density: f32,
+) -> impl PageIterator + Clone {
+    let max_def_level = column_desc.max_def_level();
+    let max_rep_level = column_desc.max_rep_level();
+    assert_eq!(max_def_level, 3);
+    assert_eq!(max_rep_level, 1);
+
+    let mut rng = seedable_rng();
+    let mut pages: Vec<Vec<parquet::column::page::Page>> = Vec::new();
+    for _i in 0..NUM_ROW_GROUPS {
+        let mut column_chunk_pages = Vec::new();
+        for _j in 0..PAGES_PER_GROUP {
+            let mut values: Vec<parquet::data_type::FixedLenByteArray> =
+                Vec::with_capacity(VALUES_PER_PAGE * MAX_LIST_LEN);
+            let mut def_levels = Vec::with_capacity(VALUES_PER_PAGE * MAX_LIST_LEN);
+            let mut rep_levels = Vec::with_capacity(VALUES_PER_PAGE * MAX_LIST_LEN);
+            for _k in 0..VALUES_PER_PAGE {
+                rep_levels.push(0);
+                if rng.random::<f32>() < null_density {
+                    def_levels.push(0);
+                    continue;
+                }
+                let len = rng.random_range(0..MAX_LIST_LEN);
+                if len == 0 {
+                    def_levels.push(1);
+                    continue;
+                }
+                (1..len).for_each(|_| rep_levels.push(1));
+                for _l in 0..len {
+                    if rng.random::<f32>() < null_density {
+                        def_levels.push(2);
+                    } else {
+                        def_levels.push(3);
+                        let mut buf = vec![0u8; FIXED_BYTE_LEN];
+                        rng.fill(&mut buf[..]);
+                        values.push(buf.into());
+                    }
+                }
+            }
+            let mut page_builder =
+                DataPageBuilderImpl::new(column_desc.clone(), values.len() as u32, true);
+            page_builder.add_rep_levels(max_rep_level, &rep_levels);
+            page_builder.add_def_levels(max_def_level, &def_levels);
+            page_builder.add_values::<FixedLenByteArrayType>(Encoding::PLAIN, &values);
+            column_chunk_pages.push(page_builder.consume());
+        }
+        pages.push(column_chunk_pages);
+    }
+
+    InMemoryPageIterator::new(pages)
+}
+
+fn create_fixed32_list_reader(
+    page_iterator: impl PageIterator + 'static,
+    column_desc: ColumnDescPtr,
+) -> Box<dyn ArrayReader> {
+    let items = make_fixed_len_byte_array_reader(
+        Box::new(page_iterator),
+        column_desc,
+        None,
+        DEFAULT_BATCH_SIZE,
+    )
+    .unwrap();
+    let field = Field::new_list_field(DataType::FixedSizeBinary(FIXED_BYTE_LEN as i32), true);
+    let data_type = DataType::List(Arc::new(field));
+    Box::new(ListArrayReader::<i32>::new(items, data_type, 2, 1, true))
+}
+
 fn bench_array_reader(mut array_reader: Box<dyn ArrayReader>) -> usize {
     // test procedure: read data in batches of 8192 until no more data
     let mut total_count = 0;
@@ -1645,6 +1799,8 @@ fn add_benches(c: &mut Criterion) {
     let optional_uint64_column_desc = schema.column(38);
     let mandatory_struct_optional_in32_column_desc = schema.column(39);
     let optional_struct_optional_in32_column_desc = schema.column(40);
+    let int32_list_desc = schema.column(41);
+    let fixed32_list_desc = schema.column(42);
 
     // primitive / int32 benchmarks
     // =============================
@@ -2228,24 +2384,59 @@ fn add_benches(c: &mut Criterion) {
     // list benchmarks
     //==============================
 
-    let list_data = build_string_list_page_iterator(string_list_desc.clone(), 0.);
-    let mut group = c.benchmark_group("arrow_array_reader/ListArray");
-    group.bench_function("plain encoded optional strings no NULLs", |b| {
-        b.iter(|| {
-            let reader = create_string_list_reader(list_data.clone(), string_list_desc.clone());
-            count = bench_array_reader(reader);
+    let mut group = c.benchmark_group("arrow_array_reader/ListArray/StringList");
+    for (label, null_density) in [
+        ("no NULLs", 0.0),
+        ("half NULLs", 0.5),
+        ("90pct NULLs", 0.9),
+        ("99pct NULLs", 0.99),
+    ] {
+        let list_data = build_string_list_page_iterator(string_list_desc.clone(), null_density);
+        group.bench_function(label, |b| {
+            b.iter(|| {
+                let reader = create_string_list_reader(list_data.clone(), string_list_desc.clone());
+                count = bench_array_reader(reader);
+            });
+            assert_eq!(count, EXPECTED_VALUE_COUNT);
         });
-        assert_eq!(count, EXPECTED_VALUE_COUNT);
-    });
-    let list_data = build_string_list_page_iterator(string_list_desc.clone(), 0.5);
-    group.bench_function("plain encoded optional strings half NULLs", |b| {
-        b.iter(|| {
-            let reader = create_string_list_reader(list_data.clone(), string_list_desc.clone());
-            count = bench_array_reader(reader);
+    }
+    group.finish();
+
+    let mut group = c.benchmark_group("arrow_array_reader/ListArray/Int32List");
+    for (label, null_density) in [
+        ("no NULLs", 0.0),
+        ("half NULLs", 0.5),
+        ("90pct NULLs", 0.9),
+        ("99pct NULLs", 0.99),
+    ] {
+        let list_data = build_int32_list_page_iterator(int32_list_desc.clone(), null_density);
+        group.bench_function(label, |b| {
+            b.iter(|| {
+                let reader = create_int32_list_reader(list_data.clone(), int32_list_desc.clone());
+                count = bench_array_reader(reader);
+            });
+            assert_eq!(count, EXPECTED_VALUE_COUNT);
         });
-        assert_eq!(count, EXPECTED_VALUE_COUNT);
-    });
+    }
+    group.finish();
 
+    let mut group = c.benchmark_group("arrow_array_reader/ListArray/Fixed32List");
+    for (label, null_density) in [
+        ("no NULLs", 0.0),
+        ("half NULLs", 0.5),
+        ("90pct NULLs", 0.9),
+        ("99pct NULLs", 0.99),
+    ] {
+        let list_data = build_fixed32_list_page_iterator(fixed32_list_desc.clone(), null_density);
+        group.bench_function(label, |b| {
+            b.iter(|| {
+                let reader =
+                    create_fixed32_list_reader(list_data.clone(), fixed32_list_desc.clone());
+                count = bench_array_reader(reader);
+            });
+            assert_eq!(count, EXPECTED_VALUE_COUNT);
+        });
+    }
     group.finish();
 
     // fixed_len_byte_array benchmarks