diff --git a/src/impl/BulkImport.cpp b/src/impl/BulkImport.cpp index 1fe725ac..064f6dd4 100644 --- a/src/impl/BulkImport.cpp +++ b/src/impl/BulkImport.cpp @@ -21,23 +21,100 @@ namespace milvus { namespace { -nlohmann::json -PostImportRequest(const std::string& url, const std::string& request_path, const std::string& api_key, - const nlohmann::json& request_payload) { - httplib::Client client(url); - httplib::Headers headers = { - {"Authorization", "Bearer " + api_key}, - }; +Status +PostImport(const std::string& url, const std::string& request_path, const std::string& api_key, + const nlohmann::json& request_payload, const std::string& db_name, const BulkImportConfig& config, + BulkImportResponse& response) { + response.SetRawJson(nlohmann::json::object()); + try { + httplib::Client client(url); + client.enable_server_certificate_verification(config.VerifyServerCert()); + if (!config.CaCertPath().empty()) { + client.set_ca_cert_path(config.CaCertPath()); + } + if (config.Timeout() > 0) { + client.set_connection_timeout(config.Timeout()); + client.set_read_timeout(config.Timeout(), 0); + client.set_write_timeout(config.Timeout(), 0); + } + httplib::Headers headers = { + {"Authorization", "Bearer " + api_key}, + }; + if (!db_name.empty()) { + headers.emplace("DB-Name", db_name); + } + + auto result = client.Post(request_path, headers, request_payload.dump(), "application/json"); + if (!result) { + return {StatusCode::RPC_FAILED, "failed to post import request: " + httplib::to_string(result.error())}; + } + if (result->status != 200) { + return {StatusCode::SERVER_FAILED, + "import request failed with HTTP status " + std::to_string(result->status) + ": " + result->body}; + } + try { + response.SetRawJson(nlohmann::json::parse(result->body)); + } catch (const std::exception& e) { + return {StatusCode::JSON_PARSE_ERROR, std::string("failed to parse import response: ") + e.what()}; + } + } catch (const std::exception& e) { + return {StatusCode::UNKNOWN_ERROR, std::string("failed to post import request: ") + e.what()}; + } + return Status::OK(); +} - auto response = client.Post(request_path, headers, request_payload.dump(), "application/json"); - if (response && response->status == 200) { - return nlohmann::json::parse(response->body); +Status +CheckImportResponse(const BulkImportResponse& response) { + if (response.Code() != 0) { + return {StatusCode::SERVER_FAILED, + "import request failed with code " + std::to_string(response.Code()) + ": " + response.Message()}; } - return nullptr; + return Status::OK(); } } // namespace +Status +BulkImport::CreateImportJobsImpl(const std::string& url, const std::string& api_key, + const nlohmann::json& request_payload, const BulkImportConfig& config, + BulkImportResponse& response) { + auto status = PostImport(url, "/v2/vectordb/jobs/import/create", api_key, request_payload, "", config, response); + return status.IsOk() ? CheckImportResponse(response) : status; +} + +Status +BulkImport::ListImportJobsImpl(const std::string& url, const std::string& api_key, + const nlohmann::json& request_payload, const BulkImportConfig& config, + BulkImportResponse& response) { + auto status = PostImport(url, "/v2/vectordb/jobs/import/list", api_key, request_payload, "", config, response); + return status.IsOk() ? CheckImportResponse(response) : status; +} + +Status +BulkImport::GetImportJobProgressImpl(const std::string& url, const std::string& api_key, + const nlohmann::json& request_payload, const std::string& db_name, + const BulkImportConfig& config, BulkImportResponse& response) { + auto status = + PostImport(url, "/v2/vectordb/jobs/import/describe", api_key, request_payload, db_name, config, response); + return status.IsOk() ? CheckImportResponse(response) : status; +} + +Status +BulkImport::CommitImportImpl(const std::string& url, const std::string& api_key, const nlohmann::json& request_payload, + const std::string& db_name, const BulkImportConfig& config, BulkImportResponse& response) { + auto status = + PostImport(url, "/v2/vectordb/jobs/import/commit", api_key, request_payload, db_name, config, response); + return status.IsOk() ? CheckImportResponse(response) : status; +} + +Status +BulkImport::AbortImportImpl(const std::string& url, const std::string& api_key, const nlohmann::json& request_payload, + const std::string& db_name, const BulkImportConfig& config, BulkImportResponse& response) { + auto status = + PostImport(url, "/v2/vectordb/jobs/import/abort", api_key, request_payload, db_name, config, response); + return status.IsOk() ? CheckImportResponse(response) : status; +} + nlohmann::json BulkImport::CreateImportJobs(const std::string& url, const std::string& collection_name, const std::vector& files, const std::string& db_name, @@ -56,7 +133,10 @@ BulkImport::CreateImportJobs(const std::string& url, const std::string& collecti if (!options.empty()) { request_payload["options"] = options; } - return PostImportRequest(url, "/v2/vectordb/jobs/import/create", api_key, request_payload); + BulkImportResponse response; + auto status = + PostImport(url, "/v2/vectordb/jobs/import/create", api_key, request_payload, "", BulkImportConfig{}, response); + return status.IsOk() ? response.RawJson() : nlohmann::json{}; } nlohmann::json @@ -66,28 +146,40 @@ BulkImport::ListImportJobs(const std::string& url, const std::string& collection {"collectionName", collection_name}, {"dbName", db_name}, }; - return PostImportRequest(url, "/v2/vectordb/jobs/import/list", api_key, request_payload); + BulkImportResponse response; + auto status = + PostImport(url, "/v2/vectordb/jobs/import/list", api_key, request_payload, "", BulkImportConfig{}, response); + return status.IsOk() ? response.RawJson() : nlohmann::json{}; } nlohmann::json BulkImport::GetImportJobProgress(const std::string& url, const std::string& job_id, const std::string& db_name, const std::string& api_key) { - nlohmann::json payload = {{"dbName", db_name}, {"jobID", job_id}}; - return PostImportRequest(url, "/v2/vectordb/jobs/import/get_progress", api_key, payload); + nlohmann::json payload = {{"dbName", db_name}, {"jobId", job_id}}; + BulkImportResponse response; + auto status = PostImport(url, "/v2/vectordb/jobs/import/get_progress", api_key, payload, db_name, + BulkImportConfig{}, response); + return status.IsOk() ? response.RawJson() : nlohmann::json{}; } nlohmann::json BulkImport::CommitImport(const std::string& url, const std::string& job_id, const std::string& db_name, const std::string& api_key) { nlohmann::json payload = {{"dbName", db_name}, {"jobId", job_id}}; - return PostImportRequest(url, "/v2/vectordb/jobs/import/commit", api_key, payload); + BulkImportResponse response; + auto status = + PostImport(url, "/v2/vectordb/jobs/import/commit", api_key, payload, db_name, BulkImportConfig{}, response); + return status.IsOk() ? response.RawJson() : nlohmann::json{}; } nlohmann::json BulkImport::AbortImport(const std::string& url, const std::string& job_id, const std::string& db_name, const std::string& api_key) { nlohmann::json payload = {{"dbName", db_name}, {"jobId", job_id}}; - return PostImportRequest(url, "/v2/vectordb/jobs/import/abort", api_key, payload); + BulkImportResponse response; + auto status = + PostImport(url, "/v2/vectordb/jobs/import/abort", api_key, payload, db_name, BulkImportConfig{}, response); + return status.IsOk() ? response.RawJson() : nlohmann::json{}; } } // namespace milvus diff --git a/src/impl/request/import/CloudDescribeImportRequest.cpp b/src/impl/request/import/CloudDescribeImportRequest.cpp new file mode 100644 index 00000000..bb34415b --- /dev/null +++ b/src/impl/request/import/CloudDescribeImportRequest.cpp @@ -0,0 +1,101 @@ +// 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/import/CloudDescribeImportRequest.h" + +namespace milvus { + +const std::string& +CloudDescribeImportRequest::ClusterId() const { + return cluster_id_; +} + +void +CloudDescribeImportRequest::SetClusterId(const std::string& cluster_id) { + cluster_id_ = cluster_id; +} + +CloudDescribeImportRequest& +CloudDescribeImportRequest::WithClusterId(const std::string& cluster_id) { + SetClusterId(cluster_id); + return *this; +} + +const std::string& +CloudDescribeImportRequest::ProjectId() const { + return project_id_; +} + +void +CloudDescribeImportRequest::SetProjectId(const std::string& project_id) { + project_id_ = project_id; +} + +CloudDescribeImportRequest& +CloudDescribeImportRequest::WithProjectId(const std::string& project_id) { + SetProjectId(project_id); + return *this; +} + +const std::string& +CloudDescribeImportRequest::RegionId() const { + return region_id_; +} + +void +CloudDescribeImportRequest::SetRegionId(const std::string& region_id) { + region_id_ = region_id; +} + +CloudDescribeImportRequest& +CloudDescribeImportRequest::WithRegionId(const std::string& region_id) { + SetRegionId(region_id); + return *this; +} + +const std::string& +CloudDescribeImportRequest::JobId() const { + return job_id_; +} + +void +CloudDescribeImportRequest::SetJobId(const std::string& job_id) { + job_id_ = job_id; +} + +CloudDescribeImportRequest& +CloudDescribeImportRequest::WithJobId(const std::string& job_id) { + SetJobId(job_id); + return *this; +} + +nlohmann::json +CloudDescribeImportRequest::ToJson() const { + nlohmann::json payload = nlohmann::json::object(); + if (!cluster_id_.empty()) { + payload["clusterId"] = cluster_id_; + } + if (!project_id_.empty()) { + payload["projectId"] = project_id_; + } + if (!region_id_.empty()) { + payload["regionId"] = region_id_; + } + payload["jobId"] = job_id_; + return payload; +} + +} // namespace milvus diff --git a/src/impl/request/import/CloudImportRequest.cpp b/src/impl/request/import/CloudImportRequest.cpp new file mode 100644 index 00000000..4fcf1661 --- /dev/null +++ b/src/impl/request/import/CloudImportRequest.cpp @@ -0,0 +1,188 @@ +// 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/import/CloudImportRequest.h" + +#include + +namespace milvus { + +const std::string& +CloudImportRequest::ClusterId() const { + return cluster_id_; +} + +void +CloudImportRequest::SetClusterId(const std::string& cluster_id) { + cluster_id_ = cluster_id; +} + +CloudImportRequest& +CloudImportRequest::WithClusterId(const std::string& cluster_id) { + SetClusterId(cluster_id); + return *this; +} + +const std::string& +CloudImportRequest::ProjectId() const { + return project_id_; +} + +void +CloudImportRequest::SetProjectId(const std::string& project_id) { + project_id_ = project_id; +} + +CloudImportRequest& +CloudImportRequest::WithProjectId(const std::string& project_id) { + SetProjectId(project_id); + return *this; +} + +const std::string& +CloudImportRequest::RegionId() const { + return region_id_; +} + +void +CloudImportRequest::SetRegionId(const std::string& region_id) { + region_id_ = region_id; +} + +CloudImportRequest& +CloudImportRequest::WithRegionId(const std::string& region_id) { + SetRegionId(region_id); + return *this; +} + +const std::vector>& +CloudImportRequest::ObjectUrls() const { + return object_urls_; +} + +void +CloudImportRequest::SetObjectUrls(std::vector>&& object_urls) { + object_urls_ = std::move(object_urls); +} + +CloudImportRequest& +CloudImportRequest::WithObjectUrls(std::vector>&& object_urls) { + SetObjectUrls(std::move(object_urls)); + return *this; +} + +const std::string& +CloudImportRequest::ObjectUrl() const { + return object_url_; +} + +void +CloudImportRequest::SetObjectUrl(const std::string& object_url) { + object_url_ = object_url; +} + +CloudImportRequest& +CloudImportRequest::WithObjectUrl(const std::string& object_url) { + SetObjectUrl(object_url); + return *this; +} + +const std::string& +CloudImportRequest::AccessKey() const { + return access_key_; +} + +void +CloudImportRequest::SetAccessKey(const std::string& access_key) { + access_key_ = access_key; +} + +CloudImportRequest& +CloudImportRequest::WithAccessKey(const std::string& access_key) { + SetAccessKey(access_key); + return *this; +} + +const std::string& +CloudImportRequest::SecretKey() const { + return secret_key_; +} + +void +CloudImportRequest::SetSecretKey(const std::string& secret_key) { + secret_key_ = secret_key; +} + +CloudImportRequest& +CloudImportRequest::WithSecretKey(const std::string& secret_key) { + SetSecretKey(secret_key); + return *this; +} + +const std::string& +CloudImportRequest::Token() const { + return token_; +} + +void +CloudImportRequest::SetToken(const std::string& token) { + token_ = token; +} + +CloudImportRequest& +CloudImportRequest::WithToken(const std::string& token) { + SetToken(token); + return *this; +} + +nlohmann::json +CloudImportRequest::ToJson() const { + auto payload = BaseImportRequest::ToJson(); + if (!cluster_id_.empty()) { + payload["clusterId"] = cluster_id_; + } + if (!project_id_.empty()) { + payload["projectId"] = project_id_; + } + if (!region_id_.empty()) { + payload["regionId"] = region_id_; + } + payload["collectionName"] = collection_name_; + if (!db_name_.empty()) { + payload["dbName"] = db_name_; + } + if (!partition_name_.empty()) { + payload["partitionName"] = partition_name_; + } + if (!object_urls_.empty()) { + payload["objectUrls"] = object_urls_; + } + if (!object_url_.empty()) { + payload["objectUrl"] = object_url_; + } + if (!access_key_.empty()) { + payload["accessKey"] = access_key_; + } + if (!secret_key_.empty()) { + payload["secretKey"] = secret_key_; + } + if (!token_.empty()) { + payload["token"] = token_; + } + return payload; +} + +} // namespace milvus diff --git a/src/impl/request/import/CloudListImportJobsRequest.cpp b/src/impl/request/import/CloudListImportJobsRequest.cpp new file mode 100644 index 00000000..1ac7e304 --- /dev/null +++ b/src/impl/request/import/CloudListImportJobsRequest.cpp @@ -0,0 +1,122 @@ +// 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/import/CloudListImportJobsRequest.h" + +namespace milvus { + +const std::string& +CloudListImportJobsRequest::ClusterId() const { + return cluster_id_; +} + +void +CloudListImportJobsRequest::SetClusterId(const std::string& cluster_id) { + cluster_id_ = cluster_id; +} + +CloudListImportJobsRequest& +CloudListImportJobsRequest::WithClusterId(const std::string& cluster_id) { + SetClusterId(cluster_id); + return *this; +} + +const std::string& +CloudListImportJobsRequest::ProjectId() const { + return project_id_; +} + +void +CloudListImportJobsRequest::SetProjectId(const std::string& project_id) { + project_id_ = project_id; +} + +CloudListImportJobsRequest& +CloudListImportJobsRequest::WithProjectId(const std::string& project_id) { + SetProjectId(project_id); + return *this; +} + +const std::string& +CloudListImportJobsRequest::RegionId() const { + return region_id_; +} + +void +CloudListImportJobsRequest::SetRegionId(const std::string& region_id) { + region_id_ = region_id; +} + +CloudListImportJobsRequest& +CloudListImportJobsRequest::WithRegionId(const std::string& region_id) { + SetRegionId(region_id); + return *this; +} + +int64_t +CloudListImportJobsRequest::PageSize() const { + return page_size_; +} + +void +CloudListImportJobsRequest::SetPageSize(int64_t page_size) { + page_size_ = page_size; +} + +CloudListImportJobsRequest& +CloudListImportJobsRequest::WithPageSize(int64_t page_size) { + SetPageSize(page_size); + return *this; +} + +int64_t +CloudListImportJobsRequest::CurrentPage() const { + return current_page_; +} + +void +CloudListImportJobsRequest::SetCurrentPage(int64_t current_page) { + current_page_ = current_page; +} + +CloudListImportJobsRequest& +CloudListImportJobsRequest::WithCurrentPage(int64_t current_page) { + SetCurrentPage(current_page); + return *this; +} + +nlohmann::json +CloudListImportJobsRequest::ToJson() const { + nlohmann::json payload = nlohmann::json::object(); + if (!cluster_id_.empty()) { + payload["clusterId"] = cluster_id_; + } + if (!project_id_.empty()) { + payload["projectId"] = project_id_; + } + if (!region_id_.empty()) { + payload["regionId"] = region_id_; + } + if (page_size_ != 0) { + payload["pageSize"] = page_size_; + } + if (current_page_ != 0) { + payload["currentPage"] = current_page_; + } + return payload; +} + +} // namespace milvus diff --git a/src/impl/request/import/MilvusDescribeImportRequest.cpp b/src/impl/request/import/MilvusDescribeImportRequest.cpp new file mode 100644 index 00000000..85549be2 --- /dev/null +++ b/src/impl/request/import/MilvusDescribeImportRequest.cpp @@ -0,0 +1,44 @@ +// 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/import/MilvusDescribeImportRequest.h" + +namespace milvus { + +const std::string& +MilvusDescribeImportRequest::JobId() const { + return job_id_; +} + +void +MilvusDescribeImportRequest::SetJobId(const std::string& job_id) { + job_id_ = job_id; +} + +MilvusDescribeImportRequest& +MilvusDescribeImportRequest::WithJobId(const std::string& job_id) { + SetJobId(job_id); + return *this; +} + +nlohmann::json +MilvusDescribeImportRequest::ToJson() const { + nlohmann::json payload = nlohmann::json::object(); + payload["jobId"] = job_id_; + return payload; +} + +} // namespace milvus diff --git a/src/impl/request/import/MilvusImportRequest.cpp b/src/impl/request/import/MilvusImportRequest.cpp new file mode 100644 index 00000000..24f72d36 --- /dev/null +++ b/src/impl/request/import/MilvusImportRequest.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/import/MilvusImportRequest.h" + +#include + +namespace milvus { + +const std::vector>& +MilvusImportRequest::Files() const { + return files_; +} + +void +MilvusImportRequest::SetFiles(std::vector>&& files) { + files_ = std::move(files); +} + +MilvusImportRequest& +MilvusImportRequest::WithFiles(std::vector>&& files) { + SetFiles(std::move(files)); + return *this; +} + +nlohmann::json +MilvusImportRequest::ToJson() const { + auto payload = BaseImportRequest::ToJson(); + payload["collectionName"] = collection_name_; + if (!db_name_.empty()) { + payload["dbName"] = db_name_; + } + if (!partition_name_.empty()) { + payload["partitionName"] = partition_name_; + } + payload["files"] = files_; + return payload; +} + +} // namespace milvus diff --git a/src/impl/request/import/MilvusListImportJobsRequest.cpp b/src/impl/request/import/MilvusListImportJobsRequest.cpp new file mode 100644 index 00000000..7520a54d --- /dev/null +++ b/src/impl/request/import/MilvusListImportJobsRequest.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/request/import/MilvusListImportJobsRequest.h" + +namespace milvus { + +nlohmann::json +MilvusListImportJobsRequest::ToJson() const { + nlohmann::json payload = nlohmann::json::object(); + if (!collection_name_.empty()) { + payload["collectionName"] = collection_name_; + } + if (!db_name_.empty()) { + payload["dbName"] = db_name_; + } + return payload; +} + +} // namespace milvus diff --git a/src/impl/request/import/VolumeImportRequest.cpp b/src/impl/request/import/VolumeImportRequest.cpp new file mode 100644 index 00000000..448621da --- /dev/null +++ b/src/impl/request/import/VolumeImportRequest.cpp @@ -0,0 +1,91 @@ +// 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/import/VolumeImportRequest.h" + +#include + +namespace milvus { + +const std::string& +VolumeImportRequest::ClusterId() const { + return cluster_id_; +} + +void +VolumeImportRequest::SetClusterId(const std::string& cluster_id) { + cluster_id_ = cluster_id; +} + +VolumeImportRequest& +VolumeImportRequest::WithClusterId(const std::string& cluster_id) { + SetClusterId(cluster_id); + return *this; +} + +const std::string& +VolumeImportRequest::VolumeName() const { + return volume_name_; +} + +void +VolumeImportRequest::SetVolumeName(const std::string& volume_name) { + volume_name_ = volume_name; +} + +VolumeImportRequest& +VolumeImportRequest::WithVolumeName(const std::string& volume_name) { + SetVolumeName(volume_name); + return *this; +} + +const std::vector>& +VolumeImportRequest::DataPaths() const { + return data_paths_; +} + +void +VolumeImportRequest::SetDataPaths(std::vector>&& data_paths) { + data_paths_ = std::move(data_paths); +} + +VolumeImportRequest& +VolumeImportRequest::WithDataPaths(std::vector>&& data_paths) { + SetDataPaths(std::move(data_paths)); + return *this; +} + +nlohmann::json +VolumeImportRequest::ToJson() const { + auto payload = BaseImportRequest::ToJson(); + if (!cluster_id_.empty()) { + payload["clusterId"] = cluster_id_; + } + payload["collectionName"] = collection_name_; + if (!db_name_.empty()) { + payload["dbName"] = db_name_; + } + if (!partition_name_.empty()) { + payload["partitionName"] = partition_name_; + } + if (!volume_name_.empty()) { + payload["volumeName"] = volume_name_; + } + payload["dataPaths"] = data_paths_; + return payload; +} + +} // namespace milvus diff --git a/src/impl/response/import/BulkImportResponse.cpp b/src/impl/response/import/BulkImportResponse.cpp new file mode 100644 index 00000000..698af402 --- /dev/null +++ b/src/impl/response/import/BulkImportResponse.cpp @@ -0,0 +1,71 @@ +// 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/import/BulkImportResponse.h" + +#include +#include + +namespace milvus { + +const nlohmann::json& +BulkImportResponse::RawJson() const { + return json_; +} + +void +BulkImportResponse::SetRawJson(nlohmann::json&& json) { + json_ = std::move(json); +} + +int64_t +BulkImportResponse::Code() const { + auto it = json_.find("code"); + if (it != json_.end() && it->is_number_integer()) { + return it->get(); + } + return -1; +} + +std::string +BulkImportResponse::Message() const { + auto it = json_.find("message"); + if (it != json_.end() && it->is_string()) { + return it->get(); + } + return ""; +} + +const nlohmann::json& +BulkImportResponse::Data() const { + static const nlohmann::json empty = nlohmann::json::object(); + auto it = json_.find("data"); + if (it != json_.end() && it->is_object()) { + return *it; + } + return empty; +} + +std::string +BulkImportResponse::JobId() const { + auto it = Data().find("jobId"); + if (it != Data().end() && it->is_string()) { + return it->get(); + } + return ""; +} + +} // namespace milvus diff --git a/src/impl/types/BulkImportConfig.cpp b/src/impl/types/BulkImportConfig.cpp new file mode 100644 index 00000000..b5dafe01 --- /dev/null +++ b/src/impl/types/BulkImportConfig.cpp @@ -0,0 +1,69 @@ +// 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/types/BulkImportConfig.h" + +namespace milvus { + +int64_t +BulkImportConfig::Timeout() const { + return timeout_; +} + +void +BulkImportConfig::SetTimeout(int64_t timeout) { + timeout_ = timeout; +} + +BulkImportConfig& +BulkImportConfig::WithTimeout(int64_t timeout) { + SetTimeout(timeout); + return *this; +} + +bool +BulkImportConfig::VerifyServerCert() const { + return verify_server_cert_; +} + +void +BulkImportConfig::SetVerifyServerCert(bool verify) { + verify_server_cert_ = verify; +} + +BulkImportConfig& +BulkImportConfig::WithVerifyServerCert(bool verify) { + SetVerifyServerCert(verify); + return *this; +} + +const std::string& +BulkImportConfig::CaCertPath() const { + return ca_cert_path_; +} + +void +BulkImportConfig::SetCaCertPath(const std::string& ca_cert_path) { + ca_cert_path_ = ca_cert_path; +} + +BulkImportConfig& +BulkImportConfig::WithCaCertPath(const std::string& ca_cert_path) { + SetCaCertPath(ca_cert_path); + return *this; +} + +} // namespace milvus diff --git a/src/include/milvus/BulkImport.h b/src/include/milvus/BulkImport.h index 5e1f4274..4120ea1a 100644 --- a/src/include/milvus/BulkImport.h +++ b/src/include/milvus/BulkImport.h @@ -21,6 +21,19 @@ #include #include "milvus/Export.h" +#include "milvus/Status.h" +#include "milvus/request/import/BaseDescribeImportRequest.h" +#include "milvus/request/import/BaseImportRequest.h" +#include "milvus/request/import/BaseListImportJobsRequest.h" +#include "milvus/request/import/CloudDescribeImportRequest.h" +#include "milvus/request/import/CloudImportRequest.h" +#include "milvus/request/import/CloudListImportJobsRequest.h" +#include "milvus/request/import/MilvusDescribeImportRequest.h" +#include "milvus/request/import/MilvusImportRequest.h" +#include "milvus/request/import/MilvusListImportJobsRequest.h" +#include "milvus/request/import/VolumeImportRequest.h" +#include "milvus/response/import/BulkImportResponse.h" +#include "milvus/types/BulkImportConfig.h" namespace milvus { @@ -28,39 +41,136 @@ class MILVUS_SDK_API BulkImport { public: /** * @brief Create an import job by restful api. + * @deprecated Use the request/response overload taking a BaseImportRequest instead. */ - static nlohmann::json + [[deprecated("Use the request/response overload instead")]] static nlohmann::json CreateImportJobs(const std::string& url, const std::string& collection_name, const std::vector& files, const std::string& db_name = "default", const std::string& api_key = "", const std::string& partition_name = "", const nlohmann::json& options = nlohmann::json{}); + /** + * @brief Create an import job by restful api. + * @param [in] url the platform endpoint. + * @param [in] request the import request. + * @param [in] config the transport options. + * @param [out] response the response. + */ + template + static Status + CreateImportJobs(const std::string& url, const BaseImportRequest& request, BulkImportResponse& response, + const BulkImportConfig& config = BulkImportConfig{}) { + return CreateImportJobsImpl(url, request.ApiKey(), request.ToJson(), config, response); + } + /** * @brief List all import jobs by restful api. + * @deprecated Use the request/response overload taking a BaseListImportJobsRequest instead. */ - static nlohmann::json + [[deprecated("Use the request/response overload instead")]] static nlohmann::json ListImportJobs(const std::string& url, const std::string& collection_name, const std::string& db_name = "default", const std::string& api_key = ""); + /** + * @brief List all import jobs by restful api. + * @param [in] url the platform endpoint. + * @param [in] request the list import jobs request. + * @param [in] config the transport options. + * @param [out] response the response. + */ + template + static Status + ListImportJobs(const std::string& url, const BaseListImportJobsRequest& request, BulkImportResponse& response, + const BulkImportConfig& config = BulkImportConfig{}) { + return ListImportJobsImpl(url, request.ApiKey(), request.ToJson(), config, response); + } + /** * @brief Get import job progress by restful api. + * @deprecated Use the request/response overload taking a BaseDescribeImportRequest instead. */ - static nlohmann::json + [[deprecated("Use the request/response overload instead")]] static nlohmann::json GetImportJobProgress(const std::string& url, const std::string& job_id, const std::string& db_name = "default", const std::string& api_key = ""); + /** + * @brief Get import job progress by restful api. + * @param [in] url the platform endpoint. + * @param [in] request the describe import request. + * @param [in] config the transport options. + * @param [out] response the response. + */ + template + static Status + GetImportJobProgress(const std::string& url, const BaseDescribeImportRequest& request, + BulkImportResponse& response, const BulkImportConfig& config = BulkImportConfig{}) { + return GetImportJobProgressImpl(url, request.ApiKey(), request.ToJson(), request.DatabaseName(), config, + response); + } + /** * @brief Commit a 2PC import job created with options.auto_commit=false, making its staged imported data visible. + * @deprecated Use the request/response overload taking a BaseDescribeImportRequest instead. */ - static nlohmann::json + [[deprecated("Use the request/response overload instead")]] static nlohmann::json CommitImport(const std::string& url, const std::string& job_id, const std::string& db_name = "default", const std::string& api_key = ""); + /** + * @brief Commit a 2PC import job created with options.auto_commit=false, making its staged imported data visible. + * @param [in] url the platform endpoint. + * @param [in] request the commit import request (a BaseDescribeImportRequest-derived DTO). + * @param [in] config the transport options. + * @param [out] response the response. + */ + template + static Status + CommitImport(const std::string& url, const BaseDescribeImportRequest& request, BulkImportResponse& response, + const BulkImportConfig& config = BulkImportConfig{}) { + return CommitImportImpl(url, request.ApiKey(), request.ToJson(), request.DatabaseName(), config, response); + } + /** * @brief Abort a 2PC import job created with options.auto_commit=false, discarding its staged imported data. + * @deprecated Use the request/response overload taking a BaseDescribeImportRequest instead. */ - static nlohmann::json + [[deprecated("Use the request/response overload instead")]] static nlohmann::json AbortImport(const std::string& url, const std::string& job_id, const std::string& db_name = "default", const std::string& api_key = ""); + + /** + * @brief Abort a 2PC import job created with options.auto_commit=false, discarding its staged imported data. + * @param [in] url the platform endpoint. + * @param [in] request the abort import request (a BaseDescribeImportRequest-derived DTO). + * @param [in] config the transport options. + * @param [out] response the response. + */ + template + static Status + AbortImport(const std::string& url, const BaseDescribeImportRequest& request, BulkImportResponse& response, + const BulkImportConfig& config = BulkImportConfig{}) { + return AbortImportImpl(url, request.ApiKey(), request.ToJson(), request.DatabaseName(), config, response); + } + + private: + static Status + CreateImportJobsImpl(const std::string& url, const std::string& api_key, const nlohmann::json& request_payload, + const BulkImportConfig& config, BulkImportResponse& response); + + static Status + ListImportJobsImpl(const std::string& url, const std::string& api_key, const nlohmann::json& request_payload, + const BulkImportConfig& config, BulkImportResponse& response); + + static Status + GetImportJobProgressImpl(const std::string& url, const std::string& api_key, const nlohmann::json& request_payload, + const std::string& db_name, const BulkImportConfig& config, BulkImportResponse& response); + + static Status + CommitImportImpl(const std::string& url, const std::string& api_key, const nlohmann::json& request_payload, + const std::string& db_name, const BulkImportConfig& config, BulkImportResponse& response); + + static Status + AbortImportImpl(const std::string& url, const std::string& api_key, const nlohmann::json& request_payload, + const std::string& db_name, const BulkImportConfig& config, BulkImportResponse& response); }; } // namespace milvus diff --git a/src/include/milvus/MilvusClientV2.h b/src/include/milvus/MilvusClientV2.h index cd1a566d..dd1774d4 100644 --- a/src/include/milvus/MilvusClientV2.h +++ b/src/include/milvus/MilvusClientV2.h @@ -22,6 +22,7 @@ #include "ClientTelemetry.h" #include "MilvusClientV2Session.h" #include "Status.h" +#include "milvus/BulkImport.h" #include "milvus/Export.h" #include "request/alias/AlterAliasRequest.h" #include "request/alias/CreateAliasRequest.h" diff --git a/src/include/milvus/request/import/BaseDescribeImportRequest.h b/src/include/milvus/request/import/BaseDescribeImportRequest.h new file mode 100644 index 00000000..02ac332a --- /dev/null +++ b/src/include/milvus/request/import/BaseDescribeImportRequest.h @@ -0,0 +1,112 @@ +// 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. + +#pragma once + +#include + +#include "milvus/thirdparty/nlohmann/json.hpp" + +namespace milvus { + +/** + * @brief Base class for describing, committing or aborting an import job. + */ +template +class BaseDescribeImportRequest { + protected: + /** + * @brief Constructor + */ + BaseDescribeImportRequest() = default; + + public: + /** + * @brief Destructor + */ + virtual ~BaseDescribeImportRequest() = default; + + /** + * @brief Get the API key used for the cloud API, or the userName:password for Milvus directly. + * @return the API key. + */ + const std::string& + ApiKey() const { + return api_key_; + } + + /** + * @brief Set the API key. + * @param [in] api_key the API key. + */ + void + SetApiKey(const std::string& api_key) { + api_key_ = api_key; + } + + /** + * @brief Set the API key. + * @param [in] api_key the API key. + */ + T& + WithApiKey(const std::string& api_key) { + SetApiKey(api_key); + return static_cast(*this); + } + + /** + * @brief Get the target database name. + * @return the database name. + */ + const std::string& + DatabaseName() const { + return db_name_; + } + + /** + * @brief Set the target database name. + * @param [in] db_name the database name. + */ + void + SetDatabaseName(const std::string& db_name) { + db_name_ = db_name; + } + + /** + * @brief Set the target database name. + * @param [in] db_name the database name. + */ + T& + WithDatabaseName(const std::string& db_name) { + SetDatabaseName(db_name); + return static_cast(*this); + } + + /** + * @brief Serialize the request into the REST import job payload. + * @return the payload. + */ + virtual nlohmann::json + ToJson() const { + return nlohmann::json::object(); + } + + protected: + std::string api_key_; + std::string db_name_; +}; + +} // namespace milvus diff --git a/src/include/milvus/request/import/BaseImportRequest.h b/src/include/milvus/request/import/BaseImportRequest.h new file mode 100644 index 00000000..2e8aea14 --- /dev/null +++ b/src/include/milvus/request/import/BaseImportRequest.h @@ -0,0 +1,205 @@ +// 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. + +#pragma once + +#include + +#include "milvus/thirdparty/nlohmann/json.hpp" + +namespace milvus { + +/** + * @brief Base class for data import requests. + */ +template +class BaseImportRequest { + protected: + /** + * @brief Constructor + */ + BaseImportRequest() = default; + + public: + /** + * @brief Destructor + */ + virtual ~BaseImportRequest() = default; + + /** + * @brief Get the API key used for the cloud API, or the userName:password for Milvus directly. + * @return the API key. + */ + const std::string& + ApiKey() const { + return api_key_; + } + + /** + * @brief Set the API key. + * @param [in] api_key the API key. + */ + void + SetApiKey(const std::string& api_key) { + api_key_ = api_key; + } + + /** + * @brief Set the API key. + * @param [in] api_key the API key. + */ + T& + WithApiKey(const std::string& api_key) { + SetApiKey(api_key); + return static_cast(*this); + } + + /** + * @brief Get the target database name. + * Note: create/list calls deliver the database in the JSON body, while describe/commit/abort + * calls send it via the DB-Name header. + * @return the database name. + */ + const std::string& + DatabaseName() const { + return db_name_; + } + + /** + * @brief Set the target database name. + * @param [in] db_name the database name. + */ + void + SetDatabaseName(const std::string& db_name) { + db_name_ = db_name; + } + + /** + * @brief Set the target database name. + * @param [in] db_name the database name. + */ + T& + WithDatabaseName(const std::string& db_name) { + SetDatabaseName(db_name); + return static_cast(*this); + } + + /** + * @brief Get the target collection name. + * @return the collection name. + */ + const std::string& + CollectionName() const { + return collection_name_; + } + + /** + * @brief Set the target collection name. + * @param [in] collection_name the collection name. + */ + void + SetCollectionName(const std::string& collection_name) { + collection_name_ = collection_name; + } + + /** + * @brief Set the target collection name. + * @param [in] collection_name the collection name. + */ + T& + WithCollectionName(const std::string& collection_name) { + SetCollectionName(collection_name); + return static_cast(*this); + } + + /** + * @brief Get the target partition name. + * @return the partition name. + */ + const std::string& + PartitionName() const { + return partition_name_; + } + + /** + * @brief Set the target partition name. + * @param [in] partition_name the partition name. + */ + void + SetPartitionName(const std::string& partition_name) { + partition_name_ = partition_name; + } + + /** + * @brief Set the target partition name. + * @param [in] partition_name the partition name. + */ + T& + WithPartitionName(const std::string& partition_name) { + SetPartitionName(partition_name); + return static_cast(*this); + } + + /** + * @brief Get the additional import options. + * @return the options. + */ + const nlohmann::json& + Options() const { + return options_; + } + + /** + * @brief Set the additional import options. + * @param [in] options the options. + */ + void + SetOptions(nlohmann::json&& options) { + options_ = std::move(options); + } + + /** + * @brief Set the additional import options. + * @param [in] options the options. + */ + T& + WithOptions(nlohmann::json&& options) { + SetOptions(std::move(options)); + return static_cast(*this); + } + + /** + * @brief Serialize the request into the REST import job payload. + * @return the payload. + */ + virtual nlohmann::json + ToJson() const { + nlohmann::json payload = nlohmann::json::object(); + if (!options_.empty()) { + payload["options"] = options_; + } + return payload; + } + + protected: + std::string api_key_; + std::string db_name_; + std::string collection_name_; + std::string partition_name_; + nlohmann::json options_; +}; + +} // namespace milvus diff --git a/src/include/milvus/request/import/BaseListImportJobsRequest.h b/src/include/milvus/request/import/BaseListImportJobsRequest.h new file mode 100644 index 00000000..63cbb256 --- /dev/null +++ b/src/include/milvus/request/import/BaseListImportJobsRequest.h @@ -0,0 +1,141 @@ +// 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. + +#pragma once + +#include + +#include "milvus/thirdparty/nlohmann/json.hpp" + +namespace milvus { + +/** + * @brief Base class for listing import jobs. + */ +template +class BaseListImportJobsRequest { + protected: + /** + * @brief Constructor + */ + BaseListImportJobsRequest() = default; + + public: + /** + * @brief Destructor + */ + virtual ~BaseListImportJobsRequest() = default; + + /** + * @brief Get the API key used for the cloud API, or the userName:password for Milvus directly. + * @return the API key. + */ + const std::string& + ApiKey() const { + return api_key_; + } + + /** + * @brief Set the API key. + * @param [in] api_key the API key. + */ + void + SetApiKey(const std::string& api_key) { + api_key_ = api_key; + } + + /** + * @brief Set the API key. + * @param [in] api_key the API key. + */ + T& + WithApiKey(const std::string& api_key) { + SetApiKey(api_key); + return static_cast(*this); + } + + /** + * @brief Get the target collection name. + * @return the collection name. + */ + const std::string& + CollectionName() const { + return collection_name_; + } + + /** + * @brief Set the target collection name. + * @param [in] collection_name the collection name. + */ + void + SetCollectionName(const std::string& collection_name) { + collection_name_ = collection_name; + } + + /** + * @brief Set the target collection name. + * @param [in] collection_name the collection name. + */ + T& + WithCollectionName(const std::string& collection_name) { + SetCollectionName(collection_name); + return static_cast(*this); + } + + /** + * @brief Get the target database name. + * @return the database name. + */ + const std::string& + DatabaseName() const { + return db_name_; + } + + /** + * @brief Set the target database name. + * @param [in] db_name the database name. + */ + void + SetDatabaseName(const std::string& db_name) { + db_name_ = db_name; + } + + /** + * @brief Set the target database name. + * @param [in] db_name the database name. + */ + T& + WithDatabaseName(const std::string& db_name) { + SetDatabaseName(db_name); + return static_cast(*this); + } + + /** + * @brief Serialize the request into the REST import job payload. + * @return the payload. + */ + virtual nlohmann::json + ToJson() const { + return nlohmann::json::object(); + } + + protected: + std::string api_key_; + std::string collection_name_; + std::string db_name_; +}; + +} // namespace milvus diff --git a/src/include/milvus/request/import/CloudDescribeImportRequest.h b/src/include/milvus/request/import/CloudDescribeImportRequest.h new file mode 100644 index 00000000..479f37d4 --- /dev/null +++ b/src/include/milvus/request/import/CloudDescribeImportRequest.h @@ -0,0 +1,134 @@ +// 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. + +#pragma once + +#include + +#include "./BaseDescribeImportRequest.h" +#include "milvus/Export.h" + +namespace milvus { + +/** + * @brief Zilliz Cloud only. Request for describing, committing or aborting an import job on a Zilliz cloud instance. + */ +class MILVUS_SDK_API CloudDescribeImportRequest : public BaseDescribeImportRequest { + public: + /** + * @brief Constructor + */ + CloudDescribeImportRequest() = default; + + /** + * @brief Get the cluster id. + * @return the cluster id. + */ + const std::string& + ClusterId() const; + + /** + * @brief Set the cluster id. + * @param [in] cluster_id the cluster id. + */ + void + SetClusterId(const std::string& cluster_id); + + /** + * @brief Set the cluster id. + * @param [in] cluster_id the cluster id. + */ + CloudDescribeImportRequest& + WithClusterId(const std::string& cluster_id); + + /** + * @brief Get the project id, used for project database deployments. + * @return the project id. + */ + const std::string& + ProjectId() const; + + /** + * @brief Set the project id. + * @param [in] project_id the project id. + */ + void + SetProjectId(const std::string& project_id); + + /** + * @brief Set the project id. + * @param [in] project_id the project id. + */ + CloudDescribeImportRequest& + WithProjectId(const std::string& project_id); + + /** + * @brief Get the region id, used for project database deployments. + * @return the region id. + */ + const std::string& + RegionId() const; + + /** + * @brief Set the region id. + * @param [in] region_id the region id. + */ + void + SetRegionId(const std::string& region_id); + + /** + * @brief Set the region id. + * @param [in] region_id the region id. + */ + CloudDescribeImportRequest& + WithRegionId(const std::string& region_id); + + /** + * @brief Get the id of the import job. + * @return the job id. + */ + const std::string& + JobId() const; + + /** + * @brief Set the id of the import job. + * @param [in] job_id the job id. + */ + void + SetJobId(const std::string& job_id); + + /** + * @brief Set the id of the import job. + * @param [in] job_id the job id. + */ + CloudDescribeImportRequest& + WithJobId(const std::string& job_id); + + /** + * @brief Serialize the request into the REST import job payload. + * @return the payload. + */ + nlohmann::json + ToJson() const override; + + private: + std::string cluster_id_; + std::string project_id_; + std::string region_id_; + std::string job_id_; +}; + +} // namespace milvus diff --git a/src/include/milvus/request/import/CloudImportRequest.h b/src/include/milvus/request/import/CloudImportRequest.h new file mode 100644 index 00000000..e1fb303d --- /dev/null +++ b/src/include/milvus/request/import/CloudImportRequest.h @@ -0,0 +1,226 @@ +// 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. + +#pragma once + +#include +#include + +#include "./BaseImportRequest.h" +#include "milvus/Export.h" + +namespace milvus { + +/** + * @brief Zilliz Cloud only. Request for importing data from a storage bucket into a Zilliz cloud instance. + */ +class MILVUS_SDK_API CloudImportRequest : public BaseImportRequest { + public: + /** + * @brief Constructor + */ + CloudImportRequest() = default; + + /** + * @brief Get the cluster id. + * @return the cluster id. + */ + const std::string& + ClusterId() const; + + /** + * @brief Set the cluster id. + * @param [in] cluster_id the cluster id. + */ + void + SetClusterId(const std::string& cluster_id); + + /** + * @brief Set the cluster id. + * @param [in] cluster_id the cluster id. + */ + CloudImportRequest& + WithClusterId(const std::string& cluster_id); + + /** + * @brief Get the project id, used for project database deployments. + * @return the project id. + */ + const std::string& + ProjectId() const; + + /** + * @brief Set the project id. + * @param [in] project_id the project id. + */ + void + SetProjectId(const std::string& project_id); + + /** + * @brief Set the project id. + * @param [in] project_id the project id. + */ + CloudImportRequest& + WithProjectId(const std::string& project_id); + + /** + * @brief Get the region id, used for project database deployments. + * @return the region id. + */ + const std::string& + RegionId() const; + + /** + * @brief Set the region id. + * @param [in] region_id the region id. + */ + void + SetRegionId(const std::string& region_id); + + /** + * @brief Set the region id. + * @param [in] region_id the region id. + */ + CloudImportRequest& + WithRegionId(const std::string& region_id); + + /** + * @brief Get the object urls to import. + * @return the object urls. + */ + const std::vector>& + ObjectUrls() const; + + /** + * @brief Set the object urls to import. + * @param [in] object_urls the object urls. + */ + void + SetObjectUrls(std::vector>&& object_urls); + + /** + * @brief Set the object urls to import. + * @param [in] object_urls the object urls. + */ + CloudImportRequest& + WithObjectUrls(std::vector>&& object_urls); + + /** + * @brief Get the deprecated single object url. + * @deprecated Use ObjectUrls() instead. + * @return the object url. + */ + [[deprecated("Use ObjectUrls() instead")]] const std::string& + ObjectUrl() const; + + /** + * @brief Set the deprecated single object url. + * @deprecated Use SetObjectUrls() instead. + * @param [in] object_url the object url. + */ + [[deprecated("Use SetObjectUrls() instead")]] void + SetObjectUrl(const std::string& object_url); + + /** + * @brief Set the deprecated single object url. + * @deprecated Use WithObjectUrls() instead. + * @param [in] object_url the object url. + */ + [[deprecated("Use WithObjectUrls() instead")]] CloudImportRequest& + WithObjectUrl(const std::string& object_url); + + /** + * @brief Get the access key for the storage bucket. + * @return the access key. + */ + const std::string& + AccessKey() const; + + /** + * @brief Set the access key. + * @param [in] access_key the access key. + */ + void + SetAccessKey(const std::string& access_key); + + /** + * @brief Set the access key. + * @param [in] access_key the access key. + */ + CloudImportRequest& + WithAccessKey(const std::string& access_key); + + /** + * @brief Get the secret key for the storage bucket. + * @return the secret key. + */ + const std::string& + SecretKey() const; + + /** + * @brief Set the secret key. + * @param [in] secret_key the secret key. + */ + void + SetSecretKey(const std::string& secret_key); + + /** + * @brief Set the secret key. + * @param [in] secret_key the secret key. + */ + CloudImportRequest& + WithSecretKey(const std::string& secret_key); + + /** + * @brief Get the token for short-term credentials. + * @return the token. + */ + const std::string& + Token() const; + + /** + * @brief Set the token for short-term credentials. + * @param [in] token the token. + */ + void + SetToken(const std::string& token); + + /** + * @brief Set the token for short-term credentials. + * @param [in] token the token. + */ + CloudImportRequest& + WithToken(const std::string& token); + + /** + * @brief Serialize the request into the REST import job payload. + * @return the payload. + */ + nlohmann::json + ToJson() const override; + + private: + std::string cluster_id_; + std::string project_id_; + std::string region_id_; + std::vector> object_urls_; + std::string object_url_; + std::string access_key_; + std::string secret_key_; + std::string token_; +}; + +} // namespace milvus diff --git a/src/include/milvus/request/import/CloudListImportJobsRequest.h b/src/include/milvus/request/import/CloudListImportJobsRequest.h new file mode 100644 index 00000000..0e23b1a2 --- /dev/null +++ b/src/include/milvus/request/import/CloudListImportJobsRequest.h @@ -0,0 +1,157 @@ +// 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. + +#pragma once + +#include +#include + +#include "./BaseListImportJobsRequest.h" +#include "milvus/Export.h" + +namespace milvus { + +/** + * @brief Zilliz Cloud only. Request for listing import jobs on a Zilliz cloud instance. + */ +class MILVUS_SDK_API CloudListImportJobsRequest : public BaseListImportJobsRequest { + public: + /** + * @brief Constructor + */ + CloudListImportJobsRequest() = default; + + /** + * @brief Get the cluster id. + * @return the cluster id. + */ + const std::string& + ClusterId() const; + + /** + * @brief Set the cluster id. + * @param [in] cluster_id the cluster id. + */ + void + SetClusterId(const std::string& cluster_id); + + /** + * @brief Set the cluster id. + * @param [in] cluster_id the cluster id. + */ + CloudListImportJobsRequest& + WithClusterId(const std::string& cluster_id); + + /** + * @brief Get the project id, used for project database deployments. + * @return the project id. + */ + const std::string& + ProjectId() const; + + /** + * @brief Set the project id. + * @param [in] project_id the project id. + */ + void + SetProjectId(const std::string& project_id); + + /** + * @brief Set the project id. + * @param [in] project_id the project id. + */ + CloudListImportJobsRequest& + WithProjectId(const std::string& project_id); + + /** + * @brief Get the region id, used for project database deployments. + * @return the region id. + */ + const std::string& + RegionId() const; + + /** + * @brief Set the region id. + * @param [in] region_id the region id. + */ + void + SetRegionId(const std::string& region_id); + + /** + * @brief Set the region id. + * @param [in] region_id the region id. + */ + CloudListImportJobsRequest& + WithRegionId(const std::string& region_id); + + /** + * @brief Get the page size. + * @return the page size. + */ + int64_t + PageSize() const; + + /** + * @brief Set the page size. + * @param [in] page_size the page size. + */ + void + SetPageSize(int64_t page_size); + + /** + * @brief Set the page size. + * @param [in] page_size the page size. + */ + CloudListImportJobsRequest& + WithPageSize(int64_t page_size); + + /** + * @brief Get the current page number. + * @return the current page. + */ + int64_t + CurrentPage() const; + + /** + * @brief Set the current page number. + * @param [in] current_page the current page. + */ + void + SetCurrentPage(int64_t current_page); + + /** + * @brief Set the current page number. + * @param [in] current_page the current page. + */ + CloudListImportJobsRequest& + WithCurrentPage(int64_t current_page); + + /** + * @brief Serialize the request into the REST import job payload. + * @return the payload. + */ + nlohmann::json + ToJson() const override; + + private: + std::string cluster_id_; + std::string project_id_; + std::string region_id_; + int64_t page_size_{0}; + int64_t current_page_{0}; +}; + +} // namespace milvus diff --git a/src/include/milvus/request/import/MilvusDescribeImportRequest.h b/src/include/milvus/request/import/MilvusDescribeImportRequest.h new file mode 100644 index 00000000..659da8cb --- /dev/null +++ b/src/include/milvus/request/import/MilvusDescribeImportRequest.h @@ -0,0 +1,68 @@ +// 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. + +#pragma once + +#include + +#include "./BaseDescribeImportRequest.h" +#include "milvus/Export.h" + +namespace milvus { + +/** + * @brief Request for describing, committing or aborting a bulk import job on a Milvus server. + */ +class MILVUS_SDK_API MilvusDescribeImportRequest : public BaseDescribeImportRequest { + public: + /** + * @brief Constructor + */ + MilvusDescribeImportRequest() = default; + + /** + * @brief Get the id of the import job. + * @return the job id. + */ + const std::string& + JobId() const; + + /** + * @brief Set the id of the import job. + * @param [in] job_id the job id. + */ + void + SetJobId(const std::string& job_id); + + /** + * @brief Set the id of the import job. + * @param [in] job_id the job id. + */ + MilvusDescribeImportRequest& + WithJobId(const std::string& job_id); + + /** + * @brief Serialize the request into the REST import job payload. + * @return the payload. + */ + nlohmann::json + ToJson() const override; + + private: + std::string job_id_; +}; + +} // namespace milvus diff --git a/src/include/milvus/request/import/MilvusImportRequest.h b/src/include/milvus/request/import/MilvusImportRequest.h new file mode 100644 index 00000000..e4d44b51 --- /dev/null +++ b/src/include/milvus/request/import/MilvusImportRequest.h @@ -0,0 +1,69 @@ +// 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. + +#pragma once + +#include +#include + +#include "./BaseImportRequest.h" +#include "milvus/Export.h" + +namespace milvus { + +/** + * @brief Request for importing data into an open-source Milvus instance. + */ +class MILVUS_SDK_API MilvusImportRequest : public BaseImportRequest { + public: + /** + * @brief Constructor + */ + MilvusImportRequest() = default; + + /** + * @brief Get the data files to import. + * @return the files. + */ + const std::vector>& + Files() const; + + /** + * @brief Set the data files to import. + * @param [in] files the files. + */ + void + SetFiles(std::vector>&& files); + + /** + * @brief Set the data files to import. + * @param [in] files the files. + */ + MilvusImportRequest& + WithFiles(std::vector>&& files); + + /** + * @brief Serialize the request into the REST import job payload. + * @return the payload. + */ + nlohmann::json + ToJson() const override; + + private: + std::vector> files_; +}; + +} // namespace milvus diff --git a/src/include/milvus/request/import/MilvusListImportJobsRequest.h b/src/include/milvus/request/import/MilvusListImportJobsRequest.h new file mode 100644 index 00000000..e3145c17 --- /dev/null +++ b/src/include/milvus/request/import/MilvusListImportJobsRequest.h @@ -0,0 +1,44 @@ +// 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. + +#pragma once + +#include + +#include "./BaseListImportJobsRequest.h" +#include "milvus/Export.h" + +namespace milvus { + +/** + * @brief Request for listing bulk import jobs on a Milvus server. + */ +class MILVUS_SDK_API MilvusListImportJobsRequest : public BaseListImportJobsRequest { + public: + /** + * @brief Constructor + */ + MilvusListImportJobsRequest() = default; + + /** + * @brief Serialize the request into the REST import job payload. + * @return the payload. + */ + nlohmann::json + ToJson() const override; +}; + +} // namespace milvus diff --git a/src/include/milvus/request/import/VolumeImportRequest.h b/src/include/milvus/request/import/VolumeImportRequest.h new file mode 100644 index 00000000..b348a813 --- /dev/null +++ b/src/include/milvus/request/import/VolumeImportRequest.h @@ -0,0 +1,113 @@ +// 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. + +#pragma once + +#include +#include + +#include "./BaseImportRequest.h" +#include "milvus/Export.h" + +namespace milvus { + +/** + * @brief Zilliz Cloud only. Request for importing data from a Zilliz volume into a Zilliz cloud instance. + */ +class MILVUS_SDK_API VolumeImportRequest : public BaseImportRequest { + public: + /** + * @brief Constructor + */ + VolumeImportRequest() = default; + + /** + * @brief Get the cluster id. + * @return the cluster id. + */ + const std::string& + ClusterId() const; + + /** + * @brief Set the cluster id. + * @param [in] cluster_id the cluster id. + */ + void + SetClusterId(const std::string& cluster_id); + + /** + * @brief Set the cluster id. + * @param [in] cluster_id the cluster id. + */ + VolumeImportRequest& + WithClusterId(const std::string& cluster_id); + + /** + * @brief Get the volume name. + * @return the volume name. + */ + const std::string& + VolumeName() const; + + /** + * @brief Set the volume name. + * @param [in] volume_name the volume name. + */ + void + SetVolumeName(const std::string& volume_name); + + /** + * @brief Set the volume name. + * @param [in] volume_name the volume name. + */ + VolumeImportRequest& + WithVolumeName(const std::string& volume_name); + + /** + * @brief Get the data paths to import. + * @return the data paths. + */ + const std::vector>& + DataPaths() const; + + /** + * @brief Set the data paths to import. + * @param [in] data_paths the data paths. + */ + void + SetDataPaths(std::vector>&& data_paths); + + /** + * @brief Set the data paths to import. + * @param [in] data_paths the data paths. + */ + VolumeImportRequest& + WithDataPaths(std::vector>&& data_paths); + + /** + * @brief Serialize the request into the REST import job payload. + * @return the payload. + */ + nlohmann::json + ToJson() const override; + + private: + std::string cluster_id_; + std::string volume_name_; + std::vector> data_paths_; +}; + +} // namespace milvus diff --git a/src/include/milvus/response/import/BulkImportResponse.h b/src/include/milvus/response/import/BulkImportResponse.h new file mode 100644 index 00000000..23cf8658 --- /dev/null +++ b/src/include/milvus/response/import/BulkImportResponse.h @@ -0,0 +1,82 @@ +// 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. + +#pragma once + +#include + +#include "milvus/Export.h" +#include "milvus/thirdparty/nlohmann/json.hpp" + +namespace milvus { + +/** + * @brief Response of a bulk import REST API call. + */ +class MILVUS_SDK_API BulkImportResponse { + public: + /** + * @brief Constructor + */ + BulkImportResponse() = default; + + /** + * @brief Get the raw JSON body of the REST response. + * @return the raw JSON. + */ + const nlohmann::json& + RawJson() const; + + /** + * @brief Set the raw JSON body of the REST response. + * @param [in] json the raw JSON. + */ + void + SetRawJson(nlohmann::json&& json); + + /** + * @brief Get the REST envelope code. + * @return the code. + */ + int64_t + Code() const; + + /** + * @brief Get the REST envelope message. + * @return the message. + */ + std::string + Message() const; + + /** + * @brief Get the REST envelope data object. + * @return the data. + */ + const nlohmann::json& + Data() const; + + /** + * @brief Get the id of the created import job, empty for other operations. + * @return the job id. + */ + std::string + JobId() const; + + private: + nlohmann::json json_; +}; + +} // namespace milvus diff --git a/src/include/milvus/types/BulkImportConfig.h b/src/include/milvus/types/BulkImportConfig.h new file mode 100644 index 00000000..da380ccc --- /dev/null +++ b/src/include/milvus/types/BulkImportConfig.h @@ -0,0 +1,105 @@ +// 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. + +#pragma once + +#include + +#include "milvus/Export.h" + +namespace milvus { + +/** + * @brief Transport options for the bulk import REST API calls. + * Client certificates (mutual TLS) are not supported. + */ +class MILVUS_SDK_API BulkImportConfig { + public: + /** + * @brief Constructor + */ + BulkImportConfig() = default; + + /** + * @brief Get the request timeout in seconds, zero keeps the httplib default. + * @return the timeout. + */ + int64_t + Timeout() const; + + /** + * @brief Set the request timeout in seconds. + * @param [in] timeout the timeout in seconds. + */ + void + SetTimeout(int64_t timeout); + + /** + * @brief Set the request timeout in seconds. + * @param [in] timeout the timeout in seconds. + */ + BulkImportConfig& + WithTimeout(int64_t timeout); + + /** + * @brief Get whether the server certificate is verified for https calls. + * @return true if the certificate is verified. + */ + bool + VerifyServerCert() const; + + /** + * @brief Set whether the server certificate is verified for https calls. + * @param [in] verify whether to verify the certificate. + */ + void + SetVerifyServerCert(bool verify); + + /** + * @brief Set whether the server certificate is verified for https calls. + * @param [in] verify whether to verify the certificate. + */ + BulkImportConfig& + WithVerifyServerCert(bool verify); + + /** + * @brief Get the path of the CA certificate bundle used to verify the server. + * @return the CA certificate path. + */ + const std::string& + CaCertPath() const; + + /** + * @brief Set the path of the CA certificate bundle used to verify the server. + * @param [in] ca_cert_path the CA certificate path. + */ + void + SetCaCertPath(const std::string& ca_cert_path); + + /** + * @brief Set the path of the CA certificate bundle used to verify the server. + * @param [in] ca_cert_path the CA certificate path. + */ + BulkImportConfig& + WithCaCertPath(const std::string& ca_cert_path); + + private: + int64_t timeout_{0}; + bool verify_server_cert_{true}; + std::string ca_cert_path_; +}; + +} // namespace milvus diff --git a/test/ut/TestBulkImport.cpp b/test/ut/TestBulkImport.cpp index d9b1e4d9..34ba422d 100644 --- a/test/ut/TestBulkImport.cpp +++ b/test/ut/TestBulkImport.cpp @@ -120,3 +120,293 @@ TEST(BulkImportTest, CreateCommitAndAbortImport) { EXPECT_EQ(abort_payload.at("jobId"), "456"); EXPECT_FALSE(abort_payload.contains("jobID")); } + +namespace { + +struct ImportServer { + httplib::Server server; + std::mutex mutex; + std::vector paths; + std::vector bodies; + std::vector authorizations; + std::vector db_names; + size_t created_jobs = 0; + int envelope_code = 0; + std::string envelope_message = "success"; + int port = -1; + std::thread thread; + + bool + Start() { + auto handler = [this](const httplib::Request& request, httplib::Response& response) { + nlohmann::json payload = { + {"code", envelope_code}, + {"message", envelope_message}, + {"data", nlohmann::json::object()}, + }; + { + std::lock_guard lock(mutex); + paths.emplace_back(request.path); + bodies.emplace_back(request.body); + authorizations.emplace_back(request.get_header_value("Authorization")); + db_names.emplace_back(request.get_header_value("DB-Name")); + if (request.path == "/v2/vectordb/jobs/import/create") { + payload["data"]["jobId"] = "job-" + std::to_string(++created_jobs); + } + } + response.set_content(payload.dump(), "application/json"); + }; + for (const auto& path : {"/v2/vectordb/jobs/import/create", "/v2/vectordb/jobs/import/list", + "/v2/vectordb/jobs/import/describe", "/v2/vectordb/jobs/import/get_progress", + "/v2/vectordb/jobs/import/commit", "/v2/vectordb/jobs/import/abort"}) { + server.Post(path, handler); + } + port = server.bind_to_any_port("127.0.0.1"); + if (port <= 0) { + return false; + } + thread = std::thread([this]() { server.listen_after_bind(); }); + server.wait_until_ready(); + return server.is_running(); + } + + ~ImportServer() { + if (server.is_running()) { + server.stop(); + } + if (thread.joinable()) { + thread.join(); + } + } + + std::string + Url() const { + return "http://127.0.0.1:" + std::to_string(port); + } +}; + +} // namespace + +TEST(BulkImportTest, DtoCreateMilvusImportJobs) { + ImportServer server; + ASSERT_TRUE(server.Start()); + + milvus::MilvusImportRequest request; + request.WithApiKey("token") + .WithDatabaseName("dto-db") + .WithCollectionName("collection") + .WithPartitionName("part") + .WithFiles({{"a.parquet", "b.parquet"}}) + .WithOptions(nlohmann::json{{"auto_commit", "false"}}); + milvus::BulkImportResponse response; + ASSERT_TRUE(milvus::BulkImport::CreateImportJobs(server.Url(), request, response).IsOk()); + EXPECT_EQ(response.Code(), 0); + EXPECT_EQ(response.JobId(), "job-1"); + + std::lock_guard lock(server.mutex); + ASSERT_EQ(server.paths.size(), 1); + EXPECT_EQ(server.paths.at(0), "/v2/vectordb/jobs/import/create"); + EXPECT_EQ(server.authorizations.at(0), "Bearer token"); + const auto payload = nlohmann::json::parse(server.bodies.at(0)); + EXPECT_EQ(payload.at("dbName"), "dto-db"); + EXPECT_EQ(payload.at("collectionName"), "collection"); + EXPECT_EQ(payload.at("partitionName"), "part"); + EXPECT_EQ(payload.at("files"), nlohmann::json::array({{"a.parquet", "b.parquet"}})); + EXPECT_EQ(payload.at("options").at("auto_commit"), "false"); +} + +TEST(BulkImportTest, DtoCreateVolumeImportJobs) { + ImportServer server; + ASSERT_TRUE(server.Start()); + + milvus::VolumeImportRequest request; + request.WithApiKey("token") + .WithClusterId("cluster-a") + .WithDatabaseName("db-1") + .WithCollectionName("collection") + .WithPartitionName("part-1") + .WithVolumeName("vol-1") + .WithDataPaths({{"parquet-folder/"}}); + milvus::BulkImportResponse response; + ASSERT_TRUE(milvus::BulkImport::CreateImportJobs(server.Url(), request, response).IsOk()); + + std::lock_guard lock(server.mutex); + ASSERT_EQ(server.paths.size(), 1); + const auto payload = nlohmann::json::parse(server.bodies.at(0)); + EXPECT_EQ(payload.at("clusterId"), "cluster-a"); + EXPECT_EQ(payload.at("dbName"), "db-1"); + EXPECT_EQ(payload.at("collectionName"), "collection"); + EXPECT_EQ(payload.at("partitionName"), "part-1"); + EXPECT_EQ(payload.at("volumeName"), "vol-1"); + EXPECT_EQ(payload.at("dataPaths"), nlohmann::json::array({{"parquet-folder/"}})); +} + +TEST(BulkImportTest, DtoCreateCloudImportJobs) { + ImportServer server; + ASSERT_TRUE(server.Start()); + + milvus::CloudImportRequest request; + request.WithApiKey("token") + .WithClusterId("cluster-a") + .WithProjectId("p-1") + .WithRegionId("r-1") + .WithCollectionName("collection") + .WithObjectUrls({{"s3://bucket/1.parquet"}}) + .WithAccessKey("ak") + .WithSecretKey("sk") + .WithToken("tok"); + milvus::BulkImportResponse response; + ASSERT_TRUE(milvus::BulkImport::CreateImportJobs(server.Url(), request, response).IsOk()); + + std::lock_guard lock(server.mutex); + ASSERT_EQ(server.paths.size(), 1); + const auto payload = nlohmann::json::parse(server.bodies.at(0)); + EXPECT_EQ(payload.at("clusterId"), "cluster-a"); + EXPECT_EQ(payload.at("projectId"), "p-1"); + EXPECT_EQ(payload.at("regionId"), "r-1"); + EXPECT_EQ(payload.at("collectionName"), "collection"); + EXPECT_EQ(payload.at("objectUrls"), nlohmann::json::array({{"s3://bucket/1.parquet"}})); + EXPECT_EQ(payload.at("accessKey"), "ak"); + EXPECT_EQ(payload.at("secretKey"), "sk"); + EXPECT_EQ(payload.at("token"), "tok"); +} + +TEST(BulkImportTest, DtoListMilvusImportJobs) { + ImportServer server; + ASSERT_TRUE(server.Start()); + + milvus::MilvusListImportJobsRequest request; + request.WithApiKey("token").WithCollectionName("collection").WithDatabaseName("db-1"); + milvus::BulkImportResponse response; + ASSERT_TRUE(milvus::BulkImport::ListImportJobs(server.Url(), request, response).IsOk()); + + std::lock_guard lock(server.mutex); + ASSERT_EQ(server.paths.size(), 1); + EXPECT_EQ(server.paths.at(0), "/v2/vectordb/jobs/import/list"); + const auto payload = nlohmann::json::parse(server.bodies.at(0)); + EXPECT_EQ(payload.at("collectionName"), "collection"); + EXPECT_EQ(payload.at("dbName"), "db-1"); +} + +TEST(BulkImportTest, DtoListCloudImportJobs) { + ImportServer server; + ASSERT_TRUE(server.Start()); + + milvus::CloudListImportJobsRequest request; + request.WithApiKey("token") + .WithClusterId("cluster-a") + .WithProjectId("p-1") + .WithRegionId("r-1") + .WithPageSize(10) + .WithCurrentPage(2); + milvus::BulkImportResponse response; + ASSERT_TRUE(milvus::BulkImport::ListImportJobs(server.Url(), request, response).IsOk()); + + std::lock_guard lock(server.mutex); + ASSERT_EQ(server.paths.size(), 1); + const auto payload = nlohmann::json::parse(server.bodies.at(0)); + EXPECT_EQ(payload.at("clusterId"), "cluster-a"); + EXPECT_EQ(payload.at("projectId"), "p-1"); + EXPECT_EQ(payload.at("regionId"), "r-1"); + EXPECT_EQ(payload.at("pageSize"), 10); + EXPECT_EQ(payload.at("currentPage"), 2); +} + +TEST(BulkImportTest, DtoDescribeCommitAndAbortImport) { + ImportServer server; + ASSERT_TRUE(server.Start()); + + milvus::MilvusDescribeImportRequest describe; + describe.WithApiKey("token").WithDatabaseName("db-1").WithJobId("123"); + milvus::BulkImportResponse response; + ASSERT_TRUE(milvus::BulkImport::GetImportJobProgress(server.Url(), describe, response).IsOk()); + + milvus::MilvusDescribeImportRequest commit; + commit.WithApiKey("token").WithDatabaseName("db-1").WithJobId("456"); + ASSERT_TRUE(milvus::BulkImport::CommitImport(server.Url(), commit, response).IsOk()); + + milvus::MilvusDescribeImportRequest abort; + abort.WithApiKey("token").WithDatabaseName("db-1").WithJobId("789"); + ASSERT_TRUE(milvus::BulkImport::AbortImport(server.Url(), abort, response).IsOk()); + + std::lock_guard lock(server.mutex); + ASSERT_EQ(server.paths.size(), 3); + EXPECT_EQ(server.paths.at(0), "/v2/vectordb/jobs/import/describe"); + EXPECT_EQ(server.paths.at(1), "/v2/vectordb/jobs/import/commit"); + EXPECT_EQ(server.paths.at(2), "/v2/vectordb/jobs/import/abort"); + for (const auto& db_name : server.db_names) { + EXPECT_EQ(db_name, "db-1"); + } + + auto payload = nlohmann::json::parse(server.bodies.at(0)); + EXPECT_EQ(payload.at("jobId"), "123"); + EXPECT_FALSE(payload.contains("jobID")); + payload = nlohmann::json::parse(server.bodies.at(1)); + EXPECT_EQ(payload.at("jobId"), "456"); + payload = nlohmann::json::parse(server.bodies.at(2)); + EXPECT_EQ(payload.at("jobId"), "789"); +} + +TEST(BulkImportTest, LegacyGetImportJobProgressUsesJobIdField) { + ImportServer server; + ASSERT_TRUE(server.Start()); + + const auto response = milvus::BulkImport::GetImportJobProgress(server.Url(), "123", "legacy-db", "token"); + ASSERT_FALSE(response.is_null()); + + std::lock_guard lock(server.mutex); + ASSERT_EQ(server.paths.size(), 1); + EXPECT_EQ(server.paths.at(0), "/v2/vectordb/jobs/import/get_progress"); + const auto payload = nlohmann::json::parse(server.bodies.at(0)); + EXPECT_EQ(payload.at("jobId"), "123"); + EXPECT_FALSE(payload.contains("jobID")); + EXPECT_EQ(server.db_names.at(0), "legacy-db"); +} + +TEST(BulkImportTest, DtoCloudDescribeImport) { + ImportServer server; + ASSERT_TRUE(server.Start()); + + milvus::CloudDescribeImportRequest request; + request.WithApiKey("token").WithClusterId("cluster-a").WithProjectId("p-1").WithRegionId("r-1").WithJobId("123"); + milvus::BulkImportResponse response; + ASSERT_TRUE(milvus::BulkImport::GetImportJobProgress(server.Url(), request, response).IsOk()); + + std::lock_guard lock(server.mutex); + ASSERT_EQ(server.paths.size(), 1); + const auto payload = nlohmann::json::parse(server.bodies.at(0)); + EXPECT_EQ(payload.at("clusterId"), "cluster-a"); + EXPECT_EQ(payload.at("projectId"), "p-1"); + EXPECT_EQ(payload.at("regionId"), "r-1"); + EXPECT_EQ(payload.at("jobId"), "123"); +} + +TEST(BulkImportTest, DtoImportReportsServerEnvelopeError) { + ImportServer server; + server.envelope_code = 1100; + server.envelope_message = "out of memory"; + ASSERT_TRUE(server.Start()); + + milvus::MilvusImportRequest request; + request.WithApiKey("token").WithCollectionName("collection").WithFiles({{"a.parquet"}}); + milvus::BulkImportResponse response; + const auto status = milvus::BulkImport::CreateImportJobs(server.Url(), request, response); + EXPECT_FALSE(status.IsOk()); + EXPECT_EQ(status.Code(), milvus::StatusCode::SERVER_FAILED); + EXPECT_NE(status.Message().find("1100"), std::string::npos); + EXPECT_NE(status.Message().find("out of memory"), std::string::npos); + EXPECT_EQ(response.Code(), 1100); +} + +TEST(BulkImportTest, DtoCreateImportWithConfig) { + ImportServer server; + ASSERT_TRUE(server.Start()); + + milvus::MilvusImportRequest request; + request.WithApiKey("token").WithCollectionName("collection").WithFiles({{"a.parquet"}}); + milvus::BulkImportConfig config; + config.WithTimeout(5).WithVerifyServerCert(false); + milvus::BulkImportResponse response; + ASSERT_TRUE(milvus::BulkImport::CreateImportJobs(server.Url(), request, response, config).IsOk()); + EXPECT_EQ(response.JobId(), "job-1"); +}