refactor: func choose api&use Error not EvalError

src/flow/Cargo.toml

@@ -14,6 +14,7 @@ common-error.workspace = true
 common-macro.workspace = true
 common-telemetry.workspace = true
 common-time.workspace = true
+datafusion-substrait.workspace = true
 datatypes.workspace = true
 enum_dispatch = "0.3"
 hydroflow = "0.5.0"
@@ -23,8 +24,20 @@ serde.workspace = true
 servers.workspace = true
 smallvec.workspace = true
 snafu.workspace = true
+strum.workspace = true
 tokio.workspace = true
 tonic.workspace = true
 
+[dependencies.substrait_proto]
+package = "substrait"
+version = "0.17"
+
 [dev-dependencies]
+catalog.workspace = true
+common-catalog.workspace = true
+prost.workspace = true
+query.workspace = true
 serde_json = "1.0"
+session.workspace = true
+substrait.workspace = true
+table.workspace = true

src/flow/src/adapter/error.rs

@@ -54,6 +54,22 @@ pub enum Error {
 
     #[snafu(display("No protobuf type for value: {value}"))]
     NoProtoType { value: Value, location: Location },
+
+    #[snafu(display("Not implement in flow: {reason}"))]
+    NotImplemented { reason: String, location: Location },
+
+    #[snafu(display("Flow plan error: {reason}"))]
+    Plan { reason: String, location: Location },
+
+    #[snafu(display("Unsupported temporal filter: {reason}"))]
+    UnsupportedTemporalFilter { reason: String, location: Location },
+
+    #[snafu(display("Datatypes error: {source} with extra message: {extra}"))]
+    Datatypes {
+        source: datatypes::Error,
+        extra: String,
+        location: Location,
+    },
 }
 
 pub type Result<T> = std::result::Result<T, Error>;
@@ -64,8 +80,13 @@ impl ErrorExt for Error {
             Self::Eval { .. } | &Self::JoinTask { .. } => StatusCode::Internal,
             &Self::TableAlreadyExist { .. } => StatusCode::TableAlreadyExists,
             Self::TableNotFound { .. } => StatusCode::TableNotFound,
-            &Self::InvalidQuery { .. } => StatusCode::PlanQuery,
+            &Self::InvalidQuery { .. } | &Self::Plan { .. } | &Self::Datatypes { .. } => {
+                StatusCode::PlanQuery
+            }
             Self::NoProtoType { .. } => StatusCode::Unexpected,
+            &Self::NotImplemented { .. } | Self::UnsupportedTemporalFilter { .. } => {
+                StatusCode::Unsupported
+            }
         }
     }
 

src/flow/src/expr.rs

@@ -20,6 +20,7 @@ mod id;
 mod linear;
 mod relation;
 mod scalar;
+mod signature;
 
 pub(crate) use error::{EvalError, InvalidArgumentSnafu, OptimizeSnafu};
 pub(crate) use func::{BinaryFunc, UnaryFunc, UnmaterializableFunc, VariadicFunc};

src/flow/src/expr/error.rs

@@ -59,9 +59,6 @@ pub enum EvalError {
     #[snafu(display("Optimize error: {reason}"))]
     Optimize { reason: String, location: Location },
 
-    #[snafu(display("Unsupported temporal filter: {reason}"))]
-    UnsupportedTemporalFilter { reason: String, location: Location },
-
     #[snafu(display("Overflowed during evaluation"))]
     Overflow { location: Location },
 }