feat(flow): flow worker (#3934)

* feat: flow worker * chore: fix after cherry pick * refactor: error handling * refactor: error handling Signed-off-by: Zhenchi <[email protected]> * chore: merge origin/main Signed-off-by: Zhenchi <[email protected]> --------- Signed-off-by: Zhenchi <[email protected]> Co-authored-by: Zhenchi <[email protected]>
GreptimeTeam · May 14, 2024 · 15d7b97 · 15d7b97
1 parent 72897a2
commit 15d7b97
Show file tree

Hide file tree

Showing 6 changed files with 554 additions and 7 deletions.
diff --git a/Cargo.lock b/Cargo.lock
diff --git a/src/flow/Cargo.toml b/src/flow/Cargo.toml
@@ -27,6 +27,7 @@ futures = "0.3"
 # it is the same with upstream repo
 async-trait.workspace = true
 common-meta.workspace = true
+enum-as-inner = "0.6.0"
 greptime-proto.workspace = true
 hydroflow = { git = "https://github.com/GreptimeTeam/hydroflow.git", branch = "main" }
 itertools.workspace = true

diff --git a/src/flow/src/adapter.rs b/src/flow/src/adapter.rs
@@ -18,4 +18,8 @@
 pub(crate) mod error;
 pub(crate) mod node_context;
 
-pub(crate) use node_context::FlownodeContext;
+pub(crate) use node_context::{FlowId, FlownodeContext, TableName};
+
+mod worker;
+
+pub const PER_REQ_MAX_ROW_CNT: usize = 8192;
diff --git a/src/flow/src/adapter/error.rs b/src/flow/src/adapter/error.rs
@@ -24,6 +24,7 @@ use datatypes::value::Value;
 use servers::define_into_tonic_status;
 use snafu::{Location, Snafu};
 
+use crate::adapter::FlowId;
 use crate::expr::EvalError;
 
 /// This error is used to represent all possible errors that can occur in the flow module.
@@ -39,7 +40,11 @@ pub enum Error {
  },
 
  #[snafu(display("Internal error"))]
- Internal { location: Location, reason: String },
+ Internal {
+ reason: String,
+ #[snafu(implicit)]
+ location: Location,
+ },
 
  /// TODO(discord9): add detailed location of column
  #[snafu(display("Failed to eval stream"))]
@@ -71,6 +76,20 @@ pub enum Error {
  location: Location,
  },
 
+ #[snafu(display("Flow not found, id={id}"))]
+ FlowNotFound {
+ id: FlowId,
+ #[snafu(implicit)]
+ location: Location,
+ },
+
+ #[snafu(display("Flow already exist, id={id}"))]
+ FlowAlreadyExist {
+ id: FlowId,
+ #[snafu(implicit)]
+ location: Location,
+ },
+
  #[snafu(display("Failed to join task"))]
  JoinTask {
  #[snafu(source)]
@@ -168,10 +187,12 @@ impl ErrorExt for Error {
  Self::Eval { .. } | &Self::JoinTask { .. } | &Self::Datafusion { .. } => {
  StatusCode::Internal
  }
- &Self::TableAlreadyExist { .. } => StatusCode::TableAlreadyExists,
- Self::TableNotFound { .. } | Self::TableNotFoundMeta { .. } => {
- StatusCode::TableNotFound
+ &Self::TableAlreadyExist { .. } | Self::FlowAlreadyExist { .. } => {
+ StatusCode::TableAlreadyExists
  }
+ Self::TableNotFound { .. }
+ | Self::TableNotFoundMeta { .. }
+ | Self::FlowNotFound { .. } => StatusCode::TableNotFound,
  Self::InvalidQueryPlan { .. }
  | Self::InvalidQuerySubstrait { .. }
  | Self::InvalidQueryProst { .. }