fix(arrow-cast): respect sliced list offsets in fixed-size list casts

arrow-cast/src/cast/list.rs

@@ -155,18 +155,10 @@ where
     // Nulls in FixedSizeListArray take up space and so we must pad the values
     let values = array.values().to_data();
     let mut mutable = MutableArrayData::new(vec![&values], nullable, cap);
-    // The end position in values of the last incorrectly-sized list slice
-    let mut last_pos = 0;
-
-    // Need to flag when previous vector(s) are empty/None to distinguish from 'All slices were correct length' cases.
-    let is_prev_empty = if array.offsets().len() < 2 {
-        false
-    } else {
-        let first_offset = array.offsets()[0].as_usize();
-        let second_offset = array.offsets()[1].as_usize();
-
-        first_offset == 0 && second_offset == 0
-    };
+    let first_pos = array.offsets()[0].as_usize();
+    // The end position in values of the last incorrectly-sized list slice,
+    // or None if no padding has been needed (including for empty slices).
+    let mut last_pos = None;
 
     for (idx, w) in array.offsets().windows(2).enumerate() {
         let start_pos = w[0].as_usize();
@@ -175,10 +167,11 @@ where
 
         if len != size as usize {
             if cast_options.safe || array.is_null(idx) {
-                if last_pos != start_pos {
+                let copy_start = last_pos.unwrap_or(first_pos);
+                if copy_start != start_pos {
                     // Extend with valid slices
                     mutable
-                        .try_extend(0, last_pos, start_pos)
+                        .try_extend(0, copy_start, start_pos)
                         .map_err(|e| ArrowError::CastError(e.to_string()))?;
                 }
                 // Pad this slice with nulls
@@ -187,7 +180,7 @@ where
                     .map_err(|e| ArrowError::CastError(e.to_string()))?;
                 null_builder.set_bit(idx, false);
                 // Set last_pos to the end of this slice's values
-                last_pos = end_pos
+                last_pos = Some(end_pos)
             } else {
                 return Err(ArrowError::CastError(format!(
                     "Cannot cast to FixedSizeList({size}): value at index {idx} has length {len}",
@@ -197,8 +190,8 @@ where
     }
 
     let values = match last_pos {
-        0 if !is_prev_empty => array.values().slice(0, cap), // All slices were the correct length
-        _ => {
+        None => array.values().slice(first_pos, cap), // All slices were the correct length
+        Some(last_pos) => {
             if mutable.len() != cap {
                 // Remaining slices were all correct length
                 let remaining = cap - mutable.len();

arrow-cast/src/cast/mod.rs

@@ -9860,6 +9860,179 @@ mod tests {
         ));
         let fsl = cast(list.as_ref(), expected.data_type()).unwrap();
         assert_eq!(&expected, &fsl);
+
+        // Direct non-zero offsets must retain the row count and validity.
+        let field = Arc::new(Field::new_list_field(DataType::Int32, true));
+        let target = DataType::FixedSizeList(field.clone(), 0);
+        let strict = CastOptions {
+            safe: false,
+            ..Default::default()
+        };
+        for nulls in [None, Some(NullBuffer::from(vec![true, false]))] {
+            let values = Arc::new(Int32Array::from(vec![1, 2, 3]));
+            let inputs: [ArrayRef; 2] = [
+                Arc::new(ListArray::new(
+                    field.clone(),
+                    OffsetBuffer::new(vec![3; 3].into()),
+                    values.clone(),
+                    nulls.clone(),
+                )),
+                Arc::new(LargeListArray::new(
+                    field.clone(),
+                    OffsetBuffer::new(vec![3; 3].into()),
+                    values,
+                    nulls.clone(),
+                )),
+            ];
+            for input in inputs {
+                let actual = cast_with_options(input.as_ref(), &target, &strict).unwrap();
+                assert_eq!(actual.len(), 2);
+                assert_eq!(actual.data_type(), &target);
+                assert_eq!(actual.nulls(), nulls.as_ref());
+                assert_eq!(actual.as_fixed_size_list().values().len(), 0);
+            }
+        }
+    }
+
+    #[test]
+    fn test_issue_10975_sliced_list_to_fsl() {
+        fn test<O: OffsetSizeTrait>() {
+            let input = GenericListArray::<O>::from_iter_primitive::<Int32Type, _, _>([
+                Some(vec![Some(1), Some(2)]),
+                Some(vec![Some(3), Some(4)]),
+                Some(vec![Some(5), Some(6)]),
+            ]);
+            let expected = FixedSizeListArray::from_iter_primitive::<Int32Type, _, _>(
+                [Some([Some(3), Some(4)]), Some([Some(5), Some(6)])],
+                2,
+            );
+            for safe in [true, false] {
+                let options = CastOptions {
+                    safe,
+                    ..Default::default()
+                };
+                let actual =
+                    cast_with_options(&input.slice(1, 2), expected.data_type(), &options).unwrap();
+                assert_eq!(actual.as_ref(), &expected as &dyn Array);
+            }
+        }
+        test::<i32>();
+        test::<i64>();
+    }
+
+    #[test]
+    fn test_issue_10975_sliced_list_to_fsl_subcast() {
+        fn test<O: OffsetSizeTrait>() {
+            // A differently sized prefix and invalid excluded children must not
+            // affect selection or the recursive child cast.
+            let input = GenericListArray::<O>::from_iter_primitive::<Int32Type, _, _>([
+                Some(vec![Some(i32::MAX); 3]),
+                Some(vec![Some(3), None]),
+                Some(vec![Some(5), Some(6)]),
+                Some(vec![Some(i32::MAX); 2]),
+            ]);
+            let selected = input.slice(1, 3).slice(0, 2);
+            let expected = FixedSizeListArray::from_iter_primitive::<Int32Type, _, _>(
+                [Some([Some(3), None]), Some([Some(5), Some(6)])],
+                2,
+            );
+            for safe in [true, false] {
+                let options = CastOptions {
+                    safe,
+                    ..Default::default()
+                };
+                for child_type in [DataType::Int32, DataType::Int64, DataType::Int16] {
+                    let target = DataType::FixedSizeList(
+                        Arc::new(Field::new_list_field(child_type, true)),
+                        2,
+                    );
+                    let actual = cast_with_options(&selected, &target, &options).unwrap();
+                    let expected = cast_with_options(&expected, &target, &options).unwrap();
+                    assert_eq!(actual.as_ref(), expected.as_ref());
+                    assert_eq!(actual.as_fixed_size_list().values().len(), 4);
+                }
+            }
+        }
+        test::<i32>();
+        test::<i64>();
+    }
+
+    #[test]
+    fn test_issue_10975_sliced_list_to_fsl_padding() {
+        fn test<O: OffsetSizeTrait>() {
+            let field = Arc::new(Field::new_list_field(DataType::Int32, true));
+            let lengths = [3, 0, 0, 2, 1, 3, 2, 2, 0, 2];
+            let values = Int32Array::from_iter_values(0..16).slice(1, 15);
+            let input = GenericListArray::<O>::new(
+                field.clone(),
+                OffsetBuffer::from_lengths(lengths),
+                Arc::new(values),
+                Some(NullBuffer::from(vec![
+                    false, false, false, true, false, false, true, false, false, true,
+                ])),
+            );
+            let target = DataType::FixedSizeList(field, 2);
+            for safe in [true, false] {
+                let options = CastOptions {
+                    safe,
+                    ..Default::default()
+                };
+                let full = cast_with_options(&input, &target, &options).unwrap();
+                for (start, len) in [
+                    (1, 8), // Leading/consecutive empty nulls, short/long and exact-width nulls.
+                    (1, 2), // Only consecutive empty nulls.
+                    (3, 4), // Short and long nulls between valid rows.
+                    (6, 2), // Valid row and exact-width null: no padding needed.
+                    (8, 1), // Only one empty null at a non-zero child offset.
+                ] {
+                    let selected = input.slice(start, len);
+                    let actual = cast_with_options(&selected, &target, &options).unwrap();
+                    assert_eq!(actual.as_ref(), full.slice(start, len).as_ref());
+                    assert_eq!(actual.as_fixed_size_list().values().len(), len * 2);
+                }
+            }
+        }
+        test::<i32>();
+        test::<i64>();
+    }
+
+    #[test]
+    fn test_issue_10975_sliced_list_to_fsl_safety() {
+        fn test<O: OffsetSizeTrait>() {
+            let input = GenericListArray::<O>::from_iter_primitive::<Int32Type, _, _>([
+                Some(vec![Some(99); 3]),
+                Some(vec![Some(1), Some(2)]),
+                Some(vec![]),
+                Some(vec![Some(3)]),
+                Some(vec![Some(4); 3]),
+                Some(vec![Some(5), Some(6)]),
+            ]);
+            let expected = FixedSizeListArray::from_iter_primitive::<Int32Type, _, _>(
+                [
+                    Some([Some(1), Some(2)]),
+                    None,
+                    None,
+                    None,
+                    Some([Some(5), Some(6)]),
+                ],
+                2,
+            );
+            let actual = cast(&input.slice(1, 5), expected.data_type()).unwrap();
+            assert_eq!(actual.as_ref(), &expected as &dyn Array);
+            assert_eq!(actual.as_fixed_size_list().values().len(), 10);
+            let strict = CastOptions {
+                safe: false,
+                ..Default::default()
+            };
+            let error =
+                cast_with_options(&input.slice(1, 5), expected.data_type(), &strict).unwrap_err();
+            assert_eq!(
+                error.to_string(),
+                "Cast error: Cannot cast to FixedSizeList(2): value at index 1 has length 0"
+            );
+        }
+        test::<i32>();
+        test::<i64>();
     }
 
     #[test]
@@ -10204,29 +10377,26 @@ mod tests {
         let target_type = DataType::FixedSizeList(inner_field.clone(), 3);
         let expected = new_empty_array(&target_type);
 
-        // list
-        let array = new_empty_array(&DataType::List(inner_field.clone()));
-        assert!(can_cast_types(array.data_type(), &target_type));
-        let actual = cast(array.as_ref(), &target_type).unwrap();
-        assert_eq!(expected.as_ref(), actual.as_ref());
-
-        // largelist
-        let array = new_empty_array(&DataType::LargeList(inner_field.clone()));
-        assert!(can_cast_types(array.data_type(), &target_type));
-        let actual = cast(array.as_ref(), &target_type).unwrap();
-        assert_eq!(expected.as_ref(), actual.as_ref());
-
-        // listview
-        let array = new_empty_array(&DataType::ListView(inner_field.clone()));
-        assert!(can_cast_types(array.data_type(), &target_type));
-        let actual = cast(array.as_ref(), &target_type).unwrap();
-        assert_eq!(expected.as_ref(), actual.as_ref());
-
-        // largelistview
-        let array = new_empty_array(&DataType::LargeListView(inner_field.clone()));
-        assert!(can_cast_types(array.data_type(), &target_type));
-        let actual = cast(array.as_ref(), &target_type).unwrap();
-        assert_eq!(expected.as_ref(), actual.as_ref());
+        let cases = [
+            new_empty_array(&DataType::List(inner_field.clone())),
+            new_empty_array(&DataType::LargeList(inner_field.clone())),
+            new_empty_array(&DataType::ListView(inner_field.clone())),
+            new_empty_array(&DataType::LargeListView(inner_field.clone())),
+            // Empty slices with non-zero child offsets (issue #10975).
+            make_list_array().slice(2, 0),
+            make_large_list_array().slice(2, 0),
+        ];
+        for array in cases {
+            assert!(can_cast_types(array.data_type(), &target_type));
+            for safe in [true, false] {
+                let options = CastOptions {
+                    safe,
+                    ..Default::default()
+                };
+                let actual = cast_with_options(array.as_ref(), &target_type, &options).unwrap();
+                assert_eq!(expected.as_ref(), actual.as_ref());
+            }
+        }
     }
 
     fn make_list_array() -> ArrayRef {