diff --git a/hyperdb-mcp/CHANGELOG.md b/hyperdb-mcp/CHANGELOG.md index b2f2bec..4a770bb 100644 --- a/hyperdb-mcp/CHANGELOG.md +++ b/hyperdb-mcp/CHANGELOG.md @@ -68,20 +68,45 @@ and this project adheres to [Semantic Versioning](https://semver.org/). both `database` and `persist` are set, `database` wins. Combining `persist: true` with `--ephemeral-only` returns a clear `InvalidArgument` error. -- **Catalog updates are database-aware.** `_table_catalog` rows are - only written when ingesting to the primary or persistent database. - User-attached databases manage their own metadata if any. - -### Limitations (deferred to a follow-up) - -- `load_files` and `watch_directory` reject `database` / `persist` - with a clear error — their connection pool is bound to the primary - database and can't reach attached databases. Use `load_file` with - `persist: true` for one-off persistent ingests until pool routing - ships. -- `load_file` with `mode: "merge"` rejects non-primary `database`. - The merge implementation uses a temp table and cross-database DML - isn't yet verified across all hyperd versions. +- **Cross-database merge.** `load_file` with `mode: "merge"` now + accepts any writable `database`. The merge path keeps the temp + table inside the target database so DELETE-USING and INSERT-SELECT + stay single-DB — no cross-database DML is required. Engine + helpers `table_exists_in`, `column_metadata_in`, and + `alter_table_add_columns_in` carry the routing. +- **`export(format="hyper", database=...)`.** The hyper-format + export now snapshots whichever database the caller named (via a + new `ExportOptions.source_db` field plumbed into + `populate_export_target`). Default behavior (snapshot primary) + unchanged. +- **`load_files` and `watch_directory` accept `database` / `persist`.** + Their connection pool now opens the resolved target's `.hyper` + file directly as its workspace, so unqualified ingest SQL routes + into the right database without further plumbing. The watcher's + reconnect-recovery path re-resolves the target so a hyperd + restart picks the right file. `WatcherHandle` records its + `target_db`; `detach_database` rejects with `InvalidArgument` if + any active watcher targets the alias (call `unwatch_directory` + first to release it). +- **Per-database `_table_catalog`.** Every writable database + receives its own catalog, lazily seeded on first ingest. The + catalog CRUD API gains `*_in(target_db)` siblings + (`ensure_exists_in`, `upsert_stub_in`, `set_metadata_in`, + `get_in`, `list_in`, `delete_for_in`, `reconcile_in`). The + per-engine catalog-presence cache is now keyed by canonical + alias (`Mutex>`); detach clears the entry. +- **`set_table_metadata.database` parameter.** Routes the catalog + write to the named database's `_table_catalog`. Read-only + attachments are rejected up front with a clear "re-attach with + writable:true" message. + +### Changed (breaking — pre-1.0) + +- `Engine::catalog_present_in_persistent` → `Engine::catalog_present_in(alias, prober)`. +- `Engine::mark_catalog_present` → `Engine::mark_catalog_present_for(alias)`. +- `Engine::catalog_present_cache` field shape: `Mutex>` → `Mutex>` keyed by lowercased alias. +- New `Engine::clear_catalog_cache_for(alias)` paired with `detach_database`. +- `table_catalog::ensure_exists_in_database(engine, alias)` is now a deprecated wrapper over `ensure_exists_in(engine, Some(alias))`. ### Removed @@ -92,10 +117,12 @@ and this project adheres to [Semantic Versioning](https://semver.org/). ### Performance -- **Per-engine `_table_catalog` presence cache.** Catalog reads/writes +- **Per-database `_table_catalog` presence cache.** Catalog reads/writes used to round-trip a `pg_catalog.pg_tables` probe on every call; - now the existence check is cached for the engine's lifetime and - primed immediately after `CREATE TABLE IF NOT EXISTS`. + now the existence check is cached per (engine, alias) and primed + immediately after `CREATE TABLE IF NOT EXISTS`. `detach_database` + clears the alias's entry so a re-attach to a different file isn't + served stale. ### Fixed diff --git a/hyperdb-mcp/README.md b/hyperdb-mcp/README.md index b086a55..62a360d 100644 --- a/hyperdb-mcp/README.md +++ b/hyperdb-mcp/README.md @@ -203,7 +203,7 @@ describe({ database: "persistent" }) sample({ table: "customers", database: "persistent" }) ``` -The `database` parameter is available on `query`, `execute`, `load_data`, `load_file`, `describe`, `sample`, `chart`, and `export`. The shorthand `persist: true` (sugar for `database: "persistent"`) is available on `load_data` and `load_file`. Pass any user-attached writable alias (created via `attach_database`) to target a custom database. +The `database` parameter is available on `query`, `execute`, `load_data`, `load_file`, `load_files`, `watch_directory`, `describe`, `sample`, `chart`, `export`, and `set_table_metadata`. The shorthand `persist: true` (sugar for `database: "persistent"`) is available on `load_data`, `load_file`, `load_files`, and `watch_directory`. Pass any user-attached writable alias (created via `attach_database`) to target a custom database. (`query_data` and `query_file` are one-shot tools that materialize the inline data into their own temp table and query it — they do not accept a `database` parameter because the data isn't in a persisted database to begin with.) @@ -218,37 +218,9 @@ CREATE TABLE "persistent"."public"."revenue_2026" AS SELECT region, SUM(amount) FROM scratch_orders GROUP BY region; ``` -The `_table_catalog` (which tracks MCP-managed metadata for your tables) lives in the persistent attachment automatically — there's nothing to manage. If you want a pristine `.hyper` file for export with no MCP bookkeeping, run `DROP TABLE "persistent"."public"."_table_catalog"` once and subsequent sessions opening that file will leave it dropped. +**Per-database `_table_catalog`:** every writable database — persistent and any user-attached writable file — gets its own `_table_catalog` lazily seeded on first ingest. MCP-managed metadata (load tool, params, timestamps, prose fields set via `set_table_metadata`) lives alongside the data file, so opening a `.hyper` file later as a primary workspace finds the catalog ready. If you want a pristine `.hyper` file for export with no MCP bookkeeping, run `DROP TABLE ""."public"."_table_catalog"` once and subsequent sessions opening that file will leave it dropped. -**v1 limitations:** `load_files` and `watch_directory` use a connection pool bound to the primary database and don't yet accept `database`/`persist` — use `load_file` with `persist: true` for one-off persistent ingests. Merge mode (`load_file` with `mode: "merge"`) only works against the primary database in v1. - -### Daemon management - -The daemon is normally invisible — it auto-spawns and idle-times-out on its own. For diagnostics: - -```bash -hyperdb-mcp daemon status # Show running daemon (PID, endpoint, started_at, version) -hyperdb-mcp daemon stop # Gracefully shut down the daemon -hyperdb-mcp daemon # Run as a daemon explicitly (rarely needed) -``` - -State files live at `~/.hyperdb/` by default (override with `HYPERDB_STATE_DIR`). - -### Recovery from hyperd crashes - -The daemon polls `hyperd` every 5 seconds. If the process has exited (crashed, OOM, killed), the daemon spawns a replacement, atomically updates `~/.hyperdb/daemon.json` with the new endpoint, and continues serving clients. Clients see one failed tool call (the request that was in flight when hyperd died); the next tool call transparently reconnects to the new hyperd via the same recovery path used for normal connection drops. - -If a client itself notices hyperd is unreachable before the next polling tick, it sends a fast-path `REPORT_HYPERD_ERROR` signal to the daemon so the restart kicks off without waiting for the timer. - -If hyperd repeatedly fails to start (3 attempts within 60 seconds — e.g., misconfigured `HYPERD_PATH`, port exhaustion, broken binary), the daemon shuts itself down and removes the discovery file. The next MCP client to start up will then spawn a fresh daemon, surfacing any persistent failure clearly to the user rather than spinning silently. - -**Known limitation:** if hyperd hangs (alive at the OS level but unresponsive to queries), the daemon's polling can't detect it and your tool call may stall indefinitely. The recovery path is `hyperdb-mcp daemon stop` followed by reconnecting from your MCP client. - -### Other behavioral flags - -| Flag | Behavior | -|---|---| -| `--read-only` | Disables `execute`, `load_data`, `load_file`, `watch_directory`, `save_query`, `delete_query`, and Hyper-format export. See [Read-Only Mode](#read-only-mode). | +**Detach safety:** `detach_database` rejects with `InvalidArgument` if any active watcher targets the alias — call `unwatch_directory` first. This prevents the watcher's pool from silently writing into a now-detached file (or worse, the wrong file if the alias is later re-attached to a different path). ### Daemon management diff --git a/hyperdb-mcp/src/attach.rs b/hyperdb-mcp/src/attach.rs index 8ca615e..d048c32 100644 --- a/hyperdb-mcp/src/attach.rs +++ b/hyperdb-mcp/src/attach.rs @@ -366,7 +366,7 @@ impl AttachRegistry { // DB" which implicitly promises a catalog; leaving an // attached-but-unseeded file would silently violate that. if file_was_created { - if let Err(e) = crate::table_catalog::ensure_exists_in_database(engine, &req.alias) { + if let Err(e) = crate::table_catalog::ensure_exists_in(engine, Some(&req.alias)) { let detach_sql = format!("DETACH DATABASE \"{}\"", req.alias.replace('"', "\"\"")); if let Err(de) = engine.execute_command(&detach_sql) { tracing::warn!( diff --git a/hyperdb-mcp/src/engine.rs b/hyperdb-mcp/src/engine.rs index 882c61f..63de8f8 100644 --- a/hyperdb-mcp/src/engine.rs +++ b/hyperdb-mcp/src/engine.rs @@ -195,17 +195,19 @@ pub struct Engine { /// to `false` after the server consumes it via /// [`Self::take_persistent_was_created`]. persistent_was_created: bool, - /// Cached "_table_catalog exists in persistent" probe. Populated on - /// first call to [`Self::catalog_present_in_persistent`] and held - /// for the lifetime of the connection. Avoids repeated - /// `pg_catalog.pg_tables` round-trips on every catalog read/write. + /// Cached "_table_catalog exists in ``" probes, keyed by + /// canonical alias (lowercase). Populated on first call to + /// [`Self::catalog_present_in`] for each `(engine, alias)` pair. + /// /// Lives on the Engine because the catalog is per-engine-lifetime /// (a `ConnectionLost` reconnect creates a fresh Engine, so the - /// cache resets at the right boundary). `None` means "not yet - /// probed"; `Some(false)` is cacheable too — once the catalog is - /// confirmed absent on a read-only or `--ephemeral-only` flow it - /// stays absent for the whole engine lifetime. - catalog_present_cache: std::sync::Mutex>, + /// cache resets at the right boundary). Detaching an alias clears + /// its entry via [`Self::clear_catalog_cache_for`] so a re-attach + /// to a different file/writability doesn't reuse a stale value. + /// `Some(false)` is cacheable too — once the catalog is confirmed + /// absent it stays absent for the rest of the engine's lifetime + /// unless explicitly cleared. + catalog_present_cache: std::sync::Mutex>, log_dir: PathBuf, } @@ -339,7 +341,7 @@ impl Engine { ephemeral_path, persistent_path, persistent_was_created, - catalog_present_cache: std::sync::Mutex::new(None), + catalog_present_cache: std::sync::Mutex::new(std::collections::HashMap::new()), log_dir, }) } @@ -414,7 +416,7 @@ impl Engine { ephemeral_path: ephemeral_path.to_path_buf(), persistent_path, persistent_was_created, - catalog_present_cache: std::sync::Mutex::new(None), + catalog_present_cache: std::sync::Mutex::new(std::collections::HashMap::new()), log_dir: log_dir.to_path_buf(), })) } @@ -596,48 +598,56 @@ impl Engine { self.persistent_was_created } - /// Returns whether `_table_catalog` exists in the persistent - /// attachment, caching the result on first call so subsequent - /// catalog read/write paths skip the `pg_catalog.pg_tables` probe. + /// Returns whether `_table_catalog` exists in `alias`, caching + /// the per-DB result on first call so subsequent catalog read/ + /// write paths skip the `pg_catalog.pg_tables` probe. /// /// `prober` is the SQL-side existence check; the cache layer here /// is intentionally generic so the catalog module can keep its - /// existing probe SQL in one place. - /// - /// Returns `Ok(false)` immediately when the engine has no - /// persistent attachment (no probe needed). + /// probe SQL in one place. /// /// # Errors /// Propagates whatever error `prober` returns on the first call. /// On subsequent calls, the cached value is returned without /// re-running the probe. - pub fn catalog_present_in_persistent(&self, prober: F) -> Result + pub fn catalog_present_in(&self, alias: &str, prober: F) -> Result where F: Fn(&Engine) -> Result, { - if !self.has_persistent() { - return Ok(false); - } + let key = alias.to_ascii_lowercase(); // Fast path: cache already populated. if let Ok(guard) = self.catalog_present_cache.lock() { - if let Some(present) = *guard { + if let Some(&present) = guard.get(&key) { return Ok(present); } } // Slow path: run the probe and cache its result. let present = prober(self)?; if let Ok(mut guard) = self.catalog_present_cache.lock() { - *guard = Some(present); + guard.insert(key, present); } Ok(present) } - /// Synchronously set the catalog-presence cache to `true` — used by - /// `table_catalog::ensure_exists` after a `CREATE TABLE IF NOT - /// EXISTS` so subsequent reads/writes skip the existence probe. - pub fn mark_catalog_present(&self) { + /// Synchronously set the catalog-presence cache to `true` for + /// `alias` — used by `table_catalog::ensure_exists_in` after a + /// successful `CREATE TABLE IF NOT EXISTS` so subsequent reads/ + /// writes against that DB skip the existence probe. + pub fn mark_catalog_present_for(&self, alias: &str) { + let key = alias.to_ascii_lowercase(); if let Ok(mut guard) = self.catalog_present_cache.lock() { - *guard = Some(true); + guard.insert(key, true); + } + } + + /// Drop the cached probe result for `alias`. Called by + /// `detach_database` so that re-attaching the same alias to a + /// different file (or with different writability) doesn't reuse a + /// stale entry. + pub fn clear_catalog_cache_for(&self, alias: &str) { + let key = alias.to_ascii_lowercase(); + if let Ok(mut guard) = self.catalog_present_cache.lock() { + guard.remove(&key); } } @@ -926,6 +936,52 @@ impl Engine { .collect()) } + /// Like [`Self::column_metadata`] but for a table in `target_db`. + /// `None` falls back to `column_metadata` (primary). `Some(alias)` + /// reads via the qualified `pg_catalog.pg_attribute` join used by + /// `describe_columns_via_pg_catalog` — the connection-bound + /// `Catalog` API can't see attached databases. + /// + /// # Errors + /// + /// Returns [`ErrorCode::TableNotFound`] when no rows come back from + /// the qualified probe. Propagates connection errors. + pub fn column_metadata_in( + &self, + target_db: Option<&str>, + table: &str, + ) -> Result, McpError> { + let Some(db) = target_db else { + return self.column_metadata(table); + }; + let rows = describe_columns_via_pg_catalog(self, db, table)?; + if rows.is_empty() { + return Err(McpError::new( + ErrorCode::TableNotFound, + format!("Table '{table}' does not exist in database '{db}'"), + )); + } + Ok(rows + .into_iter() + .map(|r| ColumnSchema { + name: r + .get("name") + .and_then(|v| v.as_str()) + .unwrap_or_default() + .to_string(), + hyper_type: r + .get("type") + .and_then(|v| v.as_str()) + .unwrap_or_default() + .to_string(), + nullable: r + .get("nullable") + .and_then(serde_json::Value::as_bool) + .unwrap_or(true), + }) + .collect()) + } + /// Returns true if `table` exists in the `public` schema. Avoids /// the per-version error-string ambiguity of /// [`Catalog::get_table_definition`] by listing names instead. @@ -940,6 +996,29 @@ impl Engine { Ok(names.iter().any(|n| n.as_str() == table)) } + /// Like [`Self::table_exists`] but for a table in `target_db`. + /// `None` falls back to `table_exists` (primary). `Some(alias)` + /// probes the qualified `pg_catalog.pg_tables` of the attached + /// database — the connection-bound `Catalog` API can't see + /// attached databases. + /// + /// # Errors + /// + /// Propagates connection errors from the probe query. + pub fn table_exists_in(&self, target_db: Option<&str>, table: &str) -> Result { + let Some(db) = target_db else { + return self.table_exists(table); + }; + let esc_db = db.replace('"', "\"\""); + let esc_tbl = table.replace('\'', "''"); + let sql = format!( + "SELECT 1 AS one FROM \"{esc_db}\".pg_catalog.pg_tables \ + WHERE schemaname = 'public' AND tablename = '{esc_tbl}'" + ); + let rows = self.execute_query_to_json(&sql)?; + Ok(!rows.is_empty()) + } + /// Issue a single `ALTER TABLE "" ADD COLUMN "" , /// ADD COLUMN "" , …` statement that adds all columns /// atomically. Hyper supports the multi-column form (verified @@ -968,6 +1047,23 @@ impl Engine { &self, table: &str, cols: &[ColumnSchema], + ) -> Result<(), McpError> { + self.alter_table_add_columns_in(None, table, cols) + } + + /// Like [`Self::alter_table_add_columns`] but for a table in + /// `target_db`. `None` keeps the unqualified identifier; `Some(alias)` + /// emits `"db"."public"."table"` so the ALTER lands in the attached + /// database. + /// + /// # Errors + /// + /// Same as [`Self::alter_table_add_columns`]. + pub fn alter_table_add_columns_in( + &self, + target_db: Option<&str>, + table: &str, + cols: &[ColumnSchema], ) -> Result<(), McpError> { if cols.is_empty() { return Ok(()); @@ -983,7 +1079,14 @@ impl Engine { )); } } - let quoted_table = format!("\"{}\"", table.replace('"', "\"\"")); + let quoted_table = match target_db { + Some(db) => { + let esc_db = db.replace('"', "\"\""); + let esc_tbl = table.replace('"', "\"\""); + format!("\"{esc_db}\".\"public\".\"{esc_tbl}\"") + } + None => format!("\"{}\"", table.replace('"', "\"\"")), + }; let add_clauses = cols .iter() .map(|c| { diff --git a/hyperdb-mcp/src/export.rs b/hyperdb-mcp/src/export.rs index ec144f9..05ef67e 100644 --- a/hyperdb-mcp/src/export.rs +++ b/hyperdb-mcp/src/export.rs @@ -61,6 +61,12 @@ pub struct ExportOptions { /// may be strings, booleans, or numbers; anything else is rejected. /// Ignored for `"hyper"` format (which is not a `COPY` at all). pub format_options: Option>, + /// Source database to snapshot when `format = "hyper"`. `None` → + /// the primary (ephemeral) workspace; `Some("persistent")` or + /// `Some("user_alias")` snapshots that attached database. Ignored + /// for the row-oriented formats which already qualify via `sql` + /// or via `scoped_search_path`. + pub source_db: Option, } /// Returned by [`export_to_file`] with the exported row count and telemetry. @@ -119,7 +125,7 @@ pub fn export_to_file(engine: &Engine, opts: &ExportOptions) -> Result std::io::Result { /// reproduced. That's acceptable for the current callers (LLMs /// exporting workspace data for Tableau Desktop), but documented here /// so a future caller that needs full catalog fidelity knows why. -fn export_hyper(engine: &Engine, path: &str, timer: &StatsTimer) -> Result { +fn export_hyper( + engine: &Engine, + path: &str, + source_db: Option<&str>, + timer: &StatsTimer, +) -> Result { // The target path is a separate file from the primary workspace, // so OS-level copy/delete on it is fine — the lock conflict only // affects the workspace hyperd has open. Pre-delete on overwrite @@ -465,7 +476,7 @@ fn export_hyper(engine: &Engine, path: &str, timer: &StatsTimer) -> Result Result Result { - let escaped_alias = alias.replace('"', "\"\""); - let primary = engine.primary_db_name(); - let escaped_primary = primary.replace('"', "\"\""); - - let schemas = list_user_schemas(engine, &escaped_primary)?; +/// Copy every user table from `source_db` (None → primary) into the +/// database attached as `target_alias`. Returns the total row count +/// written. Excludes `pg_catalog` / `information_schema` (and Hyper's +/// own system schemas) so we only touch user data. +fn populate_export_target( + engine: &Engine, + source_db: Option<&str>, + target_alias: &str, +) -> Result { + let escaped_alias = target_alias.replace('"', "\"\""); + let source = source_db.map_or_else(|| engine.primary_db_name(), str::to_string); + let escaped_source = source.replace('"', "\"\""); + + let schemas = list_user_schemas(engine, &escaped_source)?; let mut total_rows: u64 = 0; for schema in &schemas { @@ -521,7 +536,7 @@ fn populate_export_target(engine: &Engine, alias: &str) -> Result ))?; } - let tables = list_user_tables(engine, &escaped_primary, schema)?; + let tables = list_user_tables(engine, &escaped_source, schema)?; for table in &tables { if crate::engine::is_internal_table(table) { continue; @@ -529,7 +544,7 @@ fn populate_export_target(engine: &Engine, alias: &str) -> Result let escaped_table = table.replace('"', "\"\""); let rows_copied = engine.execute_command(&format!( "CREATE TABLE \"{escaped_alias}\".\"{escaped_schema}\".\"{escaped_table}\" AS \ - SELECT * FROM \"{escaped_primary}\".\"{escaped_schema}\".\"{escaped_table}\"", + SELECT * FROM \"{escaped_source}\".\"{escaped_schema}\".\"{escaped_table}\"", ))?; total_rows = total_rows.saturating_add(rows_copied); } diff --git a/hyperdb-mcp/src/ingest.rs b/hyperdb-mcp/src/ingest.rs index e53a8e2..b42b23f 100644 --- a/hyperdb-mcp/src/ingest.rs +++ b/hyperdb-mcp/src/ingest.rs @@ -171,14 +171,19 @@ pub struct IngestResult { struct TempTableGuard<'a> { engine: &'a Engine, name: String, + /// Target database the temp lives in. `None` → primary (unqualified + /// drop). `Some(alias)` → emit `"alias"."public".""` so the + /// drop lands in the same DB the temp was created in. + target_db: Option, armed: bool, } impl<'a> TempTableGuard<'a> { - fn new(engine: &'a Engine, name: String) -> Self { + fn new(engine: &'a Engine, name: String, target_db: Option) -> Self { Self { engine, name, + target_db, armed: true, } } @@ -198,10 +203,15 @@ impl Drop for TempTableGuard<'_> { // have learned about the temp leak otherwise" — the latter is // the load-bearing case for this guard's existence. let panicking = std::thread::panicking(); - let drop_sql = format!( - "DROP TABLE IF EXISTS \"{}\"", - self.name.replace('"', "\"\"") - ); + let quoted = match &self.target_db { + Some(db) => format!( + "\"{}\".\"public\".\"{}\"", + db.replace('"', "\"\""), + self.name.replace('"', "\"\""), + ), + None => format!("\"{}\"", self.name.replace('"', "\"\"")), + }; + let drop_sql = format!("DROP TABLE IF EXISTS {quoted}"); match self.engine.execute_command(&drop_sql) { Ok(_) => { if panicking { @@ -351,25 +361,32 @@ where std::process::id(), ); + // The temp table lives in the same database as the target. That + // way every DML statement below (CREATE TABLE AS, DELETE-USING, + // INSERT-SELECT) stays within a single DB — no cross-DB DML — and + // the merge works the same regardless of whether the target is + // primary, persistent, or any user-attached writable database. let tmp_opts = IngestOptions { table: tmp.clone(), mode: "replace".into(), schema_override: opts.schema_override.clone(), merge_key: None, - target_db: None, + target_db: opts.target_db.clone(), }; let tmp_result = replace_load(&tmp_opts)?; // Arm the cleanup guard immediately after the load so any later - // failure (or panic) drops the temp table on unwind. - let mut guard = TempTableGuard::new(engine, tmp.clone()); + // failure (or panic) drops the temp table on unwind. The guard + // tracks `target_db` so the DROP lands in the same DB the temp + // was created in. + let mut guard = TempTableGuard::new(engine, tmp.clone(), opts.target_db.clone()); // Belt-and-suspenders check: a per-format `replace_load` that returns // `Ok` without actually creating the table would surface as opaque // "table not found" errors from the catalog read in step 4. Catch // it here with a clear message so the contract violation is named // outright. - if !engine.table_exists(&tmp)? { + if !engine.table_exists_in(opts.target_db.as_deref(), &tmp)? { return Err(McpError::new( ErrorCode::InternalError, format!( @@ -380,10 +397,21 @@ where } // Step 3 — if target doesn't exist, rename temp → target. - if !engine.table_exists(&opts.table)? { - let quoted_tmp = format!("\"{}\"", tmp.replace('"', "\"\"")); - let quoted_tgt = format!("\"{}\"", opts.table.replace('"', "\"\"")); - engine.execute_command(&format!("ALTER TABLE {quoted_tmp} RENAME TO {quoted_tgt}"))?; + if !engine.table_exists_in(opts.target_db.as_deref(), &opts.table)? { + let qualified_tmp_opts = IngestOptions { + table: tmp.clone(), + mode: "replace".into(), + schema_override: None, + merge_key: None, + target_db: opts.target_db.clone(), + }; + let quoted_tmp = qualified_table(&qualified_tmp_opts); + // RENAME TO accepts an unqualified new name (Hyper / PostgreSQL + // semantics — the schema/database stays the same as the source). + let escaped_new = opts.table.replace('"', "\"\""); + engine.execute_command(&format!( + "ALTER TABLE {quoted_tmp} RENAME TO \"{escaped_new}\"" + ))?; // The temp is now the target; the guard must NOT try to drop it. guard.disarm(); return Ok(IngestResult { @@ -407,9 +435,13 @@ where }); } - // Step 4 — schema reconciliation. - let target_cols = engine.column_metadata(&opts.table)?; - let tmp_cols = engine.column_metadata(&tmp)?; + // Step 4 — schema reconciliation. Read both schemas from the + // target DB; in cross-DB merges the connection-bound `Catalog` + // wouldn't see the attached database, so route via + // `column_metadata_in` which falls back to the qualified + // `pg_catalog.pg_attribute` probe. + let target_cols = engine.column_metadata_in(opts.target_db.as_deref(), &opts.table)?; + let tmp_cols = engine.column_metadata_in(opts.target_db.as_deref(), &tmp)?; // Every key must exist in both, with matching type. for k in keys { let in_target = target_cols.iter().find(|c| c.name == *k); @@ -476,11 +508,20 @@ where }) .collect(); let schema_changed = !new_cols.is_empty(); - engine.alter_table_add_columns(&opts.table, &new_cols)?; + engine.alter_table_add_columns_in(opts.target_db.as_deref(), &opts.table, &new_cols)?; // Step 6 — DELETE matching rows by key, then INSERT all temp rows. - let quoted_tgt = format!("\"{}\"", opts.table.replace('"', "\"\"")); - let quoted_tmp = format!("\"{}\"", tmp.replace('"', "\"\"")); + // Both target and temp share `opts.target_db`, so qualifying via + // `qualified_table` keeps the DML scoped to one DB. + let quoted_tgt = qualified_table(opts); + let qualified_tmp_opts = IngestOptions { + table: tmp.clone(), + mode: "replace".into(), + schema_override: None, + merge_key: None, + target_db: opts.target_db.clone(), + }; + let quoted_tmp = qualified_table(&qualified_tmp_opts); let key_eq = keys .iter() .map(|k| { @@ -507,7 +548,7 @@ where let inserted = engine.execute_command(&insert_sql)?; // Re-read target schema for the result so callers see the post-merge shape. - let final_schema = engine.column_metadata(&opts.table)?; + let final_schema = engine.column_metadata_in(opts.target_db.as_deref(), &opts.table)?; // The guard drops the temp table when it falls out of scope here. Ok(IngestResult { diff --git a/hyperdb-mcp/src/readme.rs b/hyperdb-mcp/src/readme.rs index c47fe5f..dbdf0e9 100644 --- a/hyperdb-mcp/src/readme.rs +++ b/hyperdb-mcp/src/readme.rs @@ -27,6 +27,42 @@ or shell pipelines: it parses files faster, runs SQL natively, and keeps intermediate state in a workspace database the LLM can re-query without re-loading. +## Workspace model + +Every session has TWO databases, plus optional user-attached ones: + +- **Ephemeral primary** (default destination). Created fresh per + session, deleted on exit. Unqualified SQL routes here. Use for + scratch work and intermediate transformations the user doesn't + need to keep. +- **Persistent attachment** under alias `\"persistent\"`. Survives + across sessions at the platform-default path (or override via + `--persistent-db`). Use when the user wants the data to stick + around. Disabled when the server runs with `--ephemeral-only`. +- **User-attached writable databases** via `attach_database` with + `writable: true`. Each lives in its own `.hyper` file under a + user-chosen alias. + +Pick a destination in one of two ways: + +- **`database` parameter** (preferred for tools that build their own + SQL): `query`, `execute`, `load_data`, `load_file`, `load_files`, + `watch_directory`, `describe`, `sample`, `chart`, `export`, and + `set_table_metadata` accept `database: \"persistent\"`, + `database: \"local\"` (= primary), or any user-attached writable + alias. Case-insensitive. Defaults to primary. +- **`persist: true` shorthand** on `load_data`, `load_file`, + `load_files`, `watch_directory` — equivalent to + `database: \"persistent\"`. +- **Fully-qualified SQL** for power users: + `INSERT INTO \"persistent\".\"public\".\"customers\" SELECT ...` + +Each writable database carries its own `_table_catalog` table that +tracks load tool, params, timestamps, and any prose metadata set via +`set_table_metadata` — lazily seeded on first ingest into that DB. +`detach_database` rejects with `InvalidArgument` if any active +watcher targets the alias; call `unwatch_directory` first. + ## Tool index ### Query diff --git a/hyperdb-mcp/src/server.rs b/hyperdb-mcp/src/server.rs index ee80c46..a04214c 100644 --- a/hyperdb-mcp/src/server.rs +++ b/hyperdb-mcp/src/server.rs @@ -197,8 +197,6 @@ pub struct LoadFileParams { /// Target database alias. Omit (or pass `"local"`) to write to the /// ephemeral primary. Pass `"persistent"` to write to the durable /// database. Other values target a user-attached writable database. - /// Note: `mode: "merge"` with a non-primary database is not yet - /// supported and will return an error. pub database: Option, /// Shorthand for `database: "persistent"`. If both `database` and /// `persist` are set, `database` wins. @@ -308,13 +306,14 @@ pub struct LoadFilesParams { /// hyperd's side; more connections don't help past a certain point /// and can starve the primary connection. pub concurrency: Option, - /// Target database alias. **Currently only the primary (ephemeral) - /// database is supported for parallel ingest.** Passing `"persistent"` - /// or another alias returns an error — use `load_file` with - /// `persist: true` instead for single-file persistent ingest. + /// Target database alias. Omit (or pass `"local"`) to write to the + /// ephemeral primary. Pass `"persistent"` to write to the durable + /// database. Other values target a user-attached writable database. + /// Applies to every entry in the batch — multi-target batches are + /// not supported. pub database: Option, - /// Shorthand for `database: "persistent"`. Same limitation as - /// `database` applies — not yet supported for parallel ingest. + /// Shorthand for `database: "persistent"`. If both `database` and + /// `persist` are set, `database` wins. pub persist: Option, } @@ -499,14 +498,17 @@ pub struct WatchDirectoryParams { /// Each in-flight ingest holds one connection to hyperd plus a transaction. #[serde(default)] pub max_concurrent: Option, - /// Target database alias. **Currently only the primary (ephemeral) - /// database is supported by the watcher** — the connection pool is - /// bound to the primary file. Passing `"persistent"` returns an - /// error. Use `load_file` with `persist: true` for one-off persistent - /// ingests. + /// Target database alias. Omit (or pass `"local"`) for the ephemeral + /// primary. Pass `"persistent"` for the durable database, or any + /// user-attached writable alias. The watcher's connection pool is + /// built against the resolved target, so subsequent ingests land + /// in the right database without per-file routing. + /// + /// Detaching the alias while a watcher is active is rejected — call + /// `unwatch_directory` first. pub database: Option, - /// Shorthand for `database: "persistent"`. Same limitation applies — - /// not yet supported by the watcher. + /// Shorthand for `database: "persistent"`. If both `database` and + /// `persist` are set, `database` wins. pub persist: Option, } @@ -747,6 +749,14 @@ pub struct SetTableMetadataParams { pub license: Option, /// Free-form notes: refresh instructions, known gotchas, caveats. pub notes: Option, + /// Target database alias for the catalog write. Omit (or pass + /// `"local"` / `"persistent"`) to update the persistent catalog — + /// matches the default for the ephemeral primary's tables. + /// Pass any user-attached writable alias to update that DB's + /// per-database `_table_catalog` instead. Read-only attachments + /// are rejected with a clear "re-attach with writable:true" + /// message. + pub database: Option, } // --- Prompt argument structs --- @@ -1111,11 +1121,17 @@ impl HyperMcpServer { /// swallows errors — a bookkeeping failure should never fail an /// otherwise-successful load. /// - /// `&self` is kept for consistency with sibling helpers and because - /// upcoming work (per-DB catalog presence cache) will need it. + /// Routes the upsert to `target_db`'s `_table_catalog`. The + /// catalog is lazily seeded if absent. `target_db = None` and + /// `target_db = Some("persistent")` both write to the persistent + /// catalog (the single-engine ephemeral primary stubs survive + /// there for the session). User-attached writable aliases get + /// their own per-DB catalog. Read-only attachments are rejected + /// upstream by `resolve_db(require_writable=true)` so this helper + /// never sees them. #[expect( clippy::unused_self, - reason = "kept for forthcoming per-DB catalog presence cache" + reason = "kept for symmetry with after_execute_catalog_update; both run inside with_engine" )] fn after_ingest_catalog_update( &self, @@ -1124,17 +1140,20 @@ impl HyperMcpServer { load_tool: &'static str, load_params: Option<&str>, row_count: Option, + target_db: Option<&str>, ) { - if let Err(e) = crate::table_catalog::upsert_stub( + if let Err(e) = crate::table_catalog::upsert_stub_in( engine, table_name, load_tool, load_params, row_count, true, + target_db, ) { tracing::warn!( table = %table_name, + target_db = ?target_db, err = %e.message, "failed to update _table_catalog after ingest" ); @@ -1145,7 +1164,7 @@ impl HyperMcpServer { /// error-swallowing rationale as [`Self::after_ingest_catalog_update`]. #[expect( clippy::unused_self, - reason = "kept for forthcoming per-DB catalog presence cache" + reason = "kept for symmetry; runs inside with_engine" )] fn after_execute_catalog_update(&self, engine: &Engine) { if let Err(e) = crate::table_catalog::reconcile(engine) { @@ -1465,9 +1484,12 @@ impl HyperMcpServer { }) .collect(); - // Catalog bookkeeping: only for primary or persistent targets. - // User-attached databases are not tracked in the catalog. - if target_db.is_none() || target_db.as_deref() == Some(Engine::PERSISTENT_ALIAS) { + // Catalog bookkeeping: the helper routes the upsert to + // target_db's per-DB _table_catalog (lazily seeded on + // first ingest). Persistent and ephemeral primary share + // persistent's catalog; user-attached writable DBs each + // get their own. + { let load_params = serde_json::to_string(&json!({ "mode": mode, "format": fmt, @@ -1480,6 +1502,7 @@ impl HyperMcpServer { "load_data", load_params.as_deref(), i64::try_from(ingest_result.rows).ok(), + target_db.as_deref(), ); } @@ -1532,16 +1555,6 @@ impl HyperMcpServer { let result = self.with_engine(|engine| { let target_db = self.resolve_db(engine, params.database.as_deref(), params.persist, true)?; - // Merge mode + non-primary database is not yet supported - // (cross-database DML for temp tables is unverified). - if mode == "merge" && target_db.is_some() { - return Err(McpError::new( - ErrorCode::InvalidArgument, - "merge mode with a non-primary database is not yet supported. \ - Use replace or append mode, or omit the database parameter." - .to_string(), - )); - } crate::attach::validate_input_path(¶ms.path, "data file")?; let schema_override = crate::schema::normalize_schema_param(params.schema.as_ref())?; let opts = IngestOptions { @@ -1594,8 +1607,8 @@ impl HyperMcpServer { }) .collect(); - // Catalog: only for primary or persistent targets. - if target_db.is_none() || target_db.as_deref() == Some(Engine::PERSISTENT_ALIAS) { + // Catalog: helper routes to target_db's per-DB catalog. + { let load_params = serde_json::to_string(&json!({ "source_path": params.path, "mode": mode, @@ -1611,6 +1624,7 @@ impl HyperMcpServer { "load_file", load_params.as_deref(), i64::try_from(ingest_result.rows).ok(), + target_db.as_deref(), ); } @@ -1647,7 +1661,7 @@ impl HyperMcpServer { /// Each entry behaves like a standalone `load_file` call; failures are /// reported per-file rather than aborting the whole batch. #[tool( - description = "Ingest multiple files in parallel. Each entry is equivalent to a standalone `load_file` call (same formats and same format-selection guidance: prefer Parquet > CSV > Arrow IPC > JSON for large imports). The batch runs across a pool of async connections sized by `concurrency` (default `min(files.len(), 8)`), so independent files finish roughly in max-time rather than sum-time. Per-file errors are captured in the response and do not abort the rest of the batch; the top-level call still returns Ok. For Apache Iceberg tables, call `load_iceberg` per table instead — this tool only handles single-file formats. **Note: `mode = \"merge\"` is not supported here — use `load_file` once per file when you need merge/upsert semantics.**" + description = "Ingest multiple files in parallel. Each entry is equivalent to a standalone `load_file` call (same formats and same format-selection guidance: prefer Parquet > CSV > Arrow IPC > JSON for large imports). The batch runs across a pool of async connections sized by `concurrency` (default `min(files.len(), 8)`), so independent files finish roughly in max-time rather than sum-time. Per-file errors are captured in the response and do not abort the rest of the batch; the top-level call still returns Ok. For Apache Iceberg tables, call `load_iceberg` per table instead — this tool only handles single-file formats.\n\nUse `database` (or shorthand `persist: true`) to target a non-primary database; the same value applies to every entry in the batch. **Note: `mode = \"merge\"` is not supported here — use `load_file` once per file when you need merge/upsert semantics.**" )] fn load_files( &self, @@ -1666,18 +1680,6 @@ impl HyperMcpServer { )); } - // Parallel ingest uses a connection pool that only sees the ephemeral - // primary — non-primary targets are not yet supported. - if params.database.is_some() || params.persist == Some(true) { - return Self::err_content(McpError::new( - ErrorCode::InvalidArgument, - "load_files does not yet support the `database` or `persist` parameter \ - because parallel ingest uses a connection pool bound to the primary database. \ - Use `load_file` with `persist: true` for single-file persistent ingest." - .to_string(), - )); - } - // Reject `mode = "merge"` (or stray `merge_key`) up front, before // we spin up the connection pool and dispatch the parallel batch. // The async ingest paths driven from this batch loader don't @@ -1702,16 +1704,39 @@ impl HyperMcpServer { } } - // Resolve hyperd endpoint + workspace once, under the engine lock. - // After this the pool operates independently of the engine's sync - // connection. - let (endpoint, workspace) = match self.with_engine(|engine| { + // Resolve hyperd endpoint + the workspace path matching the + // resolved target database. The pool opens that .hyper file + // directly (under the same alias the engine uses) so qualified + // SQL routes correctly. Read-only attachments are rejected by + // resolve_db. + let (endpoint, workspace, target_db) = match self.with_engine(|engine| { + let target_db = + self.resolve_db(engine, params.database.as_deref(), params.persist, true)?; let endpoint = engine.hyperd_endpoint()?; - // The pool connects to the engine's primary database (ephemeral). - // Tools that target the persistent attachment use qualified SQL - // through the same connection. - let workspace = engine.ephemeral_path().to_string_lossy().to_string(); - Ok((endpoint, workspace)) + let workspace = match target_db.as_deref() { + None => engine.ephemeral_path().to_string_lossy().to_string(), + Some(alias) if alias.eq_ignore_ascii_case(Engine::PERSISTENT_ALIAS) => engine + .persistent_path() + .ok_or_else(|| { + McpError::new( + ErrorCode::InvalidArgument, + "target 'persistent' but the server is in --ephemeral-only mode", + ) + })? + .to_string_lossy() + .to_string(), + Some(alias) => { + let entry = self.attachments.get(alias).ok_or_else(|| { + McpError::new( + ErrorCode::InvalidArgument, + format!("database '{alias}' is not attached"), + ) + })?; + let crate::attach::AttachSource::LocalFile { path } = &entry.source; + path.to_string_lossy().to_string() + } + }; + Ok((endpoint, workspace, target_db)) }) { Ok(v) => v, Err(e) => return Self::err_content(e), @@ -1764,6 +1789,7 @@ impl HyperMcpServer { let mut set = tokio::task::JoinSet::new(); for (idx, entry) in params.files.into_iter().enumerate() { let pool = Arc::clone(&pool); + let entry_target_db = target_db.clone(); set.spawn(async move { let mode = entry.mode.clone().unwrap_or_else(|| "replace".into()); let replace_mode = mode == "replace"; @@ -1785,6 +1811,15 @@ impl HyperMcpServer { return (idx, out); } }; + // The pool was built against the resolved target's + // .hyper file as its workspace, so from these + // connections' perspective the target IS the primary + // database. Keep target_db unqualified (None) so SQL + // routes into the pool's primary instead of trying + // to qualify against an alias that doesn't exist on + // these connections. The `entry_target_db` is still + // used downstream for the catalog gate. + let _ = &entry_target_db; let opts = IngestOptions { table: entry.table.clone(), mode: mode.clone(), @@ -1945,7 +1980,8 @@ impl HyperMcpServer { .collect(); // Update the per-table catalog stubs for every success. Requires - // the engine, so we run this inside `with_engine`. + // the engine, so we run this inside `with_engine`. The helper + // routes the upsert to target_db's per-DB _table_catalog. if let Err(e) = self.with_engine(|engine| { for o in &outcomes { if let Some((rows, _, _)) = &o.ok { @@ -1955,6 +1991,7 @@ impl HyperMcpServer { "load_file", None, i64::try_from(*rows).ok(), + target_db.as_deref(), ); } } @@ -2037,6 +2074,7 @@ impl HyperMcpServer { "load_iceberg", load_params.as_deref(), i64::try_from(ingest_result.rows).ok(), + None, ); Ok(json!({ @@ -2351,7 +2389,7 @@ impl HyperMcpServer { /// Begin watching a directory for `.ready` sentinel files. See /// [`crate::watcher`] for the full producer/consumer protocol. #[tool( - description = "Watch a directory for files to auto-ingest. Producers write data file + companion .ready sentinel; the watcher appends the data file to the given table and deletes both on success. Disabled in read-only mode." + description = "Watch a directory for files to auto-ingest. Producers write data file + companion .ready sentinel; the watcher appends the data file to the given table and deletes both on success. Use `database` (or shorthand `persist: true`) to target a non-primary database — the watcher's connection pool opens that file directly. `detach_database` rejects while a watcher is active; call `unwatch_directory` first. Disabled in read-only mode." )] fn watch_directory( &self, @@ -2360,17 +2398,6 @@ impl HyperMcpServer { if let Err(e) = self.check_writable("watch_directory") { return Self::err_content(e); } - // Watcher uses a connection pool bound to the primary database; - // non-primary targets are not yet supported. - if params.database.is_some() || params.persist == Some(true) { - return Self::err_content(McpError::new( - ErrorCode::InvalidArgument, - "watch_directory does not yet support `database` or `persist`. \ - The watcher's connection pool is bound to the primary database. \ - Use `load_file` with `persist: true` for one-off persistent ingests." - .to_string(), - )); - } let canonical = match crate::attach::validate_input_path(¶ms.path, "watch directory") { Ok(p) => p, Err(e) => return Self::err_content(e), @@ -2382,18 +2409,31 @@ impl HyperMcpServer { Err(e) => return Self::err_content(e), } + // Resolve the target database once, under the engine lock. Read-only + // attachments are rejected here (require_writable=true) so the + // watcher can't be pointed at a destination it can't write to. + let target_db = match self.with_engine(|engine| { + self.resolve_db(engine, params.database.as_deref(), params.persist, true) + }) { + Ok(v) => v, + Err(e) => return Self::err_content(e), + }; + let path = canonical; let engine_handle = self.engine_handle(); + let attachments = self.attachments_handle(); let registry = self.watchers_handle(); let options = crate::watcher::WatchOptions { max_concurrent: params.max_concurrent.unwrap_or(0) as usize, }; let result = crate::watcher::start_watching( engine_handle, + attachments, registry, Some(self.subscriptions_handle()), path.clone(), params.table.clone(), + target_db, options, ); match result { @@ -2498,7 +2538,7 @@ impl HyperMcpServer { /// Export query results or a table to CSV, Parquet, Arrow IPC, /// Apache Iceberg, or a new `.hyper` file. #[tool( - description = "Export query results or a table to a file via hyperd's native writers. Every format listed here is server-side — hyperd writes the file directly, with zero per-row work in the MCP process — and every format round-trips cleanly through the matching loader (`load_file` or `load_iceberg`).\n\nWhen choosing a format for *data leaving* Hyper, prefer in this order:\n 1. **Parquet** (recommended default): smallest output, fastest write, preserves every type (NUMERIC precision/scale, DATE, TIMESTAMP, etc.). `path` is a single file.\n 2. **Iceberg**: produces a full Apache Iceberg table directory (`metadata/` + `data/`). Use when the consumer is a data-lake tool (Spark, Trino, DuckDB, etc.). `path` is a directory that hyperd creates.\n 3. **Arrow IPC Stream** (`arrow_ipc`): same wire shape Hyper uses internally; great for handing data to another Arrow-aware process. Larger than Parquet (no compression) but extremely fast to read back. `path` is a single file.\n 4. **CSV**: portable and human-readable but the largest output and types are lost (everything becomes text). Use for spreadsheet / shell-pipeline interop. Includes header row.\n 5. **Hyper**: an entire `.hyper` database file openable directly in Tableau Desktop. `sql`/`table` are ignored — every user table is copied.\n\nAll formats except Iceberg and Hyper require either `sql` or `table`. Iceberg output is a directory; all others are single files." + description = "Export query results or a table to a file via hyperd's native writers. Every format listed here is server-side — hyperd writes the file directly, with zero per-row work in the MCP process — and every format round-trips cleanly through the matching loader (`load_file` or `load_iceberg`).\n\nWhen choosing a format for *data leaving* Hyper, prefer in this order:\n 1. **Parquet** (recommended default): smallest output, fastest write, preserves every type (NUMERIC precision/scale, DATE, TIMESTAMP, etc.). `path` is a single file.\n 2. **Iceberg**: produces a full Apache Iceberg table directory (`metadata/` + `data/`). Use when the consumer is a data-lake tool (Spark, Trino, DuckDB, etc.). `path` is a directory that hyperd creates.\n 3. **Arrow IPC Stream** (`arrow_ipc`): same wire shape Hyper uses internally; great for handing data to another Arrow-aware process. Larger than Parquet (no compression) but extremely fast to read back. `path` is a single file.\n 4. **CSV**: portable and human-readable but the largest output and types are lost (everything becomes text). Use for spreadsheet / shell-pipeline interop. Includes header row.\n 5. **Hyper**: an entire `.hyper` database file openable directly in Tableau Desktop. `sql`/`table` are ignored — every user table is copied.\n\nAll formats except Iceberg and Hyper require either `sql` or `table`. Iceberg output is a directory; all others are single files.\n\nUse `database` to read from a non-primary source: for `format=\"hyper\"` it selects which database is snapshotted; for the row-oriented formats it routes the SELECT through the named database (when `table` is set) or pins `schema_search_path` for the call (when `sql` is set)." )] fn export( &self, @@ -2521,26 +2561,17 @@ impl HyperMcpServer { )); } }; - // Database routing. Two strategies: + // Database routing. Three strategies: + // - `hyper` format + non-primary: source_db plumbed through + // into populate_export_target so the snapshot reads from + // the requested database (no need to redirect anything; + // the cross-DB CREATE TABLE AS handles it natively). // - `table` mode + non-primary: synthesize a fully-qualified // SELECT and pass it as `sql` so export.rs's name-quoting // doesn't double-quote our identifier. // - `sql` mode + non-primary: redirect search_path for the // call duration so unqualified names resolve correctly. - // - `hyper` format always exports the *primary* database - // (export_hyper takes a snapshot of the connection's - // workspace); database+hyper would silently produce the - // wrong file, so we reject it up front. let target_db = self.resolve_db(engine, params.database.as_deref(), None, false)?; - if params.format == "hyper" && target_db.is_some() { - return Err(McpError::new( - ErrorCode::InvalidArgument, - "export with format=\"hyper\" always snapshots the primary (ephemeral) \ - database. Drop the `database` parameter, or pick a different format \ - (csv, parquet, arrow_ipc, iceberg) that supports cross-database export." - .to_string(), - )); - } let (effective_sql, effective_table) = match (¶ms.sql, ¶ms.table, &target_db) { (None, Some(t), Some(db)) => { let esc_db = db.replace('"', "\"\""); @@ -2567,6 +2598,7 @@ impl HyperMcpServer { format: params.format, overwrite: params.overwrite.unwrap_or(true), format_options, + source_db: target_db.clone(), }; let export_result = export_to_file(engine, &opts)?; Ok(json!({ @@ -2679,7 +2711,7 @@ impl HyperMcpServer { /// Update prose metadata for a table in the `_table_catalog`. #[tool( - description = "Update prose metadata for a table in the `_table_catalog`: source_url, source_description, purpose, license, notes. Fields you omit stay unchanged; pass an explicit empty string (\"\") to clear a field. Mechanical fields (load_tool, load_params, loaded_at, last_refreshed_at, row_count) are managed by the server. Requires an existing catalog entry — load the table first (load_file / load_data / execute CREATE TABLE) so the stub row is created automatically. Disabled in read-only mode and when the catalog table doesn't exist on the target database." + description = "Update prose metadata for a table in the `_table_catalog`: source_url, source_description, purpose, license, notes. Fields you omit stay unchanged; pass an explicit empty string (\"\") to clear a field. Mechanical fields (load_tool, load_params, loaded_at, last_refreshed_at, row_count) are managed by the server. Requires an existing catalog entry — load the table first (load_file / load_data / execute CREATE TABLE) so the stub row is created automatically. Use `database` to target the metadata for a table in a non-primary writable database; read-only attachments are rejected with a clear re-attach-with-writable message. Disabled in read-only mode." )] fn set_table_metadata( &self, @@ -2696,8 +2728,20 @@ impl HyperMcpServer { notes: params.notes, }; let table_name = params.table.clone(); - let result = self - .with_engine(|engine| crate::table_catalog::set_metadata(engine, &table_name, &fields)); + let result = self.with_engine(|engine| { + // Resolve target with require_writable=true so read-only + // attachments are rejected BEFORE any catalog write + // (defense-in-depth: ensure_exists_in's CREATE TABLE + // would also fail at the Hyper layer, but the resolve_db + // error is more actionable). + let target_db = self.resolve_db(engine, params.database.as_deref(), None, true)?; + crate::table_catalog::set_metadata_in( + engine, + &table_name, + &fields, + target_db.as_deref(), + ) + }); match result { Ok(entry) => Self::ok_content(entry.to_json()), Err(e) => Self::err_content(e), @@ -2842,8 +2886,40 @@ impl HyperMcpServer { Parameters(params): Parameters, ) -> Result { let alias = params.alias.clone(); + // Reject if any active watcher targets this alias. Otherwise the + // watcher's pool would keep ingesting into the now-detached + // workspace path; or, if the user re-attached the same alias to + // a different file, into the wrong database. Fixed by stopping + // the watcher first via `unwatch_directory`. + if let Ok(watchers) = self.watchers.watchers.lock() { + let conflict = watchers.values().find(|h| { + h.target_db + .as_deref() + .is_some_and(|a| a.eq_ignore_ascii_case(&alias)) + }); + if let Some(h) = conflict { + return Self::err_content(McpError::new( + ErrorCode::InvalidArgument, + format!( + "cannot detach '{alias}': an active watcher on directory '{}' targets it. \ + Call unwatch_directory(\"{}\") first.", + h.directory.display(), + h.directory.display() + ), + )); + } + } let registry = self.attachments_handle(); - let result = self.with_engine(|engine| registry.detach(engine, &alias)); + let result = self.with_engine(|engine| { + let outcome = registry.detach(engine, &alias)?; + if outcome { + // Drop any cached "_table_catalog exists in this alias" + // probe so a re-attach to a different file or with + // different writability won't reuse a stale entry. + engine.clear_catalog_cache_for(&alias); + } + Ok(outcome) + }); match result { Ok(detached) => { Self::ok_content(json!({ "alias": params.alias, "detached": detached })) @@ -3020,6 +3096,7 @@ impl HyperMcpServer { "copy_query", load_params.as_deref(), row_count, + target_db, ); } diff --git a/hyperdb-mcp/src/table_catalog.rs b/hyperdb-mcp/src/table_catalog.rs index b09e4f9..8ef28ed 100644 --- a/hyperdb-mcp/src/table_catalog.rs +++ b/hyperdb-mcp/src/table_catalog.rs @@ -19,20 +19,48 @@ //! | `license` | User / LLM | //! | `notes` | User / LLM | //! -//! The backing table is `_table_catalog`. After the ephemeral-primary -//! migration, the catalog tracks tables in the **persistent** database -//! only — ephemeral scratch tables aren't worth cataloguing because the -//! database is replaced every session. All operations therefore target -//! the persistent attachment via fully-qualified SQL -//! (`"persistent"."public"."_table_catalog"`); when the engine has no -//! persistent attachment (`--ephemeral-only`), catalog operations are -//! no-ops. +//! The backing table is `_table_catalog`. Each writable database that +//! receives an ingest gets its own catalog, lazily seeded on first +//! ingest (or first `set_table_metadata`): +//! +//! - **Ephemeral primary writes** stub into the persistent catalog +//! (so a long-running session's bookkeeping survives the +//! ephemeral file's per-session deletion). Reconciliation cleans +//! up rows whose tables no longer exist on engine bootstrap. +//! - **`database = "persistent"`** writes to the persistent catalog +//! directly (same destination as the ephemeral case). +//! - **`database = ""`** writes to that DB's +//! own `_table_catalog`. Read-only attachments never get a catalog. +//! - **`--ephemeral-only` mode (no persistent attached)** is a +//! degraded mode where catalog operations are no-ops because there +//! is nowhere durable to put bookkeeping. +//! +//! Per-DB catalog routing is exposed via the `*_in` variants +//! ([`ensure_exists_in`], [`upsert_stub_in`], [`set_metadata_in`], +//! [`get_in`], [`list_in`], [`delete_for_in`], [`reconcile_in`]). The +//! original names ([`ensure_exists`], [`upsert_stub`], …) become +//! 1-line wrappers passing `target_db = None` (which resolves to the +//! persistent catalog). //! //! `_table_catalog` is hidden from the user-visible //! [`Engine::describe_tables`] output (see //! [`crate::engine::is_internal_table`]) so the LLM doesn't see it //! alongside its data tables; users who want to inspect it directly -//! can run `SELECT * FROM "persistent"."public"."_table_catalog"`. +//! can run `SELECT * FROM ""."public"."_table_catalog"`. +//! +//! # Concurrency +//! +//! Catalog upserts use DELETE + INSERT inside a single transaction. +//! Hyper rejects PRIMARY KEY ("Index support is disabled") and +//! `INSERT … ON CONFLICT` ("syntax error: got ON"), so atomic upsert +//! at the SQL layer isn't viable. Within one MCP server engine the +//! lack of a PK is not a race risk: every catalog write runs inside +//! [`crate::server::HyperMcpServer::with_engine`], which holds the +//! engine mutex for the duration — so concurrent ingests serialize +//! at the catalog write boundary even though the data-write path is +//! parallel through the watcher pool's separate connections. +//! Cross-process concurrency on the same `.hyper` file is out of +//! scope. //! //! [`HyperMcpServer`]: crate::server::HyperMcpServer @@ -48,21 +76,39 @@ use serde_json::Value; /// run a fully-qualified SELECT. pub const TABLE_CATALOG_TABLE: &str = "_table_catalog"; -/// Returns the fully-qualified SQL reference for `_table_catalog` in the -/// engine's persistent attachment, or `None` when the engine has no -/// persistent attachment (the `--ephemeral-only` case). All catalog -/// read/write helpers consult this; `None` means catalog operations are -/// no-ops because there's nowhere durable to put bookkeeping. -fn qualified_catalog(engine: &Engine) -> Option { - if engine.has_persistent() { - Some(format!( - "\"{}\".\"public\".\"{}\"", - Engine::PERSISTENT_ALIAS, - TABLE_CATALOG_TABLE - )) - } else { - None - } +/// Returns the fully-qualified SQL reference for `_table_catalog` in +/// `target_db`, or `None` when no durable destination exists. +/// +/// Routing: +/// - `None` → persistent catalog if attached, else `None` (the +/// `--ephemeral-only` case, where catalog operations are no-ops). +/// - `Some("persistent")` (case-insensitive) → persistent catalog. +/// - `Some(alias)` → `""."public"."_table_catalog"`. The +/// caller is responsible for ensuring the catalog table exists in +/// that DB (typically by calling [`ensure_exists_in`] first); this +/// function does not validate existence or writability. +fn qualified_catalog_in(engine: &Engine, target_db: Option<&str>) -> Option { + let alias = match target_db { + None | Some("") => { + if engine.has_persistent() { + Engine::PERSISTENT_ALIAS.to_string() + } else { + return None; + } + } + Some(a) if a.eq_ignore_ascii_case(Engine::PERSISTENT_ALIAS) => { + // Persistent is only valid when the engine actually has it. + if !engine.has_persistent() { + return None; + } + Engine::PERSISTENT_ALIAS.to_string() + } + Some(a) => a.to_string(), + }; + let alias_esc = alias.replace('"', "\"\""); + Some(format!( + "\"{alias_esc}\".\"public\".\"{TABLE_CATALOG_TABLE}\"" + )) } /// One row in [`TABLE_CATALOG_TABLE`]. @@ -151,80 +197,81 @@ const CATALOG_COLUMNS: &str = "(\ notes TEXT\ )"; -/// Idempotently create the backing table in the primary workspace. Safe -/// to call on every engine init — if the table already exists this is -/// a no-op. The schema here is the only one the code targets; all -/// prose columns are nullable, so a plain `NULL` insert is always -/// well-formed. +/// Idempotently create the backing table in `target_db`. Safe to call +/// on every engine init or every catalog write — if the table +/// already exists this is a no-op. The schema is the only one the +/// code targets; all prose columns are nullable, so a plain `NULL` +/// insert is always well-formed. /// -/// The DDL uses the unqualified name, which — thanks to the -/// `schema_search_path` pin installed by [`crate::attach::AttachRegistry`] -/// — always resolves to the primary workspace even while additional -/// databases are attached. +/// Routing follows [`qualified_catalog_in`]: `None` resolves to the +/// persistent catalog (or no-op in `--ephemeral-only` mode); +/// `Some("persistent")` is the same; `Some(alias)` targets the +/// attached database directly. /// /// # Errors /// /// Propagates any error from [`Engine::execute_command`] on the /// `CREATE TABLE IF NOT EXISTS` statement — typically connection loss -/// or permission failures. -pub fn ensure_exists(engine: &Engine) -> Result<(), McpError> { - let Some(qualified) = qualified_catalog(engine) else { - // No persistent attachment — there's nowhere durable to put the - // catalog. Treat as success so callers don't have to special-case - // the ephemeral-only mode. +/// or permission failures (the latter for read-only attachments; +/// callers should pre-check writability). +pub fn ensure_exists_in(engine: &Engine, target_db: Option<&str>) -> Result<(), McpError> { + let Some(qualified) = qualified_catalog_in(engine, target_db) else { + // Nowhere durable to put the catalog (--ephemeral-only). return Ok(()); }; let ddl = format!("CREATE TABLE IF NOT EXISTS {qualified} {CATALOG_COLUMNS}"); engine.execute_command(&ddl)?; // After CREATE TABLE IF NOT EXISTS the catalog is guaranteed - // present — short-circuit subsequent existence probes. - engine.mark_catalog_present(); + // present in this DB — short-circuit subsequent existence probes. + if let Some(alias) = resolve_catalog_alias(engine, target_db) { + engine.mark_catalog_present_for(&alias); + } Ok(()) } -/// Idempotently create `_table_catalog` inside an *attached* database -/// (fully qualified as `"{db_alias}"."public"."_table_catalog"`), with -/// the same schema as the primary's catalog. +/// Backward-compatible wrapper: ensure the catalog exists in the +/// persistent attachment (or no-op in `--ephemeral-only` mode). /// -/// Used by `attach_database` when the MCP just created a fresh -/// `.hyper` file via `on_missing: create`: seeding the empty file with -/// the catalog table makes the new workspace immediately usable as a -/// primary the next time someone opens it on its own, without paying -/// a backfill sweep. Only called when the server is not in bare mode; -/// attaching an *existing* database never touches its `_table_catalog`. +/// New code should prefer [`ensure_exists_in`] with an explicit +/// `target_db`. /// /// # Errors /// -/// Propagates any error from the qualified `CREATE TABLE IF NOT EXISTS` -/// against the attached database — typically a wire error or a -/// malformed alias that slipped past validation. +/// Same as [`ensure_exists_in`]. +pub fn ensure_exists(engine: &Engine) -> Result<(), McpError> { + ensure_exists_in(engine, None) +} + +/// Backward-compatible wrapper for the old `ensure_exists_in_database` +/// name. Delegates to [`ensure_exists_in`]. +/// +/// # Errors +/// +/// Same as [`ensure_exists_in`]. +#[deprecated(since = "0.2.0", note = "use ensure_exists_in(engine, Some(db_alias))")] pub fn ensure_exists_in_database(engine: &Engine, db_alias: &str) -> Result<(), McpError> { - let alias_esc = db_alias.replace('"', "\"\""); - let ddl = format!( - "CREATE TABLE IF NOT EXISTS \"{alias_esc}\".\"public\".\"{TABLE_CATALOG_TABLE}\" \ - {CATALOG_COLUMNS}" - ); - engine.execute_command(&ddl)?; - Ok(()) + ensure_exists_in(engine, Some(db_alias)) } // --- Reads ------------------------------------------------------------------ -/// Fetch every catalog row in name-sorted order. Returns an empty `Vec` if -/// the catalog table doesn't exist (callers shouldn't need to pre-check). +/// Fetch every catalog row from `target_db` in name-sorted order. +/// Returns an empty `Vec` when no catalog exists in the target. +/// +/// Routing follows [`qualified_catalog_in`]. /// /// # Errors /// -/// - Propagates any error from `table_present` (connection failure +/// - Propagates any error from `table_present_in` (connection failure /// during the existence probe). /// - Propagates any error from [`Engine::execute_query_to_json`]. /// - Propagates [`ErrorCode::SchemaMismatch`] from `row_to_entry` if /// a persisted row cannot be decoded into a [`CatalogEntry`]. -pub fn list(engine: &Engine) -> Result, McpError> { - let Some(qualified) = qualified_catalog(engine) else { +pub fn list_in(engine: &Engine, target_db: Option<&str>) -> Result, McpError> { + let Some(qualified) = qualified_catalog_in(engine, target_db) else { return Ok(Vec::new()); }; - if !table_present(engine)? { + if !table_present_in(engine, target_db)? { return Ok(Vec::new()); } let sql = format!( @@ -237,17 +284,30 @@ pub fn list(engine: &Engine) -> Result, McpError> { rows.iter().map(row_to_entry).collect() } -/// Fetch a single row by `table_name`, or `Ok(None)` if absent. +/// Backward-compatible wrapper: list rows from the persistent catalog. /// /// # Errors /// -/// Same as [`list`]: propagates errors from `table_present`, the -/// Hyper `SELECT`, or row decoding. -pub fn get(engine: &Engine, table_name: &str) -> Result, McpError> { - let Some(qualified) = qualified_catalog(engine) else { +/// Same as [`list_in`]. +pub fn list(engine: &Engine) -> Result, McpError> { + list_in(engine, None) +} + +/// Fetch a single row from `target_db` by `table_name`, or `Ok(None)` +/// if absent. +/// +/// # Errors +/// +/// Same as [`list_in`]. +pub fn get_in( + engine: &Engine, + table_name: &str, + target_db: Option<&str>, +) -> Result, McpError> { + let Some(qualified) = qualified_catalog_in(engine, target_db) else { return Ok(None); }; - if !table_present(engine)? { + if !table_present_in(engine, target_db)? { return Ok(None); } let sql = format!( @@ -264,10 +324,22 @@ pub fn get(engine: &Engine, table_name: &str) -> Result, Mc } } +/// Backward-compatible wrapper: read from the persistent catalog. +/// +/// # Errors +/// +/// Same as [`get_in`]. +pub fn get(engine: &Engine, table_name: &str) -> Result, McpError> { + get_in(engine, table_name, None) +} + // --- Writes ----------------------------------------------------------------- -/// Upsert a catalog row for `table_name`, carrying forward prose fields -/// from any existing row and refreshing mechanical fields. +/// Upsert a catalog row for `table_name` in `target_db`, carrying +/// forward prose fields from any existing row and refreshing +/// mechanical fields. +/// +/// Routing follows [`qualified_catalog_in`]. /// /// * `load_tool` / `load_params` — overwrite the existing values. /// * `row_count` — overwrite. @@ -278,31 +350,34 @@ pub fn get(engine: &Engine, table_name: &str) -> Result, Mc /// * Prose fields (`source_url`, `source_description`, `purpose`, /// `license`, `notes`) — preserved unchanged from the existing row. /// -/// Implementation is DELETE + INSERT (Hyper lacks `UPSERT`), run inside a -/// transaction for atomicity. +/// Implementation is DELETE + INSERT inside a transaction. Hyper +/// rejects PRIMARY KEY and `INSERT … ON CONFLICT`, so this is the +/// only available shape. Within one MCP server engine, concurrent +/// upserts of the same row are serialized by the engine mutex (see +/// the module-level concurrency note). /// /// # Errors /// -/// - Propagates errors from [`ensure_exists`] and [`get`] (catalog -/// probe and read failures). +/// - Propagates errors from [`ensure_exists_in`] and [`get_in`]. /// - Propagates any transaction error from the enclosing /// [`Engine::execute_in_transaction`] — typically DELETE, INSERT, /// commit, or connection-loss failures. -pub fn upsert_stub( +pub fn upsert_stub_in( engine: &Engine, table_name: &str, load_tool: &str, load_params: Option<&str>, row_count: Option, bump_refresh: bool, + target_db: Option<&str>, ) -> Result<(), McpError> { - ensure_exists(engine)?; - let Some(qualified) = qualified_catalog(engine) else { - // No persistent attachment; nothing to write to. + ensure_exists_in(engine, target_db)?; + let Some(qualified) = qualified_catalog_in(engine, target_db) else { + // No durable destination; nothing to write to. return Ok(()); }; - let existing = get(engine, table_name)?; + let existing = get_in(engine, table_name, target_db)?; let now = Utc::now(); let loaded_at = existing.as_ref().map_or(now, |e| e.loaded_at); let last_refreshed_at = if bump_refresh { @@ -355,26 +430,57 @@ pub fn upsert_stub( Ok(()) } -/// Partial UPDATE of prose fields for one table. Errors with -/// [`ErrorCode::TableNotFound`] if there is no catalog row for -/// `table_name` (callers can decide whether to first stub via -/// [`upsert_stub`] or surface the error). +/// Backward-compatible wrapper: upsert into the persistent catalog. +/// +/// # Errors +/// +/// Same as [`upsert_stub_in`]. +pub fn upsert_stub( + engine: &Engine, + table_name: &str, + load_tool: &str, + load_params: Option<&str>, + row_count: Option, + bump_refresh: bool, +) -> Result<(), McpError> { + upsert_stub_in( + engine, + table_name, + load_tool, + load_params, + row_count, + bump_refresh, + None, + ) +} + +/// Partial UPDATE of prose fields for one table in `target_db`. Errors +/// with [`ErrorCode::TableNotFound`] if there is no catalog row for +/// `table_name` in that DB. +/// +/// Routing follows [`qualified_catalog_in`]. The catalog is lazily +/// seeded in `target_db` if absent (matches today's behavior for +/// the persistent target). /// /// # Errors /// /// - Returns [`ErrorCode::EmptyData`] if `fields` contains no values /// to update. /// - Returns [`ErrorCode::TableNotFound`] if no catalog row exists for -/// `table_name`. -/// - Propagates any error from [`ensure_exists`], [`get`], or the +/// `table_name` in `target_db`. +/// - Returns [`ErrorCode::ReadOnlyViolation`] when there is no durable +/// catalog destination (`--ephemeral-only` with `target_db = None`). +/// - Propagates any error from [`ensure_exists_in`], [`get_in`], or the /// `UPDATE` statement. -pub fn set_metadata( +pub fn set_metadata_in( engine: &Engine, table_name: &str, fields: &MetadataFields, + target_db: Option<&str>, ) -> Result { - ensure_exists(engine)?; - + // Validate caller intent BEFORE seeding the catalog. Both the + // empty-fields and missing-row paths return an error; we don't + // want them to mutate the target DB's schema as a side effect. if fields.is_empty() { return Err(McpError::new( ErrorCode::EmptyData, @@ -386,14 +492,24 @@ pub fn set_metadata( // Require an existing row so we don't accidentally create catalog // entries for tables that don't exist. The server wires the catalog // up to ingest + execute so any real table should already have a - // stub row. - let existing = get(engine, table_name)?.ok_or_else(|| { + // stub row. `get_in` returns None when the catalog table itself + // doesn't exist yet; treat that as "no row" without seeding the + // catalog (the user gets a clean TableNotFound and the .hyper file + // is left untouched). + let existing = get_in(engine, table_name, target_db)?.ok_or_else(|| { + let where_clause = match target_db { + Some(alias) if !alias.eq_ignore_ascii_case(Engine::PERSISTENT_ALIAS) => { + format!(" in database '{alias}'") + } + _ => String::new(), + }; McpError::new( ErrorCode::TableNotFound, format!( - "No catalog entry for table '{table_name}'. Load the table first \ - (load_file / load_data / execute CREATE TABLE) or create it and \ - re-run; the catalog is refreshed automatically on those paths." + "No catalog entry for table '{table_name}'{where_clause}. Load the \ + table first (load_file / load_data / execute CREATE TABLE) or \ + create it and re-run; the catalog is refreshed automatically on \ + those paths." ), ) })?; @@ -418,7 +534,7 @@ pub fn set_metadata( assignments.push(format!("notes = {}", sql_literal_or_null_if_empty(v))); } - let qualified = qualified_catalog(engine).ok_or_else(|| { + let qualified = qualified_catalog_in(engine, target_db).ok_or_else(|| { McpError::new( ErrorCode::ReadOnlyViolation, "set_table_metadata is unavailable in --ephemeral-only mode \ @@ -436,21 +552,38 @@ pub fn set_metadata( // fields). `existing` is our fallback if the re-read somehow returns // nothing — that shouldn't happen, but preserves the previous row // instead of failing spuriously. - get(engine, table_name)?.map_or(Ok(existing), Ok) + get_in(engine, table_name, target_db)?.map_or(Ok(existing), Ok) } -/// Delete the catalog row (if any) for `table_name`. Called when a table -/// is dropped. Idempotent — no error if the row doesn't exist. +/// Backward-compatible wrapper: update prose fields in the persistent +/// catalog. /// /// # Errors /// -/// Propagates any error from `table_present` or from the `DELETE` -/// statement against the catalog table. -pub fn delete_for(engine: &Engine, table_name: &str) -> Result { - let Some(qualified) = qualified_catalog(engine) else { +/// Same as [`set_metadata_in`]. +pub fn set_metadata( + engine: &Engine, + table_name: &str, + fields: &MetadataFields, +) -> Result { + set_metadata_in(engine, table_name, fields, None) +} + +/// Delete the catalog row (if any) for `table_name` in `target_db`. +/// Idempotent — no error if the row doesn't exist. +/// +/// # Errors +/// +/// Propagates any error from `table_present_in` or from the `DELETE`. +pub fn delete_for_in( + engine: &Engine, + table_name: &str, + target_db: Option<&str>, +) -> Result { + let Some(qualified) = qualified_catalog_in(engine, target_db) else { return Ok(false); }; - if !table_present(engine)? { + if !table_present_in(engine, target_db)? { return Ok(false); } let sql = format!( @@ -461,32 +594,40 @@ pub fn delete_for(engine: &Engine, table_name: &str) -> Result { Ok(affected > 0) } -/// Synchronize the catalog against the current set of user tables. +/// Backward-compatible wrapper: delete from the persistent catalog. +/// +/// # Errors +/// +/// Same as [`delete_for_in`]. +pub fn delete_for(engine: &Engine, table_name: &str) -> Result { + delete_for_in(engine, table_name, None) +} + +/// Synchronize the catalog in `target_db` against the current set of +/// user tables in that DB. /// /// * Insert a stub row for every user table missing from the catalog. /// These stubs use `load_tool = "unknown"` so callers can later tell /// the difference between "loaded via a tool and we tracked it" and /// "found during reconciliation". -/// * Delete catalog rows whose table no longer exists in Hyper. -/// * Refresh `row_count` on every remaining row so `SELECT * FROM -/// _table_catalog` always reflects current size (cheap: it's one -/// `COUNT(*)` per table, and the user set is small). +/// * Delete catalog rows whose table no longer exists in `target_db`. +/// * Refresh `row_count` on every remaining row. /// /// Does *not* bump `last_refreshed_at` — reconciliation is a housekeeping -/// pass, not a data refresh. Only explicit loads mark a refresh. +/// pass, not a data refresh. /// /// # Errors /// -/// - Propagates any error from [`ensure_exists`], `user_tables`, -/// [`list`], [`delete_for`], `refresh_row_count`, or [`upsert_stub`]. -/// - `row_count_of` failures are swallowed (the row count falls back -/// to `None`), so per-table count probe failures do not abort the -/// sweep. -pub fn reconcile(engine: &Engine) -> Result<(), McpError> { - ensure_exists(engine)?; - - let tables = user_tables(engine)?; - let catalog_entries = list(engine)?; +/// - Propagates any error from [`ensure_exists_in`], `user_tables_in`, +/// [`list_in`], [`delete_for_in`], `refresh_row_count_in`, or +/// [`upsert_stub_in`]. +/// - `row_count_of_in` failures are swallowed (the row count falls +/// back to `None`). +pub fn reconcile_in(engine: &Engine, target_db: Option<&str>) -> Result<(), McpError> { + ensure_exists_in(engine, target_db)?; + + let tables = user_tables_in(engine, target_db)?; + let catalog_entries = list_in(engine, target_db)?; let catalog_names: std::collections::HashSet = catalog_entries .iter() .map(|e| e.table_name.clone()) @@ -495,21 +636,30 @@ pub fn reconcile(engine: &Engine) -> Result<(), McpError> { for entry in &catalog_entries { if !live_tables.contains(&entry.table_name) { - delete_for(engine, &entry.table_name)?; + delete_for_in(engine, &entry.table_name, target_db)?; } } for table in &tables { - let row_count = row_count_of(engine, table).ok(); + let row_count = row_count_of_in(engine, table, target_db).ok(); if catalog_names.contains(table) { - refresh_row_count(engine, table, row_count)?; + refresh_row_count_in(engine, table, row_count, target_db)?; } else { - upsert_stub(engine, table, "unknown", None, row_count, false)?; + upsert_stub_in(engine, table, "unknown", None, row_count, false, target_db)?; } } Ok(()) } +/// Backward-compatible wrapper: reconcile the persistent catalog. +/// +/// # Errors +/// +/// Same as [`reconcile_in`]. +pub fn reconcile(engine: &Engine) -> Result<(), McpError> { + reconcile_in(engine, None) +} + // --- Internals -------------------------------------------------------------- /// List user-facing tables in the **persistent** attachment (excludes @@ -518,14 +668,17 @@ pub fn reconcile(engine: &Engine) -> Result<(), McpError> { /// `pg_catalog.pg_tables` probe so it sees inside the attachment; /// `Catalog::get_table_names` would target the connection's primary /// (ephemeral) by default. -fn user_tables(engine: &Engine) -> Result, McpError> { - if !engine.has_persistent() { +/// List user-facing tables in `target_db` (excludes `_hyperdb_*` +/// internals and the catalog itself). Returns an empty Vec when no +/// catalog destination exists for the target. +fn user_tables_in(engine: &Engine, target_db: Option<&str>) -> Result, McpError> { + let Some(alias) = resolve_catalog_alias(engine, target_db) else { return Ok(Vec::new()); - } + }; + let alias_esc = alias.replace('"', "\"\""); let sql = format!( - "SELECT tablename FROM \"{}\".pg_catalog.pg_tables \ - WHERE schemaname = 'public'", - Engine::PERSISTENT_ALIAS + "SELECT tablename FROM \"{alias_esc}\".pg_catalog.pg_tables \ + WHERE schemaname = 'public'" ); let rows = engine.execute_query_to_json(&sql)?; Ok(rows @@ -539,19 +692,23 @@ fn user_tables(engine: &Engine) -> Result, McpError> { .collect()) } -/// `true` when `_table_catalog` is present inside the persistent -/// attachment. Returns `false` if there is no persistent attachment. +/// `true` when `_table_catalog` is present inside `target_db`. +/// Returns `false` when no catalog destination exists. /// -/// Caches the result on the engine after the first probe — subsequent -/// reads/writes don't re-run the existence query. Callers that mutate -/// the catalog (`ensure_exists`) update the cache directly via -/// [`Engine::mark_catalog_present`]. -fn table_present(engine: &Engine) -> Result { - engine.catalog_present_in_persistent(|engine| { +/// Caches the per-DB result on the engine after the first probe so +/// subsequent reads/writes skip the existence query. Catalog +/// mutators ([`ensure_exists_in`]) update the cache directly via +/// [`Engine::mark_catalog_present_for`]. +fn table_present_in(engine: &Engine, target_db: Option<&str>) -> Result { + let Some(alias) = resolve_catalog_alias(engine, target_db) else { + return Ok(false); + }; + let alias_for_probe = alias.clone(); + engine.catalog_present_in(&alias, move |engine| { + let alias_esc = alias_for_probe.replace('"', "\"\""); let sql = format!( - "SELECT tablename FROM \"{}\".pg_catalog.pg_tables \ + "SELECT tablename FROM \"{alias_esc}\".pg_catalog.pg_tables \ WHERE schemaname = 'public' AND tablename = {}", - Engine::PERSISTENT_ALIAS, sql_literal(TABLE_CATALOG_TABLE) ); let rows = engine.execute_query_to_json(&sql)?; @@ -559,18 +716,20 @@ fn table_present(engine: &Engine) -> Result { }) } -/// Return `COUNT(*)` for a user table inside the persistent attachment. -/// Quoted to handle mixed-case or keyword-like names. Returns 0 if no -/// persistent attachment is present. -fn row_count_of(engine: &Engine, table_name: &str) -> Result { - if !engine.has_persistent() { +/// Return `COUNT(*)` for a user table inside `target_db`. Quoted to +/// handle mixed-case or keyword-like names. Returns 0 if no catalog +/// destination exists. +fn row_count_of_in( + engine: &Engine, + table_name: &str, + target_db: Option<&str>, +) -> Result { + let Some(alias) = resolve_catalog_alias(engine, target_db) else { return Ok(0); - } + }; + let alias_esc = alias.replace('"', "\"\""); let quoted = table_name.replace('"', "\"\""); - let sql = format!( - "SELECT COUNT(*) AS cnt FROM \"{}\".\"public\".\"{quoted}\"", - Engine::PERSISTENT_ALIAS - ); + let sql = format!("SELECT COUNT(*) AS cnt FROM \"{alias_esc}\".\"public\".\"{quoted}\""); let rows = engine.execute_query_to_json(&sql)?; Ok(rows .first() @@ -578,15 +737,15 @@ fn row_count_of(engine: &Engine, table_name: &str) -> Result { .unwrap_or(0)) } -/// Cheap UPDATE for just the `row_count` column of an existing row. -/// Used by [`reconcile`] so we don't rewrite the whole row just to -/// refresh counts. -fn refresh_row_count( +/// Cheap UPDATE for just the `row_count` column of an existing row in +/// `target_db`. Used by [`reconcile_in`]. +fn refresh_row_count_in( engine: &Engine, table_name: &str, row_count: Option, + target_db: Option<&str>, ) -> Result<(), McpError> { - let Some(qualified) = qualified_catalog(engine) else { + let Some(qualified) = qualified_catalog_in(engine, target_db) else { return Ok(()); }; let sql = format!( @@ -598,6 +757,30 @@ fn refresh_row_count( Ok(()) } +/// Resolve `target_db` to the canonical alias whose catalog this +/// operation should target. Mirrors [`qualified_catalog_in`]'s +/// routing but returns just the alias (not the qualified table +/// reference) for callers that need to build their own SQL. +fn resolve_catalog_alias(engine: &Engine, target_db: Option<&str>) -> Option { + match target_db { + None | Some("") => { + if engine.has_persistent() { + Some(Engine::PERSISTENT_ALIAS.to_string()) + } else { + None + } + } + Some(a) if a.eq_ignore_ascii_case(Engine::PERSISTENT_ALIAS) => { + if engine.has_persistent() { + Some(Engine::PERSISTENT_ALIAS.to_string()) + } else { + None + } + } + Some(a) => Some(a.to_string()), + } +} + /// Hyper emits TIMESTAMP columns as either `YYYY-MM-DDTHH:MM:SS...` (RFC /// 3339) or `YYYY-MM-DD HH:MM:SS[.fff]` depending on path; accept both. fn parse_timestamp(s: &str) -> Result, McpError> { diff --git a/hyperdb-mcp/src/watcher.rs b/hyperdb-mcp/src/watcher.rs index dd3bbad..20f2ec6 100644 --- a/hyperdb-mcp/src/watcher.rs +++ b/hyperdb-mcp/src/watcher.rs @@ -47,6 +47,7 @@ //! them into a `tokio::sync::mpsc` on a small helper thread so the tokio //! consumer can `.recv().await` naturally. +use crate::attach::{AttachRegistry, AttachSource}; use crate::engine::Engine; use crate::error::{ErrorCode, McpError}; use crate::ingest::{ @@ -71,11 +72,64 @@ use tokio::task::JoinHandle; /// this to the full data file name (e.g. `orders.csv` → `orders.csv.ready`). pub const READY_SUFFIX: &str = ".ready"; +/// Resolve the file path the watcher pool should open as its workspace. +/// `None` → primary (ephemeral). `Some("persistent")` → the engine's +/// persistent path. `Some(alias)` → user-attached writable file from +/// `AttachRegistry`. Read-only attachments are rejected because the +/// watcher needs to INSERT/COPY into the table. +fn resolve_pool_workspace( + eng: &Engine, + attachments: Option<&AttachRegistry>, + target_db: Option<&str>, +) -> Result { + let Some(alias) = target_db else { + return Ok(eng.ephemeral_path().to_string_lossy().to_string()); + }; + if alias.eq_ignore_ascii_case(Engine::PERSISTENT_ALIAS) { + let path = eng.persistent_path().ok_or_else(|| { + McpError::new( + ErrorCode::InvalidArgument, + "target 'persistent' but the server is in --ephemeral-only mode", + ) + })?; + return Ok(path.to_string_lossy().to_string()); + } + let registry = attachments.ok_or_else(|| { + McpError::new( + ErrorCode::InternalError, + "watcher pool requested for a user-attached alias but no AttachRegistry was supplied", + ) + })?; + let entry = registry.get(alias).ok_or_else(|| { + McpError::new( + ErrorCode::InvalidArgument, + format!("database '{alias}' is not attached"), + ) + })?; + if !entry.writable { + return Err(McpError::new( + ErrorCode::InvalidArgument, + format!( + "database '{alias}' was attached read-only. \ + Re-attach with writable:true to use it as a watcher target." + ), + )); + } + let AttachSource::LocalFile { path } = &entry.source; + Ok(path.to_string_lossy().to_string()) +} + /// Build a watcher connection pool from the current engine. Pulled out /// of [`start_watching`] so the recovery path (post hyperd restart) /// can call it again to swap in a fresh pool. +/// +/// `target_db` selects the pool's workspace file: `None` → ephemeral +/// primary, `Some("persistent")` → persistent attachment, +/// `Some(alias)` → user-attached writable file (from `attachments`). fn build_watcher_pool( engine: &Arc>>, + attachments: Option<&AttachRegistry>, + target_db: Option<&str>, concurrency: usize, ) -> Result, McpError> { let guard = engine @@ -88,10 +142,7 @@ fn build_watcher_pool( ) })?; let endpoint = eng.hyperd_endpoint()?; - // Watcher pool connects to the engine's primary (ephemeral). The - // user-supplied table on `watch_directory` always lives there - // unless a future flag routes it to the persistent attachment. - let workspace = eng.ephemeral_path().to_string_lossy().to_string(); + let workspace = resolve_pool_workspace(eng, attachments, target_db)?; let cfg = PoolConfig::new(endpoint, workspace) .create_mode(CreateMode::DoNotCreate) .max_size(concurrency); @@ -110,9 +161,11 @@ fn build_watcher_pool( async fn rebuild_watcher_pool( pool_slot: &tokio::sync::RwLock>, engine: &Arc>>, + attachments: Option<&AttachRegistry>, + target_db: Option<&str>, concurrency: usize, ) -> Result<(), McpError> { - let new_pool = build_watcher_pool(engine, concurrency)?; + let new_pool = build_watcher_pool(engine, attachments, target_db, concurrency)?; let mut guard = pool_slot.write().await; *guard = new_pool; Ok(()) @@ -182,6 +235,13 @@ impl WatcherStats { pub struct WatcherHandle { pub directory: PathBuf, pub table: String, + /// Resolved target database alias for this watcher. `None` → + /// primary; `Some("persistent")` → persistent attachment; + /// `Some(alias)` → user-attached writable. Stored so the + /// post-reconnect rebuild path resolves to the same target, + /// and so `detach_database` can refuse to detach an alias with + /// an active watcher. + pub target_db: Option, pub stats: Arc>, /// Live counter of in-flight ingest tasks. Decremented on task completion /// via an RAII guard. @@ -262,6 +322,7 @@ impl WatcherRegistry { json!({ "directory": h.directory.to_string_lossy(), "table": h.table, + "target_db": h.target_db.clone().unwrap_or_else(|| "local".into()), "files_ingested": stats.files_ingested, "files_failed": stats.files_failed, "last_event_ms_ago": last_event_ms_ago, @@ -329,10 +390,12 @@ impl Drop for InFlightGuard { /// (file read, schema inference, or Hyper `COPY` / `INSERT` errors). pub fn start_watching( engine: Arc>>, + attachments: Arc, registry: Arc, subscriptions: Option>, dir: PathBuf, table: String, + target_db: Option, options: WatchOptions, ) -> Result { if !dir.exists() { @@ -380,6 +443,8 @@ pub fn start_watching( // watcher. let pool = Arc::new(tokio::sync::RwLock::new(build_watcher_pool( &engine, + Some(attachments.as_ref()), + target_db.as_deref(), concurrency, )?)); @@ -423,6 +488,8 @@ pub fn start_watching( process_ready_with_recovery( &pool, &engine, + Some(attachments.as_ref()), + target_db.as_deref(), concurrency, subscriptions.as_deref(), &canonical, @@ -492,12 +559,14 @@ pub fn start_watching( let task = { let pool = Arc::clone(&pool); let engine_for_pool = Arc::clone(&engine); + let attachments_for_pool = Arc::clone(&attachments); let subs = subscriptions.clone(); let stats = Arc::clone(&stats); let in_flight = Arc::clone(&in_flight); let in_flight_paths = Arc::clone(&in_flight_paths); let dir = canonical.clone(); let table = table.clone(); + let task_target_db = target_db.clone(); tokio::spawn(async move { while let Some(event_res) = async_rx.recv().await { let Ok(event) = event_res else { continue }; @@ -522,6 +591,8 @@ pub fn start_watching( } let pool_slot = Arc::clone(&pool); let engine_handle = Arc::clone(&engine_for_pool); + let attachments_handle = Arc::clone(&attachments_for_pool); + let target_db_clone = task_target_db.clone(); let subs = subs.clone(); let stats = Arc::clone(&stats); let in_flight = Arc::clone(&in_flight); @@ -533,6 +604,8 @@ pub fn start_watching( process_ready_with_recovery( &pool_slot, &engine_handle, + Some(attachments_handle.as_ref()), + target_db_clone.as_deref(), concurrency, subs.as_deref(), &dir, @@ -553,6 +626,7 @@ pub fn start_watching( let handle = WatcherHandle { directory: canonical.clone(), table, + target_db, stats, in_flight, watcher: Some(watcher), @@ -669,6 +743,11 @@ async fn ingest_one_ready_file( format!("Failed to check out connection: {e}"), ) })?; + // The pool was built against the target's `.hyper` file as its + // workspace, so from these connections' perspective the target IS + // the primary database — qualified SQL with "persistent"."public" + // wouldn't resolve here. Keep target_db = None so the unqualified + // identifier routes into the pool's primary. let opts = IngestOptions { table: table.to_string(), mode: "append".into(), @@ -698,6 +777,8 @@ async fn ingest_one_ready_file( async fn process_ready_with_recovery( pool_slot: &tokio::sync::RwLock>, engine: &Arc>>, + attachments: Option<&AttachRegistry>, + target_db: Option<&str>, concurrency: usize, subscriptions: Option<&SubscriptionRegistry>, dir: &Path, @@ -743,7 +824,8 @@ async fn process_ready_with_recovery( err = %err.message, "watcher: detected connection-lost error, rebuilding pool and retrying" ); - match rebuild_watcher_pool(pool_slot, engine, concurrency).await { + match rebuild_watcher_pool(pool_slot, engine, attachments, target_db, concurrency).await + { Ok(()) => { let active_pool = pool_slot.read().await.clone(); result = @@ -773,6 +855,41 @@ async fn process_ready_with_recovery( data_path.display(), table ); + // Update _table_catalog on the engine's main connection + // (which sees both ephemeral and persistent + any user- + // attached aliases). The pool wrote the data into the + // target's .hyper file; the engine connection now stamps + // a stub row in that DB's per-DB _table_catalog. + // + // Errors here are logged but never block the success + // bookkeeping above — the data is in. Mirrors the sync + // load_file/load_data handlers' best-effort contract. + let row_count_i64 = i64::try_from(rows).unwrap_or(i64::MAX); + let load_params = serde_json::to_string(&json!({ + "watch_directory": dir.to_string_lossy(), + "database": target_db.unwrap_or("local"), + })) + .ok(); + if let Ok(guard) = engine.lock() { + if let Some(eng) = guard.as_ref() { + if let Err(e) = crate::table_catalog::upsert_stub_in( + eng, + table, + "watch_directory", + load_params.as_deref(), + Some(row_count_i64), + true, + target_db, + ) { + tracing::warn!( + table = %table, + target_db = ?target_db, + err = %e.message, + "watcher: failed to update _table_catalog after ingest" + ); + } + } + } if let Some(subs) = subscriptions { for uri in uris_for_table_change(table) { subs.notify_updated(&uri); diff --git a/hyperdb-mcp/tests/cross_db_dml_smoke.rs b/hyperdb-mcp/tests/cross_db_dml_smoke.rs new file mode 100644 index 0000000..68c8031 --- /dev/null +++ b/hyperdb-mcp/tests/cross_db_dml_smoke.rs @@ -0,0 +1,292 @@ +// Copyright (c) 2026, Salesforce, Inc. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 OR MIT + +//! Pre-flight smoke battery for cross-database SQL shapes that the +//! "remove v1 limitations" PR plumbs through ingest/merge/export. +//! +//! Plan: `HYPERDB_MCP_REMOVE_V1_LIMITATIONS_PLAN.md` §"Pre-flight smoke +//! battery". Each test verifies one SQL shape against a *user-attached +//! writable* database. If any shape fails, the iter-1 merge-in-target-DB +//! design is dead and the plan needs to redesign before plumbing. +//! +//! Marked `#[ignore]` so they don't run on every `cargo test` — invoke +//! explicitly via `cargo test -p hyperdb-mcp --test cross_db_dml_smoke +//! -- --ignored`. + +use hyperdb_mcp::attach::{AttachRegistry, AttachRequest, AttachSource, OnMissing}; +use hyperdb_mcp::engine::Engine; +use tempfile::TempDir; + +/// Build a primary workspace plus an attached writable database under +/// alias `"smoke"` pointing at a freshly-created `.hyper` file. Returns +/// `(engine, registry, _dir)` — drop order matters: registry → engine → +/// dir, so the temp directory outlives the engine. +fn setup() -> (Engine, AttachRegistry, TempDir) { + let dir = TempDir::new().unwrap(); + let primary_path = dir.path().join("primary.hyper"); + let attached_path = dir.path().join("attached.hyper"); + + let engine = Engine::new_no_daemon(Some(primary_path.to_string_lossy().into())).unwrap(); + let registry = AttachRegistry::new(); + registry + .attach( + &engine, + AttachRequest { + alias: "smoke".into(), + source: AttachSource::LocalFile { + path: attached_path, + }, + writable: true, + on_missing: OnMissing::Create, + }, + ) + .unwrap(); + (engine, registry, dir) +} + +/// Shape 1: `CREATE TABLE "alias"."public"."tmp" AS SELECT ...` — the +/// merge path's temp-table strategy depends on this. +#[test] +#[ignore = "pre-flight smoke; run with --ignored"] +fn cross_db_ctas_into_attached_writable() { + let (engine, _reg, _dir) = setup(); + engine + .execute_command( + "CREATE TABLE \"smoke\".\"public\".\"tmp\" AS \ + SELECT * FROM (VALUES (1, 'a'), (2, 'b')) AS v(k, val)", + ) + .expect("CTAS into attached writable DB must succeed"); + + let rows = engine + .execute_query_to_json("SELECT k, val FROM \"smoke\".\"public\".\"tmp\" ORDER BY k") + .unwrap(); + assert_eq!(rows.len(), 2); + assert_eq!(rows[0]["k"], 1); + assert_eq!(rows[1]["val"], "b"); +} + +/// Shape 2: qualified `DELETE ... USING ...` where target and source are +/// both in the attached DB — the merge path's row-by-key delete step. +#[test] +#[ignore = "pre-flight smoke; run with --ignored"] +fn cross_db_qualified_delete_using() { + let (engine, _reg, _dir) = setup(); + + engine + .execute_command("CREATE TABLE \"smoke\".\"public\".\"t\" (k INT, v TEXT)") + .unwrap(); + engine + .execute_command( + "INSERT INTO \"smoke\".\"public\".\"t\" VALUES (1, 'old'), (2, 'keep'), (3, 'old')", + ) + .unwrap(); + + engine + .execute_command( + "CREATE TABLE \"smoke\".\"public\".\"tmp\" AS \ + SELECT * FROM (VALUES (1), (3)) AS v(k)", + ) + .unwrap(); + + engine + .execute_command( + "DELETE FROM \"smoke\".\"public\".\"t\" t \ + USING \"smoke\".\"public\".\"tmp\" s \ + WHERE t.k = s.k", + ) + .expect("qualified DELETE-USING must succeed across same attached DB"); + + let rows = engine + .execute_query_to_json("SELECT k, v FROM \"smoke\".\"public\".\"t\" ORDER BY k") + .unwrap(); + assert_eq!(rows.len(), 1, "rows 1 and 3 should be deleted"); + assert_eq!(rows[0]["v"], "keep"); +} + +/// Shape 3: qualified `INSERT ... SELECT` — the merge path's append step +/// after the delete. +#[test] +#[ignore = "pre-flight smoke; run with --ignored"] +fn cross_db_qualified_insert_select() { + let (engine, _reg, _dir) = setup(); + + engine + .execute_command("CREATE TABLE \"smoke\".\"public\".\"t\" (k INT, v TEXT)") + .unwrap(); + engine + .execute_command( + "CREATE TABLE \"smoke\".\"public\".\"tmp\" AS \ + SELECT * FROM (VALUES (1, 'a'), (2, 'b')) AS v(k, v)", + ) + .unwrap(); + + engine + .execute_command( + "INSERT INTO \"smoke\".\"public\".\"t\" \ + SELECT * FROM \"smoke\".\"public\".\"tmp\"", + ) + .expect("qualified INSERT-SELECT must succeed"); + + let rows = engine + .execute_query_to_json("SELECT k, v FROM \"smoke\".\"public\".\"t\" ORDER BY k") + .unwrap(); + assert_eq!(rows.len(), 2); +} + +/// Shape 4: qualified `ALTER TABLE ... ADD COLUMN` — the merge path's +/// new-column promotion step when the incoming schema is wider. +#[test] +#[ignore = "pre-flight smoke; run with --ignored"] +fn cross_db_qualified_alter_add_column() { + let (engine, _reg, _dir) = setup(); + + engine + .execute_command("CREATE TABLE \"smoke\".\"public\".\"t\" (k INT)") + .unwrap(); + + engine + .execute_command("ALTER TABLE \"smoke\".\"public\".\"t\" ADD COLUMN extra TEXT") + .expect("qualified ALTER ADD COLUMN must succeed"); + + engine + .execute_command("INSERT INTO \"smoke\".\"public\".\"t\" VALUES (1, 'x')") + .unwrap(); + let rows = engine + .execute_query_to_json("SELECT extra FROM \"smoke\".\"public\".\"t\"") + .unwrap(); + assert_eq!(rows[0]["extra"], "x"); +} + +/// Shape 6 (Iter 4): `CREATE TABLE … (col TEXT PRIMARY KEY, …)` against +/// an attached writable DB. +/// +/// **Result: rejected by Hyper with "Index support is disabled (0A000)".** +/// Hyper backs PKs with indexes, which aren't supported in the bundled +/// build. PK-based atomic upsert is therefore not viable; Iter 4 +/// designs around it via a Rust-side per-(alias, table_name) Mutex +/// plus DELETE+INSERT in a transaction (the existing pattern, just +/// extended cross-DB). +#[test] +#[ignore = "pre-flight smoke; run with --ignored — DOCUMENTS that Hyper rejects PK"] +fn cross_db_create_table_with_primary_key_is_rejected() { + let (engine, _reg, _dir) = setup(); + + let err = engine + .execute_command( + "CREATE TABLE \"smoke\".\"public\".\"with_pk\" \ + (k TEXT PRIMARY KEY, v INT)", + ) + .expect_err("Hyper must reject PK creation — design depends on this"); + assert!( + err.message.contains("Index support is disabled") + || err.message.contains("not implemented") + || err.message.contains("not supported"), + "expected an index/PK rejection, got: {}", + err.message + ); +} + +/// Shape 7 (Iter 4): `ALTER TABLE … ADD CONSTRAINT … PRIMARY KEY` against +/// an attached writable DB. +/// +/// **Result: rejected by Hyper with "named constraints not implemented +/// yet (0A000)".** Confirms that PK migration via ALTER is also a dead +/// end. Iter 4 falls back to non-PK design — see the plan. +#[test] +#[ignore = "pre-flight smoke; run with --ignored — DOCUMENTS that Hyper rejects ALTER ADD CONSTRAINT"] +fn cross_db_alter_add_primary_key_is_rejected() { + let (engine, _reg, _dir) = setup(); + + engine + .execute_command("CREATE TABLE \"smoke\".\"public\".\"add_pk\" (k TEXT, v INT)") + .unwrap(); + let err = engine + .execute_command( + "ALTER TABLE \"smoke\".\"public\".\"add_pk\" \ + ADD CONSTRAINT add_pk_pk PRIMARY KEY (k)", + ) + .expect_err("Hyper must reject ADD CONSTRAINT"); + assert!( + err.message.contains("not implemented") + || err.message.contains("Index support is disabled"), + "expected a not-implemented/index rejection, got: {}", + err.message + ); +} + +/// Shape 8 (Iter 4): `INSERT … ON CONFLICT … DO UPDATE` against an +/// attached writable DB. +/// +/// **Result: rejected because ON CONFLICT requires a unique index on +/// the conflict columns, which Hyper doesn't support.** Atomic upsert +/// in a single statement is therefore not viable. +#[test] +#[ignore = "pre-flight smoke; run with --ignored — DOCUMENTS that ON CONFLICT is unavailable"] +fn cross_db_insert_on_conflict_is_rejected() { + let (engine, _reg, _dir) = setup(); + + engine + .execute_command("CREATE TABLE \"smoke\".\"public\".\"upsert\" (k TEXT, v INT, prose TEXT)") + .unwrap(); + engine + .execute_command("INSERT INTO \"smoke\".\"public\".\"upsert\" VALUES ('a', 1, 'hello')") + .unwrap(); + + let err = engine + .execute_command( + "INSERT INTO \"smoke\".\"public\".\"upsert\" (k, v, prose) \ + VALUES ('a', 99, 'overwritten') \ + ON CONFLICT (k) DO UPDATE SET v = EXCLUDED.v", + ) + .expect_err("Hyper must reject ON CONFLICT — syntax not supported"); + // Hyper rejects ON CONFLICT at the *parser* layer ("syntax error: got + // ON, expected FETCH, FOR, LIMIT, OFFSET") — the dialect doesn't + // include the upsert grammar at all, regardless of index support. + assert!( + err.message.contains("syntax error") + || err.message.contains("Index support is disabled") + || err.message.contains("not implemented"), + "expected a syntax/index rejection, got: {}", + err.message + ); +} + +/// Shape 5: qualified `pg_catalog.pg_tables` and column-introspection +/// probes against the attached DB — the basis for `table_exists_in` and +/// `column_metadata_in`. +#[test] +#[ignore = "pre-flight smoke; run with --ignored"] +fn cross_db_qualified_pg_catalog_probes() { + let (engine, _reg, _dir) = setup(); + + engine + .execute_command("CREATE TABLE \"smoke\".\"public\".\"probe_me\" (id INT, label TEXT)") + .unwrap(); + + let table_rows = engine + .execute_query_to_json( + "SELECT tablename FROM \"smoke\".pg_catalog.pg_tables \ + WHERE schemaname = 'public' AND tablename = 'probe_me'", + ) + .expect("qualified pg_tables probe must succeed"); + assert_eq!( + table_rows.len(), + 1, + "table 'probe_me' must be visible via attached pg_catalog" + ); + + let column_rows = engine + .execute_query_to_json( + "SELECT a.attname AS name, t.typname AS type \ + FROM \"smoke\".pg_catalog.pg_attribute a \ + JOIN \"smoke\".pg_catalog.pg_class c ON a.attrelid = c.oid \ + JOIN \"smoke\".pg_catalog.pg_type t ON a.atttypid = t.oid \ + JOIN \"smoke\".pg_catalog.pg_namespace n ON c.relnamespace = n.oid \ + WHERE n.nspname = 'public' AND c.relname = 'probe_me' AND a.attnum > 0 \ + ORDER BY a.attnum", + ) + .expect("qualified pg_attribute/pg_type join must succeed"); + assert_eq!(column_rows.len(), 2); + assert_eq!(column_rows[0]["name"], "id"); + assert_eq!(column_rows[1]["name"], "label"); +} diff --git a/hyperdb-mcp/tests/daemon_tests.rs b/hyperdb-mcp/tests/daemon_tests.rs index 2523a73..3821aad 100644 --- a/hyperdb-mcp/tests/daemon_tests.rs +++ b/hyperdb-mcp/tests/daemon_tests.rs @@ -9,7 +9,6 @@ //! are process-global, these tests MUST run sequentially. We enforce this via a //! shared mutex — every test that touches env vars acquires `ENV_LOCK` first. -use std::net::TcpListener; use std::sync::{Arc, Mutex}; use std::time::{Duration, Instant}; @@ -768,20 +767,30 @@ impl TestDaemon { let state_dir_guard = EnvGuard::set("HYPERDB_STATE_DIR", tmp.path().to_str().unwrap()); - let port = find_free_port(); - let port_guard = EnvGuard::set("HYPERDB_DAEMON_PORT", &port.to_string()); - - // Start the daemon in a background tokio runtime - let daemon_port = port; - std::thread::spawn(move || { - let rt = tokio::runtime::Runtime::new().unwrap(); + // Pass port 0 so the daemon's HealthListener binds an OS-assigned + // free port and reports it back via the discovery file. We avoid + // the find_free_port → set env → daemon binds later TOCTOU race + // where another process could grab the port between pick and bind + // (a real source of flakes on busy CI runners). The + // `HYPERDB_DAEMON_PORT` env var only matters for the + // spawn-daemon-if-missing path; once we've written the discovery + // file, clients read `health_port` directly from it. + let port_guard = EnvGuard::set("HYPERDB_DAEMON_PORT", "0"); + + // Start the daemon in a background tokio runtime. Capture errors + // via a JoinHandle so a bind/spawn failure fails the test fast + // with a real message instead of a 30s timeout. + let daemon_handle = std::thread::spawn(move || -> Result<(), String> { + let rt = tokio::runtime::Runtime::new().map_err(|e| format!("rt: {e}"))?; rt.block_on(async { let config = hyperdb_mcp::daemon::run::DaemonConfig { - port: daemon_port, + port: 0, idle_timeout: Duration::from_secs(300), }; - let _ = hyperdb_mcp::daemon::run::run_daemon(config).await; - }); + hyperdb_mcp::daemon::run::run_daemon(config) + .await + .map_err(|e| format!("run_daemon: {e}")) + }) }); // Wait for daemon to become ready. CI runners (especially macOS) @@ -796,9 +805,19 @@ impl TestDaemon { _port_guard: port_guard, }; } + // Fail fast if the daemon thread has already exited (bind error, + // spawn error, etc.). Avoids the unhelpful 30s-timeout panic. + if daemon_handle.is_finished() { + let msg = match daemon_handle.join() { + Ok(Ok(())) => "daemon thread exited cleanly without writing discovery".into(), + Ok(Err(e)) => format!("daemon thread errored: {e}"), + Err(_) => "daemon thread panicked".into(), + }; + panic!("TestDaemon failed to start: {msg}"); + } assert!( start.elapsed() <= Duration::from_secs(30), - "TestDaemon did not start within 30 seconds (port={daemon_port})" + "TestDaemon did not start within 30 seconds" ); std::thread::sleep(Duration::from_millis(200)); } @@ -822,12 +841,6 @@ impl Drop for TestDaemon { } } -/// Find a free TCP port by binding to port 0 and reading the assigned port. -fn find_free_port() -> u16 { - let listener = TcpListener::bind("127.0.0.1:0").unwrap(); - listener.local_addr().unwrap().port() -} - /// Locate the `hyperd` process by matching the listen-port portion of an /// endpoint string like `127.0.0.1:54321` against `lsof`'s view of TCP ports. /// Returns the PID of whichever process owns the port. Unix-only. diff --git a/hyperdb-mcp/tests/export_tests.rs b/hyperdb-mcp/tests/export_tests.rs index c46c310..a13acf0 100644 --- a/hyperdb-mcp/tests/export_tests.rs +++ b/hyperdb-mcp/tests/export_tests.rs @@ -7,6 +7,7 @@ mod common; use common::TestEngine; +use hyperdb_mcp::engine::Engine; use hyperdb_mcp::error::ErrorCode; use hyperdb_mcp::export::{export_to_file, ExportOptions}; @@ -41,6 +42,7 @@ fn export_csv_from_table() { format: "csv".into(), overwrite: true, format_options: None, + source_db: None, }; let result = export_to_file(&te.engine, &opts).unwrap(); assert_eq!(result.rows, 2); @@ -66,6 +68,7 @@ fn export_csv_from_query() { format: "csv".into(), overwrite: true, format_options: None, + source_db: None, }; let result = export_to_file(&te.engine, &opts).unwrap(); assert_eq!(result.rows, 1); @@ -87,6 +90,7 @@ fn export_parquet_from_table() { format: "parquet".into(), overwrite: true, format_options: None, + source_db: None, }; let result = export_to_file(&te.engine, &opts).unwrap(); assert_eq!(result.rows, 2); @@ -109,11 +113,79 @@ fn export_hyper_copies_workspace() { format: "hyper".into(), overwrite: true, format_options: None, + source_db: None, }; let _result = export_to_file(&te.engine, &opts).unwrap(); assert!(std::fs::metadata(path_str).unwrap().len() > 0); } +/// `format = "hyper"` with `source_db = Some("persistent")` snapshots +/// the persistent attachment instead of the primary. Verifies the +/// rejection lifted in iter 2 produces the right output. +#[test] +fn export_hyper_with_source_db_snapshots_persistent() { + let te = TestEngine::new_ephemeral(); + + // Create a table in primary that should NOT appear in the snapshot, + // and a different table in persistent that SHOULD. + te.engine + .execute_command("CREATE TABLE primary_only (n INT)") + .unwrap(); + te.engine + .execute_command("INSERT INTO primary_only VALUES (99)") + .unwrap(); + te.engine + .execute_command( + "CREATE TABLE \"persistent\".\"public\".\"persisted\" (id INT, label TEXT)", + ) + .unwrap(); + te.engine + .execute_command("INSERT INTO \"persistent\".\"public\".\"persisted\" VALUES (1, 'a')") + .unwrap(); + + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join("snapshot.hyper"); + let path_str = path.to_str().unwrap(); + + let opts = ExportOptions { + sql: None, + table: None, + path: path_str.into(), + format: "hyper".into(), + overwrite: true, + format_options: None, + source_db: Some("persistent".into()), + }; + export_to_file(&te.engine, &opts).expect("hyper export with source_db must succeed"); + + // Verify the snapshot file contains the persistent table and NOT + // the primary-only table by opening it as a fresh persistent + // workspace and listing tables. + let probe_engine = Engine::new_no_daemon(Some(path_str.into())).unwrap(); + let rows = probe_engine + .execute_query_to_json( + "SELECT tablename FROM \"persistent\".pg_catalog.pg_tables \ + WHERE schemaname = 'public' ORDER BY tablename", + ) + .unwrap(); + let names: Vec = rows + .iter() + .filter_map(|r| { + r.get("tablename") + .and_then(|v| v.as_str()) + .map(String::from) + }) + .collect(); + assert!( + names.contains(&"persisted".into()), + "snapshot must contain 'persisted' from source_db; got {names:?}" + ); + assert!( + !names.contains(&"primary_only".into()), + "snapshot must NOT contain 'primary_only' (lives in primary, not persistent); got {names:?}" + ); +} + /// `format = "hyper"` is a whole-workspace file copy and must succeed even /// when neither `sql` nor `table` is provided — the row-oriented /// SQL-resolution check should not apply. Regression test for a dispatcher @@ -133,6 +205,7 @@ fn export_hyper_requires_no_sql_or_table() { format: "hyper".into(), overwrite: true, format_options: None, + source_db: None, }; let result = export_to_file(&te.engine, &opts) .expect("hyper format must accept a bare (path, format) call without sql or table"); @@ -159,6 +232,7 @@ fn export_csv_without_sql_or_table_errors() { format: "csv".into(), overwrite: true, format_options: None, + source_db: None, }; let Err(err) = export_to_file(&te.engine, &opts) else { panic!("csv export must reject calls with no sql and no table") @@ -194,6 +268,7 @@ fn export_overwrite_false_rejects_existing_file() { format: "csv".into(), overwrite: false, format_options: None, + source_db: None, }, ) else { panic!("export with overwrite=false must error when target exists") @@ -224,6 +299,7 @@ fn export_overwrite_false_rejects_existing_file() { format: "hyper".into(), overwrite: false, format_options: None, + source_db: None, }, ) else { panic!("hyper export with overwrite=false must error when target exists") @@ -251,6 +327,7 @@ fn export_overwrite_false_allows_new_path() { format: "csv".into(), overwrite: false, format_options: None, + source_db: None, }; let result = export_to_file(&te.engine, &opts) .expect("overwrite=false must succeed when target doesn't exist yet"); @@ -275,6 +352,7 @@ fn export_overwrite_true_replaces_existing_file() { format: "csv".into(), overwrite: true, format_options: None, + source_db: None, }; let result = export_to_file(&te.engine, &opts) .expect("overwrite=true must succeed even when target already exists"); @@ -311,6 +389,7 @@ fn iceberg_export_round_trips_through_load_iceberg() { format: "iceberg".into(), overwrite: true, format_options: None, + source_db: None, }; let export_result = export_to_file(&te.engine, &export_opts).unwrap(); assert_eq!(export_result.rows, 2); @@ -405,6 +484,7 @@ fn iceberg_export_overwrite_replaces_directory() { format: "iceberg".into(), overwrite: true, format_options: None, + source_db: None, }; let r1 = export_to_file(&te.engine, &opts_first).unwrap(); assert_eq!(r1.rows, 2); @@ -417,6 +497,7 @@ fn iceberg_export_overwrite_replaces_directory() { format: "iceberg".into(), overwrite: true, format_options: None, + source_db: None, }; let r2 = export_to_file(&te.engine, &opts_second).unwrap(); assert_eq!(r2.rows, 1); @@ -458,6 +539,7 @@ fn iceberg_export_refuses_overwrite_when_disabled() { format: "iceberg".into(), overwrite: true, format_options: None, + source_db: None, }; export_to_file(&te.engine, &opts_first).unwrap(); @@ -469,6 +551,7 @@ fn iceberg_export_refuses_overwrite_when_disabled() { format: "iceberg".into(), overwrite: false, format_options: None, + source_db: None, }; let err = export_to_file(&te.engine, &opts_second).err().unwrap(); assert_eq!(err.code, ErrorCode::PermissionDenied); @@ -522,6 +605,7 @@ fn parquet_export_round_trips_through_load_file() { format: "parquet".into(), overwrite: true, format_options: None, + source_db: None, }, ) .unwrap(); @@ -616,6 +700,7 @@ fn arrow_ipc_export_round_trips_through_load_file() { format: "arrow_ipc".into(), overwrite: true, format_options: None, + source_db: None, }, ) .unwrap(); @@ -694,6 +779,7 @@ fn parquet_export_honors_compression_override() { format: "parquet".into(), overwrite: true, format_options: Some(opts), + source_db: None, }, ) .expect("parquet export with compression override should succeed"); @@ -748,6 +834,7 @@ fn csv_export_honors_delimiter_override() { format: "csv".into(), overwrite: true, format_options: Some(opts), + source_db: None, }, ) .unwrap(); @@ -792,6 +879,7 @@ fn format_options_invalid_shapes_reject_cleanly() { format: "parquet".into(), overwrite: true, format_options: Some(null_val), + source_db: None, }, ) .err() @@ -813,6 +901,7 @@ fn format_options_invalid_shapes_reject_cleanly() { format: "parquet".into(), overwrite: true, format_options: Some(bad_key), + source_db: None, }, ) .err() diff --git a/hyperdb-mcp/tests/integration_tests.rs b/hyperdb-mcp/tests/integration_tests.rs index 1e479df..e7ad64c 100644 --- a/hyperdb-mcp/tests/integration_tests.rs +++ b/hyperdb-mcp/tests/integration_tests.rs @@ -80,6 +80,7 @@ fn full_pipeline_csv_ingest_and_export() { format: "csv".into(), overwrite: true, format_options: None, + source_db: None, }; let result = export_to_file(&te.engine, &export_opts).unwrap(); assert_eq!(result.rows, 2); diff --git a/hyperdb-mcp/tests/per_tool_database_tests.rs b/hyperdb-mcp/tests/per_tool_database_tests.rs index 97a35d4..c2876c8 100644 --- a/hyperdb-mcp/tests/per_tool_database_tests.rs +++ b/hyperdb-mcp/tests/per_tool_database_tests.rs @@ -395,12 +395,475 @@ fn resolve_target_db_persistent_uppercase_errors_in_ephemeral_only() { assert_eq!(err.code, ErrorCode::InvalidArgument); } -// Note on coverage: server-handler-level rejection paths (load_file -// merge+database, load_files+database, watch_directory+database, -// export format=hyper+database, attach-readonly+writable-required) -// are not reachable from this integration test layer without going -// through the rmcp tool router. The compile-time signatures of -// `create_table_async` / `build_parquet_ingest_sql` exercising the -// new `target_db` parameter cover the structural contract; the -// runtime rejection paths are covered by manual smoke tests and -// will land in a follow-up end-to-end MCP test harness. +// --- Cross-database merge (Iter 1) ----------------------------------------- + +/// Merge into persistent when the target table doesn't exist yet: +/// the temp is renamed into the target slot. Verifies the rename path +/// works against a non-primary database. +#[test] +fn merge_into_persistent_creates_table_when_missing() { + let te = TestEngine::new_ephemeral(); + let opts = IngestOptions { + table: "merged_persist".into(), + mode: "merge".into(), + schema_override: None, + merge_key: Some(vec!["id".into()]), + target_db: Some("persistent".into()), + }; + let data = r#"[{"id": 1, "name": "Alice"}, {"id": 2, "name": "Bob"}]"#; + let result = ingest_json(&te.engine, data, &opts).unwrap(); + assert_eq!(result.rows, 2); + assert!( + result.stats.schema_changed, + "newly-created target should report schema_changed=true" + ); + + let rows = te + .engine + .execute_query_to_json( + "SELECT id, name FROM \"persistent\".\"public\".\"merged_persist\" ORDER BY id", + ) + .unwrap(); + assert_eq!(rows.len(), 2); + assert_eq!(rows[0]["name"], "Alice"); +} + +/// Merge into persistent when the target exists with overlapping keys: +/// matching rows are replaced, unmatched rows are appended. +#[test] +fn merge_into_persistent_replaces_matching_and_appends_new() { + let te = TestEngine::new_ephemeral(); + // Seed target with two rows. + let seed_opts = IngestOptions { + table: "merge_target".into(), + mode: "replace".into(), + schema_override: None, + merge_key: None, + target_db: Some("persistent".into()), + }; + ingest_json( + &te.engine, + r#"[{"id": 1, "name": "old1"}, {"id": 2, "name": "old2"}]"#, + &seed_opts, + ) + .unwrap(); + + // Merge: id=1 replaces, id=3 appends, id=2 stays. + let merge_opts = IngestOptions { + table: "merge_target".into(), + mode: "merge".into(), + schema_override: None, + merge_key: Some(vec!["id".into()]), + target_db: Some("persistent".into()), + }; + let result = ingest_json( + &te.engine, + r#"[{"id": 1, "name": "new1"}, {"id": 3, "name": "new3"}]"#, + &merge_opts, + ) + .unwrap(); + assert_eq!(result.rows, 2, "INSERT row count from merge"); + + let rows = te + .engine + .execute_query_to_json( + "SELECT id, name FROM \"persistent\".\"public\".\"merge_target\" ORDER BY id", + ) + .unwrap(); + assert_eq!(rows.len(), 3); + assert_eq!(rows[0]["name"], "new1", "id=1 replaced"); + assert_eq!(rows[1]["name"], "old2", "id=2 untouched"); + assert_eq!(rows[2]["name"], "new3", "id=3 appended"); +} + +/// Merge into persistent that introduces a new column triggers ALTER +/// TABLE ADD COLUMN against the persistent target, and the new column +/// shows up in the post-merge schema with NULL for pre-existing rows. +#[test] +fn merge_into_persistent_alters_when_incoming_has_new_column() { + let te = TestEngine::new_ephemeral(); + let seed_opts = IngestOptions { + table: "widens".into(), + mode: "replace".into(), + schema_override: None, + merge_key: None, + target_db: Some("persistent".into()), + }; + ingest_json(&te.engine, r#"[{"id": 1, "name": "Alice"}]"#, &seed_opts).unwrap(); + + let merge_opts = IngestOptions { + table: "widens".into(), + mode: "merge".into(), + schema_override: None, + merge_key: Some(vec!["id".into()]), + target_db: Some("persistent".into()), + }; + let result = ingest_json( + &te.engine, + r#"[{"id": 2, "name": "Bob", "email": "bob@x.com"}]"#, + &merge_opts, + ) + .unwrap(); + assert!( + result.stats.schema_changed, + "new column 'email' should signal schema_changed" + ); + + let rows = te + .engine + .execute_query_to_json( + "SELECT id, name, email FROM \"persistent\".\"public\".\"widens\" ORDER BY id", + ) + .unwrap(); + assert_eq!(rows.len(), 2); + assert!( + rows[0]["email"].is_null(), + "pre-existing row gets NULL for new column" + ); + assert_eq!(rows[1]["email"], "bob@x.com"); +} + +// --- Per-DB _table_catalog (Iter 4) ---------------------------------------- + +/// `ensure_exists_in(Some(alias))` creates `_table_catalog` inside the +/// user-attached database with the same schema as the persistent +/// catalog. Subsequent ingests can write to it via `upsert_stub_in`. +#[test] +fn ensure_exists_in_seeds_catalog_inside_user_attached_db() { + use hyperdb_mcp::attach::{AttachRegistry, AttachRequest, AttachSource, OnMissing}; + + let dir = TempDir::new().unwrap(); + let primary = dir.path().join("primary.hyper"); + let attached = dir.path().join("attached.hyper"); + + let engine = Engine::new_no_daemon(Some(primary.to_string_lossy().into())).unwrap(); + let reg = AttachRegistry::new(); + reg.attach( + &engine, + AttachRequest { + alias: "user_db".into(), + source: AttachSource::LocalFile { path: attached }, + writable: true, + on_missing: OnMissing::Create, + }, + ) + .unwrap(); + + // Catalog doesn't exist yet in user_db (attach-only doesn't seed + // for an existing file; this attached file was just created so + // attach-with-on_missing=create *does* seed — verify and then + // make a separate user-attach test below for the no-seed case). + // Here we just confirm explicit ensure_exists_in is idempotent. + hyperdb_mcp::table_catalog::ensure_exists_in(&engine, Some("user_db")).unwrap(); + hyperdb_mcp::table_catalog::ensure_exists_in(&engine, Some("user_db")).unwrap(); + + // Probe via qualified pg_tables. + let rows = engine + .execute_query_to_json( + "SELECT tablename FROM \"user_db\".pg_catalog.pg_tables \ + WHERE schemaname = 'public' AND tablename = '_table_catalog'", + ) + .unwrap(); + assert_eq!(rows.len(), 1, "_table_catalog must exist in user_db"); +} + +/// `upsert_stub_in(target_db=Some("user_db"))` writes the row into the +/// user-attached database's catalog, NOT into the persistent catalog. +#[test] +fn upsert_stub_in_routes_to_user_attached_db() { + use hyperdb_mcp::attach::{AttachRegistry, AttachRequest, AttachSource, OnMissing}; + + let dir = TempDir::new().unwrap(); + let primary = dir.path().join("primary.hyper"); + let attached = dir.path().join("attached.hyper"); + + let engine = Engine::new_no_daemon(Some(primary.to_string_lossy().into())).unwrap(); + let reg = AttachRegistry::new(); + reg.attach( + &engine, + AttachRequest { + alias: "user_db".into(), + source: AttachSource::LocalFile { path: attached }, + writable: true, + on_missing: OnMissing::Create, + }, + ) + .unwrap(); + + // Create a real table in user_db so the catalog row points to + // something that exists. + engine + .execute_command("CREATE TABLE \"user_db\".\"public\".\"my_data\" (id INT)") + .unwrap(); + + hyperdb_mcp::table_catalog::upsert_stub_in( + &engine, + "my_data", + "load_data", + Some("{\"format\":\"json\"}"), + Some(42), + true, + Some("user_db"), + ) + .unwrap(); + + // Visible via fully-qualified SELECT into user_db's catalog. + let rows = engine + .execute_query_to_json( + "SELECT table_name, load_tool, row_count \ + FROM \"user_db\".\"public\".\"_table_catalog\" \ + WHERE table_name = 'my_data'", + ) + .unwrap(); + assert_eq!(rows.len(), 1); + assert_eq!(rows[0]["load_tool"], "load_data"); + assert_eq!(rows[0]["row_count"], 42); + + // NOT in the persistent catalog. + let persistent_rows = engine + .execute_query_to_json( + "SELECT table_name FROM \"persistent\".\"public\".\"_table_catalog\" \ + WHERE table_name = 'my_data'", + ) + .unwrap_or_default(); + assert!( + persistent_rows.is_empty(), + "the row must NOT bleed into persistent's catalog" + ); +} + +/// `get_in(target_db=Some("user_db"))` reads from the user-attached +/// database's catalog and returns the row written by `upsert_stub_in`. +#[test] +fn get_in_reads_from_user_attached_db_catalog() { + use hyperdb_mcp::attach::{AttachRegistry, AttachRequest, AttachSource, OnMissing}; + + let dir = TempDir::new().unwrap(); + let primary = dir.path().join("primary.hyper"); + let attached = dir.path().join("attached.hyper"); + + let engine = Engine::new_no_daemon(Some(primary.to_string_lossy().into())).unwrap(); + let reg = AttachRegistry::new(); + reg.attach( + &engine, + AttachRequest { + alias: "user_db".into(), + source: AttachSource::LocalFile { path: attached }, + writable: true, + on_missing: OnMissing::Create, + }, + ) + .unwrap(); + + engine + .execute_command("CREATE TABLE \"user_db\".\"public\".\"events\" (id INT)") + .unwrap(); + hyperdb_mcp::table_catalog::upsert_stub_in( + &engine, + "events", + "load_file", + None, + Some(7), + true, + Some("user_db"), + ) + .unwrap(); + + let entry = hyperdb_mcp::table_catalog::get_in(&engine, "events", Some("user_db")) + .unwrap() + .expect("get_in must find the row written via upsert_stub_in"); + assert_eq!(entry.table_name, "events"); + assert_eq!(entry.row_count, Some(7)); + // Reading the persistent catalog for the same table_name returns + // None — they are independent catalogs. + let persistent_entry = hyperdb_mcp::table_catalog::get_in(&engine, "events", None).unwrap(); + assert!(persistent_entry.is_none()); +} + +/// `reconcile_in(target_db=Some("user_db"))` operates only on the +/// user-attached database's tables, leaving persistent's catalog +/// untouched. +#[test] +fn reconcile_in_per_db_does_not_touch_persistent_catalog() { + use hyperdb_mcp::attach::{AttachRegistry, AttachRequest, AttachSource, OnMissing}; + + let dir = TempDir::new().unwrap(); + let primary = dir.path().join("primary.hyper"); + let attached = dir.path().join("attached.hyper"); + + let engine = Engine::new_no_daemon(Some(primary.to_string_lossy().into())).unwrap(); + let reg = AttachRegistry::new(); + reg.attach( + &engine, + AttachRequest { + alias: "user_db".into(), + source: AttachSource::LocalFile { path: attached }, + writable: true, + on_missing: OnMissing::Create, + }, + ) + .unwrap(); + + // Two tables in persistent (existing behavior). + engine + .execute_command("CREATE TABLE \"persistent\".\"public\".\"persist_t\" (n INT)") + .unwrap(); + // One table in user_db. + engine + .execute_command("CREATE TABLE \"user_db\".\"public\".\"user_t\" (n INT)") + .unwrap(); + + hyperdb_mcp::table_catalog::reconcile_in(&engine, Some("user_db")).unwrap(); + + let user_rows = hyperdb_mcp::table_catalog::list_in(&engine, Some("user_db")).unwrap(); + let user_names: Vec = user_rows.iter().map(|e| e.table_name.clone()).collect(); + assert!( + user_names.contains(&"user_t".into()), + "user_db reconcile must stub user_t; got {user_names:?}" + ); + + let persistent_rows = hyperdb_mcp::table_catalog::list_in(&engine, None).unwrap(); + let persistent_names: Vec = persistent_rows + .iter() + .map(|e| e.table_name.clone()) + .collect(); + assert!( + !persistent_names.contains(&"user_t".into()), + "user_t must not appear in persistent's catalog; got {persistent_names:?}" + ); +} + +// --- set_table_metadata + database (Iter 5) -------------------------------- + +/// `set_metadata_in(target_db=Some("user_db"))` updates prose fields +/// in the user-attached database's catalog after a stub row exists +/// there (the typical flow: load_data into the user DB seeds the +/// catalog automatically; set_table_metadata then updates prose). +#[test] +fn set_metadata_in_routes_to_user_attached_db() { + use hyperdb_mcp::attach::{AttachRegistry, AttachRequest, AttachSource, OnMissing}; + use hyperdb_mcp::table_catalog::MetadataFields; + + let dir = TempDir::new().unwrap(); + let primary = dir.path().join("primary.hyper"); + let attached = dir.path().join("attached.hyper"); + + let engine = Engine::new_no_daemon(Some(primary.to_string_lossy().into())).unwrap(); + let reg = AttachRegistry::new(); + reg.attach( + &engine, + AttachRequest { + alias: "user_db".into(), + source: AttachSource::LocalFile { path: attached }, + writable: true, + on_missing: OnMissing::Create, + }, + ) + .unwrap(); + + // Stub a row first (set_metadata requires an existing entry). + engine + .execute_command("CREATE TABLE \"user_db\".\"public\".\"events\" (id INT)") + .unwrap(); + hyperdb_mcp::table_catalog::upsert_stub_in( + &engine, + "events", + "load_data", + None, + Some(0), + true, + Some("user_db"), + ) + .unwrap(); + + // Update prose via set_metadata_in. + let fields = MetadataFields { + source_url: Some("s3://bucket/events.parquet".into()), + source_description: Some("Tracking events".into()), + purpose: Some("daily reports".into()), + license: None, + notes: None, + }; + let entry = + hyperdb_mcp::table_catalog::set_metadata_in(&engine, "events", &fields, Some("user_db")) + .unwrap(); + assert_eq!( + entry.source_url.as_deref(), + Some("s3://bucket/events.parquet") + ); + assert_eq!(entry.purpose.as_deref(), Some("daily reports")); + + // Visible in the user_db catalog. + let rows = engine + .execute_query_to_json( + "SELECT source_url, purpose FROM \"user_db\".\"public\".\"_table_catalog\" \ + WHERE table_name = 'events'", + ) + .unwrap(); + assert_eq!(rows.len(), 1); + assert_eq!(rows[0]["source_url"], "s3://bucket/events.parquet"); + assert_eq!(rows[0]["purpose"], "daily reports"); +} + +/// `set_metadata_in` errors with TableNotFound + a target-DB-named +/// message when the row is in a different DB than the caller asked +/// for. Specifically: row in persistent's catalog, ask user_db. +#[test] +fn set_metadata_in_missing_row_names_target_database_in_error() { + use hyperdb_mcp::attach::{AttachRegistry, AttachRequest, AttachSource, OnMissing}; + use hyperdb_mcp::table_catalog::MetadataFields; + + let dir = TempDir::new().unwrap(); + let primary = dir.path().join("primary.hyper"); + let attached = dir.path().join("attached.hyper"); + + let engine = Engine::new_no_daemon(Some(primary.to_string_lossy().into())).unwrap(); + let reg = AttachRegistry::new(); + reg.attach( + &engine, + AttachRequest { + alias: "user_db".into(), + source: AttachSource::LocalFile { path: attached }, + writable: true, + on_missing: OnMissing::Create, + }, + ) + .unwrap(); + + // Stub in PERSISTENT, not user_db. + engine + .execute_command("CREATE TABLE \"persistent\".\"public\".\"per_t\" (id INT)") + .unwrap(); + hyperdb_mcp::table_catalog::upsert_stub_in( + &engine, + "per_t", + "load_data", + None, + Some(0), + true, + None, + ) + .unwrap(); + + let fields = MetadataFields { + source_url: Some("anything".into()), + ..Default::default() + }; + let err = + hyperdb_mcp::table_catalog::set_metadata_in(&engine, "per_t", &fields, Some("user_db")) + .expect_err("must error: row exists in persistent, not user_db"); + assert_eq!(err.code, ErrorCode::TableNotFound); + assert!( + err.message.contains("user_db"), + "error must name the target db; got: {}", + err.message + ); +} + +// Note on coverage: server-handler-level rejection paths +// (load_files+database, watch_directory+database, export +// format=hyper+database, attach-readonly+writable-required, +// set_table_metadata.database with read-only target) are not +// reachable from this integration test layer without going through +// the rmcp tool router. They land in the end-to-end MCP test harness +// (Iter 6). diff --git a/hyperdb-mcp/tests/two_db_model_tests.rs b/hyperdb-mcp/tests/two_db_model_tests.rs index 7dec4f7..2ae4958 100644 --- a/hyperdb-mcp/tests/two_db_model_tests.rs +++ b/hyperdb-mcp/tests/two_db_model_tests.rs @@ -178,10 +178,10 @@ fn engine_status_ephemeral_only_reports_no_persistent() { assert!(status["persistent_path"].is_null()); } -/// `catalog_present_in_persistent` caches the probe result so the -/// underlying SQL only runs once per engine lifetime. +/// `catalog_present_in` caches per-DB probe results so subsequent +/// reads/writes against the same alias don't re-run the probe. #[test] -fn catalog_presence_probe_is_cached() { +fn catalog_presence_probe_is_cached_per_db() { let dir = TempDir::new().unwrap(); let path = dir.path().join("ws.hyper"); let engine = Engine::new(Some(path.to_str().unwrap().into())).unwrap(); @@ -193,30 +193,48 @@ fn catalog_presence_probe_is_cached() { }; // First call runs the probe. - assert!(!engine.catalog_present_in_persistent(probe).unwrap()); + assert!(!engine.catalog_present_in("persistent", probe).unwrap()); assert_eq!(probe_count.load(std::sync::atomic::Ordering::SeqCst), 1); // Second call uses the cache; probe count stays at 1. - assert!(!engine.catalog_present_in_persistent(probe).unwrap()); + assert!(!engine.catalog_present_in("persistent", probe).unwrap()); assert_eq!(probe_count.load(std::sync::atomic::Ordering::SeqCst), 1); - // After mark_catalog_present, the cached value flips to true and - // the probe stays untouched. - engine.mark_catalog_present(); - assert!(engine.catalog_present_in_persistent(probe).unwrap()); + // After mark_catalog_present_for, the cached value flips to true + // and the probe stays untouched. + engine.mark_catalog_present_for("persistent"); + assert!(engine.catalog_present_in("persistent", probe).unwrap()); assert_eq!(probe_count.load(std::sync::atomic::Ordering::SeqCst), 1); + + // A different alias has its own cache entry — uncached, so the + // probe runs again. + assert!(!engine.catalog_present_in("other", probe).unwrap()); + assert_eq!(probe_count.load(std::sync::atomic::Ordering::SeqCst), 2); } -/// In ephemeral-only mode, the cache short-circuits to `false` without -/// running the probe at all. +/// `clear_catalog_cache_for` drops the entry so the next read reruns +/// the probe — used by `detach_database` to guard against re-attach +/// to a different file. #[test] -fn catalog_presence_short_circuits_in_ephemeral_only() { - let engine = Engine::new(None).unwrap(); +fn clear_catalog_cache_for_invalidates_alias_entry() { + let dir = TempDir::new().unwrap(); + let path = dir.path().join("ws.hyper"); + let engine = Engine::new(Some(path.to_str().unwrap().into())).unwrap(); + let probe_count = std::sync::atomic::AtomicUsize::new(0); let probe = |_e: &Engine| -> Result { probe_count.fetch_add(1, std::sync::atomic::Ordering::SeqCst); Ok(true) }; - assert!(!engine.catalog_present_in_persistent(probe).unwrap()); - assert_eq!(probe_count.load(std::sync::atomic::Ordering::SeqCst), 0); + + assert!(engine.catalog_present_in("foo", probe).unwrap()); + assert_eq!(probe_count.load(std::sync::atomic::Ordering::SeqCst), 1); + // Cached. + assert!(engine.catalog_present_in("foo", probe).unwrap()); + assert_eq!(probe_count.load(std::sync::atomic::Ordering::SeqCst), 1); + + // Clear, next probe runs again. + engine.clear_catalog_cache_for("foo"); + assert!(engine.catalog_present_in("foo", probe).unwrap()); + assert_eq!(probe_count.load(std::sync::atomic::Ordering::SeqCst), 2); } diff --git a/hyperdb-mcp/tests/watcher_tests.rs b/hyperdb-mcp/tests/watcher_tests.rs index 3f881dd..1b0c9e2 100644 --- a/hyperdb-mcp/tests/watcher_tests.rs +++ b/hyperdb-mcp/tests/watcher_tests.rs @@ -17,6 +17,7 @@ mod common; use common::TestEngine; +use hyperdb_mcp::attach::AttachRegistry; use hyperdb_mcp::engine::Engine; use hyperdb_mcp::watcher::{self, WatchOptions, WatcherRegistry}; use std::fmt::Write as _; @@ -64,10 +65,12 @@ async fn watcher_ingests_csv_and_cleans_up() { watcher::start_watching( Arc::clone(&engine), + Arc::new(AttachRegistry::new()), Arc::clone(®istry), None, watch_dir.path().to_path_buf(), "events".into(), + None, WatchOptions::default(), ) .unwrap(); @@ -116,10 +119,12 @@ async fn watcher_moves_bad_files_to_failed() { watcher::start_watching( Arc::clone(&engine), + Arc::new(AttachRegistry::new()), Arc::clone(®istry), None, watch_dir.path().to_path_buf(), "evts".into(), + None, WatchOptions::default(), ) .unwrap(); @@ -158,10 +163,12 @@ async fn watcher_sweep_picks_up_preexisting_files() { let initial = watcher::start_watching( Arc::clone(&engine), + Arc::new(AttachRegistry::new()), Arc::clone(®istry), None, watch_dir.path().to_path_buf(), "t".into(), + None, WatchOptions::default(), ) .unwrap(); @@ -189,10 +196,12 @@ async fn unwatch_removes_from_registry() { watcher::start_watching( Arc::clone(&engine), + Arc::new(AttachRegistry::new()), Arc::clone(®istry), None, watch_dir.path().to_path_buf(), "logs".into(), + None, WatchOptions::default(), ) .unwrap(); @@ -214,20 +223,24 @@ async fn watch_same_directory_twice_fails() { watcher::start_watching( Arc::clone(&engine), + Arc::new(AttachRegistry::new()), Arc::clone(®istry), None, watch_dir.path().to_path_buf(), "t1".into(), + None, WatchOptions::default(), ) .unwrap(); let err = watcher::start_watching( Arc::clone(&engine), + Arc::new(AttachRegistry::new()), Arc::clone(®istry), None, watch_dir.path().to_path_buf(), "t2".into(), + None, WatchOptions::default(), ) .unwrap_err(); @@ -273,10 +286,12 @@ async fn watcher_ingests_many_files_concurrently() { let initial = watcher::start_watching( Arc::clone(&engine), + Arc::new(AttachRegistry::new()), Arc::clone(®istry), None, watch_dir.path().to_path_buf(), "batches".into(), + None, WatchOptions { max_concurrent: 4 }, ) .unwrap(); @@ -332,3 +347,88 @@ async fn watcher_ingests_many_files_concurrently() { .unwrap(); assert_eq!(total, (FILES * ROWS_PER_FILE) as i64); } + +/// Iter 3: a watcher with `target_db = Some("persistent")` opens the +/// persistent file as its pool workspace and ingests rows there +/// instead of into primary. Verifies the pool resolution path picks +/// the right .hyper file. +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn watcher_ingests_into_persistent_target() { + let (engine, _td) = engine_handle(TestEngine::new_ephemeral()); + + // Pre-create the target table in persistent so the watcher's + // append mode has somewhere to land. The current watcher contract + // is "append into an existing table" — auto-create is not part of + // this iteration's scope. + { + let guard = engine.lock().unwrap(); + guard + .as_ref() + .unwrap() + .execute_command( + "CREATE TABLE \"persistent\".\"public\".\"events\" (id INT, name TEXT)", + ) + .unwrap(); + } + + let watch_dir = tempfile::TempDir::new().unwrap(); + let registry = Arc::new(WatcherRegistry::new()); + + watcher::start_watching( + Arc::clone(&engine), + Arc::new(AttachRegistry::new()), + Arc::clone(®istry), + None, + watch_dir.path().to_path_buf(), + "events".into(), + Some("persistent".into()), + WatchOptions::default(), + ) + .unwrap(); + + drop_ready_pair(watch_dir.path(), "p.csv", b"id,name\n1,Alice\n2,Bob\n"); + + let canon = watch_dir.path().canonicalize().unwrap(); + let ready = canon.join("p.csv.ready"); + let data = canon.join("p.csv"); + assert!( + wait_until( + || !ready.exists() && !data.exists(), + Duration::from_secs(10) + ), + "watcher did not finish ingesting persistent target within 10s" + ); + + // Rows must be visible in the persistent attachment, not primary. + let count: i64 = engine + .lock() + .unwrap() + .as_ref() + .unwrap() + .connection() + .execute_scalar_query("SELECT COUNT(*) FROM \"persistent\".\"public\".\"events\"") + .unwrap() + .unwrap(); + assert_eq!(count, 2); + + // The watcher must also stamp persistent's _table_catalog after the + // ingest. Without this row, set_table_metadata against the table + // would error confusingly even though the table exists. This is + // the C1 fix from the post-merge architectural review. + let catalog_rows: i64 = engine + .lock() + .unwrap() + .as_ref() + .unwrap() + .connection() + .execute_scalar_query( + "SELECT COUNT(*) FROM \"persistent\".\"public\".\"_table_catalog\" \ + WHERE table_name = 'events' AND load_tool = 'watch_directory'", + ) + .unwrap() + .unwrap(); + assert_eq!( + catalog_rows, 1, + "watcher must stamp _table_catalog with load_tool='watch_directory'" + ); +}