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 pub fn new(encoder: Box<dyn Encoder + 'a>, nulls: Option<NullBuffer>) -> Self {
266 Self { encoder, nulls }
267 }
268
269 pub fn encode(&mut self, idx: usize, out: &mut Vec<u8>) {
271 self.encoder.encode(idx, out)
272 }
273
274 pub fn is_null(&self, idx: usize) -> bool {
276 self.nulls.as_ref().is_some_and(|nulls| nulls.is_null(idx))
277 }
278
279 pub fn has_nulls(&self) -> bool {
281 match self.nulls {
282 Some(ref nulls) => nulls.null_count() > 0,
283 None => false,
284 }
285 }
286}
287
288impl Encoder for NullableEncoder<'_> {
289 fn encode(&mut self, idx: usize, out: &mut Vec<u8>) {
290 self.encoder.encode(idx, out)
291 }
292}
293
294pub trait Encoder {
298 fn encode(&mut self, idx: usize, out: &mut Vec<u8>);
302}
303
304pub fn make_encoder<'a>(
308 field: &'a FieldRef,
309 array: &'a dyn Array,
310 options: &'a EncoderOptions,
311) -> Result<NullableEncoder<'a>, ArrowError> {
312 macro_rules! primitive_helper {
313 ($t:ty) => {{
314 let array = array.as_primitive::<$t>();
315 let nulls = array.nulls().cloned();
316 NullableEncoder::new(Box::new(PrimitiveEncoder::new(array)), nulls)
317 }};
318 }
319
320 if let Some(factory) = options.encoder_factory() {
321 if let Some(encoder) = factory.make_default_encoder(field, array, options)? {
322 return Ok(encoder);
323 }
324 }
325
326 let nulls = array.nulls().cloned();
327 let encoder = downcast_integer! {
328 array.data_type() => (primitive_helper),
329 DataType::Float16 => primitive_helper!(Float16Type),
330 DataType::Float32 => primitive_helper!(Float32Type),
331 DataType::Float64 => primitive_helper!(Float64Type),
332 DataType::Boolean => {
333 let array = array.as_boolean();
334 NullableEncoder::new(Box::new(BooleanEncoder(array)), array.nulls().cloned())
335 }
336 DataType::Null => NullableEncoder::new(Box::new(NullEncoder), array.logical_nulls()),
337 DataType::Utf8 => {
338 let array = array.as_string::<i32>();
339 NullableEncoder::new(Box::new(StringEncoder(array)), array.nulls().cloned())
340 }
341 DataType::LargeUtf8 => {
342 let array = array.as_string::<i64>();
343 NullableEncoder::new(Box::new(StringEncoder(array)), array.nulls().cloned())
344 }
345 DataType::Utf8View => {
346 let array = array.as_string_view();
347 NullableEncoder::new(Box::new(StringViewEncoder(array)), array.nulls().cloned())
348 }
349 DataType::BinaryView => {
350 let array = array.as_binary_view();
351 NullableEncoder::new(Box::new(BinaryViewEncoder(array)), array.nulls().cloned())
352 }
353 DataType::List(_) => {
354 let array = array.as_list::<i32>();
355 NullableEncoder::new(Box::new(ListLikeEncoder::try_new(field, array, options)?), array.nulls().cloned())
356 }
357 DataType::LargeList(_) => {
358 let array = array.as_list::<i64>();
359 NullableEncoder::new(Box::new(ListLikeEncoder::try_new(field, array, options)?), array.nulls().cloned())
360 }
361 DataType::ListView(_) => {
362 let array = array.as_list_view::<i32>();
363 NullableEncoder::new(Box::new(ListLikeEncoder::try_new(field, array, options)?), array.nulls().cloned())
364 }
365 DataType::LargeListView(_) => {
366 let array = array.as_list_view::<i64>();
367 NullableEncoder::new(Box::new(ListLikeEncoder::try_new(field, array, options)?), array.nulls().cloned())
368 }
369 DataType::FixedSizeList(_, _) => {
370 let array = array.as_fixed_size_list();
371 NullableEncoder::new(Box::new(ListLikeEncoder::try_new(field, array, options)?), array.nulls().cloned())
372 }
373
374 DataType::Dictionary(_, _) => downcast_dictionary_array! {
375 array => {
376 NullableEncoder::new(Box::new(DictionaryEncoder::try_new(field, array, options)?), array.nulls().cloned())
377 },
378 _ => unreachable!()
379 }
380
381 DataType::RunEndEncoded(_, _) => downcast_run_array! {
382 array => {
383 NullableEncoder::new(
384 Box::new(RunEndEncodedEncoder::try_new(field, array, options)?),
385 array.logical_nulls(),
386 )
387 },
388 _ => unreachable!()
389 }
390
391 DataType::Map(_, _) => {
392 let array = array.as_map();
393 NullableEncoder::new(Box::new(MapEncoder::try_new(field, array, options)?), array.nulls().cloned())
394 }
395
396 DataType::FixedSizeBinary(_) => {
397 let array = array.as_fixed_size_binary();
398 NullableEncoder::new(Box::new(BinaryEncoder::new(array)) as _, array.nulls().cloned())
399 }
400
401 DataType::Binary => {
402 let array: &BinaryArray = array.as_binary();
403 NullableEncoder::new(Box::new(BinaryEncoder::new(array)), array.nulls().cloned())
404 }
405
406 DataType::LargeBinary => {
407 let array: &LargeBinaryArray = array.as_binary();
408 NullableEncoder::new(Box::new(BinaryEncoder::new(array)), array.nulls().cloned())
409 }
410
411 DataType::Struct(fields) => {
412 let array = array.as_struct();
413 let encoders = fields.iter().zip(array.columns()).map(|(field, array)| {
414 let encoder = make_encoder(field, array, options)?;
415
416 let mut field_name = Vec::with_capacity(field.name().len() + 3);
418 encode_string(field.name(), &mut field_name);
419 field_name.push(b':');
420
421 Ok(FieldEncoder {
422 field_name,
423 encoder,
424 })
425 }).collect::<Result<Vec<_>, ArrowError>>()?;
426
427 let encoder = StructArrayEncoder{
428 encoders,
429 explicit_nulls: options.explicit_nulls(),
430 struct_mode: options.struct_mode(),
431 };
432 let nulls = array.nulls().cloned();
433 NullableEncoder::new(Box::new(encoder) as Box<dyn Encoder + 'a>, nulls)
434 }
435 DataType::Decimal32(_, _) | DataType::Decimal64(_, _) | DataType::Decimal128(_, _) | DataType::Decimal256(_, _) => {
436 let options = FormatOptions::new().with_display_error(true);
437 let formatter = JsonArrayFormatter::new(ArrayFormatter::try_new(array, &options)?);
438 NullableEncoder::new(Box::new(RawArrayFormatter(formatter)) as Box<dyn Encoder + 'a>, nulls)
439 }
440 d => match d.is_temporal() {
441 true => {
442 let fops = FormatOptions::new().with_display_error(true)
447 .with_date_format(options.date_format.as_deref())
448 .with_datetime_format(options.datetime_format.as_deref())
449 .with_timestamp_format(options.timestamp_format.as_deref())
450 .with_timestamp_tz_format(options.timestamp_tz_format.as_deref())
451 .with_time_format(options.time_format.as_deref());
452
453 let formatter = ArrayFormatter::try_new(array, &fops)?;
454 let formatter = JsonArrayFormatter::new(formatter);
455 NullableEncoder::new(Box::new(formatter) as Box<dyn Encoder + 'a>, nulls)
456 }
457 false => return Err(ArrowError::JsonError(format!(
458 "Unsupported data type for JSON encoding: {d:?}",
459 )))
460 }
461 };
462
463 Ok(encoder)
464}
465
466fn encode_string(s: &str, out: &mut Vec<u8>) {
467 let mut serializer = serde_json::Serializer::new(out);
468 serializer.serialize_str(s).unwrap();
469}
470
471fn encode_binary(bytes: &[u8], out: &mut Vec<u8>) {
472 out.push(b'"');
473 for byte in bytes {
474 write!(out, "{byte:02x}").unwrap();
475 }
476 out.push(b'"');
477}
478
479struct FieldEncoder<'a> {
480 field_name: Vec<u8>,
481 encoder: NullableEncoder<'a>,
482}
483
484impl FieldEncoder<'_> {
485 fn is_null(&self, idx: usize) -> bool {
486 self.encoder.is_null(idx)
487 }
488}
489
490struct StructArrayEncoder<'a> {
491 encoders: Vec<FieldEncoder<'a>>,
492 explicit_nulls: bool,
493 struct_mode: StructMode,
494}
495
496impl Encoder for StructArrayEncoder<'_> {
497 fn encode(&mut self, idx: usize, out: &mut Vec<u8>) {
498 match self.struct_mode {
499 StructMode::ObjectOnly => out.push(b'{'),
500 StructMode::ListOnly => out.push(b'['),
501 }
502 let mut is_first = true;
503 let drop_nulls = (self.struct_mode == StructMode::ObjectOnly) && !self.explicit_nulls;
505
506 for field_encoder in self.encoders.iter_mut() {
507 let is_null = field_encoder.is_null(idx);
508 if is_null && drop_nulls {
509 continue;
510 }
511
512 if !is_first {
513 out.push(b',');
514 }
515 is_first = false;
516
517 if self.struct_mode == StructMode::ObjectOnly {
518 out.extend_from_slice(&field_encoder.field_name);
519 }
520
521 if is_null {
522 out.extend_from_slice(b"null");
523 } else {
524 field_encoder.encoder.encode(idx, out);
525 }
526 }
527 match self.struct_mode {
528 StructMode::ObjectOnly => out.push(b'}'),
529 StructMode::ListOnly => out.push(b']'),
530 }
531 }
532}
533
534trait PrimitiveEncode: ArrowNativeType {
535 type Buffer;
536
537 fn init_buffer() -> Self::Buffer;
539
540 fn encode(self, buf: &mut Self::Buffer) -> &[u8];
544}
545
546macro_rules! integer_encode {
547 ($($t:ty),*) => {
548 $(
549 impl PrimitiveEncode for $t {
550 type Buffer = [u8; Self::FORMATTED_SIZE];
551
552 fn init_buffer() -> Self::Buffer {
553 [0; Self::FORMATTED_SIZE]
554 }
555
556 fn encode(self, buf: &mut Self::Buffer) -> &[u8] {
557 lexical_core::write(self, buf)
558 }
559 }
560 )*
561 };
562}
563integer_encode!(i8, i16, i32, i64, u8, u16, u32, u64);
564
565macro_rules! float_encode {
566 ($($t:ty),*) => {
567 $(
568 impl PrimitiveEncode for $t {
569 type Buffer = [u8; Self::FORMATTED_SIZE];
570
571 fn init_buffer() -> Self::Buffer {
572 [0; Self::FORMATTED_SIZE]
573 }
574
575 fn encode(self, buf: &mut Self::Buffer) -> &[u8] {
576 if self.is_infinite() || self.is_nan() {
577 b"null"
578 } else {
579 lexical_core::write(self, buf)
580 }
581 }
582 }
583 )*
584 };
585}
586float_encode!(f32, f64);
587
588impl PrimitiveEncode for f16 {
589 type Buffer = <f32 as PrimitiveEncode>::Buffer;
590
591 fn init_buffer() -> Self::Buffer {
592 f32::init_buffer()
593 }
594
595 fn encode(self, buf: &mut Self::Buffer) -> &[u8] {
596 self.to_f32().encode(buf)
597 }
598}
599
600struct PrimitiveEncoder<N: PrimitiveEncode> {
601 values: ScalarBuffer<N>,
602 buffer: N::Buffer,
603}
604
605impl<N: PrimitiveEncode> PrimitiveEncoder<N> {
606 fn new<P: ArrowPrimitiveType<Native = N>>(array: &PrimitiveArray<P>) -> Self {
607 Self {
608 values: array.values().clone(),
609 buffer: N::init_buffer(),
610 }
611 }
612}
613
614impl<N: PrimitiveEncode> Encoder for PrimitiveEncoder<N> {
615 fn encode(&mut self, idx: usize, out: &mut Vec<u8>) {
616 out.extend_from_slice(self.values[idx].encode(&mut self.buffer));
617 }
618}
619
620struct BooleanEncoder<'a>(&'a BooleanArray);
621
622impl Encoder for BooleanEncoder<'_> {
623 fn encode(&mut self, idx: usize, out: &mut Vec<u8>) {
624 match self.0.value(idx) {
625 true => out.extend_from_slice(b"true"),
626 false => out.extend_from_slice(b"false"),
627 }
628 }
629}
630
631struct StringEncoder<'a, O: OffsetSizeTrait>(&'a GenericStringArray<O>);
632
633impl<O: OffsetSizeTrait> Encoder for StringEncoder<'_, O> {
634 fn encode(&mut self, idx: usize, out: &mut Vec<u8>) {
635 encode_string(self.0.value(idx), out);
636 }
637}
638
639struct StringViewEncoder<'a>(&'a StringViewArray);
640
641impl Encoder for StringViewEncoder<'_> {
642 fn encode(&mut self, idx: usize, out: &mut Vec<u8>) {
643 encode_string(self.0.value(idx), out);
644 }
645}
646
647struct BinaryViewEncoder<'a>(&'a BinaryViewArray);
648
649impl Encoder for BinaryViewEncoder<'_> {
650 fn encode(&mut self, idx: usize, out: &mut Vec<u8>) {
651 encode_binary(self.0.value(idx), out);
652 }
653}
654
655struct ListLikeEncoder<'a, L: ListLikeArray> {
656 list_array: &'a L,
657 encoder: NullableEncoder<'a>,
658}
659
660impl<'a, L: ListLikeArray> ListLikeEncoder<'a, L> {
661 fn try_new(
662 field: &'a FieldRef,
663 array: &'a L,
664 options: &'a EncoderOptions,
665 ) -> Result<Self, ArrowError> {
666 let encoder = make_encoder(field, array.values().as_ref(), options)?;
667 Ok(Self {
668 list_array: array,
669 encoder,
670 })
671 }
672}
673
674impl<L: ListLikeArray> Encoder for ListLikeEncoder<'_, L> {
675 fn encode(&mut self, idx: usize, out: &mut Vec<u8>) {
676 let range = self.list_array.element_range(idx);
677 let start = range.start;
678 let end = range.end;
679 out.push(b'[');
680 if self.encoder.has_nulls() {
681 for idx in start..end {
682 if idx != start {
683 out.push(b',')
684 }
685 if self.encoder.is_null(idx) {
686 out.extend_from_slice(b"null");
687 } else {
688 self.encoder.encode(idx, out);
689 }
690 }
691 } else {
692 for idx in start..end {
693 if idx != start {
694 out.push(b',')
695 }
696 self.encoder.encode(idx, out);
697 }
698 }
699 out.push(b']');
700 }
701}
702
703struct DictionaryEncoder<'a, K: ArrowDictionaryKeyType> {
704 keys: ScalarBuffer<K::Native>,
705 encoder: NullableEncoder<'a>,
706}
707
708impl<'a, K: ArrowDictionaryKeyType> DictionaryEncoder<'a, K> {
709 fn try_new(
710 field: &'a FieldRef,
711 array: &'a DictionaryArray<K>,
712 options: &'a EncoderOptions,
713 ) -> Result<Self, ArrowError> {
714 let encoder = make_encoder(field, array.values().as_ref(), options)?;
715
716 Ok(Self {
717 keys: array.keys().values().clone(),
718 encoder,
719 })
720 }
721}
722
723impl<K: ArrowDictionaryKeyType> Encoder for DictionaryEncoder<'_, K> {
724 fn encode(&mut self, idx: usize, out: &mut Vec<u8>) {
725 self.encoder.encode(self.keys[idx].as_usize(), out)
726 }
727}
728
729struct RunEndEncodedEncoder<'a, R: RunEndIndexType> {
730 run_array: &'a RunArray<R>,
731 encoder: NullableEncoder<'a>,
732}
733
734impl<'a, R: RunEndIndexType> RunEndEncodedEncoder<'a, R> {
735 fn try_new(
736 field: &'a FieldRef,
737 array: &'a RunArray<R>,
738 options: &'a EncoderOptions,
739 ) -> Result<Self, ArrowError> {
740 let encoder = make_encoder(field, array.values().as_ref(), options)?;
741 Ok(Self {
742 run_array: array,
743 encoder,
744 })
745 }
746}
747
748impl<R: RunEndIndexType> Encoder for RunEndEncodedEncoder<'_, R> {
749 fn encode(&mut self, idx: usize, out: &mut Vec<u8>) {
750 let physical_idx = self.run_array.get_physical_index(idx);
751 self.encoder.encode(physical_idx, out)
752 }
753}
754
755struct JsonArrayFormatter<'a> {
757 formatter: ArrayFormatter<'a>,
758}
759
760impl<'a> JsonArrayFormatter<'a> {
761 fn new(formatter: ArrayFormatter<'a>) -> Self {
762 Self { formatter }
763 }
764}
765
766impl Encoder for JsonArrayFormatter<'_> {
767 fn encode(&mut self, idx: usize, out: &mut Vec<u8>) {
768 out.push(b'"');
769 let _ = write!(out, "{}", self.formatter.value(idx));
772 out.push(b'"')
773 }
774}
775
776struct RawArrayFormatter<'a>(JsonArrayFormatter<'a>);
778
779impl Encoder for RawArrayFormatter<'_> {
780 fn encode(&mut self, idx: usize, out: &mut Vec<u8>) {
781 let _ = write!(out, "{}", self.0.formatter.value(idx));
782 }
783}
784
785struct NullEncoder;
786
787impl Encoder for NullEncoder {
788 fn encode(&mut self, _idx: usize, _out: &mut Vec<u8>) {
789 unreachable!()
790 }
791}
792
793struct MapEncoder<'a> {
794 offsets: OffsetBuffer<i32>,
795 keys: NullableEncoder<'a>,
796 values: NullableEncoder<'a>,
797 explicit_nulls: bool,
798}
799
800impl<'a> MapEncoder<'a> {
801 fn try_new(
802 field: &'a FieldRef,
803 array: &'a MapArray,
804 options: &'a EncoderOptions,
805 ) -> Result<Self, ArrowError> {
806 let values = array.values();
807 let keys = array.keys();
808
809 if !matches!(
810 keys.data_type(),
811 DataType::Utf8 | DataType::LargeUtf8 | DataType::Utf8View
812 ) {
813 return Err(ArrowError::JsonError(format!(
814 "Only UTF8 keys supported by JSON MapArray Writer: got {:?}",
815 keys.data_type()
816 )));
817 }
818
819 let keys = make_encoder(field, keys, options)?;
820 let values = make_encoder(field, values, options)?;
821
822 if keys.has_nulls() {
824 return Err(ArrowError::InvalidArgumentError(
825 "Encountered nulls in MapArray keys".to_string(),
826 ));
827 }
828
829 if array.entries().nulls().is_some_and(|x| x.null_count() != 0) {
830 return Err(ArrowError::InvalidArgumentError(
831 "Encountered nulls in MapArray entries".to_string(),
832 ));
833 }
834
835 Ok(Self {
836 offsets: array.offsets().clone(),
837 keys,
838 values,
839 explicit_nulls: options.explicit_nulls(),
840 })
841 }
842}
843
844impl Encoder for MapEncoder<'_> {
845 fn encode(&mut self, idx: usize, out: &mut Vec<u8>) {
846 let end = self.offsets[idx + 1].as_usize();
847 let start = self.offsets[idx].as_usize();
848
849 let mut is_first = true;
850
851 out.push(b'{');
852
853 for idx in start..end {
854 let is_null = self.values.is_null(idx);
855 if is_null && !self.explicit_nulls {
856 continue;
857 }
858
859 if !is_first {
860 out.push(b',');
861 }
862 is_first = false;
863
864 self.keys.encode(idx, out);
865 out.push(b':');
866
867 if is_null {
868 out.extend_from_slice(b"null");
869 } else {
870 self.values.encode(idx, out);
871 }
872 }
873 out.push(b'}');
874 }
875}
876
877struct BinaryEncoder<B>(B);
880
881impl<'a, B> BinaryEncoder<B>
882where
883 B: ArrayAccessor<Item = &'a [u8]>,
884{
885 fn new(array: B) -> Self {
886 Self(array)
887 }
888}
889
890impl<'a, B> Encoder for BinaryEncoder<B>
891where
892 B: ArrayAccessor<Item = &'a [u8]>,
893{
894 fn encode(&mut self, idx: usize, out: &mut Vec<u8>) {
895 out.push(b'"');
896 for byte in self.0.value(idx) {
897 write!(out, "{byte:02x}").unwrap();
899 }
900 out.push(b'"');
901 }
902}