1use 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
88fn 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 bytes_start_offset: usize,
101 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 const PAD_BYTE: u8 = 72;
175 let pad_len = pad_bytes_to(output.len(), alignment);
176
177 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
185fn search_next_offset_idx<N: OffsetSizeTrait>(
190 offsets: &[N],
191 last_offset_idx: usize,
192 minichunk_size: i64,
193) -> usize {
194 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 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 return offsets.len() - 1;
213 } else {
214 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 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
312fn 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 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 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
762fn 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 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 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#[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 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 output.extend_from_slice(&(32_u32).to_le_bytes());
978
979 output.extend_from_slice(&(bytes_start_offset).to_le_bytes());
981
982 output.extend_from_slice(&variable_width_data.offsets);
984
985 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 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 output.extend_from_slice(&(64_u64).to_le_bytes());
1002
1003 output.extend_from_slice(&(bytes_start_offset).to_le_bytes());
1005
1006 output.extend_from_slice(&variable_width_data.offsets);
1008
1009 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 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 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 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 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 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 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 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_", 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 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 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 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 let values = [Some("abc"), Some(""), None];
1805 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 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] async fn test_jumbo_string() {
1844 let mut string_builder = LargeStringBuilder::new();
1848 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 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 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 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 {
1928 let decompressor = BinaryMiniBlockDecompressor::new(32).unwrap();
1929
1930 let mut test_data = Vec::new();
1934
1935 test_data.extend_from_slice(&12u32.to_le_bytes()); test_data.extend_from_slice(&15u32.to_le_bytes()); test_data.extend_from_slice(&20u32.to_le_bytes()); test_data.extend_from_slice(b"ABCXYZ"); test_data.extend_from_slice(&[0, 0]); let mut padded = Vec::with_capacity(test_data.len() + 1);
1946 padded.push(0xFF); padded.extend_from_slice(&test_data);
1948
1949 let bytes = bytes::Bytes::from(padded);
1950 let misaligned = bytes.slice(1..); let buffer = LanceBuffer::from_bytes(misaligned, 1);
1954
1955 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 let result = decompressor.decompress(vec![buffer], 2);
1965 assert!(
1966 result.is_ok(),
1967 "Decompression should succeed with misaligned buffer"
1968 );
1969
1970 if let Ok(DataBlock::VariableWidth(block)) = result {
1972 assert_eq!(block.num_values, 2);
1973 assert_eq!(&block.data.as_ref()[..6], b"ABCXYZ");
1975 } else {
1976 panic!("Expected VariableWidth block");
1977 }
1978 }
1979
1980 {
1982 let decompressor = BinaryMiniBlockDecompressor::new(64).unwrap();
1983
1984 let mut test_data = Vec::new();
1986
1987 test_data.extend_from_slice(&24u64.to_le_bytes()); test_data.extend_from_slice(&29u64.to_le_bytes()); test_data.extend_from_slice(&40u64.to_le_bytes()); test_data.extend_from_slice(b"HelloWorld"); test_data.extend_from_slice(&[0, 0, 0, 0, 0, 0]); let mut padded = Vec::with_capacity(test_data.len() + 3);
1998 padded.extend_from_slice(&[0xFF, 0xFF, 0xFF]); padded.extend_from_slice(&test_data);
2000
2001 let bytes = bytes::Bytes::from(padded);
2002 let misaligned = bytes.slice(3..); let buffer = LanceBuffer::from_bytes(misaligned, 1);
2005
2006 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 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 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 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 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 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 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 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 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 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 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 #[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 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 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 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 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}