fix[vortex-array]: handle lazily sliced REE to_arrow

This previous code was short-circuiting with a zero offset, with the assumption
that ends were already correctly physically sliced. However, slicing is a lazy
operation using metadata, so this commit fixes a couple of edge cases by
executing the fast path only if the ends are correcly sliced. Otherwise, ends
are trimmed and transformed with the offset in for correct arrow array
construction.

Signed-off-by: Alfonso Subiotto Marques <alfonso.subiotto@polarsignals.com>

Author: asubiotto

encodings/runend/src/arrow.rs

@@ -64,6 +64,7 @@ mod tests {
     use arrow_array::types::Int64Type;
     use arrow_schema::DataType;
     use arrow_schema::Field;
+    use rstest::rstest;
     use vortex_array::IntoArray as _;
     use vortex_array::VortexSessionExecute as _;
     use vortex_array::arrays::PrimitiveArray;
@@ -251,19 +252,37 @@ mod tests {
         Ok(())
     }
 
-    #[test]
-    fn test_sliced_runend_to_arrow_ree() -> VortexResult<()> {
-        let array = RunEndArray::encode(
-            PrimitiveArray::from_iter(vec![10i32, 10, 20, 20, 20, 30, 30]).into_array(),
-        )?;
-        // Slicing from index 1 produces a non-zero offset in the RunEndArray.
-        let sliced = array.into_array().slice(1..5)?;
+    /// Slicing a RunEndArray and converting to Arrow REE must produce
+    /// correctly trimmed and adjusted run ends for both zero and non-zero offsets.
+    #[rstest]
+    #[case::nonzero_offset(
+        &[10i32, 10, 20, 20, 20, 30, 30],
+        1..5usize,
+        &[1i32, 4],
+        &[10i32, 20],
+    )]
+    #[case::zero_offset_excess_runs(
+        &[10i32, 10, 10, 20, 20, 30, 30, 30, 30, 30],
+        0..4usize,
+        &[3i32, 4],
+        &[10i32, 20],
+    )]
+    fn sliced_runend_to_arrow_ree(
+        #[case] input: &[i32],
+        #[case] slice_range: std::ops::Range<usize>,
+        #[case] expected_ends: &[i32],
+        #[case] expected_values: &[i32],
+    ) -> VortexResult<()> {
+        let array =
+            RunEndArray::encode(PrimitiveArray::from_iter(input.iter().copied()).into_array())?;
+        let sliced = array.into_array().slice(slice_range.clone())?;
         let target = ree_type(DataType::Int32, DataType::Int32);
         let result = execute(sliced, &target)?;
 
+        assert_eq!(result.len(), slice_range.len());
         let expected = RunArray::<Int32Type>::try_new(
-            &Int32Array::from(vec![1, 4]),
-            &Int32Array::from(vec![10, 20]),
+            &Int32Array::from(expected_ends.to_vec()),
+            &Int32Array::from(expected_values.to_vec()),
         )?;
         assert_eq!(result.as_ref(), &expected);
         Ok(())

vortex-array/src/arrow/executor/run_end.rs

@@ -118,22 +118,35 @@ fn build_run_array<R: RunEndIndexType>(
 where
     R::Native: std::ops::Sub<Output = R::Native> + Ord,
 {
-    if offset == 0 {
-        return Ok(
-            Arc::new(RunArray::<R>::try_new(ends.as_primitive::<R>(), values)?) as ArrowArrayRef,
-        );
-    }
-
     let offset_native = R::Native::from_usize(offset)
         .ok_or_else(|| vortex_err!("Offset {offset} exceeds run-end index capacity"))?;
     let length_native = R::Native::from_usize(length)
         .ok_or_else(|| vortex_err!("Length {length} exceeds run-end index capacity"))?;
 
-    let adjusted = ends
+    let ends_prim = ends.as_primitive::<R>();
+    if offset == 0 && ends_prim.values().last() == Some(&length_native) {
+        // Fast path: no trimming or adjustment needed.
+        return Ok(Arc::new(RunArray::<R>::try_new(ends_prim, values)?) as ArrowArrayRef);
+    }
+
+    // Trim to only include runs covering the [offset, offset+length) range.
+    // Runs beyond this would produce duplicate adjusted ends, violating
+    // Arrow's strict-ordering requirement for RunArray.
+    let num_runs = ends_prim
+        .values()
+        .iter()
+        .position(|&e| e - offset_native >= length_native)
+        .map(|i| i + 1)
+        .unwrap_or(ends_prim.len());
+
+    let trimmed_ends = ends.slice(0, num_runs);
+    let trimmed_values = values.slice(0, num_runs);
+
+    let adjusted = trimmed_ends
         .as_primitive::<R>()
         .unary(|end| (end - offset_native).min(length_native));
 
-    Ok(Arc::new(RunArray::<R>::try_new(&adjusted, values)?) as ArrowArrayRef)
+    Ok(Arc::new(RunArray::<R>::try_new(&adjusted, &trimmed_values)?) as ArrowArrayRef)
 }
 
 /// Convert a constant array to a run-end encoded array with a single run.
@@ -275,4 +288,34 @@ mod tests {
         assert_eq!(result.as_ref(), &expected);
         Ok(())
     }
+
+    /// Regression: build_run_array must trim excess trailing runs and
+    /// respect the `length` parameter. This happens when a vortex
+    /// RunEndArray is sliced to fewer rows than the physical run_ends cover.
+    #[rstest]
+    #[case::offset_zero(0, 5, &[3, 5], &[100, 200])]
+    #[case::nonzero_offset(2, 3, &[1, 3], &[100, 200])]
+    #[case::all_runs_needed_but_last_exceeds(0, 8, &[3, 5, 8], &[100, 200, 300])]
+    fn build_run_array_trims_excess_runs(
+        #[case] offset: usize,
+        #[case] length: usize,
+        #[case] expected_ends: &[i32],
+        #[case] expected_values: &[i64],
+    ) -> VortexResult<()> {
+        // 3 runs covering 10 rows: [0..3), [3..5), [5..10)
+        let ends: arrow_array::ArrayRef = Arc::new(Int32Array::from(vec![3i32, 5, 10]));
+        let values: arrow_array::ArrayRef = Arc::new(Int64Array::from(vec![100i64, 200, 300]));
+
+        let result = super::build_run_array::<Int32Type>(&ends, &values, offset, length)?;
+        assert_eq!(result.len(), length);
+
+        let ree = result
+            .as_any()
+            .downcast_ref::<RunArray<Int32Type>>()
+            .unwrap();
+        assert_eq!(ree.run_ends().values(), expected_ends);
+        let values = ree.values().as_any().downcast_ref::<Int64Array>().unwrap();
+        assert_eq!(values.values(), expected_values);
+        Ok(())
+    }
 }