diff --git a/Cargo.lock b/Cargo.lock
index b4c70b0dbe6e..3f60831975a7 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -4595,7 +4595,7 @@ dependencies = [
 [[package]]
 name = "greptime-proto"
 version = "0.1.0"
-source = "git+https://github.com/GreptimeTeam/greptime-proto.git?rev=75c5fb569183bb3d0fa1023df9c2214df722b9b1#75c5fb569183bb3d0fa1023df9c2214df722b9b1"
+source = "git+https://github.com/GreptimeTeam/greptime-proto.git?rev=e1070ad3e7ad3d89f8dd1e20756b0c0cc26f3365#e1070ad3e7ad3d89f8dd1e20756b0c0cc26f3365"
 dependencies = [
  "prost 0.12.6",
  "serde",
diff --git a/Cargo.toml b/Cargo.toml
index 0a06bf16aba5..ee02ab0b71cb 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -122,7 +122,7 @@ etcd-client = "0.13"
 fst = "0.4.7"
 futures = "0.3"
 futures-util = "0.3"
-greptime-proto = { git = "https://github.com/GreptimeTeam/greptime-proto.git", rev = "75c5fb569183bb3d0fa1023df9c2214df722b9b1" }
+greptime-proto = { git = "https://github.com/GreptimeTeam/greptime-proto.git", rev = "e1070ad3e7ad3d89f8dd1e20756b0c0cc26f3365" }
 hex = "0.4"
 humantime = "2.1"
 humantime-serde = "1.1"
diff --git a/src/common/meta/src/ddl/create_flow.rs b/src/common/meta/src/ddl/create_flow.rs
index d60ee1dfd380..afaf606078c0 100644
--- a/src/common/meta/src/ddl/create_flow.rs
+++ b/src/common/meta/src/ddl/create_flow.rs
@@ -286,6 +286,7 @@ impl From<&CreateFlowData> for CreateRequest {
             sink_table_name: Some(value.task.sink_table_name.clone().into()),
             // Always be true
             create_if_not_exists: true,
+            or_replace: true,
             expire_after: value.task.expire_after.map(|value| ExpireAfter { value }),
             comment: value.task.comment.clone(),
             sql: value.task.sql.clone(),
diff --git a/src/flow/src/adapter/flownode_impl.rs b/src/flow/src/adapter/flownode_impl.rs
index 739804be930f..8e5a02858820 100644
--- a/src/flow/src/adapter/flownode_impl.rs
+++ b/src/flow/src/adapter/flownode_impl.rs
@@ -57,6 +57,7 @@ impl Flownode for FlowWorkerManager {
                 comment,
                 sql,
                 flow_options,
+                or_replace: _,
             })) => {
                 let source_table_ids = source_table_ids.into_iter().map(|id| id.id).collect_vec();
                 let sink_table_name = [