1use arrow_array::types::UInt64Type;
19use arrow_array::{Array, PrimitiveArray};
20use arrow_buffer::ArrowNativeType;
21use lance_bitpacking::BitPacking;
22
23use lance_core::{Error, Result};
24
25use crate::buffer::LanceBuffer;
26use crate::compression::{BlockCompressor, BlockDecompressor, MiniBlockDecompressor};
27use crate::data::BlockInfo;
28use crate::data::{DataBlock, FixedWidthDataBlock};
29use crate::encodings::logical::primitive::miniblock::{
30 MiniBlockChunk, MiniBlockCompressed, MiniBlockCompressionContext, MiniBlockCompressor,
31};
32use crate::format::pb21::CompressiveEncoding;
33use crate::format::{ProtobufUtils21, pb21};
34use crate::statistics::{GetStat, Stat};
35use bytemuck::Pod;
36
37pub(crate) const LOG_ELEMS_PER_CHUNK: u8 = 10;
38pub const ELEMS_PER_CHUNK: u64 = 1 << LOG_ELEMS_PER_CHUNK;
40
41#[derive(Debug, Default)]
42pub struct InlineBitpacking {
43 uncompressed_bit_width: u64,
44}
45
46impl InlineBitpacking {
47 pub fn new(uncompressed_bit_width: u64) -> Self {
48 Self {
49 uncompressed_bit_width,
50 }
51 }
52
53 pub fn from_description(description: &pb21::InlineBitpacking) -> Self {
54 Self {
55 uncompressed_bit_width: description.uncompressed_bits_per_value,
56 }
57 }
58
59 pub fn min_size_bytes(compressed_bit_width: u64) -> u64 {
65 (ELEMS_PER_CHUNK * compressed_bit_width).div_ceil(8)
66 }
67
68 fn bitpack_chunked<T: ArrowNativeType + BitPacking>(
74 data: FixedWidthDataBlock,
75 ) -> MiniBlockCompressed {
76 debug_assert!(data.num_values > 0);
77 let data_buffer = data.data.borrow_to_typed_slice::<T>();
78 let data_buffer = data_buffer.as_ref();
79
80 let bit_widths = data.expect_stat(Stat::BitWidth);
81 let bit_widths_array = bit_widths
82 .as_any()
83 .downcast_ref::<PrimitiveArray<UInt64Type>>()
84 .unwrap();
85
86 let (packed_chunk_sizes, total_size) = bit_widths_array
87 .values()
88 .iter()
89 .map(|&bit_width| {
90 let chunk_size = ((1024 * bit_width) / data.bits_per_value) as usize;
91 (chunk_size, chunk_size + 1)
92 })
93 .fold(
94 (Vec::with_capacity(bit_widths_array.len()), 0),
95 |(mut sizes, total), (size, inc)| {
96 sizes.push(size);
97 (sizes, total + inc)
98 },
99 );
100
101 let mut output: Vec<T> = Vec::with_capacity(total_size);
102 let mut chunks = Vec::with_capacity(bit_widths_array.len());
103
104 for (i, packed_chunk_size) in packed_chunk_sizes
105 .iter()
106 .enumerate()
107 .take(bit_widths_array.len() - 1)
108 {
109 let start_elem = i * ELEMS_PER_CHUNK as usize;
110 let bit_width = bit_widths_array.value(i) as usize;
111 output.push(T::from_usize(bit_width).unwrap());
112 let output_len = output.len();
113 unsafe {
114 output.set_len(output_len + *packed_chunk_size);
115 BitPacking::unchecked_pack(
116 bit_width,
117 &data_buffer[start_elem..][..ELEMS_PER_CHUNK as usize],
118 &mut output[output_len..][..*packed_chunk_size],
119 );
120 }
121 chunks.push(MiniBlockChunk {
122 buffer_sizes: vec![((1 + *packed_chunk_size) * std::mem::size_of::<T>()) as u32],
123 log_num_values: LOG_ELEMS_PER_CHUNK,
124 });
125 }
126
127 let last_chunk_elem_num = if data.num_values.is_multiple_of(ELEMS_PER_CHUNK) {
129 ELEMS_PER_CHUNK
130 } else {
131 data.num_values % ELEMS_PER_CHUNK
132 };
133 let mut last_chunk: Vec<T> = vec![T::from_usize(0).unwrap(); ELEMS_PER_CHUNK as usize];
134 last_chunk[..last_chunk_elem_num as usize].clone_from_slice(
135 &data_buffer[data.num_values as usize - last_chunk_elem_num as usize..],
136 );
137 let bit_width = bit_widths_array.value(bit_widths_array.len() - 1) as usize;
138 output.push(T::from_usize(bit_width).unwrap());
139 let output_len = output.len();
140 unsafe {
141 output.set_len(output_len + packed_chunk_sizes[bit_widths_array.len() - 1]);
142 BitPacking::unchecked_pack(
143 bit_width,
144 &last_chunk,
145 &mut output[output_len..][..packed_chunk_sizes[bit_widths_array.len() - 1]],
146 );
147 }
148 chunks.push(MiniBlockChunk {
149 buffer_sizes: vec![
150 ((1 + packed_chunk_sizes[bit_widths_array.len() - 1]) * std::mem::size_of::<T>())
151 as u32,
152 ],
153 log_num_values: 0,
154 });
155
156 MiniBlockCompressed {
157 data: vec![LanceBuffer::reinterpret_vec(output)],
158 chunks,
159 num_values: data.num_values,
160 }
161 }
162
163 fn chunk_data(&self, data: FixedWidthDataBlock) -> (MiniBlockCompressed, CompressiveEncoding) {
164 assert!(data.bits_per_value.is_multiple_of(8));
165 assert_eq!(data.bits_per_value, self.uncompressed_bit_width);
166 let bits_per_value = data.bits_per_value;
167 let compressed = match bits_per_value {
168 8 => Self::bitpack_chunked::<u8>(data),
169 16 => Self::bitpack_chunked::<u16>(data),
170 32 => Self::bitpack_chunked::<u32>(data),
171 64 => Self::bitpack_chunked::<u64>(data),
172 _ => unreachable!(),
173 };
174 (
175 compressed,
176 ProtobufUtils21::inline_bitpacking(
177 bits_per_value,
178 None,
180 ),
181 )
182 }
183
184 fn unchunk<T: ArrowNativeType + BitPacking + Pod>(
185 data: LanceBuffer,
186 num_values: u64,
187 ) -> Result<DataBlock> {
188 let uncompressed_bit_width = std::mem::size_of::<T>() * 8;
190 let word_size = std::mem::size_of::<T>();
191
192 if data.len() < word_size {
193 return Err(Error::corrupt_file_named(
194 "inline_bitpacking",
195 format!(
196 "Inline bitpacking chunk is too small for {}-byte header: {} bytes",
197 word_size,
198 data.len()
199 ),
200 ));
201 }
202 if !data.len().is_multiple_of(word_size) {
203 return Err(Error::corrupt_file_named(
204 "inline_bitpacking",
205 format!(
206 "Inline bitpacking chunk size must be a multiple of {} bytes, got {} bytes",
207 word_size,
208 data.len()
209 ),
210 ));
211 }
212 if num_values > ELEMS_PER_CHUNK {
213 return Err(Error::corrupt_file_named(
214 "inline_bitpacking",
215 format!(
216 "Inline bitpacking chunk has {} values, expected at most {}",
217 num_values, ELEMS_PER_CHUNK
218 ),
219 ));
220 }
221
222 let chunk_words = data.borrow_to_typed_view::<T>();
223 let bit_width_value = chunk_words[0].as_usize();
224 if bit_width_value > uncompressed_bit_width {
225 return Err(Error::corrupt_file_named(
226 "inline_bitpacking",
227 format!(
228 "Inline bitpacking width {} exceeds {}-bit values",
229 bit_width_value, uncompressed_bit_width
230 ),
231 ));
232 }
233 let chunk = &chunk_words[1..];
234 let expected_num_bits = bit_width_value
238 .checked_mul(ELEMS_PER_CHUNK as usize)
239 .ok_or_else(|| {
240 Error::corrupt_file_named(
241 "inline_bitpacking",
242 format!(
243 "Inline bitpacking width {} overflows chunk bit count",
244 bit_width_value
245 ),
246 )
247 })?;
248 let expected_num_bytes = expected_num_bits / 8;
249 let actual_num_bytes = std::mem::size_of_val(chunk);
250 if actual_num_bytes != expected_num_bytes {
251 return Err(Error::corrupt_file_named(
252 "inline_bitpacking",
253 format!(
254 "Inline bitpacking payload has {} bytes, expected {} bytes for bit width {}",
255 actual_num_bytes, expected_num_bytes, bit_width_value
256 ),
257 ));
258 }
259
260 let mut decompressed = vec![T::default(); ELEMS_PER_CHUNK as usize];
261 unsafe {
262 BitPacking::unchecked_unpack(bit_width_value, chunk, &mut decompressed);
263 }
264
265 decompressed.truncate(num_values as usize);
266 Ok(DataBlock::FixedWidth(FixedWidthDataBlock {
267 data: LanceBuffer::reinterpret_vec(decompressed),
268 bits_per_value: uncompressed_bit_width as u64,
269 num_values,
270 block_info: BlockInfo::new(),
271 }))
272 }
273
274 fn empty_block(&self) -> DataBlock {
277 DataBlock::FixedWidth(FixedWidthDataBlock {
278 data: LanceBuffer::empty(),
279 bits_per_value: self.uncompressed_bit_width,
280 num_values: 0,
281 block_info: BlockInfo::new(),
282 })
283 }
284}
285
286impl MiniBlockCompressor for InlineBitpacking {
287 fn compress(
288 &self,
289 _context: MiniBlockCompressionContext,
290 chunk: DataBlock,
291 ) -> Result<(MiniBlockCompressed, CompressiveEncoding)> {
292 match chunk {
293 DataBlock::FixedWidth(fixed_width) => Ok(self.chunk_data(fixed_width)),
294 _ => Err(Error::invalid_input_source(
295 format!(
296 "Cannot compress a data block of type {} with BitpackMiniBlockEncoder",
297 chunk.name()
298 )
299 .into(),
300 )),
301 }
302 }
303}
304
305impl BlockCompressor for InlineBitpacking {
306 fn compress(&self, data: DataBlock) -> Result<LanceBuffer> {
307 let fixed_width = data.as_fixed_width().unwrap();
308 let (chunked, _) = self.chunk_data(fixed_width);
309 Ok(chunked.data.into_iter().next().unwrap())
310 }
311}
312
313impl MiniBlockDecompressor for InlineBitpacking {
314 fn decompress(&self, data: Vec<LanceBuffer>, num_values: u64) -> Result<DataBlock> {
315 assert_eq!(data.len(), 1);
316 let data = data.into_iter().next().unwrap();
317 if num_values == 0 {
318 return Ok(self.empty_block());
320 }
321 match self.uncompressed_bit_width {
322 8 => Self::unchunk::<u8>(data, num_values),
323 16 => Self::unchunk::<u16>(data, num_values),
324 32 => Self::unchunk::<u32>(data, num_values),
325 64 => Self::unchunk::<u64>(data, num_values),
326 _ => unimplemented!("Bitpacking word size must be 8, 16, 32, or 64"),
327 }
328 }
329}
330
331impl BlockDecompressor for InlineBitpacking {
332 fn decompress(&self, data: LanceBuffer, num_values: u64) -> Result<DataBlock> {
333 if num_values == 0 {
334 return Ok(self.empty_block());
338 }
339 match self.uncompressed_bit_width {
340 8 => Self::unchunk::<u8>(data, num_values),
341 16 => Self::unchunk::<u16>(data, num_values),
342 32 => Self::unchunk::<u32>(data, num_values),
343 64 => Self::unchunk::<u64>(data, num_values),
344 _ => unimplemented!("Bitpacking word size must be 8, 16, 32, or 64"),
345 }
346 }
347}
348
349fn bitpack_out_of_line<T: ArrowNativeType + BitPacking>(
355 data: FixedWidthDataBlock,
356 compressed_bits_per_value: usize,
357) -> LanceBuffer {
358 let data_buffer = data.data.borrow_to_typed_slice::<T>();
359 let data_buffer = data_buffer.as_ref();
360
361 let num_chunks = data_buffer.len().div_ceil(ELEMS_PER_CHUNK as usize);
362 let last_chunk_is_runt = data_buffer.len() % ELEMS_PER_CHUNK as usize != 0;
363 let words_per_chunk = (ELEMS_PER_CHUNK as usize * compressed_bits_per_value)
364 .div_ceil(data.bits_per_value as usize);
365 #[allow(clippy::uninit_vec)]
366 let mut output: Vec<T> = Vec::with_capacity(num_chunks * words_per_chunk);
367 #[allow(clippy::uninit_vec)]
368 unsafe {
369 output.set_len(num_chunks * words_per_chunk);
370 }
371
372 let num_whole_chunks = if last_chunk_is_runt {
373 num_chunks - 1
374 } else {
375 num_chunks
376 };
377
378 for i in 0..num_whole_chunks {
380 let input_start = i * ELEMS_PER_CHUNK as usize;
381 let input_end = input_start + ELEMS_PER_CHUNK as usize;
382 let output_start = i * words_per_chunk;
383 let output_end = output_start + words_per_chunk;
384 unsafe {
385 BitPacking::unchecked_pack(
386 compressed_bits_per_value,
387 &data_buffer[input_start..input_end],
388 &mut output[output_start..output_end],
389 );
390 }
391 }
392
393 if !last_chunk_is_runt {
394 return LanceBuffer::reinterpret_vec(output);
395 }
396
397 let last_chunk_start = num_whole_chunks * ELEMS_PER_CHUNK as usize;
398 unsafe {
400 output.set_len(num_whole_chunks * words_per_chunk);
401 }
402 let remaining_items = data_buffer.len() - last_chunk_start;
403
404 let uncompressed_bits = data.bits_per_value as usize;
405 let tail_bit_savings = uncompressed_bits.saturating_sub(compressed_bits_per_value);
406 let padding_cost = compressed_bits_per_value * (ELEMS_PER_CHUNK as usize - remaining_items);
407 let tail_pack_savings = tail_bit_savings.saturating_mul(remaining_items);
408 debug_assert!(remaining_items > 0, "remaining_items must be non-zero");
409 debug_assert!(tail_bit_savings > 0, "tail_bit_savings must be non-zero");
410
411 if padding_cost < tail_pack_savings {
412 let mut last_chunk: Vec<T> = vec![T::from_usize(0).unwrap(); ELEMS_PER_CHUNK as usize];
414 last_chunk[..remaining_items].copy_from_slice(&data_buffer[last_chunk_start..]);
415 let start = output.len();
416 unsafe {
417 output.set_len(start + words_per_chunk);
419 BitPacking::unchecked_pack(
420 compressed_bits_per_value,
421 &last_chunk,
422 &mut output[start..start + words_per_chunk],
423 );
424 }
425 } else {
426 output.extend_from_slice(&data_buffer[last_chunk_start..]);
428 }
429
430 LanceBuffer::reinterpret_vec(output)
431}
432
433fn unpack_out_of_line<T: ArrowNativeType + BitPacking>(
439 data: FixedWidthDataBlock,
440 num_values: usize,
441 compressed_bits_per_value: usize,
442) -> FixedWidthDataBlock {
443 let words_per_chunk = (ELEMS_PER_CHUNK as usize * compressed_bits_per_value)
444 .div_ceil(data.bits_per_value as usize);
445 let compressed_words = data.data.borrow_to_typed_slice::<T>();
446
447 let num_whole_chunks = num_values / ELEMS_PER_CHUNK as usize;
448 let tail_values = num_values % ELEMS_PER_CHUNK as usize;
449 let expected_full_words = num_whole_chunks * words_per_chunk;
450 let expected_new_len = expected_full_words + tail_values;
451 let tail_is_raw = tail_values > 0 && compressed_words.len() == expected_new_len;
452
453 let extra_tail_capacity = ELEMS_PER_CHUNK as usize;
454 #[allow(clippy::uninit_vec)]
455 let mut decompressed: Vec<T> =
456 Vec::with_capacity(num_values.saturating_add(extra_tail_capacity));
457 let chunk_value_len = num_whole_chunks * ELEMS_PER_CHUNK as usize;
458 unsafe {
459 decompressed.set_len(chunk_value_len);
460 }
461
462 for chunk_idx in 0..num_whole_chunks {
463 let input_start = chunk_idx * words_per_chunk;
464 let input_end = input_start + words_per_chunk;
465 let output_start = chunk_idx * ELEMS_PER_CHUNK as usize;
466 let output_end = output_start + ELEMS_PER_CHUNK as usize;
467 unsafe {
468 BitPacking::unchecked_unpack(
469 compressed_bits_per_value,
470 &compressed_words[input_start..input_end],
471 &mut decompressed[output_start..output_end],
472 );
473 }
474 }
475
476 if tail_values > 0 {
477 if tail_is_raw {
480 let tail_start = expected_full_words;
481 decompressed.extend_from_slice(&compressed_words[tail_start..tail_start + tail_values]);
482 } else {
483 let tail_start = expected_full_words;
484 let output_start = decompressed.len();
485 unsafe {
486 decompressed.set_len(output_start + ELEMS_PER_CHUNK as usize);
487 }
488 unsafe {
489 BitPacking::unchecked_unpack(
490 compressed_bits_per_value,
491 &compressed_words[tail_start..tail_start + words_per_chunk],
492 &mut decompressed[output_start..output_start + ELEMS_PER_CHUNK as usize],
493 );
494 }
495 decompressed.truncate(output_start + tail_values);
496 }
497 }
498
499 debug_assert_eq!(decompressed.len(), num_values);
500
501 FixedWidthDataBlock {
502 data: LanceBuffer::reinterpret_vec(decompressed),
503 bits_per_value: data.bits_per_value,
504 num_values: num_values as u64,
505 block_info: BlockInfo::new(),
506 }
507}
508
509#[derive(Debug)]
535pub struct OutOfLineBitpacking {
536 compressed_bit_width: u64,
537 uncompressed_bit_width: u64,
538}
539
540impl OutOfLineBitpacking {
541 pub fn new(compressed_bit_width: u64, uncompressed_bit_width: u64) -> Self {
542 Self {
543 compressed_bit_width,
544 uncompressed_bit_width,
545 }
546 }
547}
548
549impl BlockCompressor for OutOfLineBitpacking {
550 fn compress(&self, data: DataBlock) -> Result<LanceBuffer> {
551 let fixed_width = data.as_fixed_width().unwrap();
552 let compressed = match fixed_width.bits_per_value {
553 8 => bitpack_out_of_line::<u8>(fixed_width, self.compressed_bit_width as usize),
554 16 => bitpack_out_of_line::<u16>(fixed_width, self.compressed_bit_width as usize),
555 32 => bitpack_out_of_line::<u32>(fixed_width, self.compressed_bit_width as usize),
556 64 => bitpack_out_of_line::<u64>(fixed_width, self.compressed_bit_width as usize),
557 _ => panic!("Bitpacking word size must be 8,16,32,64"),
558 };
559 Ok(compressed)
560 }
561}
562
563impl BlockDecompressor for OutOfLineBitpacking {
564 fn decompress(&self, data: LanceBuffer, num_values: u64) -> Result<DataBlock> {
565 let word_size = match self.uncompressed_bit_width {
566 8 => std::mem::size_of::<u8>(),
567 16 => std::mem::size_of::<u16>(),
568 32 => std::mem::size_of::<u32>(),
569 64 => std::mem::size_of::<u64>(),
570 _ => panic!("Bitpacking word size must be 8,16,32,64"),
571 };
572 debug_assert_eq!(data.len() % word_size, 0);
573 let total_words = (data.len() / word_size) as u64;
574 let block = FixedWidthDataBlock {
575 data,
576 bits_per_value: self.uncompressed_bit_width,
577 num_values: total_words,
578 block_info: BlockInfo::new(),
579 };
580
581 let unpacked = match self.uncompressed_bit_width {
582 8 => unpack_out_of_line::<u8>(
583 block,
584 num_values as usize,
585 self.compressed_bit_width as usize,
586 ),
587 16 => unpack_out_of_line::<u16>(
588 block,
589 num_values as usize,
590 self.compressed_bit_width as usize,
591 ),
592 32 => unpack_out_of_line::<u32>(
593 block,
594 num_values as usize,
595 self.compressed_bit_width as usize,
596 ),
597 64 => unpack_out_of_line::<u64>(
598 block,
599 num_values as usize,
600 self.compressed_bit_width as usize,
601 ),
602 _ => unreachable!(),
603 };
604 Ok(DataBlock::FixedWidth(unpacked))
605 }
606}
607
608#[cfg(test)]
609mod test {
610 use std::{collections::HashMap, sync::Arc};
611
612 use arrow_array::{Array, Int8Array, Int64Array};
613 use arrow_buffer::ArrowNativeType;
614 use arrow_schema::DataType;
615 use bytemuck::Pod;
616 use lance_bitpacking::BitPacking;
617 use rstest::rstest;
618
619 use super::{ELEMS_PER_CHUNK, InlineBitpacking, bitpack_out_of_line, unpack_out_of_line};
620 use crate::{
621 buffer::LanceBuffer,
622 compression::{BlockDecompressor, MiniBlockDecompressor},
623 data::{BlockInfo, DataBlock, FixedWidthDataBlock},
624 testing::{TestCases, check_round_trip_encoding_of_data},
625 };
626
627 #[rstest]
628 #[case::u8(8)]
629 #[case::u16(16)]
630 #[case::u32(32)]
631 #[case::u64(64)]
632 fn test_inline_bitpacking_decompress_empty_miniblock(#[case] bit_width: u64) {
633 let decompressor = InlineBitpacking::new(bit_width);
634 let decompressed =
635 MiniBlockDecompressor::decompress(&decompressor, vec![LanceBuffer::empty()], 0)
636 .unwrap();
637
638 let DataBlock::FixedWidth(block) = decompressed else {
639 panic!("Expected FixedWidth block");
640 };
641 assert_eq!(block.bits_per_value, bit_width);
642 assert_eq!(block.num_values, 0);
643 assert_eq!(block.data.len(), 0);
644 }
645
646 #[rstest]
650 #[case::u8(8)]
651 #[case::u16(16)]
652 #[case::u32(32)]
653 #[case::u64(64)]
654 fn test_inline_bitpacking_decompress_empty_block(#[case] bit_width: u64) {
655 let decompressor = InlineBitpacking::new(bit_width);
656 let decompressed =
657 BlockDecompressor::decompress(&decompressor, LanceBuffer::empty(), 0).unwrap();
658
659 let DataBlock::FixedWidth(block) = decompressed else {
660 panic!("Expected FixedWidth block");
661 };
662 assert_eq!(block.bits_per_value, bit_width);
663 assert_eq!(block.num_values, 0);
664 assert_eq!(block.data.len(), 0);
665 }
666
667 fn roundtrip_unchunk<T>(values: &[T], bit_width: usize)
668 where
669 T: ArrowNativeType + BitPacking + Pod,
670 {
671 assert!(values.len() <= ELEMS_PER_CHUNK as usize);
672 let num_values = values.len() as u64;
673
674 let mut padded = vec![T::from_usize(0).unwrap(); ELEMS_PER_CHUNK as usize];
675 padded[..values.len()].copy_from_slice(values);
676
677 let packed_words = ELEMS_PER_CHUNK as usize * bit_width / (std::mem::size_of::<T>() * 8);
678 let mut chunk: Vec<T> = Vec::with_capacity(1 + packed_words);
679 chunk.push(T::from_usize(bit_width).unwrap());
680 let out_len = chunk.len();
681 chunk.resize(out_len + packed_words, T::from_usize(0).unwrap());
682 unsafe {
683 BitPacking::unchecked_pack(bit_width, &padded, &mut chunk[out_len..]);
684 }
685
686 let data = LanceBuffer::reinterpret_vec(chunk);
687 let decoded = InlineBitpacking::unchunk::<T>(data, num_values).unwrap();
688 let DataBlock::FixedWidth(fixed) = decoded else {
689 panic!("expected FixedWidth DataBlock");
690 };
691 let decoded_values = fixed.data.borrow_to_typed_view::<T>();
692 assert_eq!(decoded_values.as_ref(), values);
693 }
694
695 fn assert_corrupt_unchunk<T>(data: LanceBuffer, num_values: u64, expected_message: &str)
696 where
697 T: ArrowNativeType + BitPacking + Pod,
698 {
699 let err = InlineBitpacking::unchunk::<T>(data, num_values).unwrap_err();
700 assert!(matches!(err, lance_core::Error::CorruptFile { .. }));
701 let err = err.to_string();
702 assert!(
703 err.contains(expected_message),
704 "expected error containing {expected_message:?}, got {err:?}"
705 );
706 }
707
708 #[test]
709 fn unchunk_u32_bw12_tail() {
710 let values: Vec<u32> = (0..500).map(|i| ((i * 7) % (1 << 12)) as u32).collect();
711 roundtrip_unchunk(&values, 12);
712 }
713
714 #[test]
715 fn unchunk_u64_bw23_full() {
716 let values: Vec<u64> = (0..1024).map(|i| ((i * 3) % (1 << 23)) as u64).collect();
717 roundtrip_unchunk(&values, 23);
718 }
719
720 #[rstest]
721 #[case::too_small_header(LanceBuffer::from(vec![1, 2, 3]), 1, "too small")]
722 #[case::misaligned_chunk_size(LanceBuffer::from(vec![0, 0, 0, 0, 0]), 1, "multiple")]
723 #[case::too_many_values(
724 LanceBuffer::reinterpret_vec(vec![0_u32]),
725 ELEMS_PER_CHUNK + 1,
726 "expected at most"
727 )]
728 #[case::payload_size_mismatch(LanceBuffer::reinterpret_vec(vec![12_u32]), 1, "payload")]
729 #[case::invalid_bit_width(LanceBuffer::reinterpret_vec(vec![33_u32]), 1, "exceeds")]
730 fn unchunk_rejects(
731 #[case] data: LanceBuffer,
732 #[case] num_values: u64,
733 #[case] expected_message: &str,
734 ) {
735 assert_corrupt_unchunk::<u32>(data, num_values, expected_message);
736 }
737
738 #[test_log::test(tokio::test)]
739 async fn test_miniblock_bitpack() {
740 let test_cases = TestCases::default().with_structural_encodings();
741
742 let arrays = vec![
743 Arc::new(Int8Array::from(vec![100; 1024])) as Arc<dyn Array>,
744 Arc::new(Int8Array::from(vec![1; 1024])) as Arc<dyn Array>,
745 Arc::new(Int8Array::from(vec![16; 1024])) as Arc<dyn Array>,
746 Arc::new(Int8Array::from(vec![-1; 1024])) as Arc<dyn Array>,
747 Arc::new(Int8Array::from(vec![5; 1])) as Arc<dyn Array>,
748 ];
749 check_round_trip_encoding_of_data(arrays, &test_cases, HashMap::new()).await;
750
751 for data_type in [DataType::Int16, DataType::Int32, DataType::Int64] {
752 let int64_arrays = vec![
753 Int64Array::from(vec![3; 1024]),
754 Int64Array::from(vec![8; 1024]),
755 Int64Array::from(vec![16; 1024]),
756 Int64Array::from(vec![100; 1024]),
757 Int64Array::from(vec![512; 1024]),
758 Int64Array::from(vec![1000; 1024]),
759 Int64Array::from(vec![2000; 1024]),
760 Int64Array::from(vec![-1; 10]),
761 ];
762
763 let mut arrays = vec![];
764 for int64_array in int64_arrays {
765 arrays.push(arrow_cast::cast(&int64_array, &data_type).unwrap());
766 }
767
768 check_round_trip_encoding_of_data(arrays, &test_cases, HashMap::new()).await;
769 }
770 }
771
772 #[test_log::test(tokio::test)]
773 async fn test_bitpack_encoding_verification() {
774 use arrow_array::Int32Array;
775
776 let test_cases = TestCases::default()
778 .with_expected_encoding("inline_bitpacking")
779 .with_structural_encodings();
780
781 let mut values = Vec::new();
784 for i in 0..2048 {
785 values.push(i % 16); }
787
788 let arrays = vec![Arc::new(Int32Array::from(values)) as Arc<dyn Array>];
789
790 let mut metadata = HashMap::new();
792 metadata.insert("lance-encoding:bss".to_string(), "off".to_string());
793
794 check_round_trip_encoding_of_data(arrays, &test_cases, metadata.clone()).await;
795 }
796
797 #[test_log::test(tokio::test)]
798 async fn test_miniblock_bitpack_zero_chunk_selection() {
799 use arrow_array::Int32Array;
800
801 let test_cases = TestCases::default()
802 .with_expected_encoding("inline_bitpacking")
803 .with_structural_encodings();
804
805 let mut vals = vec![0i32; 1024];
808 for i in 0..1024 {
809 vals.push(i % 16);
810 }
811
812 let arrays = vec![Arc::new(Int32Array::from(vals)) as Arc<dyn Array>];
813
814 let mut metadata = HashMap::new();
816 metadata.insert("lance-encoding:bss".to_string(), "off".to_string());
817 metadata.insert("lance-encoding:rle-threshold".to_string(), "0".to_string());
818
819 check_round_trip_encoding_of_data(arrays, &test_cases, metadata).await;
820 }
821
822 #[test]
823 fn test_out_of_line_bitpack_raw_tail_roundtrip() {
824 let bit_width = 8usize;
825 let word_bits = std::mem::size_of::<u32>() as u64 * 8;
826 let values: Vec<u32> = (0..1025).map(|i| (i % 200) as u32).collect();
827 let input = FixedWidthDataBlock {
828 data: LanceBuffer::reinterpret_vec(values.clone()),
829 bits_per_value: word_bits,
830 num_values: values.len() as u64,
831 block_info: BlockInfo::new(),
832 };
833
834 let compressed = bitpack_out_of_line::<u32>(input, bit_width);
835 let compressed_words = compressed.borrow_to_typed_slice::<u32>().to_vec();
836 let words_per_chunk = (ELEMS_PER_CHUNK as usize * bit_width).div_ceil(word_bits as usize);
837 assert_eq!(
838 compressed_words.len(),
839 words_per_chunk + (values.len() - ELEMS_PER_CHUNK as usize),
840 );
841
842 let compressed_block = FixedWidthDataBlock {
843 data: LanceBuffer::reinterpret_vec(compressed_words.clone()),
844 bits_per_value: word_bits,
845 num_values: compressed_words.len() as u64,
846 block_info: BlockInfo::new(),
847 };
848
849 let decoded = unpack_out_of_line::<u32>(compressed_block, values.len(), bit_width);
850 let decoded_values = decoded.data.borrow_to_typed_slice::<u32>();
851 assert_eq!(decoded_values.as_ref(), values.as_slice());
852 }
853}