feat(flow): accumlator for aggr func (#3396)

* feat: Accumlator trait

* feat: add `OrdValue` accum&use enum_dispatch

* test: more accum test

* feat: eval aggr funcs

* chore: refactor test&fmt clippy

* refactor: less verbose

* test: more tests

* refactor: better err handling&use OrdValue for Count

* refactor: ignore null&more tests for error handle

* refactor: OrdValue accum

* chore: extract null check

* refactor: def&use fn signature

* chore: use extra cond with match guard

* chore: per review

src/flow/Cargo.toml

@@ -18,6 +18,7 @@ common-query.workspace = true
 common-telemetry.workspace = true
 common-time.workspace = true
 datatypes.workspace = true
+enum_dispatch = "0.3"
 hydroflow = "0.5.0"
 itertools.workspace = true
 num-traits = "0.2"
@@ -27,3 +28,6 @@ session.workspace = true
 snafu.workspace = true
 tokio.workspace = true
 tonic.workspace = true
+
+[dev-dependencies]
+serde_json = "1.0"

src/flow/src/expr/error.rs

@@ -61,4 +61,7 @@ pub enum EvalError {
 
     #[snafu(display("Unsupported temporal filter: {reason}"))]
     UnsupportedTemporalFilter { reason: String, location: Location },
+
+    #[snafu(display("Overflowed during evaluation"))]
+    Overflow { location: Location },
 }