Skip to main content

lance_encoding/encodings/physical/
binary.rs

1// SPDX-License-Identifier: Apache-2.0
2// SPDX-FileCopyrightText: Copyright The Lance Authors
3
4//! Basic encodings for variable width data
5//!
6//! These are not compression but represent the "leaf" encodings for variable length data
7//! where we simply match the data with the rules of the structural encoding.
8//!
9//! These encodings are transparent since we aren't actually doing any compression.  No information
10//! is needed in the encoding description.
11
12use arrow_array::OffsetSizeTrait;
13use byteorder::{ByteOrder, LittleEndian};
14use prost::Message;
15
16use crate::compression::{
17    BlockCompressor, BlockDecompressor, MiniBlockDecompressor, VariablePerValueDecompressor,
18    create_fixed_width_block_decompressor, infer_fixed_width_block_bits, require_block_payload,
19};
20
21use crate::buffer::LanceBuffer;
22use crate::compression_config::CompressionFieldParams;
23use crate::data::{BlockInfo, DataBlock, FixedWidthDataBlock, VariableWidthBlock};
24use crate::encodings::logical::primitive::fullzip::{PerValueCompressor, PerValueDataBlock};
25use crate::encodings::logical::primitive::miniblock::{
26    MAX_MINIBLOCK_VALUES, MiniBlockChunk, MiniBlockCompressed, MiniBlockCompressionContext,
27    MiniBlockCompressor,
28};
29use crate::encodings::physical::checked_fixed_values;
30use crate::format::metadata::CompressiveEncoding;
31use crate::format::metadata::compressive_encoding::Compression;
32use crate::format::{EncodingUtils, metadata};
33
34use lance_core::utils::bit::pad_bytes_to;
35use lance_core::{Error, Result};
36
37use super::sequence::{BlockCost, SequenceBlockCodec, select_block_codec};
38
39#[derive(Debug)]
40pub struct BinaryMiniBlockEncoder {
41    minichunk_size: i64,
42    generic_offsets: Option<CompressionFieldParams>,
43}
44
45impl Default for BinaryMiniBlockEncoder {
46    fn default() -> Self {
47        Self {
48            minichunk_size: *AIM_MINICHUNK_SIZE,
49            generic_offsets: None,
50        }
51    }
52}
53
54const DEFAULT_AIM_MINICHUNK_SIZE: i64 = 4 * 1024;
55
56pub static AIM_MINICHUNK_SIZE: std::sync::LazyLock<i64> = std::sync::LazyLock::new(|| {
57    std::env::var("LANCE_BINARY_MINIBLOCK_CHUNK_SIZE")
58        .unwrap_or_else(|_| DEFAULT_AIM_MINICHUNK_SIZE.to_string())
59        .parse::<i64>()
60        .unwrap_or(DEFAULT_AIM_MINICHUNK_SIZE)
61});
62
63#[derive(Debug, Clone, Copy)]
64struct BinaryChunkRange {
65    start_offset_index: usize,
66    end_offset_index: usize,
67}
68
69fn binary_chunk_ranges<N: OffsetSizeTrait>(
70    offsets: &[N],
71    minichunk_size: i64,
72) -> Vec<BinaryChunkRange> {
73    let mut ranges = Vec::new();
74    let mut start = 0;
75    loop {
76        let end = search_next_offset_idx(offsets, start, minichunk_size);
77        ranges.push(BinaryChunkRange {
78            start_offset_index: start,
79            end_offset_index: end,
80        });
81        if end == offsets.len() - 1 {
82            return ranges;
83        }
84        start = end;
85    }
86}
87
88// Make it to support both u32 and u64
89fn chunk_offsets<N: OffsetSizeTrait>(
90    offsets: &[N],
91    data: &[u8],
92    alignment: usize,
93    minichunk_size: i64,
94) -> (Vec<LanceBuffer>, Vec<MiniBlockChunk>) {
95    #[derive(Debug)]
96    struct ChunkInfo {
97        chunk_start_offset_in_orig_idx: usize,
98        chunk_last_offset_in_orig_idx: usize,
99        // the bytes in every chunk starts at `chunk.bytes_start_offset`
100        bytes_start_offset: usize,
101        // every chunk is padded to 8 bytes.
102        // we need to interpret every chunk as &[u32] so we need it to padded at least to 4 bytes,
103        // this field can actually be eliminated and I can use `num_bytes` in `MiniBlockChunk` to compute
104        // the `output_total_bytes`.
105        padded_chunk_size: usize,
106    }
107
108    let byte_width: usize = N::get_byte_width();
109    let mut chunks_info = vec![];
110    let mut chunks = vec![];
111    let mut last_offset_in_orig_idx = 0;
112    loop {
113        let this_last_offset_in_orig_idx =
114            search_next_offset_idx(offsets, last_offset_in_orig_idx, minichunk_size);
115
116        let num_values_in_this_chunk = this_last_offset_in_orig_idx - last_offset_in_orig_idx;
117        let chunk_bytes = offsets[this_last_offset_in_orig_idx] - offsets[last_offset_in_orig_idx];
118        let this_chunk_size =
119            (num_values_in_this_chunk + 1) * byte_width + chunk_bytes.to_usize().unwrap();
120
121        let padded_chunk_size = this_chunk_size.next_multiple_of(alignment);
122        debug_assert!(padded_chunk_size > 0);
123
124        let this_chunk_bytes_start_offset = (num_values_in_this_chunk + 1) * byte_width;
125        chunks_info.push(ChunkInfo {
126            chunk_start_offset_in_orig_idx: last_offset_in_orig_idx,
127            chunk_last_offset_in_orig_idx: this_last_offset_in_orig_idx,
128            bytes_start_offset: this_chunk_bytes_start_offset,
129            padded_chunk_size,
130        });
131        chunks.push(MiniBlockChunk {
132            log_num_values: if this_last_offset_in_orig_idx == offsets.len() - 1 {
133                0
134            } else {
135                num_values_in_this_chunk.trailing_zeros() as u8
136            },
137            buffer_sizes: vec![padded_chunk_size as u32],
138        });
139        if this_last_offset_in_orig_idx == offsets.len() - 1 {
140            break;
141        }
142        last_offset_in_orig_idx = this_last_offset_in_orig_idx;
143    }
144
145    let output_total_bytes = chunks_info
146        .iter()
147        .map(|chunk_info| chunk_info.padded_chunk_size)
148        .sum::<usize>();
149
150    let mut output: Vec<u8> = Vec::with_capacity(output_total_bytes);
151
152    for chunk in chunks_info {
153        let this_chunk_offsets: Vec<N> = offsets
154            [chunk.chunk_start_offset_in_orig_idx..=chunk.chunk_last_offset_in_orig_idx]
155            .iter()
156            .map(|offset| {
157                *offset - offsets[chunk.chunk_start_offset_in_orig_idx]
158                    + N::from_usize(chunk.bytes_start_offset).unwrap()
159            })
160            .collect();
161
162        let this_chunk_offsets = LanceBuffer::reinterpret_vec(this_chunk_offsets);
163        output.extend_from_slice(&this_chunk_offsets);
164
165        let start_in_orig = offsets[chunk.chunk_start_offset_in_orig_idx]
166            .to_usize()
167            .unwrap();
168        let end_in_orig = offsets[chunk.chunk_last_offset_in_orig_idx]
169            .to_usize()
170            .unwrap();
171        output.extend_from_slice(&data[start_in_orig..end_in_orig]);
172
173        // pad this chunk to make it align to desired bytes.
174        const PAD_BYTE: u8 = 72;
175        let pad_len = pad_bytes_to(output.len(), alignment);
176
177        // Compare with usize literal to avoid type mismatch with N
178        if pad_len > 0_usize {
179            output.extend(std::iter::repeat_n(PAD_BYTE, pad_len));
180        }
181    }
182    (vec![LanceBuffer::reinterpret_vec(output)], chunks)
183}
184
185// search for the next offset index to cut the values into a chunk.
186// this function incrementally peek the number of values in a chunk,
187// each time multiplies the number of values by 2.
188// It returns the offset_idx in `offsets` that belongs to this chunk.
189fn search_next_offset_idx<N: OffsetSizeTrait>(
190    offsets: &[N],
191    last_offset_idx: usize,
192    minichunk_size: i64,
193) -> usize {
194    // MiniBlockChunk uses `log_num_values == 0` as a sentinel for the final chunk. This means we
195    // must avoid creating 1-value chunks except for the final chunk, even if the configured
196    // `minichunk_size` is too small to fit more than one value.
197    let remaining_values = offsets.len().saturating_sub(last_offset_idx + 1);
198    if remaining_values <= 1 {
199        return offsets.len() - 1;
200    }
201
202    let mut num_values = 2;
203    let mut new_num_values = num_values * 2;
204    loop {
205        if last_offset_idx + new_num_values >= offsets.len() {
206            let existing_bytes = offsets[offsets.len() - 1] - offsets[last_offset_idx];
207            // existing bytes plus the new offset size
208            let new_size = existing_bytes
209                + N::from_usize((offsets.len() - last_offset_idx) * N::get_byte_width()).unwrap();
210            if new_size.to_i64().unwrap() <= minichunk_size {
211                // case 1: can fit the rest of all data into a miniblock
212                return offsets.len() - 1;
213            } else {
214                // case 2: can only fit the last tried `num_values` into a miniblock
215                return last_offset_idx + num_values;
216            }
217        }
218        let existing_bytes = offsets[last_offset_idx + new_num_values] - offsets[last_offset_idx];
219        let new_size =
220            existing_bytes + N::from_usize((new_num_values + 1) * N::get_byte_width()).unwrap();
221        if new_size.to_i64().unwrap() <= minichunk_size {
222            if new_num_values * 2 > *MAX_MINIBLOCK_VALUES as usize {
223                // hit the max number of values limit
224                break;
225            }
226            num_values = new_num_values;
227            new_num_values *= 2;
228        } else {
229            break;
230        }
231    }
232    last_offset_idx + num_values
233}
234
235fn validate_variable_offsets<N: OffsetSizeTrait>(
236    offsets: &[N],
237    num_values: u64,
238    data_len: usize,
239) -> Result<()> {
240    let expected_offsets = usize::try_from(num_values)
241        .ok()
242        .and_then(|num_values| num_values.checked_add(1))
243        .ok_or_else(|| Error::invalid_input("Variable-width offset count overflows usize"))?;
244    if offsets.len() != expected_offsets {
245        return Err(Error::invalid_input(format!(
246            "Variable-width block has {} offsets, expected {expected_offsets}",
247            offsets.len()
248        )));
249    }
250    let mut previous = None;
251    for (index, offset) in offsets.iter().enumerate() {
252        let offset = offset.to_usize().ok_or_else(|| {
253            Error::invalid_input(format!(
254                "Variable-width offset at index {index} is negative or does not fit usize"
255            ))
256        })?;
257        if previous.is_some_and(|previous| offset < previous) {
258            return Err(Error::invalid_input(format!(
259                "Variable-width offsets decrease at index {index}"
260            )));
261        }
262        previous = Some(offset);
263    }
264    if offsets[0].to_usize() != Some(0) {
265        return Err(Error::invalid_input(
266            "Variable-width offsets must start at zero",
267        ));
268    }
269    if previous != Some(data_len) {
270        return Err(Error::invalid_input(format!(
271            "Final variable-width offset {:?} does not equal {data_len} data bytes",
272            previous
273        )));
274    }
275    Ok(())
276}
277
278fn chunk_value_range<N: OffsetSizeTrait>(
279    offsets: &[N],
280    range: BinaryChunkRange,
281) -> Result<std::ops::Range<usize>> {
282    let start = offsets[range.start_offset_index]
283        .to_usize()
284        .ok_or_else(|| Error::invalid_input("Variable chunk start offset does not fit usize"))?;
285    let end = offsets[range.end_offset_index]
286        .to_usize()
287        .ok_or_else(|| Error::invalid_input("Variable chunk end offset does not fit usize"))?;
288    if start > end {
289        return Err(Error::invalid_input(
290            "Variable chunk offsets are decreasing",
291        ));
292    }
293    Ok(start..end)
294}
295
296fn serialized_variable_cost(
297    compressed: &MiniBlockCompressed,
298    encoding: &CompressiveEncoding,
299    context: MiniBlockCompressionContext,
300) -> u64 {
301    compressed
302        .chunks
303        .iter()
304        .fold(encoding.encoded_len() as u64, |total, chunk| {
305            chunk.buffer_sizes.iter().fold(
306                total.saturating_add(context.chunk_header_bytes(chunk.buffer_sizes.len() as u64)),
307                |total, size| total.saturating_add(u64::from(*size).next_multiple_of(8)),
308            )
309        })
310}
311
312// Compare complete layouts, including value-buffer alignment. Pricing offsets
313// alone can select Delta(Flat) just for omitting its first offset, then lose all
314// those bytes again to the second buffer's header and padding.
315fn offset_layout_costs<N: OffsetSizeTrait>(
316    offsets: &[N],
317    ranges: &[BinaryChunkRange],
318    codec: &SequenceBlockCodec,
319    legacy_encoding: &CompressiveEncoding,
320    context: MiniBlockCompressionContext,
321) -> Result<(u64, u64)> {
322    let encoding = EncodingUtils::variable(codec.expected_encoding().clone(), None);
323    let mut generic = codec.estimated_wire_bytes()
324        + (encoding.encoded_len() - codec.expected_encoding().encoded_len()) as u64;
325    let mut legacy = legacy_encoding.encoded_len() as u64;
326    for range in ranges {
327        let values = chunk_value_range(offsets, *range)?.len();
328        let count = range.end_offset_index - range.start_offset_index + 1;
329        let bytes = values
330            .checked_add(
331                count
332                    .checked_mul(N::get_byte_width())
333                    .ok_or_else(|| Error::invalid_input("Offset bytes overflow"))?,
334            )
335            .ok_or_else(|| Error::invalid_input("Variable chunk bytes overflow"))?;
336        legacy += context.chunk_header_bytes(1) + bytes.next_multiple_of(8) as u64;
337        generic += context.chunk_header_bytes(1) + values.next_multiple_of(8) as u64;
338    }
339    Ok((generic, legacy))
340}
341
342fn build_generic_chunks<N: OffsetSizeTrait>(
343    data: &VariableWidthBlock,
344    offsets: &[N],
345    ranges: &[BinaryChunkRange],
346    codec: &SequenceBlockCodec,
347) -> Result<(MiniBlockCompressed, CompressiveEncoding)> {
348    let bytes_per_offset = (data.bits_per_offset / 8) as usize;
349    let mut offset_data = Vec::new();
350    let mut value_data = Vec::new();
351    let mut chunks = Vec::with_capacity(ranges.len());
352    let mut actual_encoding = None;
353
354    for (index, range) in ranges.iter().enumerate() {
355        let num_offsets = range.end_offset_index - range.start_offset_index + 1;
356        let offset_start = range
357            .start_offset_index
358            .checked_mul(bytes_per_offset)
359            .ok_or_else(|| Error::invalid_input("Offset chunk start overflows usize"))?;
360        let offset_bytes = num_offsets
361            .checked_mul(bytes_per_offset)
362            .ok_or_else(|| Error::invalid_input("Offset chunk size overflows usize"))?;
363        let offset_block = FixedWidthDataBlock {
364            bits_per_value: data.bits_per_offset as u64,
365            data: data.offsets.slice_with_length(offset_start, offset_bytes),
366            num_values: num_offsets as u64,
367            block_info: BlockInfo::default(),
368        };
369        let (payload, encoding) = codec.compress(offset_block)?;
370        if actual_encoding
371            .as_ref()
372            .is_some_and(|actual| actual != &encoding)
373        {
374            return Err(Error::internal(format!(
375                "Offset codec produced inconsistent descriptors for chunk {index}"
376            )));
377        }
378        actual_encoding.get_or_insert(encoding);
379
380        let value_range = chunk_value_range(offsets, *range)?;
381        let value_bytes = data.data.get(value_range.clone()).ok_or_else(|| {
382            Error::invalid_input(format!(
383                "Variable chunk range {}..{} exceeds {} bytes",
384                value_range.start,
385                value_range.end,
386                data.data.len()
387            ))
388        })?;
389        let mut buffer_sizes = Vec::with_capacity(1 + usize::from(codec.has_payload()));
390        if let Some(payload) = payload {
391            buffer_sizes.push(
392                u32::try_from(payload.len())
393                    .map_err(|_| Error::invalid_input("Offset payload exceeds u32::MAX bytes"))?,
394            );
395            offset_data.extend_from_slice(&payload);
396        }
397        buffer_sizes.push(
398            u32::try_from(value_bytes.len()).map_err(|_| {
399                Error::invalid_input("Variable value payload exceeds u32::MAX bytes")
400            })?,
401        );
402        value_data.extend_from_slice(value_bytes);
403        let num_values = range.end_offset_index - range.start_offset_index;
404        chunks.push(MiniBlockChunk {
405            log_num_values: if index + 1 == ranges.len() {
406                0
407            } else {
408                num_values.trailing_zeros() as u8
409            },
410            buffer_sizes,
411        });
412    }
413
414    let mut buffers = Vec::with_capacity(1 + usize::from(codec.has_payload()));
415    if codec.has_payload() {
416        buffers.push(LanceBuffer::from(offset_data));
417    }
418    buffers.push(LanceBuffer::from(value_data));
419    let offsets_encoding = actual_encoding.ok_or_else(|| {
420        Error::internal("Variable-width page did not contain an offset chunk".to_string())
421    })?;
422    Ok((
423        MiniBlockCompressed {
424            data: buffers,
425            chunks,
426            num_values: data.num_values,
427        },
428        EncodingUtils::variable(offsets_encoding, None),
429    ))
430}
431
432impl BinaryMiniBlockEncoder {
433    pub fn new(minichunk_size: Option<i64>) -> Self {
434        Self {
435            minichunk_size: minichunk_size.unwrap_or(*AIM_MINICHUNK_SIZE),
436            generic_offsets: None,
437        }
438    }
439
440    pub fn with_generic_offsets(
441        minichunk_size: Option<i64>,
442        field_params: CompressionFieldParams,
443    ) -> Self {
444        Self {
445            minichunk_size: minichunk_size.unwrap_or(*AIM_MINICHUNK_SIZE),
446            generic_offsets: Some(field_params),
447        }
448    }
449
450    // put binary data into chunks, every chunk is less than or equal to `minichunk_size`.
451    // In each chunk, offsets are put first then followed by binary bytes data, each chunk is padded to 8 bytes.
452    // the offsets in the chunk points to the bytes offset in this chunk.
453    fn chunk_data(
454        &self,
455        data: VariableWidthBlock,
456        context: MiniBlockCompressionContext,
457    ) -> Result<(MiniBlockCompressed, CompressiveEncoding)> {
458        match data.bits_per_offset {
459            32 => {
460                let offsets_buffer = data.offsets.clone();
461                let offsets = offsets_buffer.borrow_to_typed_slice::<i32>();
462                self.chunk_typed_data(offsets.as_ref(), data, 4, context)
463            }
464            64 => {
465                let offsets_buffer = data.offsets.clone();
466                let offsets = offsets_buffer.borrow_to_typed_slice::<i64>();
467                self.chunk_typed_data(offsets.as_ref(), data, 8, context)
468            }
469            _ => Err(Error::invalid_input(format!(
470                "Unsupported bits_per_offset={}",
471                data.bits_per_offset
472            ))),
473        }
474    }
475
476    fn chunk_typed_data<N: OffsetSizeTrait>(
477        &self,
478        offsets: &[N],
479        data: VariableWidthBlock,
480        legacy_alignment: usize,
481        context: MiniBlockCompressionContext,
482    ) -> Result<(MiniBlockCompressed, CompressiveEncoding)> {
483        validate_variable_offsets(offsets, data.num_values, data.data.len())?;
484        let bits_per_offset = data.bits_per_offset as u64;
485        let legacy_encoding =
486            EncodingUtils::variable(EncodingUtils::flat(bits_per_offset, None), None);
487        let legacy = || {
488            let (buffers, chunks) =
489                chunk_offsets(offsets, &data.data, legacy_alignment, self.minichunk_size);
490            (
491                MiniBlockCompressed {
492                    data: buffers,
493                    chunks,
494                    num_values: data.num_values,
495                },
496                legacy_encoding.clone(),
497            )
498        };
499        let Some(field_params) = self
500            .generic_offsets
501            .as_ref()
502            .filter(|_| context.allows_generic_offsets())
503        else {
504            return Ok(legacy());
505        };
506
507        let ranges = binary_chunk_ranges(offsets, self.minichunk_size);
508        let member_ranges = ranges
509            .iter()
510            .map(|range| range.start_offset_index..range.end_offset_index + 1)
511            .collect::<Vec<_>>();
512        let offsets_block = FixedWidthDataBlock {
513            bits_per_value: bits_per_offset,
514            data: data.offsets.clone(),
515            num_values: offsets.len() as u64,
516            block_info: BlockInfo::default(),
517        };
518        let extra_payload_header = context
519            .chunk_header_bytes(2)
520            .saturating_sub(context.chunk_header_bytes(1));
521        // Bound default-policy exploration when many small chunks are likely
522        // to remain flat. Sample whole chunks across the page, not tiny prefixes
523        // of every chunk; the final selection still checks every member.
524        if field_params.compression.is_none() && member_ranges.len() > 8 {
525            let sample_ranges = (0..8)
526                .map(|i| member_ranges[i * (member_ranges.len() - 1) / 7].clone())
527                .collect::<Vec<_>>();
528            let sample = select_block_codec(
529                &offsets_block,
530                &sample_ranges,
531                field_params,
532                BlockCost::new(extra_payload_header, 8),
533                true,
534            )?;
535            let sample_chunks = (0..8)
536                .map(|i| ranges[i * (ranges.len() - 1) / 7])
537                .collect::<Vec<_>>();
538            let (generic_cost, legacy_cost) =
539                offset_layout_costs(offsets, &sample_chunks, &sample, &legacy_encoding, context)?;
540            if generic_cost >= legacy_cost {
541                return Ok(legacy());
542            }
543        }
544        let codec = select_block_codec(
545            &offsets_block,
546            &member_ranges,
547            field_params,
548            BlockCost::new(extra_payload_header, 8),
549            true,
550        )?;
551        if matches!(
552            codec.expected_encoding().compression.as_ref(),
553            Some(Compression::Flat(_))
554        ) {
555            return Ok(legacy());
556        }
557
558        let (generic_cost, legacy_cost) =
559            offset_layout_costs(offsets, &ranges, &codec, &legacy_encoding, context)?;
560        if generic_cost >= legacy_cost {
561            return Ok(legacy());
562        }
563        let (generic, generic_encoding) = build_generic_chunks(&data, offsets, &ranges, &codec)?;
564        if serialized_variable_cost(&generic, &generic_encoding, context) < legacy_cost {
565            Ok((generic, generic_encoding))
566        } else {
567            Ok(legacy())
568        }
569    }
570}
571
572impl MiniBlockCompressor for BinaryMiniBlockEncoder {
573    fn compress(
574        &self,
575        context: MiniBlockCompressionContext,
576        data: DataBlock,
577    ) -> Result<(MiniBlockCompressed, CompressiveEncoding)> {
578        match data {
579            DataBlock::VariableWidth(variable_width) => self.chunk_data(variable_width, context),
580            _ => Err(Error::invalid_input_source(
581                format!(
582                    "Cannot compress a data block of type {} with BinaryMiniBlockEncoder",
583                    data.name()
584                )
585                .into(),
586            )),
587        }
588    }
589}
590
591#[derive(Debug)]
592pub struct BinaryMiniBlockDecompressor {
593    layout: BinaryMiniBlockLayout,
594}
595
596#[derive(Debug)]
597enum BinaryMiniBlockLayout {
598    Legacy {
599        bits_per_offset: u8,
600    },
601    Generic {
602        bits_per_offset: u8,
603        offsets: Box<dyn BlockDecompressor>,
604        offsets_have_payload: bool,
605    },
606}
607
608impl BinaryMiniBlockDecompressor {
609    pub fn new(bits_per_offset: u8) -> Result<Self> {
610        if !matches!(bits_per_offset, 32 | 64) {
611            return Err(Error::invalid_input(format!(
612                "Binary mini-block offsets require 32 or 64 bits, got {bits_per_offset}"
613            )));
614        }
615        Ok(Self {
616            layout: BinaryMiniBlockLayout::Legacy { bits_per_offset },
617        })
618    }
619
620    pub fn from_variable(variable: &metadata::Variable) -> Result<Self> {
621        if variable.values.is_some() {
622            return Err(Error::invalid_input(
623                "Binary mini-block Variable encoding cannot contain a values codec",
624            ));
625        }
626        let offsets = variable.offsets.as_deref().ok_or_else(|| {
627            Error::invalid_input("Binary mini-block Variable encoding is missing offsets")
628        })?;
629        let compression = offsets.compression.as_ref().ok_or_else(|| {
630            Error::invalid_input("Binary mini-block offsets are missing a compression variant")
631        })?;
632        if let Compression::Flat(flat) = compression {
633            if flat.data.is_some() || !matches!(flat.bits_per_value, 32 | 64) {
634                return Err(Error::invalid_input(format!(
635                    "Legacy binary mini-block offsets require plain 32 or 64-bit Flat encoding, got {} bits",
636                    flat.bits_per_value
637                )));
638            }
639            return Self::new(flat.bits_per_value as u8);
640        }
641
642        let bits_per_offset = infer_fixed_width_block_bits(offsets)?;
643        let offsets = create_fixed_width_block_decompressor(offsets, bits_per_offset)?;
644        let offsets_have_payload = offsets.requires_payload();
645        Ok(Self {
646            layout: BinaryMiniBlockLayout::Generic {
647                bits_per_offset: bits_per_offset as u8,
648                offsets,
649                offsets_have_payload,
650            },
651        })
652    }
653}
654
655fn decode_generic_binary_miniblock(
656    bits_per_offset: u8,
657    offsets: &dyn BlockDecompressor,
658    offsets_have_payload: bool,
659    data: Vec<LanceBuffer>,
660    num_values: u64,
661) -> Result<DataBlock> {
662    let expected_buffers = 1 + usize::from(offsets_have_payload);
663    if data.len() != expected_buffers {
664        return Err(Error::corrupt_file_named(
665            "binary mini-block",
666            format!(
667                "generic chunk has {} buffers, expected {expected_buffers}",
668                data.len()
669            ),
670        ));
671    }
672    let mut buffers = data.into_iter();
673    let offset_payload =
674        offsets_have_payload.then(|| buffers.next().expect("buffer count was checked"));
675    let values = buffers.next().expect("buffer count was checked");
676    let num_offsets = num_values.checked_add(1).ok_or_else(|| {
677        Error::corrupt_file_named(
678            "binary mini-block",
679            format!("cannot decode offsets for {num_values} values"),
680        )
681    })?;
682    let decoded = offsets.decompress(offset_payload, num_offsets)?;
683    let DataBlock::FixedWidth(offsets) = decoded else {
684        return Err(Error::corrupt_file_named(
685            "binary mini-block",
686            "generic offset codec did not produce fixed-width offsets",
687        ));
688    };
689    if offsets.bits_per_value != u64::from(bits_per_offset) || offsets.num_values != num_offsets {
690        return Err(Error::corrupt_file_named(
691            "binary mini-block",
692            format!(
693                "generic offset codec produced {} {}-bit offsets, expected {num_offsets} {bits_per_offset}-bit offsets",
694                offsets.num_values, offsets.bits_per_value
695            ),
696        ));
697    }
698
699    match bits_per_offset {
700        32 => {
701            let typed = checked_fixed_values::<u32>(&offsets, "Binary mini-block offsets")?;
702            let mut previous = 0_u32;
703            for (position, &offset) in typed.iter().enumerate() {
704                if (position == 0 && offset != 0) || offset < previous || offset > i32::MAX as u32 {
705                    return Err(Error::corrupt_file_named(
706                        "binary mini-block",
707                        format!("invalid 32-bit generic offset {offset} at position {position}"),
708                    ));
709                }
710                previous = offset;
711            }
712            if previous as usize != values.len() {
713                return Err(Error::corrupt_file_named(
714                    "binary mini-block",
715                    format!(
716                        "final generic offset {previous} does not equal the {}-byte value buffer",
717                        values.len()
718                    ),
719                ));
720            }
721        }
722        64 => {
723            let typed = checked_fixed_values::<u64>(&offsets, "Binary mini-block offsets")?;
724            let mut previous = 0_u64;
725            for (position, &offset) in typed.iter().enumerate() {
726                if (position == 0 && offset != 0) || offset < previous || offset > i64::MAX as u64 {
727                    return Err(Error::corrupt_file_named(
728                        "binary mini-block",
729                        format!("invalid 64-bit generic offset {offset} at position {position}"),
730                    ));
731                }
732                previous = offset;
733            }
734            let value_len = u64::try_from(values.len()).map_err(|_| {
735                Error::corrupt_file_named(
736                    "binary mini-block",
737                    "value buffer length does not fit u64",
738                )
739            })?;
740            if previous != value_len {
741                return Err(Error::corrupt_file_named(
742                    "binary mini-block",
743                    format!(
744                        "final generic offset {previous} does not equal the {}-byte value buffer",
745                        values.len()
746                    ),
747                ));
748            }
749        }
750        _ => unreachable!("generic offset width was validated during decoder construction"),
751    }
752
753    Ok(DataBlock::VariableWidth(VariableWidthBlock {
754        data: values,
755        offsets: offsets.data,
756        bits_per_offset,
757        num_values,
758        block_info: BlockInfo::new(),
759    }))
760}
761
762/// Cold path: pinpoint why the chunk-relative offsets of a binary mini-block
763/// chunk failed validation.
764fn chunk_offset_violation_error<T: Copy + Into<u64>>(offsets: &[T], chunk_len: usize) -> Error {
765    let mut previous: u64 = offsets[0].into();
766    for (position, &offset) in offsets.iter().enumerate().skip(1) {
767        let offset: u64 = offset.into();
768        if offset < previous {
769            return Error::corrupt_file_named(
770                "binary mini-block",
771                format!(
772                    "value offset at position {position} decreases: {offset} < {previous} \
773                     (chunk is {chunk_len} bytes)"
774                ),
775            );
776        }
777        previous = offset;
778    }
779    Error::corrupt_file_named(
780        "binary mini-block",
781        format!("value offset {previous} is out of bounds for a chunk of {chunk_len} bytes"),
782    )
783}
784
785impl MiniBlockDecompressor for BinaryMiniBlockDecompressor {
786    fn num_buffers(&self) -> usize {
787        match self.layout {
788            BinaryMiniBlockLayout::Legacy { .. } => 1,
789            BinaryMiniBlockLayout::Generic {
790                offsets_have_payload,
791                ..
792            } => 1 + usize::from(offsets_have_payload),
793        }
794    }
795
796    // decompress a MiniBlock of binary data, the num_values must be less than or equal
797    // to the number of values this MiniBlock has, BinaryMiniBlock doesn't store `the number of values`
798    // it has so assertion can not be done here and the caller of `decompress` must ensure
799    // `num_values` <= number of values in the chunk.
800    //
801    // The chunk-relative value offsets at the front of the chunk come straight
802    // from the file and are used to slice the chunk buffer, so corrupt values
803    // must surface as a typed error instead of a panic or an out-of-bounds
804    // read.  The monotonicity check rides along the existing rebase loop (the
805    // `&=` accumulation keeps it branchless) so validation adds no extra pass.
806    fn decompress(&self, data: Vec<LanceBuffer>, num_values: u64) -> Result<DataBlock> {
807        let bits_per_offset = match &self.layout {
808            BinaryMiniBlockLayout::Legacy { bits_per_offset } => *bits_per_offset,
809            BinaryMiniBlockLayout::Generic {
810                bits_per_offset,
811                offsets,
812                offsets_have_payload,
813            } => {
814                return decode_generic_binary_miniblock(
815                    *bits_per_offset,
816                    offsets.as_ref(),
817                    *offsets_have_payload,
818                    data,
819                    num_values,
820                );
821            }
822        };
823        if data.len() != 1 {
824            return Err(Error::corrupt_file_named(
825                "binary mini-block",
826                format!("legacy chunk has {} buffers, expected 1", data.len()),
827            ));
828        }
829        let data = data.into_iter().next().expect("buffer count was checked");
830
831        let bytes_per_offset = bits_per_offset as usize / 8;
832        if !data.len().is_multiple_of(bytes_per_offset) {
833            return Err(Error::corrupt_file_named(
834                "binary mini-block",
835                format!(
836                    "chunk size {} is not a multiple of the {}-byte offset width",
837                    data.len(),
838                    bytes_per_offset
839                ),
840            ));
841        }
842        let num_offsets = (num_values as usize).checked_add(1).ok_or_else(|| {
843            Error::corrupt_file_named(
844                "binary mini-block",
845                format!("cannot decode {num_values} values from a single chunk"),
846            )
847        })?;
848        if data.len() / bytes_per_offset < num_offsets {
849            return Err(Error::corrupt_file_named(
850                "binary mini-block",
851                format!(
852                    "chunk of {} bytes holds {} offsets but decoding {} values requires {}",
853                    data.len(),
854                    data.len() / bytes_per_offset,
855                    num_values,
856                    num_offsets
857                ),
858            ));
859        }
860
861        // The value region must start past the offsets being decoded, otherwise
862        // the offset table itself aliases into the value bytes.  A lower bound
863        // (not equality) because a prefix read of the chunk legitimately leaves
864        // unrequested offsets between the requested prefix and the values.
865        let min_value_region_start = num_offsets * bytes_per_offset;
866        let value_region_overlap_error = |first: u64| {
867            Error::corrupt_file_named(
868                "binary mini-block",
869                format!(
870                    "value region starts at offset {first} which overlaps the {num_offsets} \
871                     requested offsets ({min_value_region_start} bytes)"
872                ),
873            )
874        };
875
876        if bits_per_offset == 64 {
877            let offsets_buffer = data.borrow_to_typed_slice::<u64>();
878            let offsets = &offsets_buffer.as_ref()[..num_offsets];
879
880            let first = offsets[0];
881            if first < min_value_region_start as u64 {
882                return Err(value_region_overlap_error(first));
883            }
884            let mut previous = first;
885            let mut is_monotonic = true;
886            let result_offsets = offsets
887                .iter()
888                .map(|&offset| {
889                    is_monotonic &= previous <= offset;
890                    previous = offset;
891                    offset.wrapping_sub(first)
892                })
893                .collect::<Vec<u64>>();
894            let last = offsets[num_offsets - 1];
895            if !is_monotonic || last as usize > data.len() {
896                return Err(chunk_offset_violation_error(offsets, data.len()));
897            }
898
899            Ok(DataBlock::VariableWidth(VariableWidthBlock {
900                data: LanceBuffer::from(data[first as usize..last as usize].to_vec()),
901                offsets: LanceBuffer::reinterpret_vec(result_offsets),
902                bits_per_offset: 64,
903                num_values,
904                block_info: BlockInfo::new(),
905            }))
906        } else {
907            let offsets_buffer = data.borrow_to_typed_slice::<u32>();
908            let offsets = &offsets_buffer.as_ref()[..num_offsets];
909
910            let first = offsets[0];
911            if (first as u64) < min_value_region_start as u64 {
912                return Err(value_region_overlap_error(first as u64));
913            }
914            let mut previous = first;
915            let mut is_monotonic = true;
916            let result_offsets = offsets
917                .iter()
918                .map(|&offset| {
919                    is_monotonic &= previous <= offset;
920                    previous = offset;
921                    offset.wrapping_sub(first)
922                })
923                .collect::<Vec<u32>>();
924            let last = offsets[num_offsets - 1];
925            if !is_monotonic || last as usize > data.len() {
926                return Err(chunk_offset_violation_error(offsets, data.len()));
927            }
928
929            Ok(DataBlock::VariableWidth(VariableWidthBlock {
930                data: LanceBuffer::from(data[first as usize..last as usize].to_vec()),
931                offsets: LanceBuffer::reinterpret_vec(result_offsets),
932                bits_per_offset: 32,
933                num_values,
934                block_info: BlockInfo::new(),
935            }))
936        }
937    }
938}
939
940/// Most basic encoding for variable-width data which does no compression at all
941/// The DataBlock memory layout looks like below:
942///
943/// | bits_per_offset           | bytes_start_offset        | offsets data | bytes data |
944/// | ------------------------- | ------------------------- | ------------ | ---------- |
945/// | <bits_per_offset>/8 bytes | <bits_per_offset>/8 bytes | offsets_len  | data_len   |
946///
947/// It's used in VariableEncoder and BinaryBlockDecompressor
948///
949#[derive(Debug, Default)]
950pub struct VariableEncoder {}
951
952impl BlockCompressor for VariableEncoder {
953    fn compress(&self, mut data: DataBlock) -> Result<(Option<LanceBuffer>, CompressiveEncoding)> {
954        let bits_per_offset = match &data {
955            DataBlock::VariableWidth(data) => data.bits_per_offset,
956            _ => {
957                return Err(Error::invalid_input(
958                    "BinaryBlockEncoder requires a variable-width block",
959                ));
960            }
961        };
962        match data {
963            DataBlock::VariableWidth(ref mut variable_width_data) => {
964                match variable_width_data.bits_per_offset {
965                    32 => {
966                        let offsets = variable_width_data.offsets.borrow_to_typed_slice::<u32>();
967                        let offsets = offsets.as_ref();
968                        // The first 4 bytes store the bits per offset, the next 4 bytes store the start
969                        // offset of the bytes data, then offsets data, then bytes data.
970                        let bytes_start_offset = 4 + 4 + std::mem::size_of_val(offsets) as u32;
971
972                        let output_total_bytes =
973                            bytes_start_offset as usize + variable_width_data.data.len();
974                        let mut output: Vec<u8> = Vec::with_capacity(output_total_bytes);
975
976                        // Store bit_per_offset info
977                        output.extend_from_slice(&(32_u32).to_le_bytes());
978
979                        // store `bytes_start_offset` in the next 4 bytes of output buffer
980                        output.extend_from_slice(&(bytes_start_offset).to_le_bytes());
981
982                        // store offsets
983                        output.extend_from_slice(&variable_width_data.offsets);
984
985                        // store bytes
986                        output.extend_from_slice(&variable_width_data.data);
987                        Ok(LanceBuffer::from(output))
988                    }
989                    64 => {
990                        let offsets = variable_width_data.offsets.borrow_to_typed_slice::<u64>();
991                        let offsets = offsets.as_ref();
992                        // The first 8 bytes store the bits per offset, the next 8 bytes store the start
993                        // offset of the bytes data, then offsets data, then bytes data.
994                        let bytes_start_offset = 8 + 8 + std::mem::size_of_val(offsets) as u64;
995
996                        let output_total_bytes =
997                            bytes_start_offset as usize + variable_width_data.data.len();
998                        let mut output: Vec<u8> = Vec::with_capacity(output_total_bytes);
999
1000                        // Store bit_per_offset info
1001                        output.extend_from_slice(&(64_u64).to_le_bytes());
1002
1003                        // store `bytes_start_offset` in the next 8 bytes of output buffer
1004                        output.extend_from_slice(&(bytes_start_offset).to_le_bytes());
1005
1006                        // store offsets
1007                        output.extend_from_slice(&variable_width_data.offsets);
1008
1009                        // store bytes
1010                        output.extend_from_slice(&variable_width_data.data);
1011                        Ok(LanceBuffer::from(output))
1012                    }
1013                    _ => Err(Error::invalid_input(format!(
1014                        "BinaryBlockEncoder does not support {}-bit offsets",
1015                        variable_width_data.bits_per_offset
1016                    ))),
1017                }
1018            }
1019            _ => unreachable!("variable-width input was validated above"),
1020        }
1021        .map(|payload| {
1022            (
1023                Some(payload),
1024                EncodingUtils::variable(EncodingUtils::flat(bits_per_offset as u64, None), None),
1025            )
1026        })
1027    }
1028}
1029
1030impl PerValueCompressor for VariableEncoder {
1031    fn compress(&self, data: DataBlock) -> Result<(PerValueDataBlock, CompressiveEncoding)> {
1032        let DataBlock::VariableWidth(variable) = data else {
1033            panic!("BinaryPerValueCompressor can only work with Variable Width DataBlock.");
1034        };
1035
1036        let encoding = EncodingUtils::variable(
1037            EncodingUtils::flat(variable.bits_per_offset as u64, None),
1038            None,
1039        );
1040        Ok((PerValueDataBlock::Variable(variable), encoding))
1041    }
1042}
1043
1044#[derive(Debug, Default)]
1045pub struct VariableDecoder {}
1046
1047impl VariablePerValueDecompressor for VariableDecoder {
1048    fn decompress(&self, data: VariableWidthBlock) -> Result<DataBlock> {
1049        Ok(DataBlock::VariableWidth(data))
1050    }
1051}
1052
1053#[derive(Debug, Default)]
1054pub struct BinaryBlockDecompressor {}
1055
1056impl BlockDecompressor for BinaryBlockDecompressor {
1057    fn decompress(&self, data: Option<LanceBuffer>, num_values: u64) -> Result<DataBlock> {
1058        let data = require_block_payload(data, "Binary block")?;
1059        // In older (not quite stable) versions we stored the bits per offset as a single byte and then the num_values
1060        // as four bytes.  However, this led to alignment problems and was wasteful since we already store the num_values
1061        // in higher layers.
1062        //
1063        // In the standard scheme we use 4 bytes for the bits per offset and 4 bytes for the bytes_start_offset and we
1064        // rely on the passed in num_values to be correct.
1065
1066        // This isn't perfect but it's probably good enough and the best I think we can do.  The bits per offset will
1067        // never be more than 255 and it's little endian so the last 3 bytes will always be 0.  These will be the least
1068        // significant 3 bytes of the number of values in the old scheme.  It's pretty unlikely these are all 0 (that would
1069        // mean there are at least 16M values in a single page) so we'll use this to determine if the old scheme is used.
1070        //
1071        // The header fields and the offsets themselves come straight from the file.
1072        // The structural checks below (all O(1)) reject blocks whose regions do not
1073        // line up; the offset *values* are validated later, by the mandatory layout
1074        // validation in `VariableWidthBlock::into_arrow`, so they are not rescanned
1075        // here.
1076        if data.len() < 4 {
1077            return Err(Error::corrupt_file_named(
1078                "variable-width block",
1079                format!(
1080                    "block of {} bytes is too small to hold a header",
1081                    data.len()
1082                ),
1083            ));
1084        }
1085        let is_old_scheme = data[1] != 0 || data[2] != 0 || data[3] != 0;
1086
1087        let ensure_header = |header_len: usize| {
1088            if data.len() < header_len {
1089                return Err(Error::corrupt_file_named(
1090                    "variable-width block",
1091                    format!(
1092                        "block of {} bytes is too small for a {} byte header",
1093                        data.len(),
1094                        header_len
1095                    ),
1096                ));
1097            }
1098            Ok(())
1099        };
1100        let (bits_per_offset, bytes_start_offset, offset_start) = if is_old_scheme {
1101            // Old scheme
1102            let bits_per_offset = data[0];
1103            match bits_per_offset {
1104                32 => {
1105                    ensure_header(9)?;
1106                    debug_assert_eq!(LittleEndian::read_u32(&data[1..5]), num_values as u32);
1107                    let bytes_start_offset = LittleEndian::read_u32(&data[5..9]);
1108                    (bits_per_offset, bytes_start_offset as u64, 9_u64)
1109                }
1110                64 => {
1111                    ensure_header(17)?;
1112                    debug_assert_eq!(LittleEndian::read_u64(&data[1..9]), num_values);
1113                    let bytes_start_offset = LittleEndian::read_u64(&data[9..17]);
1114                    (bits_per_offset, bytes_start_offset, 17)
1115                }
1116                _ => {
1117                    return Err(Error::invalid_input_source(
1118                        format!("Unsupported bits_per_offset={}", bits_per_offset).into(),
1119                    ));
1120                }
1121            }
1122        } else {
1123            // Standard scheme
1124            let bits_per_offset = LittleEndian::read_u32(&data[0..4]) as u8;
1125            match bits_per_offset {
1126                32 => {
1127                    ensure_header(8)?;
1128                    let bytes_start_offset = LittleEndian::read_u32(&data[4..8]);
1129                    (bits_per_offset, bytes_start_offset as u64, 8)
1130                }
1131                64 => {
1132                    ensure_header(16)?;
1133                    let bytes_start_offset = LittleEndian::read_u64(&data[8..16]);
1134                    (bits_per_offset, bytes_start_offset, 16)
1135                }
1136                _ => {
1137                    return Err(Error::invalid_input_source(
1138                        format!("Unsupported bits_per_offset={}", bits_per_offset).into(),
1139                    ));
1140                }
1141            }
1142        };
1143
1144        // The offsets region sits between the header and `bytes_start_offset`
1145        // and must hold exactly `num_values + 1` offsets starting at zero.
1146        let expected_offsets_bytes = num_values
1147            .checked_add(1)
1148            .and_then(|num_offsets| num_offsets.checked_mul(bits_per_offset as u64 / 8))
1149            .ok_or_else(|| {
1150                Error::corrupt_file_named(
1151                    "variable-width block",
1152                    format!("offsets region size overflows for {num_values} values"),
1153                )
1154            })?;
1155        if bytes_start_offset < offset_start || bytes_start_offset > data.len() as u64 {
1156            return Err(Error::corrupt_file_named(
1157                "variable-width block",
1158                format!(
1159                    "bytes start offset {} is outside the block (header: {} bytes, block: {} bytes)",
1160                    bytes_start_offset,
1161                    offset_start,
1162                    data.len()
1163                ),
1164            ));
1165        }
1166        if bytes_start_offset - offset_start != expected_offsets_bytes {
1167            return Err(Error::corrupt_file_named(
1168                "variable-width block",
1169                format!(
1170                    "expected {} offset bytes for {} values but found {}",
1171                    expected_offsets_bytes,
1172                    num_values,
1173                    bytes_start_offset - offset_start
1174                ),
1175            ));
1176        }
1177
1178        // the next `bytes_start_offset - offset_start` stores the offsets.
1179        let offsets = data.slice_with_length(
1180            offset_start as usize,
1181            (bytes_start_offset - offset_start) as usize,
1182        );
1183        let first_offset = match bits_per_offset {
1184            32 => LittleEndian::read_u32(&offsets[0..4]) as u64,
1185            _ => LittleEndian::read_u64(&offsets[0..8]),
1186        };
1187        if first_offset != 0 {
1188            return Err(Error::corrupt_file_named(
1189                "variable-width block",
1190                format!("first offset must be 0 but found {first_offset}"),
1191            ));
1192        }
1193
1194        // the rest are the binary bytes.
1195        let data = data.slice_with_length(
1196            bytes_start_offset as usize,
1197            data.len() - bytes_start_offset as usize,
1198        );
1199
1200        Ok(DataBlock::VariableWidth(VariableWidthBlock {
1201            data,
1202            offsets,
1203            bits_per_offset,
1204            num_values,
1205            block_info: BlockInfo::new(),
1206        }))
1207    }
1208}
1209
1210#[cfg(test)]
1211mod tests {
1212    use super::{BinaryMiniBlockDecompressor, BinaryMiniBlockEncoder};
1213    use arrow_array::{
1214        ArrayRef, StringArray,
1215        builder::{LargeStringBuilder, StringBuilder},
1216    };
1217    use arrow_schema::{DataType, Field};
1218    use lance_core::{Error, Result};
1219
1220    use crate::{
1221        buffer::LanceBuffer,
1222        compression::MiniBlockDecompressor,
1223        compression_config::CompressionFieldParams,
1224        constants::{
1225            COMPRESSION_META_KEY, DICT_DIVISOR_META_KEY, STRUCTURAL_ENCODING_FULLZIP,
1226            STRUCTURAL_ENCODING_META_KEY, STRUCTURAL_ENCODING_MINIBLOCK,
1227        },
1228        data::{BlockInfo, DataBlock, VariableWidthBlock},
1229        encodings::logical::primitive::miniblock::{
1230            MiniBlockCompressed, MiniBlockCompressionContext, MiniBlockCompressor,
1231        },
1232        format::metadata::compressive_encoding::Compression,
1233        format::{EncodingUtils, metadata, metadata::CompressiveEncoding},
1234        testing::check_specific_random,
1235    };
1236    use rstest::rstest;
1237    use std::{collections::HashMap, sync::Arc, vec};
1238
1239    use crate::testing::{
1240        FnArrayGeneratorProvider, TestCases, TestEncoding, check_basic_random_case,
1241        check_round_trip_encoding_generated, check_round_trip_encoding_of_data,
1242    };
1243
1244    fn miniblock_context() -> MiniBlockCompressionContext {
1245        MiniBlockCompressionContext::new(0, true, true)
1246    }
1247
1248    fn variable_block_u32(lengths: &[usize]) -> VariableWidthBlock {
1249        let mut offsets = Vec::with_capacity(lengths.len() + 1);
1250        let mut data = Vec::new();
1251        offsets.push(0_i32);
1252        for (index, length) in lengths.iter().copied().enumerate() {
1253            data.extend(std::iter::repeat_n((index % 251) as u8, length));
1254            offsets.push(i32::try_from(data.len()).unwrap());
1255        }
1256        VariableWidthBlock {
1257            data: LanceBuffer::from(data),
1258            offsets: LanceBuffer::reinterpret_vec(offsets),
1259            bits_per_offset: 32,
1260            num_values: lengths.len() as u64,
1261            block_info: BlockInfo::default(),
1262        }
1263    }
1264
1265    fn decode_binary_miniblocks(
1266        compressed: MiniBlockCompressed,
1267        encoding: &CompressiveEncoding,
1268    ) -> Result<Vec<VariableWidthBlock>> {
1269        let Some(Compression::Variable(variable)) = encoding.compression.as_ref() else {
1270            return Err(Error::invalid_input("expected Variable encoding"));
1271        };
1272        let decoder = BinaryMiniBlockDecompressor::from_variable(variable)?;
1273        let mut buffer_offsets = vec![0_usize; compressed.data.len()];
1274        let mut values_seen = 0_u64;
1275        let mut decoded = Vec::with_capacity(compressed.chunks.len());
1276        for chunk in compressed.chunks {
1277            let num_values = chunk.num_values(values_seen, compressed.num_values);
1278            values_seen += num_values;
1279            let buffers = chunk
1280                .buffer_sizes
1281                .iter()
1282                .zip(compressed.data.iter().zip(&mut buffer_offsets))
1283                .map(|(size, (buffer, offset))| {
1284                    let size = *size as usize;
1285                    let chunk = buffer.slice_with_length(*offset, size);
1286                    *offset += size;
1287                    chunk
1288                })
1289                .collect();
1290            let DataBlock::VariableWidth(block) = decoder.decompress(buffers, num_values)? else {
1291                return Err(Error::internal(
1292                    "Binary mini-block decoded a non-variable block".to_string(),
1293                ));
1294            };
1295            decoded.push(block);
1296        }
1297        Ok(decoded)
1298    }
1299
1300    fn assert_decoded_value_lengths(decoded: &[VariableWidthBlock], expected: &[usize]) {
1301        let actual = decoded
1302            .iter()
1303            .flat_map(|block| {
1304                let offsets = block.offsets.borrow_to_typed_slice::<i32>();
1305                offsets
1306                    .windows(2)
1307                    .map(|pair| (pair[1] - pair[0]) as usize)
1308                    .collect::<Vec<_>>()
1309            })
1310            .collect::<Vec<_>>();
1311        assert_eq!(actual, expected);
1312    }
1313
1314    #[test]
1315    fn generic_offsets_use_range_across_chunks() {
1316        let lengths = vec![3_usize; 2_048];
1317        let encoder = BinaryMiniBlockEncoder::with_generic_offsets(
1318            Some(256),
1319            CompressionFieldParams::default(),
1320        );
1321        let (compressed, encoding) = encoder
1322            .compress(
1323                miniblock_context(),
1324                DataBlock::VariableWidth(variable_block_u32(&lengths)),
1325            )
1326            .unwrap();
1327        let Some(Compression::Variable(variable)) = encoding.compression.as_ref() else {
1328            panic!("expected Variable encoding");
1329        };
1330        assert!(matches!(
1331            variable
1332                .offsets
1333                .as_deref()
1334                .and_then(|offsets| offsets.compression.as_ref()),
1335            Some(Compression::Range(_))
1336        ));
1337        assert_eq!(compressed.data.len(), 1);
1338        assert!(compressed.chunks.len() > 1);
1339        assert!(
1340            compressed
1341                .chunks
1342                .iter()
1343                .all(|chunk| chunk.buffer_sizes.len() == 1)
1344        );
1345        let decoded = decode_binary_miniblocks(compressed, &encoding).unwrap();
1346        assert_decoded_value_lengths(&decoded, &lengths);
1347    }
1348
1349    #[test]
1350    fn generic_offsets_use_delta_payload() {
1351        let lengths = (0..4_096)
1352            .map(|index| [1_usize, 7, 2, 5][index % 4])
1353            .collect::<Vec<_>>();
1354        let encoder = BinaryMiniBlockEncoder::with_generic_offsets(
1355            Some(1_024),
1356            CompressionFieldParams::default(),
1357        );
1358        let (compressed, encoding) = encoder
1359            .compress(
1360                miniblock_context(),
1361                DataBlock::VariableWidth(variable_block_u32(&lengths)),
1362            )
1363            .unwrap();
1364        let Some(Compression::Variable(variable)) = encoding.compression.as_ref() else {
1365            panic!("expected Variable encoding");
1366        };
1367        assert!(matches!(
1368            variable
1369                .offsets
1370                .as_deref()
1371                .and_then(|offsets| offsets.compression.as_ref()),
1372            Some(Compression::Delta(_))
1373        ));
1374        assert_eq!(compressed.data.len(), 2);
1375        assert!(
1376            compressed
1377                .chunks
1378                .iter()
1379                .all(|chunk| chunk.buffer_sizes.len() == 2)
1380        );
1381        let decoded = decode_binary_miniblocks(compressed, &encoding).unwrap();
1382        assert_decoded_value_lengths(&decoded, &lengths);
1383    }
1384
1385    #[test]
1386    fn generic_offsets_support_constant_and_u64_range() {
1387        let encoder = BinaryMiniBlockEncoder::with_generic_offsets(
1388            Some(256),
1389            CompressionFieldParams::default(),
1390        );
1391        let empty_lengths = vec![0_usize; 1_024];
1392        let (compressed, encoding) = encoder
1393            .compress(
1394                miniblock_context(),
1395                DataBlock::VariableWidth(variable_block_u32(&empty_lengths)),
1396            )
1397            .unwrap();
1398        let Some(Compression::Variable(variable)) = encoding.compression.as_ref() else {
1399            panic!("expected Variable encoding");
1400        };
1401        assert!(matches!(
1402            variable
1403                .offsets
1404                .as_deref()
1405                .and_then(|offsets| offsets.compression.as_ref()),
1406            Some(Compression::Constant(_))
1407        ));
1408        let decoded = decode_binary_miniblocks(compressed, &encoding).unwrap();
1409        assert_decoded_value_lengths(&decoded, &empty_lengths);
1410
1411        let num_values = 512_usize;
1412        let block = VariableWidthBlock {
1413            data: LanceBuffer::from(vec![1_u8; num_values * 2]),
1414            offsets: LanceBuffer::reinterpret_vec(
1415                (0..=num_values)
1416                    .map(|index| (index * 2) as i64)
1417                    .collect::<Vec<_>>(),
1418            ),
1419            bits_per_offset: 64,
1420            num_values: num_values as u64,
1421            block_info: BlockInfo::default(),
1422        };
1423        let (compressed, encoding) = encoder
1424            .compress(miniblock_context(), DataBlock::VariableWidth(block))
1425            .unwrap();
1426        let Some(Compression::Variable(variable)) = encoding.compression.as_ref() else {
1427            panic!("expected Variable encoding");
1428        };
1429        assert!(matches!(
1430            variable
1431                .offsets
1432                .as_deref()
1433                .and_then(|offsets| offsets.compression.as_ref()),
1434            Some(Compression::Range(_))
1435        ));
1436        let decoded = decode_binary_miniblocks(compressed, &encoding).unwrap();
1437        assert!(decoded.iter().all(|block| block.bits_per_offset == 64));
1438    }
1439
1440    #[test]
1441    fn binary_miniblock_preserves_legacy_wire_and_fallible_generic_reader() {
1442        assert!(BinaryMiniBlockDecompressor::new(16).is_err());
1443        let encoder = BinaryMiniBlockEncoder::new(Some(4_096));
1444        let (compressed, encoding) = encoder
1445            .compress(
1446                miniblock_context(),
1447                DataBlock::VariableWidth(variable_block_u32(&[3, 3])),
1448            )
1449            .unwrap();
1450        let Some(Compression::Variable(variable)) = encoding.compression.as_ref() else {
1451            panic!("expected Variable encoding");
1452        };
1453        assert!(matches!(
1454            variable
1455                .offsets
1456                .as_deref()
1457                .and_then(|offsets| offsets.compression.as_ref()),
1458            Some(Compression::Flat(_))
1459        ));
1460        let mut expected = Vec::new();
1461        expected.extend_from_slice(&12_i32.to_le_bytes());
1462        expected.extend_from_slice(&15_i32.to_le_bytes());
1463        expected.extend_from_slice(&18_i32.to_le_bytes());
1464        expected.extend_from_slice(&[0, 0, 0, 1, 1, 1]);
1465        expected.extend_from_slice(&[72, 72]);
1466        assert_eq!(compressed.chunks[0].buffer_sizes, [20]);
1467        assert_eq!(compressed.data[0].as_ref(), expected);
1468
1469        let generic_encoder = BinaryMiniBlockEncoder::with_generic_offsets(
1470            Some(4_096),
1471            CompressionFieldParams::default(),
1472        );
1473        let (_, encoding) = generic_encoder
1474            .compress(
1475                miniblock_context(),
1476                DataBlock::VariableWidth(variable_block_u32(&[1, 7, 2, 5])),
1477            )
1478            .unwrap();
1479        let Some(Compression::Variable(variable)) = encoding.compression.as_ref() else {
1480            panic!("expected Variable encoding");
1481        };
1482        assert!(matches!(
1483            variable
1484                .offsets
1485                .as_deref()
1486                .and_then(|offsets| offsets.compression.as_ref()),
1487            Some(Compression::Flat(_))
1488        ));
1489
1490        let variable = metadata::Variable {
1491            offsets: Some(Box::new(EncodingUtils::range(32, 0, 3))),
1492            values: None,
1493        };
1494        let decoder = BinaryMiniBlockDecompressor::from_variable(&variable).unwrap();
1495        let error = decoder
1496            .decompress(vec![LanceBuffer::empty(), LanceBuffer::empty()], 2)
1497            .unwrap_err();
1498        assert!(error.to_string().contains("2 buffers, expected 1"));
1499        let error = decoder
1500            .decompress(vec![LanceBuffer::from(vec![0_u8; 5])], 2)
1501            .unwrap_err();
1502        assert!(error.to_string().contains("final generic offset 6"));
1503    }
1504
1505    #[rstest]
1506    #[case::range([16_usize; 4], "range")]
1507    #[case::delta([4_usize, 10, 5, 8], "delta")]
1508    #[test_log::test(tokio::test)]
1509    async fn generic_offsets_support_scan_range_take(
1510        #[case] lengths: [usize; 4],
1511        #[case] expected_encoding: &str,
1512    ) {
1513        let values = StringArray::from_iter_values((0..10_000).map(|index| {
1514            let len = lengths[index % lengths.len()];
1515            format!("{index:04x}{}", "x".repeat(len - 4))
1516        }));
1517        let metadata = HashMap::from([
1518            (
1519                STRUCTURAL_ENCODING_META_KEY.to_string(),
1520                STRUCTURAL_ENCODING_MINIBLOCK.to_string(),
1521            ),
1522            (COMPRESSION_META_KEY.to_string(), "none".to_string()),
1523            (DICT_DIVISOR_META_KEY.to_string(), "100000".to_string()),
1524        ]);
1525        let test_cases = TestCases::basic()
1526            .with_encoding(TestEncoding::StructuralSparse)
1527            .with_expected_encoding(expected_encoding);
1528        check_round_trip_encoding_of_data(vec![Arc::new(values)], &test_cases, metadata).await;
1529    }
1530
1531    #[rstest]
1532    #[test_log::test(tokio::test)]
1533    async fn test_utf8_binary(
1534        #[values(
1535            TestEncoding::StructuralU16,
1536            TestEncoding::StructuralU32,
1537            TestEncoding::StructuralSparse
1538        )]
1539        encoding: TestEncoding,
1540        #[values(4096, 1024 * 1024)] page_size: u64,
1541        #[values(false, true)] use_slicing: bool,
1542    ) {
1543        let field = Field::new("", DataType::Utf8, false);
1544        check_specific_random(
1545            field,
1546            TestCases::basic()
1547                .with_encoding(encoding)
1548                .with_page_sizes(vec![page_size])
1549                .with_slicing_modes([use_slicing]),
1550        )
1551        .await;
1552    }
1553
1554    #[rstest]
1555    #[test_log::test(tokio::test)]
1556    async fn test_binary(
1557        #[values(STRUCTURAL_ENCODING_MINIBLOCK, STRUCTURAL_ENCODING_FULLZIP)]
1558        structural_encoding: &str,
1559        #[values(DataType::Utf8, DataType::Binary)] data_type: DataType,
1560        #[values(
1561            TestEncoding::Array,
1562            TestEncoding::StructuralU16,
1563            TestEncoding::StructuralU32,
1564            TestEncoding::StructuralSparse
1565        )]
1566        encoding: TestEncoding,
1567        #[values(4096, 1024 * 1024)] page_size: u64,
1568        #[values(false, true)] use_slicing: bool,
1569    ) {
1570        let mut field_metadata = HashMap::new();
1571        field_metadata.insert(
1572            STRUCTURAL_ENCODING_META_KEY.to_string(),
1573            structural_encoding.into(),
1574        );
1575
1576        let field = Field::new("", data_type, false).with_metadata(field_metadata);
1577        check_basic_random_case(field, encoding, page_size, use_slicing).await;
1578    }
1579
1580    #[rstest]
1581    #[test_log::test(tokio::test)]
1582    async fn test_binary_fsst(
1583        #[values(STRUCTURAL_ENCODING_MINIBLOCK, STRUCTURAL_ENCODING_FULLZIP)]
1584        structural_encoding: &str,
1585        #[values(DataType::Binary, DataType::Utf8)] data_type: DataType,
1586        #[values(
1587            TestEncoding::StructuralU16,
1588            TestEncoding::StructuralU32,
1589            TestEncoding::StructuralSparse
1590        )]
1591        encoding: TestEncoding,
1592        #[values(4096, 1024 * 1024)] page_size: u64,
1593        #[values(false, true)] use_slicing: bool,
1594    ) {
1595        let mut field_metadata = HashMap::new();
1596        field_metadata.insert(
1597            STRUCTURAL_ENCODING_META_KEY.to_string(),
1598            structural_encoding.into(),
1599        );
1600        field_metadata.insert(COMPRESSION_META_KEY.to_string(), "fsst".into());
1601        let field = Field::new("", data_type, true).with_metadata(field_metadata);
1602        // TODO (https://github.com/lance-format/lance/issues/4783)
1603        let test_cases = TestCases::default()
1604            .with_encoding(encoding)
1605            .with_page_sizes(vec![page_size])
1606            .with_slicing_modes([use_slicing]);
1607        check_specific_random(field, test_cases).await;
1608    }
1609
1610    #[rstest]
1611    #[test_log::test(tokio::test)]
1612    async fn test_fsst_large_binary(
1613        #[values(STRUCTURAL_ENCODING_MINIBLOCK, STRUCTURAL_ENCODING_FULLZIP)]
1614        structural_encoding: &str,
1615        #[values(DataType::LargeBinary, DataType::LargeUtf8)] data_type: DataType,
1616        #[values(
1617            TestEncoding::StructuralU16,
1618            TestEncoding::StructuralU32,
1619            TestEncoding::StructuralSparse
1620        )]
1621        encoding: TestEncoding,
1622        #[values(4096, 1024 * 1024)] page_size: u64,
1623        #[values(false, true)] use_slicing: bool,
1624    ) {
1625        let mut field_metadata = HashMap::new();
1626        field_metadata.insert(
1627            STRUCTURAL_ENCODING_META_KEY.to_string(),
1628            structural_encoding.into(),
1629        );
1630        field_metadata.insert(COMPRESSION_META_KEY.to_string(), "fsst".into());
1631        let field = Field::new("", data_type, true).with_metadata(field_metadata);
1632        check_specific_random(
1633            field,
1634            TestCases::basic()
1635                .with_encoding(encoding)
1636                .with_page_sizes(vec![page_size])
1637                .with_slicing_modes([use_slicing]),
1638        )
1639        .await;
1640    }
1641
1642    #[rstest]
1643    #[test_log::test(tokio::test)]
1644    async fn test_large_binary_types(
1645        #[values(DataType::LargeBinary, DataType::LargeUtf8)] data_type: DataType,
1646        #[values(
1647            TestEncoding::Array,
1648            TestEncoding::StructuralU16,
1649            TestEncoding::StructuralU32,
1650            TestEncoding::StructuralSparse
1651        )]
1652        encoding: TestEncoding,
1653        #[values(4096, 1024 * 1024)] page_size: u64,
1654        #[values(false, true)] use_slicing: bool,
1655    ) {
1656        let field = Field::new("", data_type, true);
1657        check_basic_random_case(field, encoding, page_size, use_slicing).await;
1658    }
1659
1660    #[rstest]
1661    #[test_log::test(tokio::test)]
1662    async fn test_small_strings(
1663        #[values(STRUCTURAL_ENCODING_MINIBLOCK, STRUCTURAL_ENCODING_FULLZIP)]
1664        structural_encoding: &str,
1665        #[values(
1666            TestEncoding::Array,
1667            TestEncoding::StructuralU16,
1668            TestEncoding::StructuralU32,
1669            TestEncoding::StructuralSparse
1670        )]
1671        encoding: TestEncoding,
1672        #[values(4096, 1024 * 1024)] page_size: u64,
1673        #[values(false, true)] use_slicing: bool,
1674    ) {
1675        let mut field_metadata = HashMap::new();
1676        field_metadata.insert(
1677            STRUCTURAL_ENCODING_META_KEY.to_string(),
1678            structural_encoding.into(),
1679        );
1680        let field = Field::new("", DataType::Utf8, true).with_metadata(field_metadata);
1681        check_round_trip_encoding_generated(
1682            field,
1683            Box::new(FnArrayGeneratorProvider::new(move || {
1684                lance_datagen::array::utf8_prefix_plus_counter("user_", /*is_large=*/ false)
1685            })),
1686            TestCases::basic()
1687                .with_encoding(encoding)
1688                .with_page_sizes(vec![page_size])
1689                .with_slicing_modes([use_slicing]),
1690        )
1691        .await;
1692    }
1693
1694    #[rstest]
1695    #[test_log::test(tokio::test)]
1696    async fn test_simple_binary(
1697        #[values(STRUCTURAL_ENCODING_MINIBLOCK, STRUCTURAL_ENCODING_FULLZIP)]
1698        structural_encoding: &str,
1699        #[values(DataType::Utf8, DataType::Binary)] data_type: DataType,
1700    ) {
1701        let string_array = StringArray::from(vec![Some("abc"), None, Some("pqr"), None, Some("m")]);
1702        let string_array = arrow_cast::cast(&string_array, &data_type).unwrap();
1703
1704        let mut field_metadata = HashMap::new();
1705        field_metadata.insert(
1706            STRUCTURAL_ENCODING_META_KEY.to_string(),
1707            structural_encoding.into(),
1708        );
1709
1710        let test_cases = TestCases::default()
1711            .with_range(0..2)
1712            .with_range(0..3)
1713            .with_range(1..3)
1714            .with_indices(vec![0, 1, 3, 4]);
1715        check_round_trip_encoding_of_data(
1716            vec![Arc::new(string_array)],
1717            &test_cases,
1718            field_metadata,
1719        )
1720        .await;
1721    }
1722
1723    #[test_log::test(tokio::test)]
1724    async fn test_sliced_utf8() {
1725        let string_array = StringArray::from(vec![Some("abc"), Some("de"), None, Some("fgh")]);
1726        let string_array = string_array.slice(1, 3);
1727
1728        let test_cases = TestCases::default()
1729            .with_range(0..1)
1730            .with_range(0..2)
1731            .with_range(1..2);
1732        check_round_trip_encoding_of_data(
1733            vec![Arc::new(string_array)],
1734            &test_cases,
1735            HashMap::new(),
1736        )
1737        .await;
1738    }
1739
1740    #[rstest]
1741    #[test_log::test(tokio::test)]
1742    async fn test_value_bigger_than_max_page_size(
1743        #[values(
1744            TestEncoding::Array,
1745            TestEncoding::StructuralU16,
1746            TestEncoding::StructuralU32,
1747            TestEncoding::StructuralSparse
1748        )]
1749        encoding: TestEncoding,
1750    ) {
1751        // Create one value larger than the configured 1MiB page budget.
1752        let big_string = String::from_iter((0..(2 * 1024 * 1024)).map(|_| '0'));
1753        let string_array = StringArray::from(vec![
1754            Some(big_string),
1755            Some("abc".to_string()),
1756            None,
1757            None,
1758            Some("xyz".to_string()),
1759        ]);
1760
1761        // Drop the max page size to 1MiB
1762        let test_cases = TestCases::default()
1763            .with_max_page_size(1024 * 1024)
1764            .with_encoding(encoding);
1765
1766        check_round_trip_encoding_of_data(
1767            vec![Arc::new(string_array)],
1768            &test_cases,
1769            HashMap::new(),
1770        )
1771        .await;
1772    }
1773
1774    #[rstest]
1775    #[test_log::test(tokio::test)]
1776    async fn test_page_split_parts_do_not_evenly_divide_rows(
1777        #[values(
1778            TestEncoding::Array,
1779            TestEncoding::StructuralU16,
1780            TestEncoding::StructuralU32,
1781            TestEncoding::StructuralSparse
1782        )]
1783        encoding: TestEncoding,
1784    ) {
1785        // Regression: split 90 rows into four parts, where the part count does
1786        // not evenly divide the row count.
1787        let big_string = String::from_iter((0..45_000).map(|_| '0'));
1788        let string_array = StringArray::from_iter_values((0..90).map(|_| big_string.clone()));
1789
1790        check_round_trip_encoding_of_data(
1791            vec![Arc::new(string_array)],
1792            &TestCases::default()
1793                .with_max_page_size(1024 * 1024)
1794                .with_encoding(encoding),
1795            HashMap::new(),
1796        )
1797        .await;
1798    }
1799
1800    #[test_log::test(tokio::test)]
1801    async fn test_empty_strings() {
1802        // Scenario 1: Some strings are empty
1803
1804        let values = [Some("abc"), Some(""), None];
1805        // Test empty list at beginning, middle, and end
1806        for order in [[0, 1, 2], [1, 0, 2], [2, 0, 1]] {
1807            let mut string_builder = StringBuilder::new();
1808            for idx in order {
1809                string_builder.append_option(values[idx]);
1810            }
1811            let string_array = Arc::new(string_builder.finish());
1812            let test_cases = TestCases::default()
1813                .with_indices(vec![1])
1814                .with_indices(vec![0])
1815                .with_indices(vec![2])
1816                .with_indices(vec![0, 1]);
1817            check_round_trip_encoding_of_data(
1818                vec![string_array.clone()],
1819                &test_cases,
1820                HashMap::new(),
1821            )
1822            .await;
1823            let test_cases = test_cases.with_batch_size(1);
1824            check_round_trip_encoding_of_data(vec![string_array], &test_cases, HashMap::new())
1825                .await;
1826        }
1827
1828        // Scenario 2: All strings are empty
1829
1830        // When encoding an array of empty strings there are no bytes to encode
1831        // which is strange and we want to ensure we handle it
1832        let string_array = Arc::new(StringArray::from(vec![Some(""), None, Some("")]));
1833
1834        let test_cases = TestCases::default().with_range(0..2).with_indices(vec![1]);
1835        check_round_trip_encoding_of_data(vec![string_array.clone()], &test_cases, HashMap::new())
1836            .await;
1837        let test_cases = test_cases.with_batch_size(1);
1838        check_round_trip_encoding_of_data(vec![string_array], &test_cases, HashMap::new()).await;
1839    }
1840
1841    #[test_log::test(tokio::test)]
1842    #[ignore] // This test is quite slow in debug mode
1843    async fn test_jumbo_string() {
1844        // This is an overflow test.  We have a list of lists where each list
1845        // has 1Mi items.  We encode 5000 of these lists and so we have over 4Gi in the
1846        // offsets range
1847        let mut string_builder = LargeStringBuilder::new();
1848        // a 1 MiB string
1849        let giant_string = String::from_iter((0..(1024 * 1024)).map(|_| '0'));
1850        for _ in 0..5000 {
1851            string_builder.append_option(Some(&giant_string));
1852        }
1853        let giant_array = Arc::new(string_builder.finish()) as ArrayRef;
1854        let arrs = vec![giant_array];
1855
1856        // // We can't validate because our validation relies on concatenating all input arrays
1857        let test_cases = TestCases::default().without_validation();
1858        check_round_trip_encoding_of_data(arrs, &test_cases, HashMap::new()).await;
1859    }
1860
1861    #[rstest]
1862    #[test_log::test(tokio::test)]
1863    async fn test_binary_dictionary_encoding(
1864        #[values(true, false)] with_nulls: bool,
1865        #[values(100, 500, 35000)] dict_size: u32,
1866    ) {
1867        let test_cases = TestCases::default().with_structural_encodings();
1868        let strings = (0..dict_size)
1869            .map(|i| i.to_string())
1870            .collect::<Vec<String>>();
1871
1872        let repeated_strings: Vec<_> = strings
1873            .iter()
1874            .cycle()
1875            .take(70000)
1876            .enumerate()
1877            .map(|(i, s)| {
1878                if with_nulls && i % 7 == 0 {
1879                    None
1880                } else {
1881                    Some(s.clone())
1882                }
1883            })
1884            .collect();
1885        let string_array = Arc::new(StringArray::from(repeated_strings)) as ArrayRef;
1886        check_round_trip_encoding_of_data(vec![string_array], &test_cases, HashMap::new()).await;
1887    }
1888
1889    #[test_log::test(tokio::test)]
1890    async fn test_binary_encoding_verification() {
1891        use lance_datagen::{ByteCount, RowCount};
1892
1893        let test_cases = TestCases::default()
1894            .with_expected_encoding("variable")
1895            .with_structural_encodings();
1896
1897        // Test both automatic selection and explicit configuration
1898        // 1. Test automatic binary encoding selection (small strings that won't trigger FSST)
1899        let arr_small = lance_datagen::gen_batch()
1900            .anon_col(lance_datagen::array::rand_utf8(ByteCount::from(10), false))
1901            .into_batch_rows(RowCount::from(1000))
1902            .unwrap()
1903            .column(0)
1904            .clone();
1905        check_round_trip_encoding_of_data(vec![arr_small], &test_cases, HashMap::new()).await;
1906
1907        // 2. Test explicit "none" compression to force binary encoding
1908        let metadata_explicit =
1909            HashMap::from([("lance-encoding:compression".to_string(), "none".to_string())]);
1910        let arr_large = lance_datagen::gen_batch()
1911            .anon_col(lance_datagen::array::rand_utf8(ByteCount::from(50), false))
1912            .into_batch_rows(RowCount::from(2000))
1913            .unwrap()
1914            .column(0)
1915            .clone();
1916        check_round_trip_encoding_of_data(vec![arr_large], &test_cases, metadata_explicit).await;
1917    }
1918
1919    #[test]
1920    fn test_binary_miniblock_with_misaligned_buffer() {
1921        use super::BinaryMiniBlockDecompressor;
1922        use crate::buffer::LanceBuffer;
1923        use crate::compression::MiniBlockDecompressor;
1924        use crate::data::DataBlock;
1925
1926        // Test case 1: u32 offsets
1927        {
1928            let decompressor = BinaryMiniBlockDecompressor::new(32).unwrap();
1929
1930            // Create test data with u32 offsets
1931            // BinaryMiniBlock format: all offsets followed by all string data
1932            // Need to ensure total size is divisible by 4 for u32
1933            let mut test_data = Vec::new();
1934
1935            // Offsets section (3 offsets for 2 values + 1 end offset)
1936            test_data.extend_from_slice(&12u32.to_le_bytes()); // offset to start of strings (after offsets)
1937            test_data.extend_from_slice(&15u32.to_le_bytes()); // offset to second string
1938            test_data.extend_from_slice(&20u32.to_le_bytes()); // offset to end
1939
1940            // String data section
1941            test_data.extend_from_slice(b"ABCXYZ"); // 6 bytes of string data
1942            test_data.extend_from_slice(&[0, 0]); // 2 bytes padding to make total 20 bytes (divisible by 4)
1943
1944            // Create a misaligned buffer by adding padding and slicing
1945            let mut padded = Vec::with_capacity(test_data.len() + 1);
1946            padded.push(0xFF); // Padding byte to misalign
1947            padded.extend_from_slice(&test_data);
1948
1949            let bytes = bytes::Bytes::from(padded);
1950            let misaligned = bytes.slice(1..); // Skip first byte to create misalignment
1951
1952            // Create LanceBuffer with bytes_per_value=1 to bypass alignment check
1953            let buffer = LanceBuffer::from_bytes(misaligned, 1);
1954
1955            // Verify the buffer is actually misaligned
1956            let ptr = buffer.as_ref().as_ptr();
1957            assert_ne!(
1958                ptr.align_offset(4),
1959                0,
1960                "Test setup: buffer should be misaligned for u32"
1961            );
1962
1963            // Decompress with misaligned buffer - should work with borrow_to_typed_slice
1964            let result = decompressor.decompress(vec![buffer], 2);
1965            assert!(
1966                result.is_ok(),
1967                "Decompression should succeed with misaligned buffer"
1968            );
1969
1970            // Verify the data is correct
1971            if let Ok(DataBlock::VariableWidth(block)) = result {
1972                assert_eq!(block.num_values, 2);
1973                // Data should be the strings (including padding from the original buffer)
1974                assert_eq!(&block.data.as_ref()[..6], b"ABCXYZ");
1975            } else {
1976                panic!("Expected VariableWidth block");
1977            }
1978        }
1979
1980        // Test case 2: u64 offsets
1981        {
1982            let decompressor = BinaryMiniBlockDecompressor::new(64).unwrap();
1983
1984            // Create test data with u64 offsets
1985            let mut test_data = Vec::new();
1986
1987            // Offsets section (3 offsets for 2 values + 1 end offset)
1988            test_data.extend_from_slice(&24u64.to_le_bytes()); // offset to start of strings (after offsets)
1989            test_data.extend_from_slice(&29u64.to_le_bytes()); // offset to second string
1990            test_data.extend_from_slice(&40u64.to_le_bytes()); // offset to end (divisible by 8)
1991
1992            // String data section
1993            test_data.extend_from_slice(b"HelloWorld"); // 10 bytes of string data
1994            test_data.extend_from_slice(&[0, 0, 0, 0, 0, 0]); // 6 bytes padding to make total 40 bytes (divisible by 8)
1995
1996            // Create misaligned buffer
1997            let mut padded = Vec::with_capacity(test_data.len() + 3);
1998            padded.extend_from_slice(&[0xFF, 0xFF, 0xFF]); // 3 bytes padding for misalignment
1999            padded.extend_from_slice(&test_data);
2000
2001            let bytes = bytes::Bytes::from(padded);
2002            let misaligned = bytes.slice(3..); // Skip 3 bytes
2003
2004            let buffer = LanceBuffer::from_bytes(misaligned, 1);
2005
2006            // Verify misalignment for u64
2007            let ptr = buffer.as_ref().as_ptr();
2008            assert_ne!(
2009                ptr.align_offset(8),
2010                0,
2011                "Test setup: buffer should be misaligned for u64"
2012            );
2013
2014            // Decompress should succeed
2015            let result = decompressor.decompress(vec![buffer], 2);
2016            assert!(
2017                result.is_ok(),
2018                "Decompression should succeed with misaligned u64 buffer"
2019            );
2020
2021            if let Ok(DataBlock::VariableWidth(block)) = result {
2022                assert_eq!(block.num_values, 2);
2023                // Data should be the strings (including padding from the original buffer)
2024                assert_eq!(&block.data.as_ref()[..10], b"HelloWorld");
2025            } else {
2026                panic!("Expected VariableWidth block");
2027            }
2028        }
2029    }
2030
2031    #[test]
2032    fn test_binary_miniblock_rejects_corrupt_offsets() {
2033        use super::BinaryMiniBlockDecompressor;
2034        use crate::compression::MiniBlockDecompressor;
2035        use lance_core::Error;
2036
2037        // Chunk layout mirrors the on-disk format for ["alpha", "beta", "gamma"]:
2038        // LE u32 offsets [16, 21, 25, 30] followed by the value bytes, padded to
2039        // a multiple of 8 bytes.
2040        fn chunk_u32(offsets: &[u32], values: &[u8]) -> LanceBuffer {
2041            let mut chunk = offsets
2042                .iter()
2043                .flat_map(|offset| offset.to_le_bytes())
2044                .collect::<Vec<u8>>();
2045            chunk.extend_from_slice(values);
2046            chunk.resize(chunk.len().next_multiple_of(8), 0);
2047            LanceBuffer::from(chunk)
2048        }
2049
2050        let decompressor = BinaryMiniBlockDecompressor::new(32).unwrap();
2051
2052        // The tail offset points past the end of the 32-byte chunk.
2053        let err = decompressor
2054            .decompress(
2055                vec![chunk_u32(&[16, 21, 25, 100_000], b"alphabetagamma")],
2056                3,
2057            )
2058            .unwrap_err();
2059        assert!(matches!(err, Error::CorruptFile { .. }), "{err:?}");
2060        assert!(err.to_string().contains("out of bounds"), "{err}");
2061
2062        // Offsets go backwards, which would underflow the rebase subtraction.
2063        let err = decompressor
2064            .decompress(vec![chunk_u32(&[16, 25, 21, 30], b"alphabetagamma")], 3)
2065            .unwrap_err();
2066        assert!(matches!(err, Error::CorruptFile { .. }), "{err:?}");
2067        assert!(err.to_string().contains("decreases"), "{err}");
2068
2069        // The first offset points inside the offset table, which would alias
2070        // the serialized offsets into the value bytes.
2071        let err = decompressor
2072            .decompress(vec![chunk_u32(&[0, 21, 25, 30], b"alphabetagamma")], 3)
2073            .unwrap_err();
2074        assert!(matches!(err, Error::CorruptFile { .. }), "{err:?}");
2075        assert!(err.to_string().contains("overlaps"), "{err}");
2076
2077        // The chunk stores fewer offsets than the requested value count needs.
2078        let err = decompressor
2079            .decompress(vec![chunk_u32(&[8, 8], &[])], 3)
2080            .unwrap_err();
2081        assert!(matches!(err, Error::CorruptFile { .. }), "{err:?}");
2082        assert!(err.to_string().contains("requires 4"), "{err}");
2083
2084        // The chunk size is not a multiple of the offset width.
2085        let err = decompressor
2086            .decompress(vec![LanceBuffer::from(vec![0u8; 10])], 1)
2087            .unwrap_err();
2088        assert!(matches!(err, Error::CorruptFile { .. }), "{err:?}");
2089        assert!(err.to_string().contains("multiple"), "{err}");
2090
2091        // 64-bit offsets take the same validation path.
2092        fn chunk_u64(offsets: &[u64], values: &[u8]) -> LanceBuffer {
2093            let mut chunk = offsets
2094                .iter()
2095                .flat_map(|offset| offset.to_le_bytes())
2096                .collect::<Vec<u8>>();
2097            chunk.extend_from_slice(values);
2098            chunk.resize(chunk.len().next_multiple_of(8), 0);
2099            LanceBuffer::from(chunk)
2100        }
2101        let decompressor = BinaryMiniBlockDecompressor::new(64).unwrap();
2102        let err = decompressor
2103            .decompress(
2104                vec![chunk_u64(&[32, 37, 41, 100_000], b"alphabetagamma")],
2105                3,
2106            )
2107            .unwrap_err();
2108        assert!(matches!(err, Error::CorruptFile { .. }), "{err:?}");
2109        assert!(err.to_string().contains("out of bounds"), "{err}");
2110        let err = decompressor
2111            .decompress(vec![chunk_u64(&[0, 37, 41, 46], b"alphabetagamma")], 3)
2112            .unwrap_err();
2113        assert!(matches!(err, Error::CorruptFile { .. }), "{err:?}");
2114        assert!(err.to_string().contains("overlaps"), "{err}");
2115
2116        // A valid chunk still decodes: offsets rebase to [0, 5, 9, 14].
2117        let decompressor = BinaryMiniBlockDecompressor::new(32).unwrap();
2118        let block = decompressor
2119            .decompress(vec![chunk_u32(&[16, 21, 25, 30], b"alphabetagamma")], 3)
2120            .unwrap();
2121        let DataBlock::VariableWidth(block) = block else {
2122            panic!("expected a variable-width block");
2123        };
2124        assert_eq!(block.data.as_ref(), b"alphabetagamma");
2125        assert_eq!(
2126            block.offsets,
2127            LanceBuffer::reinterpret_vec(vec![0_u32, 5, 9, 14])
2128        );
2129    }
2130
2131    fn encoded_binary_block(bits_per_offset: u8) -> Vec<u8> {
2132        use crate::compression::BlockCompressor;
2133
2134        let offsets = match bits_per_offset {
2135            32 => LanceBuffer::reinterpret_vec(vec![0_i32, 5, 9, 14]),
2136            64 => LanceBuffer::reinterpret_vec(vec![0_i64, 5, 9, 14]),
2137            _ => unreachable!(),
2138        };
2139        let block = DataBlock::VariableWidth(VariableWidthBlock {
2140            data: LanceBuffer::copy_slice(b"alphabetagamma"),
2141            offsets,
2142            bits_per_offset,
2143            num_values: 3,
2144            block_info: BlockInfo::new(),
2145        });
2146        BlockCompressor::compress(&super::VariableEncoder::default(), block)
2147            .unwrap()
2148            .0
2149            .as_ref()
2150            .unwrap()
2151            .to_vec()
2152    }
2153
2154    /// The block decompressor only checks the block structure (all O(1)); bad
2155    /// offset values inside a structurally-sound block are rejected by the
2156    /// mandatory layout validation when the block is converted to Arrow.
2157    #[rstest]
2158    #[case::i32_tail_out_of_bounds(32, 3, 100_000, "out of bounds")]
2159    #[case::i64_tail_out_of_bounds(64, 3, 15, "out of bounds")]
2160    #[case::i32_non_monotonic(32, 2, 4, "non-monotonic")]
2161    #[case::i64_non_monotonic(64, 2, 4, "non-monotonic")]
2162    fn test_binary_block_bad_offsets_rejected_at_arrow_conversion(
2163        #[case] bits_per_offset: u8,
2164        #[case] mutated_offset_index: usize,
2165        #[case] mutated_offset_value: u64,
2166        #[case] expected_message: &str,
2167    ) {
2168        use crate::compression::BlockDecompressor;
2169        use lance_core::Error;
2170
2171        let mut encoded = encoded_binary_block(bits_per_offset);
2172        let bytes_per_offset = (bits_per_offset / 8) as usize;
2173        // The standard scheme header is two offset-width fields.
2174        let mutated_offset_start = bytes_per_offset * (2 + mutated_offset_index);
2175        encoded[mutated_offset_start..mutated_offset_start + bytes_per_offset]
2176            .copy_from_slice(&mutated_offset_value.to_le_bytes()[..bytes_per_offset]);
2177
2178        let block = super::BinaryBlockDecompressor::default()
2179            .decompress(Some(LanceBuffer::from(encoded)), 3)
2180            .unwrap();
2181        let data_type = match bits_per_offset {
2182            32 => DataType::Binary,
2183            _ => DataType::LargeBinary,
2184        };
2185        let err = block.into_arrow(data_type, false).unwrap_err();
2186        assert!(matches!(err, Error::CorruptFile { .. }), "{err:?}");
2187        assert!(err.to_string().contains(expected_message), "{err}");
2188    }
2189
2190    #[test]
2191    fn test_binary_block_rejects_corrupt_structure() {
2192        use crate::compression::BlockDecompressor;
2193        use lance_core::Error;
2194
2195        let decompressor = super::BinaryBlockDecompressor::default();
2196
2197        // The first offset must be zero.
2198        let mut encoded = encoded_binary_block(32);
2199        encoded[8..12].copy_from_slice(&5_u32.to_le_bytes());
2200        let err = decompressor
2201            .decompress(Some(LanceBuffer::from(encoded)), 3)
2202            .unwrap_err();
2203        assert!(matches!(err, Error::CorruptFile { .. }), "{err:?}");
2204        assert!(err.to_string().contains("first offset"), "{err}");
2205
2206        // The offsets region must hold exactly num_values + 1 offsets.
2207        let encoded = encoded_binary_block(32);
2208        let err = decompressor
2209            .decompress(Some(LanceBuffer::from(encoded)), 4)
2210            .unwrap_err();
2211        assert!(matches!(err, Error::CorruptFile { .. }), "{err:?}");
2212        assert!(err.to_string().contains("offset bytes"), "{err}");
2213
2214        // A block too small to hold its header is rejected, not a panic.
2215        let err = decompressor
2216            .decompress(Some(LanceBuffer::from(vec![0_u8; 2])), 1)
2217            .unwrap_err();
2218        assert!(matches!(err, Error::CorruptFile { .. }), "{err:?}");
2219        assert!(err.to_string().contains("too small"), "{err}");
2220    }
2221}