perf(aggregation): break the Kahan-sum dependency chain with accumulator lanes

The metric reducer `collect_stats` (shared by sum/avg/min/max/count/stats)
folded values one at a time through Kahan compensated summation. The Kahan
recurrence carries `sum`/`delta` across every iteration — a strict serial
dependency chain that blocks both CPU pipelining and auto-vectorization of
the co-located min/max, and it ran over an iterator rather than a slice.

Add `ColumnBlockAccessor::vals()` to expose the fetched block as a slice,
and reduce it with 4 independent (sum, delta) Kahan lanes + 4 min/max lanes,
merged back with the same compensated combination as `merge_fruits`. The
four chains run in parallel and min/max vectorize.

Accuracy is preserved: it is still Kahan-compensated; only the summation
order changes, exactly as it already does when merging across segments.
All aggregation tests pass.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

columnar/src/block_accessor.rs

@@ -163,6 +163,15 @@ impl<T: PartialOrd + Copy + std::fmt::Debug + Send + Sync + 'static + Default>
         self.val_cache.iter().cloned()
     }
 
+    /// Returns the fetched values of the current block as a contiguous slice.
+    ///
+    /// This lets reducers (sum/min/max/stats) process the block with a tight, vectorizable
+    /// loop instead of going through the `iter_vals` iterator.
+    #[inline]
+    pub fn vals(&self) -> &[T] {
+        &self.val_cache
+    }
+
     #[inline]
     /// Returns an iterator over the docids and values
     /// The passed in `docs` slice needs to be the same slice that was passed to `fetch_block` or

src/aggregation/metric/stats.rs

@@ -300,11 +300,11 @@ impl<const COLUMN_TYPE_ID: u8> SegmentAggregationCollector
             &self.accessor,
             self.missing_u64,
         );
-        collect_stats::<COLUMN_TYPE_ID>(
+        collect_stats_slice::<COLUMN_TYPE_ID>(
             &mut self.buckets[parent_bucket_id as usize],
-            agg_data.column_block_accessor.iter_vals(),
+            agg_data.column_block_accessor.vals(),
             self.is_number_or_date_type,
-        )?;
+        );
 
         Ok(())
     }
@@ -357,6 +357,72 @@ impl<const COLUMN_TYPE_ID: u8> SegmentAggregationCollector
     }
 }
 
+/// Reduces a contiguous block of raw column values into `stats`.
+///
+/// Uses `LANES` independent (sum, delta) Kahan accumulators and `LANES` min/max
+/// accumulators so the per-element dependency chain of the serial Kahan sum is broken,
+/// letting the CPU pipeline the additions and auto-vectorize the min/max. The lanes are
+/// merged back into `stats` with the same compensated-sum combination used by
+/// [`IntermediateStats::merge_fruits`], so accuracy is preserved (the summation order
+/// differs, exactly as it already does across segment merges).
+#[inline]
+fn collect_stats_slice<const COLUMN_TYPE_ID: u8>(
+    stats: &mut IntermediateStats,
+    vals: &[u64],
+    is_number_or_date_type: bool,
+) {
+    if !is_number_or_date_type {
+        // Non-numeric: only the presence of a value matters (preserve existing behavior).
+        for _ in 0..vals.len() {
+            stats.collect(0.0);
+        }
+        return;
+    }
+
+    const LANES: usize = 4;
+    let mut sum = [0f64; LANES];
+    let mut delta = [0f64; LANES];
+    let mut min = [f64::MAX; LANES];
+    let mut max = [f64::MIN; LANES];
+
+    let mut chunks = vals.chunks_exact(LANES);
+    for chunk in chunks.by_ref() {
+        for lane in 0..LANES {
+            let val = convert_to_f64::<COLUMN_TYPE_ID>(chunk[lane]);
+            // Per-lane Kahan summation.
+            let y = val - delta[lane];
+            let t = sum[lane] + y;
+            delta[lane] = (t - sum[lane]) - y;
+            sum[lane] = t;
+            min[lane] = min[lane].min(val);
+            max[lane] = max[lane].max(val);
+        }
+    }
+
+    stats.count += vals.len() as u64;
+
+    // Merge the lanes into `stats`.
+    for lane in 0..LANES {
+        let y = sum[lane] - (stats.delta + delta[lane]);
+        let t = stats.sum + y;
+        stats.delta = (t - stats.sum) - y;
+        stats.sum = t;
+        stats.min = stats.min.min(min[lane]);
+        stats.max = stats.max.max(max[lane]);
+    }
+
+    // Tail (fewer than LANES values) — fold directly into `stats` (count already added).
+    for &raw in chunks.remainder() {
+        let val = convert_to_f64::<COLUMN_TYPE_ID>(raw);
+        let y = val - stats.delta;
+        let t = stats.sum + y;
+        stats.delta = (t - stats.sum) - y;
+        stats.sum = t;
+        stats.min = stats.min.min(val);
+        stats.max = stats.max.max(val);
+    }
+}
+
 #[inline]
 fn collect_stats<const COLUMN_TYPE_ID: u8>(
     stats: &mut IntermediateStats,