chore: Helper function to convert Vec<Value> to VectorRef

src/datatypes/src/duration.rs

@@ -111,6 +111,24 @@ macro_rules! define_duration_with_unit {
                     val.0.value()
                 }
             }
+
+            impl TryFrom<Value> for Option<[<Duration $unit>]> {
+                type Error = $crate::error::Error;
+
+                #[inline]
+                fn try_from(from: Value) -> std::result::Result<Self, Self::Error> {
+                    match from {
+                        Value::Duration(v) if v.unit() == TimeUnit::$unit => {
+                            Ok(Some([<Duration $unit>](v)))
+                        },
+                        Value::Null => Ok(None),
+                        _ => $crate::error::TryFromValueSnafu {
+                            reason: format!("{:?} is not a {}", from, stringify!([<Duration $unit>])),
+                        }
+                        .fail(),
+                    }
+                }
+            }
         }
     };
 }

src/datatypes/src/interval.rs

@@ -106,6 +106,24 @@ macro_rules! define_interval_with_unit {
                     val.0.[<to_ $native_ty>]()
                 }
             }
+
+            impl TryFrom<Value> for Option<[<Interval $unit>]> {
+                type Error = $crate::error::Error;
+
+                #[inline]
+                fn try_from(from: Value) -> std::result::Result<Self, Self::Error> {
+                    match from {
+                        Value::Interval(v) if v.unit() == common_time::interval::IntervalUnit::$unit => {
+                            Ok(Some([<Interval $unit>](v)))
+                        },
+                        Value::Null => Ok(None),
+                        _ => $crate::error::TryFromValueSnafu {
+                            reason: format!("{:?} is not a {}", from, stringify!([<Interval $unit>])),
+                        }
+                        .fail(),
+                    }
+                }
+            }
         }
     };
 }

src/datatypes/src/time.rs

@@ -109,6 +109,24 @@ macro_rules! define_time_with_unit {
                     val.0.value()
                 }
             }
+
+            impl TryFrom<Value> for Option<[<Time $unit>]> {
+                type Error = $crate::error::Error;
+
+                #[inline]
+                fn try_from(from: Value) -> std::result::Result<Self, Self::Error> {
+                    match from {
+                        Value::Time(v) if *v.unit() == TimeUnit::$unit => {
+                            Ok(Some([<Time $unit>](v)))
+                        },
+                        Value::Null => Ok(None),
+                        _ => $crate::error::TryFromValueSnafu {
+                            reason: format!("{:?} is not a {}", from, stringify!([<Time $unit>])),
+                        }
+                        .fail(),
+                    }
+                }
+            }
         }
     };
 }

src/datatypes/src/timestamp.rs

@@ -111,6 +111,24 @@ macro_rules! define_timestamp_with_unit {
                     val.0.value()
                 }
             }
+
+            impl TryFrom<Value> for Option<[<Timestamp $unit>]> {
+                type Error = $crate::error::Error;
+
+                #[inline]
+                fn try_from(from: Value) -> std::result::Result<Self, Self::Error> {
+                    match from {
+                        Value::Timestamp(v) if v.unit() == TimeUnit::$unit => {
+                            Ok(Some([<Timestamp $unit>](v)))
+                        },
+                        Value::Null => Ok(None),
+                        _ => $crate::error::TryFromValueSnafu {
+                            reason: format!("{:?} is not a {}", from, stringify!([<Timestamp $unit>])),
+                        }
+                        .fail(),
+                    }
+                }
+            }
         }
     };
 }

src/datatypes/src/vectors/helper.rs

@@ -22,22 +22,26 @@ use arrow::compute;
 use arrow::compute::kernels::comparison;
 use arrow::datatypes::{DataType as ArrowDataType, TimeUnit};
 use arrow_schema::IntervalUnit;
+use common_base::bytes::StringBytes;
 use datafusion_common::ScalarValue;
 use snafu::{OptionExt, ResultExt};
 
 use crate::data_type::ConcreteDataType;
-use crate::error::{self, ConvertArrowArrayToScalarsSnafu, Result};
+use crate::error::{self, ConvertArrowArrayToScalarsSnafu, Result, UnsupportedArrowTypeSnafu};
+use crate::prelude::DataType;
 use crate::scalars::{Scalar, ScalarVectorBuilder};
+use crate::types::{LogicalPrimitiveType, WrapperType};
 use crate::value::{ListValue, ListValueRef, Value};
 use crate::vectors::{
     BinaryVector, BooleanVector, ConstantVector, DateTimeVector, DateVector, Decimal128Vector,
     DurationMicrosecondVector, DurationMillisecondVector, DurationNanosecondVector,
     DurationSecondVector, Float32Vector, Float64Vector, Int16Vector, Int32Vector, Int64Vector,
     Int8Vector, IntervalDayTimeVector, IntervalMonthDayNanoVector, IntervalYearMonthVector,
-    ListVector, ListVectorBuilder, MutableVector, NullVector, StringVector, TimeMicrosecondVector,
-    TimeMillisecondVector, TimeNanosecondVector, TimeSecondVector, TimestampMicrosecondVector,
-    TimestampMillisecondVector, TimestampNanosecondVector, TimestampSecondVector, UInt16Vector,
-    UInt32Vector, UInt64Vector, UInt8Vector, Vector, VectorRef,
+    ListVector, ListVectorBuilder, MutableVector, NullVector, PrimitiveVector, StringVector,
+    TimeMicrosecondVector, TimeMillisecondVector, TimeNanosecondVector, TimeSecondVector,
+    TimestampMicrosecondVector, TimestampMillisecondVector, TimestampNanosecondVector,
+    TimestampSecondVector, UInt16Vector, UInt32Vector, UInt64Vector, UInt8Vector, Vector,
+    VectorRef,
 };
 
 /// Helper functions for `Vector`.
@@ -367,6 +371,185 @@ impl Helper {
         })
     }
 
+    /// Try to cast an vec of values into vector, fail if type is not the same across all values.
+    pub fn try_from_row_into_vector(row: Vec<Value>, dt: &ConcreteDataType) -> Result<VectorRef> {
+        use common_time::timestamp::TimeUnit;
+        use ConcreteDataType as CDT;
+
+        use crate::types::{
+            DateTimeType, DateType, DurationMicrosecondType, DurationMillisecondType,
+            DurationNanosecondType, DurationSecondType, Float32Type, Float64Type, Int16Type,
+            Int32Type, Int64Type, Int8Type, IntervalDayTimeType, IntervalMonthDayNanoType,
+            IntervalYearMonthType, TimeMicrosecondType, TimeMillisecondType, TimeNanosecondType,
+            TimeSecondType, TimestampMicrosecondType, TimestampMillisecondType,
+            TimestampNanosecondType, TimestampSecondType, UInt16Type, UInt32Type, UInt64Type,
+            UInt8Type,
+        };
+        fn primitive_values_to_vector<T, Native, Wrapped>(values: Vec<Value>) -> Result<VectorRef>
+        where
+            T: LogicalPrimitiveType<Native = Native>,
+            Wrapped: WrapperType<Native = Native, LogicalType = T>,
+            Option<Wrapped>: TryFrom<Value, Error = error::Error>,
+        {
+            let mut native_values: Vec<Option<Native>> = Vec::new();
+            for value in values.into_iter() {
+                let native: Option<Native> =
+                    Option::<Wrapped>::try_from(value)?.map(|inner| inner.into_native());
+                native_values.push(native);
+            }
+            let vector = PrimitiveVector::<T>::from(native_values);
+            Ok(Arc::new(vector))
+        }
+
+        match dt {
+            CDT::Null(_) => {
+                let vector = NullVector::new(row.len());
+                Ok(Arc::new(vector))
+            }
+            CDT::Boolean(_) => {
+                let values: Vec<Option<_>> = row
+                    .into_iter()
+                    .map(|v| v.try_into())
+                    .collect::<Result<_>>()?;
+                let vector = BooleanVector::from(values);
+                Ok(Arc::new(vector))
+            }
+            CDT::Int8(_) => primitive_values_to_vector::<Int8Type, i8, i8>(row),
+            CDT::Int16(_) => primitive_values_to_vector::<Int16Type, i16, i16>(row),
+            CDT::Int32(_) => primitive_values_to_vector::<Int32Type, i32, i32>(row),
+            CDT::Int64(_) => primitive_values_to_vector::<Int64Type, i64, i64>(row),
+            CDT::UInt8(_) => primitive_values_to_vector::<UInt8Type, u8, u8>(row),
+            CDT::UInt16(_) => primitive_values_to_vector::<UInt16Type, u16, u16>(row),
+            CDT::UInt32(_) => primitive_values_to_vector::<UInt32Type, u32, u32>(row),
+            CDT::UInt64(_) => primitive_values_to_vector::<UInt64Type, u64, u64>(row),
+            CDT::Float32(_) => primitive_values_to_vector::<Float32Type, f32, f32>(row),
+            CDT::Float64(_) => primitive_values_to_vector::<Float64Type, f64, f64>(row),
+            CDT::String(_) => {
+                let values: Vec<Option<String>> = row
+                    .into_iter()
+                    .map(|v| {
+                        Option::<StringBytes>::try_from(v)
+                            .map(|v| v.map(|inner| inner.into_string()))
+                    })
+                    .collect::<Result<_>>()?;
+                let vector = StringVector::from(values);
+                Ok(Arc::new(vector))
+            }
+            CDT::Binary(_) => {
+                let values: Vec<Option<Vec<u8>>> = row
+                    .into_iter()
+                    .map(|v| {
+                        Option::<common_base::bytes::Bytes>::try_from(v)
+                            .map(|v| v.map(|inner| inner.to_vec()))
+                    })
+                    .collect::<Result<_>>()?;
+                let vector = BinaryVector::from(values);
+                Ok(Arc::new(vector))
+            }
+            CDT::Date(_) => primitive_values_to_vector::<DateType, i32, common_time::Date>(row),
+            CDT::DateTime(_) => {
+                primitive_values_to_vector::<DateTimeType, i64, common_time::DateTime>(row)
+            }
+            CDT::Time(time_type) => match time_type.unit() {
+                TimeUnit::Second => {
+                    primitive_values_to_vector::<TimeSecondType, i32, crate::time::TimeSecond>(row)
+                }
+                TimeUnit::Millisecond => primitive_values_to_vector::<
+                    TimeMillisecondType,
+                    i32,
+                    crate::time::TimeMillisecond,
+                >(row),
+                TimeUnit::Microsecond => primitive_values_to_vector::<
+                    TimeMicrosecondType,
+                    i64,
+                    crate::time::TimeMicrosecond,
+                >(row),
+                TimeUnit::Nanosecond => {
+                    primitive_values_to_vector::<TimeNanosecondType, i64, crate::time::TimeNanosecond>(
+                        row,
+                    )
+                }
+            },
+            CDT::Timestamp(timestamp_type) => match timestamp_type.unit() {
+                TimeUnit::Second => primitive_values_to_vector::<
+                    TimestampSecondType,
+                    i64,
+                    crate::timestamp::TimestampSecond,
+                >(row),
+                TimeUnit::Millisecond => primitive_values_to_vector::<
+                    TimestampMillisecondType,
+                    i64,
+                    crate::timestamp::TimestampMillisecond,
+                >(row),
+                TimeUnit::Microsecond => primitive_values_to_vector::<
+                    TimestampMicrosecondType,
+                    i64,
+                    crate::timestamp::TimestampMicrosecond,
+                >(row),
+                TimeUnit::Nanosecond => primitive_values_to_vector::<
+                    TimestampNanosecondType,
+                    i64,
+                    crate::timestamp::TimestampNanosecond,
+                >(row),
+            },
+            CDT::Interval(interval_type) => match interval_type.unit() {
+                common_time::interval::IntervalUnit::YearMonth => primitive_values_to_vector::<
+                    IntervalYearMonthType,
+                    i32,
+                    crate::interval::IntervalYearMonth,
+                >(row),
+                common_time::interval::IntervalUnit::DayTime => primitive_values_to_vector::<
+                    IntervalDayTimeType,
+                    i64,
+                    crate::interval::IntervalDayTime,
+                >(row),
+                common_time::interval::IntervalUnit::MonthDayNano => primitive_values_to_vector::<
+                    IntervalMonthDayNanoType,
+                    i128,
+                    crate::interval::IntervalMonthDayNano,
+                >(row),
+            },
+            CDT::Duration(duration_type) => match duration_type.unit() {
+                TimeUnit::Second => primitive_values_to_vector::<
+                    DurationSecondType,
+                    i64,
+                    crate::duration::DurationSecond,
+                >(row),
+                TimeUnit::Millisecond => primitive_values_to_vector::<
+                    DurationMillisecondType,
+                    i64,
+                    crate::duration::DurationMillisecond,
+                >(row),
+                TimeUnit::Microsecond => primitive_values_to_vector::<
+                    DurationMicrosecondType,
+                    i64,
+                    crate::duration::DurationMicrosecond,
+                >(row),
+                TimeUnit::Nanosecond => primitive_values_to_vector::<
+                    DurationNanosecondType,
+                    i64,
+                    crate::duration::DurationNanosecond,
+                >(row),
+            },
+            CDT::Decimal128(_) => {
+                let values: Vec<_> = row
+                    .into_iter()
+                    .map(|v| {
+                        Option::<common_decimal::Decimal128>::try_from(v)
+                            .map(|v| v.map(|inner| inner.val()))
+                    })
+                    .collect::<Result<_>>()?;
+                let array: arrow_array::Decimal128Array = values.into_iter().collect();
+                let vector = Decimal128Vector::from(array);
+                Ok(Arc::new(vector))
+            }
+            _ => UnsupportedArrowTypeSnafu {
+                arrow_type: dt.as_arrow_type(),
+            }
+            .fail()?,
+        }
+    }
+
     /// Try to cast slice of `arrays` to vectors.
     pub fn try_into_vectors(arrays: &[ArrayRef]) -> Result<Vec<VectorRef>> {
         arrays.iter().map(Self::try_into_vector).collect()
@@ -681,4 +864,48 @@ mod tests {
             assert_eq!(Value::Interval(Interval::from_i128(2000)), vector.get(i));
         }
     }
+
+    fn check_try_from_row_to_vector(row: Vec<Value>, dt: &ConcreteDataType) {
+        let vector = Helper::try_from_row_into_vector(row.clone(), dt).unwrap();
+        for (i, item) in row.iter().enumerate().take(vector.len()) {
+            assert_eq!(*item, vector.get(i));
+        }
+    }
+
+    fn check_into_and_from(array: impl Array + 'static) {
+        let array: ArrayRef = Arc::new(array);
+        let vector = Helper::try_into_vector(array.clone()).unwrap();
+        assert_eq!(&array, &vector.to_arrow_array());
+        let row: Vec<Value> = (0..array.len()).map(|i| vector.get(i)).collect();
+        let dt = vector.data_type();
+        check_try_from_row_to_vector(row, &dt);
+    }
+
+    #[test]
+    fn test_try_from_row_to_vector() {
+        check_into_and_from(NullArray::new(2));
+        check_into_and_from(BooleanArray::from(vec![true, false]));
+        check_into_and_from(Int8Array::from(vec![1, 2, 3]));
+        check_into_and_from(Int16Array::from(vec![1, 2, 3]));
+        check_into_and_from(Int32Array::from(vec![1, 2, 3]));
+        check_into_and_from(Int64Array::from(vec![1, 2, 3]));
+        check_into_and_from(UInt8Array::from(vec![1, 2, 3]));
+        check_into_and_from(UInt16Array::from(vec![1, 2, 3]));
+        check_into_and_from(UInt32Array::from(vec![1, 2, 3]));
+        check_into_and_from(UInt64Array::from(vec![1, 2, 3]));
+        check_into_and_from(Float32Array::from(vec![1.0, 2.0, 3.0]));
+        check_into_and_from(Float64Array::from(vec![1.0, 2.0, 3.0]));
+        check_into_and_from(StringArray::from(vec!["hello", "world"]));
+        check_into_and_from(Date32Array::from(vec![1, 2, 3]));
+        check_into_and_from(Date64Array::from(vec![1, 2, 3]));
+
+        check_into_and_from(TimestampSecondArray::from(vec![1, 2, 3]));
+        check_into_and_from(TimestampMillisecondArray::from(vec![1, 2, 3]));
+        check_into_and_from(TimestampMicrosecondArray::from(vec![1, 2, 3]));
+        check_into_and_from(TimestampNanosecondArray::from(vec![1, 2, 3]));
+        check_into_and_from(Time32SecondArray::from(vec![1, 2, 3]));
+        check_into_and_from(Time32MillisecondArray::from(vec![1, 2, 3]));
+        check_into_and_from(Time64MicrosecondArray::from(vec![1, 2, 3]));
+        check_into_and_from(Time64NanosecondArray::from(vec![1, 2, 3]));
+    }
 }