From 1a304e4f2ccbc4111454b643746272aa97c7a141 Mon Sep 17 00:00:00 2001 From: David Dallakyan Date: Sun, 16 Aug 2026 16:37:48 +0400 Subject: [PATCH 1/4] feat(catalog-rest): add REST server-side scan planning client Port plan / fetch-result / cancel / fetch-tasks onto RestCatalog with a WaitForPlan poller. Task decoding and TableScan routing stay follow-ups, so supports_remote_scan_planning remains false. Part of #1690. --- Cargo.lock | 1 + crates/catalog/rest/Cargo.toml | 5 +- crates/catalog/rest/public-api.txt | 175 +++ crates/catalog/rest/src/catalog.rs | 21 +- crates/catalog/rest/src/endpoint.rs | 28 + crates/catalog/rest/src/lib.rs | 2 + crates/catalog/rest/src/scan_planning.rs | 1324 ++++++++++++++++++++++ crates/catalog/rest/src/types.rs | 6 + 8 files changed, 1551 insertions(+), 11 deletions(-) create mode 100644 crates/catalog/rest/src/scan_planning.rs diff --git a/Cargo.lock b/Cargo.lock index f1d00378f7..690728fe53 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -3900,6 +3900,7 @@ dependencies = [ "iceberg_test_utils", "itertools 0.13.0", "mockito", + "rand 0.9.5", "reqwest 0.12.28", "serde", "serde_derive", diff --git a/crates/catalog/rest/Cargo.toml b/crates/catalog/rest/Cargo.toml index 8dc9a86d7f..9c78c1b24f 100644 --- a/crates/catalog/rest/Cargo.toml +++ b/crates/catalog/rest/Cargo.toml @@ -35,14 +35,15 @@ chrono = { workspace = true } http = { workspace = true } iceberg = { workspace = true } itertools = { workspace = true } +rand = { workspace = true } reqwest = { workspace = true } serde = { workspace = true } serde_derive = { workspace = true } serde_json = { workspace = true } -tokio = { workspace = true } +tokio = { workspace = true, features = ["time"] } tracing = { workspace = true } typed-builder = { workspace = true } -uuid = { workspace = true, features = ["v4"] } +uuid = { workspace = true, features = ["v4", "v7"] } [dev-dependencies] bytes = { workspace = true } diff --git a/crates/catalog/rest/public-api.txt b/crates/catalog/rest/public-api.txt index 446821606f..5d692ddd23 100644 --- a/crates/catalog/rest/public-api.txt +++ b/crates/catalog/rest/public-api.txt @@ -14,6 +14,24 @@ impl<'a> core::fmt::Debug for iceberg_catalog_rest::HttpRequestBody<'a> pub fn iceberg_catalog_rest::HttpRequestBody<'a>::fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result impl<'a> core::marker::Copy for iceberg_catalog_rest::HttpRequestBody<'a> impl<'a> core::marker::StructuralPartialEq for iceberg_catalog_rest::HttpRequestBody<'a> +pub enum iceberg_catalog_rest::PlanStatus +pub iceberg_catalog_rest::PlanStatus::Cancelled +pub iceberg_catalog_rest::PlanStatus::Completed +pub iceberg_catalog_rest::PlanStatus::Failed +pub iceberg_catalog_rest::PlanStatus::Submitted +impl core::clone::Clone for iceberg_catalog_rest::PlanStatus +pub fn iceberg_catalog_rest::PlanStatus::clone(&self) -> iceberg_catalog_rest::PlanStatus +impl core::cmp::Eq for iceberg_catalog_rest::PlanStatus +impl core::cmp::PartialEq for iceberg_catalog_rest::PlanStatus +pub fn iceberg_catalog_rest::PlanStatus::eq(&self, other: &iceberg_catalog_rest::PlanStatus) -> bool +impl core::fmt::Debug for iceberg_catalog_rest::PlanStatus +pub fn iceberg_catalog_rest::PlanStatus::fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result +impl core::marker::Copy for iceberg_catalog_rest::PlanStatus +impl core::marker::StructuralPartialEq for iceberg_catalog_rest::PlanStatus +impl serde_core::ser::Serialize for iceberg_catalog_rest::PlanStatus +pub fn iceberg_catalog_rest::PlanStatus::serialize<__S>(&self, __serializer: __S) -> core::result::Result<<__S as serde_core::ser::Serializer>::Ok, <__S as serde_core::ser::Serializer>::Error> where __S: serde_core::ser::Serializer +impl<'de> serde_core::de::Deserialize<'de> for iceberg_catalog_rest::PlanStatus +pub fn iceberg_catalog_rest::PlanStatus::deserialize<__D>(__deserializer: __D) -> core::result::Result::Error> where __D: serde_core::de::Deserializer<'de> pub struct iceberg_catalog_rest::CommitTableRequest pub iceberg_catalog_rest::CommitTableRequest::identifier: core::option::Option pub iceberg_catalog_rest::CommitTableRequest::requirements: alloc::vec::Vec @@ -44,6 +62,22 @@ impl serde_core::ser::Serialize for iceberg_catalog_rest::CommitTableResponse pub fn iceberg_catalog_rest::CommitTableResponse::serialize<__S>(&self, __serializer: __S) -> core::result::Result<<__S as serde_core::ser::Serializer>::Ok, <__S as serde_core::ser::Serializer>::Error> where __S: serde_core::ser::Serializer impl<'de> serde_core::de::Deserialize<'de> for iceberg_catalog_rest::CommitTableResponse pub fn iceberg_catalog_rest::CommitTableResponse::deserialize<__D>(__deserializer: __D) -> core::result::Result::Error> where __D: serde_core::de::Deserializer<'de> +pub struct iceberg_catalog_rest::CompletedPlanningResult +pub iceberg_catalog_rest::CompletedPlanningResult::scan_tasks: iceberg_catalog_rest::ScanTasks +pub iceberg_catalog_rest::CompletedPlanningResult::status: iceberg_catalog_rest::PlanStatus +pub iceberg_catalog_rest::CompletedPlanningResult::storage_credentials: core::option::Option> +impl core::clone::Clone for iceberg_catalog_rest::CompletedPlanningResult +pub fn iceberg_catalog_rest::CompletedPlanningResult::clone(&self) -> iceberg_catalog_rest::CompletedPlanningResult +impl core::cmp::Eq for iceberg_catalog_rest::CompletedPlanningResult +impl core::cmp::PartialEq for iceberg_catalog_rest::CompletedPlanningResult +pub fn iceberg_catalog_rest::CompletedPlanningResult::eq(&self, other: &iceberg_catalog_rest::CompletedPlanningResult) -> bool +impl core::fmt::Debug for iceberg_catalog_rest::CompletedPlanningResult +pub fn iceberg_catalog_rest::CompletedPlanningResult::fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result +impl core::marker::StructuralPartialEq for iceberg_catalog_rest::CompletedPlanningResult +impl serde_core::ser::Serialize for iceberg_catalog_rest::CompletedPlanningResult +pub fn iceberg_catalog_rest::CompletedPlanningResult::serialize<__S>(&self, __serializer: __S) -> core::result::Result<<__S as serde_core::ser::Serializer>::Ok, <__S as serde_core::ser::Serializer>::Error> where __S: serde_core::ser::Serializer +impl<'de> serde_core::de::Deserialize<'de> for iceberg_catalog_rest::CompletedPlanningResult +pub fn iceberg_catalog_rest::CompletedPlanningResult::deserialize<__D>(__deserializer: __D) -> core::result::Result::Error> where __D: serde_core::de::Deserializer<'de> pub struct iceberg_catalog_rest::CreateNamespaceRequest pub iceberg_catalog_rest::CreateNamespaceRequest::namespace: iceberg::catalog::NamespaceIdent pub iceberg_catalog_rest::CreateNamespaceRequest::properties: std::collections::hash::map::HashMap @@ -124,6 +158,52 @@ impl serde_core::ser::Serialize for iceberg_catalog_rest::ErrorResponse pub fn iceberg_catalog_rest::ErrorResponse::serialize<__S>(&self, __serializer: __S) -> core::result::Result<<__S as serde_core::ser::Serializer>::Ok, <__S as serde_core::ser::Serializer>::Error> where __S: serde_core::ser::Serializer impl<'de> serde_core::de::Deserialize<'de> for iceberg_catalog_rest::ErrorResponse pub fn iceberg_catalog_rest::ErrorResponse::deserialize<__D>(__deserializer: __D) -> core::result::Result::Error> where __D: serde_core::de::Deserializer<'de> +pub struct iceberg_catalog_rest::FetchPlanningResultOptions +pub iceberg_catalog_rest::FetchPlanningResultOptions::access_delegation: core::option::Option +impl core::clone::Clone for iceberg_catalog_rest::FetchPlanningResultOptions +pub fn iceberg_catalog_rest::FetchPlanningResultOptions::clone(&self) -> iceberg_catalog_rest::FetchPlanningResultOptions +impl core::default::Default for iceberg_catalog_rest::FetchPlanningResultOptions +pub fn iceberg_catalog_rest::FetchPlanningResultOptions::default() -> iceberg_catalog_rest::FetchPlanningResultOptions +impl core::fmt::Debug for iceberg_catalog_rest::FetchPlanningResultOptions +pub fn iceberg_catalog_rest::FetchPlanningResultOptions::fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result +pub struct iceberg_catalog_rest::FetchPlanningResultResponse +pub iceberg_catalog_rest::FetchPlanningResultResponse::error: core::option::Option +pub iceberg_catalog_rest::FetchPlanningResultResponse::scan_tasks: iceberg_catalog_rest::ScanTasks +pub iceberg_catalog_rest::FetchPlanningResultResponse::status: iceberg_catalog_rest::PlanStatus +pub iceberg_catalog_rest::FetchPlanningResultResponse::storage_credentials: core::option::Option> +impl core::fmt::Debug for iceberg_catalog_rest::FetchPlanningResultResponse +pub fn iceberg_catalog_rest::FetchPlanningResultResponse::fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result +impl serde_core::ser::Serialize for iceberg_catalog_rest::FetchPlanningResultResponse +pub fn iceberg_catalog_rest::FetchPlanningResultResponse::serialize<__S>(&self, __serializer: __S) -> core::result::Result<<__S as serde_core::ser::Serializer>::Ok, <__S as serde_core::ser::Serializer>::Error> where __S: serde_core::ser::Serializer +impl<'de> serde_core::de::Deserialize<'de> for iceberg_catalog_rest::FetchPlanningResultResponse +pub fn iceberg_catalog_rest::FetchPlanningResultResponse::deserialize>(deserializer: D) -> core::result::Result::Error> +pub struct iceberg_catalog_rest::FetchScanTasksRequest +pub iceberg_catalog_rest::FetchScanTasksRequest::idempotency_key: core::option::Option +pub iceberg_catalog_rest::FetchScanTasksRequest::plan_task: alloc::string::String +impl core::clone::Clone for iceberg_catalog_rest::FetchScanTasksRequest +pub fn iceberg_catalog_rest::FetchScanTasksRequest::clone(&self) -> iceberg_catalog_rest::FetchScanTasksRequest +impl core::fmt::Debug for iceberg_catalog_rest::FetchScanTasksRequest +pub fn iceberg_catalog_rest::FetchScanTasksRequest::fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result +impl serde_core::ser::Serialize for iceberg_catalog_rest::FetchScanTasksRequest +pub fn iceberg_catalog_rest::FetchScanTasksRequest::serialize<__S>(&self, __serializer: __S) -> core::result::Result<<__S as serde_core::ser::Serializer>::Ok, <__S as serde_core::ser::Serializer>::Error> where __S: serde_core::ser::Serializer +impl<'de> serde_core::de::Deserialize<'de> for iceberg_catalog_rest::FetchScanTasksRequest +pub fn iceberg_catalog_rest::FetchScanTasksRequest::deserialize<__D>(__deserializer: __D) -> core::result::Result::Error> where __D: serde_core::de::Deserializer<'de> +pub struct iceberg_catalog_rest::FetchScanTasksResponse +pub iceberg_catalog_rest::FetchScanTasksResponse::scan_tasks: iceberg_catalog_rest::ScanTasks +impl core::clone::Clone for iceberg_catalog_rest::FetchScanTasksResponse +pub fn iceberg_catalog_rest::FetchScanTasksResponse::clone(&self) -> iceberg_catalog_rest::FetchScanTasksResponse +impl core::cmp::Eq for iceberg_catalog_rest::FetchScanTasksResponse +impl core::cmp::PartialEq for iceberg_catalog_rest::FetchScanTasksResponse +pub fn iceberg_catalog_rest::FetchScanTasksResponse::eq(&self, other: &iceberg_catalog_rest::FetchScanTasksResponse) -> bool +impl core::default::Default for iceberg_catalog_rest::FetchScanTasksResponse +pub fn iceberg_catalog_rest::FetchScanTasksResponse::default() -> iceberg_catalog_rest::FetchScanTasksResponse +impl core::fmt::Debug for iceberg_catalog_rest::FetchScanTasksResponse +pub fn iceberg_catalog_rest::FetchScanTasksResponse::fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result +impl core::marker::StructuralPartialEq for iceberg_catalog_rest::FetchScanTasksResponse +impl serde_core::ser::Serialize for iceberg_catalog_rest::FetchScanTasksResponse +pub fn iceberg_catalog_rest::FetchScanTasksResponse::serialize<__S>(&self, __serializer: __S) -> core::result::Result<<__S as serde_core::ser::Serializer>::Ok, <__S as serde_core::ser::Serializer>::Error> where __S: serde_core::ser::Serializer +impl<'de> serde_core::de::Deserialize<'de> for iceberg_catalog_rest::FetchScanTasksResponse +pub fn iceberg_catalog_rest::FetchScanTasksResponse::deserialize<__D>(__deserializer: __D) -> core::result::Result::Error> where __D: serde_core::de::Deserializer<'de> pub struct iceberg_catalog_rest::HttpClient impl iceberg_catalog_rest::HttpClient pub async fn iceberg_catalog_rest::HttpClient::post_form(&self, url: &str, headers: &http::header::map::HeaderMap, form: &std::collections::hash::map::HashMap<&str, &str>) -> iceberg::error::Result<(http::status::StatusCode, alloc::vec::Vec)> @@ -225,6 +305,40 @@ pub fn iceberg_catalog_rest::OAuth2Manager::fmt(&self, f: &mut core::fmt::Format impl iceberg_catalog_rest::AuthManager for iceberg_catalog_rest::OAuth2Manager pub fn iceberg_catalog_rest::OAuth2Manager::catalog_session<'life0, 'life1, 'life2, 'async_trait>(&'life0 self, client: &'life1 iceberg_catalog_rest::HttpClient, props: &'life2 std::collections::hash::map::HashMap) -> core::pin::Pin>> + core::marker::Send + 'async_trait)>> where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait pub fn iceberg_catalog_rest::OAuth2Manager::init_session<'life0, 'life1, 'life2, 'async_trait>(&'life0 self, client: &'life1 iceberg_catalog_rest::HttpClient, props: &'life2 std::collections::hash::map::HashMap) -> core::pin::Pin>> + core::marker::Send + 'async_trait)>> where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait +pub struct iceberg_catalog_rest::PlanTableScanRequest +pub iceberg_catalog_rest::PlanTableScanRequest::access_delegation: core::option::Option +pub iceberg_catalog_rest::PlanTableScanRequest::case_sensitive: core::option::Option +pub iceberg_catalog_rest::PlanTableScanRequest::end_snapshot_id: core::option::Option +pub iceberg_catalog_rest::PlanTableScanRequest::filter: core::option::Option +pub iceberg_catalog_rest::PlanTableScanRequest::idempotency_key: core::option::Option +pub iceberg_catalog_rest::PlanTableScanRequest::min_rows_requested: core::option::Option +pub iceberg_catalog_rest::PlanTableScanRequest::select: alloc::vec::Vec +pub iceberg_catalog_rest::PlanTableScanRequest::snapshot_id: core::option::Option +pub iceberg_catalog_rest::PlanTableScanRequest::start_snapshot_id: core::option::Option +pub iceberg_catalog_rest::PlanTableScanRequest::stats_fields: alloc::vec::Vec +pub iceberg_catalog_rest::PlanTableScanRequest::use_snapshot_schema: core::option::Option +impl core::clone::Clone for iceberg_catalog_rest::PlanTableScanRequest +pub fn iceberg_catalog_rest::PlanTableScanRequest::clone(&self) -> iceberg_catalog_rest::PlanTableScanRequest +impl core::default::Default for iceberg_catalog_rest::PlanTableScanRequest +pub fn iceberg_catalog_rest::PlanTableScanRequest::default() -> iceberg_catalog_rest::PlanTableScanRequest +impl core::fmt::Debug for iceberg_catalog_rest::PlanTableScanRequest +pub fn iceberg_catalog_rest::PlanTableScanRequest::fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result +impl serde_core::ser::Serialize for iceberg_catalog_rest::PlanTableScanRequest +pub fn iceberg_catalog_rest::PlanTableScanRequest::serialize<__S>(&self, __serializer: __S) -> core::result::Result<<__S as serde_core::ser::Serializer>::Ok, <__S as serde_core::ser::Serializer>::Error> where __S: serde_core::ser::Serializer +impl<'de> serde_core::de::Deserialize<'de> for iceberg_catalog_rest::PlanTableScanRequest +pub fn iceberg_catalog_rest::PlanTableScanRequest::deserialize<__D>(__deserializer: __D) -> core::result::Result::Error> where __D: serde_core::de::Deserializer<'de> +pub struct iceberg_catalog_rest::PlanTableScanResponse +pub iceberg_catalog_rest::PlanTableScanResponse::error: core::option::Option +pub iceberg_catalog_rest::PlanTableScanResponse::plan_id: core::option::Option +pub iceberg_catalog_rest::PlanTableScanResponse::scan_tasks: iceberg_catalog_rest::ScanTasks +pub iceberg_catalog_rest::PlanTableScanResponse::status: iceberg_catalog_rest::PlanStatus +pub iceberg_catalog_rest::PlanTableScanResponse::storage_credentials: core::option::Option> +impl core::fmt::Debug for iceberg_catalog_rest::PlanTableScanResponse +pub fn iceberg_catalog_rest::PlanTableScanResponse::fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result +impl serde_core::ser::Serialize for iceberg_catalog_rest::PlanTableScanResponse +pub fn iceberg_catalog_rest::PlanTableScanResponse::serialize<__S>(&self, __serializer: __S) -> core::result::Result<<__S as serde_core::ser::Serializer>::Ok, <__S as serde_core::ser::Serializer>::Error> where __S: serde_core::ser::Serializer +impl<'de> serde_core::de::Deserialize<'de> for iceberg_catalog_rest::PlanTableScanResponse +pub fn iceberg_catalog_rest::PlanTableScanResponse::deserialize>(deserializer: D) -> core::result::Result::Error> pub struct iceberg_catalog_rest::RegisterTableRequest pub iceberg_catalog_rest::RegisterTableRequest::metadata_location: alloc::string::String pub iceberg_catalog_rest::RegisterTableRequest::name: alloc::string::String @@ -257,6 +371,15 @@ pub fn iceberg_catalog_rest::RenameTableRequest::serialize<__S>(&self, __seriali impl<'de> serde_core::de::Deserialize<'de> for iceberg_catalog_rest::RenameTableRequest pub fn iceberg_catalog_rest::RenameTableRequest::deserialize<__D>(__deserializer: __D) -> core::result::Result::Error> where __D: serde_core::de::Deserializer<'de> pub struct iceberg_catalog_rest::RestCatalog +impl iceberg_catalog_rest::RestCatalog +pub async fn iceberg_catalog_rest::RestCatalog::cancel_planning(&self, table: &iceberg::catalog::TableIdent, plan_id: &str) -> iceberg::error::Result<()> +pub async fn iceberg_catalog_rest::RestCatalog::fetch_planning_result(&self, table: &iceberg::catalog::TableIdent, plan_id: &str, opts: iceberg_catalog_rest::FetchPlanningResultOptions) -> iceberg::error::Result +pub async fn iceberg_catalog_rest::RestCatalog::fetch_scan_tasks(&self, table: &iceberg::catalog::TableIdent, request: iceberg_catalog_rest::FetchScanTasksRequest) -> iceberg::error::Result +pub async fn iceberg_catalog_rest::RestCatalog::plan_table_scan(&self, table: &iceberg::catalog::TableIdent, request: iceberg_catalog_rest::PlanTableScanRequest) -> iceberg::error::Result +pub async fn iceberg_catalog_rest::RestCatalog::supports_full_remote_scan_planning(&self) -> iceberg::error::Result +pub async fn iceberg_catalog_rest::RestCatalog::supports_plan_table_scan(&self) -> iceberg::error::Result +pub async fn iceberg_catalog_rest::RestCatalog::supports_remote_scan_planning(&self) -> iceberg::error::Result +pub async fn iceberg_catalog_rest::RestCatalog::wait_for_plan(&self, table: &iceberg::catalog::TableIdent, plan_id: &str, opts: iceberg_catalog_rest::WaitForPlanOptions) -> iceberg::error::Result impl core::fmt::Debug for iceberg_catalog_rest::RestCatalog pub fn iceberg_catalog_rest::RestCatalog::fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result impl iceberg::catalog::Catalog for iceberg_catalog_rest::RestCatalog @@ -290,6 +413,22 @@ pub fn iceberg_catalog_rest::RestCatalogBuilder::load(self, name: impl core::con pub fn iceberg_catalog_rest::RestCatalogBuilder::with_kms_client_factory(self, kms_client_factory: alloc::sync::Arc) -> Self pub fn iceberg_catalog_rest::RestCatalogBuilder::with_runtime(self, runtime: iceberg::runtime::Runtime) -> Self pub fn iceberg_catalog_rest::RestCatalogBuilder::with_storage_factory(self, storage_factory: alloc::sync::Arc) -> Self +pub struct iceberg_catalog_rest::RestFileScanTask +pub iceberg_catalog_rest::RestFileScanTask::data_file: serde_json::value::Value +pub iceberg_catalog_rest::RestFileScanTask::delete_file_references: core::option::Option> +pub iceberg_catalog_rest::RestFileScanTask::residual_filter: core::option::Option +impl core::clone::Clone for iceberg_catalog_rest::RestFileScanTask +pub fn iceberg_catalog_rest::RestFileScanTask::clone(&self) -> iceberg_catalog_rest::RestFileScanTask +impl core::cmp::Eq for iceberg_catalog_rest::RestFileScanTask +impl core::cmp::PartialEq for iceberg_catalog_rest::RestFileScanTask +pub fn iceberg_catalog_rest::RestFileScanTask::eq(&self, other: &iceberg_catalog_rest::RestFileScanTask) -> bool +impl core::fmt::Debug for iceberg_catalog_rest::RestFileScanTask +pub fn iceberg_catalog_rest::RestFileScanTask::fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result +impl core::marker::StructuralPartialEq for iceberg_catalog_rest::RestFileScanTask +impl serde_core::ser::Serialize for iceberg_catalog_rest::RestFileScanTask +pub fn iceberg_catalog_rest::RestFileScanTask::serialize<__S>(&self, __serializer: __S) -> core::result::Result<<__S as serde_core::ser::Serializer>::Ok, <__S as serde_core::ser::Serializer>::Error> where __S: serde_core::ser::Serializer +impl<'de> serde_core::de::Deserialize<'de> for iceberg_catalog_rest::RestFileScanTask +pub fn iceberg_catalog_rest::RestFileScanTask::deserialize<__D>(__deserializer: __D) -> core::result::Result::Error> where __D: serde_core::de::Deserializer<'de> pub struct iceberg_catalog_rest::RestSessionCatalog impl core::fmt::Debug for iceberg_catalog_rest::RestSessionCatalog pub fn iceberg_catalog_rest::RestSessionCatalog::fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result @@ -321,6 +460,24 @@ impl core::default::Default for iceberg_catalog_rest::RestSessionCatalogBuilder pub fn iceberg_catalog_rest::RestSessionCatalogBuilder::default() -> Self impl core::fmt::Debug for iceberg_catalog_rest::RestSessionCatalogBuilder pub fn iceberg_catalog_rest::RestSessionCatalogBuilder::fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result +pub struct iceberg_catalog_rest::ScanTasks +pub iceberg_catalog_rest::ScanTasks::delete_files: alloc::vec::Vec +pub iceberg_catalog_rest::ScanTasks::file_scan_tasks: alloc::vec::Vec +pub iceberg_catalog_rest::ScanTasks::plan_tasks: alloc::vec::Vec +impl core::clone::Clone for iceberg_catalog_rest::ScanTasks +pub fn iceberg_catalog_rest::ScanTasks::clone(&self) -> iceberg_catalog_rest::ScanTasks +impl core::cmp::Eq for iceberg_catalog_rest::ScanTasks +impl core::cmp::PartialEq for iceberg_catalog_rest::ScanTasks +pub fn iceberg_catalog_rest::ScanTasks::eq(&self, other: &iceberg_catalog_rest::ScanTasks) -> bool +impl core::default::Default for iceberg_catalog_rest::ScanTasks +pub fn iceberg_catalog_rest::ScanTasks::default() -> iceberg_catalog_rest::ScanTasks +impl core::fmt::Debug for iceberg_catalog_rest::ScanTasks +pub fn iceberg_catalog_rest::ScanTasks::fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result +impl core::marker::StructuralPartialEq for iceberg_catalog_rest::ScanTasks +impl serde_core::ser::Serialize for iceberg_catalog_rest::ScanTasks +pub fn iceberg_catalog_rest::ScanTasks::serialize<__S>(&self, __serializer: __S) -> core::result::Result<<__S as serde_core::ser::Serializer>::Ok, <__S as serde_core::ser::Serializer>::Error> where __S: serde_core::ser::Serializer +impl<'de> serde_core::de::Deserialize<'de> for iceberg_catalog_rest::ScanTasks +pub fn iceberg_catalog_rest::ScanTasks::deserialize<__D>(__deserializer: __D) -> core::result::Result::Error> where __D: serde_core::de::Deserializer<'de> pub struct iceberg_catalog_rest::StorageCredential pub iceberg_catalog_rest::StorageCredential::config: std::collections::hash::map::HashMap pub iceberg_catalog_rest::StorageCredential::prefix: alloc::string::String @@ -367,6 +524,19 @@ impl serde_core::ser::Serialize for iceberg_catalog_rest::UpdateNamespacePropert pub fn iceberg_catalog_rest::UpdateNamespacePropertiesResponse::serialize<__S>(&self, __serializer: __S) -> core::result::Result<<__S as serde_core::ser::Serializer>::Ok, <__S as serde_core::ser::Serializer>::Error> where __S: serde_core::ser::Serializer impl<'de> serde_core::de::Deserialize<'de> for iceberg_catalog_rest::UpdateNamespacePropertiesResponse pub fn iceberg_catalog_rest::UpdateNamespacePropertiesResponse::deserialize<__D>(__deserializer: __D) -> core::result::Result::Error> where __D: serde_core::de::Deserializer<'de> +pub struct iceberg_catalog_rest::WaitForPlanOptions +pub iceberg_catalog_rest::WaitForPlanOptions::access_delegation: core::option::Option +pub iceberg_catalog_rest::WaitForPlanOptions::cancel_grace_period: core::time::Duration +pub iceberg_catalog_rest::WaitForPlanOptions::max_delay: core::time::Duration +pub iceberg_catalog_rest::WaitForPlanOptions::max_retries: u32 +pub iceberg_catalog_rest::WaitForPlanOptions::min_delay: core::time::Duration +pub iceberg_catalog_rest::WaitForPlanOptions::timeout: core::option::Option +impl core::clone::Clone for iceberg_catalog_rest::WaitForPlanOptions +pub fn iceberg_catalog_rest::WaitForPlanOptions::clone(&self) -> iceberg_catalog_rest::WaitForPlanOptions +impl core::default::Default for iceberg_catalog_rest::WaitForPlanOptions +pub fn iceberg_catalog_rest::WaitForPlanOptions::default() -> Self +impl core::fmt::Debug for iceberg_catalog_rest::WaitForPlanOptions +pub fn iceberg_catalog_rest::WaitForPlanOptions::fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result pub const iceberg_catalog_rest::AUTH_TYPE_NONE: &str pub const iceberg_catalog_rest::AUTH_TYPE_OAUTH2: &str pub const iceberg_catalog_rest::REST_CATALOG_PROP_AUTH_TYPE: &str @@ -384,3 +554,8 @@ pub fn iceberg_catalog_rest::OAuth2Manager::catalog_session<'life0, 'life1, 'lif pub fn iceberg_catalog_rest::OAuth2Manager::init_session<'life0, 'life1, 'life2, 'async_trait>(&'life0 self, client: &'life1 iceberg_catalog_rest::HttpClient, props: &'life2 std::collections::hash::map::HashMap) -> core::pin::Pin>> + core::marker::Send + 'async_trait)>> where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait pub trait iceberg_catalog_rest::AuthSession: core::fmt::Debug + core::marker::Send + core::marker::Sync pub fn iceberg_catalog_rest::AuthSession::authenticate<'life0, 'life1, 'async_trait>(&'life0 self, request: &'life1 mut iceberg_catalog_rest::HttpRequest) -> core::pin::Pin> + core::marker::Send + 'async_trait)>> where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait +pub fn iceberg_catalog_rest::is_no_such_plan_task(err: &iceberg::error::Error) -> bool +pub fn iceberg_catalog_rest::is_plan_cancelled(err: &iceberg::error::Error) -> bool +pub fn iceberg_catalog_rest::is_plan_expired(err: &iceberg::error::Error) -> bool +pub fn iceberg_catalog_rest::is_plan_failed(err: &iceberg::error::Error) -> bool +pub fn iceberg_catalog_rest::is_plan_poll_exhausted(err: &iceberg::error::Error) -> bool diff --git a/crates/catalog/rest/src/catalog.rs b/crates/catalog/rest/src/catalog.rs index e4fd204ba5..b165b2cdb0 100644 --- a/crates/catalog/rest/src/catalog.rs +++ b/crates/catalog/rest/src/catalog.rs @@ -238,7 +238,7 @@ impl RestCatalogConfig { self.url_prefixed(&["namespaces", &ns.to_url_string(), "register"]) } - fn table_endpoint(&self, table: &TableIdent) -> String { + pub(crate) fn table_endpoint(&self, table: &TableIdent) -> String { self.url_prefixed(&[ "namespaces", &table.namespace.to_url_string(), @@ -412,14 +412,14 @@ pub(crate) fn oauth_params_from_props(props: &HashMap) -> HashMa } #[derive(Debug)] -struct RestClient { +pub(crate) struct RestClient { /// Carries the session the auth manager derived from the merged /// configuration, so every request below is authenticated. - http_client: HttpClient, + pub(crate) http_client: HttpClient, /// Runtime config is fetched from rest server and stored here. /// /// It could be different from the user config. - config: RestCatalogConfig, + pub(crate) config: RestCatalogConfig, /// Capabilities the server advertises (see [`RestSessionCatalog::supports_endpoint`]). endpoints: HashSet, } @@ -479,7 +479,7 @@ impl RestClient { } /// Sends `request`, authenticated by the client's session. - async fn query_catalog(&self, request: HttpRequest) -> Result { + pub(crate) async fn query_catalog(&self, request: HttpRequest) -> Result { self.http_client.query_catalog(request).await } @@ -544,7 +544,7 @@ pub struct RestCatalog { impl RestCatalog { /// Creates a `RestCatalog` from a [`RestCatalogConfig`]. #[cfg(test)] - fn new( + pub(crate) fn new( context: SessionContext, config: RestCatalogConfig, auth_manager: Option>, @@ -570,10 +570,13 @@ impl RestCatalog { } } - #[cfg(test)] - async fn client(&self) -> Result<&RestClient> { + pub(crate) async fn client(&self) -> Result<&RestClient> { self.inner.client().await } + + pub(crate) async fn supports_endpoint(&self, endpoint: &Endpoint) -> Result { + self.inner.supports_endpoint(endpoint).await + } } /// Every operation forwards to its [`RestSessionCatalog`] equivalent with the @@ -809,7 +812,7 @@ impl RestSessionCatalog { } /// Gets the [`RestClient`] from the catalog. - async fn client(&self) -> Result<&RestClient> { + pub(crate) async fn client(&self) -> Result<&RestClient> { self.client .get_or_try_init(|| async { RestClient::init(&self.user_config, self.resolve_auth_manager()?).await diff --git a/crates/catalog/rest/src/endpoint.rs b/crates/catalog/rest/src/endpoint.rs index 3d82071f21..adc5231b6f 100644 --- a/crates/catalog/rest/src/endpoint.rs +++ b/crates/catalog/rest/src/endpoint.rs @@ -168,6 +168,10 @@ endpoints! { V1_REGISTER_TABLE => POST "/v1/{prefix}/namespaces/{namespace}/register", V1_REPORT_METRICS => POST "/v1/{prefix}/namespaces/{namespace}/tables/{table}/metrics", V1_COMMIT_TRANSACTION => POST "/v1/{prefix}/transactions/commit", + V1_PLAN_TABLE_SCAN => POST "/v1/{prefix}/namespaces/{namespace}/tables/{table}/plan", + V1_FETCH_PLAN_RESULT => GET "/v1/{prefix}/namespaces/{namespace}/tables/{table}/plan/{plan-id}", + V1_CANCEL_PLANNING => DELETE "/v1/{prefix}/namespaces/{namespace}/tables/{table}/plan/{plan-id}", + V1_FETCH_SCAN_TASKS => POST "/v1/{prefix}/namespaces/{namespace}/tables/{table}/tasks", } /// The standard v1 endpoints assumed to be supported when a server's @@ -231,4 +235,28 @@ mod tests { fn normalizes_http_method_to_uppercase() { assert_eq!("get /v1/x".parse::().unwrap().method(), "GET"); } + + #[test] + fn scan_planning_endpoints_are_optional_and_not_in_the_default_set() { + assert_eq!( + V1_PLAN_TABLE_SCAN.to_string(), + "POST /v1/{prefix}/namespaces/{namespace}/tables/{table}/plan" + ); + assert_eq!( + V1_FETCH_PLAN_RESULT.to_string(), + "GET /v1/{prefix}/namespaces/{namespace}/tables/{table}/plan/{plan-id}" + ); + assert_eq!( + V1_CANCEL_PLANNING.to_string(), + "DELETE /v1/{prefix}/namespaces/{namespace}/tables/{table}/plan/{plan-id}" + ); + assert_eq!( + V1_FETCH_SCAN_TASKS.to_string(), + "POST /v1/{prefix}/namespaces/{namespace}/tables/{table}/tasks" + ); + assert!(!DEFAULT_ENDPOINTS.contains(&V1_PLAN_TABLE_SCAN)); + assert!(!DEFAULT_ENDPOINTS.contains(&V1_FETCH_PLAN_RESULT)); + assert!(!DEFAULT_ENDPOINTS.contains(&V1_CANCEL_PLANNING)); + assert!(!DEFAULT_ENDPOINTS.contains(&V1_FETCH_SCAN_TASKS)); + } } diff --git a/crates/catalog/rest/src/lib.rs b/crates/catalog/rest/src/lib.rs index d9ad38dfab..681f5356b2 100644 --- a/crates/catalog/rest/src/lib.rs +++ b/crates/catalog/rest/src/lib.rs @@ -92,9 +92,11 @@ pub use client::HttpClient; mod request; pub use request::{HttpRequest, HttpRequestBody}; mod endpoint; +mod scan_planning; mod types; pub use auth::*; pub use catalog::*; pub use endpoint::Endpoint; +pub use scan_planning::*; pub use types::*; diff --git a/crates/catalog/rest/src/scan_planning.rs b/crates/catalog/rest/src/scan_planning.rs new file mode 100644 index 0000000000..7e7439bb66 --- /dev/null +++ b/crates/catalog/rest/src/scan_planning.rs @@ -0,0 +1,1324 @@ +// 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. + +//! REST server-side scan planning client. +//! +//! Implements the plan / fetch-result / cancel / fetch-tasks endpoints and a +//! [`RestCatalog::wait_for_plan`] poller. Task decoding and `TableScan` +//! auto-routing are follow-ups: [`RestCatalog::supports_remote_scan_planning`] +//! stays `false` until those land. + +use std::time::Duration; + +use iceberg::{Error, ErrorKind, Result, TableIdent}; +use rand::Rng; +use reqwest::{Method, Response, StatusCode}; +use serde::de::{self, Deserializer}; +use serde::{Deserialize, Serialize}; +use uuid::{Uuid, Variant, Version}; + +use crate::catalog::RestCatalog; +use crate::client::{deserialize_catalog_response, deserialize_unexpected_catalog_error}; +use crate::endpoint::{ + Endpoint, V1_CANCEL_PLANNING, V1_FETCH_PLAN_RESULT, V1_FETCH_SCAN_TASKS, V1_PLAN_TABLE_SCAN, +}; +use crate::request::HttpRequest; +use crate::types::{ErrorModel, ErrorResponse, StorageCredential}; + +const HEADER_IDEMPOTENCY_KEY: &str = "Idempotency-Key"; +const HEADER_ACCESS_DELEGATION: &str = "X-Iceberg-Access-Delegation"; + +const MSG_PLAN_EXPIRED: &str = "scan plan expired"; +const MSG_PLAN_FAILED: &str = "scan plan failed"; +const MSG_PLAN_CANCELLED: &str = "scan plan cancelled"; +const MSG_NO_SUCH_PLAN_TASK: &str = "scan plan task not found"; +const MSG_PLAN_POLL_EXHAUSTED: &str = "scan plan polling exhausted retries"; + +const ERR_TYPE_NO_SUCH_PLAN_ID: &str = "NoSuchPlanIdException"; +const ERR_TYPE_NO_SUCH_PLAN_TASK: &str = "NoSuchPlanTaskException"; +const ERR_TYPE_NO_SUCH_TABLE: &str = "NoSuchTableException"; +const ERR_TYPE_NO_SUCH_NAMESPACE: &str = "NoSuchNamespaceException"; + +/// Status of a server-side scan plan. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "kebab-case")] +pub enum PlanStatus { + /// Planning finished and tasks (or plan-task handles) are available. + Completed, + /// Planning is still running; poll [`RestCatalog::fetch_planning_result`]. + Submitted, + /// The plan was cancelled. Valid on fetch-result, not on planTableScan. + Cancelled, + /// Planning failed. The error detail is on the failed arm. + Failed, +} + +/// Task payload shared by completed planning responses and fetchScanTasks. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, Default)] +#[serde(rename_all = "kebab-case")] +pub struct ScanTasks { + /// Opaque plan-task handles that still need [`RestCatalog::fetch_scan_tasks`]. + #[serde(default, skip_serializing_if = "Vec::is_empty")] + pub plan_tasks: Vec, + /// File scan tasks. `data-file` is left as JSON until a decoder lands. + #[serde(default, skip_serializing_if = "Vec::is_empty")] + pub file_scan_tasks: Vec, + /// Delete files referenced by the scan tasks, as raw REST JSON. + #[serde(default, skip_serializing_if = "Vec::is_empty")] + pub delete_files: Vec, +} + +/// REST `FileScanTask` wire payload. Nested content-files stay opaque. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "kebab-case")] +pub struct RestFileScanTask { + /// REST ContentFile JSON for the data file. + pub data_file: serde_json::Value, + /// Indices into the sibling delete-files array. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub delete_file_references: Option>, + /// Optional residual filter in ExpressionParser JSON. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub residual_filter: Option, +} + +/// POST `.../plan` request body. Header-only fields are skipped on the wire. +#[derive(Debug, Clone, Default, Serialize, Deserialize)] +#[serde(rename_all = "kebab-case")] +pub struct PlanTableScanRequest { + /// `Idempotency-Key` header. `None` generates a fresh UUIDv7 per call. + #[serde(skip)] + pub idempotency_key: Option, + /// `X-Iceberg-Access-Delegation` header. `None` sends no such header. + #[serde(skip)] + pub access_delegation: Option, + /// Snapshot to scan. Omitted for the current snapshot. + #[serde(skip_serializing_if = "Option::is_none")] + pub snapshot_id: Option, + /// Selected schema fields. + #[serde(default, skip_serializing_if = "Vec::is_empty")] + pub select: Vec, + /// Row filter as ExpressionParser JSON, not `iceberg::expr::Predicate`. + #[serde(skip_serializing_if = "Option::is_none")] + pub filter: Option, + /// Hint for the minimum number of rows the server should return. + #[serde(skip_serializing_if = "Option::is_none")] + pub min_rows_requested: Option, + /// Case-sensitive field matching for filter and select. + #[serde(skip_serializing_if = "Option::is_none")] + pub case_sensitive: Option, + /// When true, use the schema at the scanned snapshot. + #[serde(skip_serializing_if = "Option::is_none")] + pub use_snapshot_schema: Option, + /// Incremental scan start (exclusive). Wire-only in this PR. + #[serde(skip_serializing_if = "Option::is_none")] + pub start_snapshot_id: Option, + /// Incremental scan end (inclusive). Wire-only in this PR. + #[serde(skip_serializing_if = "Option::is_none")] + pub end_snapshot_id: Option, + /// Fields for which the server should send column stats. + #[serde(default, skip_serializing_if = "Vec::is_empty")] + pub stats_fields: Vec, +} + +/// POST `.../plan` response. `completed` and `submitted` require `plan-id`. +#[derive(Debug, Serialize)] +#[serde(rename_all = "kebab-case")] +pub struct PlanTableScanResponse { + /// Discriminator for the planning-result union. + pub status: PlanStatus, + /// Server-issued plan id. Required for completed and submitted. + #[serde(skip_serializing_if = "Option::is_none")] + pub plan_id: Option, + /// Failed-arm error detail. + #[serde(skip_serializing_if = "Option::is_none")] + pub error: Option, + /// Task payload. Empty unless status is completed. + #[serde(flatten)] + pub scan_tasks: ScanTasks, + /// Optional vended credentials for reading the returned files. + #[serde(skip_serializing_if = "Option::is_none")] + pub storage_credentials: Option>, +} + +/// GET `.../plan/{plan-id}` response. +#[derive(Debug, Serialize)] +#[serde(rename_all = "kebab-case")] +pub struct FetchPlanningResultResponse { + /// Discriminator for the planning-result union. + pub status: PlanStatus, + /// Failed-arm error detail. + #[serde(skip_serializing_if = "Option::is_none")] + pub error: Option, + /// Task payload. Empty unless status is completed. + #[serde(flatten)] + pub scan_tasks: ScanTasks, + /// Optional vended credentials for reading the returned files. + #[serde(skip_serializing_if = "Option::is_none")] + pub storage_credentials: Option>, +} + +/// Completed arm of a planning result, as returned by [`RestCatalog::wait_for_plan`]. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "kebab-case")] +pub struct CompletedPlanningResult { + /// Always [`PlanStatus::Completed`]. + pub status: PlanStatus, + /// Task payload, which may still include plan-task handles. + #[serde(flatten)] + pub scan_tasks: ScanTasks, + /// Optional vended credentials for reading the returned files. + #[serde(skip_serializing_if = "Option::is_none")] + pub storage_credentials: Option>, +} + +/// Per-call options for [`RestCatalog::fetch_planning_result`]. +#[derive(Debug, Clone, Default)] +pub struct FetchPlanningResultOptions { + /// `X-Iceberg-Access-Delegation` header. `None` sends no such header. + pub access_delegation: Option, +} + +/// POST `.../tasks` request body. +#[derive(Debug, Clone, Serialize, Deserialize)] +#[serde(rename_all = "kebab-case")] +pub struct FetchScanTasksRequest { + /// `Idempotency-Key` header. `None` generates a fresh UUIDv7 per call. + #[serde(skip)] + pub idempotency_key: Option, + /// Opaque plan-task handle from a completed plan. + pub plan_task: String, +} + +/// POST `.../tasks` response. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, Default)] +#[serde(rename_all = "kebab-case")] +pub struct FetchScanTasksResponse { + /// Task payload, which may itself contain further plan-task handles. + #[serde(flatten)] + pub scan_tasks: ScanTasks, +} + +/// Polling backoff and bounds for [`RestCatalog::wait_for_plan`]. +#[derive(Debug, Clone)] +pub struct WaitForPlanOptions { + /// Backoff floor. Zero uses 100ms. + pub min_delay: Duration, + /// Backoff cap. Zero uses 5s. + pub max_delay: Duration, + /// Bound on the best-effort cancel after giving up. Zero uses 5s. + pub cancel_grace_period: Duration, + /// Poll attempts after the first. Zero uses 10. + pub max_retries: u32, + /// Optional overall deadline. `None` relies on `max_retries`. + pub timeout: Option, + /// `X-Iceberg-Access-Delegation` forwarded on each poll. + pub access_delegation: Option, +} + +impl Default for WaitForPlanOptions { + fn default() -> Self { + Self { + min_delay: Duration::from_millis(100), + max_delay: Duration::from_secs(5), + cancel_grace_period: Duration::from_secs(5), + max_retries: 10, + timeout: None, + access_delegation: None, + } + } +} + +impl<'de> Deserialize<'de> for PlanTableScanResponse { + fn deserialize>(deserializer: D) -> std::result::Result { + let raw = RawPlanningResponse::deserialize(deserializer)?; + match raw.status { + PlanStatus::Completed | PlanStatus::Submitted => { + if raw.plan_id.is_none() { + return Err(de::Error::custom(format!( + "planTableScan response with status {:?} missing plan-id", + raw.status + ))); + } + } + PlanStatus::Cancelled => { + return Err(de::Error::custom( + "planTableScan response has invalid status cancelled", + )); + } + PlanStatus::Failed => {} + } + Ok(PlanTableScanResponse { + status: raw.status, + plan_id: raw.plan_id, + error: decode_planning_error(raw.error), + scan_tasks: raw.scan_tasks, + storage_credentials: raw.storage_credentials, + }) + } +} + +impl<'de> Deserialize<'de> for FetchPlanningResultResponse { + fn deserialize>(deserializer: D) -> std::result::Result { + let raw = RawPlanningResponse::deserialize(deserializer)?; + Ok(FetchPlanningResultResponse { + status: raw.status, + error: decode_planning_error(raw.error), + scan_tasks: raw.scan_tasks, + storage_credentials: raw.storage_credentials, + }) + } +} + +#[derive(Deserialize)] +#[serde(rename_all = "kebab-case")] +struct RawPlanningResponse { + status: PlanStatus, + #[serde(default)] + plan_id: Option, + #[serde(default)] + error: Option, + #[serde(flatten)] + scan_tasks: ScanTasks, + #[serde(default)] + storage_credentials: Option>, +} + +fn decode_planning_error(raw: Option) -> Option { + let value = raw?; + serde_json::from_value(value).ok() +} + +/// True when a fetch-result 404 was a forgotten plan-id. +pub fn is_plan_expired(err: &Error) -> bool { + err.message() == MSG_PLAN_EXPIRED +} + +/// True when the server reported a failed plan. +pub fn is_plan_failed(err: &Error) -> bool { + err.message() == MSG_PLAN_FAILED || err.message().starts_with(&format!("{MSG_PLAN_FAILED}: ")) +} + +/// True when a submitted plan was cancelled. +pub fn is_plan_cancelled(err: &Error) -> bool { + err.message() == MSG_PLAN_CANCELLED +} + +/// True when a plan-task handle is gone. +pub fn is_no_such_plan_task(err: &Error) -> bool { + err.message() == MSG_NO_SUCH_PLAN_TASK +} + +/// True when [`RestCatalog::wait_for_plan`] hit its retry or timeout bound. +pub fn is_plan_poll_exhausted(err: &Error) -> bool { + err.message() == MSG_PLAN_POLL_EXHAUSTED +} + +impl RestCatalog { + /// Whether the server advertised the synchronous plan endpoint. + pub async fn supports_plan_table_scan(&self) -> Result { + self.supports_endpoint(&V1_PLAN_TABLE_SCAN).await + } + + /// Whether the server advertised all four scan-planning endpoints. + pub async fn supports_full_remote_scan_planning(&self) -> Result { + Ok(self.supports_plan_table_scan().await? + && self.supports_endpoint(&V1_FETCH_PLAN_RESULT).await? + && self.supports_endpoint(&V1_CANCEL_PLANNING).await? + && self.supports_endpoint(&V1_FETCH_SCAN_TASKS).await?) + } + + /// Whether this catalog can complete a remote plan end-to-end. + /// + /// Stays `false` until task decoding is wired into `TableScan`. Routing + /// auto-mode scans on endpoint capability alone would fail with + /// [`ErrorKind::FeatureUnsupported`] instead of falling back to local + /// planning. + pub async fn supports_remote_scan_planning(&self) -> Result { + // Touch config so a down server still surfaces as an error, but never + // claim end-to-end remote planning until TableScan can decode tasks. + let _ = self.supports_plan_table_scan().await?; + Ok(false) + } + + /// Submits a server-side scan plan. + /// + /// Completed and submitted plans return `Ok`. A failed plan returns + /// [`is_plan_failed`]. + pub async fn plan_table_scan( + &self, + table: &TableIdent, + request: PlanTableScanRequest, + ) -> Result { + self.require_endpoint(&V1_PLAN_TABLE_SCAN, "planTableScan") + .await?; + let client = self.client().await?; + let url = format!("{}/plan", client.config.table_endpoint(table)); + let headers = scan_planning_headers( + request.idempotency_key.as_deref(), + request.access_delegation.as_deref(), + true, + )?; + let mut builder = client.http_client.request(Method::POST, url).json(&request); + for (name, value) in headers { + builder = builder.header(name, value); + } + let http_response = client.query_catalog(HttpRequest::build(builder)?).await?; + let resp: PlanTableScanResponse = match http_response.status() { + StatusCode::OK => deserialize_catalog_response(http_response).await?, + StatusCode::NOT_FOUND => { + return Err(map_plan_not_found(http_response, PlanNotFoundKind::Submit).await); + } + _ => { + return Err(deserialize_unexpected_catalog_error( + http_response, + client.http_client.disable_header_redaction(), + ) + .await); + } + }; + match resp.status { + PlanStatus::Completed | PlanStatus::Submitted => Ok(resp), + PlanStatus::Failed => Err(failed_plan_error(resp.error.as_ref())), + PlanStatus::Cancelled => Err(Error::new( + ErrorKind::DataInvalid, + "planTableScan response has invalid status cancelled", + )), + } + } + + /// Polls a previously submitted plan. + pub async fn fetch_planning_result( + &self, + table: &TableIdent, + plan_id: &str, + opts: FetchPlanningResultOptions, + ) -> Result { + match self.fetch_planning_result_raw(table, plan_id, opts).await { + Ok(resp) => Ok(resp), + Err(PollError::Retry { error, .. }) | Err(PollError::Terminal(error)) => Err(error), + } + } + + /// Cancels a server-side plan. Best-effort: a 404 is returned as a generic + /// unexpected error rather than [`is_plan_expired`]. + pub async fn cancel_planning(&self, table: &TableIdent, plan_id: &str) -> Result<()> { + require_plan_id(plan_id)?; + self.require_endpoint(&V1_CANCEL_PLANNING, "cancelPlanning") + .await?; + let client = self.client().await?; + let url = format!( + "{}/plan/{}", + client.config.table_endpoint(table), + escape_opaque_path_segment(plan_id) + ); + let request = HttpRequest::build(client.http_client.request(Method::DELETE, url))?; + let http_response = client.query_catalog(request).await?; + match http_response.status() { + StatusCode::NO_CONTENT | StatusCode::OK => Ok(()), + _ => Err(deserialize_unexpected_catalog_error( + http_response, + client.http_client.disable_header_redaction(), + ) + .await), + } + } + + /// Fetches scan tasks for a plan-task handle. + pub async fn fetch_scan_tasks( + &self, + table: &TableIdent, + request: FetchScanTasksRequest, + ) -> Result { + self.require_endpoint(&V1_FETCH_SCAN_TASKS, "fetchScanTasks") + .await?; + let client = self.client().await?; + let url = format!("{}/tasks", client.config.table_endpoint(table)); + let headers = scan_planning_headers(request.idempotency_key.as_deref(), None, true)?; + let mut builder = client.http_client.request(Method::POST, url).json(&request); + for (name, value) in headers { + builder = builder.header(name, value); + } + let http_response = client.query_catalog(HttpRequest::build(builder)?).await?; + match http_response.status() { + StatusCode::OK => { + let bytes = http_response.bytes().await?; + if bytes.is_empty() { + return Err(Error::new( + ErrorKind::Unexpected, + "fetchScanTasks response was empty", + )); + } + serde_json::from_slice(&bytes).map_err(|e| { + Error::new( + ErrorKind::Unexpected, + "Failed to parse response from rest catalog server", + ) + .with_context("json", String::from_utf8_lossy(&bytes)) + .with_source(e) + }) + } + StatusCode::NOT_FOUND => { + Err(map_plan_not_found(http_response, PlanNotFoundKind::Tasks).await) + } + _ => Err(deserialize_unexpected_catalog_error( + http_response, + client.http_client.disable_header_redaction(), + ) + .await), + } + } + + /// Polls a submitted plan until it completes, fails, or the retry budget + /// is spent. Does not expand plan-task handles. + pub async fn wait_for_plan( + &self, + table: &TableIdent, + plan_id: &str, + opts: WaitForPlanOptions, + ) -> Result { + require_plan_id(plan_id)?; + self.require_endpoint(&V1_FETCH_PLAN_RESULT, "fetchPlanningResult") + .await?; + + let work = self.wait_for_plan_loop(table, plan_id, opts.clone()); + if let Some(timeout) = opts.timeout { + match tokio::time::timeout(timeout, work).await { + Ok(result) => result, + Err(_) => { + self.abandon_plan(table, plan_id, resolve_wait_options(&opts).2) + .await; + Err(Error::new(ErrorKind::Unexpected, MSG_PLAN_POLL_EXHAUSTED)) + } + } + } else { + work.await + } + } + + async fn wait_for_plan_loop( + &self, + table: &TableIdent, + plan_id: &str, + opts: WaitForPlanOptions, + ) -> Result { + let (min_delay, max_delay, grace, max_retries) = resolve_wait_options(&opts); + let fetch_opts = FetchPlanningResultOptions { + access_delegation: opts.access_delegation.clone(), + }; + let mut sleep = min_delay; + let mut retries = 0u32; + loop { + let retry_after = match self + .fetch_planning_result_raw(table, plan_id, fetch_opts.clone()) + .await + { + Ok(resp) => match resp.status { + PlanStatus::Completed => { + return Ok(CompletedPlanningResult { + status: PlanStatus::Completed, + scan_tasks: resp.scan_tasks, + storage_credentials: resp.storage_credentials, + }); + } + PlanStatus::Submitted => None, + PlanStatus::Failed => return Err(failed_plan_error(resp.error.as_ref())), + PlanStatus::Cancelled => { + return Err(Error::new(ErrorKind::Unexpected, MSG_PLAN_CANCELLED)); + } + }, + Err(PollError::Retry { retry_after, .. }) => retry_after, + Err(PollError::Terminal(err)) => return Err(err), + }; + + if retries >= max_retries { + self.abandon_plan(table, plan_id, grace).await; + return Err(Error::new(ErrorKind::Unexpected, MSG_PLAN_POLL_EXHAUSTED)); + } + retries += 1; + sleep = next_scan_plan_backoff(sleep, min_delay, max_delay); + sleep = apply_retry_after(sleep, retry_after, min_delay, max_delay); + tokio::time::sleep(sleep).await; + } + } + + async fn fetch_planning_result_raw( + &self, + table: &TableIdent, + plan_id: &str, + opts: FetchPlanningResultOptions, + ) -> std::result::Result { + if let Err(err) = require_plan_id(plan_id) { + return Err(PollError::Terminal(err)); + } + if let Err(err) = self + .require_endpoint(&V1_FETCH_PLAN_RESULT, "fetchPlanningResult") + .await + { + return Err(PollError::Terminal(err)); + } + let client = match self.client().await { + Ok(client) => client, + Err(err) => return Err(PollError::Terminal(err)), + }; + let url = format!( + "{}/plan/{}", + client.config.table_endpoint(table), + escape_opaque_path_segment(plan_id) + ); + let mut builder = client.http_client.request(Method::GET, url); + if let Some(delegation) = opts.access_delegation.as_deref() { + builder = builder.header(HEADER_ACCESS_DELEGATION, delegation); + } + let http_response = match client + .query_catalog(HttpRequest::build(builder).map_err(PollError::Terminal)?) + .await + { + Ok(resp) => resp, + Err(err) => return Err(PollError::Terminal(err)), + }; + let status = http_response.status(); + let retry_after = + parse_retry_after(http_response.headers().get(reqwest::header::RETRY_AFTER)); + match status { + StatusCode::OK => { + let resp: FetchPlanningResultResponse = deserialize_catalog_response(http_response) + .await + .map_err(PollError::Terminal)?; + match resp.status { + PlanStatus::Failed => { + Err(PollError::Terminal(failed_plan_error(resp.error.as_ref()))) + } + PlanStatus::Cancelled => Err(PollError::Terminal(Error::new( + ErrorKind::Unexpected, + MSG_PLAN_CANCELLED, + ))), + PlanStatus::Completed | PlanStatus::Submitted => Ok(resp), + } + } + StatusCode::NOT_FOUND => Err(PollError::Terminal( + map_plan_not_found(http_response, PlanNotFoundKind::Fetch).await, + )), + StatusCode::REQUEST_TIMEOUT + | StatusCode::TOO_MANY_REQUESTS + | StatusCode::INTERNAL_SERVER_ERROR + | StatusCode::BAD_GATEWAY + | StatusCode::SERVICE_UNAVAILABLE + | StatusCode::GATEWAY_TIMEOUT => { + let error = deserialize_unexpected_catalog_error( + http_response, + client.http_client.disable_header_redaction(), + ) + .await + .with_retryable(true); + Err(PollError::Retry { retry_after, error }) + } + _ => Err(PollError::Terminal( + deserialize_unexpected_catalog_error( + http_response, + client.http_client.disable_header_redaction(), + ) + .await, + )), + } + } + + async fn require_endpoint(&self, endpoint: &Endpoint, name: &str) -> Result<()> { + if self.supports_endpoint(endpoint).await? { + Ok(()) + } else { + Err(Error::new( + ErrorKind::FeatureUnsupported, + format!("{name} is not advertised by the REST catalog"), + )) + } + } + + async fn abandon_plan(&self, table: &TableIdent, plan_id: &str, grace: Duration) { + let _ = tokio::time::timeout(grace, self.cancel_planning(table, plan_id)).await; + } +} + +enum PollError { + Retry { + retry_after: Option, + error: Error, + }, + Terminal(Error), +} + +enum PlanNotFoundKind { + Submit, + Fetch, + Tasks, +} + +async fn map_plan_not_found(response: Response, kind: PlanNotFoundKind) -> Error { + let bytes = match response.bytes().await { + Ok(bytes) => bytes, + Err(err) => return err.into(), + }; + let err_type = serde_json::from_slice::(&bytes) + .ok() + .map(|parsed| parsed.error_type().to_string()); + match (kind, err_type.as_deref()) { + (_, Some(ERR_TYPE_NO_SUCH_TABLE)) => Error::new( + ErrorKind::TableNotFound, + "Tried to plan a table that does not exist", + ), + (_, Some(ERR_TYPE_NO_SUCH_NAMESPACE)) => Error::new( + ErrorKind::NamespaceNotFound, + "Tried to plan a table in a namespace that does not exist", + ), + (PlanNotFoundKind::Fetch, Some(ERR_TYPE_NO_SUCH_PLAN_ID)) => { + Error::new(ErrorKind::Unexpected, MSG_PLAN_EXPIRED) + } + (PlanNotFoundKind::Tasks, Some(ERR_TYPE_NO_SUCH_PLAN_TASK)) => { + Error::new(ErrorKind::Unexpected, MSG_NO_SUCH_PLAN_TASK) + } + _ => Error::new( + ErrorKind::Unexpected, + "Received response with unexpected status code", + ) + .with_context("status", "404") + .with_context("json", String::from_utf8_lossy(&bytes)), + } +} + +fn failed_plan_error(detail: Option<&ErrorModel>) -> Error { + match detail { + Some(detail) if !detail.message.is_empty() => Error::new( + ErrorKind::Unexpected, + format!("{}: {}", MSG_PLAN_FAILED, detail.message), + ) + .with_context("type", detail.r#type.clone()), + _ => Error::new(ErrorKind::Unexpected, MSG_PLAN_FAILED), + } +} + +fn scan_planning_headers( + idempotency_key: Option<&str>, + access_delegation: Option<&str>, + include_idempotency: bool, +) -> Result> { + let mut headers = Vec::with_capacity(2); + if include_idempotency { + headers.push(( + HEADER_IDEMPOTENCY_KEY, + idempotency_header_value(idempotency_key)?, + )); + } + if let Some(delegation) = access_delegation { + headers.push((HEADER_ACCESS_DELEGATION, delegation.to_string())); + } + Ok(headers) +} + +fn idempotency_header_value(idempotency_key: Option<&str>) -> Result { + match idempotency_key { + None => Ok(Uuid::now_v7().to_string()), + Some(key) => { + let parsed = Uuid::parse_str(key).map_err(|_| { + Error::new( + ErrorKind::DataInvalid, + format!("invalid idempotency key {key:?}"), + ) + })?; + if !parsed.to_string().eq_ignore_ascii_case(key) { + return Err(Error::new( + ErrorKind::DataInvalid, + format!("idempotency key {key:?} must be a canonical hyphenated UUID"), + )); + } + if parsed.get_version() != Some(Version::SortRand) + || parsed.get_variant() != Variant::RFC4122 + { + return Err(Error::new( + ErrorKind::DataInvalid, + format!("idempotency key {key:?} must be an RFC 4122 UUIDv7"), + )); + } + Ok(key.to_string()) + } + } +} + +fn escape_opaque_path_segment(s: &str) -> String { + let mut out = String::new(); + for b in s.bytes() { + match b { + b'A'..=b'Z' | b'a'..=b'z' | b'0'..=b'9' | b'-' | b'_' | b'.' | b'~' => { + out.push(b as char); + } + _ => out.push_str(&format!("%{b:02X}")), + } + } + match out.as_str() { + "." => "%2E".to_string(), + ".." => "%2E%2E".to_string(), + _ => out, + } +} + +fn resolve_wait_options(opts: &WaitForPlanOptions) -> (Duration, Duration, Duration, u32) { + let mut min_delay = opts.min_delay; + let mut max_delay = opts.max_delay; + let mut grace = opts.cancel_grace_period; + let mut max_retries = opts.max_retries; + if min_delay.is_zero() { + min_delay = Duration::from_millis(100); + } + if max_delay.is_zero() { + max_delay = Duration::from_secs(5); + } + if max_delay < min_delay { + max_delay = min_delay; + } + if grace.is_zero() { + grace = Duration::from_secs(5); + } + if max_retries == 0 { + max_retries = 10; + } + (min_delay, max_delay, grace, max_retries) +} + +fn next_scan_plan_backoff(prev: Duration, min_delay: Duration, max_delay: Duration) -> Duration { + if min_delay >= max_delay { + return min_delay; + } + let mut ceiling = max_delay; + if prev <= max_delay / 3 { + ceiling = prev * 3; + if ceiling < min_delay { + ceiling = min_delay; + } + if ceiling > max_delay { + ceiling = max_delay; + } + } + let lo = min_delay.as_nanos().min(u64::MAX as u128) as u64; + let hi = ceiling.as_nanos().min(u64::MAX as u128) as u64; + Duration::from_nanos(rand::rng().random_range(lo..=hi)) +} + +fn apply_retry_after( + backoff: Duration, + retry_after: Option, + min_delay: Duration, + max_delay: Duration, +) -> Duration { + let Some(retry_after) = retry_after.filter(|d| !d.is_zero()) else { + return backoff; + }; + retry_after.max(min_delay).min(max_delay) +} + +fn parse_retry_after(value: Option<&reqwest::header::HeaderValue>) -> Option { + let value = value?.to_str().ok()?.trim(); + let seconds: i64 = value.parse().ok()?; + (seconds > 0).then(|| Duration::from_secs(seconds as u64)) +} + +fn require_plan_id(plan_id: &str) -> Result<()> { + if plan_id.is_empty() { + Err(Error::new(ErrorKind::DataInvalid, "empty plan-id")) + } else { + Ok(()) + } +} + +#[cfg(test)] +mod tests { + use iceberg::{NamespaceIdent, Runtime, SessionContext}; + use mockito::Server; + use serde_json::json; + use uuid::Uuid; + + use super::*; + use crate::catalog::RestCatalogConfig; + + fn table() -> TableIdent { + TableIdent::new(NamespaceIdent::new("ns".into()), "tbl".into()) + } + + fn catalog(uri: &str) -> RestCatalog { + RestCatalog::new( + SessionContext::empty(), + RestCatalogConfig::builder().uri(uri.to_string()).build(), + None, + None, + Runtime::current(), + None, + ) + } + + async fn config_with_endpoints( + server: &mut mockito::ServerGuard, + endpoints: &[&str], + ) -> mockito::Mock { + let body = json!({ + "overrides": {}, + "defaults": {}, + "endpoints": endpoints, + }); + server + .mock("GET", "/v1/config") + .with_status(200) + .with_body(body.to_string()) + .create_async() + .await + } + + const ALL_PLAN: &[&str] = &[ + "POST /v1/{prefix}/namespaces/{namespace}/tables/{table}/plan", + "GET /v1/{prefix}/namespaces/{namespace}/tables/{table}/plan/{plan-id}", + "DELETE /v1/{prefix}/namespaces/{namespace}/tables/{table}/plan/{plan-id}", + "POST /v1/{prefix}/namespaces/{namespace}/tables/{table}/tasks", + ]; + + #[test] + fn plan_table_scan_requires_plan_id_on_completed_and_submitted() { + let completed = json!({"status": "completed"}); + assert!(serde_json::from_value::(completed).is_err()); + let submitted = json!({"status": "submitted", "plan-id": "p1"}); + let parsed: PlanTableScanResponse = serde_json::from_value(submitted).unwrap(); + assert_eq!(parsed.status, PlanStatus::Submitted); + assert_eq!(parsed.plan_id.as_deref(), Some("p1")); + } + + #[test] + fn plan_table_scan_rejects_cancelled() { + let body = json!({"status": "cancelled", "plan-id": "p1"}); + assert!(serde_json::from_value::(body).is_err()); + } + + #[test] + fn fetch_result_accepts_cancelled_and_failed_without_error() { + let cancelled: FetchPlanningResultResponse = + serde_json::from_value(json!({"status": "cancelled"})).unwrap(); + assert_eq!(cancelled.status, PlanStatus::Cancelled); + let failed: FetchPlanningResultResponse = + serde_json::from_value(json!({"status": "failed", "error": "not-an-object"})).unwrap(); + assert_eq!(failed.status, PlanStatus::Failed); + assert!(failed.error.is_none()); + } + + #[test] + fn completed_plan_round_trips_opaque_tasks_and_credentials() { + let body = json!({ + "status": "completed", + "plan-id": "p1", + "plan-tasks": ["t1"], + "file-scan-tasks": [{"data-file": {"file-path": "s3://b/f.parquet"}}], + "storage-credentials": [{"prefix": "s3://b/", "config": {"s3.access-key-id": "k"}}] + }); + let parsed: PlanTableScanResponse = serde_json::from_value(body.clone()).unwrap(); + assert_eq!(parsed.scan_tasks.plan_tasks, ["t1"]); + assert_eq!( + parsed.scan_tasks.file_scan_tasks[0].data_file["file-path"], + "s3://b/f.parquet" + ); + assert_eq!( + parsed.storage_credentials.as_ref().unwrap()[0].prefix, + "s3://b/" + ); + let encoded = serde_json::to_value(&parsed).unwrap(); + assert_eq!(encoded["plan-id"], "p1"); + assert_eq!(encoded["plan-tasks"][0], "t1"); + } + + #[test] + fn plan_request_omits_header_fields_from_json() { + let req = PlanTableScanRequest { + idempotency_key: Some("not-on-wire".into()), + access_delegation: Some("vended-credentials".into()), + snapshot_id: Some(7), + ..Default::default() + }; + let value = serde_json::to_value(&req).unwrap(); + assert_eq!(value["snapshot-id"], 7); + assert!(value.get("idempotency-key").is_none()); + assert!(value.get("access-delegation").is_none()); + } + + #[test] + fn idempotency_key_must_be_canonical_uuidv7() { + let v7 = Uuid::now_v7().to_string(); + assert!(idempotency_header_value(Some(&v7)).is_ok()); + assert!(idempotency_header_value(Some(&v7.to_uppercase())).is_ok()); + let v4 = Uuid::new_v4().to_string(); + assert!(idempotency_header_value(Some(&v4)).is_err()); + assert!(idempotency_header_value(Some("not-a-uuid")).is_err()); + let unhyphenated: String = v7.chars().filter(|c| *c != '-').collect(); + assert!(idempotency_header_value(Some(&unhyphenated)).is_err()); + } + + #[test] + fn opaque_plan_id_is_a_single_path_segment() { + assert_eq!(escape_opaque_path_segment("a/b"), "a%2Fb"); + assert_eq!(escape_opaque_path_segment("."), "%2E"); + assert_eq!(escape_opaque_path_segment(".."), "%2E%2E"); + assert_eq!(escape_opaque_path_segment("plain"), "plain"); + } + + #[tokio::test] + async fn default_config_does_not_advertise_plan() { + let mut server = Server::new_async().await; + let config = config_with_endpoints(&mut server, &[]).await; + // Empty endpoints list falls back to DEFAULT_ENDPOINTS. + let catalog = catalog(&server.url()); + assert!(!catalog.supports_plan_table_scan().await.unwrap()); + assert!(!catalog.supports_full_remote_scan_planning().await.unwrap()); + assert!(!catalog.supports_remote_scan_planning().await.unwrap()); + config.assert_async().await; + } + + #[tokio::test] + async fn remote_scan_planning_stays_false_when_all_endpoints_are_advertised() { + let mut server = Server::new_async().await; + let config = config_with_endpoints(&mut server, ALL_PLAN).await; + let catalog = catalog(&server.url()); + assert!(catalog.supports_plan_table_scan().await.unwrap()); + assert!(catalog.supports_full_remote_scan_planning().await.unwrap()); + assert!(!catalog.supports_remote_scan_planning().await.unwrap()); + config.assert_async().await; + } + + #[tokio::test] + async fn plan_table_scan_is_feature_unsupported_when_not_advertised() { + let mut server = Server::new_async().await; + let config = config_with_endpoints(&mut server, &["GET /v1/{prefix}/namespaces"]).await; + let catalog = catalog(&server.url()); + let err = catalog + .plan_table_scan(&table(), PlanTableScanRequest::default()) + .await + .unwrap_err(); + assert_eq!(err.kind(), ErrorKind::FeatureUnsupported); + config.assert_async().await; + } + + #[tokio::test] + async fn plan_table_scan_completed_sends_uuidv7_idempotency_key() { + let mut server = Server::new_async().await; + let config = config_with_endpoints(&mut server, ALL_PLAN).await; + let plan = server + .mock("POST", "/v1/namespaces/ns/tables/tbl/plan") + .match_header( + "idempotency-key", + mockito::Matcher::Regex( + r"^[0-9a-fA-F]{8}-[0-9a-fA-F]{4}-7[0-9a-fA-F]{3}-[89abAB][0-9a-fA-F]{3}-[0-9a-fA-F]{12}$" + .into(), + ), + ) + .with_status(200) + .with_body(r#"{"status":"completed","plan-id":"p1","file-scan-tasks":[]}"#) + .create_async() + .await; + let catalog = catalog(&server.url()); + let resp = catalog + .plan_table_scan(&table(), PlanTableScanRequest::default()) + .await + .unwrap(); + assert_eq!(resp.status, PlanStatus::Completed); + assert_eq!(resp.plan_id.as_deref(), Some("p1")); + config.assert_async().await; + plan.assert_async().await; + } + + #[tokio::test] + async fn plan_table_scan_failed_is_an_error() { + let mut server = Server::new_async().await; + let config = config_with_endpoints(&mut server, ALL_PLAN).await; + let plan = server + .mock("POST", "/v1/namespaces/ns/tables/tbl/plan") + .with_status(200) + .with_body(r#"{"status":"failed","error":{"message":"boom","type":"IcebergException","code":500}}"#) + .create_async() + .await; + let catalog = catalog(&server.url()); + let err = catalog + .plan_table_scan(&table(), PlanTableScanRequest::default()) + .await + .unwrap_err(); + assert!(is_plan_failed(&err)); + assert!(err.message().contains("boom")); + config.assert_async().await; + plan.assert_async().await; + } + + #[tokio::test] + async fn plan_404_splits_table_from_unrecognized() { + let mut server = Server::new_async().await; + let config = config_with_endpoints(&mut server, ALL_PLAN).await; + let missing = server + .mock("POST", "/v1/namespaces/ns/tables/tbl/plan") + .with_status(404) + .with_body(r#"{"error":{"message":"gone","type":"NoSuchTableException","code":404}}"#) + .create_async() + .await; + let catalog = catalog(&server.url()); + let err = catalog + .plan_table_scan(&table(), PlanTableScanRequest::default()) + .await + .unwrap_err(); + assert_eq!(err.kind(), ErrorKind::TableNotFound); + config.assert_async().await; + missing.assert_async().await; + } + + #[tokio::test] + async fn fetch_result_expired_plan_id() { + let mut server = Server::new_async().await; + let config = config_with_endpoints(&mut server, ALL_PLAN).await; + let fetch = server + .mock("GET", "/v1/namespaces/ns/tables/tbl/plan/p1") + .with_status(404) + .with_body( + r#"{"error":{"message":"expired","type":"NoSuchPlanIdException","code":404}}"#, + ) + .create_async() + .await; + let catalog = catalog(&server.url()); + let err = catalog + .fetch_planning_result(&table(), "p1", FetchPlanningResultOptions::default()) + .await + .unwrap_err(); + assert!(is_plan_expired(&err)); + config.assert_async().await; + fetch.assert_async().await; + } + + #[tokio::test] + async fn fetch_scan_tasks_and_empty_body() { + let mut server = Server::new_async().await; + let config = config_with_endpoints(&mut server, ALL_PLAN).await; + let tasks = server + .mock("POST", "/v1/namespaces/ns/tables/tbl/tasks") + .match_body(r#"{"plan-task":"h1"}"#) + .with_status(200) + .with_body( + r#"{"plan-tasks":["h2"],"file-scan-tasks":[{"data-file":{"file-path":"f"}}]}"#, + ) + .create_async() + .await; + let catalog = catalog(&server.url()); + let resp = catalog + .fetch_scan_tasks(&table(), FetchScanTasksRequest { + idempotency_key: None, + plan_task: "h1".into(), + }) + .await + .unwrap(); + assert_eq!(resp.scan_tasks.plan_tasks, ["h2"]); + config.assert_async().await; + tasks.assert_async().await; + } + + #[tokio::test] + async fn fetch_scan_tasks_rejects_empty_200() { + let mut server = Server::new_async().await; + let config = config_with_endpoints(&mut server, ALL_PLAN).await; + let tasks = server + .mock("POST", "/v1/namespaces/ns/tables/tbl/tasks") + .with_status(200) + .with_body("") + .create_async() + .await; + let catalog = catalog(&server.url()); + let err = catalog + .fetch_scan_tasks(&table(), FetchScanTasksRequest { + idempotency_key: None, + plan_task: "h1".into(), + }) + .await + .unwrap_err(); + assert_eq!(err.kind(), ErrorKind::Unexpected); + config.assert_async().await; + tasks.assert_async().await; + } + + #[tokio::test] + async fn fetch_scan_tasks_no_such_plan_task() { + let mut server = Server::new_async().await; + let config = config_with_endpoints(&mut server, ALL_PLAN).await; + let tasks = server + .mock("POST", "/v1/namespaces/ns/tables/tbl/tasks") + .with_status(404) + .with_body( + r#"{"error":{"message":"gone","type":"NoSuchPlanTaskException","code":404}}"#, + ) + .create_async() + .await; + let catalog = catalog(&server.url()); + let err = catalog + .fetch_scan_tasks(&table(), FetchScanTasksRequest { + idempotency_key: None, + plan_task: "h1".into(), + }) + .await + .unwrap_err(); + assert!(is_no_such_plan_task(&err)); + config.assert_async().await; + tasks.assert_async().await; + } + + #[tokio::test] + async fn wait_for_plan_polls_submitted_then_completes() { + let mut server = Server::new_async().await; + let config = config_with_endpoints(&mut server, ALL_PLAN).await; + let first = server + .mock("GET", "/v1/namespaces/ns/tables/tbl/plan/p1") + .with_status(200) + .with_body(r#"{"status":"submitted"}"#) + .expect(1) + .create_async() + .await; + let second = server + .mock("GET", "/v1/namespaces/ns/tables/tbl/plan/p1") + .with_status(200) + .with_body(r#"{"status":"completed","plan-tasks":["h1"]}"#) + .expect(1) + .create_async() + .await; + let catalog = catalog(&server.url()); + let result = catalog + .wait_for_plan(&table(), "p1", WaitForPlanOptions { + min_delay: Duration::from_millis(1), + max_delay: Duration::from_millis(1), + ..Default::default() + }) + .await + .unwrap(); + assert_eq!(result.scan_tasks.plan_tasks, ["h1"]); + config.assert_async().await; + first.assert_async().await; + second.assert_async().await; + } + + #[tokio::test] + async fn wait_for_plan_retries_503_then_completes() { + let mut server = Server::new_async().await; + let config = config_with_endpoints(&mut server, ALL_PLAN).await; + let busy = server + .mock("GET", "/v1/namespaces/ns/tables/tbl/plan/p1") + .with_status(503) + .with_header("Retry-After", "0") + .expect(1) + .create_async() + .await; + let done = server + .mock("GET", "/v1/namespaces/ns/tables/tbl/plan/p1") + .with_status(200) + .with_body(r#"{"status":"completed"}"#) + .expect(1) + .create_async() + .await; + let catalog = catalog(&server.url()); + let result = catalog + .wait_for_plan(&table(), "p1", WaitForPlanOptions { + min_delay: Duration::from_millis(1), + max_delay: Duration::from_millis(1), + ..Default::default() + }) + .await + .unwrap(); + assert_eq!(result.status, PlanStatus::Completed); + config.assert_async().await; + busy.assert_async().await; + done.assert_async().await; + } + + #[tokio::test] + async fn wait_for_plan_cancels_after_max_retries() { + let mut server = Server::new_async().await; + let config = config_with_endpoints(&mut server, ALL_PLAN).await; + let poll = server + .mock("GET", "/v1/namespaces/ns/tables/tbl/plan/p1") + .with_status(200) + .with_body(r#"{"status":"submitted"}"#) + .expect(2) + .create_async() + .await; + let cancel = server + .mock("DELETE", "/v1/namespaces/ns/tables/tbl/plan/p1") + .with_status(204) + .expect(1) + .create_async() + .await; + let catalog = catalog(&server.url()); + let err = catalog + .wait_for_plan(&table(), "p1", WaitForPlanOptions { + min_delay: Duration::from_millis(1), + max_delay: Duration::from_millis(1), + max_retries: 1, + cancel_grace_period: Duration::from_secs(1), + ..Default::default() + }) + .await + .unwrap_err(); + assert!(is_plan_poll_exhausted(&err)); + config.assert_async().await; + poll.assert_async().await; + cancel.assert_async().await; + } + + #[tokio::test] + async fn wait_for_plan_rejects_empty_plan_id() { + let catalog = catalog("http://127.0.0.1:1"); + let err = catalog + .wait_for_plan(&table(), "", WaitForPlanOptions::default()) + .await + .unwrap_err(); + assert_eq!(err.kind(), ErrorKind::DataInvalid); + } + + #[tokio::test] + async fn plan_id_with_slash_stays_one_path_segment() { + let mut server = Server::new_async().await; + let config = config_with_endpoints(&mut server, ALL_PLAN).await; + let fetch = server + .mock("GET", "/v1/namespaces/ns/tables/tbl/plan/a%2Fb") + .with_status(200) + .with_body(r#"{"status":"submitted"}"#) + .create_async() + .await; + let catalog = catalog(&server.url()); + let resp = catalog + .fetch_planning_result(&table(), "a/b", FetchPlanningResultOptions::default()) + .await + .unwrap(); + assert_eq!(resp.status, PlanStatus::Submitted); + config.assert_async().await; + fetch.assert_async().await; + } + + #[tokio::test] + async fn cancel_planning_accepts_204() { + let mut server = Server::new_async().await; + let config = config_with_endpoints(&mut server, ALL_PLAN).await; + let cancel = server + .mock("DELETE", "/v1/namespaces/ns/tables/tbl/plan/p1") + .with_status(204) + .create_async() + .await; + let catalog = catalog(&server.url()); + catalog.cancel_planning(&table(), "p1").await.unwrap(); + config.assert_async().await; + cancel.assert_async().await; + } +} diff --git a/crates/catalog/rest/src/types.rs b/crates/catalog/rest/src/types.rs index 390521229e..50cdb68a09 100644 --- a/crates/catalog/rest/src/types.rs +++ b/crates/catalog/rest/src/types.rs @@ -42,6 +42,12 @@ pub struct ErrorResponse { error: ErrorModel, } +impl ErrorResponse { + pub(crate) fn error_type(&self) -> &str { + &self.error.r#type + } +} + impl From for Error { fn from(resp: ErrorResponse) -> Error { resp.error.into() From d7abfbda7f30118a112c134f8afde2aa2943e2d2 Mon Sep 17 00:00:00 2001 From: David Dallakyan Date: Sun, 16 Aug 2026 16:51:55 +0400 Subject: [PATCH 2/4] fix(catalog-rest): cancel wait_for_plan on drop and bound Retry-After A huge Retry-After used to panic Duration::from_secs. Ignore overflow, await cancel when the poller times out, and DELETE the plan if the wait future is dropped. --- crates/catalog/rest/src/catalog.rs | 16 ++ crates/catalog/rest/src/scan_planning.rs | 259 +++++++++++++++++++++-- 2 files changed, 261 insertions(+), 14 deletions(-) diff --git a/crates/catalog/rest/src/catalog.rs b/crates/catalog/rest/src/catalog.rs index b165b2cdb0..056a40a43c 100644 --- a/crates/catalog/rest/src/catalog.rs +++ b/crates/catalog/rest/src/catalog.rs @@ -724,6 +724,22 @@ impl RestSessionCatalog { } } + /// Same catalog identity with an empty HTTP client cache, for fire-and-forget + /// work (best-effort plan cancel on drop) that must not borrow `self`. + pub(crate) fn clone_uninitialized(&self) -> Self { + Self::new( + self.user_config.clone(), + self.auth_manager.clone(), + self.storage_factory.clone(), + self.runtime.clone(), + self.kms_client.clone(), + ) + } + + pub(crate) fn runtime(&self) -> &Runtime { + &self.runtime + } + /// Sends a DELETE request for the given table, optionally requesting purge. async fn delete_table( &self, diff --git a/crates/catalog/rest/src/scan_planning.rs b/crates/catalog/rest/src/scan_planning.rs index 7e7439bb66..cf7f4225a6 100644 --- a/crates/catalog/rest/src/scan_planning.rs +++ b/crates/catalog/rest/src/scan_planning.rs @@ -222,7 +222,9 @@ pub struct WaitForPlanOptions { pub max_delay: Duration, /// Bound on the best-effort cancel after giving up. Zero uses 5s. pub cancel_grace_period: Duration, - /// Poll attempts after the first. Zero uses 10. + /// Poll attempts after the first. Zero uses 10 when [`Self::timeout`] is + /// `None`. When a timeout is set, zero means keep polling until the + /// deadline. pub max_retries: u32, /// Optional overall deadline. `None` relies on `max_retries`. pub timeout: Option, @@ -495,19 +497,24 @@ impl RestCatalog { self.require_endpoint(&V1_FETCH_PLAN_RESULT, "fetchPlanningResult") .await?; + let (_, _, grace, _, _) = resolve_wait_options(&opts); + let mut guard = + PlanAbandonGuard::new(self.clone_uninitialized(), table.clone(), plan_id, grace); + let work = self.wait_for_plan_loop(table, plan_id, opts.clone()); - if let Some(timeout) = opts.timeout { + let result = if let Some(timeout) = opts.timeout { match tokio::time::timeout(timeout, work).await { Ok(result) => result, Err(_) => { - self.abandon_plan(table, plan_id, resolve_wait_options(&opts).2) - .await; + self.abandon_plan(table, plan_id, grace).await; Err(Error::new(ErrorKind::Unexpected, MSG_PLAN_POLL_EXHAUSTED)) } } } else { work.await - } + }; + guard.disarm(); + result } async fn wait_for_plan_loop( @@ -516,7 +523,8 @@ impl RestCatalog { plan_id: &str, opts: WaitForPlanOptions, ) -> Result { - let (min_delay, max_delay, grace, max_retries) = resolve_wait_options(&opts); + let (min_delay, max_delay, grace, max_retries, clamp_retry_after) = + resolve_wait_options(&opts); let fetch_opts = FetchPlanningResultOptions { access_delegation: opts.access_delegation.clone(), }; @@ -545,13 +553,13 @@ impl RestCatalog { Err(PollError::Terminal(err)) => return Err(err), }; - if retries >= max_retries { + if max_retries > 0 && retries >= max_retries { self.abandon_plan(table, plan_id, grace).await; return Err(Error::new(ErrorKind::Unexpected, MSG_PLAN_POLL_EXHAUSTED)); } retries += 1; sleep = next_scan_plan_backoff(sleep, min_delay, max_delay); - sleep = apply_retry_after(sleep, retry_after, min_delay, max_delay); + sleep = apply_retry_after(sleep, retry_after, min_delay, max_delay, clamp_retry_after); tokio::time::sleep(sleep).await; } } @@ -653,6 +661,45 @@ impl RestCatalog { } } +/// Best-effort DELETE of a submitted plan if [`RestCatalog::wait_for_plan`] is +/// dropped (caller timeout, cancellation) before it returns. +struct PlanAbandonGuard { + catalog: Option, + table: TableIdent, + plan_id: String, + grace: Duration, +} + +impl PlanAbandonGuard { + fn new(catalog: RestCatalog, table: TableIdent, plan_id: &str, grace: Duration) -> Self { + Self { + catalog: Some(catalog), + table, + plan_id: plan_id.to_string(), + grace, + } + } + + fn disarm(&mut self) { + self.catalog = None; + } +} + +impl Drop for PlanAbandonGuard { + fn drop(&mut self) { + let Some(catalog) = self.catalog.take() else { + return; + }; + let table = self.table.clone(); + let plan_id = self.plan_id.clone(); + let grace = self.grace; + let runtime = catalog.runtime().clone(); + drop(runtime.io().spawn(async move { + let _ = tokio::time::timeout(grace, catalog.cancel_planning(&table, &plan_id)).await; + })); + } +} + enum PollError { Retry { retry_after: Option, @@ -774,7 +821,7 @@ fn escape_opaque_path_segment(s: &str) -> String { } } -fn resolve_wait_options(opts: &WaitForPlanOptions) -> (Duration, Duration, Duration, u32) { +fn resolve_wait_options(opts: &WaitForPlanOptions) -> (Duration, Duration, Duration, u32, bool) { let mut min_delay = opts.min_delay; let mut max_delay = opts.max_delay; let mut grace = opts.cancel_grace_period; @@ -791,10 +838,11 @@ fn resolve_wait_options(opts: &WaitForPlanOptions) -> (Duration, Duration, Durat if grace.is_zero() { grace = Duration::from_secs(5); } - if max_retries == 0 { + if max_retries == 0 && opts.timeout.is_none() { max_retries = 10; } - (min_delay, max_delay, grace, max_retries) + let clamp_retry_after = opts.timeout.is_none(); + (min_delay, max_delay, grace, max_retries, clamp_retry_after) } fn next_scan_plan_backoff(prev: Duration, min_delay: Duration, max_delay: Duration) -> Duration { @@ -821,17 +869,31 @@ fn apply_retry_after( retry_after: Option, min_delay: Duration, max_delay: Duration, + clamp_to_max: bool, ) -> Duration { let Some(retry_after) = retry_after.filter(|d| !d.is_zero()) else { return backoff; }; - retry_after.max(min_delay).min(max_delay) + let delay = retry_after.max(min_delay); + if clamp_to_max { + delay.min(max_delay) + } else { + delay + } } fn parse_retry_after(value: Option<&reqwest::header::HeaderValue>) -> Option { let value = value?.to_str().ok()?.trim(); - let seconds: i64 = value.parse().ok()?; - (seconds > 0).then(|| Duration::from_secs(seconds as u64)) + let seconds: u64 = value.parse().ok()?; + if seconds == 0 { + return None; + } + // Duration::from_secs panics when secs * 1e9 overflows u64. + const MAX_SECS: u64 = u64::MAX / 1_000_000_000; + if seconds > MAX_SECS { + return None; + } + Some(Duration::from_secs(seconds)) } fn require_plan_id(plan_id: &str) -> Result<()> { @@ -976,6 +1038,28 @@ mod tests { assert_eq!(escape_opaque_path_segment("plain"), "plain"); } + #[test] + fn parse_retry_after_ignores_zero_and_overflow() { + let zero = reqwest::header::HeaderValue::from_static("0"); + assert!(parse_retry_after(Some(&zero)).is_none()); + let huge = reqwest::header::HeaderValue::from_static("99999999999"); + assert!(parse_retry_after(Some(&huge)).is_none()); + let ok = reqwest::header::HeaderValue::from_static("2"); + assert_eq!(parse_retry_after(Some(&ok)), Some(Duration::from_secs(2))); + } + + #[test] + fn apply_retry_after_skips_max_clamp_when_a_deadline_is_set() { + let retry = Some(Duration::from_secs(30)); + let min = Duration::from_millis(100); + let max = Duration::from_secs(5); + assert_eq!(apply_retry_after(min, retry, min, max, true), max); + assert_eq!( + apply_retry_after(min, retry, min, max, false), + Duration::from_secs(30) + ); + } + #[tokio::test] async fn default_config_does_not_advertise_plan() { let mut server = Server::new_async().await; @@ -1277,6 +1361,153 @@ mod tests { cancel.assert_async().await; } + #[tokio::test] + async fn wait_for_plan_cancels_after_timeout() { + let mut server = Server::new_async().await; + let config = config_with_endpoints(&mut server, ALL_PLAN).await; + let poll = server + .mock("GET", "/v1/namespaces/ns/tables/tbl/plan/p1") + .with_status(200) + .with_body(r#"{"status":"submitted"}"#) + .expect_at_least(1) + .create_async() + .await; + let cancel = server + .mock("DELETE", "/v1/namespaces/ns/tables/tbl/plan/p1") + .with_status(204) + .expect(1) + .create_async() + .await; + let catalog = catalog(&server.url()); + let err = catalog + .wait_for_plan(&table(), "p1", WaitForPlanOptions { + min_delay: Duration::from_millis(1), + max_delay: Duration::from_millis(1), + timeout: Some(Duration::from_millis(80)), + max_retries: 0, + cancel_grace_period: Duration::from_secs(1), + ..Default::default() + }) + .await + .unwrap_err(); + assert!(is_plan_poll_exhausted(&err)); + config.assert_async().await; + poll.assert_async().await; + cancel.assert_async().await; + } + + #[tokio::test] + async fn wait_for_plan_cancels_when_future_is_dropped() { + let mut server = Server::new_async().await; + let body = json!({ + "overrides": {}, + "defaults": {}, + "endpoints": ALL_PLAN, + }); + let config = server + .mock("GET", "/v1/config") + .with_status(200) + .with_body(body.to_string()) + .expect_at_least(1) + .create_async() + .await; + let poll = server + .mock("GET", "/v1/namespaces/ns/tables/tbl/plan/p1") + .with_status(200) + .with_body(r#"{"status":"submitted"}"#) + .expect_at_least(1) + .create_async() + .await; + let cancel = server + .mock("DELETE", "/v1/namespaces/ns/tables/tbl/plan/p1") + .with_status(204) + .expect(1) + .create_async() + .await; + let catalog = catalog(&server.url()); + let table = table(); + let wait = catalog.wait_for_plan(&table, "p1", WaitForPlanOptions { + min_delay: Duration::from_millis(1), + max_delay: Duration::from_millis(1), + max_retries: 50, + cancel_grace_period: Duration::from_secs(1), + ..Default::default() + }); + let _ = tokio::time::timeout(Duration::from_millis(80), wait).await; + tokio::time::sleep(Duration::from_millis(250)).await; + config.assert_async().await; + poll.assert_async().await; + cancel.assert_async().await; + } + + #[tokio::test] + async fn wait_for_plan_timeout_zero_retries_keeps_polling() { + let mut server = Server::new_async().await; + let config = config_with_endpoints(&mut server, ALL_PLAN).await; + let pending = server + .mock("GET", "/v1/namespaces/ns/tables/tbl/plan/p1") + .with_status(200) + .with_body(r#"{"status":"submitted"}"#) + .expect(11) + .create_async() + .await; + let done = server + .mock("GET", "/v1/namespaces/ns/tables/tbl/plan/p1") + .with_status(200) + .with_body(r#"{"status":"completed"}"#) + .expect(1) + .create_async() + .await; + let catalog = catalog(&server.url()); + let result = catalog + .wait_for_plan(&table(), "p1", WaitForPlanOptions { + min_delay: Duration::from_millis(1), + max_delay: Duration::from_millis(1), + max_retries: 0, + timeout: Some(Duration::from_secs(5)), + ..Default::default() + }) + .await + .unwrap(); + assert_eq!(result.status, PlanStatus::Completed); + config.assert_async().await; + pending.assert_async().await; + done.assert_async().await; + } + + #[tokio::test] + async fn wait_for_plan_huge_retry_after_does_not_panic() { + let mut server = Server::new_async().await; + let config = config_with_endpoints(&mut server, ALL_PLAN).await; + let busy = server + .mock("GET", "/v1/namespaces/ns/tables/tbl/plan/p1") + .with_status(503) + .with_header("Retry-After", "99999999999") + .expect(1) + .create_async() + .await; + let done = server + .mock("GET", "/v1/namespaces/ns/tables/tbl/plan/p1") + .with_status(200) + .with_body(r#"{"status":"completed"}"#) + .expect(1) + .create_async() + .await; + let catalog = catalog(&server.url()); + let result = catalog + .wait_for_plan(&table(), "p1", WaitForPlanOptions { + min_delay: Duration::from_millis(1), + max_delay: Duration::from_millis(1), + ..Default::default() + }) + .await + .unwrap(); + assert_eq!(result.status, PlanStatus::Completed); + config.assert_async().await; + busy.assert_async().await; + done.assert_async().await; + } + #[tokio::test] async fn wait_for_plan_rejects_empty_plan_id() { let catalog = catalog("http://127.0.0.1:1"); From be82b93b0cd846515ed245ef022f950df7c2ecc4 Mon Sep 17 00:00:00 2001 From: David Dallakyan Date: Sun, 16 Aug 2026 17:08:56 +0400 Subject: [PATCH 3/4] fix(catalog-rest): honor HTTP-date Retry-After on plan polls RFC 9110 allows both delta-seconds and IMF-fixdate. Ignoring the date form made wait_for_plan retry immediately against a server that asked us to wait. Tests also cover cancelled/failed/expired poll outcomes and the namespace 404 split. --- crates/catalog/rest/src/scan_planning.rs | 218 +++++++++++++++++++++-- 1 file changed, 208 insertions(+), 10 deletions(-) diff --git a/crates/catalog/rest/src/scan_planning.rs b/crates/catalog/rest/src/scan_planning.rs index cf7f4225a6..5eb93fa951 100644 --- a/crates/catalog/rest/src/scan_planning.rs +++ b/crates/catalog/rest/src/scan_planning.rs @@ -884,16 +884,25 @@ fn apply_retry_after( fn parse_retry_after(value: Option<&reqwest::header::HeaderValue>) -> Option { let value = value?.to_str().ok()?.trim(); - let seconds: u64 = value.parse().ok()?; - if seconds == 0 { - return None; - } - // Duration::from_secs panics when secs * 1e9 overflows u64. - const MAX_SECS: u64 = u64::MAX / 1_000_000_000; - if seconds > MAX_SECS { - return None; - } - Some(Duration::from_secs(seconds)) + if let Ok(seconds) = value.parse::() { + if seconds == 0 { + return None; + } + // Duration::from_secs panics when secs * 1e9 overflows u64. + const MAX_SECS: u64 = u64::MAX / 1_000_000_000; + if seconds > MAX_SECS { + return None; + } + return Some(Duration::from_secs(seconds)); + } + // RFC 9110 HTTP-date, IMF-fixdate form (RFC 2822). Obsolete RFC 850 / + // asctime values are ignored rather than adding another parser. + let retry_at = chrono::DateTime::parse_from_rfc2822(value).ok()?; + retry_at + .signed_duration_since(chrono::Utc::now()) + .to_std() + .ok() + .filter(|d| !d.is_zero()) } fn require_plan_id(plan_id: &str) -> Result<()> { @@ -1046,6 +1055,22 @@ mod tests { assert!(parse_retry_after(Some(&huge)).is_none()); let ok = reqwest::header::HeaderValue::from_static("2"); assert_eq!(parse_retry_after(Some(&ok)), Some(Duration::from_secs(2))); + let padded = reqwest::header::HeaderValue::from_static(" 2 "); + assert_eq!( + parse_retry_after(Some(&padded)), + Some(Duration::from_secs(2)) + ); + let invalid = reqwest::header::HeaderValue::from_static("not-a-date"); + assert!(parse_retry_after(Some(&invalid)).is_none()); + } + + #[test] + fn parse_retry_after_accepts_http_date() { + let past = reqwest::header::HeaderValue::from_static("Wed, 21 Oct 2015 07:28:00 GMT"); + assert!(parse_retry_after(Some(&past)).is_none()); + let future = reqwest::header::HeaderValue::from_static("Sun, 16 Aug 2099 12:00:00 GMT"); + let delay = parse_retry_after(Some(&future)).unwrap(); + assert!(delay > Duration::from_secs(60 * 60 * 24 * 365)); } #[test] @@ -1165,6 +1190,68 @@ mod tests { missing.assert_async().await; } + #[tokio::test] + async fn plan_404_splits_namespace() { + let mut server = Server::new_async().await; + let config = config_with_endpoints(&mut server, ALL_PLAN).await; + let missing = server + .mock("POST", "/v1/namespaces/ns/tables/tbl/plan") + .with_status(404) + .with_body( + r#"{"error":{"message":"gone","type":"NoSuchNamespaceException","code":404}}"#, + ) + .create_async() + .await; + let catalog = catalog(&server.url()); + let err = catalog + .plan_table_scan(&table(), PlanTableScanRequest::default()) + .await + .unwrap_err(); + assert_eq!(err.kind(), ErrorKind::NamespaceNotFound); + config.assert_async().await; + missing.assert_async().await; + } + + #[tokio::test] + async fn fetch_planning_result_cancelled_is_an_error() { + let mut server = Server::new_async().await; + let config = config_with_endpoints(&mut server, ALL_PLAN).await; + let fetch = server + .mock("GET", "/v1/namespaces/ns/tables/tbl/plan/p1") + .with_status(200) + .with_body(r#"{"status":"cancelled"}"#) + .create_async() + .await; + let catalog = catalog(&server.url()); + let err = catalog + .fetch_planning_result(&table(), "p1", FetchPlanningResultOptions::default()) + .await + .unwrap_err(); + assert!(is_plan_cancelled(&err)); + config.assert_async().await; + fetch.assert_async().await; + } + + #[tokio::test] + async fn fetch_planning_result_failed_is_an_error() { + let mut server = Server::new_async().await; + let config = config_with_endpoints(&mut server, ALL_PLAN).await; + let fetch = server + .mock("GET", "/v1/namespaces/ns/tables/tbl/plan/p1") + .with_status(200) + .with_body(r#"{"status":"failed","error":{"message":"boom","type":"IcebergException","code":500}}"#) + .create_async() + .await; + let catalog = catalog(&server.url()); + let err = catalog + .fetch_planning_result(&table(), "p1", FetchPlanningResultOptions::default()) + .await + .unwrap_err(); + assert!(is_plan_failed(&err)); + config.assert_async().await; + fetch.assert_async().await; + } + #[tokio::test] async fn fetch_result_expired_plan_id() { let mut server = Server::new_async().await; @@ -1294,6 +1381,117 @@ mod tests { second.assert_async().await; } + #[tokio::test] + async fn wait_for_plan_propagates_cancelled() { + let mut server = Server::new_async().await; + let config = config_with_endpoints(&mut server, ALL_PLAN).await; + let poll = server + .mock("GET", "/v1/namespaces/ns/tables/tbl/plan/p1") + .with_status(200) + .with_body(r#"{"status":"cancelled"}"#) + .expect(1) + .create_async() + .await; + let catalog = catalog(&server.url()); + let err = catalog + .wait_for_plan(&table(), "p1", WaitForPlanOptions { + min_delay: Duration::from_millis(1), + max_delay: Duration::from_millis(1), + ..Default::default() + }) + .await + .unwrap_err(); + assert!(is_plan_cancelled(&err)); + config.assert_async().await; + poll.assert_async().await; + } + + #[tokio::test] + async fn wait_for_plan_propagates_expired() { + let mut server = Server::new_async().await; + let config = config_with_endpoints(&mut server, ALL_PLAN).await; + let poll = server + .mock("GET", "/v1/namespaces/ns/tables/tbl/plan/p1") + .with_status(404) + .with_body( + r#"{"error":{"message":"expired","type":"NoSuchPlanIdException","code":404}}"#, + ) + .expect(1) + .create_async() + .await; + let catalog = catalog(&server.url()); + let err = catalog + .wait_for_plan(&table(), "p1", WaitForPlanOptions { + min_delay: Duration::from_millis(1), + max_delay: Duration::from_millis(1), + ..Default::default() + }) + .await + .unwrap_err(); + assert!(is_plan_expired(&err)); + config.assert_async().await; + poll.assert_async().await; + } + + #[tokio::test] + async fn wait_for_plan_propagates_failed() { + let mut server = Server::new_async().await; + let config = config_with_endpoints(&mut server, ALL_PLAN).await; + let poll = server + .mock("GET", "/v1/namespaces/ns/tables/tbl/plan/p1") + .with_status(200) + .with_body(r#"{"status":"failed","error":{"message":"boom","type":"IcebergException","code":500}}"#) + .expect(1) + .create_async() + .await; + let catalog = catalog(&server.url()); + let err = catalog + .wait_for_plan(&table(), "p1", WaitForPlanOptions { + min_delay: Duration::from_millis(1), + max_delay: Duration::from_millis(1), + ..Default::default() + }) + .await + .unwrap_err(); + assert!(is_plan_failed(&err)); + config.assert_async().await; + poll.assert_async().await; + } + + #[tokio::test] + async fn wait_for_plan_retries_java_idempotent_get_statuses() { + for status in [408, 429, 500, 502, 503, 504] { + let mut server = Server::new_async().await; + let config = config_with_endpoints(&mut server, ALL_PLAN).await; + let busy = server + .mock("GET", "/v1/namespaces/ns/tables/tbl/plan/p1") + .with_status(status) + .expect(1) + .create_async() + .await; + let done = server + .mock("GET", "/v1/namespaces/ns/tables/tbl/plan/p1") + .with_status(200) + .with_body(r#"{"status":"completed"}"#) + .expect(1) + .create_async() + .await; + let catalog = catalog(&server.url()); + let result = catalog + .wait_for_plan(&table(), "p1", WaitForPlanOptions { + min_delay: Duration::from_millis(1), + max_delay: Duration::from_millis(1), + ..Default::default() + }) + .await + .unwrap(); + assert_eq!(result.status, PlanStatus::Completed, "status {status}"); + config.assert_async().await; + busy.assert_async().await; + done.assert_async().await; + } + } + #[tokio::test] async fn wait_for_plan_retries_503_then_completes() { let mut server = Server::new_async().await; From f67e7ba30b8d3c0d75a16f32155067e78aa25694 Mon Sep 17 00:00:00 2001 From: David Dallakyan Date: Wed, 19 Aug 2026 03:00:50 +0400 Subject: [PATCH 4/4] fix(catalog-rest): keep wait_for_plan drop-cancel on RestCatalog RestSessionCatalog now owns the HTTP client cache. Clone an uninitialized inner catalog so drop-cancel still does not borrow the wait future. --- crates/catalog/rest/src/catalog.rs | 13 +++++++++++++ 1 file changed, 13 insertions(+) diff --git a/crates/catalog/rest/src/catalog.rs b/crates/catalog/rest/src/catalog.rs index 056a40a43c..cbdc85e198 100644 --- a/crates/catalog/rest/src/catalog.rs +++ b/crates/catalog/rest/src/catalog.rs @@ -577,6 +577,19 @@ impl RestCatalog { pub(crate) async fn supports_endpoint(&self, endpoint: &Endpoint) -> Result { self.inner.supports_endpoint(endpoint).await } + + /// Same catalog identity with an empty HTTP client cache, for fire-and-forget + /// work (best-effort plan cancel on drop) that must not borrow `self`. + pub(crate) fn clone_uninitialized(&self) -> Self { + Self { + session_context: self.session_context.clone(), + inner: Arc::new(self.inner.clone_uninitialized()), + } + } + + pub(crate) fn runtime(&self) -> &Runtime { + self.inner.runtime() + } } /// Every operation forwards to its [`RestSessionCatalog`] equivalent with the