vortex-row: add fused fast-path FSST row-encode benchmark

Adds `fsst_fast_fused`: bulk-decompresses the FSST code heap straight into a
contiguous buffer (no intermediate VarBinViewArray) and block-encodes rows directly
into the row-key ListView using the stored uncompressed_lengths (free size pass), with
the same no-zero-init / no-extra-copy techniques as the row encoder. Lets us compare
the fused path head-to-head against decode-then-convert.

Signed-off-by: Joe Isaacs <joe.isaacs@live.co.uk>

Author: joseph-isaacs

vortex-row/benches/fsst_row_encode.rs

@@ -1,7 +1,11 @@
 // SPDX-License-Identifier: Apache-2.0
 // SPDX-FileCopyrightText: Copyright the Vortex contributors
 
-#![expect(clippy::unwrap_used)]
+#![expect(
+    clippy::unwrap_used,
+    clippy::expect_used,
+    clippy::cast_possible_truncation
+)]
 
 //! Row-encoding an FSST-compressed string column: the only realizable strategy is
 //! "unpack then convert" (decompress FSST to a canonical `VarBinView`, then row-encode it),
@@ -29,9 +33,17 @@ use vortex_array::Canonical;
 use vortex_array::IntoArray;
 use vortex_array::LEGACY_SESSION;
 use vortex_array::VortexSessionExecute;
+use vortex_array::arrays::ListViewArray;
+use vortex_array::arrays::PrimitiveArray;
 use vortex_array::arrays::VarBinArray;
 use vortex_array::dtype::DType;
 use vortex_array::dtype::Nullability;
+use vortex_array::match_each_integer_ptype;
+use vortex_array::validity::Validity;
+use vortex_buffer::Buffer;
+use vortex_buffer::ByteBufferMut;
+use vortex_fsst::FSST;
+use vortex_fsst::FSSTArrayExt;
 use vortex_fsst::fsst_compress;
 use vortex_fsst::fsst_train_compressor;
 use vortex_row::RowEncoder;
@@ -80,6 +92,117 @@ fn decompress(fsst: &ArrayRef) -> ArrayRef {
         .into_array()
 }
 
+const VARLEN_BLOCK: usize = 32;
+const VARLEN_BLOCK_TOTAL: usize = 33;
+// Sentinel for a non-empty varlen value (ascending, non-null) — value is irrelevant to timing.
+const NON_EMPTY_SENTINEL: u8 = 0x02;
+
+/// Encoded row-key length for a non-empty value of `len` decompressed bytes: a leading
+/// sentinel plus `ceil(len/32)` 32-byte blocks, each followed by a continuation/length byte.
+fn encoded_len(len: usize) -> u32 {
+    if len == 0 {
+        1
+    } else {
+        1 + (len.div_ceil(VARLEN_BLOCK) as u32) * VARLEN_BLOCK_TOTAL as u32
+    }
+}
+
+/// Block-encode `bytes` (ascending) into `out`, matching vortex-row's varlen body format.
+fn block_encode(bytes: &[u8], out: &mut [u8]) {
+    let len = bytes.len();
+    let full = len / VARLEN_BLOCK;
+    let partial = len % VARLEN_BLOCK;
+    let (full_to_write, partial_len) = if partial == 0 {
+        (full - 1, VARLEN_BLOCK)
+    } else {
+        (full, partial)
+    };
+    let mut src = 0;
+    let mut dst = 0;
+    for _ in 0..full_to_write {
+        out[dst..dst + VARLEN_BLOCK].copy_from_slice(&bytes[src..src + VARLEN_BLOCK]);
+        out[dst + VARLEN_BLOCK] = 0xFF;
+        src += VARLEN_BLOCK;
+        dst += VARLEN_BLOCK_TOTAL;
+    }
+    out[dst..dst + partial_len].copy_from_slice(&bytes[src..src + partial_len]);
+    for b in &mut out[dst + partial_len..dst + VARLEN_BLOCK] {
+        *b = 0;
+    }
+    out[dst + VARLEN_BLOCK] = partial_len as u8;
+}
+
+/// Fused FSST → row-key kernel: bulk-decompress the code heap into one contiguous buffer (no
+/// intermediate `VarBinViewArray`), then block-encode each row straight into the row-key
+/// `ListView<u8>` using the stored `uncompressed_lengths` for boundaries (no size-pass walk).
+fn fast_fused(fsst: &ArrayRef) -> ArrayRef {
+    let mut ctx = LEGACY_SESSION.create_execution_ctx();
+    let view = fsst.as_opt::<FSST>().expect("FSST array");
+
+    // Per-row decompressed lengths are already stored — the size pass is free.
+    let lens_arr = view
+        .uncompressed_lengths()
+        .clone()
+        .execute::<PrimitiveArray>(&mut ctx)
+        .unwrap();
+    let lens: Vec<usize> = match_each_integer_ptype!(lens_arr.ptype(), |P| {
+        lens_arr
+            .as_slice::<P>()
+            .iter()
+            .map(|x| *x as usize)
+            .collect()
+    });
+
+    // Bulk-decompress the whole code heap once into a contiguous buffer (no VarBinView).
+    let heap = view.codes_bytes();
+    let total: usize = lens.iter().sum();
+    let decompressor = view.decompressor();
+    let mut decompressed = ByteBufferMut::with_capacity(total + 7);
+    let n = decompressor.decompress_into(heap.as_slice(), decompressed.spare_capacity_mut());
+    unsafe { decompressed.set_len(n) };
+    let bytes = decompressed.as_slice();
+
+    // Size + offsets for the row-key ListView (lengths are free, no view walk).
+    let nrows = lens.len();
+    let mut offsets: Vec<u32> = Vec::with_capacity(nrows);
+    let mut sizes: Vec<u32> = Vec::with_capacity(nrows);
+    let mut acc: u32 = 0;
+    for &l in &lens {
+        offsets.push(acc);
+        let sz = encoded_len(l);
+        sizes.push(sz);
+        acc += sz;
+    }
+
+    // Block-encode every row directly into the elements buffer. No zero-init (every byte is
+    // written: sentinel + block body with zero-padded final block) and no Vec→Buffer copy.
+    let mut out = ByteBufferMut::with_capacity(acc as usize);
+    unsafe { out.set_len(acc as usize) };
+    let out_slice = out.as_mut_slice();
+    let mut src = 0usize;
+    for (i, &l) in lens.iter().enumerate() {
+        let pos = offsets[i] as usize;
+        out_slice[pos] = NON_EMPTY_SENTINEL;
+        if l != 0 {
+            block_encode(&bytes[src..src + l], &mut out_slice[pos + 1..]);
+        }
+        src += l;
+    }
+
+    let elements = PrimitiveArray::new(out.freeze(), Validity::NonNullable);
+    let offsets_arr =
+        PrimitiveArray::new(Buffer::<u32>::copy_from(&offsets), Validity::NonNullable);
+    let sizes_arr = PrimitiveArray::new(Buffer::<u32>::copy_from(&sizes), Validity::NonNullable);
+    ListViewArray::try_new(
+        elements.into_array(),
+        offsets_arr.into_array(),
+        sizes_arr.into_array(),
+        Validity::NonNullable,
+    )
+    .unwrap()
+    .into_array()
+}
+
 fn main() {
     divan::main();
 }
@@ -102,6 +225,16 @@ fn fsst_unpack_then_convert(bencher: divan::Bencher) {
         });
 }
 
+/// Fused fast path: bulk-decompress directly into the row-key block format, skipping the
+/// intermediate `VarBinViewArray` and the generic row-encoder (size pass is free).
+#[divan::bench]
+fn fsst_fast_fused(bencher: divan::Bencher) {
+    let (fsst, total_bytes) = build_fsst();
+    bencher
+        .counter(BytesCount::new(total_bytes))
+        .bench_local(|| fast_fused(&fsst));
+}
+
 /// Irreducible floor: FSST decompression alone (a direct kernel must still produce these
 /// bytes, since the sort key *is* the decompressed bytes).
 #[divan::bench]