1mod stream;
28
29pub use stream::*;
30
31use flatbuffers::{VectorIter, VerifierOptions};
32use std::collections::{HashMap, VecDeque};
33use std::fmt;
34use std::io::{BufReader, Read, Seek, SeekFrom};
35use std::sync::Arc;
36
37use arrow_array::*;
38use arrow_buffer::{ArrowNativeType, BooleanBuffer, Buffer, MutableBuffer, ScalarBuffer};
39use arrow_data::{ArrayData, ArrayDataBuilder, UnsafeFlag};
40use arrow_schema::*;
41
42use crate::compression::CompressionCodec;
43use crate::{Block, FieldNode, Message, MetadataVersion, CONTINUATION_MARKER};
44use DataType::*;
45
46fn read_buffer(
56 buf: &crate::Buffer,
57 a_data: &Buffer,
58 compression_codec: Option<CompressionCodec>,
59) -> Result<Buffer, ArrowError> {
60 let start_offset = buf.offset() as usize;
61 let buf_data = a_data.slice_with_length(start_offset, buf.length() as usize);
62 match (buf_data.is_empty(), compression_codec) {
64 (true, _) | (_, None) => Ok(buf_data),
65 (false, Some(decompressor)) => decompressor.decompress_to_buffer(&buf_data),
66 }
67}
68impl RecordBatchDecoder<'_> {
69 fn create_array(
82 &mut self,
83 field: &Field,
84 variadic_counts: &mut VecDeque<i64>,
85 ) -> Result<ArrayRef, ArrowError> {
86 let data_type = field.data_type();
87 match data_type {
88 Utf8 | Binary | LargeBinary | LargeUtf8 => {
89 let field_node = self.next_node(field)?;
90 let buffers = [
91 self.next_buffer()?,
92 self.next_buffer()?,
93 self.next_buffer()?,
94 ];
95 self.create_primitive_array(field_node, data_type, &buffers)
96 }
97 BinaryView | Utf8View => {
98 let count = variadic_counts
99 .pop_front()
100 .ok_or(ArrowError::IpcError(format!(
101 "Missing variadic count for {data_type} column"
102 )))?;
103 let count = count + 2; let buffers = (0..count)
105 .map(|_| self.next_buffer())
106 .collect::<Result<Vec<_>, _>>()?;
107 let field_node = self.next_node(field)?;
108 self.create_primitive_array(field_node, data_type, &buffers)
109 }
110 FixedSizeBinary(_) => {
111 let field_node = self.next_node(field)?;
112 let buffers = [self.next_buffer()?, self.next_buffer()?];
113 self.create_primitive_array(field_node, data_type, &buffers)
114 }
115 List(ref list_field) | LargeList(ref list_field) | Map(ref list_field, _) => {
116 let list_node = self.next_node(field)?;
117 let list_buffers = [self.next_buffer()?, self.next_buffer()?];
118 let values = self.create_array(list_field, variadic_counts)?;
119 self.create_list_array(list_node, data_type, &list_buffers, values)
120 }
121 FixedSizeList(ref list_field, _) => {
122 let list_node = self.next_node(field)?;
123 let list_buffers = [self.next_buffer()?];
124 let values = self.create_array(list_field, variadic_counts)?;
125 self.create_list_array(list_node, data_type, &list_buffers, values)
126 }
127 Struct(struct_fields) => {
128 let struct_node = self.next_node(field)?;
129 let null_buffer = self.next_buffer()?;
130
131 let mut struct_arrays = vec![];
133 for struct_field in struct_fields {
136 let child = self.create_array(struct_field, variadic_counts)?;
137 struct_arrays.push(child);
138 }
139 self.create_struct_array(struct_node, null_buffer, struct_fields, struct_arrays)
140 }
141 RunEndEncoded(run_ends_field, values_field) => {
142 let run_node = self.next_node(field)?;
143 let run_ends = self.create_array(run_ends_field, variadic_counts)?;
144 let values = self.create_array(values_field, variadic_counts)?;
145
146 let run_array_length = run_node.length() as usize;
147 let builder = ArrayData::builder(data_type.clone())
148 .len(run_array_length)
149 .offset(0)
150 .add_child_data(run_ends.into_data())
151 .add_child_data(values.into_data());
152 self.create_array_from_builder(builder)
153 }
154 Dictionary(_, _) => {
156 let index_node = self.next_node(field)?;
157 let index_buffers = [self.next_buffer()?, self.next_buffer()?];
158
159 #[allow(deprecated)]
160 let dict_id = field.dict_id().ok_or_else(|| {
161 ArrowError::ParseError(format!("Field {field} does not have dict id"))
162 })?;
163
164 let value_array = self.dictionaries_by_id.get(&dict_id).ok_or_else(|| {
165 ArrowError::ParseError(format!(
166 "Cannot find a dictionary batch with dict id: {dict_id}"
167 ))
168 })?;
169
170 self.create_dictionary_array(
171 index_node,
172 data_type,
173 &index_buffers,
174 value_array.clone(),
175 )
176 }
177 Union(fields, mode) => {
178 let union_node = self.next_node(field)?;
179 let len = union_node.length() as usize;
180
181 if self.version < MetadataVersion::V5 {
184 self.next_buffer()?;
185 }
186
187 let type_ids: ScalarBuffer<i8> =
188 self.next_buffer()?.slice_with_length(0, len).into();
189
190 let value_offsets = match mode {
191 UnionMode::Dense => {
192 let offsets: ScalarBuffer<i32> =
193 self.next_buffer()?.slice_with_length(0, len * 4).into();
194 Some(offsets)
195 }
196 UnionMode::Sparse => None,
197 };
198
199 let mut children = Vec::with_capacity(fields.len());
200
201 for (_id, field) in fields.iter() {
202 let child = self.create_array(field, variadic_counts)?;
203 children.push(child);
204 }
205
206 let array = if self.skip_validation.get() {
207 unsafe {
209 UnionArray::new_unchecked(fields.clone(), type_ids, value_offsets, children)
210 }
211 } else {
212 UnionArray::try_new(fields.clone(), type_ids, value_offsets, children)?
213 };
214 Ok(Arc::new(array))
215 }
216 Null => {
217 let node = self.next_node(field)?;
218 let length = node.length();
219 let null_count = node.null_count();
220
221 if length != null_count {
222 return Err(ArrowError::SchemaError(format!(
223 "Field {field} of NullArray has unequal null_count {null_count} and len {length}"
224 )));
225 }
226
227 let builder = ArrayData::builder(data_type.clone())
228 .len(length as usize)
229 .offset(0);
230 self.create_array_from_builder(builder)
231 }
232 _ => {
233 let field_node = self.next_node(field)?;
234 let buffers = [self.next_buffer()?, self.next_buffer()?];
235 self.create_primitive_array(field_node, data_type, &buffers)
236 }
237 }
238 }
239
240 fn create_primitive_array(
243 &self,
244 field_node: &FieldNode,
245 data_type: &DataType,
246 buffers: &[Buffer],
247 ) -> Result<ArrayRef, ArrowError> {
248 let length = field_node.length() as usize;
249 let null_buffer = (field_node.null_count() > 0).then_some(buffers[0].clone());
250 let builder = match data_type {
251 Utf8 | Binary | LargeBinary | LargeUtf8 => {
252 ArrayData::builder(data_type.clone())
254 .len(length)
255 .buffers(buffers[1..3].to_vec())
256 .null_bit_buffer(null_buffer)
257 }
258 BinaryView | Utf8View => ArrayData::builder(data_type.clone())
259 .len(length)
260 .buffers(buffers[1..].to_vec())
261 .null_bit_buffer(null_buffer),
262 _ if data_type.is_primitive() || matches!(data_type, Boolean | FixedSizeBinary(_)) => {
263 ArrayData::builder(data_type.clone())
265 .len(length)
266 .add_buffer(buffers[1].clone())
267 .null_bit_buffer(null_buffer)
268 }
269 t => unreachable!("Data type {:?} either unsupported or not primitive", t),
270 };
271
272 self.create_array_from_builder(builder)
273 }
274
275 fn create_array_from_builder(&self, builder: ArrayDataBuilder) -> Result<ArrayRef, ArrowError> {
277 let mut builder = builder.align_buffers(!self.require_alignment);
278 if self.skip_validation.get() {
279 unsafe { builder = builder.skip_validation(true) }
281 };
282 Ok(make_array(builder.build()?))
283 }
284
285 fn create_list_array(
288 &self,
289 field_node: &FieldNode,
290 data_type: &DataType,
291 buffers: &[Buffer],
292 child_array: ArrayRef,
293 ) -> Result<ArrayRef, ArrowError> {
294 let null_buffer = (field_node.null_count() > 0).then_some(buffers[0].clone());
295 let length = field_node.length() as usize;
296 let child_data = child_array.into_data();
297 let builder = match data_type {
298 List(_) | LargeList(_) | Map(_, _) => ArrayData::builder(data_type.clone())
299 .len(length)
300 .add_buffer(buffers[1].clone())
301 .add_child_data(child_data)
302 .null_bit_buffer(null_buffer),
303
304 FixedSizeList(_, _) => ArrayData::builder(data_type.clone())
305 .len(length)
306 .add_child_data(child_data)
307 .null_bit_buffer(null_buffer),
308
309 _ => unreachable!("Cannot create list or map array from {:?}", data_type),
310 };
311
312 self.create_array_from_builder(builder)
313 }
314
315 fn create_struct_array(
316 &self,
317 struct_node: &FieldNode,
318 null_buffer: Buffer,
319 struct_fields: &Fields,
320 struct_arrays: Vec<ArrayRef>,
321 ) -> Result<ArrayRef, ArrowError> {
322 let null_count = struct_node.null_count() as usize;
323 let len = struct_node.length() as usize;
324
325 let nulls = (null_count > 0).then(|| BooleanBuffer::new(null_buffer, 0, len).into());
326 if struct_arrays.is_empty() {
327 return Ok(Arc::new(StructArray::new_empty_fields(len, nulls)));
330 }
331
332 let struct_array = if self.skip_validation.get() {
333 unsafe { StructArray::new_unchecked(struct_fields.clone(), struct_arrays, nulls) }
335 } else {
336 StructArray::try_new(struct_fields.clone(), struct_arrays, nulls)?
337 };
338
339 Ok(Arc::new(struct_array))
340 }
341
342 fn create_dictionary_array(
345 &self,
346 field_node: &FieldNode,
347 data_type: &DataType,
348 buffers: &[Buffer],
349 value_array: ArrayRef,
350 ) -> Result<ArrayRef, ArrowError> {
351 if let Dictionary(_, _) = *data_type {
352 let null_buffer = (field_node.null_count() > 0).then_some(buffers[0].clone());
353 let builder = ArrayData::builder(data_type.clone())
354 .len(field_node.length() as usize)
355 .add_buffer(buffers[1].clone())
356 .add_child_data(value_array.into_data())
357 .null_bit_buffer(null_buffer);
358 self.create_array_from_builder(builder)
359 } else {
360 unreachable!("Cannot create dictionary array from {:?}", data_type)
361 }
362 }
363}
364
365struct RecordBatchDecoder<'a> {
370 batch: crate::RecordBatch<'a>,
372 schema: SchemaRef,
374 dictionaries_by_id: &'a HashMap<i64, ArrayRef>,
376 compression: Option<CompressionCodec>,
378 version: MetadataVersion,
380 data: &'a Buffer,
382 nodes: VectorIter<'a, FieldNode>,
384 buffers: VectorIter<'a, crate::Buffer>,
386 projection: Option<&'a [usize]>,
389 require_alignment: bool,
392 skip_validation: UnsafeFlag,
396}
397
398impl<'a> RecordBatchDecoder<'a> {
399 fn try_new(
401 buf: &'a Buffer,
402 batch: crate::RecordBatch<'a>,
403 schema: SchemaRef,
404 dictionaries_by_id: &'a HashMap<i64, ArrayRef>,
405 metadata: &'a MetadataVersion,
406 ) -> Result<Self, ArrowError> {
407 let buffers = batch.buffers().ok_or_else(|| {
408 ArrowError::IpcError("Unable to get buffers from IPC RecordBatch".to_string())
409 })?;
410 let field_nodes = batch.nodes().ok_or_else(|| {
411 ArrowError::IpcError("Unable to get field nodes from IPC RecordBatch".to_string())
412 })?;
413
414 let batch_compression = batch.compression();
415 let compression = batch_compression
416 .map(|batch_compression| batch_compression.codec().try_into())
417 .transpose()?;
418
419 Ok(Self {
420 batch,
421 schema,
422 dictionaries_by_id,
423 compression,
424 version: *metadata,
425 data: buf,
426 nodes: field_nodes.iter(),
427 buffers: buffers.iter(),
428 projection: None,
429 require_alignment: false,
430 skip_validation: UnsafeFlag::new(),
431 })
432 }
433
434 pub fn with_projection(mut self, projection: Option<&'a [usize]>) -> Self {
439 self.projection = projection;
440 self
441 }
442
443 pub fn with_require_alignment(mut self, require_alignment: bool) -> Self {
449 self.require_alignment = require_alignment;
450 self
451 }
452
453 pub(crate) fn with_skip_validation(mut self, skip_validation: UnsafeFlag) -> Self {
465 self.skip_validation = skip_validation;
466 self
467 }
468
469 fn read_record_batch(mut self) -> Result<RecordBatch, ArrowError> {
471 let mut variadic_counts: VecDeque<i64> = self
472 .batch
473 .variadicBufferCounts()
474 .into_iter()
475 .flatten()
476 .collect();
477
478 let options = RecordBatchOptions::new().with_row_count(Some(self.batch.length() as usize));
479
480 let schema = Arc::clone(&self.schema);
481 if let Some(projection) = self.projection {
482 let mut arrays = vec![];
483 for (idx, field) in schema.fields().iter().enumerate() {
485 if let Some(proj_idx) = projection.iter().position(|p| p == &idx) {
487 let child = self.create_array(field, &mut variadic_counts)?;
488 arrays.push((proj_idx, child));
489 } else {
490 self.skip_field(field, &mut variadic_counts)?;
491 }
492 }
493
494 arrays.sort_by_key(|t| t.0);
495
496 let schema = Arc::new(schema.project(projection)?);
497 let columns = arrays.into_iter().map(|t| t.1).collect::<Vec<_>>();
498
499 if self.skip_validation.get() {
500 unsafe {
502 Ok(RecordBatch::new_unchecked(
503 schema,
504 columns,
505 self.batch.length() as usize,
506 ))
507 }
508 } else {
509 assert!(variadic_counts.is_empty());
510 RecordBatch::try_new_with_options(schema, columns, &options)
511 }
512 } else {
513 let mut children = vec![];
514 for field in schema.fields() {
516 let child = self.create_array(field, &mut variadic_counts)?;
517 children.push(child);
518 }
519
520 if self.skip_validation.get() {
521 unsafe {
523 Ok(RecordBatch::new_unchecked(
524 schema,
525 children,
526 self.batch.length() as usize,
527 ))
528 }
529 } else {
530 assert!(variadic_counts.is_empty());
531 RecordBatch::try_new_with_options(schema, children, &options)
532 }
533 }
534 }
535
536 fn next_buffer(&mut self) -> Result<Buffer, ArrowError> {
537 read_buffer(self.buffers.next().unwrap(), self.data, self.compression)
538 }
539
540 fn skip_buffer(&mut self) {
541 self.buffers.next().unwrap();
542 }
543
544 fn next_node(&mut self, field: &Field) -> Result<&'a FieldNode, ArrowError> {
545 self.nodes.next().ok_or_else(|| {
546 ArrowError::SchemaError(format!(
547 "Invalid data for schema. {field} refers to node not found in schema",
548 ))
549 })
550 }
551
552 fn skip_field(
553 &mut self,
554 field: &Field,
555 variadic_count: &mut VecDeque<i64>,
556 ) -> Result<(), ArrowError> {
557 self.next_node(field)?;
558
559 match field.data_type() {
560 Utf8 | Binary | LargeBinary | LargeUtf8 => {
561 for _ in 0..3 {
562 self.skip_buffer()
563 }
564 }
565 Utf8View | BinaryView => {
566 let count = variadic_count
567 .pop_front()
568 .ok_or(ArrowError::IpcError(format!(
569 "Missing variadic count for {} column",
570 field.data_type()
571 )))?;
572 let count = count + 2; for _i in 0..count {
574 self.skip_buffer()
575 }
576 }
577 FixedSizeBinary(_) => {
578 self.skip_buffer();
579 self.skip_buffer();
580 }
581 List(list_field) | LargeList(list_field) | Map(list_field, _) => {
582 self.skip_buffer();
583 self.skip_buffer();
584 self.skip_field(list_field, variadic_count)?;
585 }
586 FixedSizeList(list_field, _) => {
587 self.skip_buffer();
588 self.skip_field(list_field, variadic_count)?;
589 }
590 Struct(struct_fields) => {
591 self.skip_buffer();
592
593 for struct_field in struct_fields {
595 self.skip_field(struct_field, variadic_count)?
596 }
597 }
598 RunEndEncoded(run_ends_field, values_field) => {
599 self.skip_field(run_ends_field, variadic_count)?;
600 self.skip_field(values_field, variadic_count)?;
601 }
602 Dictionary(_, _) => {
603 self.skip_buffer(); self.skip_buffer(); }
606 Union(fields, mode) => {
607 self.skip_buffer(); match mode {
610 UnionMode::Dense => self.skip_buffer(),
611 UnionMode::Sparse => {}
612 };
613
614 for (_, field) in fields.iter() {
615 self.skip_field(field, variadic_count)?
616 }
617 }
618 Null => {} _ => {
620 self.skip_buffer();
621 self.skip_buffer();
622 }
623 };
624 Ok(())
625 }
626}
627
628pub fn read_record_batch(
639 buf: &Buffer,
640 batch: crate::RecordBatch,
641 schema: SchemaRef,
642 dictionaries_by_id: &HashMap<i64, ArrayRef>,
643 projection: Option<&[usize]>,
644 metadata: &MetadataVersion,
645) -> Result<RecordBatch, ArrowError> {
646 RecordBatchDecoder::try_new(buf, batch, schema, dictionaries_by_id, metadata)?
647 .with_projection(projection)
648 .with_require_alignment(false)
649 .read_record_batch()
650}
651
652pub fn read_dictionary(
655 buf: &Buffer,
656 batch: crate::DictionaryBatch,
657 schema: &Schema,
658 dictionaries_by_id: &mut HashMap<i64, ArrayRef>,
659 metadata: &MetadataVersion,
660) -> Result<(), ArrowError> {
661 read_dictionary_impl(
662 buf,
663 batch,
664 schema,
665 dictionaries_by_id,
666 metadata,
667 false,
668 UnsafeFlag::new(),
669 )
670}
671
672fn read_dictionary_impl(
673 buf: &Buffer,
674 batch: crate::DictionaryBatch,
675 schema: &Schema,
676 dictionaries_by_id: &mut HashMap<i64, ArrayRef>,
677 metadata: &MetadataVersion,
678 require_alignment: bool,
679 skip_validation: UnsafeFlag,
680) -> Result<(), ArrowError> {
681 if batch.isDelta() {
682 return Err(ArrowError::InvalidArgumentError(
683 "delta dictionary batches not supported".to_string(),
684 ));
685 }
686
687 let id = batch.id();
688 #[allow(deprecated)]
689 let fields_using_this_dictionary = schema.fields_with_dict_id(id);
690 let first_field = fields_using_this_dictionary.first().ok_or_else(|| {
691 ArrowError::InvalidArgumentError(format!("dictionary id {id} not found in schema"))
692 })?;
693
694 let dictionary_values: ArrayRef = match first_field.data_type() {
698 DataType::Dictionary(_, ref value_type) => {
699 let value = value_type.as_ref().clone();
701 let schema = Schema::new(vec![Field::new("", value, true)]);
702 let record_batch = RecordBatchDecoder::try_new(
704 buf,
705 batch.data().unwrap(),
706 Arc::new(schema),
707 dictionaries_by_id,
708 metadata,
709 )?
710 .with_require_alignment(require_alignment)
711 .with_skip_validation(skip_validation)
712 .read_record_batch()?;
713
714 Some(record_batch.column(0).clone())
715 }
716 _ => None,
717 }
718 .ok_or_else(|| {
719 ArrowError::InvalidArgumentError(format!("dictionary id {id} not found in schema"))
720 })?;
721
722 dictionaries_by_id.insert(id, dictionary_values.clone());
726
727 Ok(())
728}
729
730fn read_block<R: Read + Seek>(mut reader: R, block: &Block) -> Result<Buffer, ArrowError> {
732 reader.seek(SeekFrom::Start(block.offset() as u64))?;
733 let body_len = block.bodyLength().to_usize().unwrap();
734 let metadata_len = block.metaDataLength().to_usize().unwrap();
735 let total_len = body_len.checked_add(metadata_len).unwrap();
736
737 let mut buf = MutableBuffer::from_len_zeroed(total_len);
738 reader.read_exact(&mut buf)?;
739 Ok(buf.into())
740}
741
742fn parse_message(buf: &[u8]) -> Result<Message, ArrowError> {
746 let buf = match buf[..4] == CONTINUATION_MARKER {
747 true => &buf[8..],
748 false => &buf[4..],
749 };
750 crate::root_as_message(buf)
751 .map_err(|err| ArrowError::ParseError(format!("Unable to get root as message: {err:?}")))
752}
753
754pub fn read_footer_length(buf: [u8; 10]) -> Result<usize, ArrowError> {
758 if buf[4..] != super::ARROW_MAGIC {
759 return Err(ArrowError::ParseError(
760 "Arrow file does not contain correct footer".to_string(),
761 ));
762 }
763
764 let footer_len = i32::from_le_bytes(buf[..4].try_into().unwrap());
766 footer_len
767 .try_into()
768 .map_err(|_| ArrowError::ParseError(format!("Invalid footer length: {footer_len}")))
769}
770
771#[derive(Debug)]
836pub struct FileDecoder {
837 schema: SchemaRef,
838 dictionaries: HashMap<i64, ArrayRef>,
839 version: MetadataVersion,
840 projection: Option<Vec<usize>>,
841 require_alignment: bool,
842 skip_validation: UnsafeFlag,
843}
844
845impl FileDecoder {
846 pub fn new(schema: SchemaRef, version: MetadataVersion) -> Self {
848 Self {
849 schema,
850 version,
851 dictionaries: Default::default(),
852 projection: None,
853 require_alignment: false,
854 skip_validation: UnsafeFlag::new(),
855 }
856 }
857
858 pub fn with_projection(mut self, projection: Vec<usize>) -> Self {
860 self.projection = Some(projection);
861 self
862 }
863
864 pub fn with_require_alignment(mut self, require_alignment: bool) -> Self {
877 self.require_alignment = require_alignment;
878 self
879 }
880
881 pub unsafe fn with_skip_validation(mut self, skip_validation: bool) -> Self {
892 self.skip_validation.set(skip_validation);
893 self
894 }
895
896 fn read_message<'a>(&self, buf: &'a [u8]) -> Result<Message<'a>, ArrowError> {
897 let message = parse_message(buf)?;
898
899 if self.version != MetadataVersion::V1 && message.version() != self.version {
901 return Err(ArrowError::IpcError(
902 "Could not read IPC message as metadata versions mismatch".to_string(),
903 ));
904 }
905 Ok(message)
906 }
907
908 pub fn read_dictionary(&mut self, block: &Block, buf: &Buffer) -> Result<(), ArrowError> {
910 let message = self.read_message(buf)?;
911 match message.header_type() {
912 crate::MessageHeader::DictionaryBatch => {
913 let batch = message.header_as_dictionary_batch().unwrap();
914 read_dictionary_impl(
915 &buf.slice(block.metaDataLength() as _),
916 batch,
917 &self.schema,
918 &mut self.dictionaries,
919 &message.version(),
920 self.require_alignment,
921 self.skip_validation.clone(),
922 )
923 }
924 t => Err(ArrowError::ParseError(format!(
925 "Expecting DictionaryBatch in dictionary blocks, found {t:?}."
926 ))),
927 }
928 }
929
930 pub fn read_record_batch(
932 &self,
933 block: &Block,
934 buf: &Buffer,
935 ) -> Result<Option<RecordBatch>, ArrowError> {
936 let message = self.read_message(buf)?;
937 match message.header_type() {
938 crate::MessageHeader::Schema => Err(ArrowError::IpcError(
939 "Not expecting a schema when messages are read".to_string(),
940 )),
941 crate::MessageHeader::RecordBatch => {
942 let batch = message.header_as_record_batch().ok_or_else(|| {
943 ArrowError::IpcError("Unable to read IPC message as record batch".to_string())
944 })?;
945 RecordBatchDecoder::try_new(
947 &buf.slice(block.metaDataLength() as _),
948 batch,
949 self.schema.clone(),
950 &self.dictionaries,
951 &message.version(),
952 )?
953 .with_projection(self.projection.as_deref())
954 .with_require_alignment(self.require_alignment)
955 .with_skip_validation(self.skip_validation.clone())
956 .read_record_batch()
957 .map(Some)
958 }
959 crate::MessageHeader::NONE => Ok(None),
960 t => Err(ArrowError::InvalidArgumentError(format!(
961 "Reading types other than record batches not yet supported, unable to read {t:?}"
962 ))),
963 }
964 }
965}
966
967#[derive(Debug)]
969pub struct FileReaderBuilder {
970 projection: Option<Vec<usize>>,
972 max_footer_fb_tables: usize,
974 max_footer_fb_depth: usize,
976}
977
978impl Default for FileReaderBuilder {
979 fn default() -> Self {
980 let verifier_options = VerifierOptions::default();
981 Self {
982 max_footer_fb_tables: verifier_options.max_tables,
983 max_footer_fb_depth: verifier_options.max_depth,
984 projection: None,
985 }
986 }
987}
988
989impl FileReaderBuilder {
990 pub fn new() -> Self {
994 Self::default()
995 }
996
997 pub fn with_projection(mut self, projection: Vec<usize>) -> Self {
999 self.projection = Some(projection);
1000 self
1001 }
1002
1003 pub fn with_max_footer_fb_tables(mut self, max_footer_fb_tables: usize) -> Self {
1016 self.max_footer_fb_tables = max_footer_fb_tables;
1017 self
1018 }
1019
1020 pub fn with_max_footer_fb_depth(mut self, max_footer_fb_depth: usize) -> Self {
1033 self.max_footer_fb_depth = max_footer_fb_depth;
1034 self
1035 }
1036
1037 pub fn build<R: Read + Seek>(self, mut reader: R) -> Result<FileReader<R>, ArrowError> {
1039 let mut buffer = [0; 10];
1041 reader.seek(SeekFrom::End(-10))?;
1042 reader.read_exact(&mut buffer)?;
1043
1044 let footer_len = read_footer_length(buffer)?;
1045
1046 let mut footer_data = vec![0; footer_len];
1048 reader.seek(SeekFrom::End(-10 - footer_len as i64))?;
1049 reader.read_exact(&mut footer_data)?;
1050
1051 let verifier_options = VerifierOptions {
1052 max_tables: self.max_footer_fb_tables,
1053 max_depth: self.max_footer_fb_depth,
1054 ..Default::default()
1055 };
1056 let footer = crate::root_as_footer_with_opts(&verifier_options, &footer_data[..]).map_err(
1057 |err| ArrowError::ParseError(format!("Unable to get root as footer: {err:?}")),
1058 )?;
1059
1060 let blocks = footer.recordBatches().ok_or_else(|| {
1061 ArrowError::ParseError("Unable to get record batches from IPC Footer".to_string())
1062 })?;
1063
1064 let total_blocks = blocks.len();
1065
1066 let ipc_schema = footer.schema().unwrap();
1067 if !ipc_schema.endianness().equals_to_target_endianness() {
1068 return Err(ArrowError::IpcError(
1069 "the endianness of the source system does not match the endianness of the target system.".to_owned()
1070 ));
1071 }
1072
1073 let schema = crate::convert::fb_to_schema(ipc_schema);
1074
1075 let mut custom_metadata = HashMap::new();
1076 if let Some(fb_custom_metadata) = footer.custom_metadata() {
1077 for kv in fb_custom_metadata.into_iter() {
1078 custom_metadata.insert(
1079 kv.key().unwrap().to_string(),
1080 kv.value().unwrap().to_string(),
1081 );
1082 }
1083 }
1084
1085 let mut decoder = FileDecoder::new(Arc::new(schema), footer.version());
1086 if let Some(projection) = self.projection {
1087 decoder = decoder.with_projection(projection)
1088 }
1089
1090 if let Some(dictionaries) = footer.dictionaries() {
1092 for block in dictionaries {
1093 let buf = read_block(&mut reader, block)?;
1094 decoder.read_dictionary(block, &buf)?;
1095 }
1096 }
1097
1098 Ok(FileReader {
1099 reader,
1100 blocks: blocks.iter().copied().collect(),
1101 current_block: 0,
1102 total_blocks,
1103 decoder,
1104 custom_metadata,
1105 })
1106 }
1107}
1108
1109pub struct FileReader<R> {
1154 reader: R,
1156
1157 decoder: FileDecoder,
1159
1160 blocks: Vec<Block>,
1164
1165 current_block: usize,
1167
1168 total_blocks: usize,
1170
1171 custom_metadata: HashMap<String, String>,
1173}
1174
1175impl<R> fmt::Debug for FileReader<R> {
1176 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> Result<(), fmt::Error> {
1177 f.debug_struct("FileReader<R>")
1178 .field("decoder", &self.decoder)
1179 .field("blocks", &self.blocks)
1180 .field("current_block", &self.current_block)
1181 .field("total_blocks", &self.total_blocks)
1182 .finish_non_exhaustive()
1183 }
1184}
1185
1186impl<R: Read + Seek> FileReader<BufReader<R>> {
1187 pub fn try_new_buffered(reader: R, projection: Option<Vec<usize>>) -> Result<Self, ArrowError> {
1191 Self::try_new(BufReader::new(reader), projection)
1192 }
1193}
1194
1195impl<R: Read + Seek> FileReader<R> {
1196 pub fn try_new(reader: R, projection: Option<Vec<usize>>) -> Result<Self, ArrowError> {
1207 let builder = FileReaderBuilder {
1208 projection,
1209 ..Default::default()
1210 };
1211 builder.build(reader)
1212 }
1213
1214 pub fn custom_metadata(&self) -> &HashMap<String, String> {
1216 &self.custom_metadata
1217 }
1218
1219 pub fn num_batches(&self) -> usize {
1221 self.total_blocks
1222 }
1223
1224 pub fn schema(&self) -> SchemaRef {
1226 self.decoder.schema.clone()
1227 }
1228
1229 pub fn set_index(&mut self, index: usize) -> Result<(), ArrowError> {
1233 if index >= self.total_blocks {
1234 Err(ArrowError::InvalidArgumentError(format!(
1235 "Cannot set batch to index {} from {} total batches",
1236 index, self.total_blocks
1237 )))
1238 } else {
1239 self.current_block = index;
1240 Ok(())
1241 }
1242 }
1243
1244 fn maybe_next(&mut self) -> Result<Option<RecordBatch>, ArrowError> {
1245 let block = &self.blocks[self.current_block];
1246 self.current_block += 1;
1247
1248 let buffer = read_block(&mut self.reader, block)?;
1250 self.decoder.read_record_batch(block, &buffer)
1251 }
1252
1253 pub fn get_ref(&self) -> &R {
1257 &self.reader
1258 }
1259
1260 pub fn get_mut(&mut self) -> &mut R {
1264 &mut self.reader
1265 }
1266
1267 pub unsafe fn with_skip_validation(mut self, skip_validation: bool) -> Self {
1273 self.decoder = self.decoder.with_skip_validation(skip_validation);
1274 self
1275 }
1276}
1277
1278impl<R: Read + Seek> Iterator for FileReader<R> {
1279 type Item = Result<RecordBatch, ArrowError>;
1280
1281 fn next(&mut self) -> Option<Self::Item> {
1282 if self.current_block < self.total_blocks {
1284 self.maybe_next().transpose()
1285 } else {
1286 None
1287 }
1288 }
1289}
1290
1291impl<R: Read + Seek> RecordBatchReader for FileReader<R> {
1292 fn schema(&self) -> SchemaRef {
1293 self.schema()
1294 }
1295}
1296
1297pub struct StreamReader<R> {
1331 reader: R,
1333
1334 schema: SchemaRef,
1336
1337 dictionaries_by_id: HashMap<i64, ArrayRef>,
1341
1342 finished: bool,
1346
1347 projection: Option<(Vec<usize>, Schema)>,
1349
1350 skip_validation: UnsafeFlag,
1354}
1355
1356impl<R> fmt::Debug for StreamReader<R> {
1357 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> std::result::Result<(), fmt::Error> {
1358 f.debug_struct("StreamReader<R>")
1359 .field("reader", &"R")
1360 .field("schema", &self.schema)
1361 .field("dictionaries_by_id", &self.dictionaries_by_id)
1362 .field("finished", &self.finished)
1363 .field("projection", &self.projection)
1364 .finish()
1365 }
1366}
1367
1368impl<R: Read> StreamReader<BufReader<R>> {
1369 pub fn try_new_buffered(reader: R, projection: Option<Vec<usize>>) -> Result<Self, ArrowError> {
1373 Self::try_new(BufReader::new(reader), projection)
1374 }
1375}
1376
1377impl<R: Read> StreamReader<R> {
1378 pub fn try_new(
1390 mut reader: R,
1391 projection: Option<Vec<usize>>,
1392 ) -> Result<StreamReader<R>, ArrowError> {
1393 let mut meta_size: [u8; 4] = [0; 4];
1395 reader.read_exact(&mut meta_size)?;
1396 let meta_len = {
1397 if meta_size == CONTINUATION_MARKER {
1400 reader.read_exact(&mut meta_size)?;
1401 }
1402 i32::from_le_bytes(meta_size)
1403 };
1404
1405 let mut meta_buffer = vec![0; meta_len as usize];
1406 reader.read_exact(&mut meta_buffer)?;
1407
1408 let message = crate::root_as_message(meta_buffer.as_slice()).map_err(|err| {
1409 ArrowError::ParseError(format!("Unable to get root as message: {err:?}"))
1410 })?;
1411 let ipc_schema: crate::Schema = message.header_as_schema().ok_or_else(|| {
1413 ArrowError::ParseError("Unable to read IPC message as schema".to_string())
1414 })?;
1415 let schema = crate::convert::fb_to_schema(ipc_schema);
1416
1417 let dictionaries_by_id = HashMap::new();
1419
1420 let projection = match projection {
1421 Some(projection_indices) => {
1422 let schema = schema.project(&projection_indices)?;
1423 Some((projection_indices, schema))
1424 }
1425 _ => None,
1426 };
1427 Ok(Self {
1428 reader,
1429 schema: Arc::new(schema),
1430 finished: false,
1431 dictionaries_by_id,
1432 projection,
1433 skip_validation: UnsafeFlag::new(),
1434 })
1435 }
1436
1437 #[deprecated(since = "53.0.0", note = "use `try_new` instead")]
1439 pub fn try_new_unbuffered(
1440 reader: R,
1441 projection: Option<Vec<usize>>,
1442 ) -> Result<Self, ArrowError> {
1443 Self::try_new(reader, projection)
1444 }
1445
1446 pub fn schema(&self) -> SchemaRef {
1448 self.schema.clone()
1449 }
1450
1451 pub fn is_finished(&self) -> bool {
1453 self.finished
1454 }
1455
1456 fn maybe_next(&mut self) -> Result<Option<RecordBatch>, ArrowError> {
1457 if self.finished {
1458 return Ok(None);
1459 }
1460 let mut meta_size: [u8; 4] = [0; 4];
1462
1463 match self.reader.read_exact(&mut meta_size) {
1464 Ok(()) => (),
1465 Err(e) => {
1466 return if e.kind() == std::io::ErrorKind::UnexpectedEof {
1467 self.finished = true;
1471 Ok(None)
1472 } else {
1473 Err(ArrowError::from(e))
1474 };
1475 }
1476 }
1477
1478 let meta_len = {
1479 if meta_size == CONTINUATION_MARKER {
1482 self.reader.read_exact(&mut meta_size)?;
1483 }
1484 i32::from_le_bytes(meta_size)
1485 };
1486
1487 if meta_len == 0 {
1488 self.finished = true;
1490 return Ok(None);
1491 }
1492
1493 let mut meta_buffer = vec![0; meta_len as usize];
1494 self.reader.read_exact(&mut meta_buffer)?;
1495
1496 let vecs = &meta_buffer.to_vec();
1497 let message = crate::root_as_message(vecs).map_err(|err| {
1498 ArrowError::ParseError(format!("Unable to get root as message: {err:?}"))
1499 })?;
1500
1501 match message.header_type() {
1502 crate::MessageHeader::Schema => Err(ArrowError::IpcError(
1503 "Not expecting a schema when messages are read".to_string(),
1504 )),
1505 crate::MessageHeader::RecordBatch => {
1506 let batch = message.header_as_record_batch().ok_or_else(|| {
1507 ArrowError::IpcError("Unable to read IPC message as record batch".to_string())
1508 })?;
1509 let mut buf = MutableBuffer::from_len_zeroed(message.bodyLength() as usize);
1511 self.reader.read_exact(&mut buf)?;
1512
1513 RecordBatchDecoder::try_new(
1514 &buf.into(),
1515 batch,
1516 self.schema(),
1517 &self.dictionaries_by_id,
1518 &message.version(),
1519 )?
1520 .with_projection(self.projection.as_ref().map(|x| x.0.as_ref()))
1521 .with_require_alignment(false)
1522 .with_skip_validation(self.skip_validation.clone())
1523 .read_record_batch()
1524 .map(Some)
1525 }
1526 crate::MessageHeader::DictionaryBatch => {
1527 let batch = message.header_as_dictionary_batch().ok_or_else(|| {
1528 ArrowError::IpcError(
1529 "Unable to read IPC message as dictionary batch".to_string(),
1530 )
1531 })?;
1532 let mut buf = MutableBuffer::from_len_zeroed(message.bodyLength() as usize);
1534 self.reader.read_exact(&mut buf)?;
1535
1536 read_dictionary_impl(
1537 &buf.into(),
1538 batch,
1539 &self.schema,
1540 &mut self.dictionaries_by_id,
1541 &message.version(),
1542 false,
1543 self.skip_validation.clone(),
1544 )?;
1545
1546 self.maybe_next()
1548 }
1549 crate::MessageHeader::NONE => Ok(None),
1550 t => Err(ArrowError::InvalidArgumentError(format!(
1551 "Reading types other than record batches not yet supported, unable to read {t:?} "
1552 ))),
1553 }
1554 }
1555
1556 pub fn get_ref(&self) -> &R {
1560 &self.reader
1561 }
1562
1563 pub fn get_mut(&mut self) -> &mut R {
1567 &mut self.reader
1568 }
1569
1570 pub unsafe fn with_skip_validation(mut self, skip_validation: bool) -> Self {
1576 self.skip_validation.set(skip_validation);
1577 self
1578 }
1579}
1580
1581impl<R: Read> Iterator for StreamReader<R> {
1582 type Item = Result<RecordBatch, ArrowError>;
1583
1584 fn next(&mut self) -> Option<Self::Item> {
1585 self.maybe_next().transpose()
1586 }
1587}
1588
1589impl<R: Read> RecordBatchReader for StreamReader<R> {
1590 fn schema(&self) -> SchemaRef {
1591 self.schema.clone()
1592 }
1593}
1594
1595#[cfg(test)]
1596mod tests {
1597 use crate::convert::fb_to_schema;
1598 use crate::writer::{
1599 unslice_run_array, write_message, DictionaryTracker, IpcDataGenerator, IpcWriteOptions,
1600 };
1601
1602 use super::*;
1603
1604 use crate::{root_as_footer, root_as_message, size_prefixed_root_as_message};
1605 use arrow_array::builder::{PrimitiveRunBuilder, UnionBuilder};
1606 use arrow_array::types::*;
1607 use arrow_buffer::{NullBuffer, OffsetBuffer};
1608 use arrow_data::ArrayDataBuilder;
1609
1610 fn create_test_projection_schema() -> Schema {
1611 let list_data_type = DataType::List(Arc::new(Field::new_list_field(DataType::Int32, true)));
1613
1614 let fixed_size_list_data_type =
1615 DataType::FixedSizeList(Arc::new(Field::new_list_field(DataType::Int32, false)), 3);
1616
1617 let union_fields = UnionFields::new(
1618 vec![0, 1],
1619 vec![
1620 Field::new("a", DataType::Int32, false),
1621 Field::new("b", DataType::Float64, false),
1622 ],
1623 );
1624
1625 let union_data_type = DataType::Union(union_fields, UnionMode::Dense);
1626
1627 let struct_fields = Fields::from(vec![
1628 Field::new("id", DataType::Int32, false),
1629 Field::new_list("list", Field::new_list_field(DataType::Int8, true), false),
1630 ]);
1631 let struct_data_type = DataType::Struct(struct_fields);
1632
1633 let run_encoded_data_type = DataType::RunEndEncoded(
1634 Arc::new(Field::new("run_ends", DataType::Int16, false)),
1635 Arc::new(Field::new("values", DataType::Int32, true)),
1636 );
1637
1638 Schema::new(vec![
1640 Field::new("f0", DataType::UInt32, false),
1641 Field::new("f1", DataType::Utf8, false),
1642 Field::new("f2", DataType::Boolean, false),
1643 Field::new("f3", union_data_type, true),
1644 Field::new("f4", DataType::Null, true),
1645 Field::new("f5", DataType::Float64, true),
1646 Field::new("f6", list_data_type, false),
1647 Field::new("f7", DataType::FixedSizeBinary(3), true),
1648 Field::new("f8", fixed_size_list_data_type, false),
1649 Field::new("f9", struct_data_type, false),
1650 Field::new("f10", run_encoded_data_type, false),
1651 Field::new("f11", DataType::Boolean, false),
1652 Field::new_dictionary("f12", DataType::Int8, DataType::Utf8, false),
1653 Field::new("f13", DataType::Utf8, false),
1654 ])
1655 }
1656
1657 fn create_test_projection_batch_data(schema: &Schema) -> RecordBatch {
1658 let array0 = UInt32Array::from(vec![1, 2, 3]);
1660 let array1 = StringArray::from(vec!["foo", "bar", "baz"]);
1661 let array2 = BooleanArray::from(vec![true, false, true]);
1662
1663 let mut union_builder = UnionBuilder::new_dense();
1664 union_builder.append::<Int32Type>("a", 1).unwrap();
1665 union_builder.append::<Float64Type>("b", 10.1).unwrap();
1666 union_builder.append_null::<Float64Type>("b").unwrap();
1667 let array3 = union_builder.build().unwrap();
1668
1669 let array4 = NullArray::new(3);
1670 let array5 = Float64Array::from(vec![Some(1.1), None, Some(3.3)]);
1671 let array6_values = vec![
1672 Some(vec![Some(10), Some(10), Some(10)]),
1673 Some(vec![Some(20), Some(20), Some(20)]),
1674 Some(vec![Some(30), Some(30)]),
1675 ];
1676 let array6 = ListArray::from_iter_primitive::<Int32Type, _, _>(array6_values);
1677 let array7_values = vec![vec![11, 12, 13], vec![22, 23, 24], vec![33, 34, 35]];
1678 let array7 = FixedSizeBinaryArray::try_from_iter(array7_values.into_iter()).unwrap();
1679
1680 let array8_values = ArrayData::builder(DataType::Int32)
1681 .len(9)
1682 .add_buffer(Buffer::from_slice_ref([40, 41, 42, 43, 44, 45, 46, 47, 48]))
1683 .build()
1684 .unwrap();
1685 let array8_data = ArrayData::builder(schema.field(8).data_type().clone())
1686 .len(3)
1687 .add_child_data(array8_values)
1688 .build()
1689 .unwrap();
1690 let array8 = FixedSizeListArray::from(array8_data);
1691
1692 let array9_id: ArrayRef = Arc::new(Int32Array::from(vec![1001, 1002, 1003]));
1693 let array9_list: ArrayRef =
1694 Arc::new(ListArray::from_iter_primitive::<Int8Type, _, _>(vec![
1695 Some(vec![Some(-10)]),
1696 Some(vec![Some(-20), Some(-20), Some(-20)]),
1697 Some(vec![Some(-30)]),
1698 ]));
1699 let array9 = ArrayDataBuilder::new(schema.field(9).data_type().clone())
1700 .add_child_data(array9_id.into_data())
1701 .add_child_data(array9_list.into_data())
1702 .len(3)
1703 .build()
1704 .unwrap();
1705 let array9: ArrayRef = Arc::new(StructArray::from(array9));
1706
1707 let array10_input = vec![Some(1_i32), None, None];
1708 let mut array10_builder = PrimitiveRunBuilder::<Int16Type, Int32Type>::new();
1709 array10_builder.extend(array10_input);
1710 let array10 = array10_builder.finish();
1711
1712 let array11 = BooleanArray::from(vec![false, false, true]);
1713
1714 let array12_values = StringArray::from(vec!["x", "yy", "zzz"]);
1715 let array12_keys = Int8Array::from_iter_values([1, 1, 2]);
1716 let array12 = DictionaryArray::new(array12_keys, Arc::new(array12_values));
1717
1718 let array13 = StringArray::from(vec!["a", "bb", "ccc"]);
1719
1720 RecordBatch::try_new(
1722 Arc::new(schema.clone()),
1723 vec![
1724 Arc::new(array0),
1725 Arc::new(array1),
1726 Arc::new(array2),
1727 Arc::new(array3),
1728 Arc::new(array4),
1729 Arc::new(array5),
1730 Arc::new(array6),
1731 Arc::new(array7),
1732 Arc::new(array8),
1733 Arc::new(array9),
1734 Arc::new(array10),
1735 Arc::new(array11),
1736 Arc::new(array12),
1737 Arc::new(array13),
1738 ],
1739 )
1740 .unwrap()
1741 }
1742
1743 #[test]
1744 fn test_projection_array_values() {
1745 let schema = create_test_projection_schema();
1747
1748 let batch = create_test_projection_batch_data(&schema);
1750
1751 let mut buf = Vec::new();
1753 {
1754 let mut writer = crate::writer::FileWriter::try_new(&mut buf, &schema).unwrap();
1755 writer.write(&batch).unwrap();
1756 writer.finish().unwrap();
1757 }
1758
1759 for index in 0..12 {
1761 let projection = vec![index];
1762 let reader = FileReader::try_new(std::io::Cursor::new(buf.clone()), Some(projection));
1763 let read_batch = reader.unwrap().next().unwrap().unwrap();
1764 let projected_column = read_batch.column(0);
1765 let expected_column = batch.column(index);
1766
1767 assert_eq!(projected_column.as_ref(), expected_column.as_ref());
1769 }
1770
1771 {
1772 let reader =
1774 FileReader::try_new(std::io::Cursor::new(buf.clone()), Some(vec![3, 2, 1]));
1775 let read_batch = reader.unwrap().next().unwrap().unwrap();
1776 let expected_batch = batch.project(&[3, 2, 1]).unwrap();
1777 assert_eq!(read_batch, expected_batch);
1778 }
1779 }
1780
1781 #[test]
1782 fn test_arrow_single_float_row() {
1783 let schema = Schema::new(vec![
1784 Field::new("a", DataType::Float32, false),
1785 Field::new("b", DataType::Float32, false),
1786 Field::new("c", DataType::Int32, false),
1787 Field::new("d", DataType::Int32, false),
1788 ]);
1789 let arrays = vec![
1790 Arc::new(Float32Array::from(vec![1.23])) as ArrayRef,
1791 Arc::new(Float32Array::from(vec![-6.50])) as ArrayRef,
1792 Arc::new(Int32Array::from(vec![2])) as ArrayRef,
1793 Arc::new(Int32Array::from(vec![1])) as ArrayRef,
1794 ];
1795 let batch = RecordBatch::try_new(Arc::new(schema.clone()), arrays).unwrap();
1796 let mut file = tempfile::tempfile().unwrap();
1798 let mut stream_writer = crate::writer::StreamWriter::try_new(&mut file, &schema).unwrap();
1799 stream_writer.write(&batch).unwrap();
1800 stream_writer.finish().unwrap();
1801
1802 drop(stream_writer);
1803
1804 file.rewind().unwrap();
1805
1806 let reader = StreamReader::try_new(&mut file, None).unwrap();
1808
1809 reader.for_each(|batch| {
1810 let batch = batch.unwrap();
1811 assert!(
1812 batch
1813 .column(0)
1814 .as_any()
1815 .downcast_ref::<Float32Array>()
1816 .unwrap()
1817 .value(0)
1818 != 0.0
1819 );
1820 assert!(
1821 batch
1822 .column(1)
1823 .as_any()
1824 .downcast_ref::<Float32Array>()
1825 .unwrap()
1826 .value(0)
1827 != 0.0
1828 );
1829 });
1830
1831 file.rewind().unwrap();
1832
1833 let reader = StreamReader::try_new(file, Some(vec![0, 3])).unwrap();
1835
1836 reader.for_each(|batch| {
1837 let batch = batch.unwrap();
1838 assert_eq!(batch.schema().fields().len(), 2);
1839 assert_eq!(batch.schema().fields()[0].data_type(), &DataType::Float32);
1840 assert_eq!(batch.schema().fields()[1].data_type(), &DataType::Int32);
1841 });
1842 }
1843
1844 fn write_ipc(rb: &RecordBatch) -> Vec<u8> {
1846 let mut buf = Vec::new();
1847 let mut writer = crate::writer::FileWriter::try_new(&mut buf, rb.schema_ref()).unwrap();
1848 writer.write(rb).unwrap();
1849 writer.finish().unwrap();
1850 buf
1851 }
1852
1853 fn read_ipc(buf: &[u8]) -> Result<RecordBatch, ArrowError> {
1855 let mut reader = FileReader::try_new(std::io::Cursor::new(buf), None)?;
1856 reader.next().unwrap()
1857 }
1858
1859 fn read_ipc_skip_validation(buf: &[u8]) -> Result<RecordBatch, ArrowError> {
1862 let mut reader = unsafe {
1863 FileReader::try_new(std::io::Cursor::new(buf), None)?.with_skip_validation(true)
1864 };
1865 reader.next().unwrap()
1866 }
1867
1868 fn roundtrip_ipc(rb: &RecordBatch) -> RecordBatch {
1869 let buf = write_ipc(rb);
1870 read_ipc(&buf).unwrap()
1871 }
1872
1873 fn read_ipc_with_decoder(buf: Vec<u8>) -> Result<RecordBatch, ArrowError> {
1876 read_ipc_with_decoder_inner(buf, false)
1877 }
1878
1879 fn read_ipc_with_decoder_skip_validation(buf: Vec<u8>) -> Result<RecordBatch, ArrowError> {
1882 read_ipc_with_decoder_inner(buf, true)
1883 }
1884
1885 fn read_ipc_with_decoder_inner(
1886 buf: Vec<u8>,
1887 skip_validation: bool,
1888 ) -> Result<RecordBatch, ArrowError> {
1889 let buffer = Buffer::from_vec(buf);
1890 let trailer_start = buffer.len() - 10;
1891 let footer_len = read_footer_length(buffer[trailer_start..].try_into().unwrap())?;
1892 let footer = root_as_footer(&buffer[trailer_start - footer_len..trailer_start])
1893 .map_err(|e| ArrowError::InvalidArgumentError(format!("Invalid footer: {e}")))?;
1894
1895 let schema = fb_to_schema(footer.schema().unwrap());
1896
1897 let mut decoder = unsafe {
1898 FileDecoder::new(Arc::new(schema), footer.version())
1899 .with_skip_validation(skip_validation)
1900 };
1901 for block in footer.dictionaries().iter().flatten() {
1903 let block_len = block.bodyLength() as usize + block.metaDataLength() as usize;
1904 let data = buffer.slice_with_length(block.offset() as _, block_len);
1905 decoder.read_dictionary(block, &data)?
1906 }
1907
1908 let batches = footer.recordBatches().unwrap();
1910 assert_eq!(batches.len(), 1); let block = batches.get(0);
1913 let block_len = block.bodyLength() as usize + block.metaDataLength() as usize;
1914 let data = buffer.slice_with_length(block.offset() as _, block_len);
1915 Ok(decoder.read_record_batch(block, &data)?.unwrap())
1916 }
1917
1918 fn write_stream(rb: &RecordBatch) -> Vec<u8> {
1920 let mut buf = Vec::new();
1921 let mut writer = crate::writer::StreamWriter::try_new(&mut buf, rb.schema_ref()).unwrap();
1922 writer.write(rb).unwrap();
1923 writer.finish().unwrap();
1924 buf
1925 }
1926
1927 fn read_stream(buf: &[u8]) -> Result<RecordBatch, ArrowError> {
1929 let mut reader = StreamReader::try_new(std::io::Cursor::new(buf), None)?;
1930 reader.next().unwrap()
1931 }
1932
1933 fn read_stream_skip_validation(buf: &[u8]) -> Result<RecordBatch, ArrowError> {
1936 let mut reader = unsafe {
1937 StreamReader::try_new(std::io::Cursor::new(buf), None)?.with_skip_validation(true)
1938 };
1939 reader.next().unwrap()
1940 }
1941
1942 fn roundtrip_ipc_stream(rb: &RecordBatch) -> RecordBatch {
1943 let buf = write_stream(rb);
1944 read_stream(&buf).unwrap()
1945 }
1946
1947 #[test]
1948 fn test_roundtrip_with_custom_metadata() {
1949 let schema = Schema::new(vec![Field::new("dummy", DataType::Float64, false)]);
1950 let mut buf = Vec::new();
1951 let mut writer = crate::writer::FileWriter::try_new(&mut buf, &schema).unwrap();
1952 let mut test_metadata = HashMap::new();
1953 test_metadata.insert("abc".to_string(), "abc".to_string());
1954 test_metadata.insert("def".to_string(), "def".to_string());
1955 for (k, v) in &test_metadata {
1956 writer.write_metadata(k, v);
1957 }
1958 writer.finish().unwrap();
1959 drop(writer);
1960
1961 let reader = crate::reader::FileReader::try_new(std::io::Cursor::new(buf), None).unwrap();
1962 assert_eq!(reader.custom_metadata(), &test_metadata);
1963 }
1964
1965 #[test]
1966 fn test_roundtrip_nested_dict() {
1967 let inner: DictionaryArray<Int32Type> = vec!["a", "b", "a"].into_iter().collect();
1968
1969 let array = Arc::new(inner) as ArrayRef;
1970
1971 let dctfield = Arc::new(Field::new("dict", array.data_type().clone(), false));
1972
1973 let s = StructArray::from(vec![(dctfield, array)]);
1974 let struct_array = Arc::new(s) as ArrayRef;
1975
1976 let schema = Arc::new(Schema::new(vec![Field::new(
1977 "struct",
1978 struct_array.data_type().clone(),
1979 false,
1980 )]));
1981
1982 let batch = RecordBatch::try_new(schema, vec![struct_array]).unwrap();
1983
1984 assert_eq!(batch, roundtrip_ipc(&batch));
1985 }
1986
1987 #[test]
1988 fn test_roundtrip_nested_dict_no_preserve_dict_id() {
1989 let inner: DictionaryArray<Int32Type> = vec!["a", "b", "a"].into_iter().collect();
1990
1991 let array = Arc::new(inner) as ArrayRef;
1992
1993 let dctfield = Arc::new(Field::new("dict", array.data_type().clone(), false));
1994
1995 let s = StructArray::from(vec![(dctfield, array)]);
1996 let struct_array = Arc::new(s) as ArrayRef;
1997
1998 let schema = Arc::new(Schema::new(vec![Field::new(
1999 "struct",
2000 struct_array.data_type().clone(),
2001 false,
2002 )]));
2003
2004 let batch = RecordBatch::try_new(schema, vec![struct_array]).unwrap();
2005
2006 let mut buf = Vec::new();
2007 let mut writer = crate::writer::FileWriter::try_new_with_options(
2008 &mut buf,
2009 batch.schema_ref(),
2010 #[allow(deprecated)]
2011 IpcWriteOptions::default().with_preserve_dict_id(false),
2012 )
2013 .unwrap();
2014 writer.write(&batch).unwrap();
2015 writer.finish().unwrap();
2016 drop(writer);
2017
2018 let mut reader = FileReader::try_new(std::io::Cursor::new(buf), None).unwrap();
2019
2020 assert_eq!(batch, reader.next().unwrap().unwrap());
2021 }
2022
2023 fn check_union_with_builder(mut builder: UnionBuilder) {
2024 builder.append::<Int32Type>("a", 1).unwrap();
2025 builder.append_null::<Int32Type>("a").unwrap();
2026 builder.append::<Float64Type>("c", 3.0).unwrap();
2027 builder.append::<Int32Type>("a", 4).unwrap();
2028 builder.append::<Int64Type>("d", 11).unwrap();
2029 let union = builder.build().unwrap();
2030
2031 let schema = Arc::new(Schema::new(vec![Field::new(
2032 "union",
2033 union.data_type().clone(),
2034 false,
2035 )]));
2036
2037 let union_array = Arc::new(union) as ArrayRef;
2038
2039 let rb = RecordBatch::try_new(schema, vec![union_array]).unwrap();
2040 let rb2 = roundtrip_ipc(&rb);
2041 assert_eq!(rb.schema(), rb2.schema());
2044 assert_eq!(rb.num_columns(), rb2.num_columns());
2045 assert_eq!(rb.num_rows(), rb2.num_rows());
2046 let union1 = rb.column(0);
2047 let union2 = rb2.column(0);
2048
2049 assert_eq!(union1, union2);
2050 }
2051
2052 #[test]
2053 fn test_roundtrip_dense_union() {
2054 check_union_with_builder(UnionBuilder::new_dense());
2055 }
2056
2057 #[test]
2058 fn test_roundtrip_sparse_union() {
2059 check_union_with_builder(UnionBuilder::new_sparse());
2060 }
2061
2062 #[test]
2063 fn test_roundtrip_struct_empty_fields() {
2064 let nulls = NullBuffer::from(&[true, true, false]);
2065 let rb = RecordBatch::try_from_iter([(
2066 "",
2067 Arc::new(StructArray::new_empty_fields(nulls.len(), Some(nulls))) as _,
2068 )])
2069 .unwrap();
2070 let rb2 = roundtrip_ipc(&rb);
2071 assert_eq!(rb, rb2);
2072 }
2073
2074 #[test]
2075 fn test_roundtrip_stream_run_array_sliced() {
2076 let run_array_1: Int32RunArray = vec!["a", "a", "a", "b", "b", "c", "c", "c"]
2077 .into_iter()
2078 .collect();
2079 let run_array_1_sliced = run_array_1.slice(2, 5);
2080
2081 let run_array_2_inupt = vec![Some(1_i32), None, None, Some(2), Some(2)];
2082 let mut run_array_2_builder = PrimitiveRunBuilder::<Int16Type, Int32Type>::new();
2083 run_array_2_builder.extend(run_array_2_inupt);
2084 let run_array_2 = run_array_2_builder.finish();
2085
2086 let schema = Arc::new(Schema::new(vec![
2087 Field::new(
2088 "run_array_1_sliced",
2089 run_array_1_sliced.data_type().clone(),
2090 false,
2091 ),
2092 Field::new("run_array_2", run_array_2.data_type().clone(), false),
2093 ]));
2094 let input_batch = RecordBatch::try_new(
2095 schema,
2096 vec![Arc::new(run_array_1_sliced.clone()), Arc::new(run_array_2)],
2097 )
2098 .unwrap();
2099 let output_batch = roundtrip_ipc_stream(&input_batch);
2100
2101 assert_eq!(input_batch.column(1), output_batch.column(1));
2105
2106 let run_array_1_unsliced = unslice_run_array(run_array_1_sliced.into_data()).unwrap();
2107 assert_eq!(run_array_1_unsliced, output_batch.column(0).into_data());
2108 }
2109
2110 #[test]
2111 fn test_roundtrip_stream_nested_dict() {
2112 let xs = vec!["AA", "BB", "AA", "CC", "BB"];
2113 let dict = Arc::new(
2114 xs.clone()
2115 .into_iter()
2116 .collect::<DictionaryArray<Int8Type>>(),
2117 );
2118 let string_array: ArrayRef = Arc::new(StringArray::from(xs.clone()));
2119 let struct_array = StructArray::from(vec![
2120 (
2121 Arc::new(Field::new("f2.1", DataType::Utf8, false)),
2122 string_array,
2123 ),
2124 (
2125 Arc::new(Field::new("f2.2_struct", dict.data_type().clone(), false)),
2126 dict.clone() as ArrayRef,
2127 ),
2128 ]);
2129 let schema = Arc::new(Schema::new(vec![
2130 Field::new("f1_string", DataType::Utf8, false),
2131 Field::new("f2_struct", struct_array.data_type().clone(), false),
2132 ]));
2133 let input_batch = RecordBatch::try_new(
2134 schema,
2135 vec![
2136 Arc::new(StringArray::from(xs.clone())),
2137 Arc::new(struct_array),
2138 ],
2139 )
2140 .unwrap();
2141 let output_batch = roundtrip_ipc_stream(&input_batch);
2142 assert_eq!(input_batch, output_batch);
2143 }
2144
2145 #[test]
2146 fn test_roundtrip_stream_nested_dict_of_map_of_dict() {
2147 let values = StringArray::from(vec![Some("a"), None, Some("b"), Some("c")]);
2148 let values = Arc::new(values) as ArrayRef;
2149 let value_dict_keys = Int8Array::from_iter_values([0, 1, 1, 2, 3, 1]);
2150 let value_dict_array = DictionaryArray::new(value_dict_keys, values.clone());
2151
2152 let key_dict_keys = Int8Array::from_iter_values([0, 0, 2, 1, 1, 3]);
2153 let key_dict_array = DictionaryArray::new(key_dict_keys, values);
2154
2155 #[allow(deprecated)]
2156 let keys_field = Arc::new(Field::new_dict(
2157 "keys",
2158 DataType::Dictionary(Box::new(DataType::Int8), Box::new(DataType::Utf8)),
2159 true, 1,
2161 false,
2162 ));
2163 #[allow(deprecated)]
2164 let values_field = Arc::new(Field::new_dict(
2165 "values",
2166 DataType::Dictionary(Box::new(DataType::Int8), Box::new(DataType::Utf8)),
2167 true,
2168 2,
2169 false,
2170 ));
2171 let entry_struct = StructArray::from(vec![
2172 (keys_field, make_array(key_dict_array.into_data())),
2173 (values_field, make_array(value_dict_array.into_data())),
2174 ]);
2175 let map_data_type = DataType::Map(
2176 Arc::new(Field::new(
2177 "entries",
2178 entry_struct.data_type().clone(),
2179 false,
2180 )),
2181 false,
2182 );
2183
2184 let entry_offsets = Buffer::from_slice_ref([0, 2, 4, 6]);
2185 let map_data = ArrayData::builder(map_data_type)
2186 .len(3)
2187 .add_buffer(entry_offsets)
2188 .add_child_data(entry_struct.into_data())
2189 .build()
2190 .unwrap();
2191 let map_array = MapArray::from(map_data);
2192
2193 let dict_keys = Int8Array::from_iter_values([0, 1, 1, 2, 2, 1]);
2194 let dict_dict_array = DictionaryArray::new(dict_keys, Arc::new(map_array));
2195
2196 let schema = Arc::new(Schema::new(vec![Field::new(
2197 "f1",
2198 dict_dict_array.data_type().clone(),
2199 false,
2200 )]));
2201 let input_batch = RecordBatch::try_new(schema, vec![Arc::new(dict_dict_array)]).unwrap();
2202 let output_batch = roundtrip_ipc_stream(&input_batch);
2203 assert_eq!(input_batch, output_batch);
2204 }
2205
2206 fn test_roundtrip_stream_dict_of_list_of_dict_impl<
2207 OffsetSize: OffsetSizeTrait,
2208 U: ArrowNativeType,
2209 >(
2210 list_data_type: DataType,
2211 offsets: &[U; 5],
2212 ) {
2213 let values = StringArray::from(vec![Some("a"), None, Some("c"), None]);
2214 let keys = Int8Array::from_iter_values([0, 0, 1, 2, 0, 1, 3]);
2215 let dict_array = DictionaryArray::new(keys, Arc::new(values));
2216 let dict_data = dict_array.to_data();
2217
2218 let value_offsets = Buffer::from_slice_ref(offsets);
2219
2220 let list_data = ArrayData::builder(list_data_type)
2221 .len(4)
2222 .add_buffer(value_offsets)
2223 .add_child_data(dict_data)
2224 .build()
2225 .unwrap();
2226 let list_array = GenericListArray::<OffsetSize>::from(list_data);
2227
2228 let keys_for_dict = Int8Array::from_iter_values([0, 3, 0, 1, 1, 2, 0, 1, 3]);
2229 let dict_dict_array = DictionaryArray::new(keys_for_dict, Arc::new(list_array));
2230
2231 let schema = Arc::new(Schema::new(vec![Field::new(
2232 "f1",
2233 dict_dict_array.data_type().clone(),
2234 false,
2235 )]));
2236 let input_batch = RecordBatch::try_new(schema, vec![Arc::new(dict_dict_array)]).unwrap();
2237 let output_batch = roundtrip_ipc_stream(&input_batch);
2238 assert_eq!(input_batch, output_batch);
2239 }
2240
2241 #[test]
2242 fn test_roundtrip_stream_dict_of_list_of_dict() {
2243 #[allow(deprecated)]
2245 let list_data_type = DataType::List(Arc::new(Field::new_dict(
2246 "item",
2247 DataType::Dictionary(Box::new(DataType::Int8), Box::new(DataType::Utf8)),
2248 true,
2249 1,
2250 false,
2251 )));
2252 let offsets: &[i32; 5] = &[0, 2, 4, 4, 6];
2253 test_roundtrip_stream_dict_of_list_of_dict_impl::<i32, i32>(list_data_type, offsets);
2254
2255 #[allow(deprecated)]
2257 let list_data_type = DataType::LargeList(Arc::new(Field::new_dict(
2258 "item",
2259 DataType::Dictionary(Box::new(DataType::Int8), Box::new(DataType::Utf8)),
2260 true,
2261 1,
2262 false,
2263 )));
2264 let offsets: &[i64; 5] = &[0, 2, 4, 4, 7];
2265 test_roundtrip_stream_dict_of_list_of_dict_impl::<i64, i64>(list_data_type, offsets);
2266 }
2267
2268 #[test]
2269 fn test_roundtrip_stream_dict_of_fixed_size_list_of_dict() {
2270 let values = StringArray::from(vec![Some("a"), None, Some("c"), None]);
2271 let keys = Int8Array::from_iter_values([0, 0, 1, 2, 0, 1, 3, 1, 2]);
2272 let dict_array = DictionaryArray::new(keys, Arc::new(values));
2273 let dict_data = dict_array.into_data();
2274
2275 #[allow(deprecated)]
2276 let list_data_type = DataType::FixedSizeList(
2277 Arc::new(Field::new_dict(
2278 "item",
2279 DataType::Dictionary(Box::new(DataType::Int8), Box::new(DataType::Utf8)),
2280 true,
2281 1,
2282 false,
2283 )),
2284 3,
2285 );
2286 let list_data = ArrayData::builder(list_data_type)
2287 .len(3)
2288 .add_child_data(dict_data)
2289 .build()
2290 .unwrap();
2291 let list_array = FixedSizeListArray::from(list_data);
2292
2293 let keys_for_dict = Int8Array::from_iter_values([0, 1, 0, 1, 1, 2, 0, 1, 2]);
2294 let dict_dict_array = DictionaryArray::new(keys_for_dict, Arc::new(list_array));
2295
2296 let schema = Arc::new(Schema::new(vec![Field::new(
2297 "f1",
2298 dict_dict_array.data_type().clone(),
2299 false,
2300 )]));
2301 let input_batch = RecordBatch::try_new(schema, vec![Arc::new(dict_dict_array)]).unwrap();
2302 let output_batch = roundtrip_ipc_stream(&input_batch);
2303 assert_eq!(input_batch, output_batch);
2304 }
2305
2306 const LONG_TEST_STRING: &str =
2307 "This is a long string to make sure binary view array handles it";
2308
2309 #[test]
2310 fn test_roundtrip_view_types() {
2311 let schema = Schema::new(vec![
2312 Field::new("field_1", DataType::BinaryView, true),
2313 Field::new("field_2", DataType::Utf8, true),
2314 Field::new("field_3", DataType::Utf8View, true),
2315 ]);
2316 let bin_values: Vec<Option<&[u8]>> = vec![
2317 Some(b"foo"),
2318 None,
2319 Some(b"bar"),
2320 Some(LONG_TEST_STRING.as_bytes()),
2321 ];
2322 let utf8_values: Vec<Option<&str>> =
2323 vec![Some("foo"), None, Some("bar"), Some(LONG_TEST_STRING)];
2324 let bin_view_array = BinaryViewArray::from_iter(bin_values);
2325 let utf8_array = StringArray::from_iter(utf8_values.iter());
2326 let utf8_view_array = StringViewArray::from_iter(utf8_values);
2327 let record_batch = RecordBatch::try_new(
2328 Arc::new(schema.clone()),
2329 vec![
2330 Arc::new(bin_view_array),
2331 Arc::new(utf8_array),
2332 Arc::new(utf8_view_array),
2333 ],
2334 )
2335 .unwrap();
2336
2337 assert_eq!(record_batch, roundtrip_ipc(&record_batch));
2338 assert_eq!(record_batch, roundtrip_ipc_stream(&record_batch));
2339
2340 let sliced_batch = record_batch.slice(1, 2);
2341 assert_eq!(sliced_batch, roundtrip_ipc(&sliced_batch));
2342 assert_eq!(sliced_batch, roundtrip_ipc_stream(&sliced_batch));
2343 }
2344
2345 #[test]
2346 fn test_roundtrip_view_types_nested_dict() {
2347 let bin_values: Vec<Option<&[u8]>> = vec![
2348 Some(b"foo"),
2349 None,
2350 Some(b"bar"),
2351 Some(LONG_TEST_STRING.as_bytes()),
2352 Some(b"field"),
2353 ];
2354 let utf8_values: Vec<Option<&str>> = vec![
2355 Some("foo"),
2356 None,
2357 Some("bar"),
2358 Some(LONG_TEST_STRING),
2359 Some("field"),
2360 ];
2361 let bin_view_array = Arc::new(BinaryViewArray::from_iter(bin_values));
2362 let utf8_view_array = Arc::new(StringViewArray::from_iter(utf8_values));
2363
2364 let key_dict_keys = Int8Array::from_iter_values([0, 0, 1, 2, 0, 1, 3]);
2365 let key_dict_array = DictionaryArray::new(key_dict_keys, utf8_view_array.clone());
2366 #[allow(deprecated)]
2367 let keys_field = Arc::new(Field::new_dict(
2368 "keys",
2369 DataType::Dictionary(Box::new(DataType::Int8), Box::new(DataType::Utf8View)),
2370 true,
2371 1,
2372 false,
2373 ));
2374
2375 let value_dict_keys = Int8Array::from_iter_values([0, 3, 0, 1, 2, 0, 1]);
2376 let value_dict_array = DictionaryArray::new(value_dict_keys, bin_view_array);
2377 #[allow(deprecated)]
2378 let values_field = Arc::new(Field::new_dict(
2379 "values",
2380 DataType::Dictionary(Box::new(DataType::Int8), Box::new(DataType::BinaryView)),
2381 true,
2382 2,
2383 false,
2384 ));
2385 let entry_struct = StructArray::from(vec![
2386 (keys_field, make_array(key_dict_array.into_data())),
2387 (values_field, make_array(value_dict_array.into_data())),
2388 ]);
2389
2390 let map_data_type = DataType::Map(
2391 Arc::new(Field::new(
2392 "entries",
2393 entry_struct.data_type().clone(),
2394 false,
2395 )),
2396 false,
2397 );
2398 let entry_offsets = Buffer::from_slice_ref([0, 2, 4, 7]);
2399 let map_data = ArrayData::builder(map_data_type)
2400 .len(3)
2401 .add_buffer(entry_offsets)
2402 .add_child_data(entry_struct.into_data())
2403 .build()
2404 .unwrap();
2405 let map_array = MapArray::from(map_data);
2406
2407 let dict_keys = Int8Array::from_iter_values([0, 1, 0, 1, 1, 2, 0, 1, 2]);
2408 let dict_dict_array = DictionaryArray::new(dict_keys, Arc::new(map_array));
2409 let schema = Arc::new(Schema::new(vec![Field::new(
2410 "f1",
2411 dict_dict_array.data_type().clone(),
2412 false,
2413 )]));
2414 let batch = RecordBatch::try_new(schema, vec![Arc::new(dict_dict_array)]).unwrap();
2415 assert_eq!(batch, roundtrip_ipc(&batch));
2416 assert_eq!(batch, roundtrip_ipc_stream(&batch));
2417
2418 let sliced_batch = batch.slice(1, 2);
2419 assert_eq!(sliced_batch, roundtrip_ipc(&sliced_batch));
2420 assert_eq!(sliced_batch, roundtrip_ipc_stream(&sliced_batch));
2421 }
2422
2423 #[test]
2424 fn test_no_columns_batch() {
2425 let schema = Arc::new(Schema::empty());
2426 let options = RecordBatchOptions::new()
2427 .with_match_field_names(true)
2428 .with_row_count(Some(10));
2429 let input_batch = RecordBatch::try_new_with_options(schema, vec![], &options).unwrap();
2430 let output_batch = roundtrip_ipc_stream(&input_batch);
2431 assert_eq!(input_batch, output_batch);
2432 }
2433
2434 #[test]
2435 fn test_unaligned() {
2436 let batch = RecordBatch::try_from_iter(vec![(
2437 "i32",
2438 Arc::new(Int32Array::from(vec![1, 2, 3, 4])) as _,
2439 )])
2440 .unwrap();
2441
2442 let gen = IpcDataGenerator {};
2443 #[allow(deprecated)]
2444 let mut dict_tracker = DictionaryTracker::new_with_preserve_dict_id(false, true);
2445 let (_, encoded) = gen
2446 .encoded_batch(&batch, &mut dict_tracker, &Default::default())
2447 .unwrap();
2448
2449 let message = root_as_message(&encoded.ipc_message).unwrap();
2450
2451 let mut buffer = MutableBuffer::with_capacity(encoded.arrow_data.len() + 1);
2453 buffer.push(0_u8);
2454 buffer.extend_from_slice(&encoded.arrow_data);
2455 let b = Buffer::from(buffer).slice(1);
2456 assert_ne!(b.as_ptr().align_offset(8), 0);
2457
2458 let ipc_batch = message.header_as_record_batch().unwrap();
2459 let roundtrip = RecordBatchDecoder::try_new(
2460 &b,
2461 ipc_batch,
2462 batch.schema(),
2463 &Default::default(),
2464 &message.version(),
2465 )
2466 .unwrap()
2467 .with_require_alignment(false)
2468 .read_record_batch()
2469 .unwrap();
2470 assert_eq!(batch, roundtrip);
2471 }
2472
2473 #[test]
2474 fn test_unaligned_throws_error_with_require_alignment() {
2475 let batch = RecordBatch::try_from_iter(vec![(
2476 "i32",
2477 Arc::new(Int32Array::from(vec![1, 2, 3, 4])) as _,
2478 )])
2479 .unwrap();
2480
2481 let gen = IpcDataGenerator {};
2482 #[allow(deprecated)]
2483 let mut dict_tracker = DictionaryTracker::new_with_preserve_dict_id(false, true);
2484 let (_, encoded) = gen
2485 .encoded_batch(&batch, &mut dict_tracker, &Default::default())
2486 .unwrap();
2487
2488 let message = root_as_message(&encoded.ipc_message).unwrap();
2489
2490 let mut buffer = MutableBuffer::with_capacity(encoded.arrow_data.len() + 1);
2492 buffer.push(0_u8);
2493 buffer.extend_from_slice(&encoded.arrow_data);
2494 let b = Buffer::from(buffer).slice(1);
2495 assert_ne!(b.as_ptr().align_offset(8), 0);
2496
2497 let ipc_batch = message.header_as_record_batch().unwrap();
2498 let result = RecordBatchDecoder::try_new(
2499 &b,
2500 ipc_batch,
2501 batch.schema(),
2502 &Default::default(),
2503 &message.version(),
2504 )
2505 .unwrap()
2506 .with_require_alignment(true)
2507 .read_record_batch();
2508
2509 let error = result.unwrap_err();
2510 assert_eq!(
2511 error.to_string(),
2512 "Invalid argument error: Misaligned buffers[0] in array of type Int32, \
2513 offset from expected alignment of 4 by 1"
2514 );
2515 }
2516
2517 #[test]
2518 fn test_file_with_massive_column_count() {
2519 let limit = 600_000;
2521
2522 let fields = (0..limit)
2523 .map(|i| Field::new(format!("{i}"), DataType::Boolean, false))
2524 .collect::<Vec<_>>();
2525 let schema = Arc::new(Schema::new(fields));
2526 let batch = RecordBatch::new_empty(schema);
2527
2528 let mut buf = Vec::new();
2529 let mut writer = crate::writer::FileWriter::try_new(&mut buf, batch.schema_ref()).unwrap();
2530 writer.write(&batch).unwrap();
2531 writer.finish().unwrap();
2532 drop(writer);
2533
2534 let mut reader = FileReaderBuilder::new()
2535 .with_max_footer_fb_tables(1_500_000)
2536 .build(std::io::Cursor::new(buf))
2537 .unwrap();
2538 let roundtrip_batch = reader.next().unwrap().unwrap();
2539
2540 assert_eq!(batch, roundtrip_batch);
2541 }
2542
2543 #[test]
2544 fn test_file_with_deeply_nested_columns() {
2545 let limit = 61;
2547
2548 let fields = (0..limit).fold(
2549 vec![Field::new("leaf", DataType::Boolean, false)],
2550 |field, index| vec![Field::new_struct(format!("{index}"), field, false)],
2551 );
2552 let schema = Arc::new(Schema::new(fields));
2553 let batch = RecordBatch::new_empty(schema);
2554
2555 let mut buf = Vec::new();
2556 let mut writer = crate::writer::FileWriter::try_new(&mut buf, batch.schema_ref()).unwrap();
2557 writer.write(&batch).unwrap();
2558 writer.finish().unwrap();
2559 drop(writer);
2560
2561 let mut reader = FileReaderBuilder::new()
2562 .with_max_footer_fb_depth(65)
2563 .build(std::io::Cursor::new(buf))
2564 .unwrap();
2565 let roundtrip_batch = reader.next().unwrap().unwrap();
2566
2567 assert_eq!(batch, roundtrip_batch);
2568 }
2569
2570 #[test]
2571 fn test_invalid_struct_array_ipc_read_errors() {
2572 let a_field = Field::new("a", DataType::Int32, false);
2573 let b_field = Field::new("b", DataType::Int32, false);
2574 let struct_fields = Fields::from(vec![a_field.clone(), b_field.clone()]);
2575
2576 let a_array_data = ArrayData::builder(a_field.data_type().clone())
2577 .len(4)
2578 .add_buffer(Buffer::from_slice_ref([1, 2, 3, 4]))
2579 .build()
2580 .unwrap();
2581 let b_array_data = ArrayData::builder(b_field.data_type().clone())
2582 .len(3)
2583 .add_buffer(Buffer::from_slice_ref([5, 6, 7]))
2584 .build()
2585 .unwrap();
2586
2587 let invalid_struct_arr = unsafe {
2588 StructArray::new_unchecked(
2589 struct_fields,
2590 vec![make_array(a_array_data), make_array(b_array_data)],
2591 None,
2592 )
2593 };
2594
2595 expect_ipc_validation_error(
2596 Arc::new(invalid_struct_arr),
2597 "Invalid argument error: Incorrect array length for StructArray field \"b\", expected 4 got 3",
2598 );
2599 }
2600
2601 #[test]
2602 fn test_invalid_nested_array_ipc_read_errors() {
2603 let a_field = Field::new("a", DataType::Int32, false);
2605 let b_field = Field::new("b", DataType::Utf8, false);
2606
2607 let schema = Arc::new(Schema::new(vec![Field::new_struct(
2608 "s",
2609 vec![a_field.clone(), b_field.clone()],
2610 false,
2611 )]));
2612
2613 let a_array_data = ArrayData::builder(a_field.data_type().clone())
2614 .len(4)
2615 .add_buffer(Buffer::from_slice_ref([1, 2, 3, 4]))
2616 .build()
2617 .unwrap();
2618 let b_array_data = {
2620 let valid: &[u8] = b" ";
2621 let mut invalid = vec![];
2622 invalid.extend_from_slice(b"ValidString");
2623 invalid.extend_from_slice(INVALID_UTF8_FIRST_CHAR);
2624 let binary_array =
2625 BinaryArray::from_iter(vec![None, Some(valid), None, Some(&invalid)]);
2626 let array = unsafe {
2627 StringArray::new_unchecked(
2628 binary_array.offsets().clone(),
2629 binary_array.values().clone(),
2630 binary_array.nulls().cloned(),
2631 )
2632 };
2633 array.into_data()
2634 };
2635 let struct_data_type = schema.field(0).data_type();
2636
2637 let invalid_struct_arr = unsafe {
2638 make_array(
2639 ArrayData::builder(struct_data_type.clone())
2640 .len(4)
2641 .add_child_data(a_array_data)
2642 .add_child_data(b_array_data)
2643 .build_unchecked(),
2644 )
2645 };
2646 expect_ipc_validation_error(
2647 Arc::new(invalid_struct_arr),
2648 "Invalid argument error: Invalid UTF8 sequence at string index 3 (3..18): invalid utf-8 sequence of 1 bytes from index 11",
2649 );
2650 }
2651
2652 #[test]
2653 fn test_same_dict_id_without_preserve() {
2654 let batch = RecordBatch::try_new(
2655 Arc::new(Schema::new(
2656 ["a", "b"]
2657 .iter()
2658 .map(|name| {
2659 #[allow(deprecated)]
2660 Field::new_dict(
2661 name.to_string(),
2662 DataType::Dictionary(
2663 Box::new(DataType::Int32),
2664 Box::new(DataType::Utf8),
2665 ),
2666 true,
2667 0,
2668 false,
2669 )
2670 })
2671 .collect::<Vec<Field>>(),
2672 )),
2673 vec![
2674 Arc::new(
2675 vec![Some("c"), Some("d")]
2676 .into_iter()
2677 .collect::<DictionaryArray<Int32Type>>(),
2678 ) as ArrayRef,
2679 Arc::new(
2680 vec![Some("e"), Some("f")]
2681 .into_iter()
2682 .collect::<DictionaryArray<Int32Type>>(),
2683 ) as ArrayRef,
2684 ],
2685 )
2686 .expect("Failed to create RecordBatch");
2687
2688 let mut buf = vec![];
2690 {
2691 let mut writer = crate::writer::StreamWriter::try_new_with_options(
2692 &mut buf,
2693 batch.schema().as_ref(),
2694 #[allow(deprecated)]
2695 crate::writer::IpcWriteOptions::default().with_preserve_dict_id(false),
2696 )
2697 .expect("Failed to create StreamWriter");
2698 writer.write(&batch).expect("Failed to write RecordBatch");
2699 writer.finish().expect("Failed to finish StreamWriter");
2700 }
2701
2702 StreamReader::try_new(std::io::Cursor::new(buf), None)
2703 .expect("Failed to create StreamReader")
2704 .for_each(|decoded_batch| {
2705 assert_eq!(decoded_batch.expect("Failed to read RecordBatch"), batch);
2706 });
2707 }
2708
2709 #[test]
2710 fn test_validation_of_invalid_list_array() {
2711 let array = unsafe {
2713 let values = Int32Array::from(vec![1, 2, 3]);
2714 let bad_offsets = ScalarBuffer::<i32>::from(vec![0, 2, 4, 2]); let offsets = OffsetBuffer::new_unchecked(bad_offsets); let field = Field::new_list_field(DataType::Int32, true);
2717 let nulls = None;
2718 ListArray::new(Arc::new(field), offsets, Arc::new(values), nulls)
2719 };
2720
2721 expect_ipc_validation_error(
2722 Arc::new(array),
2723 "Invalid argument error: Offset invariant failure: offset at position 2 out of bounds: 4 > 2"
2724 );
2725 }
2726
2727 #[test]
2728 fn test_validation_of_invalid_string_array() {
2729 let valid: &[u8] = b" ";
2730 let mut invalid = vec![];
2731 invalid.extend_from_slice(b"ThisStringIsCertainlyLongerThan12Bytes");
2732 invalid.extend_from_slice(INVALID_UTF8_FIRST_CHAR);
2733 let binary_array = BinaryArray::from_iter(vec![None, Some(valid), None, Some(&invalid)]);
2734 let array = unsafe {
2737 StringArray::new_unchecked(
2738 binary_array.offsets().clone(),
2739 binary_array.values().clone(),
2740 binary_array.nulls().cloned(),
2741 )
2742 };
2743 expect_ipc_validation_error(
2744 Arc::new(array),
2745 "Invalid argument error: Invalid UTF8 sequence at string index 3 (3..45): invalid utf-8 sequence of 1 bytes from index 38"
2746 );
2747 }
2748
2749 #[test]
2750 fn test_validation_of_invalid_string_view_array() {
2751 let valid: &[u8] = b" ";
2752 let mut invalid = vec![];
2753 invalid.extend_from_slice(b"ThisStringIsCertainlyLongerThan12Bytes");
2754 invalid.extend_from_slice(INVALID_UTF8_FIRST_CHAR);
2755 let binary_view_array =
2756 BinaryViewArray::from_iter(vec![None, Some(valid), None, Some(&invalid)]);
2757 let array = unsafe {
2760 StringViewArray::new_unchecked(
2761 binary_view_array.views().clone(),
2762 binary_view_array.data_buffers().to_vec(),
2763 binary_view_array.nulls().cloned(),
2764 )
2765 };
2766 expect_ipc_validation_error(
2767 Arc::new(array),
2768 "Invalid argument error: Encountered non-UTF-8 data at index 3: invalid utf-8 sequence of 1 bytes from index 38"
2769 );
2770 }
2771
2772 #[test]
2775 fn test_validation_of_invalid_dictionary_array() {
2776 let array = unsafe {
2777 let values = StringArray::from_iter_values(["a", "b", "c"]);
2778 let keys = Int32Array::from(vec![1, 200]); DictionaryArray::new_unchecked(keys, Arc::new(values))
2780 };
2781
2782 expect_ipc_validation_error(
2783 Arc::new(array),
2784 "Invalid argument error: Value at position 1 out of bounds: 200 (should be in [0, 2])",
2785 );
2786 }
2787
2788 #[test]
2789 fn test_validation_of_invalid_union_array() {
2790 let array = unsafe {
2791 let fields = UnionFields::new(
2792 vec![1, 3], vec![
2794 Field::new("a", DataType::Int32, false),
2795 Field::new("b", DataType::Utf8, false),
2796 ],
2797 );
2798 let type_ids = ScalarBuffer::from(vec![1i8, 2, 3]); let offsets = None;
2800 let children: Vec<ArrayRef> = vec![
2801 Arc::new(Int32Array::from(vec![10, 20, 30])),
2802 Arc::new(StringArray::from(vec![Some("a"), Some("b"), Some("c")])),
2803 ];
2804
2805 UnionArray::new_unchecked(fields, type_ids, offsets, children)
2806 };
2807
2808 expect_ipc_validation_error(
2809 Arc::new(array),
2810 "Invalid argument error: Type Ids values must match one of the field type ids",
2811 );
2812 }
2813
2814 const INVALID_UTF8_FIRST_CHAR: &[u8] = &[0xa0, 0xa1, 0x20, 0x20];
2817
2818 fn expect_ipc_validation_error(array: ArrayRef, expected_err: &str) {
2820 let rb = RecordBatch::try_from_iter([("a", array)]).unwrap();
2821
2822 let buf = write_stream(&rb); read_stream_skip_validation(&buf).unwrap();
2825 let err = read_stream(&buf).unwrap_err();
2826 assert_eq!(err.to_string(), expected_err);
2827
2828 let buf = write_ipc(&rb); read_ipc_skip_validation(&buf).unwrap();
2831 let err = read_ipc(&buf).unwrap_err();
2832 assert_eq!(err.to_string(), expected_err);
2833
2834 read_ipc_with_decoder_skip_validation(buf.clone()).unwrap();
2836 let err = read_ipc_with_decoder(buf).unwrap_err();
2837 assert_eq!(err.to_string(), expected_err);
2838 }
2839
2840 #[test]
2841 fn test_roundtrip_schema() {
2842 let schema = Schema::new(vec![
2843 Field::new(
2844 "a",
2845 DataType::Dictionary(Box::new(DataType::UInt16), Box::new(DataType::Utf8)),
2846 false,
2847 ),
2848 Field::new(
2849 "b",
2850 DataType::Dictionary(Box::new(DataType::UInt16), Box::new(DataType::Utf8)),
2851 false,
2852 ),
2853 ]);
2854
2855 let options = IpcWriteOptions::default();
2856 let data_gen = IpcDataGenerator::default();
2857 let mut dict_tracker = DictionaryTracker::new(false);
2858 let encoded_data =
2859 data_gen.schema_to_bytes_with_dictionary_tracker(&schema, &mut dict_tracker, &options);
2860 let mut schema_bytes = vec![];
2861 write_message(&mut schema_bytes, encoded_data, &options).expect("write_message");
2862
2863 let begin_offset: usize = if schema_bytes[0..4].eq(&CONTINUATION_MARKER) {
2864 4
2865 } else {
2866 0
2867 };
2868
2869 size_prefixed_root_as_message(&schema_bytes[begin_offset..])
2870 .expect_err("size_prefixed_root_as_message");
2871
2872 let msg = parse_message(&schema_bytes).expect("parse_message");
2873 let ipc_schema = msg.header_as_schema().expect("header_as_schema");
2874 let new_schema = fb_to_schema(ipc_schema);
2875
2876 assert_eq!(schema, new_schema);
2877 }
2878}