fix[vortex-array]: handle lazily sliced REE to_arrow (#7011)

## Summary

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.

<!--
Thank you for submitting a pull request! We appreciate your time and
effort.

Please make sure to provide enough information so that we can review
your pull
request. The Summary and Testing sections below contain guidance on what
to
include.
-->

<!--
If this PR is related to a tracked effort, please link to the relevant
issue
here (e.g., `Closes: #123`). Otherwise, feel free to ignore / delete
this.

In this section, please:

1. Explain the rationale for this change.
2. Summarize the changes included in this PR.

A general rule of thumb is that larger PRs should have larger summaries.
If
there are a lot of changes, please help us review the code by explaining
what
was changed and why.

If there is an issue or discussion attached, there is no need to
duplicate all
the details, but clarity is always preferred over brevity.
-->

<!--
## API Changes

Uncomment this section if there are any user-facing changes.

Consider whether the change affects users in one of the following ways:

1. Breaks public APIs in some way.
2. Changes the underlying behavior of one of the engine integrations.
3. Should some documentation be updated to reflect this change?

If a public API is changed in a breaking manner, make sure to add the
appropriate label. You can run `./scripts/public-api.sh` locally to see
if there
are any public API changes (and this also runs in our CI).
-->

## Testing

<!--
Please describe how this change was tested. Here are some common
categories for
testing in Vortex:

1. Verifying existing behavior is maintained.
2. Verifying new behavior and functionality works correctly.
3. Serialization compatibility (backwards and forwards) should be
maintained or
   explicitly broken.
-->
Enhanced tests to reproduce e2e arrow execution error and added some
unit tests to build_run_array

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.
+    // Run ends are strictly increasing, so we can binary search.
+    let num_runs = (ends_prim
+        .values()
+        .partition_point(|&e| e - offset_native < length_native)
+        + 1)
+    .min(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(())
+    }
 }