diff --git a/skills/hotdata/SKILL.md b/skills/hotdata/SKILL.md index a1e97ae7..8446b5ab 100644 --- a/skills/hotdata/SKILL.md +++ b/skills/hotdata/SKILL.md @@ -274,10 +274,17 @@ hotdata ingest status --wait # attach and poll to done/failed hotdata ingest list-imports # the SQL behind each import + its database hotdata ingest trigger-import # 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 FROM [.] [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 FROM [.
] [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). diff --git a/src/client/ingest.rs b/src/client/ingest.rs index a6025029..1a2f820c 100644 --- a/src/client/ingest.rs +++ b/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 { + 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 { self.send(self.authed(reqwest::Method::POST, "/jobs/drain"), None) } @@ -429,6 +442,18 @@ pub struct QueryToIngest { pub database_id: Option, } +/// `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, +} + /// 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`. diff --git a/src/commands/ingest.rs b/src/commands/ingest.rs index 66c70563..e289d3f4 100644 --- a/src/commands/ingest.rs +++ b/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 FROM [.
] [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, + + #[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, + 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`.