gsandeep1241 commented on code in PR #783: URL: https://github.com/apache/iceberg-cpp/pull/783#discussion_r4000961054
########## src/iceberg/catalog/rest/rest_table_scan.cc: ########## @@ -0,0 +1,298 @@ +/* + * Licensed to the Apache Software Foundation (ASF) 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 "iceberg/catalog/rest/rest_table_scan.h" + +#include <chrono> +#include <thread> + +#include <nlohmann/json.hpp> + +#include "iceberg/catalog/rest/endpoint.h" +#include "iceberg/catalog/rest/error_handlers.h" +#include "iceberg/catalog/rest/http_client.h" +#include "iceberg/catalog/rest/json_serde_internal.h" +#include "iceberg/catalog/rest/resource_paths.h" +#include "iceberg/catalog/rest/rest_file_io.h" +#include "iceberg/catalog/rest/types.h" +#include "iceberg/json_serde_internal.h" +#include "iceberg/partition_spec.h" +#include "iceberg/result.h" +#include "iceberg/schema.h" +#include "iceberg/table_metadata.h" +#include "iceberg/util/macros.h" + +namespace iceberg::rest { + +namespace { + +constexpr int64_t kMinSleepMs = 1'000; +constexpr int64_t kMaxSleepMs = 60'000; +constexpr int kMaxRetries = 10; +constexpr int64_t kMaxWaitTimeMs = 5 * 60 * 1'000; + +#define ICEBERG_ENDPOINT_CHECK(endpoints, endpoint) \ + do { \ + if (!endpoints.contains(endpoint)) { \ + return NotSupported("Not supported endpoint: {}", endpoint.ToString()); \ + } \ + } while (0) + +} // namespace + +// RestTableScan + +RestTableScan::RestTableScan(std::shared_ptr<TableMetadata> metadata, + std::shared_ptr<Schema> schema, std::shared_ptr<FileIO> io, + internal::TableScanContext context, + RestScanContext rest_context) + : DataTableScan(std::move(metadata), std::move(schema), std::move(io), + std::move(context)), + rest_context_(std::move(rest_context)) {} + +Result<std::unique_ptr<DataTableScan>> RestTableScan::Make( + std::shared_ptr<TableMetadata> metadata, std::shared_ptr<Schema> schema, + std::shared_ptr<FileIO> io, internal::TableScanContext context, + RestScanContext rest_context) { + ICEBERG_PRECHECK(metadata != nullptr, "Table metadata cannot be null"); + ICEBERG_PRECHECK(schema != nullptr, "Schema cannot be null"); + ICEBERG_PRECHECK(io != nullptr, "FileIO cannot be null"); + return std::unique_ptr<DataTableScan>( + new RestTableScan(std::move(metadata), std::move(schema), std::move(io), + std::move(context), std::move(rest_context))); +} + +Result<std::vector<std::shared_ptr<FileScanTask>>> RestTableScan::PlanFiles() const { + TableMetadataCache metadata_cache(metadata_.get()); + ICEBERG_ASSIGN_OR_RAISE(auto specs_by_id, metadata_cache.GetPartitionSpecsById()); + + std::string plan_id; + return PlanTableScan(plan_id, specs_by_id); +} + +Result<std::vector<std::shared_ptr<FileScanTask>>> RestTableScan::PlanTableScan( + std::string& plan_id, + const std::unordered_map<int32_t, std::shared_ptr<PartitionSpec>>& specs) const { + ICEBERG_ENDPOINT_CHECK(rest_context_.supported_endpoints, Endpoint::PlanTableScan()); + + // Build request from scan context + PlanTableScanRequest request; + request.select = context_.selected_columns; + request.filter = context_.filter; + request.case_sensitive = context_.case_sensitive; + request.min_rows_requested = context_.min_rows_requested; + + if (context_.from_snapshot_id.has_value() && context_.to_snapshot_id.has_value()) { + request.start_snapshot_id = context_.from_snapshot_id; + request.end_snapshot_id = context_.to_snapshot_id; + request.use_snapshot_schema = true; + } else if (context_.snapshot_id.has_value()) { + request.snapshot_id = context_.snapshot_id; + request.use_snapshot_schema = context_.use_snapshot_schema; + } + + if (!context_.columns_to_keep_stats.empty()) { + for (int32_t field_id : context_.columns_to_keep_stats) { + ICEBERG_ASSIGN_OR_RAISE(auto name, schema_->FindColumnNameById(field_id)); + if (name.has_value()) { + request.stats_fields.emplace_back(*name); + } + } + } + + ICEBERG_ASSIGN_OR_RAISE(auto path, + rest_context_.paths->Plan(rest_context_.identifier)); + ICEBERG_ASSIGN_OR_RAISE(auto request_json, ToJson(request)); + ICEBERG_ASSIGN_OR_RAISE(auto json_request, ToJsonString(request_json)); + ICEBERG_ASSIGN_OR_RAISE( + const auto response, + rest_context_.client->Post(path, json_request, /*headers=*/{}, + *PlanErrorHandler::Instance(), *rest_context_.session)); + ICEBERG_ASSIGN_OR_RAISE(auto json, FromJsonString(response.body())); + ICEBERG_ASSIGN_OR_RAISE(auto result, + PlanTableScanResponseFromJson(json, specs, *schema_)); + ICEBERG_RETURN_UNEXPECTED(result.Validate()); + + plan_id = result.plan_id; + + switch (result.plan_status) { + case PlanStatus::kCompleted: { + ICEBERG_RETURN_UNEXPECTED(ApplyStorageCredentials(result.storage_credentials)); + auto tasks = ResolveScanTasks(result.plan_tasks, result.file_scan_tasks, specs); + if (!tasks.has_value()) CancelPlanning(plan_id); + return tasks; + } + case PlanStatus::kSubmitted: + return FetchPlanningResult(plan_id, specs); + case PlanStatus::kFailed: + return IOError("Scan planning failed: {}", + result.error ? result.error->message : "unknown error"); + case PlanStatus::kCancelled: + return IOError("Scan planning was cancelled for plan_id={}", plan_id); Review Comment: Do we want to introduce a new error type? Or use InvalidArgument instead? ########## src/iceberg/catalog/rest/rest_table_scan.cc: ########## @@ -0,0 +1,269 @@ +/* + * Licensed to the Apache Software Foundation (ASF) 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 "iceberg/catalog/rest/rest_table_scan.h" + +#include <chrono> +#include <thread> + +#include <nlohmann/json.hpp> + +#include "iceberg/catalog/rest/endpoint.h" +#include "iceberg/catalog/rest/error_handlers.h" +#include "iceberg/catalog/rest/http_client.h" +#include "iceberg/catalog/rest/json_serde_internal.h" +#include "iceberg/catalog/rest/resource_paths.h" +#include "iceberg/catalog/rest/types.h" +#include "iceberg/json_serde_internal.h" +#include "iceberg/partition_spec.h" +#include "iceberg/result.h" +#include "iceberg/schema.h" +#include "iceberg/table_metadata.h" +#include "iceberg/util/macros.h" + +namespace iceberg::rest { + +namespace { + +constexpr int64_t kMinSleepMs = 1'000; +constexpr int64_t kMaxSleepMs = 60'000; +constexpr int kMaxRetries = 10; +constexpr int64_t kMaxWaitTimeMs = 5 * 60 * 1'000; + +#define ICEBERG_ENDPOINT_CHECK(endpoints, endpoint) \ + do { \ + if (!endpoints.contains(endpoint)) { \ + return NotSupported("Not supported endpoint: {}", endpoint.ToString()); \ + } \ + } while (0) + +} // namespace + +// RestTableScan + +RestTableScan::RestTableScan(std::shared_ptr<TableMetadata> metadata, + std::shared_ptr<Schema> schema, std::shared_ptr<FileIO> io, + internal::TableScanContext context, + RestScanContext rest_context) + : DataTableScan(std::move(metadata), std::move(schema), std::move(io), + std::move(context)), + rest_context_(std::move(rest_context)) {} + +Result<std::unique_ptr<DataTableScan>> RestTableScan::Make( + std::shared_ptr<TableMetadata> metadata, std::shared_ptr<Schema> schema, + std::shared_ptr<FileIO> io, internal::TableScanContext context, + RestScanContext rest_context) { + ICEBERG_PRECHECK(metadata != nullptr, "Table metadata cannot be null"); + ICEBERG_PRECHECK(schema != nullptr, "Schema cannot be null"); + ICEBERG_PRECHECK(io != nullptr, "FileIO cannot be null"); + return std::unique_ptr<DataTableScan>( + new RestTableScan(std::move(metadata), std::move(schema), std::move(io), + std::move(context), std::move(rest_context))); +} + +Result<std::vector<std::shared_ptr<FileScanTask>>> RestTableScan::PlanFiles() const { + TableMetadataCache metadata_cache(metadata_.get()); + ICEBERG_ASSIGN_OR_RAISE(auto specs_by_id, metadata_cache.GetPartitionSpecsById()); + + std::string plan_id; + return PlanTableScan(plan_id, specs_by_id); +} + +Result<std::vector<std::shared_ptr<FileScanTask>>> RestTableScan::PlanTableScan( + std::string& plan_id, + const std::unordered_map<int32_t, std::shared_ptr<PartitionSpec>>& specs) const { + ICEBERG_ENDPOINT_CHECK(rest_context_.supported_endpoints, Endpoint::PlanTableScan()); + + // Build request from scan context + PlanTableScanRequest request; + request.select = context_.selected_columns; + request.filter = context_.filter; + request.case_sensitive = context_.case_sensitive; + request.min_rows_requested = context_.min_rows_requested; + + if (context_.from_snapshot_id.has_value() && context_.to_snapshot_id.has_value()) { + request.start_snapshot_id = context_.from_snapshot_id; + request.end_snapshot_id = context_.to_snapshot_id; + } else if (context_.snapshot_id.has_value()) { + request.snapshot_id = context_.snapshot_id; + } + + if (!context_.columns_to_keep_stats.empty()) { + for (int32_t field_id : context_.columns_to_keep_stats) { + ICEBERG_ASSIGN_OR_RAISE(auto name, schema_->FindColumnNameById(field_id)); + if (name.has_value()) { + request.stats_fields.emplace_back(*name); + } + } + } + + ICEBERG_ASSIGN_OR_RAISE(auto path, + rest_context_.paths->Plan(rest_context_.identifier)); + ICEBERG_ASSIGN_OR_RAISE(auto request_json, ToJson(request)); + ICEBERG_ASSIGN_OR_RAISE(auto json_request, ToJsonString(request_json)); + ICEBERG_ASSIGN_OR_RAISE( + const auto response, + rest_context_.client->Post(path, json_request, /*headers=*/{}, + *PlanErrorHandler::Instance(), *rest_context_.session)); + ICEBERG_ASSIGN_OR_RAISE(auto json, FromJsonString(response.body())); + ICEBERG_ASSIGN_OR_RAISE(auto result, Review Comment: Done! ########## src/iceberg/catalog/rest/rest_table_scan.cc: ########## @@ -0,0 +1,269 @@ +/* + * Licensed to the Apache Software Foundation (ASF) 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 "iceberg/catalog/rest/rest_table_scan.h" + +#include <chrono> +#include <thread> + +#include <nlohmann/json.hpp> + +#include "iceberg/catalog/rest/endpoint.h" +#include "iceberg/catalog/rest/error_handlers.h" +#include "iceberg/catalog/rest/http_client.h" +#include "iceberg/catalog/rest/json_serde_internal.h" +#include "iceberg/catalog/rest/resource_paths.h" +#include "iceberg/catalog/rest/types.h" +#include "iceberg/json_serde_internal.h" +#include "iceberg/partition_spec.h" +#include "iceberg/result.h" +#include "iceberg/schema.h" +#include "iceberg/table_metadata.h" +#include "iceberg/util/macros.h" + +namespace iceberg::rest { + +namespace { + +constexpr int64_t kMinSleepMs = 1'000; +constexpr int64_t kMaxSleepMs = 60'000; +constexpr int kMaxRetries = 10; +constexpr int64_t kMaxWaitTimeMs = 5 * 60 * 1'000; + +#define ICEBERG_ENDPOINT_CHECK(endpoints, endpoint) \ + do { \ + if (!endpoints.contains(endpoint)) { \ + return NotSupported("Not supported endpoint: {}", endpoint.ToString()); \ + } \ + } while (0) + +} // namespace + +// RestTableScan + +RestTableScan::RestTableScan(std::shared_ptr<TableMetadata> metadata, + std::shared_ptr<Schema> schema, std::shared_ptr<FileIO> io, + internal::TableScanContext context, + RestScanContext rest_context) + : DataTableScan(std::move(metadata), std::move(schema), std::move(io), + std::move(context)), + rest_context_(std::move(rest_context)) {} + +Result<std::unique_ptr<DataTableScan>> RestTableScan::Make( + std::shared_ptr<TableMetadata> metadata, std::shared_ptr<Schema> schema, + std::shared_ptr<FileIO> io, internal::TableScanContext context, + RestScanContext rest_context) { + ICEBERG_PRECHECK(metadata != nullptr, "Table metadata cannot be null"); + ICEBERG_PRECHECK(schema != nullptr, "Schema cannot be null"); + ICEBERG_PRECHECK(io != nullptr, "FileIO cannot be null"); + return std::unique_ptr<DataTableScan>( + new RestTableScan(std::move(metadata), std::move(schema), std::move(io), + std::move(context), std::move(rest_context))); +} + +Result<std::vector<std::shared_ptr<FileScanTask>>> RestTableScan::PlanFiles() const { + TableMetadataCache metadata_cache(metadata_.get()); + ICEBERG_ASSIGN_OR_RAISE(auto specs_by_id, metadata_cache.GetPartitionSpecsById()); + + std::string plan_id; + return PlanTableScan(plan_id, specs_by_id); +} + +Result<std::vector<std::shared_ptr<FileScanTask>>> RestTableScan::PlanTableScan( + std::string& plan_id, + const std::unordered_map<int32_t, std::shared_ptr<PartitionSpec>>& specs) const { + ICEBERG_ENDPOINT_CHECK(rest_context_.supported_endpoints, Endpoint::PlanTableScan()); + + // Build request from scan context + PlanTableScanRequest request; + request.select = context_.selected_columns; + request.filter = context_.filter; + request.case_sensitive = context_.case_sensitive; + request.min_rows_requested = context_.min_rows_requested; + + if (context_.from_snapshot_id.has_value() && context_.to_snapshot_id.has_value()) { + request.start_snapshot_id = context_.from_snapshot_id; + request.end_snapshot_id = context_.to_snapshot_id; + } else if (context_.snapshot_id.has_value()) { + request.snapshot_id = context_.snapshot_id; + } + + if (!context_.columns_to_keep_stats.empty()) { + for (int32_t field_id : context_.columns_to_keep_stats) { + ICEBERG_ASSIGN_OR_RAISE(auto name, schema_->FindColumnNameById(field_id)); + if (name.has_value()) { + request.stats_fields.emplace_back(*name); + } + } + } + + ICEBERG_ASSIGN_OR_RAISE(auto path, + rest_context_.paths->Plan(rest_context_.identifier)); + ICEBERG_ASSIGN_OR_RAISE(auto request_json, ToJson(request)); + ICEBERG_ASSIGN_OR_RAISE(auto json_request, ToJsonString(request_json)); + ICEBERG_ASSIGN_OR_RAISE( + const auto response, + rest_context_.client->Post(path, json_request, /*headers=*/{}, + *PlanErrorHandler::Instance(), *rest_context_.session)); + ICEBERG_ASSIGN_OR_RAISE(auto json, FromJsonString(response.body())); + ICEBERG_ASSIGN_OR_RAISE(auto result, + PlanTableScanResponseFromJson(json, specs, *schema_)); + ICEBERG_RETURN_UNEXPECTED(result.Validate()); + + plan_id = result.plan_id; + + switch (result.plan_status) { + case PlanStatus::kCompleted: + return ResolveScanTasks(result.plan_tasks, result.file_scan_tasks, specs); Review Comment: Updated here. Are you also suggesting that we change the return signature of PlanFiles to have an iterable and let the blocking happen at the caller instead of here? ########## src/iceberg/catalog/rest/rest_catalog.cc: ########## @@ -873,6 +874,20 @@ Result<std::shared_ptr<Table>> RestCatalog::MakeTableFromLoadResult( TableAuthSession(identifier, table_config, std::move(contextual_session))); auto table_catalog = std::make_shared<TableScopedCatalog>( shared_from_this(), context, identifier, table_config, table_session); + + if (supported_endpoints_.contains(Endpoint::PlanTableScan())) { Review Comment: Done! ########## src/iceberg/catalog/rest/rest_table_scan.cc: ########## @@ -0,0 +1,269 @@ +/* + * Licensed to the Apache Software Foundation (ASF) 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 "iceberg/catalog/rest/rest_table_scan.h" + +#include <chrono> +#include <thread> + +#include <nlohmann/json.hpp> + +#include "iceberg/catalog/rest/endpoint.h" +#include "iceberg/catalog/rest/error_handlers.h" +#include "iceberg/catalog/rest/http_client.h" +#include "iceberg/catalog/rest/json_serde_internal.h" +#include "iceberg/catalog/rest/resource_paths.h" +#include "iceberg/catalog/rest/types.h" +#include "iceberg/json_serde_internal.h" +#include "iceberg/partition_spec.h" +#include "iceberg/result.h" +#include "iceberg/schema.h" +#include "iceberg/table_metadata.h" +#include "iceberg/util/macros.h" + +namespace iceberg::rest { + +namespace { + +constexpr int64_t kMinSleepMs = 1'000; +constexpr int64_t kMaxSleepMs = 60'000; +constexpr int kMaxRetries = 10; +constexpr int64_t kMaxWaitTimeMs = 5 * 60 * 1'000; + +#define ICEBERG_ENDPOINT_CHECK(endpoints, endpoint) \ + do { \ + if (!endpoints.contains(endpoint)) { \ + return NotSupported("Not supported endpoint: {}", endpoint.ToString()); \ + } \ + } while (0) + +} // namespace + +// RestTableScan + +RestTableScan::RestTableScan(std::shared_ptr<TableMetadata> metadata, + std::shared_ptr<Schema> schema, std::shared_ptr<FileIO> io, + internal::TableScanContext context, + RestScanContext rest_context) + : DataTableScan(std::move(metadata), std::move(schema), std::move(io), + std::move(context)), + rest_context_(std::move(rest_context)) {} + +Result<std::unique_ptr<DataTableScan>> RestTableScan::Make( + std::shared_ptr<TableMetadata> metadata, std::shared_ptr<Schema> schema, + std::shared_ptr<FileIO> io, internal::TableScanContext context, + RestScanContext rest_context) { + ICEBERG_PRECHECK(metadata != nullptr, "Table metadata cannot be null"); + ICEBERG_PRECHECK(schema != nullptr, "Schema cannot be null"); + ICEBERG_PRECHECK(io != nullptr, "FileIO cannot be null"); + return std::unique_ptr<DataTableScan>( + new RestTableScan(std::move(metadata), std::move(schema), std::move(io), + std::move(context), std::move(rest_context))); +} + +Result<std::vector<std::shared_ptr<FileScanTask>>> RestTableScan::PlanFiles() const { + TableMetadataCache metadata_cache(metadata_.get()); + ICEBERG_ASSIGN_OR_RAISE(auto specs_by_id, metadata_cache.GetPartitionSpecsById()); + + std::string plan_id; + return PlanTableScan(plan_id, specs_by_id); +} + +Result<std::vector<std::shared_ptr<FileScanTask>>> RestTableScan::PlanTableScan( + std::string& plan_id, + const std::unordered_map<int32_t, std::shared_ptr<PartitionSpec>>& specs) const { + ICEBERG_ENDPOINT_CHECK(rest_context_.supported_endpoints, Endpoint::PlanTableScan()); + + // Build request from scan context + PlanTableScanRequest request; + request.select = context_.selected_columns; + request.filter = context_.filter; + request.case_sensitive = context_.case_sensitive; + request.min_rows_requested = context_.min_rows_requested; + + if (context_.from_snapshot_id.has_value() && context_.to_snapshot_id.has_value()) { Review Comment: Done! -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
