daviddallakyan2005 commented on code in PR #3011: URL: https://github.com/apache/iceberg-rust/pull/3011#discussion_r3792525499
########## crates/catalog/rest/src/scan_planning.rs: ########## @@ -0,0 +1,1752 @@ +// 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<String>, + /// 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<RestFileScanTask>, + /// Delete files referenced by the scan tasks, as raw REST JSON. + #[serde(default, skip_serializing_if = "Vec::is_empty")] + pub delete_files: Vec<serde_json::Value>, +} + +/// 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<Vec<i32>>, + /// Optional residual filter in ExpressionParser JSON. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub residual_filter: Option<serde_json::Value>, +} + +/// 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<String>, + /// `X-Iceberg-Access-Delegation` header. `None` sends no such header. + #[serde(skip)] + pub access_delegation: Option<String>, + /// Snapshot to scan. Omitted for the current snapshot. + #[serde(skip_serializing_if = "Option::is_none")] + pub snapshot_id: Option<i64>, + /// Selected schema fields. + #[serde(default, skip_serializing_if = "Vec::is_empty")] + pub select: Vec<String>, + /// Row filter as ExpressionParser JSON, not `iceberg::expr::Predicate`. + #[serde(skip_serializing_if = "Option::is_none")] + pub filter: Option<serde_json::Value>, + /// Hint for the minimum number of rows the server should return. + #[serde(skip_serializing_if = "Option::is_none")] + pub min_rows_requested: Option<i64>, + /// Case-sensitive field matching for filter and select. + #[serde(skip_serializing_if = "Option::is_none")] + pub case_sensitive: Option<bool>, + /// When true, use the schema at the scanned snapshot. + #[serde(skip_serializing_if = "Option::is_none")] + pub use_snapshot_schema: Option<bool>, + /// Incremental scan start (exclusive). Wire-only in this PR. + #[serde(skip_serializing_if = "Option::is_none")] + pub start_snapshot_id: Option<i64>, + /// Incremental scan end (inclusive). Wire-only in this PR. + #[serde(skip_serializing_if = "Option::is_none")] + pub end_snapshot_id: Option<i64>, + /// Fields for which the server should send column stats. + #[serde(default, skip_serializing_if = "Vec::is_empty")] + pub stats_fields: Vec<String>, +} + +/// 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<String>, + /// Failed-arm error detail. + #[serde(skip_serializing_if = "Option::is_none")] + pub error: Option<ErrorModel>, + /// 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<Vec<StorageCredential>>, +} + +/// 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<ErrorModel>, + /// 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<Vec<StorageCredential>>, +} + +/// 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<Vec<StorageCredential>>, +} + +/// 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<String>, +} + +/// 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<String>, + /// 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 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<Duration>, + /// `X-Iceberg-Access-Delegation` forwarded on each poll. + pub access_delegation: Option<String>, +} + +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<D: Deserializer<'de>>(deserializer: D) -> std::result::Result<Self, D::Error> { + 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<D: Deserializer<'de>>(deserializer: D) -> std::result::Result<Self, D::Error> { + 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<String>, + #[serde(default)] + error: Option<serde_json::Value>, + #[serde(flatten)] + scan_tasks: ScanTasks, + #[serde(default)] + storage_credentials: Option<Vec<StorageCredential>>, +} + +fn decode_planning_error(raw: Option<serde_json::Value>) -> Option<ErrorModel> { + 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 { Review Comment: Agreed on both points. Matching err.message() is a bad public contract, and these helpers are test-only today. I'll unexport all five (expired / failed / cancelled / no-such-plan-task / poll-exhausted) and drop them from public-api.txt. Tests can keep crate-private checks. I would not add a planning-error enum in this PR, and I would not extend iceberg::ErrorKind (that's the core crate). Go has sentinels like ErrPlanExpired; we can add crate-local matching later if a caller (TableScan auto fallback) needs to branch on expired vs failed. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
