test: cover LargeList and scan/take path for #6580

- Refactor the sliced-list merge_with_schema test into a generic helper
  and add a LargeList case.
- Add a scan().filter().project([list_struct_col]) regression in scanner.rs
  that forces the panicking TakeExec -> merge_with_schema path via
  MaterializationStyle::AllEarlyExcept.

Author: kaan-simbe

rust/lance-arrow/src/lib.rs

@@ -2319,56 +2319,71 @@ mod tests {
         assert_eq!(merged_array.len(), 2);
     }
 
+    #[test]
+    fn test_merge_with_schema_sliced_list_struct() {
+        test_merge_with_schema_sliced_list_struct_generic::<i32>();
+    }
+
+    #[test]
+    fn test_merge_with_schema_sliced_large_list_struct() {
+        test_merge_with_schema_sliced_list_struct_generic::<i64>();
+    }
+
     // Regression for #6580: merge_with_schema panicked when the left list was a
     // sliced view whose offsets did not start at zero (common after a filtered
     // scan). Cloning those offsets alongside `trimmed_values` produced offsets
-    // larger than the trimmed child, panicking in `ListArray::new`.
-    #[test]
-    fn test_merge_with_schema_sliced_list_struct() {
+    // larger than the trimmed child, panicking in `(Large)ListArray::new`.
+    fn test_merge_with_schema_sliced_list_struct_generic<O: OffsetSizeTrait>() {
+        let make_list_dtype = |item_field: Arc<Field>| {
+            if O::IS_LARGE {
+                DataType::LargeList(item_field)
+            } else {
+                DataType::List(item_field)
+            }
+        };
+
         // Build a List<Struct> with two rows of 5 items each, then slice away
         // the first row so the remaining list's offsets start at 5, not 0.
-        let struct_fields = Fields::from(vec![Field::new("a", DataType::Int32, true)]);
-        let values = Arc::new(StructArray::new(
-            struct_fields.clone(),
+        let struct_fields_a = Fields::from(vec![Field::new("a", DataType::Int32, true)]);
+        let left_values = Arc::new(StructArray::new(
+            struct_fields_a.clone(),
             vec![Arc::new(Int32Array::from_iter_values(0..10)) as ArrayRef],
             None,
         ));
-        let full_list = ListArray::new(
-            Arc::new(Field::new("item", DataType::Struct(struct_fields), true)),
-            OffsetBuffer::from_lengths([5, 5]),
-            values,
+        let full_list = GenericListArray::<O>::new(
+            Arc::new(Field::new("item", DataType::Struct(struct_fields_a), true)),
+            OffsetBuffer::<O>::from_lengths([5, 5]),
+            left_values,
             None,
         );
         let sliced_left = full_list.slice(1, 1);
-        assert_eq!(sliced_left.offsets()[0], 5);
-        assert_eq!(sliced_left.offsets()[1], 10);
+        assert_eq!(sliced_left.offsets()[0].as_usize(), 5);
+        assert_eq!(sliced_left.offsets()[1].as_usize(), 10);
 
-        let right_struct = Arc::new(StructArray::new(
-            Fields::from(vec![Field::new("b", DataType::Int32, true)]),
+        let struct_fields_b = Fields::from(vec![Field::new("b", DataType::Int32, true)]);
+        let right_values = Arc::new(StructArray::new(
+            struct_fields_b.clone(),
             vec![Arc::new(Int32Array::from_iter_values(100..105)) as ArrayRef],
             None,
         ));
-        let right_list = ListArray::new(
-            Arc::new(Field::new(
-                "item",
-                DataType::Struct(right_struct.fields().clone()),
-                true,
-            )),
-            OffsetBuffer::from_lengths([5]),
-            right_struct,
+        let right_list = GenericListArray::<O>::new(
+            Arc::new(Field::new("item", DataType::Struct(struct_fields_b), true)),
+            OffsetBuffer::<O>::from_lengths([5]),
+            right_values,
             None,
         );
 
+        let target_item_field = Arc::new(Field::new(
+            "item",
+            DataType::Struct(Fields::from(vec![
+                Field::new("a", DataType::Int32, true),
+                Field::new("b", DataType::Int32, true),
+            ])),
+            true,
+        ));
         let target_fields = Fields::from(vec![Field::new(
             "items",
-            DataType::List(Arc::new(Field::new(
-                "item",
-                DataType::Struct(Fields::from(vec![
-                    Field::new("a", DataType::Int32, true),
-                    Field::new("b", DataType::Int32, true),
-                ])),
-                true,
-            ))),
+            make_list_dtype(target_item_field),
             true,
         )]);
 
@@ -2399,10 +2414,10 @@ mod tests {
             .column_by_name("items")
             .unwrap()
             .as_any()
-            .downcast_ref::<ListArray>()
+            .downcast_ref::<GenericListArray<O>>()
             .unwrap();
         assert_eq!(merged_list.len(), 1);
-        assert_eq!(merged_list.value_length(0), 5);
+        assert_eq!(merged_list.value_length(0).as_usize(), 5);
         let merged_struct = merged_list.values().as_struct();
         assert_eq!(merged_struct.num_columns(), 2);
         let a = merged_struct

rust/lance/src/dataset/scanner.rs

@@ -5373,6 +5373,131 @@ mod test {
         Ok(())
     }
 
+    // Regression for #6580: a scan with `filter` + `project` of a
+    // `List<Struct>` column used to panic in `merge_with_schema` (called from
+    // `TakeStream::map_batch`) because the filtered batch arrived as a sliced
+    // view of a larger batch and the cloned list offsets did not start at
+    // zero. The trigger requires (a) a `List<Struct>` projection where the
+    // struct is split across `filtered_read` and `TakeExec` and (b) a
+    // sparse-tail selectivity pattern so the trailing filter result lands deep
+    // inside the values buffer of its source batch.
+    #[rstest]
+    #[tokio::test]
+    async fn test_filter_project_list_struct_sparse_tail(
+        // The panic is specific to v2.x storage; the legacy reader takes a
+        // different code path. V2_0 and V2_2 are the versions called out in
+        // the original report.
+        #[values(
+            LanceFileVersion::V2_0,
+            LanceFileVersion::Stable,
+            LanceFileVersion::V2_2
+        )]
+        data_storage_version: LanceFileVersion,
+    ) {
+        use arrow_array::{ListArray, UInt16Array};
+        use arrow_buffer::{OffsetBuffer, ScalarBuffer};
+
+        let struct_fields = Fields::from(vec![
+            Arc::new(ArrowField::new("a", DataType::Int32, true)),
+            Arc::new(ArrowField::new("b", DataType::Int32, true)),
+        ]);
+        let item_field = Arc::new(ArrowField::new(
+            "item",
+            DataType::Struct(struct_fields.clone()),
+            true,
+        ));
+        let schema = Arc::new(ArrowSchema::new(vec![
+            ArrowField::new("id", DataType::Int32, false),
+            ArrowField::new("grp", DataType::UInt16, false),
+            ArrowField::new("items", DataType::List(item_field.clone()), false),
+        ]));
+
+        let make_batch = |start: i32, n: usize, group: u16| -> RecordBatch {
+            let ids = Int32Array::from_iter_values(start..start + n as i32);
+            let groups = UInt16Array::from(vec![group; n]);
+
+            let mut offsets = Vec::with_capacity(n + 1);
+            let mut a_vals: Vec<i32> = Vec::new();
+            let mut b_vals: Vec<i32> = Vec::new();
+            offsets.push(0i32);
+            for i in 0..n {
+                // Variable-length lists (1..=18) so offsets don't land on
+                // batch-row boundaries.
+                let len = 1 + (i % 18);
+                for j in 0..len {
+                    a_vals.push(j as i32);
+                    b_vals.push(-(j as i32));
+                }
+                offsets.push(a_vals.len() as i32);
+            }
+            let struct_arr = Arc::new(StructArray::new(
+                struct_fields.clone(),
+                vec![
+                    Arc::new(Int32Array::from(a_vals)) as ArrayRef,
+                    Arc::new(Int32Array::from(b_vals)) as ArrayRef,
+                ],
+                None,
+            ));
+            let items = ListArray::new(
+                item_field.clone(),
+                OffsetBuffer::new(ScalarBuffer::from(offsets)),
+                struct_arr,
+                None,
+            );
+            RecordBatch::try_new(
+                schema.clone(),
+                vec![
+                    Arc::new(ids) as ArrayRef,
+                    Arc::new(groups) as ArrayRef,
+                    Arc::new(items) as ArrayRef,
+                ],
+            )
+            .unwrap()
+        };
+
+        // Sparse-tail selectivity (matching the original report's shape at a
+        // smaller scale): a large leading block of matches, a large gap of
+        // non-matches, then a small trailing match. Single fragment.
+        let batches = vec![
+            make_batch(0, 100_000, 7),
+            make_batch(100_000, 400_000, 1),
+            make_batch(500_000, 7_300, 7),
+        ];
+
+        let reader = RecordBatchIterator::new(batches.into_iter().map(Ok), schema.clone());
+        let dataset = Dataset::write(
+            reader,
+            "memory://",
+            Some(WriteParams {
+                max_rows_per_file: 1_000_000,
+                data_storage_version: Some(data_storage_version),
+                ..Default::default()
+            }),
+        )
+        .await
+        .unwrap();
+
+        // Force a column split inside the `items` struct by marking `items.b`
+        // as a late-materialized field: `filtered_read` returns the batch with
+        // `items.a`, and `TakeExec` adds `items.b`. `merge_with_schema` then
+        // takes its `List<Struct>` branch, which is where the panic was.
+        let items_b_field_id = dataset
+            .schema()
+            .field("items")
+            .unwrap()
+            .child("item")
+            .unwrap()
+            .child("b")
+            .unwrap()
+            .id as u32;
+        let mut scan = dataset.scan();
+        scan.filter("grp = 7").unwrap();
+        scan.project(&["id", "items"]).unwrap();
+        scan.materialization_style(MaterializationStyle::AllEarlyExcept(vec![items_b_field_id]));
+        let result = scan.try_into_batch().await.unwrap();
+        assert_eq!(result.num_rows(), 107_300);
+    }
+
     #[tokio::test]
     async fn test_scan_regexp_match_and_non_empty_captions() {
         // Build a small dataset with three Utf8 columns and verify the full