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

lance_encoding/encodings/logical/
list.rs
