Optimise min/max computation for VarBinView arrays (#9740)

Leverage view values to perform fast min/max for VarBinView arrays.
Unify the view comparison logic across multiple places in the project
under ResolvedViews

---------

Signed-off-by: Robert Kruszewski <github@robertk.io>

Author: robert3005

vortex-array/benches/aggregate_max.rs

@@ -8,8 +8,13 @@ use mimalloc::MiMalloc;
 use rand::prelude::*;
 use vortex_array::IntoArray;
 use vortex_array::VortexSessionExecute;
+use vortex_array::aggregate_fn::NumericalAggregateOpts;
+use vortex_array::aggregate_fn::fns::min_max::min_max;
 use vortex_array::array_session;
 use vortex_array::arrays::PrimitiveArray;
+use vortex_array::arrays::VarBinViewArray;
+use vortex_array::dtype::DType;
+use vortex_array::dtype::Nullability;
 use vortex_session::VortexSession;
 
 #[global_allocator]
@@ -106,3 +111,123 @@ fn max_i32_nulls_scattered(bencher: Bencher) {
         })
         .bench_refs(|(a, ctx)| a.statistics().compute_max::<i32>(ctx));
 }
+
+// String extrema. Every view carries its value's first four bytes inline, so the shape that
+// matters is how far into a value its order is settled: `distinct_prefixes` and
+// `low_cardinality` never have to look past the view, while the log-shaped columns (every value
+// starting `thread-`/`req-`/`This is `) tie on those bytes and have to resolve the value.
+//
+// Like the primitive benches, each iteration gets a fresh array: `min_max` caches `Stat::Min`
+// and `Stat::Max` on the array it scanned, so a reused array would only be scanned once.
+
+// Resolving and comparing string values costs far more per row than a primitive, so these scan a
+// smaller column to keep each benchmark under 1ms in CodSpeed simulation.
+const STRING_N: usize = 4_096;
+
+fn bench_min_max(bencher: Bencher, values: Vec<Option<String>>) {
+    bencher
+        .with_inputs(|| {
+            (
+                VarBinViewArray::from_iter(
+                    values.iter().map(|v| v.as_deref()),
+                    DType::Utf8(Nullability::Nullable),
+                )
+                .into_array(),
+                SESSION.create_execution_ctx(),
+            )
+        })
+        .bench_refs(|(a, ctx)| min_max(a, ctx, NumericalAggregateOpts::default()));
+}
+
+/// Long values with distinct leading bytes: the prefix rejects almost every element.
+#[divan::bench]
+fn max_utf8_view_distinct_prefixes(bencher: Bencher) {
+    let mut rng = StdRng::seed_from_u64(6);
+    bench_min_max(
+        bencher,
+        (0..STRING_N)
+            .map(|_| {
+                Some(format!(
+                    "{:016x}-{:016x}",
+                    rng.random::<u64>(),
+                    rng.random::<u64>()
+                ))
+            })
+            .collect(),
+    );
+}
+
+/// Log levels: short inlined values over a five-value dictionary, with distinct leading bytes.
+#[divan::bench]
+fn max_utf8_view_low_cardinality(bencher: Bencher) {
+    bench_min_max(
+        bencher,
+        (0..STRING_N)
+            .map(|i| {
+                Some(
+                    match i % 5 {
+                        0 => "ERROR",
+                        1 => "WARN",
+                        2 => "INFO",
+                        3 => "DEBUG",
+                        _ => "TRACE",
+                    }
+                    .to_string(),
+                )
+            })
+            .collect(),
+    );
+}
+
+/// Thread names: short inlined values that tie on the inline prefix and are settled a few
+/// bytes later.
+#[divan::bench]
+fn max_utf8_view_tied_prefix_inlined(bencher: Bencher) {
+    bench_min_max(
+        bencher,
+        (0..STRING_N)
+            .map(|i| Some(format!("thread-{}", i % 8)))
+            .collect(),
+    );
+}
+
+/// Request ids: 20-byte buffer-backed values that tie on the inline prefix `req-`.
+#[divan::bench]
+fn max_utf8_view_tied_prefix_outlined(bencher: Bencher) {
+    bench_min_max(
+        bencher,
+        (0..STRING_N)
+            .map(|i| Some(format!("req-{i:016x}")))
+            .collect(),
+    );
+}
+
+/// Log lines: long, buffer-backed values that agree for their first twenty-six bytes.
+#[divan::bench]
+fn max_utf8_view_shared_prefix(bencher: Bencher) {
+    bench_min_max(
+        bencher,
+        (0..STRING_N)
+            .map(|i| {
+                Some(format!(
+                    "This is log message number {i} with some additional content to simulate real logs"
+                ))
+            })
+            .collect(),
+    );
+}
+
+/// Distinct prefixes with ~50% random nulls, producing many short valid runs.
+#[divan::bench]
+fn max_utf8_view_nulls_scattered(bencher: Bencher) {
+    let mut rng = StdRng::seed_from_u64(7);
+    bench_min_max(
+        bencher,
+        (0..STRING_N)
+            .map(|_| {
+                rng.random_bool(0.5)
+                    .then(|| format!("{:016x}-{:016x}", rng.random::<u64>(), rng.random::<u64>()))
+            })
+            .collect(),
+    );
+}

vortex-array/src/aggregate_fn/fns/min_max/mod.rs

@@ -5,7 +5,7 @@ mod bool;
 mod decimal;
 mod extension;
 mod primitive;
-mod varbin;
+mod varbinview;
 
 use std::sync::LazyLock;
 
@@ -19,7 +19,7 @@ use self::bool::accumulate_bool;
 use self::decimal::accumulate_decimal;
 use self::extension::accumulate_extension;
 use self::primitive::accumulate_primitive;
-use self::varbin::accumulate_varbinview;
+use self::varbinview::accumulate_varbinview;
 use crate::ArrayRef;
 use crate::Canonical;
 use crate::Columnar;