feat(catalog-rest): wire TableScan to REST scan planning

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 }

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<String, String>) -> 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<Endpoint>,
 }
@@ -479,7 +479,7 @@ impl RestClient {
     }
 
     /// Sends `request`, authenticated by the client's session.
-    async fn query_catalog(&self, request: HttpRequest) -> Result<Response> {
+    pub(crate) async fn query_catalog(&self, request: HttpRequest) -> Result<Response> {
         self.http_client.query_catalog(request).await
     }
 
@@ -694,7 +694,7 @@ pub struct RestSessionCatalog {
     ///
     /// It could be different from the config fetched from the server and used at runtime.
     user_config: RestCatalogConfig,
-    client: OnceCell<RestClient>,
+    client: Arc<OnceCell<RestClient>>,
     /// Storage factory for creating FileIO instances.
     storage_factory: Option<Arc<dyn StorageFactory>>,
     runtime: Runtime,
@@ -704,7 +704,7 @@ pub struct RestSessionCatalog {
 
 impl RestSessionCatalog {
     /// Creates a `RestSessionCatalog` from a [`RestCatalogConfig`].
-    fn new(
+    pub(crate) fn new(
         config: RestCatalogConfig,
         auth_manager: Option<Arc<dyn AuthManager>>,
         storage_factory: Option<Arc<dyn StorageFactory>>,
@@ -714,13 +714,43 @@ impl RestSessionCatalog {
         Self {
             auth_manager,
             user_config: config,
-            client: OnceCell::new(),
+            client: Arc::new(OnceCell::new()),
             storage_factory,
             runtime,
             kms_client,
         }
     }
 
+    /// 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(),
+        )
+    }
+
+    /// Shares the already-initialized HTTP client. Used when injecting this
+    /// catalog as a [`iceberg::scan::ScanPlanner`] so Auto/capability checks
+    /// do not repeat `GET /v1/config`.
+    pub(crate) fn clone_initialized(&self) -> Self {
+        Self {
+            auth_manager: self.auth_manager.clone(),
+            user_config: self.user_config.clone(),
+            client: Arc::clone(&self.client),
+            storage_factory: self.storage_factory.clone(),
+            runtime: self.runtime.clone(),
+            kms_client: 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,
@@ -809,7 +839,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
@@ -1204,7 +1234,8 @@ impl SessionCatalog for RestSessionCatalog {
             .identifier(table_ident.clone())
             .file_io(file_io)
             .metadata(response.metadata)
-            .runtime(self.runtime.clone());
+            .runtime(self.runtime.clone())
+            .scan_planner(Arc::new(self.clone_initialized()));
         if let Some(kms_client) = self.kms_client.clone() {
             table_builder = table_builder.kms_client(kms_client);
         }
@@ -1269,7 +1300,8 @@ impl SessionCatalog for RestSessionCatalog {
             .identifier(table_ident.clone())
             .file_io(file_io)
             .metadata(response.metadata)
-            .runtime(self.runtime.clone());
+            .runtime(self.runtime.clone())
+            .scan_planner(Arc::new(self.clone_initialized()));
         if let Some(kms_client) = self.kms_client.clone() {
             table_builder = table_builder.kms_client(kms_client);
         }
@@ -1412,7 +1444,8 @@ impl SessionCatalog for RestSessionCatalog {
             .file_io(file_io)
             .metadata(response.metadata)
             .metadata_location(metadata_location.clone())
-            .runtime(self.runtime.clone());
+            .runtime(self.runtime.clone())
+            .scan_planner(Arc::new(self.clone_initialized()));
         if let Some(kms_client) = self.kms_client.clone() {
             table_builder = table_builder.kms_client(kms_client);
         }
@@ -1493,7 +1526,8 @@ impl SessionCatalog for RestSessionCatalog {
             .file_io(file_io)
             .metadata(response.metadata)
             .metadata_location(response.metadata_location)
-            .runtime(self.runtime.clone());
+            .runtime(self.runtime.clone())
+            .scan_planner(Arc::new(self.clone_initialized()));
         if let Some(kms_client) = self.kms_client.clone() {
             table_builder = table_builder.kms_client(kms_client);
         }

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::<Endpoint>().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));
+    }
 }

crates/catalog/rest/src/lib.rs

@@ -92,9 +92,12 @@ pub use client::HttpClient;
 mod request;
 pub use request::{HttpRequest, HttpRequestBody};
 mod endpoint;
+mod scan_decode;
+mod scan_planning;
 mod types;
 
 pub use auth::*;
 pub use catalog::*;
 pub use endpoint::Endpoint;
+pub use scan_planning::*;
 pub use types::*;