perf(array): append repeated and viewed lists in bulk, not a value at a time

Three callers walked a builder one list at a time where the whole run was
available up front.

`Sparse` canonicalization fills the gaps between patches by appending the fill
value once per row. A gap is a run of one value, so it goes in as a single
`ConstantArray` instead: canonicalizing a constant list array points every view
at one copy of the value, so a gap now costs the fill value's elements once
however many rows it covers. Appending a 10,000-row gap of a three-element fill
produced 30,000 elements; it now produces 3.

`ListBuilder::append_listview_array` sliced the elements array and appended the
slice once per list, which is what its `ListViewBuilder` twin stopped doing.
`ListArray` offsets can only describe contiguous, in-order lists, so flatten
the incoming views to that layout - a no-op when they are laid out that way
already - and then append the referenced elements in one go, walking only the
metadata to rebase the offsets.

Signed-off-by: Claude <noreply@anthropic.com>
Signed-off-by: Robert Kruszewski <robert@spiraldb.com>

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

Author: robert3005

encodings/sparse/src/canonical.rs

@@ -31,7 +31,6 @@ use vortex_array::builders::ArrayBuilder;
 use vortex_array::builders::DecimalBuilder;
 use vortex_array::builders::FixedSizeListBuilder;
 use vortex_array::builders::ListViewBuilder;
-use vortex_array::builders::builder_with_capacity;
 use vortex_array::dtype::DType;
 use vortex_array::dtype::DecimalDType;
 use vortex_array::dtype::DecimalType;
@@ -47,7 +46,6 @@ use vortex_array::match_each_unsigned_integer_ptype;
 use vortex_array::match_smallest_offset_type;
 use vortex_array::patches::Patches;
 use vortex_array::scalar::DecimalScalar;
-use vortex_array::scalar::ListScalar;
 use vortex_array::scalar::Scalar;
 use vortex_array::scalar::StructScalar;
 use vortex_array::validity::Validity;
@@ -181,15 +179,19 @@ fn execute_sparse_lists(
     let values = resolved.values().as_::<ListView>().into_owned();
     let fill_list = fill_value.as_list();
 
+    // Each gap between patches is appended as one constant array, whose canonical form points
+    // every view at a single copy of the fill value, so a gap costs the fill value's elements once
+    // however many rows it covers. There is at most one gap on either side of each patch.
     let n_filled = len - resolved.num_patches();
-    let total_canonical_values = values.elements().len() + fill_list.len() * n_filled;
+    let n_fill_runs = (resolved.num_patches() + 1).min(n_filled);
+    let total_canonical_values = values.elements().len() + fill_list.len() * n_fill_runs;
 
     Ok(match_each_unsigned_integer_ptype!(indices.ptype(), |I| {
         match_smallest_offset_type!(total_canonical_values, |O| {
             execute_sparse_lists_inner::<I, O>(
                 indices.as_slice(),
                 values,
-                fill_list,
+                fill_value,
                 values_dtype,
                 len,
                 total_canonical_values,
@@ -204,7 +206,7 @@ fn execute_sparse_lists(
 fn execute_sparse_lists_inner<I: IntegerPType, O: IntegerPType>(
     patch_indices: &[I],
     patch_values: ListViewArray,
-    fill_scalar: ListScalar,
+    fill_value: &Scalar,
     values_dtype: Arc<DType>,
     len: usize,
     total_canonical_values: usize,
@@ -219,7 +221,6 @@ fn execute_sparse_lists_inner<I: IntegerPType, O: IntegerPType>(
         total_canonical_values,
         len,
     );
-    let fill_elements = list_scalar_elements_array(fill_scalar);
     let patch_values_validity = patch_values
         .listview_validity()
         .execute_mask(patch_values.len(), ctx)
@@ -236,12 +237,7 @@ fn execute_sparse_lists_inner<I: IntegerPType, O: IntegerPType>(
             .to_usize()
             .vortex_expect("patch index must fit in usize");
 
-        append_list_fill(
-            &mut builder,
-            fill_elements.as_ref(),
-            sparse_idx - next_index,
-            ctx,
-        );
+        append_fill(&mut builder, fill_value, sparse_idx - next_index, ctx);
 
         if patch_valid {
             let patch_list = patch_values
@@ -257,7 +253,7 @@ fn execute_sparse_lists_inner<I: IntegerPType, O: IntegerPType>(
         next_index = sparse_idx + 1;
     }
 
-    append_list_fill(&mut builder, fill_elements.as_ref(), len - next_index, ctx);
+    append_fill(&mut builder, fill_value, len - next_index, ctx);
 
     builder.finish()
 }
@@ -272,13 +268,12 @@ fn execute_sparse_fixed_size_list(
 ) -> VortexResult<ArrayRef> {
     let indices = resolved.indices().as_::<Primitive>().into_owned();
     let values = resolved.values().as_::<FixedSizeList>().into_owned();
-    let fill_scalar = fill_value.as_list();
 
     Ok(match_each_integer_ptype!(indices.ptype(), |I| {
         execute_sparse_fixed_size_list_inner::<I>(
             indices.as_slice(),
             values,
-            fill_scalar,
+            fill_value,
             len,
             nullability,
             ctx,
@@ -296,7 +291,7 @@ fn execute_sparse_fixed_size_list(
 fn execute_sparse_fixed_size_list_inner<I: IntegerPType>(
     indices: &[I],
     values: FixedSizeListArray,
-    fill_scalar: ListScalar,
+    fill_value: &Scalar,
     array_len: usize,
     nullability: Nullability,
     ctx: &mut ExecutionCtx,
@@ -312,7 +307,6 @@ fn execute_sparse_fixed_size_list_inner<I: IntegerPType>(
         nullability,
         array_len,
     );
-    let fill_elements = list_scalar_elements_array(fill_scalar);
     let values_validity = values
         .validity()
         .vortex_expect("sparse fixed-size-list validity should be derivable")
@@ -326,12 +320,7 @@ fn execute_sparse_fixed_size_list_inner<I: IntegerPType>(
 
     for ((patch_idx, sparse_idx), patch_valid) in indices.enumerate().zip(values_validity.iter()) {
         // Fill gap before this patch with fill values.
-        append_fixed_size_list_fill(
-            &mut builder,
-            fill_elements.as_ref(),
-            sparse_idx - next_index,
-            ctx,
-        );
+        append_fill(&mut builder, fill_value, sparse_idx - next_index, ctx);
 
         // Append the patch value, handling null patches by appending defaults.
         if patch_valid {
@@ -349,60 +338,38 @@ fn execute_sparse_fixed_size_list_inner<I: IntegerPType>(
     }
 
     // Fill remaining positions after last patch.
-    append_fixed_size_list_fill(
-        &mut builder,
-        fill_elements.as_ref(),
-        array_len - next_index,
-        ctx,
-    );
+    append_fill(&mut builder, fill_value, array_len - next_index, ctx);
 
     builder.finish_into_fixed_size_list()
 }
 
-fn list_scalar_elements_array(list: ListScalar) -> Option<ArrayRef> {
-    list.elements().map(|elements| {
-        let mut builder = builder_with_capacity(list.element_dtype(), elements.len());
-        for element in elements {
-            builder
-                .append_scalar(&element)
-                .vortex_expect("list element scalar was invalid");
-        }
-        builder.finish()
-    })
-}
-
-fn append_list_fill<O: IntegerPType, S: IntegerPType>(
-    builder: &mut ListViewBuilder<O, S>,
-    fill_elements: Option<&ArrayRef>,
+/// Appends the run of `count` fill values that covers the gap before the next patch.
+///
+/// The run goes in as a single [`ConstantArray`] rather than one appended value per row. For a
+/// list dtype that is the difference between storing the fill value once per gap and once per row:
+/// canonicalizing a constant list array points every view at one copy of the value, and the
+/// builder keeps that layout.
+fn append_fill(
+    builder: &mut dyn ArrayBuilder,
+    fill_value: &Scalar,
     count: usize,
     ctx: &mut ExecutionCtx,
 ) {
-    if let Some(fill_elements) = fill_elements {
-        for _ in 0..count {
-            builder
-                .append_array_as_list(fill_elements, ctx)
-                .vortex_expect("Failed to append sparse fill value");
-        }
-    } else {
-        builder.append_nulls(count);
+    if count == 0 {
+        return;
     }
-}
 
-fn append_fixed_size_list_fill(
-    builder: &mut FixedSizeListBuilder,
-    fill_elements: Option<&ArrayRef>,
-    count: usize,
-    ctx: &mut ExecutionCtx,
-) {
-    if let Some(fill_elements) = fill_elements {
-        for _ in 0..count {
-            builder
-                .append_array_as_list(fill_elements, ctx)
-                .vortex_expect("Failed to append sparse fixed-size-list fill value");
-        }
-    } else {
+    if fill_value.is_null() {
+        // A null fill has no elements to share, and the builder can record the nulls without
+        // going through an array at all.
         builder.append_nulls(count);
+        return;
     }
+
+    ConstantArray::new(fill_value.clone(), count)
+        .into_array()
+        .append_to_builder(builder, ctx)
+        .vortex_expect("Failed to append sparse fill value");
 }
 
 fn execute_sparse_bools(
@@ -1303,6 +1270,60 @@ mod test {
         Ok(())
     }
 
+    /// Each gap between patches is appended as a single constant array, so the fill value's
+    /// elements are stored once per gap however many rows the gap covers.
+    #[test]
+    fn test_sparse_list_fill_stores_one_copy_per_gap() -> VortexResult<()> {
+        let mut ctx = SESSION.create_execution_ctx();
+
+        // Two single-element patch lists: [1] and [2].
+        let lists = unsafe {
+            ListViewArray::new_unchecked(
+                buffer![1i32, 2].into_array(),
+                buffer![0u32, 1].into_array(),
+                buffer![1u32, 1].into_array(),
+                Validity::AllValid,
+            )
+            .with_zero_copy_to_list(true)
+        }
+        .into_array();
+
+        // Patches at 10 and 20 of 10,000 rows, so the fill covers three gaps.
+        let indices = buffer![10u32, 20].into_array();
+        let fill = vec![7i32, 8, 9];
+        let sparse =
+            Sparse::try_new(indices, lists, 10_000, Scalar::from(Some(fill.clone())))?.into_array();
+
+        let actual = sparse.execute::<ListViewArray>(&mut ctx)?;
+        assert_eq!(actual.len(), 10_000);
+        assert_eq!(
+            actual.elements().len(),
+            2 + 3 * fill.len(),
+            "the fill value should be stored once per gap, not once per row",
+        );
+
+        let fill_elements = PrimitiveArray::from_iter(fill);
+        for index in [0, 9, 11, 19, 21, 9_999] {
+            assert_arrays_eq!(
+                actual.list_elements_at(index).vortex_expect("fill list"),
+                fill_elements,
+                &mut ctx
+            );
+        }
+        assert_arrays_eq!(
+            actual.list_elements_at(10).vortex_expect("patch list"),
+            PrimitiveArray::from_iter([1i32]),
+            &mut ctx
+        );
+        assert_arrays_eq!(
+            actual.list_elements_at(20).vortex_expect("patch list"),
+            PrimitiveArray::from_iter([2i32]),
+            &mut ctx
+        );
+
+        Ok(())
+    }
+
     #[test]
     fn test_sparse_binary_varbin_null_fill() {
         let mut ctx = SESSION.create_execution_ctx();