lance_encoding/encodings/logical/
list.rs

1// SPDX-License-Identifier: Apache-2.0
2// SPDX-FileCopyrightText: Copyright The Lance Authors
3
4use std::{ops::Range, sync::Arc};
5
6use arrow_array::{Array, ArrayRef, LargeListArray, ListArray, cast::AsArray, make_array};
7use arrow_schema::DataType;
8use futures::future::BoxFuture;
9use lance_arrow::deepcopy::deep_copy_nulls;
10use lance_arrow::list::ListArrayExt;
11use lance_core::Result;
12
13use crate::{
14    decoder::{
15        DecodedArray, FilterExpression, ScheduledScanLine, SchedulerContext,
16        StructuralDecodeArrayTask, StructuralFieldDecoder, StructuralFieldScheduler,
17        StructuralSchedulingJob,
18    },
19    encoder::{EncodeTask, FieldEncoder, OutOfLineBuffers},
20    repdef::RepDefBuilder,
21};
22
23/// A structural encoder for list fields
24///
25/// The list's offsets are added to the rep/def builder
26/// and the list array's values are passed to the child encoder
27///
28/// The values will have any garbage values removed and will be trimmed
29/// to only include the values that are actually used.
30pub struct ListStructuralEncoder {
31    keep_original_array: bool,
32    child: Box<dyn FieldEncoder>,
33}
34
35impl ListStructuralEncoder {
36    pub fn new(keep_original_array: bool, child: Box<dyn FieldEncoder>) -> Self {
37        Self {
38            keep_original_array,
39            child,
40        }
41    }
42}
43
44impl FieldEncoder for ListStructuralEncoder {
45    fn maybe_encode(
46        &mut self,
47        array: ArrayRef,
48        external_buffers: &mut OutOfLineBuffers,
49        mut repdef: RepDefBuilder,
50        row_number: u64,
51        num_rows: u64,
52    ) -> Result<Vec<EncodeTask>> {
53        let values = if let Some(list_arr) = array.as_list_opt::<i32>() {
54            let has_garbage_values = if self.keep_original_array {
55                repdef.add_offsets(list_arr.offsets().clone(), array.nulls().cloned())
56            } else {
57                // there is no need to deep copy offsets, because offset buffers will be cast to a common type (i64).
58                repdef.add_offsets(list_arr.offsets().clone(), deep_copy_nulls(array.nulls()))
59            };
60            if has_garbage_values {
61                list_arr.filter_garbage_nulls().trimmed_values()
62            } else {
63                list_arr.trimmed_values()
64            }
65        } else if let Some(list_arr) = array.as_list_opt::<i64>() {
66            let has_garbage_values = if self.keep_original_array {
67                repdef.add_offsets(list_arr.offsets().clone(), array.nulls().cloned())
68            } else {
69                repdef.add_offsets(list_arr.offsets().clone(), deep_copy_nulls(array.nulls()))
70            };
71            if has_garbage_values {
72                list_arr.filter_garbage_nulls().trimmed_values()
73            } else {
74                list_arr.trimmed_values()
75            }
76        } else {
77            panic!("List encoder used for non-list data")
78        };
79        self.child
80            .maybe_encode(values, external_buffers, repdef, row_number, num_rows)
81    }
82
83    fn flush(&mut self, external_buffers: &mut OutOfLineBuffers) -> Result<Vec<EncodeTask>> {
84        self.child.flush(external_buffers)
85    }
86
87    fn num_columns(&self) -> u32 {
88        self.child.num_columns()
89    }
90
91    fn finish(
92        &mut self,
93        external_buffers: &mut OutOfLineBuffers,
94    ) -> BoxFuture<'_, Result<Vec<crate::encoder::EncodedColumn>>> {
95        self.child.finish(external_buffers)
96    }
97}
98
99#[derive(Debug)]
100pub struct StructuralListScheduler {
101    child: Box<dyn StructuralFieldScheduler>,
102}
103
104impl StructuralListScheduler {
105    pub fn new(child: Box<dyn StructuralFieldScheduler>) -> Self {
106        Self { child }
107    }
108}
109
110impl StructuralFieldScheduler for StructuralListScheduler {
111    fn schedule_ranges<'a>(
112        &'a self,
113        ranges: &[Range<u64>],
114        filter: &FilterExpression,
115    ) -> Result<Box<dyn StructuralSchedulingJob + 'a>> {
116        let child = self.child.schedule_ranges(ranges, filter)?;
117
118        Ok(Box::new(StructuralListSchedulingJob::new(child)))
119    }
120
121    fn initialize<'a>(
122        &'a mut self,
123        filter: &'a FilterExpression,
124        context: &'a SchedulerContext,
125    ) -> BoxFuture<'a, Result<()>> {
126        self.child.initialize(filter, context)
127    }
128}
129
130/// Scheduling job for list data
131///
132/// Scheduling is handled by the primitive encoder and nothing special
133/// happens here.
134#[derive(Debug)]
135struct StructuralListSchedulingJob<'a> {
136    child: Box<dyn StructuralSchedulingJob + 'a>,
137}
138
139impl<'a> StructuralListSchedulingJob<'a> {
140    fn new(child: Box<dyn StructuralSchedulingJob + 'a>) -> Self {
141        Self { child }
142    }
143}
144
145impl StructuralSchedulingJob for StructuralListSchedulingJob<'_> {
146    fn schedule_next(&mut self, context: &mut SchedulerContext) -> Result<Vec<ScheduledScanLine>> {
147        self.child.schedule_next(context)
148    }
149}
150
151#[derive(Debug)]
152pub struct StructuralListDecoder {
153    child: Box<dyn StructuralFieldDecoder>,
154    data_type: DataType,
155}
156
157impl StructuralListDecoder {
158    pub fn new(child: Box<dyn StructuralFieldDecoder>, data_type: DataType) -> Self {
159        Self { child, data_type }
160    }
161}
162
163impl StructuralFieldDecoder for StructuralListDecoder {
164    fn accept_page(&mut self, child: crate::decoder::LoadedPageShard) -> Result<()> {
165        self.child.accept_page(child)
166    }
167
168    fn drain(&mut self, num_rows: u64) -> Result<Box<dyn StructuralDecodeArrayTask>> {
169        let child_task = self.child.drain(num_rows)?;
170        Ok(Box::new(StructuralListDecodeTask::new(
171            child_task,
172            self.data_type.clone(),
173        )))
174    }
175
176    fn data_type(&self) -> &DataType {
177        &self.data_type
178    }
179}
180
181#[derive(Debug)]
182struct StructuralListDecodeTask {
183    child_task: Box<dyn StructuralDecodeArrayTask>,
184    data_type: DataType,
185}
186
187impl StructuralListDecodeTask {
188    fn new(child_task: Box<dyn StructuralDecodeArrayTask>, data_type: DataType) -> Self {
189        Self {
190            child_task,
191            data_type,
192        }
193    }
194}
195
196impl StructuralDecodeArrayTask for StructuralListDecodeTask {
197    fn decode(self: Box<Self>) -> Result<DecodedArray> {
198        let DecodedArray {
199            array,
200            mut repdef,
201            data_size,
202        } = self.child_task.decode()?;
203        match &self.data_type {
204            DataType::List(child_field) => {
205                let (offsets, validity) = repdef.unravel_offsets::<i32>()?;
206                let array = if !child_field.is_nullable() && array.null_count() == array.len() {
207                    make_array(array.into_data().into_builder().nulls(None).build()?)
208                } else {
209                    array
210                };
211                let list_array = ListArray::try_new(child_field.clone(), offsets, array, validity)?;
212
213                Ok(DecodedArray {
214                    array: Arc::new(list_array),
215                    repdef,
216                    data_size,
217                })
218            }
219            DataType::LargeList(child_field) => {
220                let (offsets, validity) = repdef.unravel_offsets::<i64>()?;
221                let list_array =
222                    LargeListArray::try_new(child_field.clone(), offsets, array, validity)?;
223                Ok(DecodedArray {
224                    array: Arc::new(list_array),
225                    repdef,
226                    data_size,
227                })
228            }
229            _ => panic!("List decoder did not have a list field"),
230        }
231    }
232}
233
234#[cfg(test)]
235mod tests {
236
237    use std::{collections::HashMap, sync::Arc};
238
239    use crate::constants::{
240        STRUCTURAL_ENCODING_FULLZIP, STRUCTURAL_ENCODING_META_KEY, STRUCTURAL_ENCODING_MINIBLOCK,
241    };
242    use arrow_array::{
243        Array, ArrayRef, BooleanArray, DictionaryArray, LargeStringArray, ListArray, StructArray,
244        UInt8Array, UInt64Array,
245        builder::{
246            Int32Builder, Int64Builder, LargeListBuilder, ListBuilder, StringBuilder, UInt32Builder,
247        },
248    };
249
250    use arrow_buffer::{BooleanBuffer, NullBuffer, OffsetBuffer, ScalarBuffer};
251    use arrow_schema::{DataType, Field, Fields};
252    use rstest::rstest;
253
254    use crate::{
255        testing::{TestCases, check_basic_random, check_round_trip_encoding_of_data},
256        version::LanceFileVersion,
257    };
258
259    fn make_list_type(inner_type: DataType) -> DataType {
260        DataType::List(Arc::new(Field::new("item", inner_type, true)))
261    }
262
263    fn make_large_list_type(inner_type: DataType) -> DataType {
264        DataType::LargeList(Arc::new(Field::new("item", inner_type, true)))
265    }
266
267    async fn try_encode_v22_pages(
268        array: ArrayRef,
269    ) -> lance_core::Result<Vec<crate::encoder::EncodedPage>> {
270        try_encode_v22_pages_with_metadata(array, HashMap::new()).await
271    }
272
273    async fn try_encode_v22_pages_with_metadata(
274        array: ArrayRef,
275        field_metadata: HashMap<String, String>,
276    ) -> lance_core::Result<Vec<crate::encoder::EncodedPage>> {
277        let arrow_field =
278            Field::new("", array.data_type().clone(), true).with_metadata(field_metadata);
279        let lance_field = lance_core::datatypes::Field::try_from(&arrow_field).unwrap();
280        let encoding_strategy = crate::encoder::default_encoding_strategy(LanceFileVersion::V2_2);
281        let mut column_index_seq = crate::encoder::ColumnIndexSequence::default();
282        let encoding_options = crate::encoder::EncodingOptions {
283            version: LanceFileVersion::V2_2,
284            ..Default::default()
285        };
286        let mut encoder = encoding_strategy
287            .create_field_encoder(
288                encoding_strategy.as_ref(),
289                &lance_field,
290                &mut column_index_seq,
291                &encoding_options,
292            )
293            .unwrap();
294        let mut external_buffers =
295            crate::encoder::OutOfLineBuffers::new(0, crate::encoder::MIN_PAGE_BUFFER_ALIGNMENT);
296        let num_rows = array.len() as u64;
297        let mut pages = Vec::new();
298        for task in encoder
299            .maybe_encode(
300                array,
301                &mut external_buffers,
302                crate::repdef::RepDefBuilder::default(),
303                0,
304                num_rows,
305            )
306            .unwrap()
307        {
308            pages.push(task.await?);
309        }
310        for task in encoder.flush(&mut external_buffers).unwrap() {
311            pages.push(task.await?);
312        }
313        Ok(pages)
314    }
315
316    async fn encode_v22_pages(array: ArrayRef) -> Vec<crate::encoder::EncodedPage> {
317        try_encode_v22_pages(array).await.unwrap()
318    }
319
320    fn assert_split_miniblock_layout(
321        pages: &[crate::encoder::EncodedPage],
322        expect_structural_only_page: bool,
323    ) {
324        let mut miniblock_pages = 0;
325        let mut fullzip_pages = 0;
326        let mut structural_only_pages = 0;
327
328        for page in pages {
329            let crate::decoder::PageEncoding::Structural(layout) = &page.description else {
330                continue;
331            };
332            match layout.layout.as_ref().unwrap() {
333                crate::format::pb21::page_layout::Layout::MiniBlockLayout(_) => {
334                    miniblock_pages += 1;
335                }
336                crate::format::pb21::page_layout::Layout::FullZipLayout(_) => {
337                    fullzip_pages += 1;
338                }
339                crate::format::pb21::page_layout::Layout::ConstantLayout(layout) => {
340                    if layout.inline_value.is_none()
341                        && (layout.num_rep_values > 0 || layout.num_def_values > 0)
342                    {
343                        structural_only_pages += 1;
344                    }
345                }
346                crate::format::pb21::page_layout::Layout::BlobLayout(_) => {}
347            }
348        }
349
350        assert!(
351            miniblock_pages > 0,
352            "expected leaf values to remain on mini-block pages"
353        );
354        assert_eq!(
355            fullzip_pages, 0,
356            "split list pages should not fall back to full-zip"
357        );
358        if expect_structural_only_page {
359            assert!(
360                structural_only_pages > 0,
361                "expected at least one structural-only page"
362            );
363        }
364    }
365
366    fn assert_has_fullzip_layout(pages: &[crate::encoder::EncodedPage]) {
367        let has_fullzip = pages.iter().any(|page| {
368            let crate::decoder::PageEncoding::Structural(layout) = &page.description else {
369                return false;
370            };
371            matches!(
372                layout.layout.as_ref().unwrap(),
373                crate::format::pb21::page_layout::Layout::FullZipLayout(_)
374            )
375        });
376        assert!(has_fullzip, "expected at least one full-zip page");
377    }
378
379    #[rstest]
380    #[test_log::test(tokio::test)]
381    async fn test_list(
382        #[values(STRUCTURAL_ENCODING_MINIBLOCK, STRUCTURAL_ENCODING_FULLZIP)]
383        structural_encoding: &str,
384    ) {
385        let mut field_metadata = HashMap::new();
386        field_metadata.insert(
387            STRUCTURAL_ENCODING_META_KEY.to_string(),
388            structural_encoding.into(),
389        );
390        let field =
391            Field::new("", make_list_type(DataType::Int32), true).with_metadata(field_metadata);
392        check_basic_random(field).await;
393    }
394
395    #[rstest]
396    #[test_log::test(tokio::test)]
397    async fn test_deeply_nested_lists(
398        #[values(STRUCTURAL_ENCODING_MINIBLOCK, STRUCTURAL_ENCODING_FULLZIP)]
399        structural_encoding: &str,
400    ) {
401        let mut field_metadata = HashMap::new();
402        field_metadata.insert(
403            STRUCTURAL_ENCODING_META_KEY.to_string(),
404            structural_encoding.into(),
405        );
406        let field = Field::new("item", DataType::Int32, true).with_metadata(field_metadata);
407        for _ in 0..5 {
408            let field = Field::new("", make_list_type(field.data_type().clone()), true);
409            check_basic_random(field).await;
410        }
411    }
412
413    #[test_log::test(tokio::test)]
414    async fn test_large_list() {
415        let field = Field::new("", make_large_list_type(DataType::Int32), true);
416        check_basic_random(field).await;
417    }
418
419    #[test_log::test(tokio::test)]
420    async fn test_nested_strings() {
421        let field = Field::new("", make_list_type(DataType::Utf8), true);
422        check_basic_random(field).await;
423    }
424
425    #[test_log::test(tokio::test)]
426    async fn test_nested_list() {
427        let field = Field::new("", make_list_type(make_list_type(DataType::Int32)), true);
428        check_basic_random(field).await;
429    }
430
431    /// Regression test: a `List<List<Float32>>` column written as MULTIPLE
432    /// batches (chunks) whose flattened leaf values cross a value-page boundary
433    /// fails to decode with "Max offset N exceeds length of values M" (Arrow
434    /// error raised by `ListArray::try_new` in `StructuralListDecodeTask::decode`).
435    ///
436    /// The trigger (verified against the production file and pylance 7.0.0b12 /
437    /// 7.0.0 / 9.0.0-beta.10) requires ALL of:
438    ///   1. >= 2 list layers (`List<List<..>>`),
439    ///   2. a leaf large enough to be chunked into multiple value pages,
440    ///   3. the column written as more than one batch.
441    /// A single batch of the identical data round-trips fine — which is why the
442    /// earlier single-chunk version of this test (and the small `test_nested_list`
443    /// cases) did not catch it. Found in production on the gaming TransNet
444    /// `dino_embedding_per_frame` column (rectangular 3 x 768 float per row).
445    ///
446    /// Each element of the `vec![..]` passed to `check_round_trip_encoding_of_data`
447    /// is encoded as a separate batch (its own `RepDefBuilder`), so we split the
448    /// rows into two chunks to exercise the multi-batch repdef accumulation path.
449    #[rstest]
450    #[test_log::test(tokio::test)]
451    async fn test_multipage_nested_float_list(
452        #[values(STRUCTURAL_ENCODING_MINIBLOCK, STRUCTURAL_ENCODING_FULLZIP)]
453        structural_encoding: &str,
454    ) {
455        use arrow_array::Float32Array;
456
457        // Production shape: 3 inner lists per row, 768 floats each.
458        let inner_per_row: usize = 3;
459        let inner_len: usize = 768;
460        // Two chunks (batches) -> two pages; a read batch that spans the page
461        // boundary is where the multi-page outer-offset bug triggered. A single
462        // [2731] chunk (one page) decodes fine, which is why this needs >= 2.
463        let chunk_rows: &[usize] = &[1366, 1365];
464
465        let make_chunk = |start_row: usize, num_rows: usize| -> Arc<dyn Array> {
466            let total_inner = num_rows * inner_per_row;
467            let total_values = total_inner * inner_len;
468            let values = Float32Array::from(
469                (0..total_values)
470                    .map(|i| (start_row + i) as f32)
471                    .collect::<Vec<_>>(),
472            );
473            let inner_offsets = ScalarBuffer::<i32>::from(
474                (0..=total_inner)
475                    .map(|i| (i * inner_len) as i32)
476                    .collect::<Vec<_>>(),
477            );
478            let inner_list = ListArray::new(
479                Arc::new(Field::new("item", DataType::Float32, true)),
480                OffsetBuffer::new(inner_offsets),
481                Arc::new(values),
482                None,
483            );
484            let outer_offsets = ScalarBuffer::<i32>::from(
485                (0..=num_rows)
486                    .map(|i| (i * inner_per_row) as i32)
487                    .collect::<Vec<_>>(),
488            );
489            Arc::new(ListArray::new(
490                Arc::new(Field::new(
491                    "item",
492                    DataType::List(Arc::new(Field::new("item", DataType::Float32, true))),
493                    true,
494                )),
495                OffsetBuffer::new(outer_offsets),
496                Arc::new(inner_list),
497                None,
498            ))
499        };
500
501        let mut start = 0;
502        let chunks: Vec<Arc<dyn Array>> = chunk_rows
503            .iter()
504            .map(|&n| {
505                let c = make_chunk(start, n);
506                start += n;
507                c
508            })
509            .collect();
510
511        let mut field_metadata = HashMap::new();
512        field_metadata.insert(
513            STRUCTURAL_ENCODING_META_KEY.to_string(),
514            structural_encoding.into(),
515        );
516
517        let test_cases = TestCases::default().with_min_file_version(LanceFileVersion::V2_1);
518        check_round_trip_encoding_of_data(chunks, &test_cases, field_metadata).await;
519    }
520
521    #[test_log::test(tokio::test)]
522    async fn test_list_struct_list() {
523        let struct_type = DataType::Struct(Fields::from(vec![Field::new(
524            "inner_str",
525            DataType::Utf8,
526            false,
527        )]));
528
529        let field = Field::new("", make_list_type(struct_type), true);
530        check_basic_random(field).await;
531    }
532
533    #[test_log::test(tokio::test)]
534    async fn test_list_struct_empty() {
535        let fields = Fields::from(vec![Field::new("inner", DataType::UInt64, true)]);
536        let items = UInt64Array::from(Vec::<u64>::new());
537        let structs = StructArray::new(fields, vec![Arc::new(items)], None);
538        let offsets = OffsetBuffer::new(ScalarBuffer::<i32>::from(vec![0; 2 * 1024 * 1024 + 1]));
539        let lists = ListArray::new(
540            Arc::new(Field::new("item", structs.data_type().clone(), true)),
541            offsets,
542            Arc::new(structs),
543            None,
544        );
545
546        check_round_trip_encoding_of_data(
547            vec![Arc::new(lists)],
548            &TestCases::default(),
549            HashMap::new(),
550        )
551        .await;
552    }
553
554    #[rstest]
555    #[test_log::test(tokio::test)]
556    async fn test_simple_list(
557        #[values(STRUCTURAL_ENCODING_MINIBLOCK, STRUCTURAL_ENCODING_FULLZIP)]
558        structural_encoding: &str,
559    ) {
560        let items_builder = Int32Builder::new();
561        let mut list_builder = ListBuilder::new(items_builder);
562        list_builder.append_value([Some(1), Some(2), Some(3)]);
563        list_builder.append_value([Some(4), Some(5)]);
564        list_builder.append_null();
565        list_builder.append_value([Some(6), Some(7), Some(8)]);
566        let list_array = list_builder.finish();
567
568        let mut field_metadata = HashMap::new();
569        field_metadata.insert(
570            STRUCTURAL_ENCODING_META_KEY.to_string(),
571            structural_encoding.into(),
572        );
573
574        let test_cases = TestCases::default()
575            .with_range(0..2)
576            .with_range(0..3)
577            .with_range(1..3)
578            .with_indices(vec![1, 3])
579            .with_indices(vec![2]);
580        check_round_trip_encoding_of_data(vec![Arc::new(list_array)], &test_cases, field_metadata)
581            .await;
582    }
583
584    #[rstest]
585    #[test_log::test(tokio::test)]
586    async fn test_simple_nested_list_ends_with_null(
587        #[values(STRUCTURAL_ENCODING_MINIBLOCK, STRUCTURAL_ENCODING_FULLZIP)]
588        structural_encoding: &str,
589    ) {
590        use arrow_array::Int32Array;
591
592        let values = Int32Array::from(vec![1, 2, 3, 4, 5]);
593        let inner_offsets = ScalarBuffer::<i32>::from(vec![0, 1, 2, 3, 4, 5, 5]);
594        let inner_validity = BooleanBuffer::from(vec![true, true, true, true, true, false]);
595        let outer_offsets = ScalarBuffer::<i32>::from(vec![0, 1, 2, 3, 4, 5, 6, 6]);
596        let outer_validity = BooleanBuffer::from(vec![true, true, true, true, true, true, false]);
597
598        let inner_list = ListArray::new(
599            Arc::new(Field::new("item", DataType::Int32, true)),
600            OffsetBuffer::new(inner_offsets),
601            Arc::new(values),
602            Some(NullBuffer::new(inner_validity)),
603        );
604        let outer_list = ListArray::new(
605            Arc::new(Field::new(
606                "item",
607                DataType::List(Arc::new(Field::new("item", DataType::Int32, true))),
608                true,
609            )),
610            OffsetBuffer::new(outer_offsets),
611            Arc::new(inner_list),
612            Some(NullBuffer::new(outer_validity)),
613        );
614
615        let mut field_metadata = HashMap::new();
616        field_metadata.insert(
617            STRUCTURAL_ENCODING_META_KEY.to_string(),
618            structural_encoding.into(),
619        );
620
621        let test_cases = TestCases::default()
622            .with_range(0..2)
623            .with_range(0..3)
624            .with_range(5..7)
625            .with_indices(vec![1, 6])
626            .with_indices(vec![6])
627            .with_min_file_version(LanceFileVersion::V2_1);
628        check_round_trip_encoding_of_data(vec![Arc::new(outer_list)], &test_cases, field_metadata)
629            .await;
630    }
631
632    #[rstest]
633    #[test_log::test(tokio::test)]
634    async fn test_simple_string_list(
635        #[values(STRUCTURAL_ENCODING_MINIBLOCK, STRUCTURAL_ENCODING_FULLZIP)]
636        structural_encoding: &str,
637    ) {
638        let items_builder = StringBuilder::new();
639        let mut list_builder = ListBuilder::new(items_builder);
640        list_builder.append_value([Some("a"), Some("bc"), Some("def")]);
641        list_builder.append_value([Some("gh"), None]);
642        list_builder.append_null();
643        list_builder.append_value([Some("ijk"), Some("lmnop"), Some("qrs")]);
644        let list_array = list_builder.finish();
645
646        let mut field_metadata = HashMap::new();
647        field_metadata.insert(
648            STRUCTURAL_ENCODING_META_KEY.to_string(),
649            structural_encoding.into(),
650        );
651
652        let test_cases = TestCases::default()
653            .with_range(0..2)
654            .with_range(0..3)
655            .with_range(1..3)
656            .with_indices(vec![1, 3])
657            .with_indices(vec![2])
658            .with_min_file_version(LanceFileVersion::V2_1);
659        check_round_trip_encoding_of_data(vec![Arc::new(list_array)], &test_cases, field_metadata)
660            .await;
661    }
662
663    #[rstest]
664    #[test_log::test(tokio::test)]
665    async fn test_simple_string_list_no_null(
666        #[values(STRUCTURAL_ENCODING_MINIBLOCK, STRUCTURAL_ENCODING_FULLZIP)]
667        structural_encoding: &str,
668    ) {
669        let items_builder = StringBuilder::new();
670        let mut list_builder = ListBuilder::new(items_builder);
671        list_builder.append_value([Some("a"), Some("bc"), Some("def")]);
672        list_builder.append_value([Some("gh"), Some("zxy")]);
673        list_builder.append_value([Some("gh"), Some("z")]);
674        list_builder.append_value([Some("ijk"), Some("lmnop"), Some("qrs")]);
675        let list_array = list_builder.finish();
676
677        let mut field_metadata = HashMap::new();
678        field_metadata.insert(
679            STRUCTURAL_ENCODING_META_KEY.to_string(),
680            structural_encoding.into(),
681        );
682
683        let test_cases = TestCases::default()
684            .with_range(0..2)
685            .with_range(0..3)
686            .with_range(1..3)
687            .with_indices(vec![1, 3])
688            .with_indices(vec![2])
689            .with_min_file_version(LanceFileVersion::V2_1);
690        check_round_trip_encoding_of_data(vec![Arc::new(list_array)], &test_cases, field_metadata)
691            .await;
692    }
693
694    #[rstest]
695    #[test_log::test(tokio::test)]
696    async fn test_simple_sliced_list(
697        #[values(STRUCTURAL_ENCODING_MINIBLOCK, STRUCTURAL_ENCODING_FULLZIP)]
698        structural_encoding: &str,
699    ) {
700        let items_builder = Int32Builder::new();
701        let mut list_builder = ListBuilder::new(items_builder);
702        list_builder.append_value([Some(1), Some(2), Some(3)]);
703        list_builder.append_value([Some(4), Some(5)]);
704        list_builder.append_null();
705        list_builder.append_value([Some(6), Some(7), Some(8)]);
706        let list_array = list_builder.finish();
707
708        let list_array = list_array.slice(1, 2);
709
710        let mut field_metadata = HashMap::new();
711        field_metadata.insert(
712            STRUCTURAL_ENCODING_META_KEY.to_string(),
713            structural_encoding.into(),
714        );
715
716        let test_cases = TestCases::default()
717            .with_range(0..2)
718            .with_range(1..2)
719            .with_indices(vec![0])
720            .with_indices(vec![1])
721            .with_min_file_version(LanceFileVersion::V2_1);
722        check_round_trip_encoding_of_data(vec![Arc::new(list_array)], &test_cases, field_metadata)
723            .await;
724    }
725
726    #[test_log::test(tokio::test)]
727    async fn test_simple_list_dict() {
728        let values = LargeStringArray::from_iter_values(["a", "bb", "ccc"]);
729        let indices = UInt8Array::from(vec![0, 1, 2, 0, 1, 2, 0, 1, 2]);
730        let dict_array = DictionaryArray::new(indices, Arc::new(values));
731        let offsets = OffsetBuffer::new(ScalarBuffer::<i32>::from(vec![0, 3, 5, 6, 9]));
732        let list_array = ListArray::new(
733            Arc::new(Field::new("item", dict_array.data_type().clone(), true)),
734            offsets,
735            Arc::new(dict_array),
736            None,
737        );
738
739        let test_cases = TestCases::default()
740            .with_range(0..2)
741            .with_range(1..3)
742            .with_range(2..4)
743            .with_indices(vec![1])
744            .with_indices(vec![2]);
745        check_round_trip_encoding_of_data(
746            vec![Arc::new(list_array)],
747            &test_cases,
748            HashMap::default(),
749        )
750        .await;
751    }
752
753    #[test_log::test(tokio::test)]
754    async fn test_simple_list_all_null() {
755        let items = UInt64Array::from(vec![0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10]);
756        let offsets = ScalarBuffer::<i32>::from(vec![0, 5, 8, 10]);
757        let offsets = OffsetBuffer::new(offsets);
758        let list_validity = NullBuffer::new(BooleanBuffer::from(vec![false, false, false]));
759
760        // The list array is nullable but the items are not.  Then, all lists are null.
761        let list_arr = ListArray::new(
762            Arc::new(Field::new("item", DataType::UInt64, false)),
763            offsets,
764            Arc::new(items),
765            Some(list_validity),
766        );
767
768        let test_cases = TestCases::default()
769            .with_range(0..3)
770            .with_range(1..2)
771            .with_indices(vec![1])
772            .with_indices(vec![2])
773            .with_min_file_version(LanceFileVersion::V2_1);
774        check_round_trip_encoding_of_data(
775            vec![Arc::new(list_arr)],
776            &test_cases,
777            HashMap::default(),
778        )
779        .await;
780    }
781
782    #[rstest]
783    #[test_log::test(tokio::test)]
784    async fn test_list_with_garbage_nulls(
785        #[values(STRUCTURAL_ENCODING_MINIBLOCK, STRUCTURAL_ENCODING_FULLZIP)]
786        structural_encoding: &str,
787    ) {
788        // In Arrow, list nulls are allowed to be non-empty, with masked garbage values
789        // Here we make a list with a null row in the middle with 3 garbage values
790        let items = UInt64Array::from(vec![0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10]);
791        let offsets = ScalarBuffer::<i32>::from(vec![0, 5, 8, 10]);
792        let offsets = OffsetBuffer::new(offsets);
793        let list_validity = NullBuffer::new(BooleanBuffer::from(vec![true, false, true]));
794        let list_arr = ListArray::new(
795            Arc::new(Field::new("item", DataType::UInt64, true)),
796            offsets,
797            Arc::new(items),
798            Some(list_validity),
799        );
800
801        let mut field_metadata = HashMap::new();
802        field_metadata.insert(
803            STRUCTURAL_ENCODING_META_KEY.to_string(),
804            structural_encoding.into(),
805        );
806
807        let test_cases = TestCases::default()
808            .with_range(0..3)
809            .with_range(1..2)
810            .with_indices(vec![1])
811            .with_indices(vec![2])
812            .with_min_file_version(LanceFileVersion::V2_1);
813        check_round_trip_encoding_of_data(vec![Arc::new(list_arr)], &test_cases, field_metadata)
814            .await;
815    }
816
817    #[rstest]
818    #[test_log::test(tokio::test)]
819    async fn test_simple_two_page_list(
820        #[values(STRUCTURAL_ENCODING_MINIBLOCK, STRUCTURAL_ENCODING_FULLZIP)]
821        structural_encoding: &str,
822    ) {
823        // This is a simple pre-defined list that spans two pages.  This test is useful for
824        // debugging the repetition index
825
826        let items_builder = Int64Builder::new();
827        let mut list_builder = ListBuilder::new(items_builder);
828        for i in 0..512 {
829            list_builder.append_value([Some(i), Some(i * 2)]);
830        }
831        let list_array_1 = list_builder.finish();
832
833        let items_builder = Int64Builder::new();
834        let mut list_builder = ListBuilder::new(items_builder);
835        for i in 0..512 {
836            let i = i + 512;
837            list_builder.append_value([Some(i), Some(i * 2)]);
838        }
839        let list_array_2 = list_builder.finish();
840
841        let mut metadata = HashMap::new();
842        metadata.insert(
843            STRUCTURAL_ENCODING_META_KEY.to_string(),
844            structural_encoding.into(),
845        );
846
847        let test_cases = TestCases::default()
848            .with_min_file_version(LanceFileVersion::V2_1)
849            .with_page_sizes(vec![100])
850            .with_range(800..900);
851        check_round_trip_encoding_of_data(
852            vec![Arc::new(list_array_1), Arc::new(list_array_2)],
853            &test_cases,
854            metadata,
855        )
856        .await;
857    }
858
859    #[test_log::test(tokio::test)]
860    async fn test_simple_large_list() {
861        let items_builder = Int32Builder::new();
862        let mut list_builder = LargeListBuilder::new(items_builder);
863        list_builder.append_value([Some(1), Some(2), Some(3)]);
864        list_builder.append_value([Some(4), Some(5)]);
865        list_builder.append_null();
866        list_builder.append_value([Some(6), Some(7), Some(8)]);
867        let list_array = list_builder.finish();
868
869        let test_cases = TestCases::default()
870            .with_range(0..2)
871            .with_range(0..3)
872            .with_range(1..3)
873            .with_indices(vec![1, 3]);
874        check_round_trip_encoding_of_data(vec![Arc::new(list_array)], &test_cases, HashMap::new())
875            .await;
876    }
877
878    #[rstest]
879    #[test_log::test(tokio::test)]
880    async fn test_empty_lists(
881        #[values(STRUCTURAL_ENCODING_MINIBLOCK, STRUCTURAL_ENCODING_FULLZIP)]
882        structural_encoding: &str,
883    ) {
884        let mut field_metadata = HashMap::new();
885        field_metadata.insert(
886            STRUCTURAL_ENCODING_META_KEY.to_string(),
887            structural_encoding.into(),
888        );
889
890        // Scenario 1: Some lists are empty
891
892        let values = [vec![Some(1), Some(2), Some(3)], vec![], vec![None]];
893        // Test empty list at beginning, middle, and end
894        for order in [[0, 1, 2], [1, 0, 2], [2, 0, 1]] {
895            let items_builder = Int32Builder::new();
896            let mut list_builder = ListBuilder::new(items_builder);
897            for idx in order {
898                list_builder.append_value(values[idx].clone());
899            }
900            let list_array = Arc::new(list_builder.finish());
901            let test_cases = TestCases::default()
902                .with_indices(vec![1])
903                .with_indices(vec![0])
904                .with_indices(vec![2])
905                .with_indices(vec![0, 1]);
906            check_round_trip_encoding_of_data(
907                vec![list_array.clone()],
908                &test_cases,
909                field_metadata.clone(),
910            )
911            .await;
912            let test_cases = test_cases.with_batch_size(1);
913            check_round_trip_encoding_of_data(
914                vec![list_array],
915                &test_cases,
916                field_metadata.clone(),
917            )
918            .await;
919        }
920
921        // Scenario 2: All lists are empty
922
923        // When encoding a list of empty lists there are no items to encode
924        // which is strange and we want to ensure we handle it
925        let items_builder = Int32Builder::new();
926        let mut list_builder = ListBuilder::new(items_builder);
927        list_builder.append(true);
928        list_builder.append_null();
929        list_builder.append(true);
930        let list_array = Arc::new(list_builder.finish());
931
932        let test_cases = TestCases::default().with_range(0..2).with_indices(vec![1]);
933        check_round_trip_encoding_of_data(
934            vec![list_array.clone()],
935            &test_cases,
936            field_metadata.clone(),
937        )
938        .await;
939        let test_cases = test_cases.with_batch_size(1);
940        check_round_trip_encoding_of_data(vec![list_array], &test_cases, field_metadata.clone())
941            .await;
942
943        // Scenario 2B: All lists are empty (but now with strings)
944
945        // When encoding a list of empty lists there are no items to encode
946        // which is strange and we want to ensure we handle it
947        let items_builder = StringBuilder::new();
948        let mut list_builder = ListBuilder::new(items_builder);
949        list_builder.append(true);
950        list_builder.append_null();
951        list_builder.append(true);
952        let list_array = Arc::new(list_builder.finish());
953
954        let test_cases = TestCases::default().with_range(0..2).with_indices(vec![1]);
955        check_round_trip_encoding_of_data(
956            vec![list_array.clone()],
957            &test_cases,
958            field_metadata.clone(),
959        )
960        .await;
961        let test_cases = test_cases.with_batch_size(1);
962        check_round_trip_encoding_of_data(vec![list_array], &test_cases, field_metadata.clone())
963            .await;
964
965        // Scenario 3: All lists are null
966
967        let items_builder = Int32Builder::new();
968        let mut list_builder = ListBuilder::new(items_builder);
969        list_builder.append_null();
970        list_builder.append_null();
971        list_builder.append_null();
972        let list_array = Arc::new(list_builder.finish());
973
974        let test_cases = TestCases::default().with_range(0..2).with_indices(vec![1]);
975        check_round_trip_encoding_of_data(
976            vec![list_array.clone()],
977            &test_cases,
978            field_metadata.clone(),
979        )
980        .await;
981        let test_cases = test_cases.with_batch_size(1);
982        check_round_trip_encoding_of_data(vec![list_array], &test_cases, field_metadata.clone())
983            .await;
984
985        // Scenario 4: All lists are null and inside a struct (only valid for 2.1 since 2.0 doesn't
986        // support null structs)
987        let items_builder = Int32Builder::new();
988        let mut list_builder = ListBuilder::new(items_builder);
989        list_builder.append_null();
990        list_builder.append_null();
991        list_builder.append_null();
992        let list_array = Arc::new(list_builder.finish());
993
994        let struct_validity = NullBuffer::new(BooleanBuffer::from(vec![true, false, true]));
995        let struct_array = Arc::new(StructArray::new(
996            Fields::from(vec![Field::new(
997                "lists",
998                list_array.data_type().clone(),
999                true,
1000            )]),
1001            vec![list_array],
1002            Some(struct_validity),
1003        ));
1004
1005        let test_cases = TestCases::default()
1006            .with_range(0..2)
1007            .with_indices(vec![1])
1008            .with_min_file_version(LanceFileVersion::V2_1);
1009        check_round_trip_encoding_of_data(
1010            vec![struct_array.clone()],
1011            &test_cases,
1012            field_metadata.clone(),
1013        )
1014        .await;
1015        let test_cases = test_cases.with_batch_size(1);
1016        check_round_trip_encoding_of_data(vec![struct_array], &test_cases, field_metadata.clone())
1017            .await;
1018    }
1019
1020    #[test_log::test(tokio::test)]
1021    async fn test_empty_list_list() {
1022        let items_builder = Int32Builder::new();
1023        let list_builder = ListBuilder::new(items_builder);
1024        let mut outer_list_builder = ListBuilder::new(list_builder);
1025        outer_list_builder.append_null();
1026        outer_list_builder.append_null();
1027        outer_list_builder.append_null();
1028        let list_array = Arc::new(outer_list_builder.finish());
1029
1030        let test_cases = TestCases::default().with_min_file_version(LanceFileVersion::V2_1);
1031        check_round_trip_encoding_of_data(vec![list_array], &test_cases, HashMap::new()).await;
1032    }
1033
1034    #[test_log::test(tokio::test)]
1035    #[ignore] // This test is quite slow in debug mode
1036    async fn test_jumbo_list() {
1037        // This is an overflow test.  We have a list of lists where each list
1038        // has 1Mi items.  We encode 5000 of these lists and so we have over 4Gi in the
1039        // offsets range
1040        let items = BooleanArray::new_null(1024 * 1024);
1041        let offsets = OffsetBuffer::new(ScalarBuffer::from(vec![0, 1024 * 1024]));
1042        let list_arr = Arc::new(ListArray::new(
1043            Arc::new(Field::new("item", DataType::Boolean, true)),
1044            offsets,
1045            Arc::new(items),
1046            None,
1047        )) as ArrayRef;
1048        let arrs = vec![list_arr; 5000];
1049
1050        // We can't validate because our validation relies on concatenating all input arrays
1051        let test_cases = TestCases::default().without_validation();
1052        check_round_trip_encoding_of_data(arrs, &test_cases, HashMap::new()).await;
1053    }
1054
1055    // Regression test for issue with ListArray encoding when crossing 1024 value boundary
1056    // This test reproduces the bug where rows_avail assertion fails in schedule_instructions
1057    // when encoding a ListArray with specific size patterns that cross the 1024 value boundary
1058    #[tokio::test]
1059    async fn test_fuzz_issue_4466() {
1060        // This specific pattern of list sizes triggers the bug when total values cross 1024
1061        // 94 lists total 1009 values (passes), 95 lists total 1025 values (fails)
1062        let list_sizes = vec![
1063            13, 18, 12, 7, 14, 12, 6, 13, 18, 8, // 0-9: 119 values
1064            6, 11, 17, 12, 8, 19, 5, 6, 10, 13, // 10-19: 107 values
1065            8, 6, 10, 4, 8, 16, 14, 12, 18, 9, // 20-29: 105 values
1066            17, 8, 14, 18, 15, 3, 2, 4, 5, 1, // 30-39: 82 values
1067            3, 13, 1, 2, 10, 4, 10, 18, 7, 14, // 40-49: 75 values
1068            18, 13, 9, 17, 3, 13, 10, 14, 8, 19, // 50-59: 125 values
1069            17, 10, 5, 11, 6, 15, 10, 18, 18, 20, // 60-69: 130 values
1070            16, 11, 12, 15, 7, 9, 3, 10, 20, 5, // 70-79: 102 values
1071            2, 3, 17, 4, 8, 12, 15, 6, 3, 20, // 80-89: 90 values
1072            15, 20, 1, 19, 16, // 90-94: 71 values
1073        ];
1074
1075        // Build the ListArray
1076        let mut list_builder = ListBuilder::new(Int32Builder::new());
1077        let mut total_values = 0;
1078
1079        for size in &list_sizes {
1080            for i in 0..*size {
1081                list_builder.values().append_value(i);
1082            }
1083            list_builder.append(true);
1084            total_values += size;
1085        }
1086
1087        let list_array = Arc::new(list_builder.finish());
1088
1089        // Verify we have the expected number of values
1090        assert_eq!(list_array.len(), 95);
1091        assert_eq!(total_values, 1025);
1092
1093        // This should trigger the assertion failure at primitive.rs:1362
1094        // debug_assert!(rows_avail > 0)
1095        let test_cases = TestCases::default().with_min_file_version(LanceFileVersion::V2_1);
1096
1097        // The bug manifests when encoding this specific pattern
1098        // Expected: successful round-trip encoding
1099        // Actual: panic at primitive.rs:1362 - assertion failed: rows_avail > 0
1100        check_round_trip_encoding_of_data(vec![list_array], &test_cases, HashMap::new()).await;
1101    }
1102
1103    #[rstest]
1104    #[test_log::test(tokio::test)]
1105    async fn test_sparse_large_string_list(
1106        #[values(STRUCTURAL_ENCODING_MINIBLOCK, STRUCTURAL_ENCODING_FULLZIP)]
1107        structural_encoding: &str,
1108    ) {
1109        // 2.5 million rows, mostly empty lists. ~100 lists have 10 short strings each.
1110        let num_rows = 2_500_000u32;
1111        let num_non_empty = 100u32;
1112        let strings_per_list = 10;
1113
1114        let items_builder = StringBuilder::new();
1115        let mut list_builder = ListBuilder::new(items_builder);
1116
1117        // Spread non-empty lists evenly across the range
1118        let step = num_rows / num_non_empty;
1119        let mut next_non_empty = step / 2;
1120
1121        for i in 0..num_rows {
1122            if i == next_non_empty {
1123                let vals: Vec<Option<&str>> = (0..strings_per_list)
1124                    .map(|j| match j % 4 {
1125                        0 => Some("a"),
1126                        1 => Some("bb"),
1127                        2 => Some("ccc"),
1128                        _ => Some("d"),
1129                    })
1130                    .collect();
1131                list_builder.append_value(vals);
1132                next_non_empty = next_non_empty.saturating_add(step);
1133            } else {
1134                list_builder.append_value([] as [Option<&str>; 0]);
1135            }
1136        }
1137        let list_array = list_builder.finish();
1138
1139        let mut field_metadata = HashMap::new();
1140        field_metadata.insert(
1141            STRUCTURAL_ENCODING_META_KEY.to_string(),
1142            structural_encoding.into(),
1143        );
1144
1145        let test_cases = TestCases::default()
1146            .with_range(0..1000)
1147            .with_range(0..num_rows as u64)
1148            .with_indices(vec![0, (step / 2) as u64, num_rows as u64 - 1])
1149            .with_max_file_version(LanceFileVersion::V2_2);
1150        check_round_trip_encoding_of_data(vec![Arc::new(list_array)], &test_cases, field_metadata)
1151            .await;
1152    }
1153
1154    #[test_log::test(tokio::test)]
1155    async fn test_sparse_boolean_list_uses_miniblock() {
1156        // Redacted reproduction from a production schema shape containing ARRAY(BOOLEAN).
1157        // The field names are not relevant; the failure requires sparse list structure
1158        // with a 1-bit Boolean leaf value.
1159        let num_rows = 200_000usize;
1160        let num_non_empty = 10usize;
1161        let booleans_per_list = 8usize;
1162        let step = num_rows / num_non_empty;
1163
1164        let mut offsets = Vec::with_capacity(num_rows + 1);
1165        let mut values = Vec::with_capacity(num_non_empty * booleans_per_list);
1166        offsets.push(0i32);
1167
1168        let mut next_non_empty = step / 2;
1169        for row in 0..num_rows {
1170            if row == next_non_empty {
1171                values.extend((0..booleans_per_list).map(|idx| idx % 2 == 0));
1172                next_non_empty += step;
1173            }
1174            offsets.push(values.len() as i32);
1175        }
1176
1177        let items = BooleanArray::from(values);
1178        let list_array = ListArray::new(
1179            Arc::new(Field::new("item", DataType::Boolean, true)),
1180            OffsetBuffer::new(ScalarBuffer::from(offsets)),
1181            Arc::new(items),
1182            None,
1183        );
1184
1185        let test_cases = TestCases::default()
1186            .with_range(0..1000)
1187            .with_range(0..num_rows as u64)
1188            .with_indices(vec![0, (step / 2) as u64, num_rows as u64 - 1])
1189            .with_max_file_version(LanceFileVersion::V2_2);
1190        let list_array = Arc::new(list_array) as ArrayRef;
1191        let pages = encode_v22_pages(list_array.clone()).await;
1192        assert_split_miniblock_layout(&pages, false);
1193        check_round_trip_encoding_of_data(vec![list_array], &test_cases, HashMap::new()).await;
1194    }
1195
1196    #[test_log::test(tokio::test)]
1197    async fn test_sparse_boolean_list_with_long_empty_prefix() {
1198        let empty_prefix_rows = 70_000usize;
1199        let trailing_empty_rows = 9usize;
1200        let booleans_per_list = 8usize;
1201        let num_rows = empty_prefix_rows + 1 + trailing_empty_rows;
1202
1203        let mut offsets = Vec::with_capacity(num_rows + 1);
1204        offsets.extend(std::iter::repeat_n(0i32, empty_prefix_rows + 1));
1205        let values = (0..booleans_per_list)
1206            .map(|idx| idx % 2 == 0)
1207            .collect::<Vec<_>>();
1208        offsets.push(values.len() as i32);
1209        offsets.extend(std::iter::repeat_n(
1210            values.len() as i32,
1211            trailing_empty_rows,
1212        ));
1213
1214        let items = BooleanArray::from(values);
1215        let list_array = ListArray::new(
1216            Arc::new(Field::new("item", DataType::Boolean, true)),
1217            OffsetBuffer::new(ScalarBuffer::from(offsets)),
1218            Arc::new(items),
1219            None,
1220        );
1221
1222        let test_cases = TestCases::default()
1223            .with_range(0..num_rows as u64)
1224            .with_indices(vec![0, empty_prefix_rows as u64, num_rows as u64 - 1])
1225            .with_max_file_version(LanceFileVersion::V2_2);
1226        let list_array = Arc::new(list_array) as ArrayRef;
1227        let pages = encode_v22_pages(list_array.clone()).await;
1228        assert_split_miniblock_layout(&pages, true);
1229        check_round_trip_encoding_of_data(vec![list_array], &test_cases, HashMap::new()).await;
1230    }
1231
1232    #[test_log::test(tokio::test)]
1233    async fn test_sparse_boolean_list_with_long_null_prefix() {
1234        let null_prefix_rows = 70_000usize;
1235        let trailing_empty_rows = 9usize;
1236        let booleans_per_list = 8usize;
1237        let num_rows = null_prefix_rows + 1 + trailing_empty_rows;
1238
1239        let mut offsets = Vec::with_capacity(num_rows + 1);
1240        offsets.extend(std::iter::repeat_n(0i32, null_prefix_rows + 1));
1241        let values = (0..booleans_per_list)
1242            .map(|idx| idx % 2 == 0)
1243            .collect::<Vec<_>>();
1244        offsets.push(values.len() as i32);
1245        offsets.extend(std::iter::repeat_n(
1246            values.len() as i32,
1247            trailing_empty_rows,
1248        ));
1249        let validity = BooleanBuffer::from_iter((0..num_rows).map(|row| row >= null_prefix_rows));
1250
1251        let items = BooleanArray::from(values);
1252        let list_array = ListArray::new(
1253            Arc::new(Field::new("item", DataType::Boolean, true)),
1254            OffsetBuffer::new(ScalarBuffer::from(offsets)),
1255            Arc::new(items),
1256            Some(NullBuffer::new(validity)),
1257        );
1258
1259        let test_cases = TestCases::default()
1260            .with_range(0..num_rows as u64)
1261            .with_indices(vec![0, null_prefix_rows as u64, num_rows as u64 - 1])
1262            .with_max_file_version(LanceFileVersion::V2_2);
1263        let list_array = Arc::new(list_array) as ArrayRef;
1264        let pages = encode_v22_pages(list_array.clone()).await;
1265        assert_split_miniblock_layout(&pages, true);
1266        check_round_trip_encoding_of_data(vec![list_array], &test_cases, HashMap::new()).await;
1267    }
1268
1269    #[test_log::test(tokio::test)]
1270    async fn test_sparse_boolean_list_with_amortized_long_empty_prefix() {
1271        let empty_prefix_rows = 62_000usize;
1272        let booleans_per_list = 8_192usize;
1273        let num_rows = empty_prefix_rows + 1;
1274
1275        let mut offsets = Vec::with_capacity(num_rows + 1);
1276        offsets.extend(std::iter::repeat_n(0i32, empty_prefix_rows + 1));
1277        let values = (0..booleans_per_list)
1278            .map(|idx| idx % 2 == 0)
1279            .collect::<Vec<_>>();
1280        offsets.push(values.len() as i32);
1281
1282        let items = BooleanArray::from(values);
1283        let list_array = ListArray::new(
1284            Arc::new(Field::new("item", DataType::Boolean, true)),
1285            OffsetBuffer::new(ScalarBuffer::from(offsets)),
1286            Arc::new(items),
1287            None,
1288        );
1289
1290        let test_cases = TestCases::default()
1291            .with_range(0..num_rows as u64)
1292            .with_indices(vec![0, empty_prefix_rows as u64])
1293            .with_max_file_version(LanceFileVersion::V2_2);
1294        let list_array = Arc::new(list_array) as ArrayRef;
1295        let pages = encode_v22_pages(list_array.clone()).await;
1296        assert_split_miniblock_layout(&pages, true);
1297        check_round_trip_encoding_of_data(vec![list_array], &test_cases, HashMap::new()).await;
1298    }
1299
1300    #[test_log::test(tokio::test)]
1301    async fn test_nested_sparse_boolean_list_fails_without_panic() {
1302        let empty_inner_lists = 70_000usize;
1303        let booleans_per_list = 8usize;
1304
1305        let mut inner_offsets = vec![0i32; empty_inner_lists + 1];
1306        let values = (0..booleans_per_list)
1307            .map(|idx| idx % 2 == 0)
1308            .collect::<Vec<_>>();
1309        inner_offsets.push(values.len() as i32);
1310
1311        let inner_items = BooleanArray::from(values);
1312        let inner_list = ListArray::new(
1313            Arc::new(Field::new("item", DataType::Boolean, true)),
1314            OffsetBuffer::new(ScalarBuffer::from(inner_offsets)),
1315            Arc::new(inner_items),
1316            None,
1317        );
1318        let outer_list = ListArray::new(
1319            Arc::new(Field::new("item", inner_list.data_type().clone(), true)),
1320            OffsetBuffer::new(ScalarBuffer::from(vec![0i32, empty_inner_lists as i32 + 1])),
1321            Arc::new(inner_list),
1322            None,
1323        );
1324
1325        let err = try_encode_v22_pages(Arc::new(outer_list))
1326            .await
1327            .unwrap_err();
1328        assert!(
1329            err.to_string().contains("Mini-block cannot encode"),
1330            "unexpected error: {err}"
1331        );
1332    }
1333
1334    #[test_log::test(tokio::test)]
1335    async fn test_nested_sparse_string_single_row_falls_back_to_fullzip() {
1336        let empty_inner_lists = 70_000usize;
1337
1338        let mut inner_offsets = vec![0i32; empty_inner_lists + 1];
1339        inner_offsets.push(1);
1340        inner_offsets.push(2);
1341
1342        let mut strings = StringBuilder::new();
1343        strings.append_value("value");
1344        strings.append_value("other");
1345        let inner_items = strings.finish();
1346        let inner_list = ListArray::new(
1347            Arc::new(Field::new("item", DataType::Utf8, true)),
1348            OffsetBuffer::new(ScalarBuffer::from(inner_offsets)),
1349            Arc::new(inner_items),
1350            None,
1351        );
1352        let outer_list = ListArray::new(
1353            Arc::new(Field::new("item", inner_list.data_type().clone(), true)),
1354            OffsetBuffer::new(ScalarBuffer::from(vec![0i32, empty_inner_lists as i32 + 2])),
1355            Arc::new(inner_list),
1356            None,
1357        );
1358
1359        let outer_list = Arc::new(outer_list) as ArrayRef;
1360        let pages = encode_v22_pages(outer_list.clone()).await;
1361        assert_has_fullzip_layout(&pages);
1362
1363        let test_cases = TestCases::default()
1364            .with_range(0..1)
1365            .with_indices(vec![0])
1366            .with_max_file_version(LanceFileVersion::V2_2);
1367        check_round_trip_encoding_of_data(vec![outer_list], &test_cases, HashMap::new()).await;
1368    }
1369
1370    /// Builds the HNSW-flush repro shape: a dense prefix where every row has
1371    /// `NEIGHBORS_PER_ROW` distinct values, followed by a long tail of empty
1372    /// lists. Mirrors `HNSW::schema()` `__neighbors` / `__dists` columns:
1373    /// dense level-0 lists, then ~6x as many mostly-empty higher-level rows.
1374    fn make_hnsw_shaped_list_u32() -> ListArray {
1375        const DENSE_ROWS: u32 = 40_000;
1376        const NEIGHBORS_PER_ROW: u32 = 32;
1377        const EMPTY_TAIL_ROWS: u32 = 240_000;
1378
1379        let mut list_builder = ListBuilder::new(UInt32Builder::new());
1380        let mut next_val: u32 = 0;
1381        for _ in 0..DENSE_ROWS {
1382            for _ in 0..NEIGHBORS_PER_ROW {
1383                list_builder.values().append_value(next_val);
1384                next_val = next_val.wrapping_add(1);
1385            }
1386            list_builder.append(true);
1387        }
1388        for _ in 0..EMPTY_TAIL_ROWS {
1389            list_builder.append(true);
1390        }
1391        list_builder.finish()
1392    }
1393
1394    /// Reproduces the HNSW-flush shape at v2.2 on the auto path (no
1395    /// `STRUCTURAL_ENCODING` metadata): a dense level-0 prefix followed by a
1396    /// long tail of empty lists. The global levels/values ratio looks dense,
1397    /// so this used to encode as a single mini-block page whose final chunk
1398    /// absorbed every trailing empty list and overflowed the per-chunk `u16`
1399    /// `num_levels`, corrupting the read. The structural page planner now
1400    /// splits on top-level row boundaries: the dense prefix stays on
1401    /// mini-block pages and the empty tail becomes structural-only pages, so
1402    /// the round-trip is lossless without falling back to full-zip.
1403    #[test_log::test(tokio::test)]
1404    async fn test_list_hnsw_shape_splits_to_miniblock_v2_2() {
1405        let list_array = make_hnsw_shaped_list_u32();
1406        let dense_rows: u64 = 40_000;
1407        let total_rows = list_array.len() as u64;
1408
1409        let test_cases = TestCases::default()
1410            .with_range(0..1000)
1411            .with_range(dense_rows.saturating_sub(8)..(dense_rows + 8))
1412            .with_range(0..total_rows)
1413            .with_indices(vec![0, dense_rows - 1, dense_rows, total_rows - 1])
1414            .with_min_file_version(LanceFileVersion::V2_2)
1415            .with_max_file_version(LanceFileVersion::V2_2);
1416        let list_array = Arc::new(list_array) as ArrayRef;
1417        let pages = encode_v22_pages(list_array.clone()).await;
1418        assert_split_miniblock_layout(&pages, true);
1419        check_round_trip_encoding_of_data(vec![list_array], &test_cases, HashMap::new()).await;
1420    }
1421
1422    /// Companion to the auto-path test: even when the user explicitly requests
1423    /// `STRUCTURAL_ENCODING_MINIBLOCK`, the structural page planner splits the
1424    /// HNSW shape so every emitted page fits the mini-block per-chunk budget.
1425    /// The request is honored (the dense prefix stays on mini-block pages
1426    /// rather than being forced to full-zip) and the round-trip is lossless.
1427    #[test_log::test(tokio::test)]
1428    async fn test_forced_miniblock_hnsw_shape_splits_to_miniblock_v2_2() {
1429        let list_array = make_hnsw_shaped_list_u32();
1430        let total_rows = list_array.len() as u64;
1431
1432        let mut field_metadata = HashMap::new();
1433        field_metadata.insert(
1434            STRUCTURAL_ENCODING_META_KEY.to_string(),
1435            STRUCTURAL_ENCODING_MINIBLOCK.into(),
1436        );
1437
1438        let test_cases = TestCases::default()
1439            .with_range(0..total_rows)
1440            .with_min_file_version(LanceFileVersion::V2_2)
1441            .with_max_file_version(LanceFileVersion::V2_2);
1442        let list_array = Arc::new(list_array) as ArrayRef;
1443        let pages = try_encode_v22_pages_with_metadata(list_array.clone(), field_metadata.clone())
1444            .await
1445            .unwrap();
1446        assert_split_miniblock_layout(&pages, true);
1447        check_round_trip_encoding_of_data(vec![list_array], &test_cases, field_metadata).await;
1448    }
1449}
lance_encoding/encodings/logical/list.rs

lance_encoding/encodings/logical/
list.rs
