diff --git a/c++/greptime/v1/flow/server.pb.cc b/c++/greptime/v1/flow/server.pb.cc
index 09775ccc..21743ff4 100644
--- a/c++/greptime/v1/flow/server.pb.cc
+++ b/c++/greptime/v1/flow/server.pb.cc
@@ -77,8 +77,8 @@ struct FlowResponse_ExtensionEntry_DoNotUseDefaultTypeInternal {
 PROTOBUF_ATTRIBUTE_NO_DESTROY PROTOBUF_CONSTINIT PROTOBUF_ATTRIBUTE_INIT_PRIORITY1 FlowResponse_ExtensionEntry_DoNotUseDefaultTypeInternal _FlowResponse_ExtensionEntry_DoNotUse_default_instance_;
 PROTOBUF_CONSTEXPR FlowResponse::FlowResponse(
     ::_pbi::ConstantInitialized): _impl_{
-    /*decltype(_impl_.extension_)*/{::_pbi::ConstantInitialized()}
-  , /*decltype(_impl_.affected_tasks_)*/{}
+    /*decltype(_impl_.affected_flows_)*/{}
+  , /*decltype(_impl_.extension_)*/{::_pbi::ConstantInitialized()}
   , /*decltype(_impl_.header_)*/nullptr
   , /*decltype(_impl_.affected_rows_)*/uint64_t{0u}
   , /*decltype(_impl_._cached_size_)*/{}} {}
@@ -185,8 +185,8 @@ const uint32_t TableStruct_greptime_2fv1_2fflow_2fserver_2eproto::offsets[] PROT
   ~0u,  // no _inlined_string_donated_
   PROTOBUF_FIELD_OFFSET(::greptime::v1::flow::FlowResponse, _impl_.header_),
   PROTOBUF_FIELD_OFFSET(::greptime::v1::flow::FlowResponse, _impl_.affected_rows_),
+  PROTOBUF_FIELD_OFFSET(::greptime::v1::flow::FlowResponse, _impl_.affected_flows_),
   PROTOBUF_FIELD_OFFSET(::greptime::v1::flow::FlowResponse, _impl_.extension_),
-  PROTOBUF_FIELD_OFFSET(::greptime::v1::flow::FlowResponse, _impl_.affected_tasks_),
   PROTOBUF_FIELD_OFFSET(::greptime::v1::flow::CreateRequest_FlowOptionsEntry_DoNotUse, _has_bits_),
   PROTOBUF_FIELD_OFFSET(::greptime::v1::flow::CreateRequest_FlowOptionsEntry_DoNotUse, _internal_metadata_),
   ~0u,  // no _extensions_
@@ -253,10 +253,10 @@ const char descriptor_table_protodef_greptime_2fv1_2fflow_2fserver_2eproto[] PRO
   "estH\000\022-\n\004drop\030\002 \001(\0132\035.greptime.v1.flow.D"
   "ropRequestH\000B\006\n\004body\"\363\001\n\014FlowResponse\022+\n"
   "\006header\030\001 \001(\0132\033.greptime.v1.ResponseHead"
-  "er\022\025\n\raffected_rows\030\002 \001(\004\022@\n\textension\030\003"
-  " \003(\0132-.greptime.v1.flow.FlowResponse.Ext"
-  "ensionEntry\022+\n\016affected_tasks\030\004 \003(\0132\023.gr"
-  "eptime.v1.FlowId\0320\n\016ExtensionEntry\022\013\n\003ke"
+  "er\022\025\n\raffected_rows\030\002 \001(\004\022+\n\016affected_fl"
+  "ows\030\003 \003(\0132\023.greptime.v1.FlowId\022@\n\textens"
+  "ion\030\004 \003(\0132-.greptime.v1.flow.FlowRespons"
+  "e.ExtensionEntry\0320\n\016ExtensionEntry\022\013\n\003ke"
   "y\030\001 \001(\t\022\r\n\005value\030\002 \001(\014:\0028\001\"\343\002\n\rCreateReq"
   "uest\022$\n\007flow_id\030\001 \001(\0132\023.greptime.v1.Flow"
   "Id\022.\n\020source_table_ids\030\002 \003(\0132\024.greptime."
@@ -1051,8 +1051,8 @@ void FlowResponse::clear_header() {
   }
   _impl_.header_ = nullptr;
 }
-void FlowResponse::clear_affected_tasks() {
-  _impl_.affected_tasks_.Clear();
+void FlowResponse::clear_affected_flows() {
+  _impl_.affected_flows_.Clear();
 }
 FlowResponse::FlowResponse(::PROTOBUF_NAMESPACE_ID::Arena* arena,
                          bool is_message_owned)
@@ -1067,8 +1067,8 @@ FlowResponse::FlowResponse(const FlowResponse& from)
   : ::PROTOBUF_NAMESPACE_ID::Message() {
   FlowResponse* const _this = this; (void)_this;
   new (&_impl_) Impl_{
-      /*decltype(_impl_.extension_)*/{}
-    , decltype(_impl_.affected_tasks_){from._impl_.affected_tasks_}
+      decltype(_impl_.affected_flows_){from._impl_.affected_flows_}
+    , /*decltype(_impl_.extension_)*/{}
     , decltype(_impl_.header_){nullptr}
     , decltype(_impl_.affected_rows_){}
     , /*decltype(_impl_._cached_size_)*/{}};
@@ -1087,8 +1087,8 @@ inline void FlowResponse::SharedCtor(
   (void)arena;
   (void)is_message_owned;
   new (&_impl_) Impl_{
-      /*decltype(_impl_.extension_)*/{::_pbi::ArenaInitialized(), arena}
-    , decltype(_impl_.affected_tasks_){arena}
+      decltype(_impl_.affected_flows_){arena}
+    , /*decltype(_impl_.extension_)*/{::_pbi::ArenaInitialized(), arena}
     , decltype(_impl_.header_){nullptr}
     , decltype(_impl_.affected_rows_){uint64_t{0u}}
     , /*decltype(_impl_._cached_size_)*/{}
@@ -1107,9 +1107,9 @@ FlowResponse::~FlowResponse() {
 
 inline void FlowResponse::SharedDtor() {
   GOOGLE_DCHECK(GetArenaForAllocation() == nullptr);
+  _impl_.affected_flows_.~RepeatedPtrField();
   _impl_.extension_.Destruct();
   _impl_.extension_.~MapField();
-  _impl_.affected_tasks_.~RepeatedPtrField();
   if (this != internal_default_instance()) delete _impl_.header_;
 }
 
@@ -1127,8 +1127,8 @@ void FlowResponse::Clear() {
   // Prevent compiler warnings about cached_has_bits being unused
   (void) cached_has_bits;
 
+  _impl_.affected_flows_.Clear();
   _impl_.extension_.Clear();
-  _impl_.affected_tasks_.Clear();
   if (GetArenaForAllocation() == nullptr && _impl_.header_ != nullptr) {
     delete _impl_.header_;
   }
@@ -1159,26 +1159,26 @@ const char* FlowResponse::_InternalParse(const char* ptr, ::_pbi::ParseContext*
         } else
           goto handle_unusual;
         continue;
-      // map<string, bytes> extension = 3;
+      // repeated .greptime.v1.FlowId affected_flows = 3;
       case 3:
         if (PROTOBUF_PREDICT_TRUE(static_cast<uint8_t>(tag) == 26)) {
           ptr -= 1;
           do {
             ptr += 1;
-            ptr = ctx->ParseMessage(&_impl_.extension_, ptr);
+            ptr = ctx->ParseMessage(_internal_add_affected_flows(), ptr);
             CHK_(ptr);
             if (!ctx->DataAvailable(ptr)) break;
           } while (::PROTOBUF_NAMESPACE_ID::internal::ExpectTag<26>(ptr));
         } else
           goto handle_unusual;
         continue;
-      // repeated .greptime.v1.FlowId affected_tasks = 4;
+      // map<string, bytes> extension = 4;
       case 4:
         if (PROTOBUF_PREDICT_TRUE(static_cast<uint8_t>(tag) == 34)) {
           ptr -= 1;
           do {
             ptr += 1;
-            ptr = ctx->ParseMessage(_internal_add_affected_tasks(), ptr);
+            ptr = ctx->ParseMessage(&_impl_.extension_, ptr);
             CHK_(ptr);
             if (!ctx->DataAvailable(ptr)) break;
           } while (::PROTOBUF_NAMESPACE_ID::internal::ExpectTag<34>(ptr));
@@ -1227,7 +1227,15 @@ uint8_t* FlowResponse::_InternalSerialize(
     target = ::_pbi::WireFormatLite::WriteUInt64ToArray(2, this->_internal_affected_rows(), target);
   }
 
-  // map<string, bytes> extension = 3;
+  // repeated .greptime.v1.FlowId affected_flows = 3;
+  for (unsigned i = 0,
+      n = static_cast<unsigned>(this->_internal_affected_flows_size()); i < n; i++) {
+    const auto& repfield = this->_internal_affected_flows(i);
+    target = ::PROTOBUF_NAMESPACE_ID::internal::WireFormatLite::
+        InternalWriteMessage(3, repfield, repfield.GetCachedSize(), target, stream);
+  }
+
+  // map<string, bytes> extension = 4;
   if (!this->_internal_extension().empty()) {
     using MapType = ::_pb::Map<std::string, std::string>;
     using WireHelper = FlowResponse_ExtensionEntry_DoNotUse::Funcs;
@@ -1242,25 +1250,17 @@ uint8_t* FlowResponse::_InternalSerialize(
 
     if (stream->IsSerializationDeterministic() && map_field.size() > 1) {
       for (const auto& entry : ::_pbi::MapSorterPtr<MapType>(map_field)) {
-        target = WireHelper::InternalSerialize(3, entry.first, entry.second, target, stream);
+        target = WireHelper::InternalSerialize(4, entry.first, entry.second, target, stream);
         check_utf8(entry);
       }
     } else {
       for (const auto& entry : map_field) {
-        target = WireHelper::InternalSerialize(3, entry.first, entry.second, target, stream);
+        target = WireHelper::InternalSerialize(4, entry.first, entry.second, target, stream);
         check_utf8(entry);
       }
     }
   }
 
-  // repeated .greptime.v1.FlowId affected_tasks = 4;
-  for (unsigned i = 0,
-      n = static_cast<unsigned>(this->_internal_affected_tasks_size()); i < n; i++) {
-    const auto& repfield = this->_internal_affected_tasks(i);
-    target = ::PROTOBUF_NAMESPACE_ID::internal::WireFormatLite::
-        InternalWriteMessage(4, repfield, repfield.GetCachedSize(), target, stream);
-  }
-
   if (PROTOBUF_PREDICT_FALSE(_internal_metadata_.have_unknown_fields())) {
     target = ::_pbi::WireFormat::InternalSerializeUnknownFieldsToArray(
         _internal_metadata_.unknown_fields<::PROTOBUF_NAMESPACE_ID::UnknownFieldSet>(::PROTOBUF_NAMESPACE_ID::UnknownFieldSet::default_instance), target, stream);
@@ -1277,7 +1277,14 @@ size_t FlowResponse::ByteSizeLong() const {
   // Prevent compiler warnings about cached_has_bits being unused
   (void) cached_has_bits;
 
-  // map<string, bytes> extension = 3;
+  // repeated .greptime.v1.FlowId affected_flows = 3;
+  total_size += 1UL * this->_internal_affected_flows_size();
+  for (const auto& msg : this->_impl_.affected_flows_) {
+    total_size +=
+      ::PROTOBUF_NAMESPACE_ID::internal::WireFormatLite::MessageSize(msg);
+  }
+
+  // map<string, bytes> extension = 4;
   total_size += 1 *
       ::PROTOBUF_NAMESPACE_ID::internal::FromIntSize(this->_internal_extension_size());
   for (::PROTOBUF_NAMESPACE_ID::Map< std::string, std::string >::const_iterator
@@ -1286,13 +1293,6 @@ size_t FlowResponse::ByteSizeLong() const {
     total_size += FlowResponse_ExtensionEntry_DoNotUse::Funcs::ByteSizeLong(it->first, it->second);
   }
 
-  // repeated .greptime.v1.FlowId affected_tasks = 4;
-  total_size += 1UL * this->_internal_affected_tasks_size();
-  for (const auto& msg : this->_impl_.affected_tasks_) {
-    total_size +=
-      ::PROTOBUF_NAMESPACE_ID::internal::WireFormatLite::MessageSize(msg);
-  }
-
   // .greptime.v1.ResponseHeader header = 1;
   if (this->_internal_has_header()) {
     total_size += 1 +
@@ -1323,8 +1323,8 @@ void FlowResponse::MergeImpl(::PROTOBUF_NAMESPACE_ID::Message& to_msg, const ::P
   uint32_t cached_has_bits = 0;
   (void) cached_has_bits;
 
+  _this->_impl_.affected_flows_.MergeFrom(from._impl_.affected_flows_);
   _this->_impl_.extension_.MergeFrom(from._impl_.extension_);
-  _this->_impl_.affected_tasks_.MergeFrom(from._impl_.affected_tasks_);
   if (from._internal_has_header()) {
     _this->_internal_mutable_header()->::greptime::v1::ResponseHeader::MergeFrom(
         from._internal_header());
@@ -1349,8 +1349,8 @@ bool FlowResponse::IsInitialized() const {
 void FlowResponse::InternalSwap(FlowResponse* other) {
   using std::swap;
   _internal_metadata_.InternalSwap(&other->_internal_metadata_);
+  _impl_.affected_flows_.InternalSwap(&other->_impl_.affected_flows_);
   _impl_.extension_.InternalSwap(&other->_impl_.extension_);
-  _impl_.affected_tasks_.InternalSwap(&other->_impl_.affected_tasks_);
   ::PROTOBUF_NAMESPACE_ID::internal::memswap<
       PROTOBUF_FIELD_OFFSET(FlowResponse, _impl_.affected_rows_)
       + sizeof(FlowResponse::_impl_.affected_rows_)
diff --git a/c++/greptime/v1/flow/server.pb.h b/c++/greptime/v1/flow/server.pb.h
index dc0d13ad..f4c6d36c 100644
--- a/c++/greptime/v1/flow/server.pb.h
+++ b/c++/greptime/v1/flow/server.pb.h
@@ -767,12 +767,30 @@ class FlowResponse final :
   // accessors -------------------------------------------------------
 
   enum : int {
-    kExtensionFieldNumber = 3,
-    kAffectedTasksFieldNumber = 4,
+    kAffectedFlowsFieldNumber = 3,
+    kExtensionFieldNumber = 4,
     kHeaderFieldNumber = 1,
     kAffectedRowsFieldNumber = 2,
   };
-  // map<string, bytes> extension = 3;
+  // repeated .greptime.v1.FlowId affected_flows = 3;
+  int affected_flows_size() const;
+  private:
+  int _internal_affected_flows_size() const;
+  public:
+  void clear_affected_flows();
+  ::greptime::v1::FlowId* mutable_affected_flows(int index);
+  ::PROTOBUF_NAMESPACE_ID::RepeatedPtrField< ::greptime::v1::FlowId >*
+      mutable_affected_flows();
+  private:
+  const ::greptime::v1::FlowId& _internal_affected_flows(int index) const;
+  ::greptime::v1::FlowId* _internal_add_affected_flows();
+  public:
+  const ::greptime::v1::FlowId& affected_flows(int index) const;
+  ::greptime::v1::FlowId* add_affected_flows();
+  const ::PROTOBUF_NAMESPACE_ID::RepeatedPtrField< ::greptime::v1::FlowId >&
+      affected_flows() const;
+
+  // map<string, bytes> extension = 4;
   int extension_size() const;
   private:
   int _internal_extension_size() const;
@@ -789,24 +807,6 @@ class FlowResponse final :
   ::PROTOBUF_NAMESPACE_ID::Map< std::string, std::string >*
       mutable_extension();
 
-  // repeated .greptime.v1.FlowId affected_tasks = 4;
-  int affected_tasks_size() const;
-  private:
-  int _internal_affected_tasks_size() const;
-  public:
-  void clear_affected_tasks();
-  ::greptime::v1::FlowId* mutable_affected_tasks(int index);
-  ::PROTOBUF_NAMESPACE_ID::RepeatedPtrField< ::greptime::v1::FlowId >*
-      mutable_affected_tasks();
-  private:
-  const ::greptime::v1::FlowId& _internal_affected_tasks(int index) const;
-  ::greptime::v1::FlowId* _internal_add_affected_tasks();
-  public:
-  const ::greptime::v1::FlowId& affected_tasks(int index) const;
-  ::greptime::v1::FlowId* add_affected_tasks();
-  const ::PROTOBUF_NAMESPACE_ID::RepeatedPtrField< ::greptime::v1::FlowId >&
-      affected_tasks() const;
-
   // .greptime.v1.ResponseHeader header = 1;
   bool has_header() const;
   private:
@@ -842,12 +842,12 @@ class FlowResponse final :
   typedef void InternalArenaConstructable_;
   typedef void DestructorSkippable_;
   struct Impl_ {
+    ::PROTOBUF_NAMESPACE_ID::RepeatedPtrField< ::greptime::v1::FlowId > affected_flows_;
     ::PROTOBUF_NAMESPACE_ID::internal::MapField<
         FlowResponse_ExtensionEntry_DoNotUse,
         std::string, std::string,
         ::PROTOBUF_NAMESPACE_ID::internal::WireFormatLite::TYPE_STRING,
         ::PROTOBUF_NAMESPACE_ID::internal::WireFormatLite::TYPE_BYTES> extension_;
-    ::PROTOBUF_NAMESPACE_ID::RepeatedPtrField< ::greptime::v1::FlowId > affected_tasks_;
     ::greptime::v1::ResponseHeader* header_;
     uint64_t affected_rows_;
     mutable ::PROTOBUF_NAMESPACE_ID::internal::CachedSize _cached_size_;
@@ -1754,7 +1754,44 @@ inline void FlowResponse::set_affected_rows(uint64_t value) {
   // @@protoc_insertion_point(field_set:greptime.v1.flow.FlowResponse.affected_rows)
 }
 
-// map<string, bytes> extension = 3;
+// repeated .greptime.v1.FlowId affected_flows = 3;
+inline int FlowResponse::_internal_affected_flows_size() const {
+  return _impl_.affected_flows_.size();
+}
+inline int FlowResponse::affected_flows_size() const {
+  return _internal_affected_flows_size();
+}
+inline ::greptime::v1::FlowId* FlowResponse::mutable_affected_flows(int index) {
+  // @@protoc_insertion_point(field_mutable:greptime.v1.flow.FlowResponse.affected_flows)
+  return _impl_.affected_flows_.Mutable(index);
+}
+inline ::PROTOBUF_NAMESPACE_ID::RepeatedPtrField< ::greptime::v1::FlowId >*
+FlowResponse::mutable_affected_flows() {
+  // @@protoc_insertion_point(field_mutable_list:greptime.v1.flow.FlowResponse.affected_flows)
+  return &_impl_.affected_flows_;
+}
+inline const ::greptime::v1::FlowId& FlowResponse::_internal_affected_flows(int index) const {
+  return _impl_.affected_flows_.Get(index);
+}
+inline const ::greptime::v1::FlowId& FlowResponse::affected_flows(int index) const {
+  // @@protoc_insertion_point(field_get:greptime.v1.flow.FlowResponse.affected_flows)
+  return _internal_affected_flows(index);
+}
+inline ::greptime::v1::FlowId* FlowResponse::_internal_add_affected_flows() {
+  return _impl_.affected_flows_.Add();
+}
+inline ::greptime::v1::FlowId* FlowResponse::add_affected_flows() {
+  ::greptime::v1::FlowId* _add = _internal_add_affected_flows();
+  // @@protoc_insertion_point(field_add:greptime.v1.flow.FlowResponse.affected_flows)
+  return _add;
+}
+inline const ::PROTOBUF_NAMESPACE_ID::RepeatedPtrField< ::greptime::v1::FlowId >&
+FlowResponse::affected_flows() const {
+  // @@protoc_insertion_point(field_list:greptime.v1.flow.FlowResponse.affected_flows)
+  return _impl_.affected_flows_;
+}
+
+// map<string, bytes> extension = 4;
 inline int FlowResponse::_internal_extension_size() const {
   return _impl_.extension_.size();
 }
@@ -1783,43 +1820,6 @@ FlowResponse::mutable_extension() {
   return _internal_mutable_extension();
 }
 
-// repeated .greptime.v1.FlowId affected_tasks = 4;
-inline int FlowResponse::_internal_affected_tasks_size() const {
-  return _impl_.affected_tasks_.size();
-}
-inline int FlowResponse::affected_tasks_size() const {
-  return _internal_affected_tasks_size();
-}
-inline ::greptime::v1::FlowId* FlowResponse::mutable_affected_tasks(int index) {
-  // @@protoc_insertion_point(field_mutable:greptime.v1.flow.FlowResponse.affected_tasks)
-  return _impl_.affected_tasks_.Mutable(index);
-}
-inline ::PROTOBUF_NAMESPACE_ID::RepeatedPtrField< ::greptime::v1::FlowId >*
-FlowResponse::mutable_affected_tasks() {
-  // @@protoc_insertion_point(field_mutable_list:greptime.v1.flow.FlowResponse.affected_tasks)
-  return &_impl_.affected_tasks_;
-}
-inline const ::greptime::v1::FlowId& FlowResponse::_internal_affected_tasks(int index) const {
-  return _impl_.affected_tasks_.Get(index);
-}
-inline const ::greptime::v1::FlowId& FlowResponse::affected_tasks(int index) const {
-  // @@protoc_insertion_point(field_get:greptime.v1.flow.FlowResponse.affected_tasks)
-  return _internal_affected_tasks(index);
-}
-inline ::greptime::v1::FlowId* FlowResponse::_internal_add_affected_tasks() {
-  return _impl_.affected_tasks_.Add();
-}
-inline ::greptime::v1::FlowId* FlowResponse::add_affected_tasks() {
-  ::greptime::v1::FlowId* _add = _internal_add_affected_tasks();
-  // @@protoc_insertion_point(field_add:greptime.v1.flow.FlowResponse.affected_tasks)
-  return _add;
-}
-inline const ::PROTOBUF_NAMESPACE_ID::RepeatedPtrField< ::greptime::v1::FlowId >&
-FlowResponse::affected_tasks() const {
-  // @@protoc_insertion_point(field_list:greptime.v1.flow.FlowResponse.affected_tasks)
-  return _impl_.affected_tasks_;
-}
-
 // -------------------------------------------------------------------
 
 // -------------------------------------------------------------------
diff --git a/java/src/main/java/io/greptime/v1/flow/Server.java b/java/src/main/java/io/greptime/v1/flow/Server.java
index 65d4b76e..b82c9eed 100644
--- a/java/src/main/java/io/greptime/v1/flow/Server.java
+++ b/java/src/main/java/io/greptime/v1/flow/Server.java
@@ -2480,11 +2480,55 @@ public interface FlowResponseOrBuilder extends
     long getAffectedRows();
 
     /**
-     * <code>map&lt;string, bytes&gt; extension = 3;</code>
+     * <pre>
+     * affected flow ids
+     * </pre>
+     *
+     * <code>repeated .greptime.v1.FlowId affected_flows = 3;</code>
+     */
+    java.util.List<io.greptime.v1.Ddl.FlowId> 
+        getAffectedFlowsList();
+    /**
+     * <pre>
+     * affected flow ids
+     * </pre>
+     *
+     * <code>repeated .greptime.v1.FlowId affected_flows = 3;</code>
+     */
+    io.greptime.v1.Ddl.FlowId getAffectedFlows(int index);
+    /**
+     * <pre>
+     * affected flow ids
+     * </pre>
+     *
+     * <code>repeated .greptime.v1.FlowId affected_flows = 3;</code>
+     */
+    int getAffectedFlowsCount();
+    /**
+     * <pre>
+     * affected flow ids
+     * </pre>
+     *
+     * <code>repeated .greptime.v1.FlowId affected_flows = 3;</code>
+     */
+    java.util.List<? extends io.greptime.v1.Ddl.FlowIdOrBuilder> 
+        getAffectedFlowsOrBuilderList();
+    /**
+     * <pre>
+     * affected flow ids
+     * </pre>
+     *
+     * <code>repeated .greptime.v1.FlowId affected_flows = 3;</code>
+     */
+    io.greptime.v1.Ddl.FlowIdOrBuilder getAffectedFlowsOrBuilder(
+        int index);
+
+    /**
+     * <code>map&lt;string, bytes&gt; extension = 4;</code>
      */
     int getExtensionCount();
     /**
-     * <code>map&lt;string, bytes&gt; extension = 3;</code>
+     * <code>map&lt;string, bytes&gt; extension = 4;</code>
      */
     boolean containsExtension(
         java.lang.String key);
@@ -2495,12 +2539,12 @@ boolean containsExtension(
     java.util.Map<java.lang.String, com.google.protobuf.ByteString>
     getExtension();
     /**
-     * <code>map&lt;string, bytes&gt; extension = 3;</code>
+     * <code>map&lt;string, bytes&gt; extension = 4;</code>
      */
     java.util.Map<java.lang.String, com.google.protobuf.ByteString>
     getExtensionMap();
     /**
-     * <code>map&lt;string, bytes&gt; extension = 3;</code>
+     * <code>map&lt;string, bytes&gt; extension = 4;</code>
      */
 
     /* nullable */
@@ -2509,55 +2553,11 @@ com.google.protobuf.ByteString getExtensionOrDefault(
         /* nullable */
 com.google.protobuf.ByteString defaultValue);
     /**
-     * <code>map&lt;string, bytes&gt; extension = 3;</code>
+     * <code>map&lt;string, bytes&gt; extension = 4;</code>
      */
 
     com.google.protobuf.ByteString getExtensionOrThrow(
         java.lang.String key);
-
-    /**
-     * <pre>
-     * affected task ids
-     * </pre>
-     *
-     * <code>repeated .greptime.v1.FlowId affected_tasks = 4;</code>
-     */
-    java.util.List<io.greptime.v1.Ddl.FlowId> 
-        getAffectedTasksList();
-    /**
-     * <pre>
-     * affected task ids
-     * </pre>
-     *
-     * <code>repeated .greptime.v1.FlowId affected_tasks = 4;</code>
-     */
-    io.greptime.v1.Ddl.FlowId getAffectedTasks(int index);
-    /**
-     * <pre>
-     * affected task ids
-     * </pre>
-     *
-     * <code>repeated .greptime.v1.FlowId affected_tasks = 4;</code>
-     */
-    int getAffectedTasksCount();
-    /**
-     * <pre>
-     * affected task ids
-     * </pre>
-     *
-     * <code>repeated .greptime.v1.FlowId affected_tasks = 4;</code>
-     */
-    java.util.List<? extends io.greptime.v1.Ddl.FlowIdOrBuilder> 
-        getAffectedTasksOrBuilderList();
-    /**
-     * <pre>
-     * affected task ids
-     * </pre>
-     *
-     * <code>repeated .greptime.v1.FlowId affected_tasks = 4;</code>
-     */
-    io.greptime.v1.Ddl.FlowIdOrBuilder getAffectedTasksOrBuilder(
-        int index);
   }
   /**
    * Protobuf type {@code greptime.v1.flow.FlowResponse}
@@ -2572,7 +2572,7 @@ private FlowResponse(com.google.protobuf.GeneratedMessageV3.Builder<?> builder)
       super(builder);
     }
     private FlowResponse() {
-      affectedTasks_ = java.util.Collections.emptyList();
+      affectedFlows_ = java.util.Collections.emptyList();
     }
 
     @java.lang.Override
@@ -2626,24 +2626,24 @@ private FlowResponse(
             }
             case 26: {
               if (!((mutable_bitField0_ & 0x00000001) != 0)) {
-                extension_ = com.google.protobuf.MapField.newMapField(
-                    ExtensionDefaultEntryHolder.defaultEntry);
+                affectedFlows_ = new java.util.ArrayList<io.greptime.v1.Ddl.FlowId>();
                 mutable_bitField0_ |= 0x00000001;
               }
-              com.google.protobuf.MapEntry<java.lang.String, com.google.protobuf.ByteString>
-              extension__ = input.readMessage(
-                  ExtensionDefaultEntryHolder.defaultEntry.getParserForType(), extensionRegistry);
-              extension_.getMutableMap().put(
-                  extension__.getKey(), extension__.getValue());
+              affectedFlows_.add(
+                  input.readMessage(io.greptime.v1.Ddl.FlowId.parser(), extensionRegistry));
               break;
             }
             case 34: {
               if (!((mutable_bitField0_ & 0x00000002) != 0)) {
-                affectedTasks_ = new java.util.ArrayList<io.greptime.v1.Ddl.FlowId>();
+                extension_ = com.google.protobuf.MapField.newMapField(
+                    ExtensionDefaultEntryHolder.defaultEntry);
                 mutable_bitField0_ |= 0x00000002;
               }
-              affectedTasks_.add(
-                  input.readMessage(io.greptime.v1.Ddl.FlowId.parser(), extensionRegistry));
+              com.google.protobuf.MapEntry<java.lang.String, com.google.protobuf.ByteString>
+              extension__ = input.readMessage(
+                  ExtensionDefaultEntryHolder.defaultEntry.getParserForType(), extensionRegistry);
+              extension_.getMutableMap().put(
+                  extension__.getKey(), extension__.getValue());
               break;
             }
             default: {
@@ -2663,8 +2663,8 @@ private FlowResponse(
         throw new com.google.protobuf.InvalidProtocolBufferException(
             e).setUnfinishedMessage(this);
       } finally {
-        if (((mutable_bitField0_ & 0x00000002) != 0)) {
-          affectedTasks_ = java.util.Collections.unmodifiableList(affectedTasks_);
+        if (((mutable_bitField0_ & 0x00000001) != 0)) {
+          affectedFlows_ = java.util.Collections.unmodifiableList(affectedFlows_);
         }
         this.unknownFields = unknownFields.build();
         makeExtensionsImmutable();
@@ -2680,7 +2680,7 @@ private FlowResponse(
     protected com.google.protobuf.MapField internalGetMapField(
         int number) {
       switch (number) {
-        case 3:
+        case 4:
           return internalGetExtension();
         default:
           throw new RuntimeException(
@@ -2732,7 +2732,67 @@ public long getAffectedRows() {
       return affectedRows_;
     }
 
-    public static final int EXTENSION_FIELD_NUMBER = 3;
+    public static final int AFFECTED_FLOWS_FIELD_NUMBER = 3;
+    private java.util.List<io.greptime.v1.Ddl.FlowId> affectedFlows_;
+    /**
+     * <pre>
+     * affected flow ids
+     * </pre>
+     *
+     * <code>repeated .greptime.v1.FlowId affected_flows = 3;</code>
+     */
+    @java.lang.Override
+    public java.util.List<io.greptime.v1.Ddl.FlowId> getAffectedFlowsList() {
+      return affectedFlows_;
+    }
+    /**
+     * <pre>
+     * affected flow ids
+     * </pre>
+     *
+     * <code>repeated .greptime.v1.FlowId affected_flows = 3;</code>
+     */
+    @java.lang.Override
+    public java.util.List<? extends io.greptime.v1.Ddl.FlowIdOrBuilder> 
+        getAffectedFlowsOrBuilderList() {
+      return affectedFlows_;
+    }
+    /**
+     * <pre>
+     * affected flow ids
+     * </pre>
+     *
+     * <code>repeated .greptime.v1.FlowId affected_flows = 3;</code>
+     */
+    @java.lang.Override
+    public int getAffectedFlowsCount() {
+      return affectedFlows_.size();
+    }
+    /**
+     * <pre>
+     * affected flow ids
+     * </pre>
+     *
+     * <code>repeated .greptime.v1.FlowId affected_flows = 3;</code>
+     */
+    @java.lang.Override
+    public io.greptime.v1.Ddl.FlowId getAffectedFlows(int index) {
+      return affectedFlows_.get(index);
+    }
+    /**
+     * <pre>
+     * affected flow ids
+     * </pre>
+     *
+     * <code>repeated .greptime.v1.FlowId affected_flows = 3;</code>
+     */
+    @java.lang.Override
+    public io.greptime.v1.Ddl.FlowIdOrBuilder getAffectedFlowsOrBuilder(
+        int index) {
+      return affectedFlows_.get(index);
+    }
+
+    public static final int EXTENSION_FIELD_NUMBER = 4;
     private static final class ExtensionDefaultEntryHolder {
       static final com.google.protobuf.MapEntry<
           java.lang.String, com.google.protobuf.ByteString> defaultEntry =
@@ -2759,7 +2819,7 @@ public int getExtensionCount() {
       return internalGetExtension().getMap().size();
     }
     /**
-     * <code>map&lt;string, bytes&gt; extension = 3;</code>
+     * <code>map&lt;string, bytes&gt; extension = 4;</code>
      */
 
     @java.lang.Override
@@ -2777,7 +2837,7 @@ public java.util.Map<java.lang.String, com.google.protobuf.ByteString> getExtens
       return getExtensionMap();
     }
     /**
-     * <code>map&lt;string, bytes&gt; extension = 3;</code>
+     * <code>map&lt;string, bytes&gt; extension = 4;</code>
      */
     @java.lang.Override
 
@@ -2785,7 +2845,7 @@ public java.util.Map<java.lang.String, com.google.protobuf.ByteString> getExtens
       return internalGetExtension().getMap();
     }
     /**
-     * <code>map&lt;string, bytes&gt; extension = 3;</code>
+     * <code>map&lt;string, bytes&gt; extension = 4;</code>
      */
     @java.lang.Override
 
@@ -2798,7 +2858,7 @@ public com.google.protobuf.ByteString getExtensionOrDefault(
       return map.containsKey(key) ? map.get(key) : defaultValue;
     }
     /**
-     * <code>map&lt;string, bytes&gt; extension = 3;</code>
+     * <code>map&lt;string, bytes&gt; extension = 4;</code>
      */
     @java.lang.Override
 
@@ -2813,66 +2873,6 @@ public com.google.protobuf.ByteString getExtensionOrThrow(
       return map.get(key);
     }
 
-    public static final int AFFECTED_TASKS_FIELD_NUMBER = 4;
-    private java.util.List<io.greptime.v1.Ddl.FlowId> affectedTasks_;
-    /**
-     * <pre>
-     * affected task ids
-     * </pre>
-     *
-     * <code>repeated .greptime.v1.FlowId affected_tasks = 4;</code>
-     */
-    @java.lang.Override
-    public java.util.List<io.greptime.v1.Ddl.FlowId> getAffectedTasksList() {
-      return affectedTasks_;
-    }
-    /**
-     * <pre>
-     * affected task ids
-     * </pre>
-     *
-     * <code>repeated .greptime.v1.FlowId affected_tasks = 4;</code>
-     */
-    @java.lang.Override
-    public java.util.List<? extends io.greptime.v1.Ddl.FlowIdOrBuilder> 
-        getAffectedTasksOrBuilderList() {
-      return affectedTasks_;
-    }
-    /**
-     * <pre>
-     * affected task ids
-     * </pre>
-     *
-     * <code>repeated .greptime.v1.FlowId affected_tasks = 4;</code>
-     */
-    @java.lang.Override
-    public int getAffectedTasksCount() {
-      return affectedTasks_.size();
-    }
-    /**
-     * <pre>
-     * affected task ids
-     * </pre>
-     *
-     * <code>repeated .greptime.v1.FlowId affected_tasks = 4;</code>
-     */
-    @java.lang.Override
-    public io.greptime.v1.Ddl.FlowId getAffectedTasks(int index) {
-      return affectedTasks_.get(index);
-    }
-    /**
-     * <pre>
-     * affected task ids
-     * </pre>
-     *
-     * <code>repeated .greptime.v1.FlowId affected_tasks = 4;</code>
-     */
-    @java.lang.Override
-    public io.greptime.v1.Ddl.FlowIdOrBuilder getAffectedTasksOrBuilder(
-        int index) {
-      return affectedTasks_.get(index);
-    }
-
     private byte memoizedIsInitialized = -1;
     @java.lang.Override
     public final boolean isInitialized() {
@@ -2893,15 +2893,15 @@ public void writeTo(com.google.protobuf.CodedOutputStream output)
       if (affectedRows_ != 0L) {
         output.writeUInt64(2, affectedRows_);
       }
+      for (int i = 0; i < affectedFlows_.size(); i++) {
+        output.writeMessage(3, affectedFlows_.get(i));
+      }
       com.google.protobuf.GeneratedMessageV3
         .serializeStringMapTo(
           output,
           internalGetExtension(),
           ExtensionDefaultEntryHolder.defaultEntry,
-          3);
-      for (int i = 0; i < affectedTasks_.size(); i++) {
-        output.writeMessage(4, affectedTasks_.get(i));
-      }
+          4);
       unknownFields.writeTo(output);
     }
 
@@ -2919,6 +2919,10 @@ public int getSerializedSize() {
         size += com.google.protobuf.CodedOutputStream
           .computeUInt64Size(2, affectedRows_);
       }
+      for (int i = 0; i < affectedFlows_.size(); i++) {
+        size += com.google.protobuf.CodedOutputStream
+          .computeMessageSize(3, affectedFlows_.get(i));
+      }
       for (java.util.Map.Entry<java.lang.String, com.google.protobuf.ByteString> entry
            : internalGetExtension().getMap().entrySet()) {
         com.google.protobuf.MapEntry<java.lang.String, com.google.protobuf.ByteString>
@@ -2927,11 +2931,7 @@ public int getSerializedSize() {
             .setValue(entry.getValue())
             .build();
         size += com.google.protobuf.CodedOutputStream
-            .computeMessageSize(3, extension__);
-      }
-      for (int i = 0; i < affectedTasks_.size(); i++) {
-        size += com.google.protobuf.CodedOutputStream
-          .computeMessageSize(4, affectedTasks_.get(i));
+            .computeMessageSize(4, extension__);
       }
       size += unknownFields.getSerializedSize();
       memoizedSize = size;
@@ -2955,10 +2955,10 @@ public boolean equals(final java.lang.Object obj) {
       }
       if (getAffectedRows()
           != other.getAffectedRows()) return false;
+      if (!getAffectedFlowsList()
+          .equals(other.getAffectedFlowsList())) return false;
       if (!internalGetExtension().equals(
           other.internalGetExtension())) return false;
-      if (!getAffectedTasksList()
-          .equals(other.getAffectedTasksList())) return false;
       if (!unknownFields.equals(other.unknownFields)) return false;
       return true;
     }
@@ -2977,14 +2977,14 @@ public int hashCode() {
       hash = (37 * hash) + AFFECTED_ROWS_FIELD_NUMBER;
       hash = (53 * hash) + com.google.protobuf.Internal.hashLong(
           getAffectedRows());
+      if (getAffectedFlowsCount() > 0) {
+        hash = (37 * hash) + AFFECTED_FLOWS_FIELD_NUMBER;
+        hash = (53 * hash) + getAffectedFlowsList().hashCode();
+      }
       if (!internalGetExtension().getMap().isEmpty()) {
         hash = (37 * hash) + EXTENSION_FIELD_NUMBER;
         hash = (53 * hash) + internalGetExtension().hashCode();
       }
-      if (getAffectedTasksCount() > 0) {
-        hash = (37 * hash) + AFFECTED_TASKS_FIELD_NUMBER;
-        hash = (53 * hash) + getAffectedTasksList().hashCode();
-      }
       hash = (29 * hash) + unknownFields.hashCode();
       memoizedHashCode = hash;
       return hash;
@@ -3096,7 +3096,7 @@ public static final class Builder extends
       protected com.google.protobuf.MapField internalGetMapField(
           int number) {
         switch (number) {
-          case 3:
+          case 4:
             return internalGetExtension();
           default:
             throw new RuntimeException(
@@ -3107,7 +3107,7 @@ protected com.google.protobuf.MapField internalGetMapField(
       protected com.google.protobuf.MapField internalGetMutableMapField(
           int number) {
         switch (number) {
-          case 3:
+          case 4:
             return internalGetMutableExtension();
           default:
             throw new RuntimeException(
@@ -3135,7 +3135,7 @@ private Builder(
       private void maybeForceBuilderInitialization() {
         if (com.google.protobuf.GeneratedMessageV3
                 .alwaysUseFieldBuilders) {
-          getAffectedTasksFieldBuilder();
+          getAffectedFlowsFieldBuilder();
         }
       }
       @java.lang.Override
@@ -3149,13 +3149,13 @@ public Builder clear() {
         }
         affectedRows_ = 0L;
 
-        internalGetMutableExtension().clear();
-        if (affectedTasksBuilder_ == null) {
-          affectedTasks_ = java.util.Collections.emptyList();
-          bitField0_ = (bitField0_ & ~0x00000002);
+        if (affectedFlowsBuilder_ == null) {
+          affectedFlows_ = java.util.Collections.emptyList();
+          bitField0_ = (bitField0_ & ~0x00000001);
         } else {
-          affectedTasksBuilder_.clear();
+          affectedFlowsBuilder_.clear();
         }
+        internalGetMutableExtension().clear();
         return this;
       }
 
@@ -3189,17 +3189,17 @@ public io.greptime.v1.flow.Server.FlowResponse buildPartial() {
           result.header_ = headerBuilder_.build();
         }
         result.affectedRows_ = affectedRows_;
-        result.extension_ = internalGetExtension();
-        result.extension_.makeImmutable();
-        if (affectedTasksBuilder_ == null) {
-          if (((bitField0_ & 0x00000002) != 0)) {
-            affectedTasks_ = java.util.Collections.unmodifiableList(affectedTasks_);
-            bitField0_ = (bitField0_ & ~0x00000002);
+        if (affectedFlowsBuilder_ == null) {
+          if (((bitField0_ & 0x00000001) != 0)) {
+            affectedFlows_ = java.util.Collections.unmodifiableList(affectedFlows_);
+            bitField0_ = (bitField0_ & ~0x00000001);
           }
-          result.affectedTasks_ = affectedTasks_;
+          result.affectedFlows_ = affectedFlows_;
         } else {
-          result.affectedTasks_ = affectedTasksBuilder_.build();
+          result.affectedFlows_ = affectedFlowsBuilder_.build();
         }
+        result.extension_ = internalGetExtension();
+        result.extension_.makeImmutable();
         onBuilt();
         return result;
       }
@@ -3254,34 +3254,34 @@ public Builder mergeFrom(io.greptime.v1.flow.Server.FlowResponse other) {
         if (other.getAffectedRows() != 0L) {
           setAffectedRows(other.getAffectedRows());
         }
-        internalGetMutableExtension().mergeFrom(
-            other.internalGetExtension());
-        if (affectedTasksBuilder_ == null) {
-          if (!other.affectedTasks_.isEmpty()) {
-            if (affectedTasks_.isEmpty()) {
-              affectedTasks_ = other.affectedTasks_;
-              bitField0_ = (bitField0_ & ~0x00000002);
+        if (affectedFlowsBuilder_ == null) {
+          if (!other.affectedFlows_.isEmpty()) {
+            if (affectedFlows_.isEmpty()) {
+              affectedFlows_ = other.affectedFlows_;
+              bitField0_ = (bitField0_ & ~0x00000001);
             } else {
-              ensureAffectedTasksIsMutable();
-              affectedTasks_.addAll(other.affectedTasks_);
+              ensureAffectedFlowsIsMutable();
+              affectedFlows_.addAll(other.affectedFlows_);
             }
             onChanged();
           }
         } else {
-          if (!other.affectedTasks_.isEmpty()) {
-            if (affectedTasksBuilder_.isEmpty()) {
-              affectedTasksBuilder_.dispose();
-              affectedTasksBuilder_ = null;
-              affectedTasks_ = other.affectedTasks_;
-              bitField0_ = (bitField0_ & ~0x00000002);
-              affectedTasksBuilder_ = 
+          if (!other.affectedFlows_.isEmpty()) {
+            if (affectedFlowsBuilder_.isEmpty()) {
+              affectedFlowsBuilder_.dispose();
+              affectedFlowsBuilder_ = null;
+              affectedFlows_ = other.affectedFlows_;
+              bitField0_ = (bitField0_ & ~0x00000001);
+              affectedFlowsBuilder_ = 
                 com.google.protobuf.GeneratedMessageV3.alwaysUseFieldBuilders ?
-                   getAffectedTasksFieldBuilder() : null;
+                   getAffectedFlowsFieldBuilder() : null;
             } else {
-              affectedTasksBuilder_.addAllMessages(other.affectedTasks_);
+              affectedFlowsBuilder_.addAllMessages(other.affectedFlows_);
             }
           }
         }
+        internalGetMutableExtension().mergeFrom(
+            other.internalGetExtension());
         this.mergeUnknownFields(other.unknownFields);
         onChanged();
         return this;
@@ -3462,447 +3462,447 @@ public Builder clearAffectedRows() {
         return this;
       }
 
-      private com.google.protobuf.MapField<
-          java.lang.String, com.google.protobuf.ByteString> extension_;
-      private com.google.protobuf.MapField<java.lang.String, com.google.protobuf.ByteString>
-      internalGetExtension() {
-        if (extension_ == null) {
-          return com.google.protobuf.MapField.emptyMapField(
-              ExtensionDefaultEntryHolder.defaultEntry);
-        }
-        return extension_;
-      }
-      private com.google.protobuf.MapField<java.lang.String, com.google.protobuf.ByteString>
-      internalGetMutableExtension() {
-        onChanged();;
-        if (extension_ == null) {
-          extension_ = com.google.protobuf.MapField.newMapField(
-              ExtensionDefaultEntryHolder.defaultEntry);
-        }
-        if (!extension_.isMutable()) {
-          extension_ = extension_.copy();
-        }
-        return extension_;
-      }
-
-      public int getExtensionCount() {
-        return internalGetExtension().getMap().size();
-      }
-      /**
-       * <code>map&lt;string, bytes&gt; extension = 3;</code>
-       */
-
-      @java.lang.Override
-      public boolean containsExtension(
-          java.lang.String key) {
-        if (key == null) { throw new NullPointerException("map key"); }
-        return internalGetExtension().getMap().containsKey(key);
-      }
-      /**
-       * Use {@link #getExtensionMap()} instead.
-       */
-      @java.lang.Override
-      @java.lang.Deprecated
-      public java.util.Map<java.lang.String, com.google.protobuf.ByteString> getExtension() {
-        return getExtensionMap();
-      }
-      /**
-       * <code>map&lt;string, bytes&gt; extension = 3;</code>
-       */
-      @java.lang.Override
-
-      public java.util.Map<java.lang.String, com.google.protobuf.ByteString> getExtensionMap() {
-        return internalGetExtension().getMap();
-      }
-      /**
-       * <code>map&lt;string, bytes&gt; extension = 3;</code>
-       */
-      @java.lang.Override
-
-      public com.google.protobuf.ByteString getExtensionOrDefault(
-          java.lang.String key,
-          com.google.protobuf.ByteString defaultValue) {
-        if (key == null) { throw new NullPointerException("map key"); }
-        java.util.Map<java.lang.String, com.google.protobuf.ByteString> map =
-            internalGetExtension().getMap();
-        return map.containsKey(key) ? map.get(key) : defaultValue;
-      }
-      /**
-       * <code>map&lt;string, bytes&gt; extension = 3;</code>
-       */
-      @java.lang.Override
-
-      public com.google.protobuf.ByteString getExtensionOrThrow(
-          java.lang.String key) {
-        if (key == null) { throw new NullPointerException("map key"); }
-        java.util.Map<java.lang.String, com.google.protobuf.ByteString> map =
-            internalGetExtension().getMap();
-        if (!map.containsKey(key)) {
-          throw new java.lang.IllegalArgumentException();
-        }
-        return map.get(key);
-      }
-
-      public Builder clearExtension() {
-        internalGetMutableExtension().getMutableMap()
-            .clear();
-        return this;
-      }
-      /**
-       * <code>map&lt;string, bytes&gt; extension = 3;</code>
-       */
-
-      public Builder removeExtension(
-          java.lang.String key) {
-        if (key == null) { throw new NullPointerException("map key"); }
-        internalGetMutableExtension().getMutableMap()
-            .remove(key);
-        return this;
-      }
-      /**
-       * Use alternate mutation accessors instead.
-       */
-      @java.lang.Deprecated
-      public java.util.Map<java.lang.String, com.google.protobuf.ByteString>
-      getMutableExtension() {
-        return internalGetMutableExtension().getMutableMap();
-      }
-      /**
-       * <code>map&lt;string, bytes&gt; extension = 3;</code>
-       */
-      public Builder putExtension(
-          java.lang.String key,
-          com.google.protobuf.ByteString value) {
-        if (key == null) { throw new NullPointerException("map key"); }
-        if (value == null) {
-  throw new NullPointerException("map value");
-}
-
-        internalGetMutableExtension().getMutableMap()
-            .put(key, value);
-        return this;
-      }
-      /**
-       * <code>map&lt;string, bytes&gt; extension = 3;</code>
-       */
-
-      public Builder putAllExtension(
-          java.util.Map<java.lang.String, com.google.protobuf.ByteString> values) {
-        internalGetMutableExtension().getMutableMap()
-            .putAll(values);
-        return this;
-      }
-
-      private java.util.List<io.greptime.v1.Ddl.FlowId> affectedTasks_ =
+      private java.util.List<io.greptime.v1.Ddl.FlowId> affectedFlows_ =
         java.util.Collections.emptyList();
-      private void ensureAffectedTasksIsMutable() {
-        if (!((bitField0_ & 0x00000002) != 0)) {
-          affectedTasks_ = new java.util.ArrayList<io.greptime.v1.Ddl.FlowId>(affectedTasks_);
-          bitField0_ |= 0x00000002;
+      private void ensureAffectedFlowsIsMutable() {
+        if (!((bitField0_ & 0x00000001) != 0)) {
+          affectedFlows_ = new java.util.ArrayList<io.greptime.v1.Ddl.FlowId>(affectedFlows_);
+          bitField0_ |= 0x00000001;
          }
       }
 
       private com.google.protobuf.RepeatedFieldBuilderV3<
-          io.greptime.v1.Ddl.FlowId, io.greptime.v1.Ddl.FlowId.Builder, io.greptime.v1.Ddl.FlowIdOrBuilder> affectedTasksBuilder_;
+          io.greptime.v1.Ddl.FlowId, io.greptime.v1.Ddl.FlowId.Builder, io.greptime.v1.Ddl.FlowIdOrBuilder> affectedFlowsBuilder_;
 
       /**
        * <pre>
-       * affected task ids
+       * affected flow ids
        * </pre>
        *
-       * <code>repeated .greptime.v1.FlowId affected_tasks = 4;</code>
+       * <code>repeated .greptime.v1.FlowId affected_flows = 3;</code>
        */
-      public java.util.List<io.greptime.v1.Ddl.FlowId> getAffectedTasksList() {
-        if (affectedTasksBuilder_ == null) {
-          return java.util.Collections.unmodifiableList(affectedTasks_);
+      public java.util.List<io.greptime.v1.Ddl.FlowId> getAffectedFlowsList() {
+        if (affectedFlowsBuilder_ == null) {
+          return java.util.Collections.unmodifiableList(affectedFlows_);
         } else {
-          return affectedTasksBuilder_.getMessageList();
+          return affectedFlowsBuilder_.getMessageList();
         }
       }
       /**
        * <pre>
-       * affected task ids
+       * affected flow ids
        * </pre>
        *
-       * <code>repeated .greptime.v1.FlowId affected_tasks = 4;</code>
+       * <code>repeated .greptime.v1.FlowId affected_flows = 3;</code>
        */
-      public int getAffectedTasksCount() {
-        if (affectedTasksBuilder_ == null) {
-          return affectedTasks_.size();
+      public int getAffectedFlowsCount() {
+        if (affectedFlowsBuilder_ == null) {
+          return affectedFlows_.size();
         } else {
-          return affectedTasksBuilder_.getCount();
+          return affectedFlowsBuilder_.getCount();
         }
       }
       /**
        * <pre>
-       * affected task ids
+       * affected flow ids
        * </pre>
        *
-       * <code>repeated .greptime.v1.FlowId affected_tasks = 4;</code>
+       * <code>repeated .greptime.v1.FlowId affected_flows = 3;</code>
        */
-      public io.greptime.v1.Ddl.FlowId getAffectedTasks(int index) {
-        if (affectedTasksBuilder_ == null) {
-          return affectedTasks_.get(index);
+      public io.greptime.v1.Ddl.FlowId getAffectedFlows(int index) {
+        if (affectedFlowsBuilder_ == null) {
+          return affectedFlows_.get(index);
         } else {
-          return affectedTasksBuilder_.getMessage(index);
+          return affectedFlowsBuilder_.getMessage(index);
         }
       }
       /**
        * <pre>
-       * affected task ids
+       * affected flow ids
        * </pre>
        *
-       * <code>repeated .greptime.v1.FlowId affected_tasks = 4;</code>
+       * <code>repeated .greptime.v1.FlowId affected_flows = 3;</code>
        */
-      public Builder setAffectedTasks(
+      public Builder setAffectedFlows(
           int index, io.greptime.v1.Ddl.FlowId value) {
-        if (affectedTasksBuilder_ == null) {
+        if (affectedFlowsBuilder_ == null) {
           if (value == null) {
             throw new NullPointerException();
           }
-          ensureAffectedTasksIsMutable();
-          affectedTasks_.set(index, value);
+          ensureAffectedFlowsIsMutable();
+          affectedFlows_.set(index, value);
           onChanged();
         } else {
-          affectedTasksBuilder_.setMessage(index, value);
+          affectedFlowsBuilder_.setMessage(index, value);
         }
         return this;
       }
       /**
        * <pre>
-       * affected task ids
+       * affected flow ids
        * </pre>
        *
-       * <code>repeated .greptime.v1.FlowId affected_tasks = 4;</code>
+       * <code>repeated .greptime.v1.FlowId affected_flows = 3;</code>
        */
-      public Builder setAffectedTasks(
+      public Builder setAffectedFlows(
           int index, io.greptime.v1.Ddl.FlowId.Builder builderForValue) {
-        if (affectedTasksBuilder_ == null) {
-          ensureAffectedTasksIsMutable();
-          affectedTasks_.set(index, builderForValue.build());
+        if (affectedFlowsBuilder_ == null) {
+          ensureAffectedFlowsIsMutable();
+          affectedFlows_.set(index, builderForValue.build());
           onChanged();
         } else {
-          affectedTasksBuilder_.setMessage(index, builderForValue.build());
+          affectedFlowsBuilder_.setMessage(index, builderForValue.build());
         }
         return this;
       }
       /**
        * <pre>
-       * affected task ids
+       * affected flow ids
        * </pre>
        *
-       * <code>repeated .greptime.v1.FlowId affected_tasks = 4;</code>
+       * <code>repeated .greptime.v1.FlowId affected_flows = 3;</code>
        */
-      public Builder addAffectedTasks(io.greptime.v1.Ddl.FlowId value) {
-        if (affectedTasksBuilder_ == null) {
+      public Builder addAffectedFlows(io.greptime.v1.Ddl.FlowId value) {
+        if (affectedFlowsBuilder_ == null) {
           if (value == null) {
             throw new NullPointerException();
           }
-          ensureAffectedTasksIsMutable();
-          affectedTasks_.add(value);
+          ensureAffectedFlowsIsMutable();
+          affectedFlows_.add(value);
           onChanged();
         } else {
-          affectedTasksBuilder_.addMessage(value);
+          affectedFlowsBuilder_.addMessage(value);
         }
         return this;
       }
       /**
        * <pre>
-       * affected task ids
+       * affected flow ids
        * </pre>
        *
-       * <code>repeated .greptime.v1.FlowId affected_tasks = 4;</code>
+       * <code>repeated .greptime.v1.FlowId affected_flows = 3;</code>
        */
-      public Builder addAffectedTasks(
+      public Builder addAffectedFlows(
           int index, io.greptime.v1.Ddl.FlowId value) {
-        if (affectedTasksBuilder_ == null) {
+        if (affectedFlowsBuilder_ == null) {
           if (value == null) {
             throw new NullPointerException();
           }
-          ensureAffectedTasksIsMutable();
-          affectedTasks_.add(index, value);
+          ensureAffectedFlowsIsMutable();
+          affectedFlows_.add(index, value);
           onChanged();
         } else {
-          affectedTasksBuilder_.addMessage(index, value);
+          affectedFlowsBuilder_.addMessage(index, value);
         }
         return this;
       }
       /**
        * <pre>
-       * affected task ids
+       * affected flow ids
        * </pre>
        *
-       * <code>repeated .greptime.v1.FlowId affected_tasks = 4;</code>
+       * <code>repeated .greptime.v1.FlowId affected_flows = 3;</code>
        */
-      public Builder addAffectedTasks(
+      public Builder addAffectedFlows(
           io.greptime.v1.Ddl.FlowId.Builder builderForValue) {
-        if (affectedTasksBuilder_ == null) {
-          ensureAffectedTasksIsMutable();
-          affectedTasks_.add(builderForValue.build());
+        if (affectedFlowsBuilder_ == null) {
+          ensureAffectedFlowsIsMutable();
+          affectedFlows_.add(builderForValue.build());
           onChanged();
         } else {
-          affectedTasksBuilder_.addMessage(builderForValue.build());
+          affectedFlowsBuilder_.addMessage(builderForValue.build());
         }
         return this;
       }
       /**
        * <pre>
-       * affected task ids
+       * affected flow ids
        * </pre>
        *
-       * <code>repeated .greptime.v1.FlowId affected_tasks = 4;</code>
+       * <code>repeated .greptime.v1.FlowId affected_flows = 3;</code>
        */
-      public Builder addAffectedTasks(
+      public Builder addAffectedFlows(
           int index, io.greptime.v1.Ddl.FlowId.Builder builderForValue) {
-        if (affectedTasksBuilder_ == null) {
-          ensureAffectedTasksIsMutable();
-          affectedTasks_.add(index, builderForValue.build());
+        if (affectedFlowsBuilder_ == null) {
+          ensureAffectedFlowsIsMutable();
+          affectedFlows_.add(index, builderForValue.build());
           onChanged();
         } else {
-          affectedTasksBuilder_.addMessage(index, builderForValue.build());
+          affectedFlowsBuilder_.addMessage(index, builderForValue.build());
         }
         return this;
       }
       /**
        * <pre>
-       * affected task ids
+       * affected flow ids
        * </pre>
        *
-       * <code>repeated .greptime.v1.FlowId affected_tasks = 4;</code>
+       * <code>repeated .greptime.v1.FlowId affected_flows = 3;</code>
        */
-      public Builder addAllAffectedTasks(
+      public Builder addAllAffectedFlows(
           java.lang.Iterable<? extends io.greptime.v1.Ddl.FlowId> values) {
-        if (affectedTasksBuilder_ == null) {
-          ensureAffectedTasksIsMutable();
+        if (affectedFlowsBuilder_ == null) {
+          ensureAffectedFlowsIsMutable();
           com.google.protobuf.AbstractMessageLite.Builder.addAll(
-              values, affectedTasks_);
+              values, affectedFlows_);
           onChanged();
         } else {
-          affectedTasksBuilder_.addAllMessages(values);
+          affectedFlowsBuilder_.addAllMessages(values);
         }
         return this;
       }
       /**
        * <pre>
-       * affected task ids
+       * affected flow ids
        * </pre>
        *
-       * <code>repeated .greptime.v1.FlowId affected_tasks = 4;</code>
+       * <code>repeated .greptime.v1.FlowId affected_flows = 3;</code>
        */
-      public Builder clearAffectedTasks() {
-        if (affectedTasksBuilder_ == null) {
-          affectedTasks_ = java.util.Collections.emptyList();
-          bitField0_ = (bitField0_ & ~0x00000002);
+      public Builder clearAffectedFlows() {
+        if (affectedFlowsBuilder_ == null) {
+          affectedFlows_ = java.util.Collections.emptyList();
+          bitField0_ = (bitField0_ & ~0x00000001);
           onChanged();
         } else {
-          affectedTasksBuilder_.clear();
+          affectedFlowsBuilder_.clear();
         }
         return this;
       }
       /**
        * <pre>
-       * affected task ids
+       * affected flow ids
        * </pre>
        *
-       * <code>repeated .greptime.v1.FlowId affected_tasks = 4;</code>
+       * <code>repeated .greptime.v1.FlowId affected_flows = 3;</code>
        */
-      public Builder removeAffectedTasks(int index) {
-        if (affectedTasksBuilder_ == null) {
-          ensureAffectedTasksIsMutable();
-          affectedTasks_.remove(index);
+      public Builder removeAffectedFlows(int index) {
+        if (affectedFlowsBuilder_ == null) {
+          ensureAffectedFlowsIsMutable();
+          affectedFlows_.remove(index);
           onChanged();
         } else {
-          affectedTasksBuilder_.remove(index);
+          affectedFlowsBuilder_.remove(index);
         }
         return this;
       }
       /**
        * <pre>
-       * affected task ids
+       * affected flow ids
        * </pre>
        *
-       * <code>repeated .greptime.v1.FlowId affected_tasks = 4;</code>
+       * <code>repeated .greptime.v1.FlowId affected_flows = 3;</code>
        */
-      public io.greptime.v1.Ddl.FlowId.Builder getAffectedTasksBuilder(
+      public io.greptime.v1.Ddl.FlowId.Builder getAffectedFlowsBuilder(
           int index) {
-        return getAffectedTasksFieldBuilder().getBuilder(index);
+        return getAffectedFlowsFieldBuilder().getBuilder(index);
       }
       /**
        * <pre>
-       * affected task ids
+       * affected flow ids
        * </pre>
        *
-       * <code>repeated .greptime.v1.FlowId affected_tasks = 4;</code>
+       * <code>repeated .greptime.v1.FlowId affected_flows = 3;</code>
        */
-      public io.greptime.v1.Ddl.FlowIdOrBuilder getAffectedTasksOrBuilder(
+      public io.greptime.v1.Ddl.FlowIdOrBuilder getAffectedFlowsOrBuilder(
           int index) {
-        if (affectedTasksBuilder_ == null) {
-          return affectedTasks_.get(index);  } else {
-          return affectedTasksBuilder_.getMessageOrBuilder(index);
+        if (affectedFlowsBuilder_ == null) {
+          return affectedFlows_.get(index);  } else {
+          return affectedFlowsBuilder_.getMessageOrBuilder(index);
         }
       }
       /**
        * <pre>
-       * affected task ids
+       * affected flow ids
        * </pre>
        *
-       * <code>repeated .greptime.v1.FlowId affected_tasks = 4;</code>
+       * <code>repeated .greptime.v1.FlowId affected_flows = 3;</code>
        */
       public java.util.List<? extends io.greptime.v1.Ddl.FlowIdOrBuilder> 
-           getAffectedTasksOrBuilderList() {
-        if (affectedTasksBuilder_ != null) {
-          return affectedTasksBuilder_.getMessageOrBuilderList();
+           getAffectedFlowsOrBuilderList() {
+        if (affectedFlowsBuilder_ != null) {
+          return affectedFlowsBuilder_.getMessageOrBuilderList();
         } else {
-          return java.util.Collections.unmodifiableList(affectedTasks_);
+          return java.util.Collections.unmodifiableList(affectedFlows_);
         }
       }
       /**
        * <pre>
-       * affected task ids
+       * affected flow ids
        * </pre>
        *
-       * <code>repeated .greptime.v1.FlowId affected_tasks = 4;</code>
+       * <code>repeated .greptime.v1.FlowId affected_flows = 3;</code>
        */
-      public io.greptime.v1.Ddl.FlowId.Builder addAffectedTasksBuilder() {
-        return getAffectedTasksFieldBuilder().addBuilder(
+      public io.greptime.v1.Ddl.FlowId.Builder addAffectedFlowsBuilder() {
+        return getAffectedFlowsFieldBuilder().addBuilder(
             io.greptime.v1.Ddl.FlowId.getDefaultInstance());
       }
       /**
        * <pre>
-       * affected task ids
+       * affected flow ids
        * </pre>
        *
-       * <code>repeated .greptime.v1.FlowId affected_tasks = 4;</code>
+       * <code>repeated .greptime.v1.FlowId affected_flows = 3;</code>
        */
-      public io.greptime.v1.Ddl.FlowId.Builder addAffectedTasksBuilder(
+      public io.greptime.v1.Ddl.FlowId.Builder addAffectedFlowsBuilder(
           int index) {
-        return getAffectedTasksFieldBuilder().addBuilder(
+        return getAffectedFlowsFieldBuilder().addBuilder(
             index, io.greptime.v1.Ddl.FlowId.getDefaultInstance());
       }
       /**
        * <pre>
-       * affected task ids
+       * affected flow ids
        * </pre>
        *
-       * <code>repeated .greptime.v1.FlowId affected_tasks = 4;</code>
+       * <code>repeated .greptime.v1.FlowId affected_flows = 3;</code>
        */
       public java.util.List<io.greptime.v1.Ddl.FlowId.Builder> 
-           getAffectedTasksBuilderList() {
-        return getAffectedTasksFieldBuilder().getBuilderList();
+           getAffectedFlowsBuilderList() {
+        return getAffectedFlowsFieldBuilder().getBuilderList();
       }
       private com.google.protobuf.RepeatedFieldBuilderV3<
           io.greptime.v1.Ddl.FlowId, io.greptime.v1.Ddl.FlowId.Builder, io.greptime.v1.Ddl.FlowIdOrBuilder> 
-          getAffectedTasksFieldBuilder() {
-        if (affectedTasksBuilder_ == null) {
-          affectedTasksBuilder_ = new com.google.protobuf.RepeatedFieldBuilderV3<
+          getAffectedFlowsFieldBuilder() {
+        if (affectedFlowsBuilder_ == null) {
+          affectedFlowsBuilder_ = new com.google.protobuf.RepeatedFieldBuilderV3<
               io.greptime.v1.Ddl.FlowId, io.greptime.v1.Ddl.FlowId.Builder, io.greptime.v1.Ddl.FlowIdOrBuilder>(
-                  affectedTasks_,
-                  ((bitField0_ & 0x00000002) != 0),
+                  affectedFlows_,
+                  ((bitField0_ & 0x00000001) != 0),
                   getParentForChildren(),
                   isClean());
-          affectedTasks_ = null;
+          affectedFlows_ = null;
+        }
+        return affectedFlowsBuilder_;
+      }
+
+      private com.google.protobuf.MapField<
+          java.lang.String, com.google.protobuf.ByteString> extension_;
+      private com.google.protobuf.MapField<java.lang.String, com.google.protobuf.ByteString>
+      internalGetExtension() {
+        if (extension_ == null) {
+          return com.google.protobuf.MapField.emptyMapField(
+              ExtensionDefaultEntryHolder.defaultEntry);
         }
-        return affectedTasksBuilder_;
+        return extension_;
+      }
+      private com.google.protobuf.MapField<java.lang.String, com.google.protobuf.ByteString>
+      internalGetMutableExtension() {
+        onChanged();;
+        if (extension_ == null) {
+          extension_ = com.google.protobuf.MapField.newMapField(
+              ExtensionDefaultEntryHolder.defaultEntry);
+        }
+        if (!extension_.isMutable()) {
+          extension_ = extension_.copy();
+        }
+        return extension_;
+      }
+
+      public int getExtensionCount() {
+        return internalGetExtension().getMap().size();
+      }
+      /**
+       * <code>map&lt;string, bytes&gt; extension = 4;</code>
+       */
+
+      @java.lang.Override
+      public boolean containsExtension(
+          java.lang.String key) {
+        if (key == null) { throw new NullPointerException("map key"); }
+        return internalGetExtension().getMap().containsKey(key);
+      }
+      /**
+       * Use {@link #getExtensionMap()} instead.
+       */
+      @java.lang.Override
+      @java.lang.Deprecated
+      public java.util.Map<java.lang.String, com.google.protobuf.ByteString> getExtension() {
+        return getExtensionMap();
+      }
+      /**
+       * <code>map&lt;string, bytes&gt; extension = 4;</code>
+       */
+      @java.lang.Override
+
+      public java.util.Map<java.lang.String, com.google.protobuf.ByteString> getExtensionMap() {
+        return internalGetExtension().getMap();
+      }
+      /**
+       * <code>map&lt;string, bytes&gt; extension = 4;</code>
+       */
+      @java.lang.Override
+
+      public com.google.protobuf.ByteString getExtensionOrDefault(
+          java.lang.String key,
+          com.google.protobuf.ByteString defaultValue) {
+        if (key == null) { throw new NullPointerException("map key"); }
+        java.util.Map<java.lang.String, com.google.protobuf.ByteString> map =
+            internalGetExtension().getMap();
+        return map.containsKey(key) ? map.get(key) : defaultValue;
+      }
+      /**
+       * <code>map&lt;string, bytes&gt; extension = 4;</code>
+       */
+      @java.lang.Override
+
+      public com.google.protobuf.ByteString getExtensionOrThrow(
+          java.lang.String key) {
+        if (key == null) { throw new NullPointerException("map key"); }
+        java.util.Map<java.lang.String, com.google.protobuf.ByteString> map =
+            internalGetExtension().getMap();
+        if (!map.containsKey(key)) {
+          throw new java.lang.IllegalArgumentException();
+        }
+        return map.get(key);
+      }
+
+      public Builder clearExtension() {
+        internalGetMutableExtension().getMutableMap()
+            .clear();
+        return this;
+      }
+      /**
+       * <code>map&lt;string, bytes&gt; extension = 4;</code>
+       */
+
+      public Builder removeExtension(
+          java.lang.String key) {
+        if (key == null) { throw new NullPointerException("map key"); }
+        internalGetMutableExtension().getMutableMap()
+            .remove(key);
+        return this;
+      }
+      /**
+       * Use alternate mutation accessors instead.
+       */
+      @java.lang.Deprecated
+      public java.util.Map<java.lang.String, com.google.protobuf.ByteString>
+      getMutableExtension() {
+        return internalGetMutableExtension().getMutableMap();
+      }
+      /**
+       * <code>map&lt;string, bytes&gt; extension = 4;</code>
+       */
+      public Builder putExtension(
+          java.lang.String key,
+          com.google.protobuf.ByteString value) {
+        if (key == null) { throw new NullPointerException("map key"); }
+        if (value == null) {
+  throw new NullPointerException("map value");
+}
+
+        internalGetMutableExtension().getMutableMap()
+            .put(key, value);
+        return this;
+      }
+      /**
+       * <code>map&lt;string, bytes&gt; extension = 4;</code>
+       */
+
+      public Builder putAllExtension(
+          java.util.Map<java.lang.String, com.google.protobuf.ByteString> values) {
+        internalGetMutableExtension().getMutableMap()
+            .putAll(values);
+        return this;
       }
       @java.lang.Override
       public final Builder setUnknownFields(
@@ -6698,10 +6698,10 @@ public io.greptime.v1.flow.Server.DropRequest getDefaultInstanceForType() {
       "estH\000\022-\n\004drop\030\002 \001(\0132\035.greptime.v1.flow.D" +
       "ropRequestH\000B\006\n\004body\"\363\001\n\014FlowResponse\022+\n" +
       "\006header\030\001 \001(\0132\033.greptime.v1.ResponseHead" +
-      "er\022\025\n\raffected_rows\030\002 \001(\004\022@\n\textension\030\003" +
-      " \003(\0132-.greptime.v1.flow.FlowResponse.Ext" +
-      "ensionEntry\022+\n\016affected_tasks\030\004 \003(\0132\023.gr" +
-      "eptime.v1.FlowId\0320\n\016ExtensionEntry\022\013\n\003ke" +
+      "er\022\025\n\raffected_rows\030\002 \001(\004\022+\n\016affected_fl" +
+      "ows\030\003 \003(\0132\023.greptime.v1.FlowId\022@\n\textens" +
+      "ion\030\004 \003(\0132-.greptime.v1.flow.FlowRespons" +
+      "e.ExtensionEntry\0320\n\016ExtensionEntry\022\013\n\003ke" +
       "y\030\001 \001(\t\022\r\n\005value\030\002 \001(\014:\0028\001\"\343\002\n\rCreateReq" +
       "uest\022$\n\007flow_id\030\001 \001(\0132\023.greptime.v1.Flow" +
       "Id\022.\n\020source_table_ids\030\002 \003(\0132\024.greptime." +
@@ -6751,7 +6751,7 @@ public io.greptime.v1.flow.Server.DropRequest getDefaultInstanceForType() {
     internal_static_greptime_v1_flow_FlowResponse_fieldAccessorTable = new
       com.google.protobuf.GeneratedMessageV3.FieldAccessorTable(
         internal_static_greptime_v1_flow_FlowResponse_descriptor,
-        new java.lang.String[] { "Header", "AffectedRows", "Extension", "AffectedTasks", });
+        new java.lang.String[] { "Header", "AffectedRows", "AffectedFlows", "Extension", });
     internal_static_greptime_v1_flow_FlowResponse_ExtensionEntry_descriptor =
       internal_static_greptime_v1_flow_FlowResponse_descriptor.getNestedTypes().get(0);
     internal_static_greptime_v1_flow_FlowResponse_ExtensionEntry_fieldAccessorTable = new