diff --git a/README.md b/README.md index b2770cf..e16c4db 100644 --- a/README.md +++ b/README.md @@ -42,17 +42,20 @@ PostgreSQL-dialect SQL. Everything else builds on that. ## Getting your data in -**Upload a parquet file** directly (convert CSV/JSON first): +**Upload a file** directly — csv, newline-delimited json, or parquet: ```sh -hotdata databases load --catalog demo --table listings --file ./listings.parquet +hotdata databases load --catalog demo --table listings --file ./listings.csv ``` -A load **replaces** the table by default. Add `--append` to add rows to an +The format comes from the extension; pass `--format csv|json|parquet` when the +extension is missing or misleading. + +A load **replaces** the table by default. `--mode append` adds rows to an existing table instead: ```sh -hotdata databases load --catalog demo --table listings --file ./more-listings.parquet --append +hotdata databases load --catalog demo --table listings --file ./more-listings.parquet --mode append ``` **Import from an external source** — Postgres/MySQL, S3/GCS buckets, Iceberg, @@ -163,10 +166,11 @@ The full command surface. The top level has nine groups — `auth`, `workspaces` | `databases use` | Set the current (default) database | | `databases unset` | Clear the current database | | `databases remove` | Delete a database and all its tables | -| `databases load` | Load a parquet file or saved result into a table (replace, or `--append`) | +| `databases load` | Load a csv/json/parquet file or saved result into a table (`--mode replace\|append\|delete\|update\|upsert`) | +| `databases tables add` | Declare a table with its key and storage layout | | `databases tables list` | List tables in a database | | `databases tables show` | Show column definitions for a table | -| `databases tables load` | Load parquet/result into a table (replace, or `--append`) | +| `databases tables load` | Same as `databases load`, addressed by database instead of catalog | | `databases tables remove` | Delete a table from a database | | `databases context list` | List named contexts in a database | | `databases context show` | Print context content to stdout | diff --git a/skills/hotdata/SKILL.md b/skills/hotdata/SKILL.md index 73bf443..3682eac 100644 --- a/skills/hotdata/SKILL.md +++ b/skills/hotdata/SKILL.md @@ -87,7 +87,7 @@ Returns workspaces with `public_id`, `name`, `active`, `favorite`, `provision_st **Instant databases** are Hotdata-owned catalogs you create and populate yourself — no remote source to sync. Query them in SQL as **`..`**. Prefer **`hotdata databases`** for this workflow. -**Parquet only:** `databases tables load` accepts **parquet** files (local `--file`, remote `--url`, or a pre-staged `--upload-id`). +**File formats:** `databases tables load` accepts **csv**, **newline-delimited json**, and **parquet** (local `--file`, remote `--url`, or a pre-staged `--upload-id`). The format is read from the file's extension; `--format csv|json|parquet` overrides it, and an unrecognised extension is left to the server to resolve from the bytes. **Active database:** `hotdata databases use ` saves the active database to config. `databases tables list`/`load`/`remove`, `databases queries`/`results`, and all `databases context` commands default to the active database; pass **`--database `** to override per-command. (`databases tables show` instead takes a fully-qualified `catalog.schema.table`.) @@ -108,12 +108,14 @@ hotdata databases remove [--workspace-id ] hotdata databases attach [--database ] [--alias ] hotdata databases detach [--database ] -# Preferred: load by catalog alias (auto-declares table if needed). --append adds rows instead of replacing. -hotdata databases load --catalog --table
[--schema public] (--file | --url | --upload-id | --result-id ) [--append] [--workspace-id ] +# Preferred: load by catalog alias (server declares the table/schema if missing). +# Loads csv, newline-delimited json, or parquet — format read from the extension. +hotdata databases load --catalog --table
[--schema public] (--file | --url | --upload-id | --result-id ) [--mode replace|append|delete|update|upsert] [--append] [--format csv|json|parquet] [--key ]... [--workspace-id ] # Also available via tables subcommand hotdata databases tables list [--database ] [--schema ] [--workspace-id ] [--output table|json|yaml] -hotdata databases tables load
[--database ] [--schema public] (--file | --url | --upload-id | --result-id ) [--append] [--workspace-id ] +hotdata databases tables add [--database ] [--schema public] [--key ]... [--key-determines ]... [--sorted-by [=asc|desc]]... [--partition-by [=identity|year|month|day|hour]]... [--output table|json|yaml] +hotdata databases tables load
[--database ] [--schema public] (--file | --url | --upload-id | --result-id ) [--mode replace|append|delete|update|upsert] [--append] [--format csv|json|parquet] [--key ]... [--workspace-id ] hotdata databases tables remove
[--database ] [--schema public] [--workspace-id ] ``` @@ -126,9 +128,11 @@ hotdata databases tables remove
[--database ] [--schema public] [--w - `unset` — clears the active database from config. - `` — inspect one database (returns id, catalog, name, expires_at; a fork also shows its `forked_from` record). - `remove` — removes the instant database; clears the active-database config if it matched. -- `load` (top-level shorthand) — loads parquet into `--catalog.--schema.--table`. Accepts `--file`, `--url`, `--upload-id`, or `--result-id` (load a saved query result by id — from `hotdata databases results` or a query's `[result-id: …]` footer — instead of a file; the result must belong to the target database). Replaces the table by default; pass `--append` to add rows to the existing table instead. If the table was not declared at create time, the CLI automatically deletes and recreates the database with the table declared, then retries the load. +- `load` (top-level shorthand) — loads a file into `--catalog.--schema.--table`. Accepts `--file`, `--url`, `--upload-id`, or `--result-id` (load a saved query result by id — from `hotdata databases results` or a query's `[result-id: …]` footer — instead of a file; the result must belong to the target database). **Formats:** csv, newline-delimited json (`.json`/`.jsonl`/`.ndjson`), and parquet; the format comes from the file's extension, and `--format` overrides it (needed when the extension is absent or misleading). An unrecognised extension is not rejected — the server reads the bytes. A table or schema that was never declared is declared by the server as part of the load, so no up-front `--table` is required. +- **Load modes** (`--mode`, default `replace`) — `replace` supersedes the table's contents; `append` adds rows; `delete`, `update`, and `upsert` match existing rows **by key**. `--append` is the old shorthand for `--mode append` and still works, but the two cannot be combined. The keyed modes need a key: declare one with `databases tables add --key`, or name it per-load with `--key` (repeat for a composite key). `delete` uploads only the key columns; `update` replaces matching rows and ignores unmatched ones; `upsert` inserts the unmatched instead. Keyed modes are not available with `--result-id`. - `tables list` — lists tables with `TABLE` (`..
`), `SYNCED`, `LAST_SYNC`. Uses active database when `--database` is omitted. -- `tables load` — publishes to an instant-database table from a local parquet file (`--file`), a remote parquet URL (`--url`), a pre-staged upload (`--upload-id`), or a saved query result (`--result-id`, must belong to the target database). Defaults to **replace** mode; pass `--append` to add rows to the existing table instead. +- `tables add` — declares a table **with its key and storage layout**, which a load cannot infer. `--key` (repeatable) is what enables the `delete`/`update`/`upsert` load modes on that table. `--sorted-by ` or `=desc` sets sort order; `--partition-by ` partitions on the value, `=month` (or `year`/`day`/`hour`) on a calendar part — one partition per calendar month needs **both** `=year` and `=month`, or every March shares a partition. Sort and partition are fixed once the table exists. `--key-determines` (repeatable) asserts a column's value is fixed by the key: it prunes keyed loads harder, and is **correctness-affecting** — declare it only where the invariant really holds, or a keyed load can leave a duplicate key behind. Re-adding an existing table is a conflict (409), and `tables remove` does not clear the declaration — the table leaves the listing but the name stays declared and still conflicts. So **a key cannot be retrofitted onto a table declared without one**: declare it with `--key` up front, or use a new table name. +- `tables load` — publishes to an instant-database table from a local file (`--file`), a remote URL (`--url`), a pre-staged upload (`--upload-id`), or a saved query result (`--result-id`, must belong to the target database). Same `--mode`, `--format`, and `--key` flags as the top-level `load` above. - `tables remove` — drops a table from the instant database. - `attach` — attaches a **catalog** to an instant database, so the catalog's **live** tables become visible inside that database's query scope. Defaults to the active database; target another with `--database`. `--alias` sets the SQL name the catalog answers to (defaults to the catalog's name). This is how you query an attached catalog's tables and **join across catalogs** — see [Querying across catalogs](#querying-across-catalogs-attach). - `detach` — removes an attached catalog. Accepts the catalog name/id **or** the alias you attached it under. Defaults to the active database. @@ -142,6 +146,20 @@ hotdata databases load --catalog airbnb --table listings --url https://example.c hotdata query "SELECT count(*) FROM airbnb.public.listings" ``` +Keeping a table in sync by key. Declare the key **before the table's first +load** — `tables add` on a table that already exists returns 409, and a key +cannot be added afterwards: + +``` +hotdata databases tables add bookings --key booking_id --sorted-by updated_at=desc +hotdata databases load --catalog airbnb --table bookings --file bookings.csv +hotdata databases load --catalog airbnb --table bookings --file changed.csv --mode upsert +hotdata databases load --catalog airbnb --table bookings --file removed.csv --mode delete +``` + +`removed.csv` carries only the key columns. On a table already declared without +a key, name one per load instead: `--mode upsert --key booking_id`. + #### Querying across catalogs (attach) **A `hotdata query` runs inside exactly one instant database** — the active database (`hotdata databases use `) or the one named by `--database`. With none set, the query fails with *"a database is required."* That database's query scope sees **only its own catalog plus any catalogs explicitly attached to it** — a workspace catalog is **not** visible just because it exists. Referencing an unattached catalog fails with *"table '\.\.\' not found."* diff --git a/skills/hotdata/references/WORKFLOWS.md b/skills/hotdata/references/WORKFLOWS.md index a395652..9c5d417 100644 --- a/skills/hotdata/references/WORKFLOWS.md +++ b/skills/hotdata/references/WORKFLOWS.md @@ -11,7 +11,7 @@ The `hotdata` skill is always loaded first (auth and workspace setup). The three | User goal | Skill | Key commands | |-----------|--------|----------------| | Login, workspaces, datasources, tables, context | **`hotdata`** | `auth`, `workspaces`, `ingest sources`, `ingest`, `databases tables`, `databases context` | -| Load parquet files into an instant database | **`hotdata`** | `databases create` + `databases load` | +| Load csv/json/parquet files into an instant database | **`hotdata`** | `databases create` + `databases tables add` + `databases load` | | SQL analytics, aggregations, history, Chain | **`hotdata-analytics`** (`subskills/analytics/SKILL.md`) | `query`, `databases queries`, `databases results` | | BM25 / vector search, retrieval indexes | **`hotdata-search`** (`subskills/search/SKILL.md`) | `search`, `search create`, `search embeddings` | | Geospatial / PostGIS-style SQL | **`hotdata-geospatial`** (`subskills/geospatial/SKILL.md`) | `query` with `ST_*`, WKB columns | @@ -94,16 +94,16 @@ A `hotdata query` runs inside **one** instant database; its scope sees that data | | **Instant databases** | |---|------------------------| -| **Best for** | Parquet files you own; catalog-style `alias.schema.table` | +| **Best for** | Files you own (csv, newline-delimited json, parquet); catalog-style `alias.schema.table` | | **SQL prefix** | `..
` where catalog = `--catalog` alias | | **CLI** | `hotdata databases create --catalog` + `databases load` | -| **Declare schema up front** | Yes — `--table` on create (auto-declared on first `databases load`) | -| **Parquet file uploads** | `databases load --file` / `--url` / `--upload-id` | -| **Refresh** | Replace via `databases load` again | +| **Declare schema up front** | Optional — the load declares a missing table/schema. Declare with `databases tables add --key` when you need the keyed load modes; a key cannot be added later | +| **File uploads** | `databases load --file` / `--url` / `--upload-id`; format from the extension, or `--format` | +| **Refresh** | `databases load` again — `--mode replace` (default) or `append`, or `delete`/`update`/`upsert` against a declared key | -**Rule of thumb:** Parquet files you control as **`mydb.public.orders`** → **instant databases**. +**Rule of thumb:** Files you control as **`mydb.public.orders`** → **instant databases**. -### Workflow: instant database (parquet) +### Workflow: instant database (file upload) 1. Create the database with a catalog alias: @@ -111,16 +111,39 @@ A `hotdata query` runs inside **one** instant database; its scope sees that data hotdata databases create --catalog sales ``` -2. Load parquet per table (tables are auto-declared if needed): +2. **Before the first load**, declare any table that needs a key, a sort order, + or partitioning — the load cannot infer these, and a key cannot be added to + a table that already exists: ```bash - hotdata databases load --catalog sales --table orders --file ./orders.parquet + hotdata databases tables add orders --key order_id --sorted-by created_at=desc + ``` + + Skip this for a table you will only ever replace or append to. + +3. Load a file per table. A table or schema not declared above is declared as + part of the load, and the format comes from the file's extension: + + ```bash + hotdata databases load --catalog sales --table orders --file ./orders.csv hotdata databases load --catalog sales --table customers --url https://example.com/customers.parquet ``` - > Auto-declaring a *new* table recreates the database (no add-table API), which **changes its `id`** — the id returned by `databases create` goes stale after the next `load` of an undeclared table. Declare tables up front (`databases create --table orders --table customers`) to avoid the recreate, and don't cache ids across loads: re-read the current id from `databases list` at time of use. (Selection is still always by id — names and catalogs are not unique.) + > The database keeps its `id` and its other tables across a load into an + > undeclared table — nothing is recreated. Selection is still always by id, + > since names and catalogs are not unique. + +4. Keep a keyed table in sync by loading only what changed: + + ```bash + hotdata databases load --catalog sales --table orders --file ./changed.csv --mode upsert + hotdata databases load --catalog sales --table orders --file ./removed.csv --mode delete + ``` + + `removed.csv` carries only the key columns. This needs the key declared in + step 2, or named per load with `--key order_id`. -3. Confirm and query: +5. Confirm and query: ```bash hotdata databases tables list diff --git a/skills/hotdata/subskills/search/SKILL.md b/skills/hotdata/subskills/search/SKILL.md index 3f05268..6f68a24 100644 --- a/skills/hotdata/subskills/search/SKILL.md +++ b/skills/hotdata/subskills/search/SKILL.md @@ -37,6 +37,7 @@ hotdata search "" --in | **`vector`** | Pass plain-text query; name the **source text column** (e.g. `title`). Server embeds using the same provider/metric/dimensions as the index. SQL uses `vector_distance(col, 'text')`. Results sort by distance (ascending). | - **Index name:** the index carries its own type, column, and provider — you only name the index. Use `search list` to see available index names. +- **`--select`:** defaults to the table's own columns. An auto-embed vector index materialises a `{column}_embedding` column on the table, and a search leaves it out — it is a 1536-float list in every row, tens of kilobytes per search nobody asked for. Ask for it back with `--select '*'`, or name it (`--select 'id,body_embedding'`); pair either with `--output json`, since `--output table` abbreviates long lists for display. - **Custom embedding model, raw query vector, or no vector index?** Use `hotdata query` directly (e.g. `cosine_distance(col, [])`) — `search` only auto-embeds the query text via the index's own provider. - **Before search:** create the right index (`search create --type text` or `--type vector`). See [references/INDEXES.md](references/INDEXES.md). - Default `--limit` is 10. @@ -69,7 +70,7 @@ hotdata search remove [-d ] - **`--type` is required** on create: `text` (BM25; one or more text columns, comma-separated in `--column`) or `vector` (exactly one column; often embeddings or auto-embedded text). (`sorted` is also a valid `--type`, covered in **`hotdata-analytics`** — [`../analytics/SKILL.md`](../analytics/SKILL.md).) - **`sorted`** indexes (range/equality for OLAP filters) are documented in **`hotdata-analytics`** ([`../analytics/SKILL.md`](../analytics/SKILL.md)) — this skill focuses on retrieval types. - **`--async`:** poll with `hotdata jobs ` (see **`hotdata`** skill **Jobs**). -- **Auto-embedding:** `--type vector` on a **text** column generates embeddings server-side. Optional `--provider`; default output column `{column}_embedding` (override with `--output-column`). +- **Auto-embedding:** `--type vector` on a **text** column generates embeddings server-side. Optional `--provider`; default output column `{column}_embedding` (override with `--output-column`). The generated column becomes part of the table, so it shows up in `information_schema.columns` and in a hand-written `SELECT *` — `hotdata search` leaves it out of its own projection (see `--select` above). Full workflow (gather workload → compare existing → create → verify): [references/INDEXES.md](references/INDEXES.md). diff --git a/src/cli.rs b/src/cli.rs index fbc5bdc..7d3b187 100644 --- a/src/cli.rs +++ b/src/cli.rs @@ -119,7 +119,9 @@ pub enum Commands { #[arg(long, short = 'd')] database: Option, - /// Columns to display (comma-separated, defaults to all) + /// Columns to display (comma-separated). Defaults to the table's own + /// columns; a vector index's generated embedding column is left out. + /// Pass `--select '*'` to include it. #[arg(long)] select: Option, diff --git a/src/client/sdk.rs b/src/client/sdk.rs index 0636b6d..d6bb6e8 100644 --- a/src/client/sdk.rs +++ b/src/client/sdk.rs @@ -97,11 +97,54 @@ fn upload_reqwest_client() -> reqwest::Client { .expect("reqwest client should build without a timeout") } -/// Content type recorded for a managed-table parquet upload. Advisory only — -/// the managed-table load keys off the parquet file extension, not the upload's -/// recorded content type — but a correct MIME type is the right metadata to -/// persist alongside the file. const PARQUET_CONTENT_TYPE: &str = "application/vnd.apache.parquet"; +const CSV_CONTENT_TYPE: &str = "text/csv"; +/// Newline-delimited JSON — one object per line, which is what the load reads +/// for `json`. Not `application/json`, which implies a single document. +const JSON_CONTENT_TYPE: &str = "application/x-ndjson"; +/// Announced when the extension names no format we recognise, which is the +/// case the load resolves by reading the bytes. +const UNKNOWN_CONTENT_TYPE: &str = "application/octet-stream"; + +/// The MIME type to record for `name`, keyed off its extension. +/// +/// The recorded content type is **not** advisory to the managed-table load: the +/// load resolves the file's format from it first, and takes a confidently wrong +/// one at its word rather than sniffing past it — CSV bytes announced as +/// parquet fail with "failed to parse parquet metadata: Corrupt footer". So an +/// unrecognised extension announces [`UNKNOWN_CONTENT_TYPE`] rather than +/// guessing, leaving the server to read the bytes. +/// +/// `name` is the *user-supplied* file name or URL, never a temp path: `--url` +/// stages its download under a generated name, and typing that would announce +/// the wrong format for the bytes inside. +pub fn content_type_for_path(name: &str) -> &'static str { + match extension_of(name).to_ascii_lowercase().as_str() { + "parquet" => PARQUET_CONTENT_TYPE, + "csv" => CSV_CONTENT_TYPE, + "json" | "jsonl" | "ndjson" => JSON_CONTENT_TYPE, + _ => UNKNOWN_CONTENT_TYPE, + } +} + +/// The extension of a file name or URL, as written, or `""` when it has none. +/// Callers lowercase it before matching — extensions arrive in any case. +/// +/// A URL's query and fragment are dropped first: a presigned storage URL ends +/// in `…/listings.parquet?X-Amz-Signature=…`, and reading the extension off the +/// raw string would yield `parquet?x-amz-signature=…` and match nothing. Any +/// `/` after the last `.` also means the last path segment has no extension of +/// its own (`…/data.d/export`). +pub fn extension_of(name: &str) -> &str { + let path = name + .split_once(['?', '#']) + .map(|(before, _)| before) + .unwrap_or(name); + match path.rsplit_once('.') { + Some((_, ext)) if !ext.contains('/') => ext, + _ => "", + } +} /// Default number of multipart part `PUT`s the SDK keeps in flight for an /// upload. 12 saturates a typical uplink without overwhelming the socket pool @@ -681,7 +724,7 @@ impl Api { &self.client } - /// Upload a local parquet file directly to object storage via the SDK's + /// Upload a local data file directly to object storage via the SDK's /// presigned-upload flow ([`Client::upload_file`]), returning the upload id. /// /// The flow is `POST /v1/uploads` (open a session) → direct `PUT`(s) to @@ -700,14 +743,24 @@ impl Api { /// /// `progress` is the SDK [`UploadProgress`] callback, invoked with /// cumulative `(bytes_done, total)` as bytes flow; the caller drives a - /// progress bar from it. The recorded content type is parquet (advisory). - pub fn upload(&self, path: &Path, progress: UploadProgress) -> Result { + /// progress bar from it. + /// + /// `content_type` is recorded with the upload and is what the managed-table + /// load resolves the file's format from, so it must describe the bytes — + /// see [`content_type_for_path`], which callers use to derive it from the + /// name the user gave (not from `path`, which may be a staged temp file). + pub fn upload( + &self, + path: &Path, + content_type: &str, + progress: UploadProgress, + ) -> Result { let mut cfg = self.client.configuration().clone(); cfg.client = upload_reqwest_client(); let upload_client = Client::from_configuration(cfg); let opts = UploadOptions { - content_type: Some(PARQUET_CONTENT_TYPE.to_string()), + content_type: Some(content_type.to_string()), progress: Some(progress), max_concurrency: Some(upload_concurrency()), ..UploadOptions::default() @@ -799,10 +852,10 @@ impl Api { /// `Configuration`, returning the raw status + body text. /// /// The seam's DELETE counterpart to [`post_raw`](Self::post_raw): used by - /// `databases.rs`, where the delete bodies feed the same CLI-side - /// `(status, body)` control flow as the old raw `delete_raw` (e.g. the - /// delete+recreate path inspects the failure body), so non-success is - /// returned as `Ok((status, body))` rather than an error. + /// `databases.rs`, where a delete's response body feeds the same CLI-side + /// `(status, body)` control flow the raw calls used, so non-success is + /// returned as `Ok((status, body))` rather than an error and the caller + /// renders the server's own message. pub fn delete_raw(&self, path: &str) -> Result<(reqwest::StatusCode, String), ApiError> { let cfg = self.client.configuration(); let url = format!("{}/v1{path}", cfg.base_path); @@ -1517,6 +1570,37 @@ mod tests { m.assert(); } + #[test] + fn content_type_follows_the_extension_in_any_case() { + assert_eq!(content_type_for_path("a.parquet"), PARQUET_CONTENT_TYPE); + assert_eq!(content_type_for_path("A.PARQUET"), PARQUET_CONTENT_TYPE); + assert_eq!(content_type_for_path("a.csv"), CSV_CONTENT_TYPE); + assert_eq!(content_type_for_path("a.CSV"), CSV_CONTENT_TYPE); + for name in ["a.json", "a.jsonl", "a.ndjson"] { + assert_eq!(content_type_for_path(name), JSON_CONTENT_TYPE, "{name}"); + } + } + + #[test] + fn an_unknown_extension_announces_nothing_specific() { + // The load sniffs an octet-stream, so this is the permissive answer — + // announcing parquet here is what made a csv fail with "Corrupt footer". + for name in ["a.txt", "a", "archive.tar.zst", ""] { + assert_eq!(content_type_for_path(name), UNKNOWN_CONTENT_TYPE, "{name}"); + } + } + + #[test] + fn extension_of_reads_past_a_query_and_fragment() { + assert_eq!(extension_of("https://h/b/f.parquet?sig=a.b.c"), "parquet"); + assert_eq!(extension_of("https://h/f.csv#frag"), "csv"); + assert_eq!(extension_of("/local/f.csv"), "csv"); + // No extension on the last segment, despite dots earlier in the path. + assert_eq!(extension_of("https://h/v1.2/export"), ""); + assert_eq!(extension_of("/data.d/export"), ""); + assert_eq!(extension_of("plain"), ""); + } + // --- presigned direct-to-storage upload --------------------------------- /// A deterministic ASCII payload of `len` bytes written to a temp parquet @@ -1578,7 +1662,7 @@ mod tests { let api = Api::test_new(&server.url(), "test-jwt", Some("ws-1")); let id = api - .upload(tf.path(), noop_progress()) + .upload(tf.path(), "application/vnd.apache.parquet", noop_progress()) .expect("presigned upload should succeed"); assert_eq!(id, "upload_test"); @@ -1604,7 +1688,7 @@ mod tests { let api = Api::test_new(&server.url(), "test-jwt", Some("ws-1")); let err = api - .upload(tf.path(), noop_progress()) + .upload(tf.path(), "application/vnd.apache.parquet", noop_progress()) .expect_err("a 501 must map to an error, not a fallback"); match err { @@ -1642,7 +1726,7 @@ mod tests { let api = Api::test_new(&server.url(), "test-jwt", Some("ws-1")); let err = api - .upload(tf.path(), noop_progress()) + .upload(tf.path(), "application/vnd.apache.parquet", noop_progress()) .expect_err("a storage 403 must map to an error"); match err { @@ -1807,7 +1891,7 @@ mod tests { write_session(SESSION_JWT, 5); let id = api - .upload(tf.path(), noop_progress()) + .upload(tf.path(), "application/vnd.apache.parquet", noop_progress()) .expect("upload must succeed when every leg resolves its own bearer"); assert_eq!(id, "upload_fresh"); diff --git a/src/commands/databases.rs b/src/commands/databases.rs index 4daa1c0..2480f8b 100644 --- a/src/commands/databases.rs +++ b/src/commands/databases.rs @@ -172,7 +172,8 @@ pub enum DatabasesCommands { name_or_id: String, }, - /// Load a parquet file or a saved query result into an instant database table + /// Load a csv, json, or parquet file — or a saved query result — into an + /// instant database table Load { /// SQL catalog alias of the target database (e.g. `--catalog airbnb`) #[arg(long)] @@ -186,11 +187,11 @@ pub enum DatabasesCommands { #[arg(long)] table: String, - /// Path to a local parquet file to upload and load + /// Path to a local file to upload and load (csv, json, or parquet) #[arg(long, conflicts_with_all = ["upload_id", "url", "result_id"])] file: Option, - /// URL of a remote parquet file to download and load + /// URL of a remote file to download and load (csv, json, or parquet) #[arg(long, conflicts_with_all = ["file", "upload_id", "result_id"])] url: Option, @@ -205,9 +206,30 @@ pub enum DatabasesCommands { #[arg(long, conflicts_with_all = ["file", "url", "upload_id"])] result_id: Option, - /// Append rows to the table instead of replacing it (default: replace) - #[arg(long)] + /// Append rows to the table instead of replacing it (default: replace). + /// Shorthand for `--mode append`. + #[arg(long, conflicts_with = "mode")] append: bool, + + /// How the upload is applied to the table (default: replace). + /// `delete`, `update`, and `upsert` match existing rows by the table's + /// key — declare one with `databases tables add --key`, or name it here + /// with `--key`. + #[arg(long, value_parser = ["replace", "append", "delete", "update", "upsert"])] + mode: Option, + + /// Format of the uploaded file: csv, json (newline-delimited), or + /// parquet. Detected from the file's extension when omitted; pass it + /// when the extension is absent or misleading. Not valid with + /// `--result-id`, which is always parquet. + #[arg(long, value_parser = ["csv", "json", "parquet"], conflicts_with = "result_id")] + format: Option, + + /// Key column for a `delete`/`update`/`upsert` load, repeatable for a + /// composite key (`--key tenant --key id`). Defaults to the key the + /// table was declared with. + #[arg(long = "key")] + key: Vec, }, /// Manage tables inside an instant database @@ -316,6 +338,57 @@ pub enum DatabaseTablesCommands { output: String, }, + /// Declare a new table in an instant database, with its key and layout + /// + /// Declaring a key is what enables the `delete`, `update`, and `upsert` + /// load modes on the table: those loads match existing rows by the key's + /// values. `--sorted-by` and `--partition-by` set the table's on-disk + /// layout and are fixed once the table exists. + Add { + /// Table name, or `schema.table` to target a schema other than --schema + table: String, + + /// Database id or name (defaults to current database) + #[arg(long)] + database: Option, + + /// Schema for a bare table name (default: public) + #[arg(long, default_value = "public")] + schema: String, + + /// Column that uniquely identifies a row, repeatable for a composite + /// key (`--key tenant --key id`). Omit to declare no key — the table + /// still loads with replace and append. + #[arg(long = "key")] + key: Vec, + + /// Column whose value is fixed by the key, repeatable. An optimisation + /// for keyed loads, and correctness-affecting: if the claim is false, a + /// keyed load can leave a duplicate of the key behind. Declare it only + /// where the invariant really holds. + #[arg(long = "key-determines")] + key_determines: Vec, + + /// Sort the table on this column, repeatable to sort on several in + /// order: `--sorted-by ts` or `--sorted-by ts=desc` (default asc). + /// Fixed once the table is created. + #[arg(long = "sorted-by")] + sorted_by: Vec, + + /// Partition the table by this column, repeatable to nest partitions in + /// order. `--partition-by region` partitions on the value itself; + /// `--partition-by created_at=month` on a calendar part (year, month, + /// day, hour). One partition per month needs both `created_at=year` and + /// `created_at=month`, or every March shares a partition. Fixed once + /// the table is created. + #[arg(long = "partition-by")] + partition_by: Vec, + + /// Output format + #[arg(long = "output", short = 'o', default_value = "table", value_parser = ["table", "json", "yaml"])] + output: String, + }, + /// Show column definitions for a table Show { /// Table as catalog.schema.table (or schema.table with an active database) @@ -326,7 +399,8 @@ pub enum DatabaseTablesCommands { output: String, }, - /// Load a parquet file or a saved query result into a table (replaces or appends) + /// Load a csv, json, or parquet file — or a saved query result — into a + /// table, replacing it, appending, or matching rows by key Load { /// Database id or name (defaults to current database) #[arg(long)] @@ -339,11 +413,11 @@ pub enum DatabaseTablesCommands { #[arg(long, default_value = "public")] schema: String, - /// Path to a local parquet file to upload and load + /// Path to a local file to upload and load (csv, json, or parquet) #[arg(long, conflicts_with_all = ["upload_id", "url", "result_id"])] file: Option, - /// URL of a remote parquet file to download and load + /// URL of a remote file to download and load (csv, json, or parquet) #[arg(long, conflicts_with_all = ["file", "upload_id", "result_id"])] url: Option, @@ -358,9 +432,30 @@ pub enum DatabaseTablesCommands { #[arg(long, conflicts_with_all = ["file", "url", "upload_id"])] result_id: Option, - /// Append rows to the table instead of replacing it (default: replace) - #[arg(long)] + /// Append rows to the table instead of replacing it (default: replace). + /// Shorthand for `--mode append`. + #[arg(long, conflicts_with = "mode")] append: bool, + + /// How the upload is applied to the table (default: replace). + /// `delete`, `update`, and `upsert` match existing rows by the table's + /// key — declare one with `databases tables add --key`, or name it here + /// with `--key`. + #[arg(long, value_parser = ["replace", "append", "delete", "update", "upsert"])] + mode: Option, + + /// Format of the uploaded file: csv, json (newline-delimited), or + /// parquet. Detected from the file's extension when omitted; pass it + /// when the extension is absent or misleading. Not valid with + /// `--result-id`, which is always parquet. + #[arg(long, value_parser = ["csv", "json", "parquet"], conflicts_with = "result_id")] + format: Option, + + /// Key column for a `delete`/`update`/`upsert` load, repeatable for a + /// composite key (`--key tenant --key id`). Defaults to the key the + /// table was declared with. + #[arg(long = "key")] + key: Vec, }, /// Delete a table from an instant database @@ -782,10 +877,9 @@ pub fn fork_database_request( } /// Build the typed `CreateDatabaseRequest` for the SDK's `databases().create` -/// handle, reusing [`create_database_request`] as the single source of truth for -/// the body shape. The delete+recreate path still consumes the raw JSON form -/// (it inspects raw error bodies), so the JSON builder stays; this just adapts -/// it for the typed call sites. +/// handle by adapting [`create_database_request`], which stays the single +/// source of truth for the body shape — the schema-seeding rules it encodes are +/// specified against the JSON form in this module's tests. fn create_database_typed_request( name: Option<&str>, catalog: Option<&str>, @@ -807,24 +901,296 @@ pub fn managed_table_delete_path(connection_id: &str, schema: &str, table: &str) format!("/connections/{connection_id}/schemas/{schema}/tables/{table}") } -pub fn load_table_request(upload_id: &str, mode: &str) -> serde_json::Value { - serde_json::json!({ +/// Build the `sorted_by` entries for a table declaration from `--sorted-by` +/// values, each `column` or `column=direction`. +fn sort_keys(values: &[String]) -> Result, String> { + values + .iter() + .map(|v| match v.split_once('=') { + None => Ok(serde_json::json!({ "column": v })), + Some((column, dir)) => { + let dir = dir.to_ascii_lowercase(); + if !matches!(dir.as_str(), "asc" | "desc") { + return Err(format!( + "--sorted-by '{v}': direction must be asc or desc, got '{dir}'" + )); + } + Ok(serde_json::json!({ "column": column, "direction": dir })) + } + }) + .collect() +} + +/// Build the `partition_by` entries from `--partition-by` values, each +/// `column` (identity) or `column=transform`. +fn partition_keys(values: &[String]) -> Result, String> { + values + .iter() + .map(|v| { + let (column, transform) = match v.split_once('=') { + None => (v.as_str(), "identity".to_string()), + Some((column, t)) => (column, t.to_ascii_lowercase()), + }; + if !matches!( + transform.as_str(), + "identity" | "year" | "month" | "day" | "hour" + ) { + return Err(format!( + "--partition-by '{v}': transform must be identity, year, month, day, or \ + hour, got '{transform}'" + )); + } + Ok(serde_json::json!({ "column": column, "transform": transform })) + }) + .collect() +} + +/// `databases tables add` — declare a table on an existing instant database. +#[allow(clippy::too_many_arguments)] +pub fn add_table( + workspace_id: &str, + database: Option<&str>, + table: &str, + schema: &str, + key: &[String], + key_determines: &[String], + sorted_by: &[String], + partition_by: &[String], + output: &str, +) { + use crossterm::style::Stylize; + + // `schema.table` overrides --schema, matching `databases create --table`. + let (schema, table) = match table.split_once('.') { + Some((s, t)) => (s, t), + None => (schema, table), + }; + + // key_determines names columns the key fixes, so a key must exist for it to + // mean anything; the server would take it and quietly ignore it. + if !key_determines.is_empty() && key.is_empty() { + eprintln!( + "{}", + "error: --key-determines describes columns fixed by the key, so it needs --key.".red() + ); + std::process::exit(1); + } + + let sorted_by = sort_keys(sorted_by).unwrap_or_else(|e| { + eprintln!("{}", format!("error: {e}").red()); + std::process::exit(1); + }); + let partition_by = partition_keys(partition_by).unwrap_or_else(|e| { + eprintln!("{}", format!("error: {e}").red()); + std::process::exit(1); + }); + + let database = resolve_current_database(database, workspace_id); + let api = Api::new(Some(workspace_id)); + let db = resolve_database(&api, &database); + + let mut body = serde_json::json!({ "name": table }); + if !key.is_empty() { + body["key"] = serde_json::json!(key); + } + if !key_determines.is_empty() { + body["key_determines"] = serde_json::json!(key_determines); + } + if !sorted_by.is_empty() { + body["sorted_by"] = serde_json::json!(sorted_by); + } + if !partition_by.is_empty() { + body["partition_by"] = serde_json::json!(partition_by); + } + + let (status, resp) = declare_table(&api, &db.id, schema, &body); + + if !status.is_success() { + eprintln!("{}", crate::util::api_error(resp).red()); + std::process::exit(1); + } + + let catalog = db + .default_catalog + .as_deref() + .or(db.name.as_deref()) + .unwrap_or(&db.id); + let declared = serde_json::json!({ + "table": format!("{catalog}.{schema}.{table}"), + "schema": schema, + "name": table, + "key": key, + "key_determines": key_determines, + "sorted_by": sorted_by, + "partition_by": partition_by, + }); + match output { + "json" => println!("{}", serde_json::to_string_pretty(&declared).unwrap()), + "yaml" => print!("{}", serde_yaml::to_string(&declared).unwrap()), + _ => { + println!("{}", format!("Declared {catalog}.{schema}.{table}").green()); + if key.is_empty() { + println!( + "{}", + "no key — loads with replace and append only. A key is fixed at \ + declaration and cannot be added later, so declare the next \ + table with --key if you need delete/update/upsert on it." + .dark_grey() + ); + } else { + println!("key: {}", key.join(", ")); + } + if !key_determines.is_empty() { + println!("determined by key: {}", key_determines.join(", ")); + } + // Sort and partition are fixed once the table exists, so echo what + // the server accepted — there is no second chance to check. + if !sorted_by.is_empty() { + println!("sorted: {}", describe_sort_keys(&sorted_by)); + } + if !partition_by.is_empty() { + println!("parts: {}", describe_partition_keys(&partition_by)); + } + } + } +} + +/// Render sort keys for the table output, e.g. `created_at desc, id`. +/// A key with no explicit direction prints bare, matching what was sent. +fn describe_sort_keys(keys: &[serde_json::Value]) -> String { + keys.iter() + .map(|k| { + let column = k["column"].as_str().unwrap_or_default(); + match k["direction"].as_str() { + Some(dir) => format!("{column} {dir}"), + None => column.to_string(), + } + }) + .collect::>() + .join(", ") +} + +/// Render partition keys for the table output, e.g. `created_at:year, +/// created_at:month, region`. `identity` prints as the bare column, since the +/// partition is the value itself. +fn describe_partition_keys(keys: &[serde_json::Value]) -> String { + keys.iter() + .map(|k| { + let column = k["column"].as_str().unwrap_or_default(); + match k["transform"].as_str() { + Some("identity") | None => column.to_string(), + Some(t) => format!("{column}:{t}"), + } + }) + .collect::>() + .join(", ") +} + +/// Declare `table` in `schema` on an existing instant database, via +/// `POST /databases/{id}/schemas/{schema}/tables`, with `body` carrying the +/// name and whatever key/layout the caller declared. +/// +/// Declares the schema first when the server says it is missing: the table +/// route 404s with "Schema '' is not declared" for a schema the database +/// was never created with. An already-declared table comes back 409, which is +/// left to the caller — re-declaring is a real conflict here, since the +/// existing table's key and layout are fixed and this call would not change +/// them. `tables remove` does not clear the declaration either: the table +/// leaves the listing but the name stays declared and still 409s, so a key +/// cannot be retrofitted onto a table that was declared without one. +fn declare_table( + api: &Api, + database_id: &str, + schema: &str, + body: &serde_json::Value, +) -> (reqwest::StatusCode, String) { + let tables_path = format!("/databases/{database_id}/schemas/{schema}/tables"); + let (status, resp) = api + .post_raw(&tables_path, body) + .unwrap_or_else(|e| e.exit()); + + if status.as_u16() == 404 && crate::util::api_error(resp.clone()).contains("not declared") { + let (s_status, s_resp) = api + .post_raw( + &format!("/databases/{database_id}/schemas"), + &serde_json::json!({ "name": schema }), + ) + .unwrap_or_else(|e| e.exit()); + // A concurrent declaration of the same schema (409) is not a failure — + // the schema exists either way, which is all the retry needs. + if !s_status.is_success() && s_status.as_u16() != 409 { + return (s_status, s_resp); + } + return api + .post_raw(&tables_path, body) + .unwrap_or_else(|e| e.exit()); + } + + (status, resp) +} + +/// Body for a load from a staged upload. +/// +/// `format` is omitted when `None` so the server resolves the format itself, +/// from the upload's recorded content type and then from the bytes. `key` is +/// omitted when empty, which tells the server to use the key the table was +/// declared with; it is ignored outside the keyed modes. +pub fn load_table_request( + upload_id: &str, + mode: &str, + format: Option<&str>, + key: &[String], +) -> serde_json::Value { + let mut body = serde_json::json!({ "mode": mode, "upload_id": upload_id, - }) + }); + if let Some(f) = format { + body["format"] = serde_json::json!(f); + } + if !key.is_empty() { + body["key"] = serde_json::json!(key); + } + body } -pub fn load_table_request_from_result(result_id: &str, mode: &str) -> serde_json::Value { - serde_json::json!({ +/// Body for a load from a persisted query result. +/// +/// No `format`: a stored result is always parquet, and the server rejects the +/// field alongside `result_id`. +pub fn load_table_request_from_result( + result_id: &str, + mode: &str, + key: &[String], +) -> serde_json::Value { + let mut body = serde_json::json!({ "mode": mode, "result_id": result_id, - }) + }); + if !key.is_empty() { + body["key"] = serde_json::json!(key); + } + body } -/// 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") - || Path::new(path).extension().and_then(|e| e.to_str()) == Some("parquet") +/// The load-request `format` implied by `name`'s extension, or `None` when the +/// extension names nothing we recognise. +/// +/// `None` is not an error: an upload announcing an unrecognised content type is +/// sniffed server-side, so the load simply omits `format` and lets the server +/// decide. `--format` overrides this either way. +pub fn format_for_path(name: &str) -> Option<&'static str> { + // Shares `extension_of` with the content type the upload announces, so the + // two can never disagree about what a name means. + match crate::client::sdk::extension_of(name) + .to_ascii_lowercase() + .as_str() + { + "parquet" => Some("parquet"), + "csv" => Some("csv"), + "json" | "jsonl" | "ndjson" => Some("json"), + _ => None, + } } fn table_rows(catalog: &str, tables: Vec) -> Vec { @@ -850,13 +1216,18 @@ fn upload_progress_style() -> ProgressStyle { .progress_chars("=>-") } -/// Upload an already-on-disk parquet file via the SDK's presigned direct-to- +/// Upload an already-on-disk data file via the SDK's presigned direct-to- /// storage flow, driving a single aggregate progress bar from the SDK's /// byte-granular progress callback. Returns the finalized upload id, or the /// seam's error (a `501 PRESIGN_UNSUPPORTED` surfaces an actionable message, /// not a fallback). The caller decides how to surface failure — `--url` must /// clean up its temp file before exiting, so this returns rather than exits. -fn upload_parquet_path(api: &Api, path: &Path, size: u64) -> Result { +fn upload_data_path( + api: &Api, + path: &Path, + content_type: &str, + size: u64, +) -> Result { let pb = ProgressBar::new(size); pb.set_style(upload_progress_style()); @@ -869,21 +1240,15 @@ fn upload_parquet_path(api: &Api, path: &Path, size: u64) -> Result String { - if !is_parquet_path(path) { - eprintln!( - "error: managed table loads require a parquet file (got '{}'). \ - Convert your data to parquet first.", - path - ); - std::process::exit(1); - } - +/// Upload `path`, announcing the content type its extension implies. No +/// extension is rejected: an unrecognised one announces +/// `application/octet-stream`, which the load sniffs. +fn upload_data_file(api: &Api, path: &str) -> String { let file_size = match std::fs::metadata(path) { Ok(m) => m.len(), Err(e) => { @@ -892,17 +1257,19 @@ fn upload_parquet_file(api: &Api, path: &str) -> String { } }; - upload_parquet_path(api, Path::new(path), file_size).unwrap_or_else(|e| e.exit()) + upload_data_path( + api, + Path::new(path), + crate::client::sdk::content_type_for_path(path), + file_size, + ) + .unwrap_or_else(|e| e.exit()) } -fn upload_parquet_url(api: &Api, url: &str) -> String { - if !is_parquet_path(url) { - eprintln!( - "error: managed table loads require a parquet URL ending in .parquet (got '{url}')." - ); - std::process::exit(1); - } - +/// Download `url` and upload it, announcing the content type the URL's own +/// extension implies — the staged temp file's name is generated and says +/// nothing about the bytes. +fn upload_data_url(api: &Api, url: &str) -> String { // The presigned upload needs a seekable, size-known source (the SDK opens // the path, declares its byte count, and PUTs it directly to storage), so // download the URL to a temp file first, then upload that file on the same @@ -942,7 +1309,15 @@ fn upload_parquet_url(api: &Api, url: &str) -> String { } }; - let temp = match download_to_temp(resp, &dl_pb) { + // The staged name carries the source URL's own extension, or none when the + // URL has none. It is advisory — the load resolves the format from the + // recorded content type and then the bytes, never the file name — but a + // staged name should not claim a format it cannot know. + let suffix = match crate::client::sdk::extension_of(url) { + "" => String::new(), + ext => format!(".{ext}"), + }; + let temp = match download_to_temp(resp, &suffix, &dl_pb) { Ok(t) => t, Err(e) => { dl_pb.finish_and_clear(); @@ -953,7 +1328,9 @@ fn upload_parquet_url(api: &Api, url: &str) -> String { dl_pb.finish_and_clear(); let size = std::fs::metadata(temp.path()).map(|m| m.len()).unwrap_or(0); - upload_temp_file(temp, |path| upload_parquet_path(api, path, size)).unwrap_or_else(|e| e.exit()) + let content_type = crate::client::sdk::content_type_for_path(url); + upload_temp_file(temp, |path| upload_data_path(api, path, content_type, size)) + .unwrap_or_else(|e| e.exit()) } /// Upload an already-downloaded temp file, guaranteeing the file is deleted @@ -975,19 +1352,20 @@ where result } -/// Stream a blocking HTTP response body to a freshly created temp file, -/// advancing `pb` as bytes land. Returns the open [`NamedTempFile`], which +/// Stream a blocking HTTP response body to a freshly created temp file named +/// with `suffix`, advancing `pb` as bytes land. Returns the open [`NamedTempFile`], which /// deletes the file on drop. Created atomically with `O_EXCL` + 0600 perms via /// `tempfile`, so it can't be redirected by a pre-planted symlink. fn download_to_temp( resp: reqwest::blocking::Response, + suffix: &str, pb: &ProgressBar, ) -> std::io::Result { use std::io::Write; let mut temp = tempfile::Builder::new() .prefix("hotdata-upload-") - .suffix(".parquet") + .suffix(suffix) .tempfile()?; let mut reader = pb.wrap_read(resp); @@ -2215,10 +2593,27 @@ pub fn tables_load( upload_id: Option<&str>, result_id: Option<&str>, append: bool, + mode: Option<&str>, + format: Option<&str>, + key: &[String], ) { use crossterm::style::Stylize; - let mode = if append { "append" } else { "replace" }; + // `--append` predates `--mode` and stays supported; clap keeps them mutually + // exclusive, so at most one is set. + let mode = match (mode, append) { + (Some(m), _) => m, + (None, true) => "append", + (None, false) => "replace", + }; + if matches!(mode, "delete" | "update" | "upsert") && result_id.is_some() { + eprintln!( + "error: --result-id loads a stored result, which carries every column — \ + mode '{mode}' matches rows by key and needs a key-shaped upload. \ + Use --file/--url/--upload-id, or mode replace/append." + ); + std::process::exit(1); + } // NOTE: this used to detect a database API token and route it through the // database-scoped endpoints (the connection-scoped managed paths and the @@ -2263,13 +2658,25 @@ pub fn tables_load( // 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. + // An explicit --format always wins; otherwise the source's own extension + // names the format, and an unrecognised one sends nothing so the server + // sniffs. A staged --upload-id carries the content type it was created + // with, so it too sends only what --format asked for. let body = match (result_id, upload_id, file, url) { - (Some(rid), None, None, None) => load_table_request_from_result(rid, mode), - (None, Some(id), None, None) => load_table_request(id, mode), - (None, None, Some(path), None) => { - load_table_request(&upload_parquet_file(&api, path), mode) - } - (None, None, None, Some(u)) => load_table_request(&upload_parquet_url(&api, u), mode), + (Some(rid), None, None, None) => load_table_request_from_result(rid, mode, key), + (None, Some(id), None, None) => load_table_request(id, mode, format, key), + (None, None, Some(path), None) => load_table_request( + &upload_data_file(&api, path), + mode, + format.or_else(|| format_for_path(path)), + key, + ), + (None, None, None, Some(u)) => load_table_request( + &upload_data_url(&api, u), + mode, + format.or_else(|| format_for_path(u)), + key, + ), (None, None, None, None) => { eprintln!( "error: one of --file , --url , --upload-id , or --result-id is required" @@ -2288,136 +2695,10 @@ pub fn tables_load( }); spinner.finish_and_clear(); - let (status, resp_body) = if !status.is_success() - // Upload-only recovery: the delete + recreate below mints a new database - // id, which would orphan a result (the server scopes it to the original - // database). A result load into an undeclared table is auto-declared - // server-side, so this path isn't needed for it — surface the error. - && result_id.is_none() - && crate::util::api_error(resp_body.clone()).contains("not declared") - { - // The table wasn't declared at create time. Collect existing tables so - // they are re-declared in the replacement database, then delete and - // recreate with all tables (including the new one) declared. - let (existing, _, _) = - collect_tables(&api, &db.default_connection_id, None, None, None, None); - let mut all_tables: Vec = existing - .iter() - .map(|t| format!("{}.{}", t.schema, t.table)) - .collect(); - let new_table_key = format!("{schema}.{table}"); - if !all_tables.contains(&new_table_key) { - all_tables.push(new_table_key); - } - - // Warn if any existing table has synced data — delete+recreate will lose it. - let synced: Vec = existing - .iter() - .filter(|t| t.synced) - .map(|t| format!("{}.{}", t.schema, t.table)) - .collect(); - if !synced.is_empty() { - use crossterm::style::Stylize; - let catalog = db - .default_catalog - .as_deref() - .or(db.name.as_deref()) - .unwrap_or(&db.id); - eprintln!( - "{}", - format!( - "warning: declaring '{}' requires recreating the database '{catalog}'. \ - The following tables have loaded data that will be lost:\n {}", - table, - synced.join(", ") - ) - .yellow() - ); - if crate::util::is_interactive() { - use std::io::Write; - eprint!("Proceed and lose this data? [y/N] "); - std::io::stderr().flush().unwrap(); - let mut input = String::new(); - std::io::stdin().read_line(&mut input).unwrap(); - if !input.trim().eq_ignore_ascii_case("y") { - eprintln!("{}", "Aborted.".red()); - std::process::exit(1); - } - } else { - eprintln!( - "{}", - "error: cannot auto-declare table in non-interactive mode — existing data would be lost. \ - Declare all tables up front with 'databases create --table '." - .red() - ); - std::process::exit(1); - } - } - - let (del_status, del_body) = api - .delete_raw(&format!("/databases/{}", db.id)) - .unwrap_or_else(|e| e.exit()); - if !del_status.is_success() { - eprintln!("{}", crate::util::api_error(del_body).red()); - std::process::exit(1); - } - let create_body = create_database_request( - db.name.as_deref(), - db.default_catalog.as_deref(), - schema, - &all_tables, - db.expires_at.as_deref(), - ); - let (create_status, create_body_resp) = api - .post_raw("/databases", &create_body) - .unwrap_or_else(|e| e.exit()); - if !create_status.is_success() { - eprintln!("{}", crate::util::api_error(create_body_resp).red()); - std::process::exit(1); - } - let new_db: CreateDatabaseResponse = match serde_json::from_str(&create_body_resp) { - Ok(v) => v, - Err(e) => { - eprintln!("error parsing create response: {e}"); - std::process::exit(1); - } - }; - let _ = crate::config::save_current_database("default", workspace_id, &new_db.id); - // Instant databases have no add-table endpoint, so declaring a new table - // is a delete + recreate — which mints a NEW database id. Surface that - // explicitly: the id printed by `databases create` is now stale, and - // id-based automation (e.g. `databases delete `) would - // otherwise fail with "no database with id". Reference by catalog instead. - { - use crossterm::style::Stylize; - let catalog = db - .default_catalog - .as_deref() - .or(db.name.as_deref()) - .unwrap_or(&db.id); - eprintln!( - "{}", - format!( - "note: table '{table}' was not declared — recreated database '{catalog}' to add it \ - (id {} → {}). Instant databases are recreated when a new table is loaded; \ - reference them by catalog ('{catalog}'), not the create-time id.", - db.id, new_db.id - ) - .yellow() - ); - } - let new_path = managed_table_load_path(&new_db.default_connection_id, schema, table); - let spinner = crate::util::spinner("Loading table..."); - let result = api.post_raw(&new_path, &body).unwrap_or_else(|e| { - spinner.finish_and_clear(); - e.exit() - }); - spinner.finish_and_clear(); - result - } else { - (status, resp_body) - }; - + // No undeclared-table recovery here: the load endpoint declares a missing + // table (and a missing schema) itself, so a load never fails for want of a + // declaration. `databases tables add` is for declaring a table *with* a + // key or a layout, which a load cannot infer. if !status.is_success() { eprintln!("{}", crate::util::api_error(resp_body).red()); std::process::exit(1); @@ -3170,36 +3451,128 @@ mod tests { #[test] fn load_table_request_carries_mode() { - let body = load_table_request("upl_abc", "replace"); + let body = load_table_request("upl_abc", "replace", None, &[]); assert_eq!(body["mode"], "replace"); assert_eq!(body["upload_id"], "upl_abc"); - let body = load_table_request("upl_abc", "append"); + let body = load_table_request("upl_abc", "append", None, &[]); assert_eq!(body["mode"], "append"); assert_eq!(body["upload_id"], "upl_abc"); } + #[test] + fn load_table_request_omits_format_and_key_unless_given() { + // Omitted, not null: the server resolves the format itself from the + // upload's content type, and an absent key means "the table's own". + let body = load_table_request("upl_abc", "replace", None, &[]); + assert!(body.get("format").is_none()); + assert!(body.get("key").is_none()); + + let key = vec!["tenant".to_string(), "id".to_string()]; + let body = load_table_request("upl_abc", "upsert", Some("csv"), &key); + assert_eq!(body["format"], "csv"); + assert_eq!(body["key"], serde_json::json!(["tenant", "id"])); + } + #[test] fn load_table_request_from_result_carries_mode() { - let body = load_table_request_from_result("rslt_abc", "replace"); + let body = load_table_request_from_result("rslt_abc", "replace", &[]); 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()); - let body = load_table_request_from_result("rslt_abc", "append"); + let body = load_table_request_from_result("rslt_abc", "append", &[]); assert_eq!(body["mode"], "append"); assert_eq!(body["result_id"], "rslt_abc"); assert!(body.get("upload_id").is_none()); } #[test] - fn is_parquet_path_by_extension() { - assert!(is_parquet_path("/data/orders.parquet")); - assert!(is_parquet_path("/data/ORDERS.PARQUET")); - assert!(is_parquet_path("file.parquet")); - assert!(!is_parquet_path("/data/orders.csv")); - assert!(!is_parquet_path("/data/orders")); + fn result_load_never_sends_a_format() { + // A stored result is always parquet and the server rejects `format` + // alongside `result_id`, so the builder has no way to send one. + let body = load_table_request_from_result("rslt_abc", "replace", &["id".to_string()]); + assert!(body.get("format").is_none()); + assert_eq!(body["key"], serde_json::json!(["id"])); + } + + #[test] + fn format_for_path_reads_the_extension() { + assert_eq!(format_for_path("/data/orders.parquet"), Some("parquet")); + assert_eq!(format_for_path("/data/ORDERS.PARQUET"), Some("parquet")); + assert_eq!(format_for_path("/data/orders.csv"), Some("csv")); + // Newline-delimited JSON answers to three spellings, all `json`. + assert_eq!(format_for_path("a.json"), Some("json")); + assert_eq!(format_for_path("a.jsonl"), Some("json")); + assert_eq!(format_for_path("a.ndjson"), Some("json")); + // Unknown and absent extensions send no format at all, leaving the + // server to sniff the bytes — never a client-side rejection. + assert_eq!(format_for_path("/data/orders.txt"), None); + assert_eq!(format_for_path("/data/orders"), None); + } + + #[test] + fn sort_and_partition_keys_render_for_the_table_output() { + // The layout is fixed at declaration, so the echo is the user's only + // confirmation of what the server took. + let sorted = sort_keys(&["created_at=desc".to_string(), "id".to_string()]).unwrap(); + assert_eq!(describe_sort_keys(&sorted), "created_at desc, id"); + + let parts = partition_keys(&[ + "created_at=year".to_string(), + "created_at=month".to_string(), + "region".to_string(), + ]) + .unwrap(); + assert_eq!( + describe_partition_keys(&parts), + "created_at:year, created_at:month, region", + "identity prints bare — the partition is the value itself" + ); + } + + #[test] + fn format_for_path_ignores_a_url_query_string() { + // A presigned storage URL carries its signature in the query, so the + // extension has to be read from the path alone. + assert_eq!( + format_for_path("https://s3.example.com/b/listings.parquet?X-Amz-Signature=abc123"), + Some("parquet") + ); + assert_eq!( + format_for_path("https://example.com/export.csv?v=2#top"), + Some("csv") + ); + // A dot in an earlier path segment is not the file's extension. + assert_eq!(format_for_path("https://example.com/v1.2/export"), None); + } + + #[test] + fn sort_keys_default_to_ascending_and_reject_a_bad_direction() { + assert_eq!( + sort_keys(&["ts".to_string()]).unwrap(), + vec![serde_json::json!({"column": "ts"})], + "no direction is omitted, so the server applies its own default" + ); + assert_eq!( + sort_keys(&["ts=DESC".to_string()]).unwrap(), + vec![serde_json::json!({"column": "ts", "direction": "desc"})] + ); + assert!(sort_keys(&["ts=sideways".to_string()]).is_err()); + } + + #[test] + fn partition_keys_default_to_identity_and_reject_a_bad_transform() { + assert_eq!( + partition_keys(&["region".to_string()]).unwrap(), + vec![serde_json::json!({"column": "region", "transform": "identity"})] + ); + assert_eq!( + partition_keys(&["created_at=Month".to_string()]).unwrap(), + vec![serde_json::json!({"column": "created_at", "transform": "month"})] + ); + assert!(partition_keys(&["created_at=week".to_string()]).is_err()); } #[test] @@ -3530,7 +3903,8 @@ mod tests { "/v1/connections/conn_default/schemas/public/tables/orders/loads", ) .match_body(mockito::Matcher::JsonString( - serde_json::to_string(&load_table_request("upl_123", "replace")).unwrap(), + serde_json::to_string(&load_table_request("upl_123", "replace", None, &[])) + .unwrap(), )) .with_status(200) .with_body( @@ -3547,7 +3921,7 @@ mod tests { 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("upl_123", "replace"); + let body = load_table_request("upl_123", "replace", None, &[]); let (status, resp_body) = api.post_raw(&path, &body).unwrap(); assert!(status.is_success()); let parsed: LoadManagedTableResponse = serde_json::from_str(&resp_body).unwrap(); @@ -3577,7 +3951,7 @@ mod tests { ) .match_header("X-Database-Id", "db_1") .match_body(mockito::Matcher::JsonString( - serde_json::to_string(&load_table_request_from_result("rslt_123", "replace")) + serde_json::to_string(&load_table_request_from_result("rslt_123", "replace", &[])) .unwrap(), )) .with_status(200) @@ -3598,7 +3972,7 @@ mod tests { // carries X-Database-Id. let api = api.scoped_to_database_opt(Some(db.id.as_str())); let path = managed_table_load_path(&db.default_connection_id, "public", "orders"); - let body = load_table_request_from_result("rslt_123", "replace"); + let body = load_table_request_from_result("rslt_123", "replace", &[]); let (status, resp_body) = api.post_raw(&path, &body).unwrap(); assert!(status.is_success()); let parsed: LoadManagedTableResponse = serde_json::from_str(&resp_body).unwrap(); diff --git a/src/commands/indexes.rs b/src/commands/indexes.rs index 510b9cb..fd74694 100644 --- a/src/commands/indexes.rs +++ b/src/commands/indexes.rs @@ -33,6 +33,17 @@ impl Index { .clone() .or_else(|| self.columns.first().cloned()) } + + /// Columns this index added to the table. Only an auto-embed vector index + /// has any: `source_column` is the text it reads, and `columns` is the + /// embedding column it wrote. A direct vector, BM25, or sorted index + /// indexes columns that were already there, so it generated none. + fn generated_columns(&self) -> Vec { + match self.source_column { + Some(_) => self.columns.clone(), + None => Vec::new(), + } + } } #[derive(Serialize)] @@ -521,6 +532,12 @@ pub struct LocatedIndex { pub table: String, pub index_type: String, pub search_column: String, + /// Columns the index generated on the table rather than columns that were + /// already there — the `{column}_embedding` an auto-embed vector index + /// materialises. Empty for every other index kind. A search excludes these + /// from its default projection: they are 1536-wide float lists nobody + /// asked for. + pub generated_columns: Vec, pub status: String, pub metric: Option, } @@ -578,6 +595,7 @@ pub fn locate_by_name( table, index_type: one.inner.index_type.clone(), search_column, + generated_columns: one.inner.generated_columns(), status: one.inner.status.clone(), metric: one.inner.metric.clone(), }) diff --git a/src/commands/query.rs b/src/commands/query.rs index b54c007..eff4390 100644 --- a/src/commands/query.rs +++ b/src/commands/query.rs @@ -105,13 +105,12 @@ fn value_to_string(v: &Value) -> String { Value::Bool(b) => b.to_string(), Value::Number(n) => n.to_string(), Value::String(s) => s.clone(), - Value::Array(arr) => { - let (formatted, count) = crate::output::table::truncate_array(arr); - match count { - Some(n) => format!("{formatted} ({n} items)"), - None => formatted, - } - } + // Rendered whole, not abbreviated. `value_to_string` feeds `-o csv`, + // whose consumer is a program: a long list shortened to + // "[1, 2, 3, ..., 9] (1536 items)" is silent data loss in a format + // nobody re-reads by eye. The table renderer abbreviates separately, + // where a human is the reader and the width is the constraint. + Value::Array(_) => v.to_string(), Value::Object(_) => v.to_string(), } } diff --git a/src/commands/search.rs b/src/commands/search.rs index b13b5d9..d2cd4e7 100644 --- a/src/commands/search.rs +++ b/src/commands/search.rs @@ -199,6 +199,30 @@ fn parse_table(workspace_id: &str, table: &str) -> (FromTarget, String, String) } } +/// Quote a column for use inside a wildcard `EXCLUDE` list. Index columns are +/// user-named (`search create --output-column`), so they can need quoting and +/// can contain a quote character. +fn quote_ident(name: &str) -> String { + format!("\"{}\"", name.replace('"', "\"\"")) +} + +/// The default projection for a search: everything the table has, minus the +/// columns the index generated. +/// +/// An auto-embed vector index materialises a `{column}_embedding` column on the +/// table, so a bare `*` returns a 1536-float list in every row — tens of +/// kilobytes per search the caller did not ask for, and a column claiming +/// terminal width the real ones need. Excluding it by default keeps the wire +/// small and the row readable; `--select '*'` asks for it back, and naming it +/// in `--select` still works. +fn default_projection(generated_columns: &[String]) -> String { + if generated_columns.is_empty() { + return "*".to_string(); + } + let excluded: Vec = generated_columns.iter().map(|c| quote_ident(c)).collect(); + format!("* EXCLUDE ({})", excluded.join(", ")) +} + /// Build the SQL a search runs: `bm25_search(...)` for text, server-side /// `vector_distance(...)` for vector. fn build_search_sql( @@ -207,6 +231,7 @@ fn build_search_sql( column: &str, query: &str, select: Option<&str>, + generated_columns: &[String], limit: u32, ) -> String { match index_type { @@ -214,7 +239,7 @@ fn build_search_sql( let bm25_columns = match select { Some(cols) if cols.split(',').any(|c| c.trim() == "score") => cols.to_string(), Some(cols) => format!("{}, score", cols), - None => "*".to_string(), + None => default_projection(generated_columns), }; format!( "SELECT {} FROM bm25_search('{}', '{}', '{}') ORDER BY score DESC LIMIT {}", @@ -229,7 +254,9 @@ fn build_search_sql( // metric from the index metadata; the caller names the source column. _ => format!( "SELECT {}, vector_distance({}, '{}') AS dist FROM {} ORDER BY dist LIMIT {}", - select.unwrap_or("*"), + select + .map(str::to_string) + .unwrap_or_else(|| default_projection(generated_columns)), column, query.replace('\'', "''"), table_fqn, @@ -391,6 +418,7 @@ pub fn run( &loc.search_column, text, select, + &loc.generated_columns, limit, ); // Search generates HotSQL directly — never a foreign dialect. @@ -409,3 +437,109 @@ fn remove(workspace_id: &str, database: Option<&str>, name: &str) { name, ); } + +#[cfg(test)] +mod tests { + use super::*; + + const EMB: &str = "txt_embedding"; + + fn vector_sql(select: Option<&str>, generated: &[String]) -> String { + build_search_sql( + "vector", + "cat.public.d", + "txt", + "puppy", + select, + generated, + 10, + ) + } + + #[test] + fn vector_search_excludes_the_generated_embedding_column() { + let sql = vector_sql(None, &[EMB.to_string()]); + assert!( + sql.starts_with(r#"SELECT * EXCLUDE ("txt_embedding"), vector_distance("#), + "sql: {sql}" + ); + } + + #[test] + fn a_direct_vector_index_generates_nothing_so_the_star_stays_bare() { + // No embedding column was materialised, so there is nothing to exclude + // and the projection must not grow an empty EXCLUDE list. + let sql = vector_sql(None, &[]); + assert!(sql.starts_with("SELECT *, vector_distance("), "sql: {sql}"); + assert!(!sql.contains("EXCLUDE"), "sql: {sql}"); + } + + #[test] + fn an_explicit_select_is_honoured_verbatim() { + // `--select '*'` is the documented way to ask for the embedding back, + // so it must survive untouched even when a generated column exists. + let sql = vector_sql(Some("*"), &[EMB.to_string()]); + assert!(sql.starts_with("SELECT *, vector_distance("), "sql: {sql}"); + assert!(!sql.contains("EXCLUDE"), "sql: {sql}"); + + // And naming the embedding column explicitly still reaches it. + let sql = vector_sql(Some("id, txt_embedding"), &[EMB.to_string()]); + assert!( + sql.starts_with("SELECT id, txt_embedding, vector_distance("), + "sql: {sql}" + ); + } + + #[test] + fn bm25_search_keeps_a_bare_star_and_the_score() { + // A BM25 index generates no columns, so its projection never grows an + // EXCLUDE list. Nor can another index supply one: `locate_by_name` + // reports only the named index's own generated columns, and the server + // refuses to put an embedding-backed vector index on a table that + // carries any other index ("Embedding-backed vector indexes cannot + // coexist with other indexes on the same table"). Both arms of that + // are why this stays `*`. + let sql = build_search_sql("bm25", "cat.public.d", "body", "puppy", None, &[], 10); + assert!(sql.starts_with("SELECT * FROM bm25_search("), "sql: {sql}"); + assert!(!sql.contains("EXCLUDE"), "sql: {sql}"); + + // An explicit --select still gets `score` appended, unchanged. + let sql = build_search_sql("bm25", "cat.public.d", "body", "puppy", Some("id"), &[], 10); + assert!( + sql.starts_with("SELECT id, score FROM bm25_search("), + "sql: {sql}" + ); + } + + #[test] + fn the_projection_helper_is_shared_and_general() { + // `default_projection` serves both index kinds, so it is specified on + // its own rather than only through the branch that can reach it today. + assert_eq!(default_projection(&[]), "*"); + assert_eq!( + default_projection(&["txt_embedding".to_string()]), + r#"* EXCLUDE ("txt_embedding")"# + ); + } + + #[test] + fn a_generated_column_name_is_quoted_for_the_exclude_list() { + // Index columns are user-named via `search create --output-column`, so + // a name needing quotes (or containing one) must not break the SQL. + assert_eq!(quote_ident("plain"), r#""plain""#); + assert_eq!(quote_ident("has space"), r#""has space""#); + assert_eq!(quote_ident(r#"we"ird"#), r#""we""ird""#); + } + + #[test] + fn several_generated_columns_are_all_excluded() { + let sql = vector_sql( + None, + &["a_embedding".to_string(), "b_embedding".to_string()], + ); + assert!( + sql.contains(r#"* EXCLUDE ("a_embedding", "b_embedding")"#), + "sql: {sql}" + ); + } +} diff --git a/src/main.rs b/src/main.rs index efc890c..1400746 100644 --- a/src/main.rs +++ b/src/main.rs @@ -315,6 +315,9 @@ fn main() { upload_id, result_id, append, + mode, + format, + key, }) => databases::tables_load( &workspace_id, Some(catalog.as_str()), @@ -325,6 +328,9 @@ fn main() { upload_id.as_deref(), result_id.as_deref(), append, + mode.as_deref(), + format.as_deref(), + &key, ), Some(DatabasesCommands::Tables { database, command }) => match command { Some(DatabaseTablesCommands::List { @@ -366,6 +372,26 @@ fn main() { ) } } + Some(DatabaseTablesCommands::Add { + table, + database: db_flag, + schema, + key, + key_determines, + sorted_by, + partition_by, + output, + }) => databases::add_table( + &workspace_id, + db_flag.as_deref().or(database.as_deref()), + &table, + &schema, + &key, + &key_determines, + &sorted_by, + &partition_by, + &output, + ), Some(DatabaseTablesCommands::Show { table, output }) => { tables::show(&workspace_id, &table, &output) } @@ -378,6 +404,9 @@ fn main() { upload_id, result_id, append, + mode, + format, + key, }) => databases::tables_load( &workspace_id, db_flag.as_deref().or(database.as_deref()), @@ -388,6 +417,9 @@ fn main() { upload_id.as_deref(), result_id.as_deref(), append, + mode.as_deref(), + format.as_deref(), + &key, ), Some(DatabaseTablesCommands::Delete { database: db_flag, diff --git a/tests/databases_cli.rs b/tests/databases_cli.rs index df1c39a..830d271 100644 --- a/tests/databases_cli.rs +++ b/tests/databases_cli.rs @@ -130,3 +130,304 @@ fn databases_tables_load_rejects_both_file_and_upload_id_at_parse_time() { "output: {combined}" ); } + +#[test] +fn databases_tables_help_lists_add() { + let output = hotdata() + .args(["databases", "tables", "--help"]) + .output() + .unwrap(); + assert!(output.status.success()); + let help = String::from_utf8_lossy(&output.stdout); + assert!(help.contains("add"), "help: {help}"); + assert!(help.contains("load"), "help: {help}"); + assert!(help.contains("show"), "help: {help}"); +} + +#[test] +fn databases_tables_add_help_documents_key_and_layout_flags() { + let output = hotdata() + .args(["databases", "tables", "add", "--help"]) + .output() + .unwrap(); + assert!(output.status.success()); + let help = String::from_utf8_lossy(&output.stdout); + assert!(help.contains("--key"), "help: {help}"); + assert!(help.contains("--key-determines"), "help: {help}"); + assert!(help.contains("--sorted-by"), "help: {help}"); + assert!(help.contains("--partition-by"), "help: {help}"); +} + +#[test] +fn databases_tables_add_requires_a_table_argument() { + let output = hotdata() + .args(["databases", "tables", "add"]) + .output() + .unwrap(); + assert!(!output.status.success()); + let combined = format!( + "{}{}", + String::from_utf8_lossy(&output.stdout), + String::from_utf8_lossy(&output.stderr) + ); + assert!( + combined.contains("required") || combined.contains("TABLE"), + "output: {combined}" + ); +} + +#[test] +fn databases_load_help_documents_mode_format_and_key() { + let output = hotdata() + .args(["databases", "load", "--help"]) + .output() + .unwrap(); + assert!(output.status.success()); + let help = String::from_utf8_lossy(&output.stdout); + assert!(help.contains("--mode"), "help: {help}"); + assert!(help.contains("--format"), "help: {help}"); + assert!(help.contains("--key"), "help: {help}"); + // The keyed modes are the point of --mode; they must be discoverable. + for mode in ["replace", "append", "delete", "update", "upsert"] { + assert!(help.contains(mode), "help missing '{mode}': {help}"); + } +} + +#[test] +fn databases_load_rejects_an_unknown_mode_at_parse_time() { + let output = hotdata() + .args([ + "databases", + "load", + "--catalog", + "c", + "--table", + "t", + "--file", + "a.csv", + "--mode", + "merge", + ]) + .output() + .unwrap(); + assert!(!output.status.success()); + let combined = format!( + "{}{}", + String::from_utf8_lossy(&output.stdout), + String::from_utf8_lossy(&output.stderr) + ); + assert!( + combined.contains("invalid value") || combined.contains("possible values"), + "output: {combined}" + ); +} + +#[test] +fn databases_load_rejects_append_together_with_mode() { + // `--append` is the old shorthand for `--mode append`; accepting both would + // leave the effective mode ambiguous, so clap refuses the pair. + let output = hotdata() + .args([ + "databases", + "load", + "--catalog", + "c", + "--table", + "t", + "--file", + "a.csv", + "--append", + "--mode", + "upsert", + ]) + .output() + .unwrap(); + assert!(!output.status.success()); + let combined = format!( + "{}{}", + String::from_utf8_lossy(&output.stdout), + String::from_utf8_lossy(&output.stderr) + ); + assert!( + combined.contains("cannot be used with"), + "output: {combined}" + ); +} + +#[test] +fn databases_load_accepts_a_non_parquet_file_at_parse_time() { + // A csv must get past argument parsing and the client entirely — the load + // reads csv, json, and parquet, and an unrecognised extension is the + // server's call, not a client-side rejection. Without credentials the run + // fails later, on auth or the network, never on the file's extension. + let output = hotdata() + .args([ + "databases", + "load", + "--catalog", + "c", + "--table", + "t", + "--file", + "/nonexistent/data.csv", + ]) + .env("HOTDATA_CONFIG_DIR", "/nonexistent-config-dir") + .env("HOTDATA_WORKSPACE", "workffffffffffffffffffffffffffff") + .output() + .unwrap(); + let combined = format!( + "{}{}", + String::from_utf8_lossy(&output.stdout), + String::from_utf8_lossy(&output.stderr) + ); + assert!( + !combined.contains("require a parquet"), + "csv was rejected client-side: {combined}" + ); +} + +#[test] +fn databases_load_rejects_format_together_with_result_id() { + // A stored result is always parquet and the load endpoint rejects `format` + // beside `result_id`, so the request builder has no field to put it in. + // Without this conflict the flag would be accepted and silently dropped. + let output = hotdata() + .args([ + "databases", + "load", + "--catalog", + "c", + "--table", + "t", + "--result-id", + "rslt_1", + "--format", + "csv", + ]) + .output() + .unwrap(); + assert!(!output.status.success()); + let combined = format!( + "{}{}", + String::from_utf8_lossy(&output.stdout), + String::from_utf8_lossy(&output.stderr) + ); + assert!( + combined.contains("cannot be used with"), + "output: {combined}" + ); +} + +#[test] +fn databases_load_rejects_a_keyed_mode_with_result_id() { + // Not a clap conflict — `--mode` conflicts with `--result-id` only for the + // three keyed values, so the check is hand-rolled. It runs before any + // config or network access, which is what makes it reachable here. + for mode in ["delete", "update", "upsert"] { + let output = hotdata() + .args([ + "databases", + "load", + "--catalog", + "c", + "--table", + "t", + "--result-id", + "rslt_1", + "--mode", + mode, + ]) + .env("HOTDATA_CONFIG_DIR", "/nonexistent-config-dir") + .env("HOTDATA_WORKSPACE", "workffffffffffffffffffffffffffff") + .output() + .unwrap(); + assert!(!output.status.success(), "mode {mode} was accepted"); + let combined = format!( + "{}{}", + String::from_utf8_lossy(&output.stdout), + String::from_utf8_lossy(&output.stderr) + ); + assert!( + combined.contains("matches rows by key"), + "mode {mode} output: {combined}" + ); + } + + // The non-keyed modes must still reach the network on the same input. + for mode in ["replace", "append"] { + let output = hotdata() + .args([ + "databases", + "load", + "--catalog", + "c", + "--table", + "t", + "--result-id", + "rslt_1", + "--mode", + mode, + ]) + .env("HOTDATA_CONFIG_DIR", "/nonexistent-config-dir") + .env("HOTDATA_WORKSPACE", "workffffffffffffffffffffffffffff") + .output() + .unwrap(); + let combined = format!( + "{}{}", + String::from_utf8_lossy(&output.stdout), + String::from_utf8_lossy(&output.stderr) + ); + assert!( + !combined.contains("matches rows by key"), + "mode {mode} was rejected as keyed: {combined}" + ); + } +} + +#[test] +fn databases_tables_add_rejects_key_determines_without_a_key() { + // `--key-determines` names columns the key fixes, so it is meaningless + // without `--key` — the server would accept and ignore it. + let output = hotdata() + .args([ + "databases", + "tables", + "add", + "t", + "--key-determines", + "tenant", + ]) + .env("HOTDATA_CONFIG_DIR", "/nonexistent-config-dir") + .env("HOTDATA_WORKSPACE", "workffffffffffffffffffffffffffff") + .output() + .unwrap(); + assert!(!output.status.success()); + let combined = format!( + "{}{}", + String::from_utf8_lossy(&output.stdout), + String::from_utf8_lossy(&output.stderr) + ); + assert!(combined.contains("needs --key"), "output: {combined}"); +} + +#[test] +fn databases_tables_add_rejects_a_bad_sort_direction_and_transform() { + for (flag, value, needle) in [ + ("--sorted-by", "ts=sideways", "asc or desc"), + ("--partition-by", "created_at=week", "identity, year, month"), + ] { + let output = hotdata() + .args(["databases", "tables", "add", "t", flag, value]) + .env("HOTDATA_CONFIG_DIR", "/nonexistent-config-dir") + .env("HOTDATA_WORKSPACE", "workffffffffffffffffffffffffffff") + .output() + .unwrap(); + assert!(!output.status.success(), "{flag} {value} was accepted"); + let combined = format!( + "{}{}", + String::from_utf8_lossy(&output.stdout), + String::from_utf8_lossy(&output.stderr) + ); + assert!(combined.contains(needle), "{flag} output: {combined}"); + } +}