feat: introduce partition_rule_version (#306)

* feat: introduce `partition_rule_version` Signed-off-by: WenyXu <wenymedia@gmail.com> * feat: introduce `partition_rule_version` Signed-off-by: WenyXu <wenymedia@gmail.com> * feat: introduce PartitionRuleVersion Signed-off-by: WenyXu <wenymedia@gmail.com> * chore: apply suggestions Signed-off-by: WenyXu <wenymedia@gmail.com> --------- Signed-off-by: WenyXu <wenymedia@gmail.com>
2026-02-06 19:34:18 +08:00
parent f1e0ea3a71
commit b9452e3411
15 changed files with 3348 additions and 884 deletions
@@ -92,6 +92,7 @@ PROTOBUF_ATTRIBUTE_NO_DESTROY PROTOBUF_CONSTINIT PROTOBUF_ATTRIBUTE_INIT_PRIORIT
 PROTOBUF_CONSTEXPR InsertRequest::InsertRequest(
    ::_pbi::ConstantInitialized): _impl_{
    /*decltype(_impl_.rows_)*/nullptr
+  , /*decltype(_impl_.partition_rule_version_)*/nullptr
  , /*decltype(_impl_.region_id_)*/uint64_t{0u}
  , /*decltype(_impl_._cached_size_)*/{}} {}
 struct InsertRequestDefaultTypeInternal {
@@ -260,6 +261,7 @@ const uint32_t TableStruct_greptime_2fv1_2fflow_2fserver_2eproto::offsets[] PROT
  ~0u,  // no _inlined_string_donated_
  PROTOBUF_FIELD_OFFSET(::greptime::v1::flow::InsertRequest, _impl_.region_id_),
  PROTOBUF_FIELD_OFFSET(::greptime::v1::flow::InsertRequest, _impl_.rows_),
+  PROTOBUF_FIELD_OFFSET(::greptime::v1::flow::InsertRequest, _impl_.partition_rule_version_),
  ~0u,  // no _has_bits_
  PROTOBUF_FIELD_OFFSET(::greptime::v1::flow::FlowRequest, _internal_metadata_),
  ~0u,  // no _extensions_
@@ -339,13 +341,13 @@ static const ::_pbi::MigrationSchema schemas[] PROTOBUF_SECTION_VARIABLE(protode
  { 25, -1, -1, sizeof(::greptime::v1::flow::FlowRequestHeader)},
  { 33, -1, -1, sizeof(::greptime::v1::flow::InsertRequests)},
  { 40, -1, -1, sizeof(::greptime::v1::flow::InsertRequest)},
-  { 48, -1, -1, sizeof(::greptime::v1::flow::FlowRequest)},
-  { 59, 67, -1, sizeof(::greptime::v1::flow::FlowResponse_ExtensionsEntry_DoNotUse)},
-  { 69, -1, -1, sizeof(::greptime::v1::flow::FlowResponse)},
-  { 79, 87, -1, sizeof(::greptime::v1::flow::CreateRequest_FlowOptionsEntry_DoNotUse)},
-  { 89, -1, -1, sizeof(::greptime::v1::flow::CreateRequest)},
-  { 105, -1, -1, sizeof(::greptime::v1::flow::DropRequest)},
-  { 112, -1, -1, sizeof(::greptime::v1::flow::FlushFlow)},
+  { 49, -1, -1, sizeof(::greptime::v1::flow::FlowRequest)},
+  { 60, 68, -1, sizeof(::greptime::v1::flow::FlowResponse_ExtensionsEntry_DoNotUse)},
+  { 70, -1, -1, sizeof(::greptime::v1::flow::FlowResponse)},
+  { 80, 88, -1, sizeof(::greptime::v1::flow::CreateRequest_FlowOptionsEntry_DoNotUse)},
+  { 90, -1, -1, sizeof(::greptime::v1::flow::CreateRequest)},
+  { 106, -1, -1, sizeof(::greptime::v1::flow::DropRequest)},
+  { 113, -1, -1, sizeof(::greptime::v1::flow::FlushFlow)},
 };

 static const ::_pb::Message* const file_default_instances[] = {
@@ -378,44 +380,46 @@ const char descriptor_table_protodef_greptime_2fv1_2fflow_2fserver_2eproto[] PRO
  "1.QueryContext\0325\n\023TracingContextEntry\022\013\n"
  "\003key\030\001 \001(\t\022\r\n\005value\030\002 \001(\t:\0028\001\"C\n\016InsertR"
  "equests\0221\n\010requests\030\001 \003(\0132\037.greptime.v1."
-  "flow.InsertRequest\"C\n\rInsertRequest\022\021\n\tr"
-  "egion_id\030\001 \001(\004\022\037\n\004rows\030\002 \001(\0132\021.greptime."
-  "v1.Rows\"\332\001\n\013FlowRequest\0223\n\006header\030@ \001(\0132"
-  "#.greptime.v1.flow.FlowRequestHeader\0221\n\006"
-  "create\030\001 \001(\0132\037.greptime.v1.flow.CreateRe"
-  "questH\000\022-\n\004drop\030\002 \001(\0132\035.greptime.v1.flow"
-  ".DropRequestH\000\022,\n\005flush\030\003 \001(\0132\033.greptime"
-  ".v1.flow.FlushFlowH\000B\006\n\004body\"\366\001\n\014FlowRes"
-  "ponse\022+\n\006header\030\001 \001(\0132\033.greptime.v1.Resp"
-  "onseHeader\022\025\n\raffected_rows\030\002 \001(\004\022+\n\016aff"
-  "ected_flows\030\003 \003(\0132\023.greptime.v1.FlowId\022B"
-  "\n\nextensions\030\004 \003(\0132..greptime.v1.flow.Fl"
-  "owResponse.ExtensionsEntry\0321\n\017Extensions"
-  "Entry\022\013\n\003key\030\001 \001(\t\022\r\n\005value\030\002 \001(\014:\0028\001\"\304\003"
-  "\n\rCreateRequest\022$\n\007flow_id\030\001 \001(\0132\023.grept"
-  "ime.v1.FlowId\022.\n\020source_table_ids\030\002 \003(\0132"
-  "\024.greptime.v1.TableId\022/\n\017sink_table_name"
-  "\030\003 \001(\0132\026.greptime.v1.TableName\022\034\n\024create"
-  "_if_not_exists\030\004 \001(\010\022.\n\014expire_after\030\005 \001"
-  "(\0132\030.greptime.v1.ExpireAfter\0220\n\reval_int"
-  "erval\030\n \001(\0132\031.greptime.v1.EvalInterval\022\017"
-  "\n\007comment\030\006 \001(\t\022\013\n\003sql\030\007 \001(\t\022F\n\014flow_opt"
-  "ions\030\010 \003(\01320.greptime.v1.flow.CreateRequ"
-  "est.FlowOptionsEntry\022\022\n\nor_replace\030\t \001(\010"
-  "\0322\n\020FlowOptionsEntry\022\013\n\003key\030\001 \001(\t\022\r\n\005val"
-  "ue\030\002 \001(\t:\0028\001\"3\n\013DropRequest\022$\n\007flow_id\030\001"
-  " \001(\0132\023.greptime.v1.FlowId\"1\n\tFlushFlow\022$"
-  "\n\007flow_id\030\001 \001(\0132\023.greptime.v1.FlowId2\230\002\n"
-  "\004Flow\022S\n\022HandleCreateRemove\022\035.greptime.v"
-  "1.flow.FlowRequest\032\036.greptime.v1.flow.Fl"
-  "owResponse\022W\n\023HandleMirrorRequest\022 .grep"
-  "time.v1.flow.InsertRequests\032\036.greptime.v"
-  "1.flow.FlowResponse\022b\n\031HandleMarkDirtyTi"
-  "meWindow\022%.greptime.v1.flow.DirtyWindowR"
-  "equests\032\036.greptime.v1.flow.FlowResponseB"
-  "Y\n\023io.greptime.v1.flowB\006ServerZ:github.c"
-  "om/GreptimeTeam/greptime-proto/go/grepti"
-  "me/v1/flowb\006proto3"
+  "flow.InsertRequest\"\206\001\n\rInsertRequest\022\021\n\t"
+  "region_id\030\001 \001(\004\022\037\n\004rows\030\002 \001(\0132\021.greptime"
+  ".v1.Rows\022A\n\026partition_rule_version\030\003 \001(\013"
+  "2!.greptime.v1.PartitionRuleVersion\"\332\001\n\013"
+  "FlowRequest\0223\n\006header\030@ \001(\0132#.greptime.v"
+  "1.flow.FlowRequestHeader\0221\n\006create\030\001 \001(\013"
+  "2\037.greptime.v1.flow.CreateRequestH\000\022-\n\004d"
+  "rop\030\002 \001(\0132\035.greptime.v1.flow.DropRequest"
+  "H\000\022,\n\005flush\030\003 \001(\0132\033.greptime.v1.flow.Flu"
+  "shFlowH\000B\006\n\004body\"\366\001\n\014FlowResponse\022+\n\006hea"
+  "der\030\001 \001(\0132\033.greptime.v1.ResponseHeader\022\025"
+  "\n\raffected_rows\030\002 \001(\004\022+\n\016affected_flows\030"
+  "\003 \003(\0132\023.greptime.v1.FlowId\022B\n\nextensions"
+  "\030\004 \003(\0132..greptime.v1.flow.FlowResponse.E"
+  "xtensionsEntry\0321\n\017ExtensionsEntry\022\013\n\003key"
+  "\030\001 \001(\t\022\r\n\005value\030\002 \001(\014:\0028\001\"\304\003\n\rCreateRequ"
+  "est\022$\n\007flow_id\030\001 \001(\0132\023.greptime.v1.FlowI"
+  "d\022.\n\020source_table_ids\030\002 \003(\0132\024.greptime.v"
+  "1.TableId\022/\n\017sink_table_name\030\003 \001(\0132\026.gre"
+  "ptime.v1.TableName\022\034\n\024create_if_not_exis"
+  "ts\030\004 \001(\010\022.\n\014expire_after\030\005 \001(\0132\030.greptim"
+  "e.v1.ExpireAfter\0220\n\reval_interval\030\n \001(\0132"
+  "\031.greptime.v1.EvalInterval\022\017\n\007comment\030\006 "
+  "\001(\t\022\013\n\003sql\030\007 \001(\t\022F\n\014flow_options\030\010 \003(\01320"
+  ".greptime.v1.flow.CreateRequest.FlowOpti"
+  "onsEntry\022\022\n\nor_replace\030\t \001(\010\0322\n\020FlowOpti"
+  "onsEntry\022\013\n\003key\030\001 \001(\t\022\r\n\005value\030\002 \001(\t:\0028\001"
+  "\"3\n\013DropRequest\022$\n\007flow_id\030\001 \001(\0132\023.grept"
+  "ime.v1.FlowId\"1\n\tFlushFlow\022$\n\007flow_id\030\001 "
+  "\001(\0132\023.greptime.v1.FlowId2\230\002\n\004Flow\022S\n\022Han"
+  "dleCreateRemove\022\035.greptime.v1.flow.FlowR"
+  "equest\032\036.greptime.v1.flow.FlowResponse\022W"
+  "\n\023HandleMirrorRequest\022 .greptime.v1.flow"
+  ".InsertRequests\032\036.greptime.v1.flow.FlowR"
+  "esponse\022b\n\031HandleMarkDirtyTimeWindow\022%.g"
+  "reptime.v1.flow.DirtyWindowRequests\032\036.gr"
+  "eptime.v1.flow.FlowResponseBY\n\023io.grepti"
+  "me.v1.flowB\006ServerZ:github.com/GreptimeT"
+  "eam/greptime-proto/go/greptime/v1/flowb\006"
+  "proto3"
  ;
 static const ::_pbi::DescriptorTable* const descriptor_table_greptime_2fv1_2fflow_2fserver_2eproto_deps[3] = {
  &::descriptor_table_greptime_2fv1_2fcommon_2eproto,
@@ -424,7 +428,7 @@ static const ::_pbi::DescriptorTable* const descriptor_table_greptime_2fv1_2fflo
 };
 static ::_pbi::once_flag descriptor_table_greptime_2fv1_2fflow_2fserver_2eproto_once;
 const ::_pbi::DescriptorTable descriptor_table_greptime_2fv1_2fflow_2fserver_2eproto = {
-    false, false, 2018, descriptor_table_protodef_greptime_2fv1_2fflow_2fserver_2eproto,
+    false, false, 2086, descriptor_table_protodef_greptime_2fv1_2fflow_2fserver_2eproto,
    "greptime/v1/flow/server.proto",
    &descriptor_table_greptime_2fv1_2fflow_2fserver_2eproto_once, descriptor_table_greptime_2fv1_2fflow_2fserver_2eproto_deps, 3, 13,
    schemas, file_default_instances, TableStruct_greptime_2fv1_2fflow_2fserver_2eproto::offsets,
@@ -1317,18 +1321,29 @@ void InsertRequests::InternalSwap(InsertRequests* other) {
 class InsertRequest::_Internal {
 public:
  static const ::greptime::v1::Rows& rows(const InsertRequest* msg);
+  static const ::greptime::v1::PartitionRuleVersion& partition_rule_version(const InsertRequest* msg);
 };

 const ::greptime::v1::Rows&
 InsertRequest::_Internal::rows(const InsertRequest* msg) {
  return *msg->_impl_.rows_;
 }
+const ::greptime::v1::PartitionRuleVersion&
+InsertRequest::_Internal::partition_rule_version(const InsertRequest* msg) {
+  return *msg->_impl_.partition_rule_version_;
+}
 void InsertRequest::clear_rows() {
  if (GetArenaForAllocation() == nullptr && _impl_.rows_ != nullptr) {
    delete _impl_.rows_;
  }
  _impl_.rows_ = nullptr;
 }
+void InsertRequest::clear_partition_rule_version() {
+  if (GetArenaForAllocation() == nullptr && _impl_.partition_rule_version_ != nullptr) {
+    delete _impl_.partition_rule_version_;
+  }
+  _impl_.partition_rule_version_ = nullptr;
+}
 InsertRequest::InsertRequest(::PROTOBUF_NAMESPACE_ID::Arena* arena,
                         bool is_message_owned)
  : ::PROTOBUF_NAMESPACE_ID::Message(arena, is_message_owned) {
@@ -1340,6 +1355,7 @@ InsertRequest::InsertRequest(const InsertRequest& from)
  InsertRequest* const _this = this; (void)_this;
  new (&_impl_) Impl_{
      decltype(_impl_.rows_){nullptr}
+    , decltype(_impl_.partition_rule_version_){nullptr}
    , decltype(_impl_.region_id_){}
    , /*decltype(_impl_._cached_size_)*/{}};

@@ -1347,6 +1363,9 @@ InsertRequest::InsertRequest(const InsertRequest& from)
  if (from._internal_has_rows()) {
    _this->_impl_.rows_ = new ::greptime::v1::Rows(*from._impl_.rows_);
  }
+  if (from._internal_has_partition_rule_version()) {
+    _this->_impl_.partition_rule_version_ = new ::greptime::v1::PartitionRuleVersion(*from._impl_.partition_rule_version_);
+  }
  _this->_impl_.region_id_ = from._impl_.region_id_;
  // @@protoc_insertion_point(copy_constructor:greptime.v1.flow.InsertRequest)
 }
@@ -1357,6 +1376,7 @@ inline void InsertRequest::SharedCtor(
  (void)is_message_owned;
  new (&_impl_) Impl_{
      decltype(_impl_.rows_){nullptr}
+    , decltype(_impl_.partition_rule_version_){nullptr}
    , decltype(_impl_.region_id_){uint64_t{0u}}
    , /*decltype(_impl_._cached_size_)*/{}
  };
@@ -1374,6 +1394,7 @@ InsertRequest::~InsertRequest() {
 inline void InsertRequest::SharedDtor() {
  GOOGLE_DCHECK(GetArenaForAllocation() == nullptr);
  if (this != internal_default_instance()) delete _impl_.rows_;
+  if (this != internal_default_instance()) delete _impl_.partition_rule_version_;
 }

 void InsertRequest::SetCachedSize(int size) const {
@@ -1390,6 +1411,10 @@ void InsertRequest::Clear() {
    delete _impl_.rows_;
  }
  _impl_.rows_ = nullptr;
+  if (GetArenaForAllocation() == nullptr && _impl_.partition_rule_version_ != nullptr) {
+    delete _impl_.partition_rule_version_;
+  }
+  _impl_.partition_rule_version_ = nullptr;
  _impl_.region_id_ = uint64_t{0u};
  _internal_metadata_.Clear<::PROTOBUF_NAMESPACE_ID::UnknownFieldSet>();
 }
@@ -1416,6 +1441,14 @@ const char* InsertRequest::_InternalParse(const char* ptr, ::_pbi::ParseContext*
        } else
          goto handle_unusual;
        continue;
+      // .greptime.v1.PartitionRuleVersion partition_rule_version = 3;
+      case 3:
+        if (PROTOBUF_PREDICT_TRUE(static_cast<uint8_t>(tag) == 26)) {
+          ptr = ctx->ParseMessage(_internal_mutable_partition_rule_version(), ptr);
+          CHK_(ptr);
+        } else
+          goto handle_unusual;
+        continue;
      default:
        goto handle_unusual;
    }  // switch
@@ -1458,6 +1491,13 @@ uint8_t* InsertRequest::_InternalSerialize(
        _Internal::rows(this).GetCachedSize(), target, stream);
  }

+  // .greptime.v1.PartitionRuleVersion partition_rule_version = 3;
+  if (this->_internal_has_partition_rule_version()) {
+    target = ::PROTOBUF_NAMESPACE_ID::internal::WireFormatLite::
+      InternalWriteMessage(3, _Internal::partition_rule_version(this),
+        _Internal::partition_rule_version(this).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);
@@ -1481,6 +1521,13 @@ size_t InsertRequest::ByteSizeLong() const {
        *_impl_.rows_);
  }

+  // .greptime.v1.PartitionRuleVersion partition_rule_version = 3;
+  if (this->_internal_has_partition_rule_version()) {
+    total_size += 1 +
+      ::PROTOBUF_NAMESPACE_ID::internal::WireFormatLite::MessageSize(
+        *_impl_.partition_rule_version_);
+  }
+
  // uint64 region_id = 1;
  if (this->_internal_region_id() != 0) {
    total_size += ::_pbi::WireFormatLite::UInt64SizePlusOne(this->_internal_region_id());
@@ -1508,6 +1555,10 @@ void InsertRequest::MergeImpl(::PROTOBUF_NAMESPACE_ID::Message& to_msg, const ::
    _this->_internal_mutable_rows()->::greptime::v1::Rows::MergeFrom(
        from._internal_rows());
  }
+  if (from._internal_has_partition_rule_version()) {
+    _this->_internal_mutable_partition_rule_version()->::greptime::v1::PartitionRuleVersion::MergeFrom(
+        from._internal_partition_rule_version());
+  }
  if (from._internal_region_id() != 0) {
    _this->_internal_set_region_id(from._internal_region_id());
  }
@@ -936,6 +936,7 @@ class InsertRequest final :

  enum : int {
    kRowsFieldNumber = 2,
+    kPartitionRuleVersionFieldNumber = 3,
    kRegionIdFieldNumber = 1,
  };
  // .greptime.v1.Rows rows = 2;
@@ -956,6 +957,24 @@ class InsertRequest final :
      ::greptime::v1::Rows* rows);
  ::greptime::v1::Rows* unsafe_arena_release_rows();

+  // .greptime.v1.PartitionRuleVersion partition_rule_version = 3;
+  bool has_partition_rule_version() const;
+  private:
+  bool _internal_has_partition_rule_version() const;
+  public:
+  void clear_partition_rule_version();
+  const ::greptime::v1::PartitionRuleVersion& partition_rule_version() const;
+  PROTOBUF_NODISCARD ::greptime::v1::PartitionRuleVersion* release_partition_rule_version();
+  ::greptime::v1::PartitionRuleVersion* mutable_partition_rule_version();
+  void set_allocated_partition_rule_version(::greptime::v1::PartitionRuleVersion* partition_rule_version);
+  private:
+  const ::greptime::v1::PartitionRuleVersion& _internal_partition_rule_version() const;
+  ::greptime::v1::PartitionRuleVersion* _internal_mutable_partition_rule_version();
+  public:
+  void unsafe_arena_set_allocated_partition_rule_version(
+      ::greptime::v1::PartitionRuleVersion* partition_rule_version);
+  ::greptime::v1::PartitionRuleVersion* unsafe_arena_release_partition_rule_version();
+
  // uint64 region_id = 1;
  void clear_region_id();
  uint64_t region_id() const;
@@ -974,6 +993,7 @@ class InsertRequest final :
  typedef void DestructorSkippable_;
  struct Impl_ {
    ::greptime::v1::Rows* rows_;
+    ::greptime::v1::PartitionRuleVersion* partition_rule_version_;
    uint64_t region_id_;
    mutable ::PROTOBUF_NAMESPACE_ID::internal::CachedSize _cached_size_;
  };
@@ -2512,6 +2532,91 @@ inline void InsertRequest::set_allocated_rows(::greptime::v1::Rows* rows) {
  // @@protoc_insertion_point(field_set_allocated:greptime.v1.flow.InsertRequest.rows)
 }

+// .greptime.v1.PartitionRuleVersion partition_rule_version = 3;
+inline bool InsertRequest::_internal_has_partition_rule_version() const {
+  return this != internal_default_instance() && _impl_.partition_rule_version_ != nullptr;
+}
+inline bool InsertRequest::has_partition_rule_version() const {
+  return _internal_has_partition_rule_version();
+}
+inline const ::greptime::v1::PartitionRuleVersion& InsertRequest::_internal_partition_rule_version() const {
+  const ::greptime::v1::PartitionRuleVersion* p = _impl_.partition_rule_version_;
+  return p != nullptr ? *p : reinterpret_cast<const ::greptime::v1::PartitionRuleVersion&>(
+      ::greptime::v1::_PartitionRuleVersion_default_instance_);
+}
+inline const ::greptime::v1::PartitionRuleVersion& InsertRequest::partition_rule_version() const {
+  // @@protoc_insertion_point(field_get:greptime.v1.flow.InsertRequest.partition_rule_version)
+  return _internal_partition_rule_version();
+}
+inline void InsertRequest::unsafe_arena_set_allocated_partition_rule_version(
+    ::greptime::v1::PartitionRuleVersion* partition_rule_version) {
+  if (GetArenaForAllocation() == nullptr) {
+    delete reinterpret_cast<::PROTOBUF_NAMESPACE_ID::MessageLite*>(_impl_.partition_rule_version_);
+  }
+  _impl_.partition_rule_version_ = partition_rule_version;
+  if (partition_rule_version) {
+    
+  } else {
+    
+  }
+  // @@protoc_insertion_point(field_unsafe_arena_set_allocated:greptime.v1.flow.InsertRequest.partition_rule_version)
+}
+inline ::greptime::v1::PartitionRuleVersion* InsertRequest::release_partition_rule_version() {
+  
+  ::greptime::v1::PartitionRuleVersion* temp = _impl_.partition_rule_version_;
+  _impl_.partition_rule_version_ = nullptr;
+#ifdef PROTOBUF_FORCE_COPY_IN_RELEASE
+  auto* old =  reinterpret_cast<::PROTOBUF_NAMESPACE_ID::MessageLite*>(temp);
+  temp = ::PROTOBUF_NAMESPACE_ID::internal::DuplicateIfNonNull(temp);
+  if (GetArenaForAllocation() == nullptr) { delete old; }
+#else  // PROTOBUF_FORCE_COPY_IN_RELEASE
+  if (GetArenaForAllocation() != nullptr) {
+    temp = ::PROTOBUF_NAMESPACE_ID::internal::DuplicateIfNonNull(temp);
+  }
+#endif  // !PROTOBUF_FORCE_COPY_IN_RELEASE
+  return temp;
+}
+inline ::greptime::v1::PartitionRuleVersion* InsertRequest::unsafe_arena_release_partition_rule_version() {
+  // @@protoc_insertion_point(field_release:greptime.v1.flow.InsertRequest.partition_rule_version)
+  
+  ::greptime::v1::PartitionRuleVersion* temp = _impl_.partition_rule_version_;
+  _impl_.partition_rule_version_ = nullptr;
+  return temp;
+}
+inline ::greptime::v1::PartitionRuleVersion* InsertRequest::_internal_mutable_partition_rule_version() {
+  
+  if (_impl_.partition_rule_version_ == nullptr) {
+    auto* p = CreateMaybeMessage<::greptime::v1::PartitionRuleVersion>(GetArenaForAllocation());
+    _impl_.partition_rule_version_ = p;
+  }
+  return _impl_.partition_rule_version_;
+}
+inline ::greptime::v1::PartitionRuleVersion* InsertRequest::mutable_partition_rule_version() {
+  ::greptime::v1::PartitionRuleVersion* _msg = _internal_mutable_partition_rule_version();
+  // @@protoc_insertion_point(field_mutable:greptime.v1.flow.InsertRequest.partition_rule_version)
+  return _msg;
+}
+inline void InsertRequest::set_allocated_partition_rule_version(::greptime::v1::PartitionRuleVersion* partition_rule_version) {
+  ::PROTOBUF_NAMESPACE_ID::Arena* message_arena = GetArenaForAllocation();
+  if (message_arena == nullptr) {
+    delete reinterpret_cast< ::PROTOBUF_NAMESPACE_ID::MessageLite*>(_impl_.partition_rule_version_);
+  }
+  if (partition_rule_version) {
+    ::PROTOBUF_NAMESPACE_ID::Arena* submessage_arena =
+        ::PROTOBUF_NAMESPACE_ID::Arena::InternalGetOwningArena(
+                reinterpret_cast<::PROTOBUF_NAMESPACE_ID::MessageLite*>(partition_rule_version));
+    if (message_arena != submessage_arena) {
+      partition_rule_version = ::PROTOBUF_NAMESPACE_ID::internal::GetOwnedMessage(
+          message_arena, partition_rule_version, submessage_arena);
+    }
+    
+  } else {
+    
+  }
+  _impl_.partition_rule_version_ = partition_rule_version;
+  // @@protoc_insertion_point(field_set_allocated:greptime.v1.flow.InsertRequest.partition_rule_version)
+}
+
 // -------------------------------------------------------------------

 // FlowRequest