1#![doc(
33 html_logo_url = "https://arrow.apache.org/img/arrow-logo_chevrons_black-txt_white-bg.svg",
34 html_favicon_url = "https://arrow.apache.org/img/arrow-logo_chevrons_black-txt_transparent-bg.svg"
35)]
36#![cfg_attr(docsrs, feature(doc_cfg))]
37#![warn(missing_docs)]
38use arrow_buffer::{IntervalDayTime, IntervalMonthDayNano, ScalarBuffer};
39use hex::decode;
40use num_bigint::BigInt;
41use num_traits::Signed;
42use serde::{Deserialize, Serialize};
43use serde_json::{Map as SJMap, Value};
44use std::collections::HashMap;
45use std::sync::Arc;
46
47use arrow::array::*;
48use arrow::buffer::{Buffer, MutableBuffer};
49use arrow::datatypes::*;
50use arrow::error::{ArrowError, Result};
51use arrow::util::bit_util;
52
53mod datatype;
54mod field;
55mod schema;
56
57pub use datatype::*;
58pub use field::*;
59pub use schema::*;
60
61#[derive(Deserialize, Serialize, Debug)]
65pub struct ArrowJson {
66 pub schema: ArrowJsonSchema,
68 pub batches: Vec<ArrowJsonBatch>,
70 #[serde(skip_serializing_if = "Option::is_none")]
72 pub dictionaries: Option<Vec<ArrowJsonDictionaryBatch>>,
73}
74
75#[derive(Deserialize, Serialize, Debug)]
79pub struct ArrowJsonSchema {
80 pub fields: Vec<ArrowJsonField>,
82 #[serde(skip_serializing_if = "Option::is_none")]
84 pub metadata: Option<Vec<HashMap<String, String>>>,
85}
86
87#[derive(Deserialize, Serialize, Debug)]
89pub struct ArrowJsonField {
90 pub name: String,
92 #[serde(rename = "type")]
95 pub field_type: Value,
96 pub nullable: bool,
98 pub children: Vec<ArrowJsonField>,
100 #[serde(skip_serializing_if = "Option::is_none")]
102 pub dictionary: Option<ArrowJsonFieldDictionary>,
103 #[serde(skip_serializing_if = "Option::is_none")]
105 pub metadata: Option<Value>,
106}
107
108impl From<&FieldRef> for ArrowJsonField {
109 fn from(value: &FieldRef) -> Self {
110 Self::from(value.as_ref())
111 }
112}
113
114impl From<&Field> for ArrowJsonField {
115 fn from(field: &Field) -> Self {
116 let metadata_value = match field.metadata().is_empty() {
117 false => {
118 let mut array = Vec::new();
119 for (k, v) in field.metadata() {
120 let mut kv_map = SJMap::new();
121 kv_map.insert(k.clone(), Value::String(v.clone()));
122 array.push(Value::Object(kv_map));
123 }
124 if !array.is_empty() {
125 Some(Value::Array(array))
126 } else {
127 None
128 }
129 }
130 _ => None,
131 };
132
133 Self {
134 name: field.name().clone(),
135 field_type: data_type_to_json(field.data_type()),
136 nullable: field.is_nullable(),
137 children: vec![],
138 dictionary: None, metadata: metadata_value,
140 }
141 }
142}
143
144#[derive(Deserialize, Serialize, Debug)]
146pub struct ArrowJsonFieldDictionary {
147 pub id: i64,
149 #[serde(rename = "indexType")]
151 pub index_type: DictionaryIndexType,
152 #[serde(rename = "isOrdered")]
154 pub is_ordered: bool,
155}
156
157#[derive(Deserialize, Serialize, Debug)]
159pub struct DictionaryIndexType {
160 pub name: String,
162 #[serde(rename = "isSigned")]
164 pub is_signed: bool,
165 #[serde(rename = "bitWidth")]
167 pub bit_width: i64,
168}
169
170#[derive(Deserialize, Serialize, Debug, Clone)]
172pub struct ArrowJsonBatch {
173 count: usize,
174 pub columns: Vec<ArrowJsonColumn>,
176}
177
178#[derive(Deserialize, Serialize, Debug, Clone)]
180pub struct ArrowJsonDictionaryBatch {
181 pub id: i64,
183 pub data: ArrowJsonBatch,
185}
186
187#[derive(Deserialize, Serialize, Clone, Debug)]
189pub struct ArrowJsonColumn {
190 name: String,
191 pub count: usize,
193 #[serde(rename = "VALIDITY")]
195 pub validity: Option<Vec<u8>>,
196 #[serde(rename = "DATA")]
198 pub data: Option<Vec<Value>>,
199 #[serde(rename = "OFFSET")]
201 pub offset: Option<Vec<Value>>, #[serde(rename = "TYPE_ID")]
204 pub type_id: Option<Vec<i8>>,
205 #[serde(rename = "SIZE")]
207 pub size: Option<Vec<Value>>,
208 #[serde(rename = "VIEWS")]
210 pub views: Option<Vec<Value>>,
211 #[serde(rename = "VARIADIC_DATA_BUFFERS")]
213 pub variadic_data_buffers: Option<Vec<String>>,
214 pub children: Option<Vec<ArrowJsonColumn>>,
216}
217
218impl ArrowJson {
219 pub fn equals_reader(&self, reader: &mut dyn RecordBatchReader) -> Result<bool> {
221 if !self.schema.equals_schema(&reader.schema()) {
222 return Ok(false);
223 }
224
225 for json_batch in self.get_record_batches()? {
226 let batch = reader.next();
227 match batch {
228 Some(Ok(batch)) => {
229 if json_batch != batch {
230 println!("json: {json_batch:?}");
231 println!("batch: {batch:?}");
232 return Ok(false);
233 }
234 }
235 Some(Err(e)) => return Err(e),
236 None => return Ok(false),
237 }
238 }
239
240 Ok(true)
241 }
242
243 pub fn get_record_batches(&self) -> Result<Vec<RecordBatch>> {
245 let schema = self.schema.to_arrow_schema()?;
246
247 let mut dictionaries = HashMap::new();
248 self.dictionaries.iter().for_each(|dict_batches| {
249 dict_batches.iter().for_each(|d| {
250 dictionaries.insert(d.id, d.clone());
251 });
252 });
253
254 let batches: Result<Vec<_>> = self
255 .batches
256 .iter()
257 .map(|col| record_batch_from_json(&schema, col.clone(), Some(&dictionaries)))
258 .collect();
259
260 batches
261 }
262}
263
264impl ArrowJsonSchema {
265 fn equals_schema(&self, schema: &Schema) -> bool {
267 let field_len = self.fields.len();
268 if field_len != schema.fields().len() {
269 return false;
270 }
271 for i in 0..field_len {
272 let json_field = &self.fields[i];
273 let field = schema.field(i);
274 if !json_field.equals_field(field) {
275 return false;
276 }
277 }
278 true
279 }
280
281 fn to_arrow_schema(&self) -> Result<Schema> {
282 let arrow_fields: Result<Vec<_>> = self
283 .fields
284 .iter()
285 .map(|field| field.to_arrow_field())
286 .collect();
287
288 if let Some(metadatas) = &self.metadata {
289 let mut metadata: HashMap<String, String> = HashMap::new();
290
291 metadatas.iter().for_each(|pair| {
292 let key = pair.get("key").unwrap();
293 let value = pair.get("value").unwrap();
294 metadata.insert(key.clone(), value.clone());
295 });
296
297 Ok(Schema::new_with_metadata(arrow_fields?, metadata))
298 } else {
299 Ok(Schema::new(arrow_fields?))
300 }
301 }
302}
303
304impl ArrowJsonField {
305 fn equals_field(&self, field: &Field) -> bool {
307 match self.to_arrow_field() {
309 Ok(self_field) => {
310 assert_eq!(&self_field, field, "Arrow fields not the same");
311 true
312 }
313 Err(e) => {
314 eprintln!("Encountered error while converting JSON field to Arrow field: {e:?}");
315 false
316 }
317 }
318 }
319
320 fn to_arrow_field(&self) -> Result<Field> {
323 let field =
325 serde_json::to_value(self).map_err(|error| ArrowError::JsonError(error.to_string()))?;
326 field_from_json(&field)
327 }
328}
329
330pub fn record_batch_from_json(
332 schema: &Schema,
333 json_batch: ArrowJsonBatch,
334 json_dictionaries: Option<&HashMap<i64, ArrowJsonDictionaryBatch>>,
335) -> Result<RecordBatch> {
336 let mut columns = vec![];
337
338 for (field, json_col) in schema.fields().iter().zip(json_batch.columns) {
339 let col = array_from_json(field, json_col, json_dictionaries)?;
340 columns.push(col);
341 }
342
343 RecordBatch::try_new(Arc::new(schema.clone()), columns)
344}
345
346pub fn array_from_json(
348 field: &Field,
349 json_col: ArrowJsonColumn,
350 dictionaries: Option<&HashMap<i64, ArrowJsonDictionaryBatch>>,
351) -> Result<ArrayRef> {
352 match field.data_type() {
353 DataType::Null => Ok(Arc::new(NullArray::new(json_col.count))),
354 DataType::Boolean => {
355 let mut b = BooleanBuilder::with_capacity(json_col.count);
356 for (is_valid, value) in json_col
357 .validity
358 .as_ref()
359 .unwrap()
360 .iter()
361 .zip(json_col.data.unwrap())
362 {
363 match is_valid {
364 1 => b.append_value(value.as_bool().unwrap()),
365 _ => b.append_null(),
366 }
367 }
368 Ok(Arc::new(b.finish()))
369 }
370 DataType::Int8 => {
371 let mut b = Int8Builder::with_capacity(json_col.count);
372 for (is_valid, value) in json_col
373 .validity
374 .as_ref()
375 .unwrap()
376 .iter()
377 .zip(json_col.data.unwrap())
378 {
379 match is_valid {
380 1 => b.append_value(value.as_i64().ok_or_else(|| {
381 ArrowError::JsonError(format!("Unable to get {value:?} as int64"))
382 })? as i8),
383 _ => b.append_null(),
384 }
385 }
386 Ok(Arc::new(b.finish()))
387 }
388 DataType::Int16 => {
389 let mut b = Int16Builder::with_capacity(json_col.count);
390 for (is_valid, value) in json_col
391 .validity
392 .as_ref()
393 .unwrap()
394 .iter()
395 .zip(json_col.data.unwrap())
396 {
397 match is_valid {
398 1 => b.append_value(value.as_i64().unwrap() as i16),
399 _ => b.append_null(),
400 }
401 }
402 Ok(Arc::new(b.finish()))
403 }
404 DataType::Int32 | DataType::Date32 | DataType::Time32(_) => {
405 let mut b = Int32Builder::with_capacity(json_col.count);
406 for (is_valid, value) in json_col
407 .validity
408 .as_ref()
409 .unwrap()
410 .iter()
411 .zip(json_col.data.unwrap())
412 {
413 match is_valid {
414 1 => b.append_value(value.as_i64().unwrap() as i32),
415 _ => b.append_null(),
416 }
417 }
418 let array = Arc::new(b.finish()) as ArrayRef;
419 arrow::compute::cast(&array, field.data_type())
420 }
421 DataType::Interval(IntervalUnit::YearMonth) => {
422 let mut b = IntervalYearMonthBuilder::with_capacity(json_col.count);
423 for (is_valid, value) in json_col
424 .validity
425 .as_ref()
426 .unwrap()
427 .iter()
428 .zip(json_col.data.unwrap())
429 {
430 match is_valid {
431 1 => b.append_value(value.as_i64().unwrap() as i32),
432 _ => b.append_null(),
433 }
434 }
435 Ok(Arc::new(b.finish()))
436 }
437 DataType::Int64
438 | DataType::Date64
439 | DataType::Time64(_)
440 | DataType::Timestamp(_, _)
441 | DataType::Duration(_) => {
442 let mut b = Int64Builder::with_capacity(json_col.count);
443 for (is_valid, value) in json_col
444 .validity
445 .as_ref()
446 .unwrap()
447 .iter()
448 .zip(json_col.data.unwrap())
449 {
450 match is_valid {
451 1 => b.append_value(match value {
452 Value::Number(n) => n.as_i64().unwrap(),
453 Value::String(s) => s.parse().expect("Unable to parse string as i64"),
454 _ => panic!("Unable to parse {value:?} as number"),
455 }),
456 _ => b.append_null(),
457 }
458 }
459 let array = Arc::new(b.finish()) as ArrayRef;
460 arrow::compute::cast(&array, field.data_type())
461 }
462 DataType::Interval(IntervalUnit::DayTime) => {
463 let mut b = IntervalDayTimeBuilder::with_capacity(json_col.count);
464 for (is_valid, value) in json_col
465 .validity
466 .as_ref()
467 .unwrap()
468 .iter()
469 .zip(json_col.data.unwrap())
470 {
471 match is_valid {
472 1 => b.append_value(match value {
473 Value::Object(ref map)
474 if map.contains_key("days") && map.contains_key("milliseconds") =>
475 {
476 match field.data_type() {
477 DataType::Interval(IntervalUnit::DayTime) => {
478 let days = map.get("days").unwrap();
479 let milliseconds = map.get("milliseconds").unwrap();
480
481 match (days, milliseconds) {
482 (Value::Number(d), Value::Number(m)) => {
483 let days = d.as_i64().unwrap() as _;
484 let millis = m.as_i64().unwrap() as _;
485 IntervalDayTime::new(days, millis)
486 }
487 _ => {
488 panic!("Unable to parse {value:?} as interval daytime")
489 }
490 }
491 }
492 _ => panic!("Unable to parse {value:?} as interval daytime"),
493 }
494 }
495 _ => panic!("Unable to parse {value:?} as number"),
496 }),
497 _ => b.append_null(),
498 }
499 }
500 Ok(Arc::new(b.finish()))
501 }
502 DataType::UInt8 => {
503 let mut b = UInt8Builder::with_capacity(json_col.count);
504 for (is_valid, value) in json_col
505 .validity
506 .as_ref()
507 .unwrap()
508 .iter()
509 .zip(json_col.data.unwrap())
510 {
511 match is_valid {
512 1 => b.append_value(value.as_u64().unwrap() as u8),
513 _ => b.append_null(),
514 }
515 }
516 Ok(Arc::new(b.finish()))
517 }
518 DataType::UInt16 => {
519 let mut b = UInt16Builder::with_capacity(json_col.count);
520 for (is_valid, value) in json_col
521 .validity
522 .as_ref()
523 .unwrap()
524 .iter()
525 .zip(json_col.data.unwrap())
526 {
527 match is_valid {
528 1 => b.append_value(value.as_u64().unwrap() as u16),
529 _ => b.append_null(),
530 }
531 }
532 Ok(Arc::new(b.finish()))
533 }
534 DataType::UInt32 => {
535 let mut b = UInt32Builder::with_capacity(json_col.count);
536 for (is_valid, value) in json_col
537 .validity
538 .as_ref()
539 .unwrap()
540 .iter()
541 .zip(json_col.data.unwrap())
542 {
543 match is_valid {
544 1 => b.append_value(value.as_u64().unwrap() as u32),
545 _ => b.append_null(),
546 }
547 }
548 Ok(Arc::new(b.finish()))
549 }
550 DataType::UInt64 => {
551 let mut b = UInt64Builder::with_capacity(json_col.count);
552 for (is_valid, value) in json_col
553 .validity
554 .as_ref()
555 .unwrap()
556 .iter()
557 .zip(json_col.data.unwrap())
558 {
559 match is_valid {
560 1 => {
561 if value.is_string() {
562 b.append_value(
563 value
564 .as_str()
565 .unwrap()
566 .parse()
567 .expect("Unable to parse string as u64"),
568 )
569 } else if value.is_number() {
570 b.append_value(value.as_u64().expect("Unable to read number as u64"))
571 } else {
572 panic!("Unable to parse value {value:?} as u64")
573 }
574 }
575 _ => b.append_null(),
576 }
577 }
578 Ok(Arc::new(b.finish()))
579 }
580 DataType::Interval(IntervalUnit::MonthDayNano) => {
581 let mut b = IntervalMonthDayNanoBuilder::with_capacity(json_col.count);
582 for (is_valid, value) in json_col
583 .validity
584 .as_ref()
585 .unwrap()
586 .iter()
587 .zip(json_col.data.unwrap())
588 {
589 match is_valid {
590 1 => b.append_value(match value {
591 Value::Object(v) => {
592 let months = v.get("months").unwrap();
593 let days = v.get("days").unwrap();
594 let nanoseconds = v.get("nanoseconds").unwrap();
595 match (months, days, nanoseconds) {
596 (
597 Value::Number(months),
598 Value::Number(days),
599 Value::Number(nanoseconds),
600 ) => {
601 let months = months.as_i64().unwrap() as i32;
602 let days = days.as_i64().unwrap() as i32;
603 let nanoseconds = nanoseconds.as_i64().unwrap();
604 IntervalMonthDayNano::new(months, days, nanoseconds)
605 }
606 (_, _, _) => {
607 panic!("Unable to parse {v:?} as MonthDayNano")
608 }
609 }
610 }
611 _ => panic!("Unable to parse {value:?} as MonthDayNano"),
612 }),
613 _ => b.append_null(),
614 }
615 }
616 Ok(Arc::new(b.finish()))
617 }
618 DataType::Float32 => {
619 let mut b = Float32Builder::with_capacity(json_col.count);
620 for (is_valid, value) in json_col
621 .validity
622 .as_ref()
623 .unwrap()
624 .iter()
625 .zip(json_col.data.unwrap())
626 {
627 match is_valid {
628 1 => b.append_value(value.as_f64().unwrap() as f32),
629 _ => b.append_null(),
630 }
631 }
632 Ok(Arc::new(b.finish()))
633 }
634 DataType::Float64 => {
635 let mut b = Float64Builder::with_capacity(json_col.count);
636 for (is_valid, value) in json_col
637 .validity
638 .as_ref()
639 .unwrap()
640 .iter()
641 .zip(json_col.data.unwrap())
642 {
643 match is_valid {
644 1 => b.append_value(value.as_f64().unwrap()),
645 _ => b.append_null(),
646 }
647 }
648 Ok(Arc::new(b.finish()))
649 }
650 DataType::Binary => {
651 let mut b = BinaryBuilder::with_capacity(json_col.count, 1024);
652 for (is_valid, value) in json_col
653 .validity
654 .as_ref()
655 .unwrap()
656 .iter()
657 .zip(json_col.data.unwrap())
658 {
659 match is_valid {
660 1 => {
661 let v = decode(value.as_str().unwrap()).unwrap();
662 b.append_value(&v)
663 }
664 _ => b.append_null(),
665 }
666 }
667 Ok(Arc::new(b.finish()))
668 }
669 DataType::LargeBinary => {
670 let mut b = LargeBinaryBuilder::with_capacity(json_col.count, 1024);
671 for (is_valid, value) in json_col
672 .validity
673 .as_ref()
674 .unwrap()
675 .iter()
676 .zip(json_col.data.unwrap())
677 {
678 match is_valid {
679 1 => {
680 let v = decode(value.as_str().unwrap()).unwrap();
681 b.append_value(&v)
682 }
683 _ => b.append_null(),
684 }
685 }
686 Ok(Arc::new(b.finish()))
687 }
688 DataType::Utf8 => {
689 let mut b = StringBuilder::with_capacity(json_col.count, 1024);
690 for (is_valid, value) in json_col
691 .validity
692 .as_ref()
693 .unwrap()
694 .iter()
695 .zip(json_col.data.unwrap())
696 {
697 match is_valid {
698 1 => b.append_value(value.as_str().unwrap()),
699 _ => b.append_null(),
700 }
701 }
702 Ok(Arc::new(b.finish()))
703 }
704 DataType::LargeUtf8 => {
705 let mut b = LargeStringBuilder::with_capacity(json_col.count, 1024);
706 for (is_valid, value) in json_col
707 .validity
708 .as_ref()
709 .unwrap()
710 .iter()
711 .zip(json_col.data.unwrap())
712 {
713 match is_valid {
714 1 => b.append_value(value.as_str().unwrap()),
715 _ => b.append_null(),
716 }
717 }
718 Ok(Arc::new(b.finish()))
719 }
720 DataType::FixedSizeBinary(len) => {
721 let mut b = FixedSizeBinaryBuilder::with_capacity(json_col.count, *len);
722 for (is_valid, value) in json_col
723 .validity
724 .as_ref()
725 .unwrap()
726 .iter()
727 .zip(json_col.data.unwrap())
728 {
729 match is_valid {
730 1 => {
731 let v = hex::decode(value.as_str().unwrap()).unwrap();
732 b.append_value(&v)?
733 }
734 _ => b.append_null(),
735 }
736 }
737 Ok(Arc::new(b.finish()))
738 }
739 DataType::List(child_field) => {
740 let null_buf = create_null_buf(&json_col);
741 let children = json_col.children.clone().unwrap();
742 let child_array = array_from_json(child_field, children[0].clone(), dictionaries)?;
743 let offsets: Vec<i32> = json_col
744 .offset
745 .unwrap()
746 .iter()
747 .map(|v| v.as_i64().unwrap() as i32)
748 .collect();
749 let list_data = ArrayData::builder(field.data_type().clone())
750 .len(json_col.count)
751 .offset(0)
752 .add_buffer(Buffer::from(offsets.to_byte_slice()))
753 .add_child_data(child_array.into_data())
754 .null_bit_buffer(Some(null_buf))
755 .build()
756 .unwrap();
757 Ok(Arc::new(ListArray::from(list_data)))
758 }
759 DataType::LargeList(child_field) => {
760 let null_buf = create_null_buf(&json_col);
761 let children = json_col.children.clone().unwrap();
762 let child_array = array_from_json(child_field, children[0].clone(), dictionaries)?;
763 let offsets: Vec<i64> = json_col
764 .offset
765 .unwrap()
766 .iter()
767 .map(|v| match v {
768 Value::Number(n) => n.as_i64().unwrap(),
769 Value::String(s) => s.parse::<i64>().unwrap(),
770 _ => panic!("64-bit offset must be either string or number"),
771 })
772 .collect();
773 let list_data = ArrayData::builder(field.data_type().clone())
774 .len(json_col.count)
775 .offset(0)
776 .add_buffer(Buffer::from(offsets.to_byte_slice()))
777 .add_child_data(child_array.into_data())
778 .null_bit_buffer(Some(null_buf))
779 .build()
780 .unwrap();
781 Ok(Arc::new(LargeListArray::from(list_data)))
782 }
783 DataType::ListView(child_field) => {
784 let null_buf = create_null_buf(&json_col);
785 let children = json_col.children.clone().unwrap();
786 let child_array = array_from_json(child_field, children[0].clone(), dictionaries)?;
787 let offsets: Vec<i32> = json_col
788 .offset
789 .unwrap()
790 .iter()
791 .map(|v| v.as_i64().unwrap() as i32)
792 .collect();
793 let sizes: Vec<i32> = json_col
794 .size
795 .unwrap()
796 .iter()
797 .map(|v| v.as_i64().unwrap() as i32)
798 .collect();
799 let list_data = ArrayData::builder(field.data_type().clone())
800 .len(json_col.count)
801 .add_buffer(Buffer::from(offsets.to_byte_slice()))
802 .add_buffer(Buffer::from(sizes.to_byte_slice()))
803 .add_child_data(child_array.into_data())
804 .null_bit_buffer(Some(null_buf))
805 .build()
806 .unwrap();
807 Ok(Arc::new(ListViewArray::from(list_data)))
808 }
809 DataType::LargeListView(child_field) => {
810 let null_buf = create_null_buf(&json_col);
811 let children = json_col.children.clone().unwrap();
812 let child_array = array_from_json(child_field, children[0].clone(), dictionaries)?;
813 let offsets: Vec<i64> = json_col
814 .offset
815 .unwrap()
816 .iter()
817 .map(|v| match v {
818 Value::Number(n) => n.as_i64().unwrap(),
819 Value::String(s) => s.parse::<i64>().unwrap(),
820 _ => panic!("64-bit offset must be either string or number"),
821 })
822 .collect();
823 let sizes: Vec<i64> = json_col
824 .size
825 .unwrap()
826 .iter()
827 .map(|v| match v {
828 Value::Number(n) => n.as_i64().unwrap(),
829 Value::String(s) => s.parse::<i64>().unwrap(),
830 _ => panic!("64-bit size must be either string or number"),
831 })
832 .collect();
833 let list_data = ArrayData::builder(field.data_type().clone())
834 .len(json_col.count)
835 .add_buffer(Buffer::from(offsets.to_byte_slice()))
836 .add_buffer(Buffer::from(sizes.to_byte_slice()))
837 .add_child_data(child_array.into_data())
838 .null_bit_buffer(Some(null_buf))
839 .build()
840 .unwrap();
841 Ok(Arc::new(LargeListViewArray::from(list_data)))
842 }
843 DataType::FixedSizeList(child_field, _) => {
844 let children = json_col.children.clone().unwrap();
845 let child_array = array_from_json(child_field, children[0].clone(), dictionaries)?;
846 let null_buf = create_null_buf(&json_col);
847 let list_data = ArrayData::builder(field.data_type().clone())
848 .len(json_col.count)
849 .add_child_data(child_array.into_data())
850 .null_bit_buffer(Some(null_buf))
851 .build()
852 .unwrap();
853 Ok(Arc::new(FixedSizeListArray::from(list_data)))
854 }
855 DataType::Struct(fields) => {
856 let null_buf = create_null_buf(&json_col);
858 let mut array_data = ArrayData::builder(field.data_type().clone())
859 .len(json_col.count)
860 .null_bit_buffer(Some(null_buf));
861
862 for (field, col) in fields.iter().zip(json_col.children.unwrap()) {
863 let array = array_from_json(field, col, dictionaries)?;
864 array_data = array_data.add_child_data(array.into_data());
865 }
866
867 let array = StructArray::from(array_data.build().unwrap());
868 Ok(Arc::new(array))
869 }
870 DataType::Dictionary(key_type, value_type) => {
871 #[expect(deprecated)]
872 let dict_id = field.dict_id().ok_or_else(|| {
873 ArrowError::JsonError(format!("Unable to find dict_id for field {field}"))
874 })?;
875 let dictionary = dictionaries
877 .ok_or_else(|| {
878 ArrowError::JsonError(format!(
879 "Unable to find any dictionaries for field {field}"
880 ))
881 })?
882 .get(&dict_id);
883 match dictionary {
884 Some(dictionary) => dictionary_array_from_json(
885 field,
886 json_col,
887 key_type,
888 value_type,
889 dictionary,
890 dictionaries,
891 ),
892 None => Err(ArrowError::JsonError(format!(
893 "Unable to find dictionary for field {field}"
894 ))),
895 }
896 }
897 DataType::Decimal32(precision, scale) => {
898 let mut b = Decimal32Builder::with_capacity(json_col.count);
899 for (is_valid, value) in json_col
900 .validity
901 .as_ref()
902 .unwrap()
903 .iter()
904 .zip(json_col.data.unwrap())
905 {
906 match is_valid {
907 1 => b.append_value(value.as_str().unwrap().parse::<i32>().unwrap()),
908 _ => b.append_null(),
909 }
910 }
911 Ok(Arc::new(
912 b.finish().with_precision_and_scale(*precision, *scale)?,
913 ))
914 }
915 DataType::Decimal64(precision, scale) => {
916 let mut b = Decimal64Builder::with_capacity(json_col.count);
917 for (is_valid, value) in json_col
918 .validity
919 .as_ref()
920 .unwrap()
921 .iter()
922 .zip(json_col.data.unwrap())
923 {
924 match is_valid {
925 1 => b.append_value(value.as_str().unwrap().parse::<i64>().unwrap()),
926 _ => b.append_null(),
927 }
928 }
929 Ok(Arc::new(
930 b.finish().with_precision_and_scale(*precision, *scale)?,
931 ))
932 }
933 DataType::Decimal128(precision, scale) => {
934 let mut b = Decimal128Builder::with_capacity(json_col.count);
935 for (is_valid, value) in json_col
936 .validity
937 .as_ref()
938 .unwrap()
939 .iter()
940 .zip(json_col.data.unwrap())
941 {
942 match is_valid {
943 1 => b.append_value(value.as_str().unwrap().parse::<i128>().unwrap()),
944 _ => b.append_null(),
945 }
946 }
947 Ok(Arc::new(
948 b.finish().with_precision_and_scale(*precision, *scale)?,
949 ))
950 }
951 DataType::Decimal256(precision, scale) => {
952 let mut b = Decimal256Builder::with_capacity(json_col.count);
953 for (is_valid, value) in json_col
954 .validity
955 .as_ref()
956 .unwrap()
957 .iter()
958 .zip(json_col.data.unwrap())
959 {
960 match is_valid {
961 1 => {
962 let str = value.as_str().unwrap();
963 let integer = BigInt::parse_bytes(str.as_bytes(), 10).unwrap();
964 let integer_bytes = integer.to_signed_bytes_le();
965 let mut bytes = if integer.is_negative() {
969 [255_u8; 32]
970 } else {
971 [0_u8; 32]
972 };
973 bytes[0..integer_bytes.len()].copy_from_slice(integer_bytes.as_slice());
974 b.append_value(i256::from_le_bytes(bytes));
975 }
976 _ => b.append_null(),
977 }
978 }
979 Ok(Arc::new(
980 b.finish().with_precision_and_scale(*precision, *scale)?,
981 ))
982 }
983 DataType::Map(child_field, _) => {
984 let null_buf = create_null_buf(&json_col);
985 let children = json_col.children.clone().unwrap();
986 let child_array = array_from_json(child_field, children[0].clone(), dictionaries)?;
987 let offsets: Vec<i32> = json_col
988 .offset
989 .unwrap()
990 .iter()
991 .map(|v| v.as_i64().unwrap() as i32)
992 .collect();
993 let array_data = ArrayData::builder(field.data_type().clone())
994 .len(json_col.count)
995 .add_buffer(Buffer::from(offsets.to_byte_slice()))
996 .add_child_data(child_array.into_data())
997 .null_bit_buffer(Some(null_buf))
998 .build()
999 .unwrap();
1000
1001 let array = MapArray::from(array_data);
1002 Ok(Arc::new(array))
1003 }
1004 DataType::Union(fields, _) => {
1005 let Some(type_ids) = json_col.type_id else {
1006 return Err(ArrowError::JsonError(
1007 "Cannot find expected type_id in json column".to_string(),
1008 ));
1009 };
1010
1011 let offset: Option<ScalarBuffer<i32>> = json_col
1012 .offset
1013 .map(|offsets| offsets.iter().map(|v| v.as_i64().unwrap() as i32).collect());
1014
1015 let mut children = Vec::with_capacity(fields.len());
1016 for ((_, field), col) in fields.iter().zip(json_col.children.unwrap()) {
1017 let array = array_from_json(field, col, dictionaries)?;
1018 children.push(array);
1019 }
1020
1021 let array =
1022 UnionArray::try_new(fields.clone(), type_ids.into(), offset, children).unwrap();
1023 Ok(Arc::new(array))
1024 }
1025 DataType::Utf8View => {
1026 let views = json_col.views.ok_or_else(|| {
1027 ArrowError::JsonError("Utf8View requires VIEWS field".to_string())
1028 })?;
1029 let variadic_buffers = json_col.variadic_data_buffers.unwrap_or_default();
1030 let validity = json_col.validity.as_ref();
1031
1032 let mut builder = StringViewBuilder::new();
1033 for (i, view) in views.iter().enumerate() {
1034 let is_valid = validity.map_or(1, |v| v[i]);
1035 if is_valid == 0 {
1036 builder.append_null();
1037 } else {
1038 let view_obj = view.as_object().unwrap();
1039 let size = view_obj["SIZE"].as_u64().unwrap() as usize;
1040 if let Some(inlined) = view_obj.get("INLINED") {
1042 builder.append_value(inlined.as_str().unwrap());
1043 } else {
1044 let buffer_index = view_obj["BUFFER_INDEX"].as_u64().unwrap() as usize;
1046 let offset = view_obj["OFFSET"].as_u64().unwrap() as usize;
1047 let buffer_data = hex::decode(&variadic_buffers[buffer_index]).unwrap();
1048 let s = std::str::from_utf8(&buffer_data[offset..offset + size]).unwrap();
1049 builder.append_value(s);
1050 }
1051 }
1052 }
1053 Ok(Arc::new(builder.finish()))
1054 }
1055 DataType::BinaryView => {
1056 let views = json_col.views.ok_or_else(|| {
1057 ArrowError::JsonError("BinaryView requires VIEWS field".to_string())
1058 })?;
1059 let variadic_buffers = json_col.variadic_data_buffers.unwrap_or_default();
1060 let validity = json_col.validity.as_ref();
1061
1062 let mut builder = BinaryViewBuilder::new();
1063 for (i, view) in views.iter().enumerate() {
1064 let is_valid = validity.map_or(1, |v| v[i]);
1065 if is_valid == 0 {
1066 builder.append_null();
1067 } else {
1068 let view_obj = view.as_object().unwrap();
1069 let size = view_obj["SIZE"].as_u64().unwrap() as usize;
1070 if let Some(inlined) = view_obj.get("INLINED") {
1072 let data = hex::decode(inlined.as_str().unwrap()).unwrap();
1073 builder.append_value(&data);
1074 } else {
1075 let buffer_index = view_obj["BUFFER_INDEX"].as_u64().unwrap() as usize;
1077 let offset = view_obj["OFFSET"].as_u64().unwrap() as usize;
1078 let buffer_data = hex::decode(&variadic_buffers[buffer_index]).unwrap();
1079 builder.append_value(&buffer_data[offset..offset + size]);
1080 }
1081 }
1082 }
1083 Ok(Arc::new(builder.finish()))
1084 }
1085 DataType::RunEndEncoded(run_ends_field, values_field) => {
1086 let children = json_col.children.clone().unwrap();
1087 if children.len() != 2 {
1088 return Err(ArrowError::JsonError(
1089 "RunEndEncoded requires exactly 2 children".to_string(),
1090 ));
1091 }
1092 let run_ends_array =
1093 array_from_json(run_ends_field, children[0].clone(), dictionaries)?;
1094 let values_array = array_from_json(values_field, children[1].clone(), dictionaries)?;
1095
1096 let run_array_data = ArrayData::builder(field.data_type().clone())
1097 .len(json_col.count)
1098 .add_child_data(run_ends_array.into_data())
1099 .add_child_data(values_array.into_data())
1100 .build()
1101 .unwrap();
1102
1103 Ok(make_array(run_array_data))
1104 }
1105 t => Err(ArrowError::JsonError(format!(
1106 "data type {t} not supported"
1107 ))),
1108 }
1109}
1110
1111pub fn dictionary_array_from_json(
1113 field: &Field,
1114 json_col: ArrowJsonColumn,
1115 dict_key: &DataType,
1116 dict_value: &DataType,
1117 dictionary: &ArrowJsonDictionaryBatch,
1118 dictionaries: Option<&HashMap<i64, ArrowJsonDictionaryBatch>>,
1119) -> Result<ArrayRef> {
1120 match dict_key {
1121 DataType::Int8
1122 | DataType::Int16
1123 | DataType::Int32
1124 | DataType::Int64
1125 | DataType::UInt8
1126 | DataType::UInt16
1127 | DataType::UInt32
1128 | DataType::UInt64 => {
1129 let null_buf = create_null_buf(&json_col);
1130
1131 #[expect(deprecated)]
1133 let key_field = Field::new_dict(
1134 "key",
1135 dict_key.clone(),
1136 field.is_nullable(),
1137 #[expect(deprecated)]
1138 field
1139 .dict_id()
1140 .expect("Dictionary fields must have a dict_id value"),
1141 field
1142 .dict_is_ordered()
1143 .expect("Dictionary fields must have a dict_is_ordered value"),
1144 );
1145 let keys = array_from_json(&key_field, json_col, None)?;
1146 let value_field = Field::new("value", dict_value.clone(), true);
1148 let values = array_from_json(
1149 &value_field,
1150 dictionary.data.columns[0].clone(),
1151 dictionaries,
1152 )?;
1153
1154 let dict_data = ArrayData::builder(field.data_type().clone())
1156 .len(keys.len())
1157 .add_buffer(keys.to_data().buffers()[0].clone())
1158 .null_bit_buffer(Some(null_buf))
1159 .add_child_data(values.into_data())
1160 .build()
1161 .unwrap();
1162
1163 let array = match dict_key {
1164 DataType::Int8 => Arc::new(Int8DictionaryArray::from(dict_data)) as ArrayRef,
1165 DataType::Int16 => Arc::new(Int16DictionaryArray::from(dict_data)),
1166 DataType::Int32 => Arc::new(Int32DictionaryArray::from(dict_data)),
1167 DataType::Int64 => Arc::new(Int64DictionaryArray::from(dict_data)),
1168 DataType::UInt8 => Arc::new(UInt8DictionaryArray::from(dict_data)),
1169 DataType::UInt16 => Arc::new(UInt16DictionaryArray::from(dict_data)),
1170 DataType::UInt32 => Arc::new(UInt32DictionaryArray::from(dict_data)),
1171 DataType::UInt64 => Arc::new(UInt64DictionaryArray::from(dict_data)),
1172 _ => unreachable!(),
1173 };
1174 Ok(array)
1175 }
1176 _ => Err(ArrowError::JsonError(format!(
1177 "Dictionary key type {dict_key:?} not supported"
1178 ))),
1179 }
1180}
1181
1182fn create_null_buf(json_col: &ArrowJsonColumn) -> Buffer {
1184 let num_bytes = bit_util::ceil(json_col.count, 8);
1185 let mut null_buf = MutableBuffer::new(num_bytes).with_bitset(num_bytes, false);
1186 json_col
1187 .validity
1188 .clone()
1189 .unwrap()
1190 .iter()
1191 .enumerate()
1192 .for_each(|(i, v)| {
1193 let null_slice = null_buf.as_slice_mut();
1194 if *v != 0 {
1195 bit_util::set_bit(null_slice, i);
1196 }
1197 });
1198 null_buf.into()
1199}
1200
1201impl ArrowJsonBatch {
1202 pub fn from_batch(batch: &RecordBatch) -> ArrowJsonBatch {
1214 let mut json_batch = ArrowJsonBatch {
1215 count: batch.num_rows(),
1216 columns: Vec::with_capacity(batch.num_columns()),
1217 };
1218
1219 for (col, field) in batch.columns().iter().zip(batch.schema().fields.iter()) {
1220 let json_col = match field.data_type() {
1221 DataType::Int8 => {
1222 let col = col.as_any().downcast_ref::<Int8Array>().unwrap();
1223
1224 let mut validity: Vec<u8> = Vec::with_capacity(col.len());
1225 let mut data: Vec<Value> = Vec::with_capacity(col.len());
1226
1227 for i in 0..col.len() {
1228 if col.is_null(i) {
1229 validity.push(1);
1230 data.push(0i8.into());
1231 } else {
1232 validity.push(0);
1233 data.push(col.value(i).into());
1234 }
1235 }
1236
1237 ArrowJsonColumn {
1238 name: field.name().clone(),
1239 count: col.len(),
1240 validity: Some(validity),
1241 data: Some(data),
1242 offset: None,
1243 type_id: None,
1244 size: None,
1245 views: None,
1246 variadic_data_buffers: None,
1247 children: None,
1248 }
1249 }
1250 _ => ArrowJsonColumn {
1251 name: field.name().clone(),
1252 count: col.len(),
1253 validity: None,
1254 data: None,
1255 offset: None,
1256 type_id: None,
1257 size: None,
1258 views: None,
1259 variadic_data_buffers: None,
1260 children: None,
1261 },
1262 };
1263
1264 json_batch.columns.push(json_col);
1265 }
1266
1267 json_batch
1268 }
1269}
1270
1271#[cfg(test)]
1272mod tests {
1273 use super::*;
1274
1275 #[test]
1276 fn test_decimal256_from_json() {
1277 let field = Field::new("c", DataType::Decimal256(76, 0), false);
1278 let col: ArrowJsonColumn = serde_json::from_str(
1279 r#"{
1280 "name": "c",
1281 "count": 5,
1282 "VALIDITY": [1, 1, 1, 1, 1],
1283 "DATA": [
1284 "0",
1285 "1",
1286 "-1",
1287 "123456789012345678901234567890",
1288 "-123456789012345678901234567890"
1289 ]
1290 }"#,
1291 )
1292 .unwrap();
1293 let arr = array_from_json(&field, col, None).unwrap();
1294 let arr = arr.as_any().downcast_ref::<Decimal256Array>().unwrap();
1295 assert_eq!(arr.value(0), i256::ZERO);
1296 assert_eq!(arr.value(1), i256::from_i128(1));
1297 assert_eq!(arr.value(2), i256::from_i128(-1));
1298 let big = i256::from_string("123456789012345678901234567890").unwrap();
1299 assert_eq!(arr.value(3), big);
1300 assert_eq!(arr.value(4), i256::ZERO - big);
1301 }
1302
1303 #[test]
1304 fn test_schema_equality() {
1305 let json = r#"
1306 {
1307 "fields": [
1308 {
1309 "name": "c1",
1310 "type": {"name": "int", "isSigned": true, "bitWidth": 32},
1311 "nullable": true,
1312 "children": []
1313 },
1314 {
1315 "name": "c2",
1316 "type": {"name": "floatingpoint", "precision": "DOUBLE"},
1317 "nullable": true,
1318 "children": []
1319 },
1320 {
1321 "name": "c3",
1322 "type": {"name": "utf8"},
1323 "nullable": true,
1324 "children": []
1325 },
1326 {
1327 "name": "c4",
1328 "type": {
1329 "name": "list"
1330 },
1331 "nullable": true,
1332 "children": [
1333 {
1334 "name": "custom_item",
1335 "type": {
1336 "name": "int",
1337 "isSigned": true,
1338 "bitWidth": 32
1339 },
1340 "nullable": false,
1341 "children": []
1342 }
1343 ]
1344 }
1345 ]
1346 }"#;
1347 let json_schema: ArrowJsonSchema = serde_json::from_str(json).unwrap();
1348 let schema = Schema::new(vec![
1349 Field::new("c1", DataType::Int32, true),
1350 Field::new("c2", DataType::Float64, true),
1351 Field::new("c3", DataType::Utf8, true),
1352 Field::new(
1353 "c4",
1354 DataType::List(Arc::new(Field::new("custom_item", DataType::Int32, false))),
1355 true,
1356 ),
1357 ]);
1358 assert!(json_schema.equals_schema(&schema));
1359 }
1360
1361 #[test]
1362 fn test_arrow_data_equality() {
1363 let secs_tz = Some("Europe/Budapest".into());
1364 let millis_tz = Some("America/New_York".into());
1365 let micros_tz = Some("UTC".into());
1366 let nanos_tz = Some("Africa/Johannesburg".into());
1367
1368 let schema = Schema::new(vec![
1369 Field::new("bools-with-metadata-map", DataType::Boolean, true)
1370 .with_metadata([("k", "v")]),
1371 Field::new("bools-with-metadata-vec", DataType::Boolean, true)
1372 .with_metadata([("k2", "v2")]),
1373 Field::new("bools", DataType::Boolean, true),
1374 Field::new("int8s", DataType::Int8, true),
1375 Field::new("int16s", DataType::Int16, true),
1376 Field::new("int32s", DataType::Int32, true),
1377 Field::new("int64s", DataType::Int64, true),
1378 Field::new("uint8s", DataType::UInt8, true),
1379 Field::new("uint16s", DataType::UInt16, true),
1380 Field::new("uint32s", DataType::UInt32, true),
1381 Field::new("uint64s", DataType::UInt64, true),
1382 Field::new("float32s", DataType::Float32, true),
1383 Field::new("float64s", DataType::Float64, true),
1384 Field::new("date_days", DataType::Date32, true),
1385 Field::new("date_millis", DataType::Date64, true),
1386 Field::new("time_secs", DataType::Time32(TimeUnit::Second), true),
1387 Field::new("time_millis", DataType::Time32(TimeUnit::Millisecond), true),
1388 Field::new("time_micros", DataType::Time64(TimeUnit::Microsecond), true),
1389 Field::new("time_nanos", DataType::Time64(TimeUnit::Nanosecond), true),
1390 Field::new("ts_secs", DataType::Timestamp(TimeUnit::Second, None), true),
1391 Field::new(
1392 "ts_millis",
1393 DataType::Timestamp(TimeUnit::Millisecond, None),
1394 true,
1395 ),
1396 Field::new(
1397 "ts_micros",
1398 DataType::Timestamp(TimeUnit::Microsecond, None),
1399 true,
1400 ),
1401 Field::new(
1402 "ts_nanos",
1403 DataType::Timestamp(TimeUnit::Nanosecond, None),
1404 true,
1405 ),
1406 Field::new(
1407 "ts_secs_tz",
1408 DataType::Timestamp(TimeUnit::Second, secs_tz.clone()),
1409 true,
1410 ),
1411 Field::new(
1412 "ts_millis_tz",
1413 DataType::Timestamp(TimeUnit::Millisecond, millis_tz.clone()),
1414 true,
1415 ),
1416 Field::new(
1417 "ts_micros_tz",
1418 DataType::Timestamp(TimeUnit::Microsecond, micros_tz.clone()),
1419 true,
1420 ),
1421 Field::new(
1422 "ts_nanos_tz",
1423 DataType::Timestamp(TimeUnit::Nanosecond, nanos_tz.clone()),
1424 true,
1425 ),
1426 Field::new("utf8s", DataType::Utf8, true),
1427 Field::new(
1428 "lists",
1429 DataType::List(Arc::new(Field::new_list_field(DataType::Int32, true))),
1430 true,
1431 ),
1432 Field::new(
1433 "structs",
1434 DataType::Struct(Fields::from(vec![
1435 Field::new("int32s", DataType::Int32, true),
1436 Field::new("utf8s", DataType::Utf8, true),
1437 ])),
1438 true,
1439 ),
1440 Field::new("utf8views", DataType::Utf8View, true),
1441 Field::new("binaryviews", DataType::BinaryView, true),
1442 Field::new(
1443 "listviews",
1444 DataType::ListView(Arc::new(Field::new_list_field(DataType::Int32, true))),
1445 true,
1446 ),
1447 Field::new(
1448 "largelistviews",
1449 DataType::LargeListView(Arc::new(Field::new_list_field(DataType::Int32, true))),
1450 true,
1451 ),
1452 Field::new(
1453 "runendencoded",
1454 DataType::RunEndEncoded(
1455 Arc::new(Field::new("run_ends", DataType::Int16, false)),
1456 Arc::new(Field::new("values", DataType::Int32, true)),
1457 ),
1458 true,
1459 ),
1460 ]);
1461
1462 let bools_with_metadata_map = BooleanArray::from(vec![Some(true), None, Some(false)]);
1463 let bools_with_metadata_vec = BooleanArray::from(vec![Some(true), None, Some(false)]);
1464 let bools = BooleanArray::from(vec![Some(true), None, Some(false)]);
1465 let int8s = Int8Array::from(vec![Some(1), None, Some(3)]);
1466 let int16s = Int16Array::from(vec![Some(1), None, Some(3)]);
1467 let int32s = Int32Array::from(vec![Some(1), None, Some(3)]);
1468 let int64s = Int64Array::from(vec![Some(1), None, Some(3)]);
1469 let uint8s = UInt8Array::from(vec![Some(1), None, Some(3)]);
1470 let uint16s = UInt16Array::from(vec![Some(1), None, Some(3)]);
1471 let uint32s = UInt32Array::from(vec![Some(1), None, Some(3)]);
1472 let uint64s = UInt64Array::from(vec![Some(1), None, Some(3)]);
1473 let float32s = Float32Array::from(vec![Some(1.0), None, Some(3.0)]);
1474 let float64s = Float64Array::from(vec![Some(1.0), None, Some(3.0)]);
1475 let date_days = Date32Array::from(vec![Some(1196848), None, None]);
1476 let date_millis = Date64Array::from(vec![
1477 Some(167903550396207),
1478 Some(29923997007884),
1479 Some(30612271819236),
1480 ]);
1481 let time_secs = Time32SecondArray::from(vec![Some(27974), Some(78592), Some(43207)]);
1482 let time_millis =
1483 Time32MillisecondArray::from(vec![Some(6613125), Some(74667230), Some(52260079)]);
1484 let time_micros = Time64MicrosecondArray::from(vec![Some(62522958593), None, None]);
1485 let time_nanos =
1486 Time64NanosecondArray::from(vec![Some(73380123595985), None, Some(16584393546415)]);
1487 let ts_secs = TimestampSecondArray::from(vec![None, Some(193438817552), None]);
1488 let ts_millis =
1489 TimestampMillisecondArray::from(vec![None, Some(38606916383008), Some(58113709376587)]);
1490 let ts_micros = TimestampMicrosecondArray::from(vec![None, None, None]);
1491 let ts_nanos = TimestampNanosecondArray::from(vec![None, None, Some(-6473623571954960143)]);
1492 let ts_secs_tz = TimestampSecondArray::from(vec![None, Some(193438817552), None])
1493 .with_timezone_opt(secs_tz);
1494 let ts_millis_tz =
1495 TimestampMillisecondArray::from(vec![None, Some(38606916383008), Some(58113709376587)])
1496 .with_timezone_opt(millis_tz);
1497 let ts_micros_tz =
1498 TimestampMicrosecondArray::from(vec![None, None, None]).with_timezone_opt(micros_tz);
1499 let ts_nanos_tz =
1500 TimestampNanosecondArray::from(vec![None, None, Some(-6473623571954960143)])
1501 .with_timezone_opt(nanos_tz);
1502 let utf8s = StringArray::from(vec![Some("aa"), None, Some("bbb")]);
1503
1504 let value_data = Int32Array::from(vec![None, Some(2), None, None]);
1505 let value_offsets = Buffer::from_slice_ref([0, 3, 4, 4]);
1506 let list_data_type = DataType::List(Arc::new(Field::new_list_field(DataType::Int32, true)));
1507 let list_data = ArrayData::builder(list_data_type)
1508 .len(3)
1509 .add_buffer(value_offsets)
1510 .add_child_data(value_data.into_data())
1511 .null_bit_buffer(Some(Buffer::from([0b00000011])))
1512 .build()
1513 .unwrap();
1514 let lists = ListArray::from(list_data);
1515
1516 let structs_int32s = Int32Array::from(vec![None, Some(-2), None]);
1517 let structs_utf8s = StringArray::from(vec![None, None, Some("aaaaaa")]);
1518 let struct_data_type = DataType::Struct(Fields::from(vec![
1519 Field::new("int32s", DataType::Int32, true),
1520 Field::new("utf8s", DataType::Utf8, true),
1521 ]));
1522 let struct_data = ArrayData::builder(struct_data_type)
1523 .len(3)
1524 .add_child_data(structs_int32s.into_data())
1525 .add_child_data(structs_utf8s.into_data())
1526 .null_bit_buffer(Some(Buffer::from([0b00000011])))
1527 .build()
1528 .unwrap();
1529 let structs = StructArray::from(struct_data);
1530
1531 let utf8views =
1532 StringViewArray::from(vec![Some("hello"), None, Some("this is not inlined")]);
1533 let binaryviews = BinaryViewArray::from_iter(vec![
1534 Some(b"\xf3\x4d".as_slice()),
1535 Some(b"\x00\x01\x02\x03\x04\x05\x06\x07\x08\x09\x0a\x0b\x0c\x0d\x0e\x0f".as_slice()),
1536 None,
1537 ]);
1538
1539 let listview_value_data = Int32Array::from(vec![Some(1), Some(2), Some(3), None, Some(5)]);
1540 let listview_offsets = Buffer::from_slice_ref([0i32, 2, 2]);
1541 let listview_sizes = Buffer::from_slice_ref([2i32, 0, 3]);
1542 let listview_data_type =
1543 DataType::ListView(Arc::new(Field::new_list_field(DataType::Int32, true)));
1544 let listview_data = ArrayData::builder(listview_data_type)
1545 .len(3)
1546 .add_buffer(listview_offsets)
1547 .add_buffer(listview_sizes)
1548 .add_child_data(listview_value_data.into_data())
1549 .null_bit_buffer(Some(Buffer::from([0b00000101])))
1550 .build()
1551 .unwrap();
1552 let listviews = ListViewArray::from(listview_data);
1553
1554 let largelistview_value_data = Int32Array::from(vec![Some(10), None, Some(30)]);
1555 let largelistview_offsets = Buffer::from_slice_ref([0i64, 2, 3]);
1556 let largelistview_sizes = Buffer::from_slice_ref([2i64, 1, 0]);
1557 let largelistview_data_type =
1558 DataType::LargeListView(Arc::new(Field::new_list_field(DataType::Int32, true)));
1559 let largelistview_data = ArrayData::builder(largelistview_data_type)
1560 .len(3)
1561 .add_buffer(largelistview_offsets)
1562 .add_buffer(largelistview_sizes)
1563 .add_child_data(largelistview_value_data.into_data())
1564 .null_bit_buffer(Some(Buffer::from([0b00000011])))
1565 .build()
1566 .unwrap();
1567 let largelistviews = LargeListViewArray::from(largelistview_data);
1568
1569 let ree_run_ends = Int16Array::from(vec![2, 3]);
1570 let ree_values = Int32Array::from(vec![Some(100), None]);
1571 let ree_data_type = DataType::RunEndEncoded(
1572 Arc::new(Field::new("run_ends", DataType::Int16, false)),
1573 Arc::new(Field::new("values", DataType::Int32, true)),
1574 );
1575 let ree_data = ArrayData::builder(ree_data_type)
1576 .len(3)
1577 .add_child_data(ree_run_ends.into_data())
1578 .add_child_data(ree_values.into_data())
1579 .build()
1580 .unwrap();
1581 let runendencoded = RunArray::<Int16Type>::from(ree_data);
1582
1583 let record_batch = RecordBatch::try_new(
1584 Arc::new(schema.clone()),
1585 vec![
1586 Arc::new(bools_with_metadata_map),
1587 Arc::new(bools_with_metadata_vec),
1588 Arc::new(bools),
1589 Arc::new(int8s),
1590 Arc::new(int16s),
1591 Arc::new(int32s),
1592 Arc::new(int64s),
1593 Arc::new(uint8s),
1594 Arc::new(uint16s),
1595 Arc::new(uint32s),
1596 Arc::new(uint64s),
1597 Arc::new(float32s),
1598 Arc::new(float64s),
1599 Arc::new(date_days),
1600 Arc::new(date_millis),
1601 Arc::new(time_secs),
1602 Arc::new(time_millis),
1603 Arc::new(time_micros),
1604 Arc::new(time_nanos),
1605 Arc::new(ts_secs),
1606 Arc::new(ts_millis),
1607 Arc::new(ts_micros),
1608 Arc::new(ts_nanos),
1609 Arc::new(ts_secs_tz),
1610 Arc::new(ts_millis_tz),
1611 Arc::new(ts_micros_tz),
1612 Arc::new(ts_nanos_tz),
1613 Arc::new(utf8s),
1614 Arc::new(lists),
1615 Arc::new(structs),
1616 Arc::new(utf8views),
1617 Arc::new(binaryviews),
1618 Arc::new(listviews),
1619 Arc::new(largelistviews),
1620 Arc::new(runendencoded),
1621 ],
1622 )
1623 .unwrap();
1624 let json = std::fs::read_to_string("data/integration.json").unwrap();
1625 let arrow_json: ArrowJson = serde_json::from_str(&json).unwrap();
1626 assert!(arrow_json.schema.equals_schema(&schema));
1628 assert_eq!(arrow_json.get_record_batches().unwrap()[0], record_batch);
1630 }
1631}