Constant row-encode kernel

Replace the stub `RowSizeKernel` / `RowEncodeKernel` impls for
`ConstantArray` with real implementations that skip canonicalization.

The size pass adds the (constant) per-row scalar size to every entry of
the shared `sizes` slice. The encode pass encodes the scalar bytes once
into a small heap buffer, then `copy_nonoverlapping`s those bytes into
each row's slot. Per-row work is one `copy_nonoverlapping(N)` plus one
cursor add, where `N` is typically 9 (i64), 5 (i32), or 17 (i128).

Add a `constant_i64_*` bench triplet (arrow-row baseline, vortex with
kernel, vortex through canonicalization) and a
`constant_path_matches_canonical` test that round-trips bytes both
ways and asserts they're identical.

Signed-off-by: Claude <noreply@anthropic.com>

Author: claude

vortex-row/benches/row_encode.rs

@@ -30,9 +30,11 @@ use rand::RngExt;
 use rand::SeedableRng;
 use rand::distr::Alphanumeric;
 use rand::rngs::StdRng;
+use vortex_array::Canonical;
 use vortex_array::IntoArray;
 use vortex_array::LEGACY_SESSION;
 use vortex_array::VortexSessionExecute;
+use vortex_array::arrays::ConstantArray;
 use vortex_array::arrays::PrimitiveArray;
 use vortex_array::arrays::StructArray;
 use vortex_array::arrays::VarBinViewArray;
@@ -175,3 +177,40 @@ fn struct_mixed_vortex(bencher: divan::Bencher) {
         convert_columns(&[struct_arr.clone()], &[SortField::default()], &mut ctx).unwrap()
     })
 }
+
+// ---------- constant_i64 ----------
+
+#[divan::bench]
+fn constant_i64_arrow_row(bencher: divan::Bencher) {
+    let arr = Arc::new(Int64Array::from(vec![42i64; N])) as arrow_array::ArrayRef;
+    let conv = RowConverter::new(vec![ArrowSortField::new(DataType::Int64)]).unwrap();
+    let total = (N * (1 + 8)) as u64;
+    bencher
+        .counter(BytesCount::new(total))
+        .bench_local(|| conv.convert_columns(&[arr.clone()]).unwrap())
+}
+
+#[divan::bench]
+fn constant_i64_vortex_with_kernel(bencher: divan::Bencher) {
+    let arr = ConstantArray::new(42i64, N).into_array();
+    let total = (N * (1 + 8)) as u64;
+    bencher.counter(BytesCount::new(total)).bench_local(|| {
+        let mut ctx = LEGACY_SESSION.create_execution_ctx();
+        convert_columns(&[arr.clone()], &[SortField::default()], &mut ctx).unwrap()
+    })
+}
+
+#[divan::bench]
+fn constant_i64_vortex_without_kernel(bencher: divan::Bencher) {
+    let arr = ConstantArray::new(42i64, N).into_array();
+    let total = (N * (1 + 8)) as u64;
+    bencher.counter(BytesCount::new(total)).bench_local(|| {
+        let mut ctx = LEGACY_SESSION.create_execution_ctx();
+        let canonical = arr
+            .clone()
+            .execute::<Canonical>(&mut ctx)
+            .unwrap()
+            .into_array();
+        convert_columns(&[canonical], &[SortField::default()], &mut ctx).unwrap()
+    })
+}

vortex-row/src/kernels/constant.rs

@@ -2,39 +2,66 @@
 // SPDX-FileCopyrightText: Copyright the Vortex contributors
 
 //! Row-encode kernels for `ConstantArray`.
-//!
-//! Stubs in this commit return `Ok(None)` so the dispatch loop falls back to
-//! canonicalization. The real impls land in a follow-up commit.
+
+#![allow(
+    clippy::cast_possible_truncation,
+    reason = "row encoding indexes into u32-sized buffers; lengths are validated to fit in u32"
+)]
 
 use vortex_array::ArrayView;
 use vortex_array::ExecutionCtx;
 use vortex_array::arrays::Constant;
 use vortex_error::VortexResult;
 
+use crate::codec;
 use crate::encode::RowEncodeKernel;
 use crate::options::SortField;
 use crate::size::RowSizeKernel;
 
 impl RowSizeKernel for Constant {
     fn row_size_contribution(
-        _column: ArrayView<'_, Self>,
-        _field: SortField,
-        _sizes: &mut [u32],
+        column: ArrayView<'_, Self>,
+        field: SortField,
+        sizes: &mut [u32],
         _ctx: &mut ExecutionCtx,
     ) -> VortexResult<Option<()>> {
-        Ok(None)
+        let add = codec::encoded_size_for_scalar(column.scalar(), field)?;
+        for s in sizes.iter_mut().take(column.len()) {
+            *s += add;
+        }
+        Ok(Some(()))
     }
 }
 
 impl RowEncodeKernel for Constant {
     fn row_encode_into(
-        _column: ArrayView<'_, Self>,
-        _field: SortField,
-        _offsets: &[u32],
-        _cursors: &mut [u32],
-        _out: &mut [u8],
+        column: ArrayView<'_, Self>,
+        field: SortField,
+        offsets: &[u32],
+        cursors: &mut [u32],
+        out: &mut [u8],
         _ctx: &mut ExecutionCtx,
     ) -> VortexResult<Option<()>> {
-        Ok(None)
+        let bytes = codec::encode_scalar(column.scalar(), field)?;
+        let len = bytes.len();
+        let len_u32 = len as u32;
+        let n = column.len();
+        if len == 0 {
+            return Ok(Some(()));
+        }
+        // SAFETY: bytes is len bytes; offsets[i] + cursors[i] + len <= out.len() by
+        // construction of the buffer (the size pass already accounted for this column's
+        // contribution). copy_nonoverlapping elides the bounds check + slice creation
+        // that copy_from_slice would do per row.
+        unsafe {
+            let src = bytes.as_ptr();
+            let out_ptr = out.as_mut_ptr();
+            for i in 0..n {
+                let pos = (offsets[i] + cursors[i]) as usize;
+                std::ptr::copy_nonoverlapping(src, out_ptr.add(pos), len);
+                cursors[i] += len_u32;
+            }
+        }
+        Ok(Some(()))
     }
 }

vortex-row/src/tests.rs

@@ -15,6 +15,7 @@ use vortex_array::IntoArray;
 use vortex_array::LEGACY_SESSION;
 use vortex_array::VortexSessionExecute;
 use vortex_array::arrays::BoolArray;
+use vortex_array::arrays::ConstantArray;
 use vortex_array::arrays::ListViewArray;
 use vortex_array::arrays::PrimitiveArray;
 use vortex_array::arrays::VarBinViewArray;
@@ -222,6 +223,22 @@ fn nulls_first_and_last() -> VortexResult<()> {
     Ok(())
 }
 
+#[test]
+fn constant_path_matches_canonical() -> VortexResult<()> {
+    let mut ctx = LEGACY_SESSION.create_execution_ctx();
+    let nrows = 8usize;
+    let const_arr = ConstantArray::new(42i64, nrows).into_array();
+    let canonical = PrimitiveArray::from_iter(vec![42i64; nrows]).into_array();
+
+    let from_const = convert_columns(&[const_arr], &[SortField::default()], &mut ctx)?;
+    let from_canon = convert_columns(&[canonical], &[SortField::default()], &mut ctx)?;
+    assert_eq!(
+        collect_row_bytes(&from_const),
+        collect_row_bytes(&from_canon)
+    );
+    Ok(())
+}
+
 #[test]
 fn struct_sort_order() -> VortexResult<()> {
     use vortex_array::arrays::StructArray;