feat(ingest): add hotdata ingest raw-sql verb

skills/hotdata/SKILL.md

@@ -274,10 +274,17 @@ hotdata ingest status <import-id> --wait      # attach and poll to done/failed
 
 hotdata ingest list-imports                   # the SQL behind each import + its database
 hotdata ingest trigger-import <import-id>     # re-run: refresh the same DB from source
+
+# raw-sql (SQL datasources only): run a source-native query VERBATIM in the
+# source dialect and ingest its result — joins, GROUP BY, window functions, CTEs,
+# engine-specific functions all run at the source. Single read-only statement.
+hotdata ingest raw-sql --source prod_pg --table rev_by_seg \
+  "SELECT c_mktsegment, SUM(o_totalprice) AS rev FROM customer c JOIN orders o ON o.o_custkey=c.c_custkey GROUP BY 1"
 ```
 
 Agent tips:
-- Import SQL is a **restricted grammar**: `SELECT <cols|*> FROM <datasource>[.<table>] [WHERE …] [LIMIT n]` — no joins/GROUP BY. Run real analytics with `hotdata query` against the imported database afterward.
+- `new-import` SQL is a **restricted grammar**: `SELECT <cols|*> FROM <datasource>[.<table>] [WHERE …] [LIMIT n]` — no joins/GROUP BY. For those, either use `raw-sql` (pushes the full query down to a SQL source) or run `hotdata query` against the imported database afterward.
+- `raw-sql` is SQL-datasources-only and read-only (a single SELECT; writes/DDL are refused). Prefer a **read-only source credential**. Cast ambiguous/mixed-precision decimal columns explicitly in the query.
 - Prefer `-o json` + the `status` exit codes for scripting; poll one-shot `status` rather than holding `--wait` when doing other work in between.
 - `status` is a **closed set**: `pending` | `running` | `done` | `failed`. While running, the finer progress state (e.g. `extracting`, `loading`) appears in `stage` — informational only, never switch on it.
 - Tables print oldest→newest; `-o json` is newest-first (`[0]` = latest).

src/client/ingest.rs

@@ -268,6 +268,19 @@ impl IngestClient {
         )
     }
 
+    /// Ingest the result of a source-native SQL query (SQL datasources only).
+    /// The SQL runs verbatim in the source dialect; the server enforces a single
+    /// read-only statement.
+    pub fn create_raw_query(&self, req: &RawSqlIngest) -> Result<IngestAck, IngestError> {
+        self.require_api_key()?;
+        let body = serde_json::to_value(req).expect("RawSqlIngest serializes");
+        self.send(
+            self.authed(reqwest::Method::POST, "/queries/raw")
+                .json(&body),
+            Some(&body),
+        )
+    }
+
     pub fn drain(&self) -> Result<serde_json::Value, IngestError> {
         self.send(self.authed(reqwest::Method::POST, "/jobs/drain"), None)
     }
@@ -429,6 +442,18 @@ pub struct QueryToIngest {
     pub database_id: Option<String>,
 }
 
+/// `POST /ingest/queries/raw` body: source-native SQL run verbatim against a SQL
+/// datasource, its result loaded into `table`. The workspace rides the
+/// X-Workspace-Id header, so it is not in the body.
+#[derive(Debug, Serialize)]
+pub struct RawSqlIngest {
+    pub sql: String,
+    pub source: String,
+    pub table: String,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub limit: Option<u64>,
+}
+
 /// 202 body from `POST /sources` and `POST /queries`. Common fields are typed;
 /// everything else (e.g. `resource`, `columns`, `where`, `limit` on queries) is
 /// captured for `--output json`.

src/commands/ingest.rs

@@ -37,7 +37,7 @@
 //! `name`, `type`, `status`, `stage`, …), not the wire rows.
 
 use crate::client::ingest::{
-    ConnectorEntry, IngestAck, IngestClient, IngestRequest, JobStatus, QueryToIngest,
+    ConnectorEntry, IngestAck, IngestClient, IngestRequest, JobStatus, QueryToIngest, RawSqlIngest,
 };
 use crate::commands::prompt::{
     ask_secret, ask_text, optional, optional_default, prompt_list, select_optional,
@@ -160,6 +160,34 @@ pub enum IngestCommands {
         poll: PollArgs,
     },
 
+    /// Ingest the result of a source-native SQL query (SQL datasources only)
+    ///
+    /// Unlike `new-import` (SELECT <cols|*> FROM <ds>[.<table>] [WHERE] [LIMIT]),
+    /// the SQL runs verbatim against the source engine in its OWN dialect —
+    /// joins, aggregation, GROUP BY, CTEs, window functions, and engine-specific
+    /// functions all execute at the source; only the result set transfers. It
+    /// must be a single read-only statement; the result lands in `--table` in a
+    /// fresh managed database.
+    RawSql {
+        /// A single read-only SQL statement, in the SOURCE engine's dialect
+        sql: String,
+
+        /// SQL datasource to run against: a name or a datasource id
+        #[arg(long)]
+        source: String,
+
+        /// Result table name in the new managed database
+        #[arg(long)]
+        table: String,
+
+        /// Cap the result at N rows
+        #[arg(long)]
+        limit: Option<u64>,
+
+        #[command(flatten)]
+        poll: PollArgs,
+    },
+
     /// List your imports: the SQL behind each and the database it landed in
     ListImports,
 
@@ -252,6 +280,13 @@ pub fn dispatch(workspace_id: &str, output: &str, command: IngestCommands) {
             database_id,
             poll,
         } => new_import(workspace_id, output, sql, all, source, database_id, &poll),
+        IngestCommands::RawSql {
+            sql,
+            source,
+            table,
+            limit,
+            poll,
+        } => raw_sql(workspace_id, output, sql, source, table, limit, &poll),
         IngestCommands::ListImports => list_imports(workspace_id, output),
         IngestCommands::TriggerImport { id, poll } => {
             trigger_import(workspace_id, output, &id, &poll)
@@ -940,6 +975,48 @@ fn looks_like_ingest_id(s: &str) -> bool {
     s.len() == 32 && s.chars().all(|c| c.is_ascii_hexdigit())
 }
 
+// --- raw-sql: source-native SQL front-door --------------------------------
+
+fn raw_sql(
+    workspace_id: &str,
+    output: &str,
+    sql: String,
+    source: String,
+    table: String,
+    limit: Option<u64>,
+    poll: &PollArgs,
+) {
+    let client = IngestClient::new(workspace_id);
+    let req = RawSqlIngest {
+        sql,
+        source,
+        table,
+        limit,
+    };
+    let spinner = util::spinner("submitting raw-sql import…");
+    let ack = client.create_raw_query(&req).unwrap_or_else(|e| {
+        spinner.finish_and_clear();
+        e.exit()
+    });
+    spinner.finish_and_clear();
+    if poll.wait {
+        let st = poll_ingest(
+            &client,
+            output,
+            &ack.ingest_id,
+            poll.wait_timeout,
+            "importing",
+            true,
+        );
+        render_done(&st, output);
+        return;
+    }
+    // Non-blocking default: kick the drain so the enqueued job actually runs,
+    // then hand the terminal back (same as new-import).
+    let _ = client.drain();
+    render_ack(&ack, "import id:", output);
+}
+
 /// The SQL a `new-import` runs: explicit SQL wins; `--all` becomes a full
 /// `SELECT *` against the datasource's name (`pinned_name` when --source was
 /// a datasource id). Errors are messages for `fail`.