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