1use std::{
18 ops::Range,
19 sync::{Arc, RwLock},
20};
21
22use arrow_array::{
23 Array, ArrayRef, OffsetSizeTrait, UInt64Array,
24 cast::AsArray,
25 new_empty_array, new_null_array,
26 types::{ArrowDictionaryKeyType, UInt8Type, UInt16Type, UInt32Type, UInt64Type},
27};
28use arrow_buffer::{ArrowNativeType, BooleanBuffer, BooleanBufferBuilder, NullBuffer};
29use arrow_data::{ArrayData, ArrayDataBuilder};
30use arrow_schema::DataType;
31use lance_arrow::DataTypeExt;
32
33use lance_core::{Error, Result};
34
35use crate::{
36 buffer::LanceBuffer,
37 statistics::{ComputeStat, Stat},
38};
39
40#[derive(Debug, Clone)]
46pub struct AllNullDataBlock {
47 pub num_values: u64,
49}
50
51impl AllNullDataBlock {
52 fn into_arrow(self, data_type: DataType, _validate: bool) -> Result<ArrayData> {
53 Ok(ArrayData::new_null(&data_type, self.num_values as usize))
54 }
55
56 fn into_buffers(self) -> Vec<LanceBuffer> {
57 vec![]
58 }
59}
60
61use std::collections::HashMap;
62
63#[derive(Debug, Clone)]
66pub struct BlockInfo(pub Arc<RwLock<HashMap<Stat, Arc<dyn Array>>>>);
67
68impl Default for BlockInfo {
69 fn default() -> Self {
70 Self::new()
71 }
72}
73
74impl BlockInfo {
75 pub fn new() -> Self {
76 Self(Arc::new(RwLock::new(HashMap::new())))
77 }
78}
79
80impl PartialEq for BlockInfo {
81 fn eq(&self, other: &Self) -> bool {
82 let self_info = self.0.read().unwrap();
83 let other_info = other.0.read().unwrap();
84 *self_info == *other_info
85 }
86}
87
88#[derive(Debug, Clone)]
94pub struct NullableDataBlock {
95 pub data: Box<DataBlock>,
97 pub nulls: LanceBuffer,
99
100 pub block_info: BlockInfo,
101}
102
103impl NullableDataBlock {
104 fn into_arrow(self, data_type: DataType, validate: bool) -> Result<ArrayData> {
105 let nulls = self.nulls.into_buffer();
106 let data = self.data.into_arrow(data_type, validate)?.into_builder();
107 let data = data.null_bit_buffer(Some(nulls));
108 if validate {
109 Ok(data.build()?)
110 } else {
111 Ok(unsafe { data.build_unchecked() })
112 }
113 }
114
115 fn into_buffers(self) -> Vec<LanceBuffer> {
116 let mut buffers = vec![self.nulls];
117 buffers.extend(self.data.into_buffers());
118 buffers
119 }
120
121 pub fn data_size(&self) -> u64 {
122 self.data.data_size() + self.nulls.len() as u64
123 }
124}
125
126#[derive(Debug, PartialEq, Clone)]
128pub struct ConstantDataBlock {
129 pub data: LanceBuffer,
131 pub num_values: u64,
133}
134
135impl ConstantDataBlock {
136 fn into_buffers(self) -> Vec<LanceBuffer> {
137 vec![self.data]
138 }
139
140 fn into_arrow(self, _data_type: DataType, _validate: bool) -> Result<ArrayData> {
141 todo!()
144 }
145
146 pub fn try_clone(&self) -> Result<Self> {
147 Ok(Self {
148 data: self.data.clone(),
149 num_values: self.num_values,
150 })
151 }
152
153 pub fn data_size(&self) -> u64 {
154 self.data.len() as u64
155 }
156}
157
158#[derive(Debug, PartialEq, Clone)]
160pub struct FixedWidthDataBlock {
161 pub data: LanceBuffer,
163 pub bits_per_value: u64,
165 pub num_values: u64,
167
168 pub block_info: BlockInfo,
169}
170
171impl FixedWidthDataBlock {
172 fn do_into_arrow(
173 self,
174 data_type: DataType,
175 num_values: u64,
176 validate: bool,
177 ) -> Result<ArrayData> {
178 let data_buffer = if matches!(data_type, DataType::Boolean) && self.bits_per_value == 8 {
181 let mut builder = BooleanBufferBuilder::new(num_values as usize);
182 for &byte in self.data.as_ref().iter().take(num_values as usize) {
183 builder.append(byte != 0);
184 }
185 builder.finish().into_inner()
186 } else {
187 self.data.into_buffer()
188 };
189 let builder = ArrayDataBuilder::new(data_type)
190 .add_buffer(data_buffer)
191 .len(num_values as usize)
192 .null_count(0);
193 if validate {
194 Ok(builder.build()?)
195 } else {
196 Ok(unsafe { builder.build_unchecked() })
197 }
198 }
199
200 pub fn into_arrow(self, data_type: DataType, validate: bool) -> Result<ArrayData> {
201 let root_num_values = self.num_values;
202 self.do_into_arrow(data_type, root_num_values, validate)
203 }
204
205 pub fn into_buffers(self) -> Vec<LanceBuffer> {
206 vec![self.data]
207 }
208
209 pub fn try_clone(&self) -> Result<Self> {
210 Ok(Self {
211 data: self.data.clone(),
212 bits_per_value: self.bits_per_value,
213 num_values: self.num_values,
214 block_info: self.block_info.clone(),
215 })
216 }
217
218 pub fn data_size(&self) -> u64 {
219 self.data.len() as u64
220 }
221}
222
223#[derive(Debug)]
224pub struct VariableWidthDataBlockBuilder<T: OffsetSizeTrait> {
225 offsets: Vec<T>,
226 bytes: Vec<u8>,
227}
228
229impl<T: OffsetSizeTrait> VariableWidthDataBlockBuilder<T> {
230 fn new(estimated_size_bytes: u64) -> Self {
231 Self {
232 offsets: vec![T::from_usize(0).unwrap()],
233 bytes: Vec::with_capacity(estimated_size_bytes as usize),
234 }
235 }
236}
237
238impl<T: OffsetSizeTrait + bytemuck::Pod> DataBlockBuilderImpl for VariableWidthDataBlockBuilder<T> {
239 fn append(&mut self, data_block: &DataBlock, selection: Range<u64>) {
240 let block = data_block.as_variable_width_ref().unwrap();
241 assert!(block.bits_per_offset == T::get_byte_width() as u8 * 8);
242 let offsets = block.offsets.borrow_to_typed_view::<T>();
243
244 let start_offset = offsets[selection.start as usize];
245 let end_offset = offsets[selection.end as usize];
246 let mut previous_len = self.bytes.len();
247
248 self.bytes
249 .extend_from_slice(&block.data[start_offset.as_usize()..end_offset.as_usize()]);
250
251 self.offsets.extend(
252 offsets[selection.start as usize..selection.end as usize]
253 .iter()
254 .zip(&offsets[selection.start as usize + 1..=selection.end as usize])
255 .map(|(¤t, &next)| {
256 let this_value_len = next - current;
257 previous_len += this_value_len.as_usize();
258 T::from_usize(previous_len).unwrap()
259 }),
260 );
261 }
262
263 fn finish(self: Box<Self>) -> DataBlock {
264 let num_values = (self.offsets.len() - 1) as u64;
265 DataBlock::VariableWidth(VariableWidthBlock {
266 data: LanceBuffer::from(self.bytes),
267 offsets: LanceBuffer::reinterpret_vec(self.offsets),
268 bits_per_offset: T::get_byte_width() as u8 * 8,
269 num_values,
270 block_info: BlockInfo::new(),
271 })
272 }
273}
274
275#[derive(Debug)]
276struct BitmapDataBlockBuilder {
277 values: BooleanBufferBuilder,
278}
279
280impl BitmapDataBlockBuilder {
281 fn new(estimated_size_bytes: u64) -> Self {
282 Self {
283 values: BooleanBufferBuilder::new(estimated_size_bytes as usize * 8),
284 }
285 }
286}
287
288impl DataBlockBuilderImpl for BitmapDataBlockBuilder {
289 fn append(&mut self, data_block: &DataBlock, selection: Range<u64>) {
290 let bitmap_blk = data_block.as_fixed_width_ref().unwrap();
291 self.values.append_packed_range(
292 selection.start as usize..selection.end as usize,
293 &bitmap_blk.data,
294 );
295 }
296
297 fn finish(mut self: Box<Self>) -> DataBlock {
298 let bool_buf = self.values.finish();
299 let num_values = bool_buf.len() as u64;
300 let bits_buf = bool_buf.into_inner();
301 DataBlock::FixedWidth(FixedWidthDataBlock {
302 data: LanceBuffer::from(bits_buf),
303 bits_per_value: 1,
304 num_values,
305 block_info: BlockInfo::new(),
306 })
307 }
308}
309
310#[derive(Debug)]
311struct FixedWidthDataBlockBuilder {
312 bits_per_value: u64,
313 bytes_per_value: u64,
314 values: Vec<u8>,
315}
316
317impl FixedWidthDataBlockBuilder {
318 fn new(bits_per_value: u64, estimated_size_bytes: u64) -> Self {
319 assert!(bits_per_value.is_multiple_of(8));
320 Self {
321 bits_per_value,
322 bytes_per_value: bits_per_value / 8,
323 values: Vec::with_capacity(estimated_size_bytes as usize),
324 }
325 }
326}
327
328impl DataBlockBuilderImpl for FixedWidthDataBlockBuilder {
329 fn append(&mut self, data_block: &DataBlock, selection: Range<u64>) {
330 let block = data_block.as_fixed_width_ref().unwrap();
331 assert_eq!(self.bits_per_value, block.bits_per_value);
332 let start = selection.start as usize * self.bytes_per_value as usize;
333 let end = selection.end as usize * self.bytes_per_value as usize;
334 self.values.extend_from_slice(&block.data[start..end]);
335 }
336
337 fn finish(self: Box<Self>) -> DataBlock {
338 let num_values = (self.values.len() / self.bytes_per_value as usize) as u64;
339 DataBlock::FixedWidth(FixedWidthDataBlock {
340 data: LanceBuffer::from(self.values),
341 bits_per_value: self.bits_per_value,
342 num_values,
343 block_info: BlockInfo::new(),
344 })
345 }
346}
347
348#[derive(Debug)]
349struct StructDataBlockBuilder {
350 children: Vec<Box<dyn DataBlockBuilderImpl>>,
351}
352
353impl StructDataBlockBuilder {
354 fn new(children: Vec<Box<dyn DataBlockBuilderImpl>>) -> Self {
355 Self { children }
356 }
357}
358
359impl DataBlockBuilderImpl for StructDataBlockBuilder {
360 fn append(&mut self, data_block: &DataBlock, selection: Range<u64>) {
361 let data_block = data_block.as_struct_ref().unwrap();
362 for i in 0..self.children.len() {
363 self.children[i].append(&data_block.children[i], selection.clone());
364 }
365 }
366
367 fn finish(self: Box<Self>) -> DataBlock {
368 let mut children_data_block = Vec::new();
369 for child in self.children {
370 let child_data_block = child.finish();
371 children_data_block.push(child_data_block);
372 }
373 DataBlock::Struct(StructDataBlock {
374 children: children_data_block,
375 block_info: BlockInfo::new(),
376 validity: None,
377 })
378 }
379}
380
381#[derive(Debug, Default)]
382struct AllNullDataBlockBuilder {
383 num_values: u64,
384}
385
386impl DataBlockBuilderImpl for AllNullDataBlockBuilder {
387 fn append(&mut self, _data_block: &DataBlock, selection: Range<u64>) {
388 self.num_values += selection.end - selection.start;
389 }
390
391 fn finish(self: Box<Self>) -> DataBlock {
392 DataBlock::AllNull(AllNullDataBlock {
393 num_values: self.num_values,
394 })
395 }
396}
397
398#[derive(Debug, Clone)]
400pub struct FixedSizeListBlock {
401 pub child: Box<DataBlock>,
403 pub dimension: u64,
405}
406
407impl FixedSizeListBlock {
408 pub fn num_values(&self) -> u64 {
409 self.child.num_values() / self.dimension
410 }
411
412 pub fn try_into_flat(self) -> Option<FixedWidthDataBlock> {
416 match *self.child {
417 DataBlock::Nullable(_) => None,
419 DataBlock::FixedSizeList(inner) => {
420 let mut flat = inner.try_into_flat()?;
421 flat.bits_per_value *= self.dimension;
422 flat.num_values /= self.dimension;
423 Some(flat)
424 }
425 DataBlock::FixedWidth(mut inner) => {
426 inner.bits_per_value *= self.dimension;
427 inner.num_values /= self.dimension;
428 Some(inner)
429 }
430 _ => panic!(
431 "Expected FixedSizeList or FixedWidth data block but found {:?}",
432 self
433 ),
434 }
435 }
436
437 pub fn flatten_as_fixed(&mut self) -> FixedWidthDataBlock {
438 match self.child.as_mut() {
439 DataBlock::FixedSizeList(fsl) => fsl.flatten_as_fixed(),
440 DataBlock::FixedWidth(fw) => fw.clone(),
441 _ => panic!("Expected FixedSizeList or FixedWidth data block"),
442 }
443 }
444
445 pub fn from_flat(data: FixedWidthDataBlock, data_type: &DataType) -> DataBlock {
447 match data_type {
448 DataType::FixedSizeList(child_field, dimension) => {
449 let mut data = data;
450 data.bits_per_value /= *dimension as u64;
451 data.num_values *= *dimension as u64;
452 let child_data = Self::from_flat(data, child_field.data_type());
453 DataBlock::FixedSizeList(Self {
454 child: Box::new(child_data),
455 dimension: *dimension as u64,
456 })
457 }
458 _ => DataBlock::FixedWidth(data),
460 }
461 }
462
463 fn into_arrow(self, data_type: DataType, validate: bool) -> Result<ArrayData> {
464 let num_values = self.num_values();
465 let builder = match &data_type {
466 DataType::FixedSizeList(child_field, _) => {
467 let child_data = self
468 .child
469 .into_arrow(child_field.data_type().clone(), validate)?;
470 ArrayDataBuilder::new(data_type)
471 .add_child_data(child_data)
472 .len(num_values as usize)
473 .null_count(0)
474 }
475 _ => panic!("Expected FixedSizeList data type and got {:?}", data_type),
476 };
477 if validate {
478 Ok(builder.build()?)
479 } else {
480 Ok(unsafe { builder.build_unchecked() })
481 }
482 }
483
484 fn into_buffers(self) -> Vec<LanceBuffer> {
485 self.child.into_buffers()
486 }
487
488 fn data_size(&self) -> u64 {
489 self.child.data_size()
490 }
491}
492
493#[derive(Debug)]
494struct FixedSizeListBlockBuilder {
495 inner: Box<dyn DataBlockBuilderImpl>,
496 dimension: u64,
497}
498
499impl FixedSizeListBlockBuilder {
500 fn new(inner: Box<dyn DataBlockBuilderImpl>, dimension: u64) -> Self {
501 Self { inner, dimension }
502 }
503}
504
505impl DataBlockBuilderImpl for FixedSizeListBlockBuilder {
506 fn append(&mut self, data_block: &DataBlock, selection: Range<u64>) {
507 let selection = selection.start * self.dimension..selection.end * self.dimension;
508 let fsl = data_block.as_fixed_size_list_ref().unwrap();
509 self.inner.append(fsl.child.as_ref(), selection);
510 }
511
512 fn finish(self: Box<Self>) -> DataBlock {
513 let inner_block = self.inner.finish();
514 DataBlock::FixedSizeList(FixedSizeListBlock {
515 child: Box::new(inner_block),
516 dimension: self.dimension,
517 })
518 }
519}
520
521#[derive(Debug)]
522struct NullableDataBlockBuilder {
523 inner: Box<dyn DataBlockBuilderImpl>,
524 validity: BooleanBufferBuilder,
525}
526
527impl NullableDataBlockBuilder {
528 fn new(inner: Box<dyn DataBlockBuilderImpl>, estimated_size_bytes: usize) -> Self {
529 Self {
530 inner,
531 validity: BooleanBufferBuilder::new(estimated_size_bytes * 8),
532 }
533 }
534}
535
536impl DataBlockBuilderImpl for NullableDataBlockBuilder {
537 fn append(&mut self, data_block: &DataBlock, selection: Range<u64>) {
538 let nullable = data_block.as_nullable_ref().unwrap();
539 let bool_buf = BooleanBuffer::new(
540 nullable.nulls.clone().into_buffer(),
541 selection.start as usize,
542 (selection.end - selection.start) as usize,
543 );
544 self.validity.append_buffer(&bool_buf);
545 self.inner.append(nullable.data.as_ref(), selection);
546 }
547
548 fn finish(mut self: Box<Self>) -> DataBlock {
549 let inner_block = self.inner.finish();
550 DataBlock::Nullable(NullableDataBlock {
551 data: Box::new(inner_block),
552 nulls: LanceBuffer::from(self.validity.finish().into_inner()),
553 block_info: BlockInfo::new(),
554 })
555 }
556}
557
558#[derive(Debug, Clone)]
562pub struct OpaqueBlock {
563 pub buffers: Vec<LanceBuffer>,
564 pub num_values: u64,
565 pub block_info: BlockInfo,
566}
567
568impl OpaqueBlock {
569 pub fn data_size(&self) -> u64 {
570 self.buffers.iter().map(|b| b.len() as u64).sum()
571 }
572}
573
574#[derive(Debug, Clone)]
576pub struct VariableWidthBlock {
577 pub data: LanceBuffer,
579 pub offsets: LanceBuffer,
583 pub bits_per_offset: u8,
585 pub num_values: u64,
587
588 pub block_info: BlockInfo,
589}
590
591struct ValidVariableWidthLayout;
599
600fn corrupt_file_named(name: &str, message: impl Into<String>) -> Error {
601 Error::corrupt_file(name.into(), message)
602}
603
604impl VariableWidthBlock {
605 fn into_arrow(self, data_type: DataType, _validate: bool) -> Result<ArrayData> {
613 let Some(expected_bits_per_offset) = Self::expected_bits_per_offset(&data_type) else {
614 return self.into_arrow_checked(data_type);
617 };
618 if self.bits_per_offset != expected_bits_per_offset {
619 return Err(self.layout_error(
620 &data_type,
621 format!(
622 "expected {}-bit offsets but got {}-bit offsets",
623 expected_bits_per_offset, self.bits_per_offset
624 ),
625 ));
626 }
627 if self.num_values == 0 {
628 return self.into_arrow_checked(data_type);
630 }
631 let proof = self.validate_layout(&data_type)?;
632 Ok(self.into_arrow_unchecked(data_type, proof))
633 }
634
635 fn expected_bits_per_offset(data_type: &DataType) -> Option<u8> {
638 match data_type {
639 DataType::Binary | DataType::Utf8 => Some(32),
640 DataType::LargeBinary | DataType::LargeUtf8 => Some(64),
641 _ => None,
642 }
643 }
644
645 fn layout_error(&self, data_type: &DataType, detail: impl std::fmt::Display) -> Error {
646 Self::format_layout_error(
647 data_type,
648 detail,
649 self.num_values,
650 self.bits_per_offset,
651 self.offsets.len(),
652 self.data.len(),
653 )
654 }
655
656 fn format_layout_error(
657 data_type: &DataType,
658 detail: impl std::fmt::Display,
659 num_values: u64,
660 bits_per_offset: u8,
661 offsets_size: usize,
662 data_size: usize,
663 ) -> Error {
664 corrupt_file_named(
665 "variable width data block",
666 format!(
667 "invalid variable-width layout for {}: {} (num_values: {}, bits_per_offset: {}, \
668 offsets buffer size: {} bytes, data buffer size: {} bytes)",
669 data_type, detail, num_values, bits_per_offset, offsets_size, data_size,
670 ),
671 )
672 }
673
674 fn validate_layout(&self, data_type: &DataType) -> Result<ValidVariableWidthLayout> {
675 let bytes_per_offset = (self.bits_per_offset / 8) as u64;
676 let required_bytes = self
677 .num_values
678 .checked_add(1)
679 .and_then(|num_offsets| num_offsets.checked_mul(bytes_per_offset))
680 .ok_or_else(|| self.layout_error(data_type, "offsets buffer size overflows"))?;
681 if (self.offsets.len() as u64) < required_bytes {
682 return Err(self.layout_error(
683 data_type,
684 format!(
685 "offsets buffer must hold at least {} offsets ({} bytes)",
686 self.num_values + 1,
687 required_bytes
688 ),
689 ));
690 }
691 let validate_utf8 = matches!(data_type, DataType::Utf8 | DataType::LargeUtf8);
692 match self.bits_per_offset {
693 32 => self.validate_offsets_and_values::<i32>(data_type, validate_utf8),
694 64 => self.validate_offsets_and_values::<i64>(data_type, validate_utf8),
695 other => Err(self.layout_error(
696 data_type,
697 format!("unsupported offset width: {} bits", other),
698 )),
699 }
700 }
701
702 fn validate_offsets_and_values<T: ArrowNativeType + Ord>(
703 &self,
704 data_type: &DataType,
705 validate_utf8: bool,
706 ) -> Result<ValidVariableWidthLayout> {
707 let num_offsets = self.num_values as usize + 1;
708 let offsets = self
711 .offsets
712 .slice_with_length(0, num_offsets * std::mem::size_of::<T>());
713 let offsets = offsets.borrow_to_typed_slice::<T>();
714 let offsets: &[T] = offsets.as_ref();
715 let data = self.data.as_ref();
716
717 let mut is_monotonic = true;
722 for window in offsets.windows(2) {
723 is_monotonic &= window[0] <= window[1];
724 }
725 let first = offsets[0];
726 let last = offsets[num_offsets - 1];
727 let bounds_ok =
728 first >= T::usize_as(0) && last.to_usize().is_some_and(|last| last <= data.len());
729 if !is_monotonic || !bounds_ok {
730 return Err(self.offset_violation_error::<T>(data_type, offsets));
731 }
732
733 if validate_utf8 {
734 let (first, last) = (first.as_usize(), last.as_usize());
735 let values = std::str::from_utf8(&data[first..last])
736 .map_err(|utf8_err| self.layout_error(data_type, utf8_err))?;
737 let mut on_char_boundaries = true;
738 for &offset in offsets {
739 on_char_boundaries &= values.is_char_boundary(offset.as_usize() - first);
740 }
741 if !on_char_boundaries {
742 let position = offsets
744 .iter()
745 .position(|offset| !values.is_char_boundary(offset.as_usize() - first))
746 .expect("the fast scan found a non-boundary offset");
747 return Err(self.layout_error(
748 data_type,
749 format!("offset at position {position} splits a UTF-8 character"),
750 ));
751 }
752 }
753
754 Ok(ValidVariableWidthLayout)
755 }
756
757 fn offset_violation_error<T: ArrowNativeType + Ord>(
759 &self,
760 data_type: &DataType,
761 offsets: &[T],
762 ) -> Error {
763 let data_size = self.data.len();
764 for (position, window) in offsets.windows(2).enumerate() {
765 if window[0] > window[1] {
766 return self.layout_error(
767 data_type,
768 format!(
769 "non-monotonic offset at position {}: {:?} > {:?}",
770 position, window[0], window[1]
771 ),
772 );
773 }
774 }
775 for (position, offset) in offsets.iter().enumerate() {
776 match offset.to_usize() {
777 None => {
778 return self.layout_error(
779 data_type,
780 format!("negative offset at position {}: {:?}", position, offset),
781 );
782 }
783 Some(offset) if offset > data_size => {
784 return self.layout_error(
785 data_type,
786 format!(
787 "offset at position {} out of bounds: {} > {}",
788 position, offset, data_size
789 ),
790 );
791 }
792 Some(_) => {}
793 }
794 }
795 self.layout_error(data_type, "offsets failed validation")
798 }
799
800 fn into_arrow_checked(self, data_type: DataType) -> Result<ArrayData> {
801 let num_values = self.num_values;
802 let bits_per_offset = self.bits_per_offset;
803 let offsets_size = self.offsets.len();
804 let data_size = self.data.len();
805 let builder = self.into_arrow_builder(data_type.clone());
806 builder.build().map_err(|arrow_err| {
807 Self::format_layout_error(
808 &data_type,
809 arrow_err,
810 num_values,
811 bits_per_offset,
812 offsets_size,
813 data_size,
814 )
815 })
816 }
817
818 fn into_arrow_unchecked(
819 self,
820 data_type: DataType,
821 _proof: ValidVariableWidthLayout,
822 ) -> ArrayData {
823 let builder = self.into_arrow_builder(data_type);
824 unsafe { builder.build_unchecked() }
827 }
828
829 fn into_arrow_builder(self, data_type: DataType) -> ArrayDataBuilder {
830 let num_values = self.num_values;
831 let data_buffer = self.data.into_buffer();
832 let offsets_buffer = self.offsets.into_buffer();
833 ArrayDataBuilder::new(data_type)
834 .add_buffer(offsets_buffer)
835 .add_buffer(data_buffer)
836 .len(num_values as usize)
837 .null_count(0)
838 }
839
840 fn into_buffers(self) -> Vec<LanceBuffer> {
841 vec![self.offsets, self.data]
842 }
843
844 pub fn offsets_as_block(&mut self) -> DataBlock {
845 let offsets = self.offsets.clone();
846 DataBlock::FixedWidth(FixedWidthDataBlock {
847 data: offsets,
848 bits_per_value: self.bits_per_offset as u64,
849 num_values: self.num_values + 1,
850 block_info: BlockInfo::new(),
851 })
852 }
853
854 pub fn data_size(&self) -> u64 {
855 (self.data.len() + self.offsets.len()) as u64
856 }
857}
858
859#[derive(Debug, Clone)]
861pub struct StructDataBlock {
862 pub children: Vec<DataBlock>,
864 pub block_info: BlockInfo,
865 pub validity: Option<NullBuffer>,
867}
868
869impl StructDataBlock {
870 fn into_arrow(self, data_type: DataType, validate: bool) -> Result<ArrayData> {
871 if let DataType::Struct(fields) = &data_type {
872 let mut builder = ArrayDataBuilder::new(DataType::Struct(fields.clone()));
873 let mut num_rows = 0;
874 for (field, child) in fields.iter().zip(self.children) {
875 let child_data = child.into_arrow(field.data_type().clone(), validate)?;
876 num_rows = child_data.len();
877 builder = builder.add_child_data(child_data);
878 }
879
880 let builder = if let Some(validity) = self.validity {
882 let null_count = validity.null_count();
883 builder
884 .null_bit_buffer(Some(validity.into_inner().into_inner()))
885 .null_count(null_count)
886 } else {
887 builder.null_count(0)
888 };
889
890 let builder = builder.len(num_rows);
891 if validate {
892 Ok(builder.build()?)
893 } else {
894 Ok(unsafe { builder.build_unchecked() })
895 }
896 } else {
897 Err(Error::internal(format!(
898 "Expected Struct, got {:?}",
899 data_type
900 )))
901 }
902 }
903
904 fn remove_outer_validity(self) -> Self {
905 Self {
906 children: self
907 .children
908 .into_iter()
909 .map(|c| c.remove_outer_validity())
910 .collect(),
911 block_info: self.block_info,
912 validity: None, }
914 }
915
916 fn into_buffers(self) -> Vec<LanceBuffer> {
917 self.children
918 .into_iter()
919 .flat_map(|c| c.into_buffers())
920 .collect()
921 }
922
923 pub fn has_variable_width_child(&self) -> bool {
924 self.children
925 .iter()
926 .any(|child| !matches!(child, DataBlock::FixedWidth(_)))
927 }
928
929 pub fn data_size(&self) -> u64 {
930 self.children
931 .iter()
932 .map(|data_block| data_block.data_size())
933 .sum()
934 }
935}
936
937#[derive(Debug, Clone)]
939pub struct DictionaryDataBlock {
940 pub indices: FixedWidthDataBlock,
942 pub dictionary: Box<DataBlock>,
944}
945
946impl DictionaryDataBlock {
947 fn decode_helper<K: ArrowDictionaryKeyType>(self) -> Result<DataBlock> {
948 if self.indices.num_values == 0 {
951 return Ok(DataBlock::AllNull(AllNullDataBlock { num_values: 0 }));
952 }
953
954 let estimated_size_bytes = self.dictionary.data_size()
956 * (self.indices.num_values + self.dictionary.num_values() - 1)
957 / self.dictionary.num_values();
958 let mut data_builder = DataBlockBuilder::with_capacity_estimate(estimated_size_bytes);
959
960 let indices = self.indices.data.borrow_to_typed_slice::<K::Native>();
961 let indices = indices.as_ref();
962
963 indices
964 .iter()
965 .map(|idx| idx.to_usize().unwrap() as u64)
966 .for_each(|idx| {
967 data_builder.append(&self.dictionary, idx..idx + 1);
968 });
969
970 Ok(data_builder.finish())
971 }
972
973 pub fn decode(self) -> Result<DataBlock> {
974 match self.indices.bits_per_value {
975 8 => self.decode_helper::<UInt8Type>(),
976 16 => self.decode_helper::<UInt16Type>(),
977 32 => self.decode_helper::<UInt32Type>(),
978 64 => self.decode_helper::<UInt64Type>(),
979 _ => Err(lance_core::Error::internal(format!(
980 "Unsupported dictionary index bit width: {} bits",
981 self.indices.bits_per_value
982 ))),
983 }
984 }
985
986 fn into_arrow_dict(
987 self,
988 key_type: Box<DataType>,
989 value_type: Box<DataType>,
990 validate: bool,
991 ) -> Result<ArrayData> {
992 let indices = self.indices.into_arrow((*key_type).clone(), validate)?;
993 let dictionary = self
994 .dictionary
995 .into_arrow((*value_type).clone(), validate)?;
996
997 let builder = indices
998 .into_builder()
999 .add_child_data(dictionary)
1000 .data_type(DataType::Dictionary(key_type, value_type));
1001
1002 if validate {
1003 Ok(builder.build()?)
1004 } else {
1005 Ok(unsafe { builder.build_unchecked() })
1006 }
1007 }
1008
1009 fn into_arrow(self, data_type: DataType, validate: bool) -> Result<ArrayData> {
1010 if let DataType::Dictionary(key_type, value_type) = data_type {
1011 self.into_arrow_dict(key_type, value_type, validate)
1012 } else {
1013 self.decode()?.into_arrow(data_type, validate)
1014 }
1015 }
1016
1017 fn into_buffers(self) -> Vec<LanceBuffer> {
1018 let mut buffers = self.indices.into_buffers();
1019 buffers.extend(self.dictionary.into_buffers());
1020 buffers
1021 }
1022
1023 pub fn into_parts(self) -> (DataBlock, DataBlock) {
1024 (DataBlock::FixedWidth(self.indices), *self.dictionary)
1025 }
1026
1027 pub fn from_parts(indices: FixedWidthDataBlock, dictionary: DataBlock) -> Self {
1028 Self {
1029 indices,
1030 dictionary: Box::new(dictionary),
1031 }
1032 }
1033}
1034
1035#[derive(Debug, Clone)]
1050pub enum DataBlock {
1051 Empty(),
1052 Constant(ConstantDataBlock),
1053 AllNull(AllNullDataBlock),
1054 Nullable(NullableDataBlock),
1055 FixedWidth(FixedWidthDataBlock),
1056 FixedSizeList(FixedSizeListBlock),
1057 VariableWidth(VariableWidthBlock),
1058 Opaque(OpaqueBlock),
1059 Struct(StructDataBlock),
1060 Dictionary(DictionaryDataBlock),
1061}
1062
1063impl DataBlock {
1064 pub fn into_arrow(self, data_type: DataType, validate: bool) -> Result<ArrayData> {
1066 match self {
1067 Self::Empty() => Ok(new_empty_array(&data_type).to_data()),
1068 Self::Constant(inner) => inner.into_arrow(data_type, validate),
1069 Self::AllNull(inner) => inner.into_arrow(data_type, validate),
1070 Self::Nullable(inner) => inner.into_arrow(data_type, validate),
1071 Self::FixedWidth(inner) => inner.into_arrow(data_type, validate),
1072 Self::FixedSizeList(inner) => inner.into_arrow(data_type, validate),
1073 Self::VariableWidth(inner) => inner.into_arrow(data_type, validate),
1074 Self::Struct(inner) => inner.into_arrow(data_type, validate),
1075 Self::Dictionary(inner) => inner.into_arrow(data_type, validate),
1076 Self::Opaque(_) => Err(Error::internal(
1077 "Cannot convert OpaqueBlock to Arrow".to_string(),
1078 )),
1079 }
1080 }
1081
1082 pub fn into_buffers(self) -> Vec<LanceBuffer> {
1086 match self {
1087 Self::Empty() => Vec::default(),
1088 Self::Constant(inner) => inner.into_buffers(),
1089 Self::AllNull(inner) => inner.into_buffers(),
1090 Self::Nullable(inner) => inner.into_buffers(),
1091 Self::FixedWidth(inner) => inner.into_buffers(),
1092 Self::FixedSizeList(inner) => inner.into_buffers(),
1093 Self::VariableWidth(inner) => inner.into_buffers(),
1094 Self::Struct(inner) => inner.into_buffers(),
1095 Self::Dictionary(inner) => inner.into_buffers(),
1096 Self::Opaque(inner) => inner.buffers,
1097 }
1098 }
1099
1100 pub fn try_clone(&self) -> Result<Self> {
1109 match self {
1110 Self::Empty() => Ok(Self::Empty()),
1111 Self::Constant(inner) => Ok(Self::Constant(inner.clone())),
1112 Self::AllNull(inner) => Ok(Self::AllNull(inner.clone())),
1113 Self::Nullable(inner) => Ok(Self::Nullable(inner.clone())),
1114 Self::FixedWidth(inner) => Ok(Self::FixedWidth(inner.clone())),
1115 Self::FixedSizeList(inner) => Ok(Self::FixedSizeList(inner.clone())),
1116 Self::VariableWidth(inner) => Ok(Self::VariableWidth(inner.clone())),
1117 Self::Struct(inner) => Ok(Self::Struct(inner.clone())),
1118 Self::Dictionary(inner) => Ok(Self::Dictionary(inner.clone())),
1119 Self::Opaque(inner) => Ok(Self::Opaque(inner.clone())),
1120 }
1121 }
1122
1123 pub fn name(&self) -> &'static str {
1124 match self {
1125 Self::Constant(_) => "Constant",
1126 Self::Empty() => "Empty",
1127 Self::AllNull(_) => "AllNull",
1128 Self::Nullable(_) => "Nullable",
1129 Self::FixedWidth(_) => "FixedWidth",
1130 Self::FixedSizeList(_) => "FixedSizeList",
1131 Self::VariableWidth(_) => "VariableWidth",
1132 Self::Struct(_) => "Struct",
1133 Self::Dictionary(_) => "Dictionary",
1134 Self::Opaque(_) => "Opaque",
1135 }
1136 }
1137
1138 pub fn is_variable(&self) -> bool {
1139 match self {
1140 Self::Constant(_) => false,
1141 Self::Empty() => false,
1142 Self::AllNull(_) => false,
1143 Self::Nullable(nullable) => nullable.data.is_variable(),
1144 Self::FixedWidth(_) => false,
1145 Self::FixedSizeList(fsl) => fsl.child.is_variable(),
1146 Self::VariableWidth(_) => true,
1147 Self::Struct(strct) => strct.children.iter().any(|c| c.is_variable()),
1148 Self::Dictionary(_) => {
1149 todo!("is_variable for DictionaryDataBlock is not implemented yet")
1150 }
1151 Self::Opaque(_) => panic!("Does not make sense to ask if an Opaque block is variable"),
1152 }
1153 }
1154
1155 pub fn is_nullable(&self) -> bool {
1156 match self {
1157 Self::AllNull(_) => true,
1158 Self::Nullable(_) => true,
1159 Self::FixedSizeList(fsl) => fsl.child.is_nullable(),
1160 Self::Struct(strct) => strct.children.iter().any(|c| c.is_nullable()),
1161 Self::Dictionary(_) => {
1162 todo!("is_nullable for DictionaryDataBlock is not implemented yet")
1163 }
1164 Self::Opaque(_) => panic!("Does not make sense to ask if an Opaque block is nullable"),
1165 _ => false,
1166 }
1167 }
1168
1169 pub fn num_values(&self) -> u64 {
1174 match self {
1175 Self::Empty() => 0,
1176 Self::Constant(inner) => inner.num_values,
1177 Self::AllNull(inner) => inner.num_values,
1178 Self::Nullable(inner) => inner.data.num_values(),
1179 Self::FixedWidth(inner) => inner.num_values,
1180 Self::FixedSizeList(inner) => inner.num_values(),
1181 Self::VariableWidth(inner) => inner.num_values,
1182 Self::Struct(inner) => inner.children[0].num_values(),
1183 Self::Dictionary(inner) => inner.indices.num_values,
1184 Self::Opaque(inner) => inner.num_values,
1185 }
1186 }
1187
1188 pub fn items_per_row(&self) -> u64 {
1192 match self {
1193 Self::Empty() => todo!(), Self::Constant(_) => todo!(), Self::AllNull(_) => todo!(), Self::Nullable(nullable) => nullable.data.items_per_row(),
1197 Self::FixedWidth(_) => 1,
1198 Self::FixedSizeList(fsl) => fsl.dimension * fsl.child.items_per_row(),
1199 Self::VariableWidth(_) => 1,
1200 Self::Struct(_) => todo!(), Self::Dictionary(_) => 1,
1202 Self::Opaque(_) => 1,
1203 }
1204 }
1205
1206 pub fn data_size(&self) -> u64 {
1208 match self {
1209 Self::Empty() => 0,
1210 Self::Constant(inner) => inner.data_size(),
1211 Self::AllNull(_) => 0,
1212 Self::Nullable(inner) => inner.data_size(),
1213 Self::FixedWidth(inner) => inner.data_size(),
1214 Self::FixedSizeList(inner) => inner.data_size(),
1215 Self::VariableWidth(inner) => inner.data_size(),
1216 Self::Struct(inner) => inner.children.iter().map(|child| child.data_size()).sum(),
1217 Self::Dictionary(inner) => inner.indices.data_size() + inner.dictionary.data_size(),
1218 Self::Opaque(inner) => inner.data_size(),
1219 }
1220 }
1221
1222 pub fn remove_outer_validity(self) -> Self {
1230 match self {
1231 Self::AllNull(_) => panic!("Cannot remove validity on all-null data"),
1232 Self::Nullable(inner) => *inner.data,
1233 Self::Struct(inner) => Self::Struct(inner.remove_outer_validity()),
1234 other => other,
1235 }
1236 }
1237
1238 pub fn make_builder(&self, estimated_size_bytes: u64) -> Box<dyn DataBlockBuilderImpl> {
1239 match self {
1240 Self::FixedWidth(inner) => {
1241 if inner.bits_per_value == 1 {
1242 Box::new(BitmapDataBlockBuilder::new(estimated_size_bytes))
1243 } else {
1244 Box::new(FixedWidthDataBlockBuilder::new(
1245 inner.bits_per_value,
1246 estimated_size_bytes,
1247 ))
1248 }
1249 }
1250 Self::VariableWidth(inner) => {
1251 if inner.bits_per_offset == 32 {
1252 Box::new(VariableWidthDataBlockBuilder::<i32>::new(
1253 estimated_size_bytes,
1254 ))
1255 } else if inner.bits_per_offset == 64 {
1256 Box::new(VariableWidthDataBlockBuilder::<i64>::new(
1257 estimated_size_bytes,
1258 ))
1259 } else {
1260 todo!()
1261 }
1262 }
1263 Self::FixedSizeList(inner) => {
1264 let inner_builder = inner.child.make_builder(estimated_size_bytes);
1265 Box::new(FixedSizeListBlockBuilder::new(
1266 inner_builder,
1267 inner.dimension,
1268 ))
1269 }
1270 Self::Nullable(nullable) => {
1271 let estimated_validity_size_bytes = estimated_size_bytes / 16;
1274 let inner_builder = nullable
1275 .data
1276 .make_builder(estimated_size_bytes - estimated_validity_size_bytes);
1277 Box::new(NullableDataBlockBuilder::new(
1278 inner_builder,
1279 estimated_validity_size_bytes as usize,
1280 ))
1281 }
1282 Self::Struct(struct_data_block) => {
1283 let num_children = struct_data_block.children.len();
1284 let per_child_estimate = if num_children == 0 {
1285 0
1286 } else {
1287 estimated_size_bytes / num_children as u64
1288 };
1289 let child_builders = struct_data_block
1290 .children
1291 .iter()
1292 .map(|child| child.make_builder(per_child_estimate))
1293 .collect();
1294 Box::new(StructDataBlockBuilder::new(child_builders))
1295 }
1296 Self::AllNull(_) => Box::new(AllNullDataBlockBuilder::default()),
1297 _ => todo!("make_builder for {:?}", self),
1298 }
1299 }
1300}
1301
1302macro_rules! as_type {
1303 ($fn_name:ident, $inner:tt, $inner_type:ident) => {
1304 pub fn $fn_name(self) -> Option<$inner_type> {
1305 match self {
1306 Self::$inner(inner) => Some(inner),
1307 _ => None,
1308 }
1309 }
1310 };
1311}
1312
1313macro_rules! as_type_ref {
1314 ($fn_name:ident, $inner:tt, $inner_type:ident) => {
1315 pub fn $fn_name(&self) -> Option<&$inner_type> {
1316 match self {
1317 Self::$inner(inner) => Some(inner),
1318 _ => None,
1319 }
1320 }
1321 };
1322}
1323
1324macro_rules! as_type_ref_mut {
1325 ($fn_name:ident, $inner:tt, $inner_type:ident) => {
1326 pub fn $fn_name(&mut self) -> Option<&mut $inner_type> {
1327 match self {
1328 Self::$inner(inner) => Some(inner),
1329 _ => None,
1330 }
1331 }
1332 };
1333}
1334
1335impl DataBlock {
1337 as_type!(as_all_null, AllNull, AllNullDataBlock);
1338 as_type!(as_nullable, Nullable, NullableDataBlock);
1339 as_type!(as_fixed_width, FixedWidth, FixedWidthDataBlock);
1340 as_type!(as_fixed_size_list, FixedSizeList, FixedSizeListBlock);
1341 as_type!(as_variable_width, VariableWidth, VariableWidthBlock);
1342 as_type!(as_struct, Struct, StructDataBlock);
1343 as_type!(as_dictionary, Dictionary, DictionaryDataBlock);
1344 as_type_ref!(as_all_null_ref, AllNull, AllNullDataBlock);
1345 as_type_ref!(as_nullable_ref, Nullable, NullableDataBlock);
1346 as_type_ref!(as_fixed_width_ref, FixedWidth, FixedWidthDataBlock);
1347 as_type_ref!(as_fixed_size_list_ref, FixedSizeList, FixedSizeListBlock);
1348 as_type_ref!(as_variable_width_ref, VariableWidth, VariableWidthBlock);
1349 as_type_ref!(as_struct_ref, Struct, StructDataBlock);
1350 as_type_ref!(as_dictionary_ref, Dictionary, DictionaryDataBlock);
1351 as_type_ref_mut!(as_all_null_ref_mut, AllNull, AllNullDataBlock);
1352 as_type_ref_mut!(as_nullable_ref_mut, Nullable, NullableDataBlock);
1353 as_type_ref_mut!(as_fixed_width_ref_mut, FixedWidth, FixedWidthDataBlock);
1354 as_type_ref_mut!(
1355 as_fixed_size_list_ref_mut,
1356 FixedSizeList,
1357 FixedSizeListBlock
1358 );
1359 as_type_ref_mut!(as_variable_width_ref_mut, VariableWidth, VariableWidthBlock);
1360 as_type_ref_mut!(as_struct_ref_mut, Struct, StructDataBlock);
1361 as_type_ref_mut!(as_dictionary_ref_mut, Dictionary, DictionaryDataBlock);
1362}
1363
1364fn get_byte_range<T: ArrowNativeType>(offsets: &mut LanceBuffer) -> Range<usize> {
1367 let offsets = offsets.borrow_to_typed_slice::<T>();
1368 if offsets.as_ref().is_empty() {
1369 0..0
1370 } else {
1371 offsets.as_ref().first().unwrap().as_usize()..offsets.as_ref().last().unwrap().as_usize()
1372 }
1373}
1374
1375fn stitch_offsets<T: ArrowNativeType + std::ops::Add<Output = T> + std::ops::Sub<Output = T>>(
1381 offsets: Vec<LanceBuffer>,
1382) -> (LanceBuffer, Vec<Range<usize>>) {
1383 if offsets.is_empty() {
1384 return (LanceBuffer::empty(), Vec::default());
1385 }
1386 let len = offsets.iter().map(|b| b.len()).sum::<usize>();
1387 let mut dest = Vec::with_capacity(len);
1391 let mut byte_ranges = Vec::with_capacity(offsets.len());
1392
1393 dest.push(T::from_usize(0).unwrap());
1395
1396 for mut o in offsets.into_iter() {
1397 if !o.is_empty() {
1398 let last_offset = *dest.last().unwrap();
1399 let o = o.borrow_to_typed_slice::<T>();
1400 let start = *o.as_ref().first().unwrap();
1401 dest.extend(o.as_ref()[1..].iter().map(|&x| x + last_offset - start));
1415 }
1416 byte_ranges.push(get_byte_range::<T>(&mut o));
1417 }
1418 (LanceBuffer::reinterpret_vec(dest), byte_ranges)
1419}
1420
1421fn arrow_binary_to_data_block(
1422 arrays: &[ArrayRef],
1423 num_values: u64,
1424 bits_per_offset: u8,
1425) -> DataBlock {
1426 let data_vec = arrays.iter().map(|arr| arr.to_data()).collect::<Vec<_>>();
1427 let bytes_per_offset = bits_per_offset as usize / 8;
1428 let offsets = data_vec
1429 .iter()
1430 .map(|d| {
1431 LanceBuffer::from(
1432 d.buffers()[0].slice_with_length(d.offset(), (d.len() + 1) * bytes_per_offset),
1433 )
1434 })
1435 .collect::<Vec<_>>();
1436 let (offsets, data_ranges) = if bits_per_offset == 32 {
1437 stitch_offsets::<i32>(offsets)
1438 } else {
1439 stitch_offsets::<i64>(offsets)
1440 };
1441 let data = data_vec
1442 .iter()
1443 .zip(data_ranges)
1444 .map(|(d, byte_range)| {
1445 LanceBuffer::from(
1446 d.buffers()[1]
1447 .slice_with_length(byte_range.start, byte_range.end - byte_range.start),
1448 )
1449 })
1450 .collect::<Vec<_>>();
1451 let data = LanceBuffer::concat_into_one(data);
1452 DataBlock::VariableWidth(VariableWidthBlock {
1453 data,
1454 offsets,
1455 bits_per_offset,
1456 num_values,
1457 block_info: BlockInfo::new(),
1458 })
1459}
1460
1461fn encode_flat_data(arrays: &[ArrayRef], num_values: u64) -> LanceBuffer {
1462 let bytes_per_value = arrays[0].data_type().byte_width();
1463 let mut buffer = Vec::with_capacity(num_values as usize * bytes_per_value);
1464 for arr in arrays {
1465 let data = arr.to_data();
1466 buffer.extend_from_slice(data.buffers()[0].as_slice());
1467 }
1468 LanceBuffer::from(buffer)
1469}
1470
1471fn do_encode_bitmap_data(bitmaps: &[BooleanBuffer], num_values: u64) -> LanceBuffer {
1472 let mut builder = BooleanBufferBuilder::new(num_values as usize);
1473
1474 for buf in bitmaps {
1475 builder.append_buffer(buf);
1476 }
1477
1478 let buffer = builder.finish().into_inner();
1479 LanceBuffer::from(buffer)
1480}
1481
1482fn encode_bitmap_data(arrays: &[ArrayRef], num_values: u64) -> LanceBuffer {
1483 let bitmaps = arrays
1484 .iter()
1485 .map(|arr| arr.as_boolean().values().clone())
1486 .collect::<Vec<_>>();
1487 do_encode_bitmap_data(&bitmaps, num_values)
1488}
1489
1490fn concat_dict_arrays(arrays: &[ArrayRef]) -> ArrayRef {
1493 let value_type = arrays[0].as_any_dictionary().values().data_type();
1494 let array_refs = arrays.iter().map(|arr| arr.as_ref()).collect::<Vec<_>>();
1495 match arrow_select::concat::concat(&array_refs) {
1496 Ok(array) => array,
1497 Err(arrow_schema::ArrowError::DictionaryKeyOverflowError) => {
1498 let upscaled = array_refs
1500 .iter()
1501 .map(|arr| {
1502 match arrow_cast::cast(
1503 *arr,
1504 &DataType::Dictionary(
1505 Box::new(DataType::UInt32),
1506 Box::new(value_type.clone()),
1507 ),
1508 ) {
1509 Ok(arr) => arr,
1510 Err(arrow_schema::ArrowError::DictionaryKeyOverflowError) => {
1511 unimplemented!("Dictionary arrays with more than 2^32 unique values")
1513 }
1514 err => err.unwrap(),
1515 }
1516 })
1517 .collect::<Vec<_>>();
1518 let array_refs = upscaled.iter().map(|arr| arr.as_ref()).collect::<Vec<_>>();
1519 match arrow_select::concat::concat(&array_refs) {
1521 Ok(array) => array,
1522 Err(arrow_schema::ArrowError::DictionaryKeyOverflowError) => {
1523 unimplemented!("Dictionary arrays with more than 2^32 unique values")
1524 }
1525 err => err.unwrap(),
1526 }
1527 }
1528 err => err.unwrap(),
1530 }
1531}
1532
1533fn max_index_val(index_type: &DataType) -> u64 {
1534 match index_type {
1535 DataType::Int8 => i8::MAX as u64,
1536 DataType::Int16 => i16::MAX as u64,
1537 DataType::Int32 => i32::MAX as u64,
1538 DataType::Int64 => i64::MAX as u64,
1539 DataType::UInt8 => u8::MAX as u64,
1540 DataType::UInt16 => u16::MAX as u64,
1541 DataType::UInt32 => u32::MAX as u64,
1542 DataType::UInt64 => u64::MAX,
1543 _ => panic!("Invalid dictionary index type"),
1544 }
1545}
1546
1547fn arrow_dictionary_to_data_block(arrays: &[ArrayRef], validity: Option<NullBuffer>) -> DataBlock {
1566 let array = concat_dict_arrays(arrays);
1567 let array_dict = array.as_any_dictionary();
1568 let mut indices = array_dict.keys();
1569 let num_values = indices.len() as u64;
1570 let mut values = array_dict.values().clone();
1571 let mut upcast = None;
1573
1574 let indices_block = if let Some(validity) = validity {
1578 let mut first_invalid_index = None;
1582 if let Some(values_validity) = values.nulls() {
1583 first_invalid_index = (!values_validity.inner()).set_indices().next();
1584 }
1585 let first_invalid_index = first_invalid_index.unwrap_or_else(|| {
1586 let null_arr = new_null_array(values.data_type(), 1);
1587 values = arrow_select::concat::concat(&[values.as_ref(), null_arr.as_ref()]).unwrap();
1588 let null_index = values.len() - 1;
1589 let max_index_val = max_index_val(indices.data_type());
1590 if null_index as u64 > max_index_val {
1591 if max_index_val >= u32::MAX as u64 {
1593 unimplemented!("Dictionary arrays with 2^32 unique value (or more) and a null")
1594 }
1595 upcast = Some(arrow_cast::cast(indices, &DataType::UInt32).unwrap());
1596 indices = upcast.as_ref().unwrap();
1597 }
1598 null_index
1599 });
1600 let null_index_arr = arrow_cast::cast(
1602 &UInt64Array::from(vec![first_invalid_index as u64]),
1603 indices.data_type(),
1604 )
1605 .unwrap();
1606
1607 let bytes_per_index = indices.data_type().byte_width();
1608 let bits_per_index = bytes_per_index as u64 * 8;
1609
1610 let null_index_arr = null_index_arr.into_data();
1611 let null_index_bytes = &null_index_arr.buffers()[0];
1612 let mut indices_bytes = indices.to_data().buffers()[0].to_vec();
1614 for invalid_idx in (!validity.inner()).set_indices() {
1615 indices_bytes[invalid_idx * bytes_per_index..(invalid_idx + 1) * bytes_per_index]
1616 .copy_from_slice(null_index_bytes.as_slice());
1617 }
1618 FixedWidthDataBlock {
1619 data: LanceBuffer::from(indices_bytes),
1620 bits_per_value: bits_per_index,
1621 num_values,
1622 block_info: BlockInfo::new(),
1623 }
1624 } else {
1625 FixedWidthDataBlock {
1626 data: LanceBuffer::from(indices.to_data().buffers()[0].clone()),
1627 bits_per_value: indices.data_type().byte_width() as u64 * 8,
1628 num_values,
1629 block_info: BlockInfo::new(),
1630 }
1631 };
1632
1633 let items = DataBlock::from(values);
1634 DataBlock::Dictionary(DictionaryDataBlock {
1635 indices: indices_block,
1636 dictionary: Box::new(items),
1637 })
1638}
1639
1640enum Nullability {
1641 None,
1642 All,
1643 Some(NullBuffer),
1644}
1645
1646impl Nullability {
1647 fn to_option(&self) -> Option<NullBuffer> {
1648 match self {
1649 Self::Some(nulls) => Some(nulls.clone()),
1650 _ => None,
1651 }
1652 }
1653}
1654
1655fn extract_nulls(arrays: &[ArrayRef], num_values: u64) -> Nullability {
1656 let mut has_nulls = false;
1657 let nulls_and_lens = arrays
1658 .iter()
1659 .map(|arr| {
1660 let nulls = arr.logical_nulls();
1661 has_nulls |= nulls.is_some();
1662 (nulls, arr.len())
1663 })
1664 .collect::<Vec<_>>();
1665 if !has_nulls {
1666 return Nullability::None;
1667 }
1668 let mut builder = BooleanBufferBuilder::new(num_values as usize);
1669 let mut num_nulls = 0;
1670 for (null, len) in nulls_and_lens {
1671 if let Some(null) = null {
1672 num_nulls += null.null_count();
1673 builder.append_buffer(&null.into_inner());
1674 } else {
1675 builder.append_n(len, true);
1676 }
1677 }
1678 if num_nulls == num_values as usize {
1679 Nullability::All
1680 } else {
1681 Nullability::Some(NullBuffer::new(builder.finish()))
1682 }
1683}
1684
1685impl DataBlock {
1686 pub fn from_arrays(arrays: &[ArrayRef], num_values: u64) -> Self {
1687 if arrays.is_empty() || num_values == 0 {
1688 return Self::AllNull(AllNullDataBlock { num_values: 0 });
1689 }
1690
1691 let data_type = arrays[0].data_type();
1692 let nulls = extract_nulls(arrays, num_values);
1693
1694 if let Nullability::All = nulls {
1695 return Self::AllNull(AllNullDataBlock { num_values });
1696 }
1697
1698 let mut encoded = match data_type {
1699 DataType::Binary | DataType::Utf8 => arrow_binary_to_data_block(arrays, num_values, 32),
1700 DataType::Utf8View => {
1702 let casted: Vec<ArrayRef> = arrays
1703 .iter()
1704 .map(|a| {
1705 arrow_cast::cast(a.as_ref(), &DataType::Utf8)
1706 .expect("Utf8View to Utf8 cast is always valid")
1707 })
1708 .collect();
1709 arrow_binary_to_data_block(&casted, num_values, 32)
1710 }
1711 DataType::BinaryView => {
1712 let casted: Vec<ArrayRef> = arrays
1713 .iter()
1714 .map(|a| {
1715 arrow_cast::cast(a.as_ref(), &DataType::Binary)
1716 .expect("BinaryView to Binary cast is always valid")
1717 })
1718 .collect();
1719 arrow_binary_to_data_block(&casted, num_values, 32)
1720 }
1721 DataType::LargeBinary | DataType::LargeUtf8 => {
1722 arrow_binary_to_data_block(arrays, num_values, 64)
1723 }
1724 DataType::Boolean => {
1725 let data = encode_bitmap_data(arrays, num_values);
1726 Self::FixedWidth(FixedWidthDataBlock {
1727 data,
1728 bits_per_value: 1,
1729 num_values,
1730 block_info: BlockInfo::new(),
1731 })
1732 }
1733 DataType::Date32
1734 | DataType::Date64
1735 | DataType::Decimal32(_, _)
1736 | DataType::Decimal64(_, _)
1737 | DataType::Decimal128(_, _)
1738 | DataType::Decimal256(_, _)
1739 | DataType::Duration(_)
1740 | DataType::FixedSizeBinary(_)
1741 | DataType::Float16
1742 | DataType::Float32
1743 | DataType::Float64
1744 | DataType::Int16
1745 | DataType::Int32
1746 | DataType::Int64
1747 | DataType::Int8
1748 | DataType::Interval(_)
1749 | DataType::Time32(_)
1750 | DataType::Time64(_)
1751 | DataType::Timestamp(_, _)
1752 | DataType::UInt16
1753 | DataType::UInt32
1754 | DataType::UInt64
1755 | DataType::UInt8 => {
1756 let data = encode_flat_data(arrays, num_values);
1757 Self::FixedWidth(FixedWidthDataBlock {
1758 data,
1759 bits_per_value: data_type.byte_width() as u64 * 8,
1760 num_values,
1761 block_info: BlockInfo::new(),
1762 })
1763 }
1764 DataType::Null => Self::AllNull(AllNullDataBlock { num_values }),
1765 DataType::Dictionary(_, _) => arrow_dictionary_to_data_block(arrays, nulls.to_option()),
1766 DataType::Struct(fields) => {
1767 let structs = arrays.iter().map(|arr| arr.as_struct()).collect::<Vec<_>>();
1768 let mut children = Vec::with_capacity(fields.len());
1769 for child_idx in 0..fields.len() {
1770 let child_vec = structs
1771 .iter()
1772 .map(|s| s.column(child_idx).clone())
1773 .collect::<Vec<_>>();
1774 children.push(Self::from_arrays(&child_vec, num_values));
1775 }
1776
1777 let validity = match &nulls {
1779 Nullability::None => None,
1780 Nullability::Some(null_buffer) => Some(null_buffer.clone()),
1781 Nullability::All => unreachable!("Should have returned AllNull earlier"),
1782 };
1783
1784 Self::Struct(StructDataBlock {
1785 children,
1786 block_info: BlockInfo::default(),
1787 validity,
1788 })
1789 }
1790 DataType::FixedSizeList(_, dim) => {
1791 let children = arrays
1792 .iter()
1793 .map(|arr| arr.as_fixed_size_list().values().clone())
1794 .collect::<Vec<_>>();
1795 let child_block = Self::from_arrays(&children, num_values * *dim as u64);
1796 Self::FixedSizeList(FixedSizeListBlock {
1797 child: Box::new(child_block),
1798 dimension: *dim as u64,
1799 })
1800 }
1801 DataType::LargeList(_)
1802 | DataType::List(_)
1803 | DataType::ListView(_)
1804 | DataType::LargeListView(_)
1805 | DataType::Map(_, _)
1806 | DataType::RunEndEncoded(_, _)
1807 | DataType::Union(_, _) => {
1808 panic!(
1809 "Field with data type {} cannot be converted to data block",
1810 data_type
1811 )
1812 }
1813 };
1814
1815 encoded.compute_stat();
1817
1818 if !matches!(data_type, DataType::Dictionary(_, _)) {
1819 match nulls {
1820 Nullability::None => encoded,
1821 Nullability::Some(nulls) => Self::Nullable(NullableDataBlock {
1822 data: Box::new(encoded),
1823 nulls: LanceBuffer::from(nulls.into_inner().into_inner()),
1824 block_info: BlockInfo::new(),
1825 }),
1826 _ => unreachable!(),
1827 }
1828 } else {
1829 encoded
1831 }
1832 }
1833
1834 pub fn from_array<T: Array + 'static>(array: T) -> Self {
1835 let num_values = array.len();
1836 Self::from_arrays(&[Arc::new(array)], num_values as u64)
1837 }
1838}
1839
1840impl From<ArrayRef> for DataBlock {
1841 fn from(array: ArrayRef) -> Self {
1842 let num_values = array.len() as u64;
1843 Self::from_arrays(&[array], num_values)
1844 }
1845}
1846
1847pub trait DataBlockBuilderImpl: std::fmt::Debug {
1848 fn append(&mut self, data_block: &DataBlock, selection: Range<u64>);
1849 fn finish(self: Box<Self>) -> DataBlock;
1850}
1851
1852#[derive(Debug)]
1853pub struct DataBlockBuilder {
1854 estimated_size_bytes: u64,
1855 builder: Option<Box<dyn DataBlockBuilderImpl>>,
1856}
1857
1858impl DataBlockBuilder {
1859 pub fn with_capacity_estimate(estimated_size_bytes: u64) -> Self {
1860 Self {
1861 estimated_size_bytes,
1862 builder: None,
1863 }
1864 }
1865
1866 fn get_builder(&mut self, block: &DataBlock) -> &mut dyn DataBlockBuilderImpl {
1867 if self.builder.is_none() {
1868 self.builder = Some(block.make_builder(self.estimated_size_bytes));
1869 }
1870 self.builder.as_mut().unwrap().as_mut()
1871 }
1872
1873 pub fn append(&mut self, data_block: &DataBlock, selection: Range<u64>) {
1874 self.get_builder(data_block).append(data_block, selection);
1875 }
1876
1877 pub fn finish(self) -> DataBlock {
1878 let builder = self.builder.expect("DataBlockBuilder didn't see any data");
1879 builder.finish()
1880 }
1881}
1882
1883#[cfg(test)]
1884mod tests {
1885 use std::sync::Arc;
1886
1887 use arrow_array::{
1888 ArrayRef, BinaryArray, BinaryViewArray, DictionaryArray, Int8Array, LargeBinaryArray,
1889 LargeStringArray, StringArray, StringViewArray, UInt8Array, UInt16Array, make_array,
1890 new_null_array,
1891 types::{Int8Type, Int32Type},
1892 };
1893 use arrow_buffer::{BooleanBuffer, NullBuffer};
1894
1895 use arrow_schema::{DataType, Field, Fields};
1896 use lance_core::Error;
1897 use lance_datagen::{ArrayGeneratorExt, DEFAULT_SEED, RowCount, array};
1898 use rand::SeedableRng;
1899 use rstest::rstest;
1900
1901 use crate::buffer::LanceBuffer;
1902
1903 use super::{
1904 AllNullDataBlock, BlockInfo, DataBlock, DictionaryDataBlock, FixedWidthDataBlock,
1905 VariableWidthBlock,
1906 };
1907
1908 use arrow_array::Array;
1909
1910 #[test]
1911 fn test_sliced_to_data_block() {
1912 let ints = UInt16Array::from(vec![0, 1, 2, 3, 4, 5, 6, 7, 8]);
1913 let ints = ints.slice(2, 4);
1914 let data = DataBlock::from_array(ints);
1915
1916 let fixed_data = data.as_fixed_width().unwrap();
1917 assert_eq!(fixed_data.num_values, 4);
1918 assert_eq!(fixed_data.data.len(), 8);
1919
1920 let nullable_ints =
1921 UInt16Array::from(vec![Some(0), None, Some(2), None, Some(4), None, Some(6)]);
1922 let nullable_ints = nullable_ints.slice(1, 3);
1923 let data = DataBlock::from_array(nullable_ints);
1924
1925 let nullable = data.as_nullable().unwrap();
1926 assert_eq!(nullable.nulls, LanceBuffer::from(vec![0b00000010]));
1927 }
1928
1929 #[test]
1930 fn test_string_to_data_block() {
1931 let strings1 = StringArray::from(vec![Some("hello"), None, Some("world")]);
1933 let strings2 = StringArray::from(vec![Some("a"), Some("b")]);
1934 let strings3 = StringArray::from(vec![Option::<&'static str>::None, None]);
1935
1936 let arrays = &[strings1, strings2, strings3]
1937 .iter()
1938 .map(|arr| Arc::new(arr.clone()) as ArrayRef)
1939 .collect::<Vec<_>>();
1940
1941 let block = DataBlock::from_arrays(arrays, 7);
1942
1943 assert_eq!(block.num_values(), 7);
1944 let block = block.as_nullable().unwrap();
1945
1946 assert_eq!(block.nulls, LanceBuffer::from(vec![0b00011101]));
1947
1948 let data = block.data.as_variable_width().unwrap();
1949 assert_eq!(
1950 data.offsets,
1951 LanceBuffer::reinterpret_vec(vec![0, 5, 5, 10, 11, 12, 12, 12])
1952 );
1953
1954 assert_eq!(data.data, LanceBuffer::copy_slice(b"helloworldab"));
1955
1956 let strings1 = StringArray::from(vec![Some("a"), Some("bc")]);
1958 let strings2 = StringArray::from(vec![Some("def")]);
1959
1960 let arrays = &[strings1, strings2]
1961 .iter()
1962 .map(|arr| Arc::new(arr.clone()) as ArrayRef)
1963 .collect::<Vec<_>>();
1964
1965 let block = DataBlock::from_arrays(arrays, 3);
1966
1967 assert_eq!(block.num_values(), 3);
1968 let data = block.as_variable_width().unwrap();
1970 assert_eq!(data.offsets, LanceBuffer::reinterpret_vec(vec![0, 1, 3, 6]));
1971 assert_eq!(data.data, LanceBuffer::copy_slice(b"abcdef"));
1972 }
1973
1974 #[test]
1975 fn test_string_view_to_data_block() {
1976 let views1 = StringViewArray::from(vec![Some("hello"), None, Some("world")]);
1977 let views2 = StringViewArray::from(vec![Some("a"), Some("b")]);
1978 let views3 = StringViewArray::from(vec![Option::<&'static str>::None, None]);
1979
1980 let arrays = &[views1, views2, views3]
1981 .iter()
1982 .map(|arr| Arc::new(arr.clone()) as ArrayRef)
1983 .collect::<Vec<_>>();
1984
1985 let block = DataBlock::from_arrays(arrays, 7);
1986
1987 assert_eq!(block.num_values(), 7);
1988 let block = block.as_nullable().unwrap();
1989 assert_eq!(block.nulls, LanceBuffer::from(vec![0b00011101]));
1990 let data = block.data.as_variable_width().unwrap();
1991 assert_eq!(
1992 data.offsets,
1993 LanceBuffer::reinterpret_vec(vec![0, 5, 5, 10, 11, 12, 12, 12])
1994 );
1995 assert_eq!(data.data, LanceBuffer::copy_slice(b"helloworldab"));
1996
1997 let views1 = StringViewArray::from(vec![Some("a"), Some("bc")]);
1998 let views2 = StringViewArray::from(vec![Some("def")]);
1999
2000 let arrays = &[views1, views2]
2001 .iter()
2002 .map(|arr| Arc::new(arr.clone()) as ArrayRef)
2003 .collect::<Vec<_>>();
2004
2005 let block = DataBlock::from_arrays(arrays, 3);
2006
2007 assert_eq!(block.num_values(), 3);
2008 let data = block.as_variable_width().unwrap();
2009 assert_eq!(data.offsets, LanceBuffer::reinterpret_vec(vec![0, 1, 3, 6]));
2010 assert_eq!(data.data, LanceBuffer::copy_slice(b"abcdef"));
2011 }
2012
2013 #[test]
2014 fn test_binary_view_to_data_block() {
2015 let arr: ArrayRef = Arc::new(BinaryViewArray::from(vec![
2016 Some(b"foo".as_slice()),
2017 None,
2018 Some(b"bar".as_slice()),
2019 ]));
2020 let block = DataBlock::from_arrays(&[arr], 3);
2021 let block = block.as_nullable().unwrap();
2022 let data = block.data.as_variable_width().unwrap();
2023 assert_eq!(data.data, LanceBuffer::copy_slice(b"foobar"));
2024 }
2025
2026 #[test]
2027 fn test_string_sliced() {
2028 let check = |arr: Vec<StringArray>, expected_off: Vec<i32>, expected_data: &[u8]| {
2029 let arrs = arr
2030 .into_iter()
2031 .map(|a| Arc::new(a) as ArrayRef)
2032 .collect::<Vec<_>>();
2033 let num_rows = arrs.iter().map(|a| a.len()).sum::<usize>() as u64;
2034 let data = DataBlock::from_arrays(&arrs, num_rows);
2035
2036 assert_eq!(data.num_values(), num_rows);
2037
2038 let data = data.as_variable_width().unwrap();
2039 assert_eq!(data.offsets, LanceBuffer::reinterpret_vec(expected_off));
2040 assert_eq!(data.data, LanceBuffer::copy_slice(expected_data));
2041 };
2042
2043 let string = StringArray::from(vec![Some("hello"), Some("world")]);
2044 check(vec![string.slice(1, 1)], vec![0, 5], b"world");
2045 check(vec![string.slice(0, 1)], vec![0, 5], b"hello");
2046 check(
2047 vec![string.slice(0, 1), string.slice(1, 1)],
2048 vec![0, 5, 10],
2049 b"helloworld",
2050 );
2051
2052 let string2 = StringArray::from(vec![Some("foo"), Some("bar")]);
2053 check(
2054 vec![string.slice(0, 1), string2.slice(0, 1)],
2055 vec![0, 5, 8],
2056 b"hellofoo",
2057 );
2058 }
2059
2060 #[test]
2061 fn test_large() {
2062 let arr = LargeBinaryArray::from_vec(vec![b"hello", b"world"]);
2063 let data = DataBlock::from_array(arr);
2064
2065 assert_eq!(data.num_values(), 2);
2066 let data = data.as_variable_width().unwrap();
2067 assert_eq!(data.bits_per_offset, 64);
2068 assert_eq!(data.num_values, 2);
2069 assert_eq!(data.data, LanceBuffer::copy_slice(b"helloworld"));
2070 assert_eq!(
2071 data.offsets,
2072 LanceBuffer::reinterpret_vec(vec![0_u64, 5, 10])
2073 );
2074 }
2075
2076 #[test]
2077 fn test_dictionary_indices_normalized() {
2078 let arr1 = DictionaryArray::<Int8Type>::from_iter([Some("a"), Some("a"), Some("b")]);
2079 let arr2 = DictionaryArray::<Int8Type>::from_iter([Some("b"), Some("c")]);
2080
2081 let data = DataBlock::from_arrays(&[Arc::new(arr1), Arc::new(arr2)], 5);
2082
2083 assert_eq!(data.num_values(), 5);
2084 let data = data.as_dictionary().unwrap();
2085 let indices = data.indices;
2086 assert_eq!(indices.bits_per_value, 8);
2087 assert_eq!(indices.num_values, 5);
2088 assert_eq!(
2089 indices.data,
2090 LanceBuffer::reinterpret_vec::<i8>(vec![0, 0, 1, 2, 3])
2094 );
2095
2096 let items = data.dictionary.as_variable_width().unwrap();
2097 assert_eq!(items.bits_per_offset, 32);
2098 assert_eq!(items.num_values, 4);
2099 assert_eq!(items.data, LanceBuffer::copy_slice(b"abbc"));
2100 assert_eq!(
2101 items.offsets,
2102 LanceBuffer::reinterpret_vec(vec![0, 1, 2, 3, 4],)
2103 );
2104 }
2105
2106 #[test]
2107 fn test_dictionary_nulls() {
2108 let arr1 = DictionaryArray::<Int8Type>::from_iter([None, Some("a"), Some("b")]);
2112 let arr2 = DictionaryArray::<Int8Type>::from_iter([Some("c"), None]);
2113
2114 let data = DataBlock::from_arrays(&[Arc::new(arr1), Arc::new(arr2)], 5);
2115
2116 let check_common = |data: DataBlock| {
2117 assert_eq!(data.num_values(), 5);
2118 let dict = data.as_dictionary().unwrap();
2119
2120 let nullable_items = dict.dictionary.as_nullable().unwrap();
2121 assert_eq!(nullable_items.nulls, LanceBuffer::from(vec![0b00000111]));
2122 assert_eq!(nullable_items.data.num_values(), 4);
2123
2124 let items = nullable_items.data.as_variable_width().unwrap();
2125 assert_eq!(items.bits_per_offset, 32);
2126 assert_eq!(items.num_values, 4);
2127 assert_eq!(items.data, LanceBuffer::copy_slice(b"abc"));
2128 assert_eq!(
2129 items.offsets,
2130 LanceBuffer::reinterpret_vec(vec![0, 1, 2, 3, 3],)
2131 );
2132
2133 let indices = dict.indices;
2134 assert_eq!(indices.bits_per_value, 8);
2135 assert_eq!(indices.num_values, 5);
2136 assert_eq!(
2137 indices.data,
2138 LanceBuffer::reinterpret_vec::<i8>(vec![3, 0, 1, 2, 3])
2139 );
2140 };
2141 check_common(data);
2142
2143 let items = StringArray::from(vec![Some("a"), Some("b"), Some("c"), None]);
2145 let indices = Int8Array::from(vec![Some(3), Some(0), Some(1), Some(2), Some(3)]);
2146 let dict = DictionaryArray::new(indices, Arc::new(items));
2147
2148 let data = DataBlock::from_array(dict);
2149
2150 check_common(data);
2151 }
2152
2153 #[test]
2154 fn test_dictionary_cannot_add_null() {
2155 let items = StringArray::from(
2157 (0..256)
2158 .map(|i| Some(String::from_utf8(vec![0; i]).unwrap()))
2159 .collect::<Vec<_>>(),
2160 );
2161 let indices = UInt8Array::from(
2163 (0..=256)
2164 .map(|i| if i == 256 { None } else { Some(i as u8) })
2165 .collect::<Vec<_>>(),
2166 );
2167 let dict = DictionaryArray::new(indices, Arc::new(items));
2170 let data = DataBlock::from_array(dict);
2171
2172 assert_eq!(data.num_values(), 257);
2173
2174 let dict = data.as_dictionary().unwrap();
2175
2176 assert_eq!(dict.indices.bits_per_value, 32);
2177 assert_eq!(
2178 dict.indices.data,
2179 LanceBuffer::reinterpret_vec((0_u32..257).collect::<Vec<_>>())
2180 );
2181
2182 let nullable_items = dict.dictionary.as_nullable().unwrap();
2183 let null_buffer = NullBuffer::new(BooleanBuffer::new(
2184 nullable_items.nulls.into_buffer(),
2185 0,
2186 257,
2187 ));
2188 for i in 0..256 {
2189 assert!(!null_buffer.is_null(i));
2190 }
2191 assert!(null_buffer.is_null(256));
2192
2193 assert_eq!(
2194 nullable_items.data.as_variable_width().unwrap().data.len(),
2195 32640
2196 );
2197 }
2198
2199 #[test]
2200 fn test_all_null() {
2201 for data_type in [
2202 DataType::UInt32,
2203 DataType::FixedSizeBinary(2),
2204 DataType::List(Arc::new(Field::new("item", DataType::UInt32, true))),
2205 DataType::Struct(Fields::from(vec![Field::new("a", DataType::UInt32, true)])),
2206 ] {
2207 let block = DataBlock::AllNull(AllNullDataBlock { num_values: 10 });
2208 let arr = block.into_arrow(data_type.clone(), true).unwrap();
2209 let arr = make_array(arr);
2210 let expected = new_null_array(&data_type, 10);
2211 assert_eq!(&arr, &expected);
2212 }
2213 }
2214
2215 #[test]
2216 fn test_dictionary_cannot_concatenate() {
2217 let items = StringArray::from(
2219 (0..256)
2220 .map(|i| Some(String::from_utf8(vec![0; i]).unwrap()))
2221 .collect::<Vec<_>>(),
2222 );
2223 let other_items = StringArray::from(
2225 (0..256)
2226 .map(|i| Some(String::from_utf8(vec![1; i + 1]).unwrap()))
2227 .collect::<Vec<_>>(),
2228 );
2229 let indices = UInt8Array::from_iter_values(0..=255);
2230 let dict1 = DictionaryArray::new(indices.clone(), Arc::new(items));
2231 let dict2 = DictionaryArray::new(indices, Arc::new(other_items));
2232 let data = DataBlock::from_arrays(&[Arc::new(dict1), Arc::new(dict2)], 512);
2233 assert_eq!(data.num_values(), 512);
2234
2235 let dict = data.as_dictionary().unwrap();
2236
2237 assert_eq!(dict.indices.bits_per_value, 32);
2238 assert_eq!(
2239 dict.indices.data,
2240 LanceBuffer::reinterpret_vec::<u32>((0..512).collect::<Vec<_>>())
2241 );
2242 assert_eq!(
2244 dict.dictionary.as_variable_width().unwrap().data.len(),
2245 65536
2246 );
2247 }
2248
2249 #[test]
2250 fn test_data_size() {
2251 let mut rng = rand_xoshiro::Xoshiro256PlusPlus::seed_from_u64(DEFAULT_SEED.0);
2252 let mut genn = array::rand::<Int32Type>().with_nulls(&[false, false, false]);
2254
2255 let arr = genn.generate(RowCount::from(3), &mut rng).unwrap();
2256 let block = DataBlock::from_array(arr.clone());
2257 assert!(block.data_size() == arr.get_buffer_memory_size() as u64);
2258
2259 let arr = genn.generate(RowCount::from(400), &mut rng).unwrap();
2260 let block = DataBlock::from_array(arr.clone());
2261 assert!(block.data_size() == arr.get_buffer_memory_size() as u64);
2262
2263 let mut genn = array::rand::<Int32Type>().with_nulls(&[false, true, false]);
2265 let arr = genn.generate(RowCount::from(3), &mut rng).unwrap();
2266 let block = DataBlock::from_array(arr.clone());
2267
2268 let array_data = arr.to_data();
2269 let total_buffer_size: usize = array_data.buffers().iter().map(|buffer| buffer.len()).sum();
2270 let array_nulls_size_in_bytes = arr.nulls().unwrap().len().div_ceil(8);
2272 assert!(block.data_size() == (total_buffer_size + array_nulls_size_in_bytes) as u64);
2273
2274 let arr = genn.generate(RowCount::from(400), &mut rng).unwrap();
2275 let block = DataBlock::from_array(arr.clone());
2276
2277 let array_data = arr.to_data();
2278 let total_buffer_size: usize = array_data.buffers().iter().map(|buffer| buffer.len()).sum();
2279 let array_nulls_size_in_bytes = arr.nulls().unwrap().len().div_ceil(8);
2280 assert!(block.data_size() == (total_buffer_size + array_nulls_size_in_bytes) as u64);
2281
2282 let mut genn = array::rand::<Int32Type>().with_nulls(&[true, true, false]);
2283 let arr = genn.generate(RowCount::from(3), &mut rng).unwrap();
2284 let block = DataBlock::from_array(arr.clone());
2285
2286 let array_data = arr.to_data();
2287 let total_buffer_size: usize = array_data.buffers().iter().map(|buffer| buffer.len()).sum();
2288 let array_nulls_size_in_bytes = arr.nulls().unwrap().len().div_ceil(8);
2289 assert!(block.data_size() == (total_buffer_size + array_nulls_size_in_bytes) as u64);
2290
2291 let arr = genn.generate(RowCount::from(400), &mut rng).unwrap();
2292 let block = DataBlock::from_array(arr.clone());
2293
2294 let array_data = arr.to_data();
2295 let total_buffer_size: usize = array_data.buffers().iter().map(|buffer| buffer.len()).sum();
2296 let array_nulls_size_in_bytes = arr.nulls().unwrap().len().div_ceil(8);
2297 assert!(block.data_size() == (total_buffer_size + array_nulls_size_in_bytes) as u64);
2298
2299 let mut genn = array::rand::<Int32Type>().with_nulls(&[false, true, false]);
2300 let arr1 = genn.generate(RowCount::from(3), &mut rng).unwrap();
2301 let arr2 = genn.generate(RowCount::from(3), &mut rng).unwrap();
2302 let arr3 = genn.generate(RowCount::from(3), &mut rng).unwrap();
2303 let block = DataBlock::from_arrays(&[arr1.clone(), arr2.clone(), arr3.clone()], 9);
2304
2305 let concatenated_array = arrow_select::concat::concat(&[
2306 &*Arc::new(arr1.clone()) as &dyn Array,
2307 &*Arc::new(arr2.clone()) as &dyn Array,
2308 &*Arc::new(arr3.clone()) as &dyn Array,
2309 ])
2310 .unwrap();
2311 let total_buffer_size: usize = concatenated_array
2312 .to_data()
2313 .buffers()
2314 .iter()
2315 .map(|buffer| buffer.len())
2316 .sum();
2317
2318 let total_nulls_size_in_bytes = concatenated_array.nulls().unwrap().len().div_ceil(8);
2319 assert!(block.data_size() == (total_buffer_size + total_nulls_size_in_bytes) as u64);
2320 }
2321
2322 #[test]
2323 fn variable_width_rejects_out_of_bounds_offsets_without_optional_validation() {
2324 let block = VariableWidthBlock {
2325 data: LanceBuffer::copy_slice(b"alphabetagamma"),
2326 offsets: LanceBuffer::reinterpret_vec(vec![0_i32, 5, 9, 100_000]),
2327 bits_per_offset: 32,
2328 num_values: 3,
2329 block_info: BlockInfo::new(),
2330 };
2331
2332 let error = block
2333 .into_arrow(DataType::Binary, false)
2334 .expect_err("out-of-bounds offsets must be rejected");
2335 assert!(
2336 matches!(error, Error::CorruptFile { .. }),
2337 "expected CorruptFile, got: {error}"
2338 );
2339 let message = error.to_string();
2340 assert!(
2341 message.contains("100000") && message.contains("data buffer size: 14 bytes"),
2342 "error must report the offending offset and the data buffer size: {message}"
2343 );
2344 }
2345
2346 #[rstest]
2347 #[case::binary_i32_tail_out_of_bounds(
2348 DataType::Binary,
2349 LanceBuffer::reinterpret_vec(vec![0_i32, 5, 9, 100_000]),
2350 32,
2351 3,
2352 b"alphabetagamma".as_slice()
2353 )]
2354 #[case::utf8_i32_tail_out_of_bounds(
2355 DataType::Utf8,
2356 LanceBuffer::reinterpret_vec(vec![0_i32, 5, 9, 100_000]),
2357 32,
2358 3,
2359 b"alphabetagamma".as_slice()
2360 )]
2361 #[case::large_binary_i64_tail_out_of_bounds(
2362 DataType::LargeBinary,
2363 LanceBuffer::reinterpret_vec(vec![0_i64, 5, 9, 100_000]),
2364 64,
2365 3,
2366 b"alphabetagamma".as_slice()
2367 )]
2368 #[case::large_utf8_i64_tail_out_of_bounds(
2369 DataType::LargeUtf8,
2370 LanceBuffer::reinterpret_vec(vec![0_i64, 5, 9, 100_000]),
2371 64,
2372 3,
2373 b"alphabetagamma".as_slice()
2374 )]
2375 #[case::binary_negative_offset(
2376 DataType::Binary,
2377 LanceBuffer::reinterpret_vec(vec![0_i32, -1, 9, 14]),
2378 32,
2379 3,
2380 b"alphabetagamma".as_slice()
2381 )]
2382 #[case::binary_non_monotonic_offsets(
2383 DataType::Binary,
2384 LanceBuffer::reinterpret_vec(vec![0_i32, 9, 5, 14]),
2385 32,
2386 3,
2387 b"alphabetagamma".as_slice()
2388 )]
2389 #[case::binary_interior_offset_out_of_bounds(
2390 DataType::Binary,
2391 LanceBuffer::reinterpret_vec(vec![0_i32, 100_000, 100_000, 14]),
2392 32,
2393 3,
2394 b"alphabetagamma".as_slice()
2395 )]
2396 #[case::binary_offsets_buffer_too_short(
2397 DataType::Binary,
2398 LanceBuffer::reinterpret_vec(vec![0_i32, 5, 9]),
2399 32,
2400 3,
2401 b"alphabetagamma".as_slice()
2402 )]
2403 #[case::utf8_invalid_byte_sequence(
2404 DataType::Utf8,
2405 LanceBuffer::reinterpret_vec(vec![0_i32, 1, 2, 3]),
2406 32,
2407 3,
2408 &[b'a', 0xFF, b'b']
2409 )]
2410 #[case::utf8_offset_splits_multibyte_char(
2411 DataType::Utf8,
2412 LanceBuffer::reinterpret_vec(vec![0_i32, 1, 2]),
2413 32,
2414 2,
2415 "é".as_bytes()
2416 )]
2417 #[case::large_utf8_invalid_byte_sequence(
2418 DataType::LargeUtf8,
2419 LanceBuffer::reinterpret_vec(vec![0_i64, 1, 2, 3]),
2420 64,
2421 3,
2422 &[b'a', 0xFF, b'b']
2423 )]
2424 fn variable_width_rejects_malformed_layout(
2425 #[case] data_type: DataType,
2426 #[case] offsets: LanceBuffer,
2427 #[case] bits_per_offset: u8,
2428 #[case] num_values: u64,
2429 #[case] data: &[u8],
2430 ) {
2431 let block = VariableWidthBlock {
2432 data: LanceBuffer::copy_slice(data),
2433 offsets,
2434 bits_per_offset,
2435 num_values,
2436 block_info: BlockInfo::new(),
2437 };
2438
2439 for validate in [false, true] {
2443 let error = DataBlock::VariableWidth(block.clone())
2444 .into_arrow(data_type.clone(), validate)
2445 .expect_err("malformed variable-width layout must be rejected");
2446 assert!(
2447 matches!(error, Error::CorruptFile { .. }),
2448 "expected CorruptFile with validate={validate}, got: {error}"
2449 );
2450 }
2451 }
2452
2453 #[test]
2454 fn dictionary_rejects_malformed_variable_width_values_without_optional_validation() {
2455 let values = VariableWidthBlock {
2456 data: LanceBuffer::copy_slice(b"alphabetagamma"),
2457 offsets: LanceBuffer::reinterpret_vec(vec![0_i32, 5, 9, 100_000]),
2458 bits_per_offset: 32,
2459 num_values: 3,
2460 block_info: BlockInfo::new(),
2461 };
2462 let dictionary = DataBlock::Dictionary(DictionaryDataBlock {
2463 indices: FixedWidthDataBlock {
2464 data: LanceBuffer::reinterpret_vec(vec![0_i32, 1, 2]),
2465 bits_per_value: 32,
2466 num_values: 3,
2467 block_info: BlockInfo::new(),
2468 },
2469 dictionary: Box::new(DataBlock::VariableWidth(values)),
2470 });
2471
2472 let data_type = DataType::Dictionary(Box::new(DataType::Int32), Box::new(DataType::Binary));
2473 let error = dictionary
2474 .into_arrow(data_type, false)
2475 .expect_err("dictionary with out-of-bounds value offsets must be rejected");
2476 assert!(
2477 matches!(error, Error::CorruptFile { .. }),
2478 "expected CorruptFile, got: {error}"
2479 );
2480 }
2481
2482 #[rstest]
2483 #[case::binary(Arc::new(BinaryArray::from_vec(vec![b"alpha", b"", b"gamma"])) as ArrayRef)]
2484 #[case::large_binary(
2485 Arc::new(LargeBinaryArray::from_vec(vec![b"alpha", b"", b"gamma"])) as ArrayRef
2486 )]
2487 #[case::utf8(Arc::new(StringArray::from(vec!["héllo", "", "world"])) as ArrayRef)]
2488 #[case::large_utf8(Arc::new(LargeStringArray::from(vec!["héllo", "", "world"])) as ArrayRef)]
2489 fn variable_width_valid_data_survives_mandatory_validation(#[case] array: ArrayRef) {
2490 let block = DataBlock::from_array(array.clone());
2491 for validate in [false, true] {
2492 let round_tripped = make_array(
2493 block
2494 .clone()
2495 .into_arrow(array.data_type().clone(), validate)
2496 .unwrap(),
2497 );
2498 assert_eq!(&round_tripped, &array);
2499 }
2500 }
2501}