Add GetReplicateInfo interface (#501)

Signed-off-by: yhmo <yihua.mo@zilliz.com>

Author: yhmo

src/impl/MilvusClientV2Impl.cpp

@@ -2458,6 +2458,27 @@ MilvusClientV2Impl::UpdateReplicateConfiguration(const UpdateReplicateConfigurat
         pre, &MilvusConnection::UpdateReplicateConfiguration, nullptr);
 }
 
+Status
+MilvusClientV2Impl::GetReplicateInfo(const GetReplicateInfoRequest& request, GetReplicateInfoResponse& response) {
+    auto pre = [&request](proto::milvus::GetReplicateInfoRequest& rpc_request) {
+        rpc_request.set_source_cluster_id(request.SourceClusterID());
+        rpc_request.set_target_pchannel(request.TargetPChannel());
+        return Status::OK();
+    };
+
+    auto post = [&response](const proto::milvus::GetReplicateInfoResponse& rpc_response) {
+        ReplicateCheckpoint checkpoint;
+        if (rpc_response.has_checkpoint()) {
+            ConvertReplicateCheckpoint(rpc_response.checkpoint(), checkpoint);
+        }
+        response.SetCheckpoint(std::move(checkpoint));
+        return Status::OK();
+    };
+
+    return connection_.Invoke<proto::milvus::GetReplicateInfoRequest, proto::milvus::GetReplicateInfoResponse>(
+        pre, &MilvusConnection::GetReplicateInfo, post);
+}
+
 Status
 MilvusClientV2Impl::CreateResourceGroup(const CreateResourceGroupRequest& request) {
     auto pre = [&request](proto::milvus::CreateResourceGroupRequest& rpc_request) {

src/impl/MilvusClientV2Impl.h

@@ -265,6 +265,9 @@ class MilvusClientV2Impl : public MilvusClientV2, public std::enable_shared_from
     Status
     UpdateReplicateConfiguration(const UpdateReplicateConfigurationRequest& request) final;
 
+    Status
+    GetReplicateInfo(const GetReplicateInfoRequest& request, GetReplicateInfoResponse& response) final;
+
     Status
     CreateResourceGroup(const CreateResourceGroupRequest& request) final;
 

src/impl/MilvusConnection.cpp

@@ -73,6 +73,12 @@ MilvusConnection::StatusByProtoResponse(const proto::common::Status& status) {
     return Status::OK();
 }
 
+Status
+MilvusConnection::StatusByProtoResponse(const proto::milvus::GetReplicateInfoResponse& response) {
+    (void)response;
+    return Status::OK();
+}
+
 template <typename Response>
 Status
 MilvusConnection::StatusByProtoResponse(const Response& response) {
@@ -670,6 +676,13 @@ MilvusConnection::UpdateReplicateConfiguration(const proto::milvus::UpdateReplic
     return grpcCall("UpdateReplicateConfiguration", &Stub::UpdateReplicateConfiguration, request, response, options);
 }
 
+Status
+MilvusConnection::GetReplicateInfo(const proto::milvus::GetReplicateInfoRequest& request,
+                                    proto::milvus::GetReplicateInfoResponse& response,
+                                    const GrpcContextOptions& options) {
+    return grpcCall("GetReplicateInfo", &Stub::GetReplicateInfo, request, response, options);
+}
+
 Status
 MilvusConnection::SelectUser(const proto::milvus::SelectUserRequest& request,
                              proto::milvus::SelectUserResponse& response, const GrpcContextOptions& options) {

src/impl/MilvusConnection.h

@@ -373,6 +373,10 @@ class MilvusConnection {
     UpdateReplicateConfiguration(const proto::milvus::UpdateReplicateConfigurationRequest& request,
                                  proto::common::Status& response, const GrpcContextOptions& options);
 
+    Status
+    GetReplicateInfo(const proto::milvus::GetReplicateInfoRequest& request,
+                     proto::milvus::GetReplicateInfoResponse& response, const GrpcContextOptions& options);
+
     Status
     SelectUser(const proto::milvus::SelectUserRequest& request, proto::milvus::SelectUserResponse& response,
                const GrpcContextOptions& options);
@@ -426,6 +430,9 @@ class MilvusConnection {
     static Status
     StatusByProtoResponse(const proto::common::Status& status);
 
+    static Status
+    StatusByProtoResponse(const proto::milvus::GetReplicateInfoResponse& response);
+
     template <typename Response>
     static Status
     StatusByProtoResponse(const Response& response);

src/impl/request/cdc/GetReplicateInfoRequest.cpp

@@ -0,0 +1,53 @@
+// Licensed to the LF AI & Data foundation under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+#include "milvus/request/cdc/GetReplicateInfoRequest.h"
+
+namespace milvus {
+
+const std::string&
+GetReplicateInfoRequest::SourceClusterID() const {
+    return source_cluster_id_;
+}
+
+void
+GetReplicateInfoRequest::SetSourceClusterID(const std::string& source_cluster_id) {
+    source_cluster_id_ = source_cluster_id;
+}
+
+GetReplicateInfoRequest&
+GetReplicateInfoRequest::WithSourceClusterID(const std::string& source_cluster_id) {
+    SetSourceClusterID(source_cluster_id);
+    return *this;
+}
+
+const std::string&
+GetReplicateInfoRequest::TargetPChannel() const {
+    return target_pchannel_;
+}
+
+void
+GetReplicateInfoRequest::SetTargetPChannel(const std::string& target_pchannel) {
+    target_pchannel_ = target_pchannel;
+}
+
+GetReplicateInfoRequest&
+GetReplicateInfoRequest::WithTargetPChannel(const std::string& target_pchannel) {
+    SetTargetPChannel(target_pchannel);
+    return *this;
+}
+
+}  // namespace milvus

src/impl/response/cdc/GetReplicateInfoResponse.cpp

@@ -0,0 +1,33 @@
+// Licensed to the LF AI & Data foundation under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+#include "milvus/response/cdc/GetReplicateInfoResponse.h"
+
+#include <utility>
+
+namespace milvus {
+
+const ReplicateCheckpoint&
+GetReplicateInfoResponse::Checkpoint() const {
+    return checkpoint_;
+}
+
+void
+GetReplicateInfoResponse::SetCheckpoint(ReplicateCheckpoint&& checkpoint) {
+    checkpoint_ = std::move(checkpoint);
+}
+
+}  // namespace milvus

src/impl/types/ReplicateConfiguration.cpp

@@ -122,6 +122,102 @@ CrossClusterTopology::WithTargetClusterID(const std::string& cluster_id) {
     return *this;
 }
 
+const std::string&
+ReplicateMessageID::ID() const {
+    return id_;
+}
+
+void
+ReplicateMessageID::SetID(const std::string& id) {
+    id_ = id;
+}
+
+ReplicateMessageID&
+ReplicateMessageID::WithID(const std::string& id) {
+    SetID(id);
+    return *this;
+}
+
+const std::string&
+ReplicateMessageID::WalName() const {
+    return wal_name_;
+}
+
+void
+ReplicateMessageID::SetWalName(const std::string& wal_name) {
+    wal_name_ = wal_name;
+}
+
+ReplicateMessageID&
+ReplicateMessageID::WithWalName(const std::string& wal_name) {
+    SetWalName(wal_name);
+    return *this;
+}
+
+const std::string&
+ReplicateCheckpoint::ClusterID() const {
+    return cluster_id_;
+}
+
+void
+ReplicateCheckpoint::SetClusterID(const std::string& cluster_id) {
+    cluster_id_ = cluster_id;
+}
+
+ReplicateCheckpoint&
+ReplicateCheckpoint::WithClusterID(const std::string& cluster_id) {
+    SetClusterID(cluster_id);
+    return *this;
+}
+
+const std::string&
+ReplicateCheckpoint::PChannel() const {
+    return pchannel_;
+}
+
+void
+ReplicateCheckpoint::SetPChannel(const std::string& pchannel) {
+    pchannel_ = pchannel;
+}
+
+ReplicateCheckpoint&
+ReplicateCheckpoint::WithPChannel(const std::string& pchannel) {
+    SetPChannel(pchannel);
+    return *this;
+}
+
+const ReplicateMessageID&
+ReplicateCheckpoint::MessageID() const {
+    return message_id_;
+}
+
+void
+ReplicateCheckpoint::SetMessageID(ReplicateMessageID&& message_id) {
+    message_id_ = std::move(message_id);
+}
+
+ReplicateCheckpoint&
+ReplicateCheckpoint::WithMessageID(ReplicateMessageID&& message_id) {
+    SetMessageID(std::move(message_id));
+    return *this;
+}
+
+uint64_t
+ReplicateCheckpoint::TimeTick() const {
+    return time_tick_;
+}
+
+void
+ReplicateCheckpoint::SetTimeTick(uint64_t time_tick) {
+    time_tick_ = time_tick;
+}
+
+ReplicateCheckpoint&
+ReplicateCheckpoint::WithTimeTick(uint64_t time_tick) {
+    SetTimeTick(time_tick);
+    return *this;
+}
+
 const std::vector<MilvusCluster>&
 ReplicateConfiguration::Clusters() const {
     return clusters_;

src/impl/utils/TypeUtils.cpp

@@ -886,6 +886,21 @@ ConvertReplicateConfiguration(const proto::common::ReplicateConfiguration& rpc_c
     configuration.SetCrossClusterTopologies(std::move(topologies));
 }
 
+void
+ConvertReplicateCheckpoint(const proto::common::ReplicateCheckpoint& rpc_checkpoint,
+                           ReplicateCheckpoint& checkpoint) {
+    checkpoint.SetClusterID(rpc_checkpoint.cluster_id());
+    checkpoint.SetPChannel(rpc_checkpoint.pchannel());
+    checkpoint.SetTimeTick(rpc_checkpoint.time_tick());
+
+    ReplicateMessageID message_id;
+    if (rpc_checkpoint.has_message_id()) {
+        message_id.SetID(rpc_checkpoint.message_id().id());
+        message_id.SetWalName(proto::common::WALName_Name(rpc_checkpoint.message_id().wal_name()));
+    }
+    checkpoint.SetMessageID(std::move(message_id));
+}
+
 bool
 IsValidTemplate(const nlohmann::json& filter_template) {
     return filter_template.is_boolean() || filter_template.is_number() || filter_template.is_string();

src/impl/utils/TypeUtils.h

@@ -136,6 +136,10 @@ void
 ConvertReplicateConfiguration(const proto::common::ReplicateConfiguration& rpc_config,
                               ReplicateConfiguration& configuration);
 
+void
+ConvertReplicateCheckpoint(const proto::common::ReplicateCheckpoint& rpc_checkpoint,
+                           ReplicateCheckpoint& checkpoint);
+
 bool
 IsValidTemplate(const nlohmann::json& filter_template);
 

src/include/milvus/MilvusClientV2.h

@@ -47,6 +47,7 @@
 #include "request/collection/RenameCollectionRequest.h"
 #include "request/collection/TruncateCollectionRequest.h"
 #include "request/cdc/GetReplicateConfigurationRequest.h"
+#include "request/cdc/GetReplicateInfoRequest.h"
 #include "request/cdc/UpdateReplicateConfigurationRequest.h"
 #include "request/database/AlterDatabasePropertiesRequest.h"
 #include "request/database/CreateDatabaseRequest.h"
@@ -107,6 +108,7 @@
 #include "response/alias/DescribeAliasResponse.h"
 #include "response/alias/ListAliasesResponse.h"
 #include "response/cdc/GetReplicateConfigurationResponse.h"
+#include "response/cdc/GetReplicateInfoResponse.h"
 #include "response/collection/BatchDescribeCollectionsResponse.h"
 #include "response/collection/DescribeCollectionResponse.h"
 #include "response/collection/DescribeReplicasResponse.h"
@@ -932,6 +934,16 @@ class MILVUS_SDK_API MilvusClientV2 {
     virtual Status
     UpdateReplicateConfiguration(const UpdateReplicateConfigurationRequest& request) = 0;
 
+    /**
+     * @brief Get replicate information for a channel.
+     *
+     * @param [in] request input parameters
+     * @param [out] response output results
+     * @return Status operation successfully or not
+     */
+    virtual Status
+    GetReplicateInfo(const GetReplicateInfoRequest& request, GetReplicateInfoResponse& response) = 0;
+
     /**
      * @brief Create a resource group. A resource group to physically isolate certain query nodes from others.
      * Read the doc for more info: https://milvus.io/docs/resource_group.md#Manage-Resource-Groups