fix: Fix tests and clippy for common-function subcrate (#726)

* further fixing

Signed-off-by: Ruihang Xia <waynestxia@gmail.com>

* fix all compile errors in common function

Signed-off-by: Ruihang Xia <waynestxia@gmail.com>

* fix tests

Signed-off-by: Ruihang Xia <waynestxia@gmail.com>

* fix clippy

Signed-off-by: Ruihang Xia <waynestxia@gmail.com>

* revert test changes

Signed-off-by: Ruihang Xia <waynestxia@gmail.com>

Signed-off-by: Ruihang Xia <waynestxia@gmail.com>

src/common/function/src/scalars/numpy/clip.rs

@@ -158,11 +158,15 @@ impl fmt::Display for ClipFunction {
 mod tests {
     use common_query::prelude::TypeSignature;
     use datatypes::value::Value;
-    use datatypes::vectors::{ConstantVector, Float32Vector, Int32Vector, UInt32Vector};
+    use datatypes::vectors::{
+        ConstantVector, Float32Vector, Int16Vector, Int32Vector, Int8Vector, UInt16Vector,
+        UInt32Vector, UInt8Vector,
+    };
 
     use super::*;
+
     #[test]
-    fn test_clip_function() {
+    fn test_clip_signature() {
         let clip = ClipFunction::default();
 
         assert_eq!("clip", clip.name());
@@ -205,16 +209,21 @@ mod tests {
                              volatility: Volatility::Immutable
                          } if  valid_types == ConcreteDataType::numerics()
         ));
+    }
+
+    #[test]
+    fn test_clip_fn_signed() {
+        let clip = ClipFunction::default();
 
         // eval with signed integers
         let args: Vec<VectorRef> = vec![
             Arc::new(Int32Vector::from_values(0..10)),
             Arc::new(ConstantVector::new(
-                Arc::new(Int32Vector::from_vec(vec![3])),
+                Arc::new(Int8Vector::from_vec(vec![3])),
                 10,
             )),
             Arc::new(ConstantVector::new(
-                Arc::new(Int32Vector::from_vec(vec![6])),
+                Arc::new(Int16Vector::from_vec(vec![6])),
                 10,
             )),
         ];
@@ -232,16 +241,21 @@ mod tests {
                 assert!(matches!(vector.get(i), Value::Int64(v) if v == 6));
             }
         }
+    }
+
+    #[test]
+    fn test_clip_fn_unsigned() {
+        let clip = ClipFunction::default();
 
         // eval with unsigned integers
         let args: Vec<VectorRef> = vec![
-            Arc::new(UInt32Vector::from_values(0..10)),
+            Arc::new(UInt8Vector::from_values(0..10)),
             Arc::new(ConstantVector::new(
                 Arc::new(UInt32Vector::from_vec(vec![3])),
                 10,
             )),
             Arc::new(ConstantVector::new(
-                Arc::new(UInt32Vector::from_vec(vec![6])),
+                Arc::new(UInt16Vector::from_vec(vec![6])),
                 10,
             )),
         ];
@@ -259,12 +273,17 @@ mod tests {
                 assert!(matches!(vector.get(i), Value::UInt64(v) if v == 6));
             }
         }
+    }
+
+    #[test]
+    fn test_clip_fn_float() {
+        let clip = ClipFunction::default();
 
         // eval with floats
         let args: Vec<VectorRef> = vec![
-            Arc::new(Int32Vector::from_values(0..10)),
+            Arc::new(Int8Vector::from_values(0..10)),
             Arc::new(ConstantVector::new(
-                Arc::new(Int32Vector::from_vec(vec![3])),
+                Arc::new(UInt32Vector::from_vec(vec![3])),
                 10,
             )),
             Arc::new(ConstantVector::new(

src/common/function/src/scalars/timestamp/from_unixtime.rs

@@ -18,11 +18,12 @@ use std::fmt;
 use std::sync::Arc;
 
 use common_query::error::{
-    ArrowComputeSnafu, IntoVectorSnafu, Result, UnsupportedInputDataTypeSnafu,
+    ArrowComputeSnafu, IntoVectorSnafu, Result, TypeCastSnafu, UnsupportedInputDataTypeSnafu,
 };
 use common_query::prelude::{Signature, Volatility};
 use datatypes::arrow::compute;
 use datatypes::arrow::datatypes::{DataType as ArrowDatatype, Int64Type};
+use datatypes::data_type::DataType;
 use datatypes::prelude::ConcreteDataType;
 use datatypes::vectors::{TimestampMillisecondVector, VectorRef};
 use snafu::ResultExt;
@@ -59,20 +60,23 @@ impl Function for FromUnixtimeFunction {
                 let array = compute::multiply_scalar_dyn::<Int64Type>(&array, 1000i64)
                     .context(ArrowComputeSnafu)?;
 
+                let arrow_datatype = &self.return_type(&[]).unwrap().as_arrow_type();
                 Ok(Arc::new(
-                    TimestampMillisecondVector::try_from_arrow_array(array).context(
-                        IntoVectorSnafu {
-                            data_type: ArrowDatatype::Int64,
-                        },
-                    )?,
+                    TimestampMillisecondVector::try_from_arrow_array(
+                        compute::cast(&array, arrow_datatype).context(TypeCastSnafu {
+                            typ: ArrowDatatype::Int64,
+                        })?,
+                    )
+                    .context(IntoVectorSnafu {
+                        data_type: arrow_datatype.clone(),
+                    })?,
                 ))
             }
             _ => UnsupportedInputDataTypeSnafu {
                 function: NAME,
                 datatypes: columns.iter().map(|c| c.data_type()).collect::<Vec<_>>(),
             }
-            .fail()
-            .map_err(|e| e.into()),
+            .fail(),
         }
     }
 }

src/common/query/src/error.rs

@@ -214,7 +214,7 @@ impl From<Error> for DataFusionError {
 
 impl From<BoxedError> for Error {
     fn from(source: BoxedError) -> Self {
-        Error::ExecutePhysicalPlan { source }.into()
+        Error::ExecutePhysicalPlan { source }
     }
 }
 

src/common/query/src/logical_plan/expr.rs

@@ -16,7 +16,7 @@ pub use datafusion_expr::expr::Expr as DfExpr;
 
 /// Central struct of query API.
 /// Represent logical expressions such as `A + 1`, or `CAST(c1 AS int)`.
-#[derive(Clone, PartialEq, Hash, Debug)]
+#[derive(Clone, PartialEq, Eq, Hash, Debug)]
 pub struct Expr {
     df_expr: DfExpr,
 }

src/common/recordbatch/src/adapter.rs

@@ -168,8 +168,7 @@ impl Stream for AsyncRecordBatchStreamAdapter {
                             error::CreateRecordBatchesSnafu {
                                 reason: format!("Read error {:?} from stream", e),
                             }
-                            .fail()
-                            .map_err(|e| e.into()),
+                            .fail(),
                         ))
                     }
                 },

src/common/recordbatch/src/lib.rs

@@ -98,7 +98,7 @@ impl RecordBatches {
             .iter()
             .map(|x| x.df_recordbatch.clone())
             .collect::<Vec<_>>();
-        let result = pretty::pretty_format_batches(&df_batches).context(error::FormatSnafu)?;
+        let result = pretty::pretty_format_batches(df_batches).context(error::FormatSnafu)?;
 
         Ok(result.to_string())
     }

src/common/recordbatch/src/recordbatch.rs

@@ -112,7 +112,7 @@ impl<'a> Iterator for RecordBatchRowIterator<'a> {
                     .context(error::DataTypesSnafu)
                 {
                     Ok(field) => row.push(field),
-                    Err(e) => return Some(Err(e.into())),
+                    Err(e) => return Some(Err(e)),
                 }
             }
 

src/common/time/src/date.rs

@@ -56,7 +56,7 @@ impl Display for Date {
     /// [Date] is formatted according to ISO-8601 standard.
     fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
         if let Some(abs_date) = NaiveDate::from_num_days_from_ce_opt(UNIX_EPOCH_FROM_CE + self.0) {
-            write!(f, "{}", abs_date.format("%F").to_string())
+            write!(f, "{}", abs_date.format("%F"))
         } else {
             write!(f, "Date({})", self.0)
         }