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