feat(databases): load a query result via --result

`databases load` gains a `--result <id>` flag, mutually exclusive with
--file/--url/--upload-id, that loads a saved query result into a table
by posting {mode, result_id} to the load endpoint — skipping the upload
step entirely. Works for both the standard and database-scoped load
paths, and reuses the existing spinner, auto-declare recovery, and
success output.

Author: eddietejeda

src/commands/databases.rs

@@ -107,7 +107,7 @@ pub enum DatabasesCommands {
         name_or_id: String,
     },
 
-    /// Load a parquet file into a managed database table
+    /// Load a parquet file or a saved query result into a managed database table
     Load {
         /// SQL catalog alias of the target database (e.g. `--catalog airbnb`)
         #[arg(long)]
@@ -122,16 +122,21 @@ pub enum DatabasesCommands {
         table: String,
 
         /// Path to a local parquet file to upload and load
-        #[arg(long, conflicts_with_all = ["upload_id", "url"])]
+        #[arg(long, conflicts_with_all = ["upload_id", "url", "result"])]
         file: Option<String>,
 
         /// URL of a remote parquet file to download and load
-        #[arg(long, conflicts_with_all = ["file", "upload_id"])]
+        #[arg(long, conflicts_with_all = ["file", "upload_id", "result"])]
         url: Option<String>,
 
         /// Use a previously staged upload ID from `POST /v1/uploads` instead of uploading
-        #[arg(long, conflicts_with_all = ["file", "url"])]
+        #[arg(long, conflicts_with_all = ["file", "url", "result"])]
         upload_id: Option<String>,
+
+        /// Load a saved query result by id (e.g. `--result rslt…`, from
+        /// `hotdata results` or a query's `[result-id: …]` footer) instead of a file
+        #[arg(long, conflicts_with_all = ["file", "url", "upload_id"])]
+        result: Option<String>,
     },
 
     /// Manage tables inside a managed database
@@ -190,7 +195,7 @@ pub enum DatabaseTablesCommands {
         output: String,
     },
 
-    /// Load a parquet file into a table (creates or replaces the table)
+    /// Load a parquet file or a saved query result into a table (creates or replaces the table)
     Load {
         /// Database id or name (defaults to current database)
         #[arg(long)]
@@ -204,16 +209,21 @@ pub enum DatabaseTablesCommands {
         schema: String,
 
         /// Path to a local parquet file to upload and load
-        #[arg(long, conflicts_with_all = ["upload_id", "url"])]
+        #[arg(long, conflicts_with_all = ["upload_id", "url", "result"])]
         file: Option<String>,
 
         /// URL of a remote parquet file to download and load
-        #[arg(long, conflicts_with_all = ["file", "upload_id"])]
+        #[arg(long, conflicts_with_all = ["file", "upload_id", "result"])]
         url: Option<String>,
 
         /// Use a previously staged upload ID from `POST /v1/uploads` instead of uploading
-        #[arg(long, conflicts_with_all = ["file", "url"])]
+        #[arg(long, conflicts_with_all = ["file", "url", "result"])]
         upload_id: Option<String>,
+
+        /// Load a saved query result by id (e.g. `--result rslt…`, from
+        /// `hotdata results` or a query's `[result-id: …]` footer) instead of a file
+        #[arg(long, conflicts_with_all = ["file", "url", "upload_id"])]
+        result: Option<String>,
     },
 
     /// Delete a table from a managed database
@@ -561,6 +571,13 @@ pub fn load_table_request(upload_id: &str) -> serde_json::Value {
     })
 }
 
+pub fn load_table_request_from_result(result_id: &str) -> serde_json::Value {
+    serde_json::json!({
+        "mode": "replace",
+        "result_id": result_id,
+    })
+}
+
 /// Returns true when `path` looks like a parquet file by extension.
 pub fn is_parquet_path(path: &str) -> bool {
     path.to_ascii_lowercase().ends_with(".parquet")
@@ -1332,6 +1349,7 @@ pub fn tables_list(workspace_id: &str, database: Option<&str>, schema: Option<&s
     }
 }
 
+#[allow(clippy::too_many_arguments)]
 pub fn tables_load(
     workspace_id: &str,
     database: Option<&str>,
@@ -1340,6 +1358,7 @@ pub fn tables_load(
     file: Option<&str>,
     url: Option<&str>,
     upload_id: Option<&str>,
+    result: Option<&str>,
 ) {
     use crossterm::style::Stylize;
 
@@ -1350,7 +1369,16 @@ pub fn tables_load(
         && crate::client::credentials::api_key_jwt_source(&profile).as_deref()
             == Some("database_api_token")
     {
-        tables_load_database_scoped(workspace_id, database, table, schema, file, url, upload_id);
+        tables_load_database_scoped(
+            workspace_id,
+            database,
+            table,
+            schema,
+            file,
+            url,
+            upload_id,
+            result,
+        );
         return;
     }
 
@@ -1378,20 +1406,24 @@ pub fn tables_load(
     let db = resolve_database(&api, &lookup_key);
     let schema = schema_name(schema);
 
-    // clap enforces mutual exclusion; only one of these is ever Some.
-    let upload_id = match (upload_id, file, url) {
-        (Some(id), None, None) => id.to_string(),
-        (None, Some(path), None) => upload_parquet_file(&api, path),
-        (None, None, Some(u)) => upload_parquet_url(&api, u),
-        (None, None, None) => {
-            eprintln!("error: --file <path>, --url <url>, or --upload-id <id> is required");
+    // clap enforces mutual exclusion; only one source is ever Some. A file/URL
+    // is uploaded first and loaded by upload id; a result is loaded by reference
+    // with no upload step.
+    let body = match (result, upload_id, file, url) {
+        (Some(result_id), None, None, None) => load_table_request_from_result(result_id),
+        (None, Some(id), None, None) => load_table_request(id),
+        (None, None, Some(path), None) => load_table_request(&upload_parquet_file(&api, path)),
+        (None, None, None, Some(u)) => load_table_request(&upload_parquet_url(&api, u)),
+        (None, None, None, None) => {
+            eprintln!(
+                "error: one of --file <path>, --url <url>, --upload-id <id>, or --result <result-id> is required"
+            );
             std::process::exit(1);
         }
         _ => unreachable!(),
     };
 
     let path = managed_table_load_path(&db.default_connection_id, schema, table);
-    let body = load_table_request(&upload_id);
 
     let spinner = crate::util::spinner("Loading table...");
     let (status, resp_body) = api.post_raw(&path, &body).unwrap_or_else(|e| {
@@ -1565,6 +1597,7 @@ pub fn tables_load(
 /// Load path for a database API token: addresses the database by id and uses
 /// the database-scoped endpoints (the connection-scoped managed paths and the
 /// name/catalog resolve used by the standard flow are denied for this token).
+#[allow(clippy::too_many_arguments)]
 fn tables_load_database_scoped(
     workspace_id: &str,
     database: Option<&str>,
@@ -1573,6 +1606,7 @@ fn tables_load_database_scoped(
     file: Option<&str>,
     url: Option<&str>,
     upload_id: Option<&str>,
+    result: Option<&str>,
 ) {
     use crossterm::style::Stylize;
 
@@ -1596,20 +1630,24 @@ fn tables_load_database_scoped(
             std::process::exit(1);
         });
 
-    // clap enforces mutual exclusion; only one of these is ever Some.
-    let upload_id = match (upload_id, file, url) {
-        (Some(id), None, None) => id.to_string(),
-        (None, Some(path), None) => upload_parquet_file(&api, path),
-        (None, None, Some(u)) => upload_parquet_url(&api, u),
-        (None, None, None) => {
-            eprintln!("error: --file <path>, --url <url>, or --upload-id <id> is required");
+    // clap enforces mutual exclusion; only one source is ever Some. A file/URL
+    // is uploaded first and loaded by upload id; a result is loaded by reference
+    // with no upload step.
+    let body = match (result, upload_id, file, url) {
+        (Some(result_id), None, None, None) => load_table_request_from_result(result_id),
+        (None, Some(id), None, None) => load_table_request(id),
+        (None, None, Some(path), None) => load_table_request(&upload_parquet_file(&api, path)),
+        (None, None, None, Some(u)) => load_table_request(&upload_parquet_url(&api, u)),
+        (None, None, None, None) => {
+            eprintln!(
+                "error: one of --file <path>, --url <url>, --upload-id <id>, or --result <result-id> is required"
+            );
             std::process::exit(1);
         }
         _ => unreachable!(),
     };
 
     let load_path = database_table_load_path(&db_id, schema, table);
-    let body = load_table_request(&upload_id);
 
     let load = || {
         let spinner = crate::util::spinner("Loading table...");
@@ -1960,6 +1998,15 @@ mod tests {
         assert_eq!(body["upload_id"], "upl_abc");
     }
 
+    #[test]
+    fn load_table_request_from_result_is_replace_mode() {
+        let body = load_table_request_from_result("rslt_abc");
+        assert_eq!(body["mode"], "replace");
+        assert_eq!(body["result_id"], "rslt_abc");
+        // A result load must not send an upload_id (the server rejects both).
+        assert!(body.get("upload_id").is_none());
+    }
+
     #[test]
     fn is_parquet_path_by_extension() {
         assert!(is_parquet_path("/data/orders.parquet"));
@@ -2097,6 +2144,50 @@ mod tests {
         load.assert();
     }
 
+    #[test]
+    fn tables_load_from_result_posts_result_id() {
+        let mut server = mockito::Server::new();
+        let resolve = server
+            .mock("GET", "/v1/databases/db_1")
+            .with_status(200)
+            .with_header("content-type", "application/json")
+            .with_body(full_detail("db_1", "sales", "conn_default"))
+            .create();
+        // A result load hits the same loads endpoint but with a result_id body
+        // and no upload step (no POST /v1/uploads mock is registered).
+        let load = server
+            .mock(
+                "POST",
+                "/v1/connections/conn_default/schemas/public/tables/orders/loads",
+            )
+            .match_body(mockito::Matcher::JsonString(
+                serde_json::to_string(&load_table_request_from_result("rslt_123")).unwrap(),
+            ))
+            .with_status(200)
+            .with_body(
+                r#"{
+                "connection_id":"conn_default",
+                "schema_name":"public",
+                "table_name":"orders",
+                "row_count":42,
+                "arrow_schema_json":"{}"
+            }"#,
+            )
+            .create();
+
+        let api = Api::test_new(&server.url(), "k", Some("ws1"));
+        let db = resolve_database(&api, "db_1");
+        let path = managed_table_load_path(&db.default_connection_id, "public", "orders");
+        let body = load_table_request_from_result("rslt_123");
+        let (status, resp_body) = api.post_raw(&path, &body).unwrap();
+        assert!(status.is_success());
+        let parsed: LoadManagedTableResponse = serde_json::from_str(&resp_body).unwrap();
+        assert_eq!(parsed.row_count, 42);
+        assert_eq!(parsed.table_name, "orders");
+        resolve.assert();
+        load.assert();
+    }
+
     #[test]
     fn tables_delete_uses_default_connection_id() {
         let mut server = mockito::Server::new();

src/main.rs

@@ -399,6 +399,7 @@ fn main() {
                             file,
                             url,
                             upload_id,
+                            result,
                         }) => databases::tables_load(
                             &workspace_id,
                             Some(catalog.as_str()),
@@ -407,6 +408,7 @@ fn main() {
                             file.as_deref(),
                             url.as_deref(),
                             upload_id.as_deref(),
+                            result.as_deref(),
                         ),
                         Some(DatabasesCommands::Tables { database, command }) => match command {
                             Some(DatabaseTablesCommands::List {
@@ -426,6 +428,7 @@ fn main() {
                                 file,
                                 url,
                                 upload_id,
+                                result,
                             }) => databases::tables_load(
                                 &workspace_id,
                                 db_flag.as_deref().or(database.as_deref()),
@@ -434,6 +437,7 @@ fn main() {
                                 file.as_deref(),
                                 url.as_deref(),
                                 upload_id.as_deref(),
+                                result.as_deref(),
                             ),
                             Some(DatabaseTablesCommands::Delete {
                                 database: db_flag,