1use super::error::{ArrowCompatError, ArrowResult};
11use super::traits::{FromArrowArray, ToArrowArray, ZeroCopyFromArrow};
12use arrow::array::{
13 Array as ArrowArray, ArrayRef, AsArray, BooleanArray, Float32Array, Float64Array, Int32Array,
14 Int64Array, StringArray,
15};
16use arrow::buffer::Buffer;
17use arrow::datatypes::{
18 ArrowPrimitiveType, DataType, Field, Float32Type, Float64Type, Int32Type, Int64Type, Schema,
19};
20use arrow::record_batch::RecordBatch;
21use ndarray::{Array1, Array2, ArrayView1, Axis};
22use std::sync::Arc;
23
24macro_rules! impl_primitive_arrow_conversion {
29 ($rust_type:ty, $arrow_type:ty, $data_type:expr, $array_type:ty, $type_name:expr) => {
30 impl ToArrowArray for $rust_type {
31 fn to_arrow_array(data: &[Self]) -> ArrowResult<ArrayRef> {
32 Ok(Arc::new(<$array_type>::from(Vec::from(data))))
35 }
36
37 fn arrow_data_type() -> DataType {
38 $data_type
39 }
40 }
41
42 impl FromArrowArray for $rust_type {
43 fn from_arrow_array(array: &ArrayRef) -> ArrowResult<Array1<Self>> {
44 let typed = array.as_primitive_opt::<$arrow_type>().ok_or_else(|| {
45 ArrowCompatError::TypeMismatch {
46 expected: $type_name.to_string(),
47 actual: format!("{:?}", array.data_type()),
48 }
49 })?;
50
51 if typed.null_count() > 0 {
53 return Err(ArrowCompatError::NullValuesPresent {
54 null_count: typed.null_count(),
55 total_count: typed.len(),
56 });
57 }
58
59 let values: Vec<$rust_type> = typed.values().iter().copied().collect();
60 Ok(Array1::from_vec(values))
61 }
62
63 fn from_arrow_array_nullable(array: &ArrayRef) -> ArrowResult<Array1<Option<Self>>> {
64 let typed = array.as_primitive_opt::<$arrow_type>().ok_or_else(|| {
65 ArrowCompatError::TypeMismatch {
66 expected: $type_name.to_string(),
67 actual: format!("{:?}", array.data_type()),
68 }
69 })?;
70
71 let values: Vec<Option<$rust_type>> = (0..typed.len())
72 .map(|i| {
73 if typed.is_null(i) {
74 None
75 } else {
76 Some(typed.value(i))
77 }
78 })
79 .collect();
80 Ok(Array1::from_vec(values))
81 }
82 }
83
84 impl ZeroCopyFromArrow for $rust_type {
85 fn try_zero_copy_view(array: &ArrayRef) -> ArrowResult<Option<ArrayView1<'_, Self>>> {
86 let typed = array.as_primitive_opt::<$arrow_type>().ok_or_else(|| {
87 ArrowCompatError::TypeMismatch {
88 expected: $type_name.to_string(),
89 actual: format!("{:?}", array.data_type()),
90 }
91 })?;
92
93 if typed.null_count() > 0 {
95 return Ok(None);
96 }
97
98 let values_slice: &[$rust_type] = typed.values();
101 let view = ArrayView1::from(values_slice);
102 Ok(Some(view))
103 }
104 }
105 };
106}
107
108impl_primitive_arrow_conversion!(f64, Float64Type, DataType::Float64, Float64Array, "Float64");
110impl_primitive_arrow_conversion!(f32, Float32Type, DataType::Float32, Float32Array, "Float32");
111impl_primitive_arrow_conversion!(i64, Int64Type, DataType::Int64, Int64Array, "Int64");
112impl_primitive_arrow_conversion!(i32, Int32Type, DataType::Int32, Int32Array, "Int32");
113
114impl ToArrowArray for bool {
119 fn to_arrow_array(data: &[Self]) -> ArrowResult<ArrayRef> {
120 Ok(Arc::new(BooleanArray::from(Vec::from(data))))
121 }
122
123 fn arrow_data_type() -> DataType {
124 DataType::Boolean
125 }
126}
127
128impl FromArrowArray for bool {
129 fn from_arrow_array(array: &ArrayRef) -> ArrowResult<Array1<Self>> {
130 let bool_array = array
131 .as_boolean_opt()
132 .ok_or_else(|| ArrowCompatError::TypeMismatch {
133 expected: "Boolean".to_string(),
134 actual: format!("{:?}", array.data_type()),
135 })?;
136
137 if bool_array.null_count() > 0 {
138 return Err(ArrowCompatError::NullValuesPresent {
139 null_count: bool_array.null_count(),
140 total_count: bool_array.len(),
141 });
142 }
143
144 let values: Vec<bool> = (0..bool_array.len()).map(|i| bool_array.value(i)).collect();
145 Ok(Array1::from_vec(values))
146 }
147
148 fn from_arrow_array_nullable(array: &ArrayRef) -> ArrowResult<Array1<Option<Self>>> {
149 let bool_array = array
150 .as_boolean_opt()
151 .ok_or_else(|| ArrowCompatError::TypeMismatch {
152 expected: "Boolean".to_string(),
153 actual: format!("{:?}", array.data_type()),
154 })?;
155
156 let values: Vec<Option<bool>> = (0..bool_array.len())
157 .map(|i| {
158 if bool_array.is_null(i) {
159 None
160 } else {
161 Some(bool_array.value(i))
162 }
163 })
164 .collect();
165 Ok(Array1::from_vec(values))
166 }
167}
168
169impl ToArrowArray for String {
174 fn to_arrow_array(data: &[Self]) -> ArrowResult<ArrayRef> {
175 let refs: Vec<&str> = data.iter().map(|s| s.as_str()).collect();
176 Ok(Arc::new(StringArray::from(refs)))
177 }
178
179 fn arrow_data_type() -> DataType {
180 DataType::Utf8
181 }
182}
183
184impl FromArrowArray for String {
185 fn from_arrow_array(array: &ArrayRef) -> ArrowResult<Array1<Self>> {
186 let string_array = array
187 .as_any()
188 .downcast_ref::<StringArray>()
189 .ok_or_else(|| ArrowCompatError::TypeMismatch {
190 expected: "Utf8 (String)".to_string(),
191 actual: format!("{:?}", array.data_type()),
192 })?;
193
194 if string_array.null_count() > 0 {
195 return Err(ArrowCompatError::NullValuesPresent {
196 null_count: string_array.null_count(),
197 total_count: string_array.len(),
198 });
199 }
200
201 let values: Vec<String> = (0..string_array.len())
202 .map(|i| string_array.value(i).to_string())
203 .collect();
204 Ok(Array1::from_vec(values))
205 }
206
207 fn from_arrow_array_nullable(array: &ArrayRef) -> ArrowResult<Array1<Option<Self>>> {
208 let string_array = array
209 .as_any()
210 .downcast_ref::<StringArray>()
211 .ok_or_else(|| ArrowCompatError::TypeMismatch {
212 expected: "Utf8 (String)".to_string(),
213 actual: format!("{:?}", array.data_type()),
214 })?;
215
216 let values: Vec<Option<String>> = (0..string_array.len())
217 .map(|i| {
218 if string_array.is_null(i) {
219 None
220 } else {
221 Some(string_array.value(i).to_string())
222 }
223 })
224 .collect();
225 Ok(Array1::from_vec(values))
226 }
227}
228
229pub fn array1_to_arrow<T>(array: &Array1<T>) -> ArrowResult<ArrayRef>
249where
250 T: ToArrowArray + Clone,
251{
252 let data: Vec<T> = array.iter().cloned().collect();
253 T::to_arrow_array(&data)
254}
255
256pub fn array1_to_arrow_zero_copy(array: &Array1<f64>) -> ArrowResult<ArrayRef> {
261 if let Some(slice) = array.as_slice() {
263 let buffer = Buffer::from_slice_ref(slice);
268 let arrow_array = Float64Array::new(buffer.into(), None);
269 Ok(Arc::new(arrow_array))
270 } else {
271 let data: Vec<f64> = array.iter().copied().collect();
273 Ok(Arc::new(Float64Array::from(data)))
274 }
275}
276
277pub fn arrow_to_array1<T>(array: &ArrayRef) -> ArrowResult<Array1<T>>
294where
295 T: FromArrowArray,
296{
297 T::from_arrow_array(array)
298}
299
300pub fn arrow_to_array1_nullable<T>(array: &ArrayRef) -> ArrowResult<Array1<Option<T>>>
304where
305 T: FromArrowArray,
306{
307 T::from_arrow_array_nullable(array)
308}
309
310pub fn array2_to_record_batch<T>(
337 array: &Array2<T>,
338 column_names: Option<&[&str]>,
339) -> ArrowResult<RecordBatch>
340where
341 T: ToArrowArray + Clone,
342{
343 let (nrows, ncols) = (array.nrows(), array.ncols());
344
345 if let Some(names) = column_names {
347 if names.len() != ncols {
348 return Err(ArrowCompatError::ShapeMismatch {
349 expected: vec![ncols],
350 actual: vec![names.len()],
351 });
352 }
353 }
354
355 let mut fields = Vec::with_capacity(ncols);
357 let mut arrays: Vec<ArrayRef> = Vec::with_capacity(ncols);
358
359 for col_idx in 0..ncols {
360 let col_name = column_names
361 .and_then(|names| names.get(col_idx).copied())
362 .unwrap_or_else(|| {
363 ""
367 });
368
369 let name = if col_name.is_empty() {
371 format!("col_{}", col_idx)
372 } else {
373 col_name.to_string()
374 };
375
376 let col_data: Vec<T> = array.column(col_idx).iter().cloned().collect();
378 let arrow_array = T::to_arrow_array(&col_data)?;
379
380 fields.push(Field::new(&name, T::arrow_data_type(), false));
381 arrays.push(arrow_array);
382 }
383
384 let schema = Arc::new(Schema::new(fields));
385 RecordBatch::try_new(schema, arrays).map_err(ArrowCompatError::from)
386}
387
388pub fn record_batch_to_array2<T>(batch: &RecordBatch) -> ArrowResult<Array2<T>>
408where
409 T: FromArrowArray + Clone + Default,
410{
411 let nrows = batch.num_rows();
412 let ncols = batch.num_columns();
413
414 if ncols == 0 {
415 return Err(ArrowCompatError::SchemaError(
416 "RecordBatch has no columns".to_string(),
417 ));
418 }
419
420 let mut data = Vec::with_capacity(nrows * ncols);
422
423 let columns: Vec<Array1<T>> = (0..ncols)
425 .map(|col_idx| T::from_arrow_array(batch.column(col_idx)))
426 .collect::<ArrowResult<Vec<_>>>()?;
427
428 for (col_idx, col) in columns.iter().enumerate() {
430 if col.len() != nrows {
431 return Err(ArrowCompatError::InconsistentColumnLengths {
432 expected_len: nrows,
433 column_index: col_idx,
434 column_len: col.len(),
435 });
436 }
437 }
438
439 for row_idx in 0..nrows {
441 for col in &columns {
442 data.push(col[row_idx].clone());
443 }
444 }
445
446 let data_len = data.len();
447 Array2::from_shape_vec((nrows, ncols), data).map_err(|_| ArrowCompatError::ShapeMismatch {
448 expected: vec![nrows, ncols],
449 actual: vec![data_len],
450 })
451}
452
453pub fn record_batch_column_to_array1<T>(
455 batch: &RecordBatch,
456 column_index: usize,
457) -> ArrowResult<Array1<T>>
458where
459 T: FromArrowArray,
460{
461 if column_index >= batch.num_columns() {
462 return Err(ArrowCompatError::ColumnOutOfBounds {
463 index: column_index,
464 num_columns: batch.num_columns(),
465 });
466 }
467
468 T::from_arrow_array(batch.column(column_index))
469}
470
471pub fn record_batch_column_by_name<T>(
473 batch: &RecordBatch,
474 column_name: &str,
475) -> ArrowResult<Array1<T>>
476where
477 T: FromArrowArray,
478{
479 let schema = batch.schema();
480 let col_idx = schema
481 .fields()
482 .iter()
483 .position(|f| f.name() == column_name)
484 .ok_or_else(|| ArrowCompatError::ColumnNotFound {
485 name: column_name.to_string(),
486 })?;
487
488 T::from_arrow_array(batch.column(col_idx))
489}
490
491pub fn nullable_array1_to_arrow<T>(array: &Array1<Option<T>>) -> ArrowResult<ArrayRef>
499where
500 T: NullableToArrow + Clone,
501{
502 let data: Vec<Option<T>> = array.iter().cloned().collect();
503 T::nullable_to_arrow(&data)
504}
505
506pub trait NullableToArrow: Sized {
508 fn nullable_to_arrow(data: &[Option<Self>]) -> ArrowResult<ArrayRef>;
510}
511
512macro_rules! impl_nullable_to_arrow {
513 ($rust_type:ty, $arrow_array_type:ty) => {
514 impl NullableToArrow for $rust_type {
515 fn nullable_to_arrow(data: &[Option<Self>]) -> ArrowResult<ArrayRef> {
516 let array: $arrow_array_type = data.iter().copied().collect();
517 Ok(Arc::new(array))
518 }
519 }
520 };
521}
522
523impl_nullable_to_arrow!(f64, Float64Array);
524impl_nullable_to_arrow!(f32, Float32Array);
525impl_nullable_to_arrow!(i64, Int64Array);
526impl_nullable_to_arrow!(i32, Int32Array);
527
528impl NullableToArrow for bool {
529 fn nullable_to_arrow(data: &[Option<Self>]) -> ArrowResult<ArrayRef> {
530 let array: BooleanArray = data.iter().copied().collect();
531 Ok(Arc::new(array))
532 }
533}
534
535impl NullableToArrow for String {
536 fn nullable_to_arrow(data: &[Option<Self>]) -> ArrowResult<ArrayRef> {
537 let refs: Vec<Option<&str>> = data.iter().map(|s| s.as_deref()).collect();
538 let array = StringArray::from(refs);
539 Ok(Arc::new(array))
540 }
541}
542
543#[cfg(test)]
544mod tests {
545 use super::*;
546
547 #[test]
552 fn test_array1_f64_roundtrip() {
553 let original = Array1::from_vec(vec![1.0_f64, 2.5, -1.23, 0.0, f64::MAX]);
554 let arrow = array1_to_arrow(&original).expect("to_arrow failed");
555 let recovered: Array1<f64> = arrow_to_array1(&arrow).expect("from_arrow failed");
556 assert_eq!(original, recovered);
557 }
558
559 #[test]
560 fn test_array1_f32_roundtrip() {
561 let original = Array1::from_vec(vec![1.0_f32, 2.5, -1.23, 0.0]);
562 let arrow = array1_to_arrow(&original).expect("to_arrow failed");
563 let recovered: Array1<f32> = arrow_to_array1(&arrow).expect("from_arrow failed");
564 assert_eq!(original, recovered);
565 }
566
567 #[test]
568 fn test_array1_i64_roundtrip() {
569 let original = Array1::from_vec(vec![1_i64, -100, i64::MAX, i64::MIN, 0]);
570 let arrow = array1_to_arrow(&original).expect("to_arrow failed");
571 let recovered: Array1<i64> = arrow_to_array1(&arrow).expect("from_arrow failed");
572 assert_eq!(original, recovered);
573 }
574
575 #[test]
576 fn test_array1_i32_roundtrip() {
577 let original = Array1::from_vec(vec![10_i32, 20, 30, -40]);
578 let arrow = array1_to_arrow(&original).expect("to_arrow failed");
579 let recovered: Array1<i32> = arrow_to_array1(&arrow).expect("from_arrow failed");
580 assert_eq!(original, recovered);
581 }
582
583 #[test]
584 fn test_array1_bool_roundtrip() {
585 let original = Array1::from_vec(vec![true, false, true, false, true]);
586 let arrow = array1_to_arrow(&original).expect("to_arrow failed");
587 let recovered: Array1<bool> = arrow_to_array1(&arrow).expect("from_arrow failed");
588 assert_eq!(original, recovered);
589 }
590
591 #[test]
592 fn test_array1_string_roundtrip() {
593 let original = Array1::from_vec(vec![
594 "hello".to_string(),
595 "world".to_string(),
596 "".to_string(),
597 "test 123".to_string(),
598 ]);
599 let arrow = array1_to_arrow(&original).expect("to_arrow failed");
600 let recovered: Array1<String> = arrow_to_array1(&arrow).expect("from_arrow failed");
601 assert_eq!(original, recovered);
602 }
603
604 #[test]
609 fn test_zero_copy_f64() {
610 let original = Array1::from_vec(vec![1.0_f64, 2.0, 3.0, 4.0]);
611 let arrow = array1_to_arrow_zero_copy(&original).expect("zero_copy to_arrow failed");
612
613 let recovered: Array1<f64> = arrow_to_array1(&arrow).expect("from_arrow failed");
615 assert_eq!(original, recovered);
616 }
617
618 #[test]
619 fn test_zero_copy_view_f64() {
620 let arrow_arr: ArrayRef = Arc::new(Float64Array::from(vec![10.0, 20.0, 30.0]));
621 let view = f64::try_zero_copy_view(&arrow_arr).expect("zero_copy_view failed");
622 assert!(view.is_some());
623 let view = view.expect("should have view");
624 assert_eq!(view.len(), 3);
625 assert!((view[0] - 10.0).abs() < f64::EPSILON);
626 assert!((view[1] - 20.0).abs() < f64::EPSILON);
627 assert!((view[2] - 30.0).abs() < f64::EPSILON);
628 }
629
630 #[test]
635 fn test_nullable_f64() {
636 let data = Array1::from_vec(vec![Some(1.0_f64), None, Some(3.0), None, Some(5.0)]);
637 let arrow = nullable_array1_to_arrow(&data).expect("nullable to_arrow failed");
638 let recovered: Array1<Option<f64>> =
639 arrow_to_array1_nullable(&arrow).expect("nullable from_arrow failed");
640 assert_eq!(data, recovered);
641 }
642
643 #[test]
644 fn test_nullable_i32() {
645 let data = Array1::from_vec(vec![Some(10_i32), None, Some(30)]);
646 let arrow = nullable_array1_to_arrow(&data).expect("nullable to_arrow failed");
647 let recovered: Array1<Option<i32>> =
648 arrow_to_array1_nullable(&arrow).expect("nullable from_arrow failed");
649 assert_eq!(data, recovered);
650 }
651
652 #[test]
653 fn test_nullable_bool() {
654 let data = Array1::from_vec(vec![Some(true), None, Some(false)]);
655 let arrow = nullable_array1_to_arrow(&data).expect("nullable to_arrow failed");
656 let recovered: Array1<Option<bool>> =
657 arrow_to_array1_nullable(&arrow).expect("nullable from_arrow failed");
658 assert_eq!(data, recovered);
659 }
660
661 #[test]
662 fn test_nullable_string() {
663 let data = Array1::from_vec(vec![
664 Some("hello".to_string()),
665 None,
666 Some("world".to_string()),
667 ]);
668 let arrow = nullable_array1_to_arrow(&data).expect("nullable to_arrow failed");
669 let recovered: Array1<Option<String>> =
670 arrow_to_array1_nullable(&arrow).expect("nullable from_arrow failed");
671 assert_eq!(data, recovered);
672 }
673
674 #[test]
675 fn test_null_values_rejected_by_non_nullable() {
676 let arrow_arr: ArrayRef = Arc::new(Float64Array::from(vec![Some(1.0), None, Some(3.0)]));
677 let result: ArrowResult<Array1<f64>> = arrow_to_array1(&arrow_arr);
678 assert!(result.is_err());
679 }
680
681 #[test]
686 fn test_array2_f64_to_record_batch() {
687 let arr = Array2::from_shape_vec((3, 2), vec![1.0, 2.0, 3.0, 4.0, 5.0, 6.0])
688 .expect("shape error");
689 let batch = array2_to_record_batch(&arr, None).expect("to_batch failed");
690
691 assert_eq!(batch.num_rows(), 3);
692 assert_eq!(batch.num_columns(), 2);
693 assert_eq!(batch.schema().field(0).name(), "col_0");
694 assert_eq!(batch.schema().field(1).name(), "col_1");
695 }
696
697 #[test]
698 fn test_array2_f64_roundtrip() {
699 let arr = Array2::from_shape_vec(
700 (4, 3),
701 vec![
702 1.0, 2.0, 3.0, 4.0, 5.0, 6.0, 7.0, 8.0, 9.0, 10.0, 11.0, 12.0,
703 ],
704 )
705 .expect("shape error");
706 let batch = array2_to_record_batch(&arr, None).expect("to_batch failed");
707 let recovered: Array2<f64> = record_batch_to_array2(&batch).expect("from_batch failed");
708 assert_eq!(arr, recovered);
709 }
710
711 #[test]
712 fn test_array2_with_custom_column_names() {
713 let arr = Array2::from_shape_vec((2, 3), vec![1.0, 2.0, 3.0, 4.0, 5.0, 6.0])
714 .expect("shape error");
715 let names = vec!["x", "y", "z"];
716 let batch = array2_to_record_batch(&arr, Some(&names)).expect("to_batch failed");
717
718 assert_eq!(batch.schema().field(0).name(), "x");
719 assert_eq!(batch.schema().field(1).name(), "y");
720 assert_eq!(batch.schema().field(2).name(), "z");
721 }
722
723 #[test]
724 fn test_record_batch_column_by_name() {
725 let arr = Array2::from_shape_vec((3, 2), vec![1.0, 10.0, 2.0, 20.0, 3.0, 30.0])
726 .expect("shape error");
727 let names = vec!["values", "scores"];
728 let batch = array2_to_record_batch(&arr, Some(&names)).expect("to_batch failed");
729
730 let values: Array1<f64> =
731 record_batch_column_by_name(&batch, "values").expect("column lookup failed");
732 assert_eq!(values, Array1::from_vec(vec![1.0, 2.0, 3.0]));
733
734 let scores: Array1<f64> =
735 record_batch_column_by_name(&batch, "scores").expect("column lookup failed");
736 assert_eq!(scores, Array1::from_vec(vec![10.0, 20.0, 30.0]));
737 }
738
739 #[test]
740 fn test_record_batch_column_not_found() {
741 let arr = Array2::from_shape_vec((2, 1), vec![1.0, 2.0]).expect("shape error");
742 let batch = array2_to_record_batch(&arr, None).expect("to_batch failed");
743
744 let result: ArrowResult<Array1<f64>> = record_batch_column_by_name(&batch, "nonexistent");
745 assert!(result.is_err());
746 }
747
748 #[test]
749 fn test_column_out_of_bounds() {
750 let arr = Array2::from_shape_vec((2, 1), vec![1.0, 2.0]).expect("shape error");
751 let batch = array2_to_record_batch(&arr, None).expect("to_batch failed");
752
753 let result: ArrowResult<Array1<f64>> = record_batch_column_to_array1(&batch, 5);
754 assert!(result.is_err());
755 }
756
757 #[test]
762 fn test_type_mismatch_error() {
763 let arrow_arr: ArrayRef = Arc::new(Float64Array::from(vec![1.0, 2.0]));
764 let result: ArrowResult<Array1<i32>> = arrow_to_array1(&arrow_arr);
765 assert!(result.is_err());
766 }
767
768 #[test]
773 fn test_empty_array() {
774 let original: Array1<f64> = Array1::from_vec(vec![]);
775 let arrow = array1_to_arrow(&original).expect("to_arrow failed");
776 let recovered: Array1<f64> = arrow_to_array1(&arrow).expect("from_arrow failed");
777 assert_eq!(original, recovered);
778 assert_eq!(recovered.len(), 0);
779 }
780
781 #[test]
782 fn test_single_element() {
783 let original = Array1::from_vec(vec![42.0_f64]);
784 let arrow = array1_to_arrow(&original).expect("to_arrow failed");
785 let recovered: Array1<f64> = arrow_to_array1(&arrow).expect("from_arrow failed");
786 assert_eq!(original, recovered);
787 }
788}