1use std::io::Write;
18use std::sync::Arc;
19
20use crate::StructMode;
21use arrow_array::cast::AsArray;
22use arrow_array::types::*;
23use arrow_array::*;
24use arrow_buffer::{ArrowNativeType, NullBuffer, OffsetBuffer, ScalarBuffer};
25use arrow_cast::display::{ArrayFormatter, FormatOptions};
26use arrow_schema::{ArrowError, DataType, FieldRef};
27use half::f16;
28use lexical_core::FormattedSize;
29use serde_core::Serializer;
30
31#[derive(Debug, Clone, Default)]
33pub struct EncoderOptions {
34 explicit_nulls: bool,
36 struct_mode: StructMode,
38 encoder_factory: Option<Arc<dyn EncoderFactory>>,
40 date_format: Option<String>,
42 datetime_format: Option<String>,
44 timestamp_format: Option<String>,
46 timestamp_tz_format: Option<String>,
48 time_format: Option<String>,
50}
51
52impl EncoderOptions {
53 pub fn with_explicit_nulls(mut self, explicit_nulls: bool) -> Self {
55 self.explicit_nulls = explicit_nulls;
56 self
57 }
58
59 pub fn with_struct_mode(mut self, struct_mode: StructMode) -> Self {
61 self.struct_mode = struct_mode;
62 self
63 }
64
65 pub fn with_encoder_factory(mut self, encoder_factory: Arc<dyn EncoderFactory>) -> Self {
67 self.encoder_factory = Some(encoder_factory);
68 self
69 }
70
71 pub fn explicit_nulls(&self) -> bool {
73 self.explicit_nulls
74 }
75
76 pub fn struct_mode(&self) -> StructMode {
78 self.struct_mode
79 }
80
81 pub fn encoder_factory(&self) -> Option<&Arc<dyn EncoderFactory>> {
83 self.encoder_factory.as_ref()
84 }
85
86 pub fn with_date_format(mut self, format: String) -> Self {
88 self.date_format = Some(format);
89 self
90 }
91
92 pub fn date_format(&self) -> Option<&str> {
94 self.date_format.as_deref()
95 }
96
97 pub fn with_datetime_format(mut self, format: String) -> Self {
99 self.datetime_format = Some(format);
100 self
101 }
102
103 pub fn datetime_format(&self) -> Option<&str> {
105 self.datetime_format.as_deref()
106 }
107
108 pub fn with_time_format(mut self, format: String) -> Self {
110 self.time_format = Some(format);
111 self
112 }
113
114 pub fn time_format(&self) -> Option<&str> {
116 self.time_format.as_deref()
117 }
118
119 pub fn with_timestamp_format(mut self, format: String) -> Self {
121 self.timestamp_format = Some(format);
122 self
123 }
124
125 pub fn timestamp_format(&self) -> Option<&str> {
127 self.timestamp_format.as_deref()
128 }
129
130 pub fn with_timestamp_tz_format(mut self, tz_format: String) -> Self {
132 self.timestamp_tz_format = Some(tz_format);
133 self
134 }
135
136 pub fn timestamp_tz_format(&self) -> Option<&str> {
138 self.timestamp_tz_format.as_deref()
139 }
140}
141
142pub trait EncoderFactory: std::fmt::Debug + Send + Sync {
239 fn make_default_encoder<'a>(
247 &self,
248 _field: &'a FieldRef,
249 _array: &'a dyn Array,
250 _options: &'a EncoderOptions,
251 ) -> Result<Option<NullableEncoder<'a>>, ArrowError> {
252 Ok(None)
253 }
254}
255
256pub struct NullableEncoder<'a> {
259 encoder: Box<dyn Encoder + 'a>,
260 nulls: Option<NullBuffer>,
261}
262
263impl<'a> NullableEncoder<'a> {
264 #[inline]
266 pub fn new(encoder: Box<dyn Encoder + 'a>, nulls: Option<NullBuffer>) -> Self {
267 Self { encoder, nulls }
268 }
269
270 #[inline]
272 pub fn encode(&mut self, idx: usize, out: &mut Vec<u8>) {
273 self.encoder.encode(idx, out)
274 }
275
276 #[inline]
278 pub fn is_null(&self, idx: usize) -> bool {
279 match self.nulls {
280 Some(ref nulls) => nulls.is_null(idx),
281 None => false,
282 }
283 }
284
285 #[inline]
287 pub fn has_nulls(&self) -> bool {
288 match self.nulls {
289 Some(ref nulls) => nulls.null_count() > 0,
290 None => false,
291 }
292 }
293}
294
295impl Encoder for NullableEncoder<'_> {
296 #[inline]
297 fn encode(&mut self, idx: usize, out: &mut Vec<u8>) {
298 self.encoder.encode(idx, out)
299 }
300}
301
302pub trait Encoder {
306 fn encode(&mut self, idx: usize, out: &mut Vec<u8>);
310}
311
312pub fn make_encoder<'a>(
316 field: &'a FieldRef,
317 array: &'a dyn Array,
318 options: &'a EncoderOptions,
319) -> Result<NullableEncoder<'a>, ArrowError> {
320 macro_rules! primitive_helper {
321 ($t:ty) => {{
322 let array = array.as_primitive::<$t>();
323 let nulls = array.nulls().cloned();
324 NullableEncoder::new(Box::new(PrimitiveEncoder::new(array)), nulls)
325 }};
326 }
327
328 if let Some(factory) = options.encoder_factory()
329 && let Some(encoder) = factory.make_default_encoder(field, array, options)?
330 {
331 return Ok(encoder);
332 }
333
334 let nulls = array.nulls().cloned();
335 let encoder = downcast_integer! {
336 array.data_type() => (primitive_helper),
337 DataType::Float16 => primitive_helper!(Float16Type),
338 DataType::Float32 => primitive_helper!(Float32Type),
339 DataType::Float64 => primitive_helper!(Float64Type),
340 DataType::Boolean => {
341 let array = array.as_boolean();
342 NullableEncoder::new(Box::new(BooleanEncoder(array)), array.nulls().cloned())
343 }
344 DataType::Null => NullableEncoder::new(Box::new(NullEncoder), array.logical_nulls()),
345 DataType::Utf8 => {
346 let array = array.as_string::<i32>();
347 NullableEncoder::new(Box::new(StringEncoder(array)), array.nulls().cloned())
348 }
349 DataType::LargeUtf8 => {
350 let array = array.as_string::<i64>();
351 NullableEncoder::new(Box::new(StringEncoder(array)), array.nulls().cloned())
352 }
353 DataType::Utf8View => {
354 let array = array.as_string_view();
355 NullableEncoder::new(Box::new(StringViewEncoder(array)), array.nulls().cloned())
356 }
357 DataType::BinaryView => {
358 let array = array.as_binary_view();
359 NullableEncoder::new(Box::new(BinaryViewEncoder(array)), array.nulls().cloned())
360 }
361 DataType::List(_) => {
362 let array = array.as_list::<i32>();
363 NullableEncoder::new(Box::new(ListLikeEncoder::try_new(field, array, options)?), array.nulls().cloned())
364 }
365 DataType::LargeList(_) => {
366 let array = array.as_list::<i64>();
367 NullableEncoder::new(Box::new(ListLikeEncoder::try_new(field, array, options)?), array.nulls().cloned())
368 }
369 DataType::ListView(_) => {
370 let array = array.as_list_view::<i32>();
371 NullableEncoder::new(Box::new(ListLikeEncoder::try_new(field, array, options)?), array.nulls().cloned())
372 }
373 DataType::LargeListView(_) => {
374 let array = array.as_list_view::<i64>();
375 NullableEncoder::new(Box::new(ListLikeEncoder::try_new(field, array, options)?), array.nulls().cloned())
376 }
377 DataType::FixedSizeList(_, _) => {
378 let array = array.as_fixed_size_list();
379 NullableEncoder::new(Box::new(ListLikeEncoder::try_new(field, array, options)?), array.nulls().cloned())
380 }
381
382 DataType::Dictionary(_, _) => downcast_dictionary_array! {
383 array => {
384 NullableEncoder::new(Box::new(DictionaryEncoder::try_new(field, array, options)?), array.nulls().cloned())
385 },
386 _ => unreachable!()
387 }
388
389 DataType::RunEndEncoded(_, _) => downcast_run_array! {
390 array => {
391 NullableEncoder::new(
392 Box::new(RunEndEncodedEncoder::try_new(field, array, options)?),
393 array.logical_nulls(),
394 )
395 },
396 _ => unreachable!()
397 }
398
399 DataType::Map(_, _) => {
400 let array = array.as_map();
401 NullableEncoder::new(Box::new(MapEncoder::try_new(field, array, options)?), array.nulls().cloned())
402 }
403
404 DataType::FixedSizeBinary(_) => {
405 let array = array.as_fixed_size_binary();
406 NullableEncoder::new(Box::new(BinaryEncoder::new(array)) as _, array.nulls().cloned())
407 }
408
409 DataType::Binary => {
410 let array: &BinaryArray = array.as_binary();
411 NullableEncoder::new(Box::new(BinaryEncoder::new(array)), array.nulls().cloned())
412 }
413
414 DataType::LargeBinary => {
415 let array: &LargeBinaryArray = array.as_binary();
416 NullableEncoder::new(Box::new(BinaryEncoder::new(array)), array.nulls().cloned())
417 }
418
419 DataType::Struct(fields) => {
420 let array = array.as_struct();
421 let encoders = fields.iter().zip(array.columns()).map(|(field, array)| {
422 let encoder = make_encoder(field, array, options)?;
423
424 let mut field_name = Vec::with_capacity(field.name().len() + 3);
426 encode_string(field.name(), &mut field_name);
427 field_name.push(b':');
428
429 Ok(FieldEncoder {
430 field_name,
431 encoder,
432 })
433 }).collect::<Result<Vec<_>, ArrowError>>()?;
434
435 let encoder = StructArrayEncoder{
436 encoders,
437 explicit_nulls: options.explicit_nulls(),
438 struct_mode: options.struct_mode(),
439 };
440 let nulls = array.nulls().cloned();
441 NullableEncoder::new(Box::new(encoder) as Box<dyn Encoder + 'a>, nulls)
442 }
443 DataType::Decimal32(_, _) | DataType::Decimal64(_, _) | DataType::Decimal128(_, _) | DataType::Decimal256(_, _) => {
444 let options = FormatOptions::new().with_display_error(true);
445 let formatter = JsonArrayFormatter::new(ArrayFormatter::try_new(array, &options)?);
446 NullableEncoder::new(Box::new(RawArrayFormatter(formatter)) as Box<dyn Encoder + 'a>, nulls)
447 }
448 d => match d.is_temporal() {
449 true => {
450 let fops = FormatOptions::new().with_display_error(true)
455 .with_date_format(options.date_format.as_deref())
456 .with_datetime_format(options.datetime_format.as_deref())
457 .with_timestamp_format(options.timestamp_format.as_deref())
458 .with_timestamp_tz_format(options.timestamp_tz_format.as_deref())
459 .with_time_format(options.time_format.as_deref());
460
461 let formatter = ArrayFormatter::try_new(array, &fops)?;
462 let formatter = JsonArrayFormatter::new(formatter);
463 NullableEncoder::new(Box::new(formatter) as Box<dyn Encoder + 'a>, nulls)
464 }
465 false => return Err(ArrowError::JsonError(format!(
466 "Unsupported data type for JSON encoding: {d:?}",
467 )))
468 }
469 };
470
471 Ok(encoder)
472}
473
474fn encode_string(s: &str, out: &mut Vec<u8>) {
475 let mut serializer = serde_json::Serializer::new(out);
476 serializer.serialize_str(s).unwrap();
477}
478
479fn encode_binary(bytes: &[u8], out: &mut Vec<u8>) {
480 out.push(b'"');
481 for byte in bytes {
482 write!(out, "{byte:02x}").unwrap();
483 }
484 out.push(b'"');
485}
486
487struct FieldEncoder<'a> {
488 field_name: Vec<u8>,
489 encoder: NullableEncoder<'a>,
490}
491
492impl FieldEncoder<'_> {
493 #[inline]
494 fn is_null(&self, idx: usize) -> bool {
495 self.encoder.is_null(idx)
496 }
497}
498
499struct StructArrayEncoder<'a> {
500 encoders: Vec<FieldEncoder<'a>>,
501 explicit_nulls: bool,
502 struct_mode: StructMode,
503}
504
505impl Encoder for StructArrayEncoder<'_> {
506 fn encode(&mut self, idx: usize, out: &mut Vec<u8>) {
507 match self.struct_mode {
508 StructMode::ObjectOnly => out.push(b'{'),
509 StructMode::ListOnly => out.push(b'['),
510 }
511 let mut is_first = true;
512 let drop_nulls = (self.struct_mode == StructMode::ObjectOnly) && !self.explicit_nulls;
514
515 for field_encoder in self.encoders.iter_mut() {
516 let is_null = field_encoder.is_null(idx);
517 if is_null && drop_nulls {
518 continue;
519 }
520
521 if !is_first {
522 out.push(b',');
523 }
524 is_first = false;
525
526 if self.struct_mode == StructMode::ObjectOnly {
527 out.extend_from_slice(&field_encoder.field_name);
528 }
529
530 if is_null {
531 out.extend_from_slice(b"null");
532 } else {
533 field_encoder.encoder.encode(idx, out);
534 }
535 }
536 match self.struct_mode {
537 StructMode::ObjectOnly => out.push(b'}'),
538 StructMode::ListOnly => out.push(b']'),
539 }
540 }
541}
542
543trait PrimitiveEncode: ArrowNativeType {
544 type Buffer;
545
546 fn init_buffer() -> Self::Buffer;
548
549 fn encode(self, buf: &mut Self::Buffer) -> &[u8];
553}
554
555macro_rules! integer_encode {
556 ($($t:ty),*) => {
557 $(
558 impl PrimitiveEncode for $t {
559 type Buffer = [u8; Self::FORMATTED_SIZE];
560
561 fn init_buffer() -> Self::Buffer {
562 [0; Self::FORMATTED_SIZE]
563 }
564
565 fn encode(self, buf: &mut Self::Buffer) -> &[u8] {
566 lexical_core::write(self, buf)
567 }
568 }
569 )*
570 };
571}
572integer_encode!(i8, i16, i32, i64, u8, u16, u32, u64);
573
574macro_rules! float_encode {
575 ($($t:ty),*) => {
576 $(
577 impl PrimitiveEncode for $t {
578 type Buffer = [u8; Self::FORMATTED_SIZE];
579
580 fn init_buffer() -> Self::Buffer {
581 [0; Self::FORMATTED_SIZE]
582 }
583
584 fn encode(self, buf: &mut Self::Buffer) -> &[u8] {
585 if self.is_infinite() || self.is_nan() {
586 b"null"
587 } else {
588 lexical_core::write(self, buf)
589 }
590 }
591 }
592 )*
593 };
594}
595float_encode!(f32, f64);
596
597impl PrimitiveEncode for f16 {
598 type Buffer = <f32 as PrimitiveEncode>::Buffer;
599
600 fn init_buffer() -> Self::Buffer {
601 f32::init_buffer()
602 }
603
604 fn encode(self, buf: &mut Self::Buffer) -> &[u8] {
605 self.to_f32().encode(buf)
606 }
607}
608
609struct PrimitiveEncoder<N: PrimitiveEncode> {
610 values: ScalarBuffer<N>,
611 buffer: N::Buffer,
612}
613
614impl<N: PrimitiveEncode> PrimitiveEncoder<N> {
615 fn new<P: ArrowPrimitiveType<Native = N>>(array: &PrimitiveArray<P>) -> Self {
616 Self {
617 values: array.values().clone(),
618 buffer: N::init_buffer(),
619 }
620 }
621}
622
623impl<N: PrimitiveEncode> Encoder for PrimitiveEncoder<N> {
624 fn encode(&mut self, idx: usize, out: &mut Vec<u8>) {
625 out.extend_from_slice(self.values[idx].encode(&mut self.buffer));
626 }
627}
628
629struct BooleanEncoder<'a>(&'a BooleanArray);
630
631impl Encoder for BooleanEncoder<'_> {
632 fn encode(&mut self, idx: usize, out: &mut Vec<u8>) {
633 match self.0.value(idx) {
634 true => out.extend_from_slice(b"true"),
635 false => out.extend_from_slice(b"false"),
636 }
637 }
638}
639
640struct StringEncoder<'a, O: OffsetSizeTrait>(&'a GenericStringArray<O>);
641
642impl<O: OffsetSizeTrait> Encoder for StringEncoder<'_, O> {
643 fn encode(&mut self, idx: usize, out: &mut Vec<u8>) {
644 encode_string(self.0.value(idx), out);
645 }
646}
647
648struct StringViewEncoder<'a>(&'a StringViewArray);
649
650impl Encoder for StringViewEncoder<'_> {
651 fn encode(&mut self, idx: usize, out: &mut Vec<u8>) {
652 encode_string(self.0.value(idx), out);
653 }
654}
655
656struct BinaryViewEncoder<'a>(&'a BinaryViewArray);
657
658impl Encoder for BinaryViewEncoder<'_> {
659 fn encode(&mut self, idx: usize, out: &mut Vec<u8>) {
660 encode_binary(self.0.value(idx), out);
661 }
662}
663
664struct ListLikeEncoder<'a, L: ListLikeArray> {
665 list_array: &'a L,
666 encoder: NullableEncoder<'a>,
667}
668
669impl<'a, L: ListLikeArray> ListLikeEncoder<'a, L> {
670 fn try_new(
671 field: &'a FieldRef,
672 array: &'a L,
673 options: &'a EncoderOptions,
674 ) -> Result<Self, ArrowError> {
675 let encoder = make_encoder(field, array.values().as_ref(), options)?;
676 Ok(Self {
677 list_array: array,
678 encoder,
679 })
680 }
681}
682
683impl<L: ListLikeArray> Encoder for ListLikeEncoder<'_, L> {
684 fn encode(&mut self, idx: usize, out: &mut Vec<u8>) {
685 let range = self.list_array.element_range(idx);
686 let start = range.start;
687 let end = range.end;
688 out.push(b'[');
689 if self.encoder.has_nulls() {
690 for idx in start..end {
691 if idx != start {
692 out.push(b',')
693 }
694 if self.encoder.is_null(idx) {
695 out.extend_from_slice(b"null");
696 } else {
697 self.encoder.encode(idx, out);
698 }
699 }
700 } else {
701 for idx in start..end {
702 if idx != start {
703 out.push(b',')
704 }
705 self.encoder.encode(idx, out);
706 }
707 }
708 out.push(b']');
709 }
710}
711
712struct DictionaryEncoder<'a, K: ArrowDictionaryKeyType> {
713 keys: ScalarBuffer<K::Native>,
714 encoder: NullableEncoder<'a>,
715}
716
717impl<'a, K: ArrowDictionaryKeyType> DictionaryEncoder<'a, K> {
718 fn try_new(
719 field: &'a FieldRef,
720 array: &'a DictionaryArray<K>,
721 options: &'a EncoderOptions,
722 ) -> Result<Self, ArrowError> {
723 let encoder = make_encoder(field, array.values().as_ref(), options)?;
724
725 Ok(Self {
726 keys: array.keys().values().clone(),
727 encoder,
728 })
729 }
730}
731
732impl<K: ArrowDictionaryKeyType> Encoder for DictionaryEncoder<'_, K> {
733 fn encode(&mut self, idx: usize, out: &mut Vec<u8>) {
734 self.encoder.encode(self.keys[idx].as_usize(), out)
735 }
736}
737
738struct RunEndEncodedEncoder<'a, R: RunEndIndexType> {
739 run_array: &'a RunArray<R>,
740 encoder: NullableEncoder<'a>,
741}
742
743impl<'a, R: RunEndIndexType> RunEndEncodedEncoder<'a, R> {
744 fn try_new(
745 field: &'a FieldRef,
746 array: &'a RunArray<R>,
747 options: &'a EncoderOptions,
748 ) -> Result<Self, ArrowError> {
749 let encoder = make_encoder(field, array.values().as_ref(), options)?;
750 Ok(Self {
751 run_array: array,
752 encoder,
753 })
754 }
755}
756
757impl<R: RunEndIndexType> Encoder for RunEndEncodedEncoder<'_, R> {
758 fn encode(&mut self, idx: usize, out: &mut Vec<u8>) {
759 let physical_idx = self.run_array.get_physical_index(idx);
760 self.encoder.encode(physical_idx, out)
761 }
762}
763
764struct JsonArrayFormatter<'a> {
766 formatter: ArrayFormatter<'a>,
767}
768
769impl<'a> JsonArrayFormatter<'a> {
770 fn new(formatter: ArrayFormatter<'a>) -> Self {
771 Self { formatter }
772 }
773}
774
775impl Encoder for JsonArrayFormatter<'_> {
776 fn encode(&mut self, idx: usize, out: &mut Vec<u8>) {
777 out.push(b'"');
778 let _ = write!(out, "{}", self.formatter.value(idx));
781 out.push(b'"')
782 }
783}
784
785struct RawArrayFormatter<'a>(JsonArrayFormatter<'a>);
787
788impl Encoder for RawArrayFormatter<'_> {
789 fn encode(&mut self, idx: usize, out: &mut Vec<u8>) {
790 let _ = write!(out, "{}", self.0.formatter.value(idx));
791 }
792}
793
794struct NullEncoder;
795
796impl Encoder for NullEncoder {
797 fn encode(&mut self, _idx: usize, _out: &mut Vec<u8>) {
798 unreachable!()
799 }
800}
801
802struct MapEncoder<'a> {
803 offsets: OffsetBuffer<i32>,
804 keys: NullableEncoder<'a>,
805 values: NullableEncoder<'a>,
806 explicit_nulls: bool,
807}
808
809impl<'a> MapEncoder<'a> {
810 fn try_new(
811 field: &'a FieldRef,
812 array: &'a MapArray,
813 options: &'a EncoderOptions,
814 ) -> Result<Self, ArrowError> {
815 let values = array.values();
816 let keys = array.keys();
817
818 if !matches!(
819 keys.data_type(),
820 DataType::Utf8 | DataType::LargeUtf8 | DataType::Utf8View
821 ) {
822 return Err(ArrowError::JsonError(format!(
823 "Only UTF8 keys supported by JSON MapArray Writer: got {:?}",
824 keys.data_type()
825 )));
826 }
827
828 let keys = make_encoder(field, keys, options)?;
829 let values = make_encoder(field, values, options)?;
830
831 if keys.has_nulls() {
833 return Err(ArrowError::InvalidArgumentError(
834 "Encountered nulls in MapArray keys".to_string(),
835 ));
836 }
837
838 if array.entries().nulls().is_some_and(|x| x.null_count() != 0) {
839 return Err(ArrowError::InvalidArgumentError(
840 "Encountered nulls in MapArray entries".to_string(),
841 ));
842 }
843
844 Ok(Self {
845 offsets: array.offsets().clone(),
846 keys,
847 values,
848 explicit_nulls: options.explicit_nulls(),
849 })
850 }
851}
852
853impl Encoder for MapEncoder<'_> {
854 fn encode(&mut self, idx: usize, out: &mut Vec<u8>) {
855 let end = self.offsets[idx + 1].as_usize();
856 let start = self.offsets[idx].as_usize();
857
858 let mut is_first = true;
859
860 out.push(b'{');
861
862 for idx in start..end {
863 let is_null = self.values.is_null(idx);
864 if is_null && !self.explicit_nulls {
865 continue;
866 }
867
868 if !is_first {
869 out.push(b',');
870 }
871 is_first = false;
872
873 self.keys.encode(idx, out);
874 out.push(b':');
875
876 if is_null {
877 out.extend_from_slice(b"null");
878 } else {
879 self.values.encode(idx, out);
880 }
881 }
882 out.push(b'}');
883 }
884}
885
886struct BinaryEncoder<B>(B);
889
890impl<'a, B> BinaryEncoder<B>
891where
892 B: ArrayAccessor<Item = &'a [u8]>,
893{
894 fn new(array: B) -> Self {
895 Self(array)
896 }
897}
898
899impl<'a, B> Encoder for BinaryEncoder<B>
900where
901 B: ArrayAccessor<Item = &'a [u8]>,
902{
903 fn encode(&mut self, idx: usize, out: &mut Vec<u8>) {
904 out.push(b'"');
905 for byte in self.0.value(idx) {
906 write!(out, "{byte:02x}").unwrap();
908 }
909 out.push(b'"');
910 }
911}