bench(parquet): cover DELTA_BYTE_ARRAY large values in the arrow_writer benchmark

parquet/benches/arrow_writer.rs

@@ -20,7 +20,7 @@ extern crate criterion;
 
 use criterion::{Bencher, Criterion, Throughput};
 use parquet::arrow::ArrowWriter;
-use parquet::basic::{Compression, ZstdLevel};
+use parquet::basic::{Compression, Encoding, ZstdLevel};
 
 extern crate arrow;
 extern crate parquet;
@@ -123,6 +123,31 @@ fn create_large_string_bench_batch(size: usize, value_size: usize) -> Result<Rec
     Ok(RecordBatch::try_from_iter([("col", array)])?)
 }
 
+/// `size` rows of `value_size`-byte strings sharing a long common prefix and
+/// ending in a short distinct suffix — the case `DELTA_BYTE_ARRAY` exists
+/// for: consecutive values dedup to a prefix length plus a few suffix bytes.
+fn create_large_string_shared_prefix_bench_batch(
+    size: usize,
+    value_size: usize,
+) -> Result<RecordBatch> {
+    let prefix = "x".repeat(value_size - 8);
+    let array = Arc::new(StringArray::from_iter_values(
+        (0..size).map(|i| format!("{prefix}{i:08}")),
+    )) as _;
+    Ok(RecordBatch::try_from_iter([("col", array)])?)
+}
+
+/// `size` rows of `value_size`-byte strings whose leading bytes differ — the
+/// adversarial case for `DELTA_BYTE_ARRAY`, where every prefix length is ~0
+/// and the encoding stores each value in full.
+fn create_large_string_distinct_bench_batch(size: usize, value_size: usize) -> Result<RecordBatch> {
+    let filler = "x".repeat(value_size - 8);
+    let array = Arc::new(StringArray::from_iter_values(
+        (0..size).map(|i| format!("{i:08}{filler}")),
+    )) as _;
+    Ok(RecordBatch::try_from_iter([("col", array)])?)
+}
+
 fn create_string_and_binary_view_bench_batch(
     size: usize,
     null_density: f32,
@@ -651,5 +676,55 @@ fn bench_all_writers(c: &mut Criterion) {
     }
 }
 
-criterion_group!(benches, bench_all_writers);
+/// Large-value BYTE_ARRAY columns under `DELTA_BYTE_ARRAY`, which the
+/// property matrix above never exercises (it only varies writer version,
+/// compression, bloom filters, and CDC — all on the default encoding).
+///
+/// Values are sized so a single value exceeds the default 1 MiB data page
+/// limit, putting the writer in the byte-budget sub-batching regime of
+/// `write_batch_internal`. That budget measures raw payload bytes, so under
+/// `DELTA_BYTE_ARRAY` it sub-batches by the values' *pre-dedup* size however
+/// well they compress; benchmarking against `PLAIN` on the same data bounds
+/// what an encoded-size-aware budget could recover. The shared-prefix and
+/// distinct batches bracket the encoding's best and worst case.
+fn bench_delta_byte_array_writers(c: &mut Criterion) {
+    // 128 rows × 2 MiB: the same total bytes as `large_string_non_null`
+    // above, but with each value alone exceeding the default 1 MiB page
+    // limit (the regime of #10489, where the byte budget resolves to
+    // one-value mini-batches).
+    let shared = create_large_string_shared_prefix_bench_batch(128, 2 * 1024 * 1024).unwrap();
+    let distinct = create_large_string_distinct_bench_batch(128, 2 * 1024 * 1024).unwrap();
+
+    let plain = WriterProperties::builder()
+        .set_dictionary_enabled(false)
+        .set_encoding(Encoding::PLAIN)
+        .build();
+    let delta = WriterProperties::builder()
+        .set_dictionary_enabled(false)
+        .set_encoding(Encoding::DELTA_BYTE_ARRAY)
+        .build();
+
+    for (batch_name, batch) in [
+        ("large_string_shared_prefix", &shared),
+        ("large_string_distinct", &distinct),
+    ] {
+        let mut group = c.benchmark_group(batch_name);
+        group.throughput(Throughput::Bytes(
+            batch
+                .columns()
+                .iter()
+                .map(|f| f.get_array_memory_size() as u64)
+                .sum(),
+        ));
+
+        for (prop_name, prop) in [("plain", &plain), ("delta_byte_array", &delta)] {
+            group.bench_function(prop_name, |b| {
+                write_batch_with_option(b, batch, Some((*prop).clone())).unwrap()
+            });
+        }
+        group.finish();
+    }
+}
+
+criterion_group!(benches, bench_all_writers, bench_delta_byte_array_writers);
 criterion_main!(benches);