From 446c24df53d2706cb5fbb783f5bff26cc00f5914 Mon Sep 17 00:00:00 2001 From: therecluse26 Date: Sun, 19 Jul 2026 22:09:56 -0400 Subject: [PATCH 1/4] feat: add DuckDB engine with full feature coverage (REF-290) - New src/engine/duckdb module implementing DatabaseEngine: validate_connection, introspect (databases/schemas/tables/views/ indexes/table details incl. comments + row estimates via duckdb_* catalog functions), and capability-enforced execute - File-backed connections open with AccessMode::ReadOnly (defense in depth); :memory: opens read-write with the capability parser as the enforcement boundary - Rich type mapping: HUGEINT/DECIMAL stringified to preserve precision, DATE/TIME/TIMESTAMP as ISO-8601, BLOB as Base64, LIST/STRUCT/MAP/ ENUM/UNION converted recursively to JSON - Statement timeouts via InterruptHandle + timer thread -> QUERY_TIMEOUT - Structured EXPLAIN via EXPLAIN (FORMAT JSON) normalized to ExplainPlanNode - Read-only validator for the DuckDB dialect: SELECT/CTE/SHOW/DESCRIBE/ SUMMARIZE/EXPLAIN/transaction control plus a PRAGMA allowlist; COPY/ATTACH/INSTALL/LOAD/EXPORT and all DML/DDL rejected - duckdb: DSN scheme (same path forms as sqlite:) - CLI, MCP server, and schema-diff dispatch wired for the new engine Co-Authored-By: Claude Fable 5 Co-Authored-By: Paperclip --- Cargo.toml | 4 +- src/capability/mod.rs | 204 ++++++ src/diff.rs | 9 + src/dsn.rs | 78 +- src/engine/duckdb/mod.rs | 1498 ++++++++++++++++++++++++++++++++++++++ src/engine/mod.rs | 28 + src/main.rs | 72 +- src/mcp.rs | 60 +- 8 files changed, 1933 insertions(+), 20 deletions(-) create mode 100644 src/engine/duckdb/mod.rs diff --git a/Cargo.toml b/Cargo.toml index 719be21..1d18ce4 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -61,6 +61,7 @@ postgres-native-tls = { version = "0.5", optional = true } # TLS for tokio-post native-tls = { version = "0.2", optional = true, features = ["vendored"] } # Platform TLS (OpenSSL/Secure Transport/SChannel) mysql_async = { version = "0.34", default-features = false, features = ["default-rustls"], optional = true } # MySQL native async driver rusqlite = { version = "0.32", features = ["bundled"], optional = true } # SQLite native driver with bundled lib +duckdb = { version = "1.10504.0", features = ["bundled"], optional = true } # DuckDB native driver with bundled lib # BLOB encoding (Base64) - used by SQLite and PostgreSQL base64 = "0.22" @@ -74,7 +75,8 @@ default = ["all-engines"] # Enable all database engines by default postgres = ["dep:tokio-postgres", "dep:chrono", "dep:uuid", "dep:postgres-native-tls", "dep:native-tls"] mysql = ["dep:mysql_async"] sqlite = ["dep:rusqlite"] -all-engines = ["postgres", "mysql", "sqlite"] +duckdb = ["dep:duckdb", "dep:chrono"] +all-engines = ["postgres", "mysql", "sqlite", "duckdb"] [dev-dependencies] # Testing utilities diff --git a/src/capability/mod.rs b/src/capability/mod.rs index 25d52a1..c68bce6 100644 --- a/src/capability/mod.rs +++ b/src/capability/mod.rs @@ -122,6 +122,7 @@ fn is_read_only(sql: &str, engine: DatabaseType) -> bool { DatabaseType::Postgres => is_read_only_postgres(sql), DatabaseType::MySQL => is_read_only_mysql(sql), DatabaseType::SQLite => is_read_only_sqlite(sql), + DatabaseType::DuckDB => is_read_only_duckdb(sql), } } @@ -513,6 +514,102 @@ fn is_read_only_sqlite(sql: &str) -> bool { || sql.starts_with("RELEASE") } +/// `DuckDB` PRAGMAs that are read-only when invoked in argument form +/// (`PRAGMA name(arg)`). These treat the argument as a query parameter rather +/// than a setter value, so the parenthesized form does not write state. +/// +/// Names are uppercase to match the preprocessed SQL produced by +/// `preprocess_sql`. +const READ_ONLY_DUCKDB_PRAGMAS_WITH_ARGS: &[&str] = + &["TABLE_INFO", "STORAGE_INFO", "SHOW", "DATABASE_SIZE"]; + +/// `DuckDB` PRAGMAs whose bare form (`PRAGMA name`) is a pure read. +/// +/// `DuckDB` setter PRAGMAs (`memory_limit`, `threads`, `enable_progress_bar`, +/// …) use the `= value` assignment form, which is rejected unconditionally by +/// `is_safe_duckdb_pragma`. Only pure introspection names are admitted here. +const READ_ONLY_DUCKDB_PRAGMAS_BARE: &[&str] = &[ + "DATABASE_LIST", + "DATABASE_SIZE", + "SHOW_TABLES", + "SHOW_TABLES_EXPANDED", + "SHOW_DATABASES", + "FUNCTIONS", + "COLLATIONS", + "VERSION", + "PLATFORM", + "USER_AGENT", + "METADATA_INFO", +]; + +/// Validate a `PRAGMA …` statement against the `DuckDB` read-only allowlist. +/// +/// Same shape as the `SQLite` PRAGMA guard: the `= value` setter form is +/// always rejected, the argument form is allowed only for names in +/// [`READ_ONLY_DUCKDB_PRAGMAS_WITH_ARGS`], and the bare form only for names +/// in [`READ_ONLY_DUCKDB_PRAGMAS_BARE`]. +fn is_safe_duckdb_pragma(sql: &str) -> bool { + let Some(rest) = sql.strip_prefix("PRAGMA ") else { + return false; + }; + let rest = rest.trim().trim_end_matches(';').trim(); + if rest.is_empty() { + return false; + } + + // `=` always indicates the assignment / setter form (e.g. + // `PRAGMA memory_limit='1GB'`, `PRAGMA threads=4`). Reject unconditionally. + if rest.contains('=') { + return false; + } + + let (name, has_args) = match rest.find('(') { + Some(open) => { + if !rest.ends_with(')') { + return false; + } + (rest[..open].trim(), true) + } + None => (rest, false), + }; + + if name.is_empty() || name.contains(|c: char| c.is_whitespace()) { + return false; + } + + if has_args { + READ_ONLY_DUCKDB_PRAGMAS_WITH_ARGS.contains(&name) + } else { + READ_ONLY_DUCKDB_PRAGMAS_BARE.contains(&name) + } +} + +// DuckDB read-only check +fn is_read_only_duckdb(sql: &str) -> bool { + // Strip EXPLAIN prefix (DuckDB supports EXPLAIN and EXPLAIN ANALYZE) + let sql = strip_explain_prefix(sql); + let sql = sql.trim(); + + if sql.starts_with("SELECT ") { + // Guard against `SELECT ... INTO new_table` (DuckDB supports the + // CREATE-TABLE-AS shorthand) with the same quote-aware keyword scan + // used for Postgres (REF-42) and WITH-CTE queries (REF-41). + return scan_for_write_keyword(sql).is_none(); + } + + is_safe_cte_query(sql) + || is_safe_duckdb_pragma(sql) + || sql.starts_with("SHOW ") + || sql.starts_with("DESCRIBE ") + || sql.starts_with("DESC ") + || sql.starts_with("SUMMARIZE ") + || sql.starts_with("BEGIN") + || sql.starts_with("COMMIT") + || sql.starts_with("ROLLBACK") + || sql.starts_with("SAVEPOINT") + || sql.starts_with("RELEASE") +} + #[cfg(test)] mod tests { use super::*; @@ -1541,4 +1638,111 @@ mod tests { fn test_sqlite_explain_query_plan_delete_rejected() { assert_rejected("EXPLAIN QUERY PLAN DELETE FROM items", DatabaseType::SQLite); } + + // ========================================================================= + // DuckDB dialect (REF-290) + // ========================================================================= + + #[test] + fn test_duckdb_select_allowed() { + assert_allowed("SELECT * FROM users", DatabaseType::DuckDB); + assert_allowed("SELECT count(*) FROM range(10)", DatabaseType::DuckDB); + } + + #[test] + fn test_duckdb_cte_select_allowed() { + assert_allowed("WITH t AS (SELECT 1 AS x) SELECT x FROM t", DatabaseType::DuckDB); + } + + #[test] + fn test_duckdb_show_describe_summarize_allowed() { + assert_allowed("SHOW TABLES", DatabaseType::DuckDB); + assert_allowed("DESCRIBE users", DatabaseType::DuckDB); + assert_allowed("DESC users", DatabaseType::DuckDB); + assert_allowed("SUMMARIZE users", DatabaseType::DuckDB); + } + + #[test] + fn test_duckdb_explain_allowed() { + assert_allowed("EXPLAIN SELECT * FROM users", DatabaseType::DuckDB); + assert_allowed("EXPLAIN ANALYZE SELECT * FROM users", DatabaseType::DuckDB); + assert_allowed("EXPLAIN (FORMAT JSON) SELECT * FROM users", DatabaseType::DuckDB); + } + + #[test] + fn test_duckdb_explain_hidden_write_rejected() { + assert_rejected("EXPLAIN DELETE FROM users", DatabaseType::DuckDB); + assert_rejected("EXPLAIN (FORMAT JSON) INSERT INTO t VALUES (1)", DatabaseType::DuckDB); + } + + #[test] + fn test_duckdb_transaction_control_allowed() { + assert_allowed("BEGIN", DatabaseType::DuckDB); + assert_allowed("BEGIN TRANSACTION", DatabaseType::DuckDB); + assert_allowed("COMMIT", DatabaseType::DuckDB); + assert_allowed("ROLLBACK", DatabaseType::DuckDB); + } + + #[test] + fn test_duckdb_writes_rejected() { + assert_rejected("INSERT INTO users VALUES (1)", DatabaseType::DuckDB); + assert_rejected("UPDATE users SET name = 'x'", DatabaseType::DuckDB); + assert_rejected("DELETE FROM users", DatabaseType::DuckDB); + assert_rejected("CREATE TABLE t (id INTEGER)", DatabaseType::DuckDB); + assert_rejected("DROP TABLE users", DatabaseType::DuckDB); + assert_rejected("ALTER TABLE users ADD COLUMN c INTEGER", DatabaseType::DuckDB); + assert_rejected("TRUNCATE users", DatabaseType::DuckDB); + } + + #[test] + fn test_duckdb_file_and_extension_operations_rejected() { + // COPY writes to the filesystem; ATTACH/DETACH mutate catalog state; + // INSTALL/LOAD pull in extensions. + assert_rejected("COPY users TO 'out.csv'", DatabaseType::DuckDB); + assert_rejected("ATTACH 'other.duckdb' AS other", DatabaseType::DuckDB); + assert_rejected("DETACH other", DatabaseType::DuckDB); + assert_rejected("LOAD httpfs", DatabaseType::DuckDB); + assert_rejected("INSTALL httpfs", DatabaseType::DuckDB); + } + + #[test] + fn test_duckdb_select_into_rejected() { + // CREATE-TABLE-AS shorthand hidden behind a SELECT prefix + assert_rejected("SELECT * INTO new_table FROM users", DatabaseType::DuckDB); + } + + #[test] + fn test_duckdb_cte_hidden_write_rejected() { + assert_rejected( + "WITH t AS (SELECT 1) INSERT INTO users SELECT * FROM t", + DatabaseType::DuckDB, + ); + assert_rejected("WITH t AS (SELECT 1) DELETE FROM users", DatabaseType::DuckDB); + } + + #[test] + fn test_duckdb_read_only_pragmas_allowed() { + assert_allowed("PRAGMA database_list", DatabaseType::DuckDB); + assert_allowed("PRAGMA show_tables", DatabaseType::DuckDB); + assert_allowed("PRAGMA version", DatabaseType::DuckDB); + assert_allowed("PRAGMA table_info('users')", DatabaseType::DuckDB); + assert_allowed("PRAGMA storage_info('users')", DatabaseType::DuckDB); + assert_allowed("PRAGMA database_size", DatabaseType::DuckDB); + } + + #[test] + fn test_duckdb_setter_pragmas_rejected() { + assert_rejected("PRAGMA memory_limit='1GB'", DatabaseType::DuckDB); + assert_rejected("PRAGMA threads=4", DatabaseType::DuckDB); + assert_rejected("PRAGMA enable_progress_bar", DatabaseType::DuckDB); + assert_rejected("PRAGMA enable_profiling", DatabaseType::DuckDB); + assert_rejected("PRAGMA disable_progress_bar", DatabaseType::DuckDB); + } + + #[test] + fn test_duckdb_multi_statement_rejected() { + let caps = Capabilities::default(); + let result = validate_query("SELECT 1; DROP TABLE users", &caps, DatabaseType::DuckDB); + assert!(result.is_err()); + } } diff --git a/src/diff.rs b/src/diff.rs index 20085a2..9acf39a 100644 --- a/src/diff.rs +++ b/src/diff.rs @@ -13,6 +13,8 @@ use crate::engine::{ }; use crate::error::{PlenumError, Result}; +#[cfg(feature = "duckdb")] +use crate::engine::duckdb::DuckDbEngine; #[cfg(feature = "mysql")] use crate::engine::mysql::MySqlEngine; #[cfg(feature = "postgres")] @@ -50,6 +52,13 @@ async fn engine_introspect( DatabaseType::MySQL => Err(PlenumError::invalid_input( "MySQL engine not enabled. Build with --features mysql.", )), + + #[cfg(feature = "duckdb")] + DatabaseType::DuckDB => DuckDbEngine::introspect(config, operation, database, schema).await, + #[cfg(not(feature = "duckdb"))] + DatabaseType::DuckDB => Err(PlenumError::invalid_input( + "DuckDB engine not enabled. Build with --features duckdb.", + )), } } diff --git a/src/dsn.rs b/src/dsn.rs index e39051d..3554e7f 100644 --- a/src/dsn.rs +++ b/src/dsn.rs @@ -16,6 +16,7 @@ use crate::error::{PlenumError, Result}; /// - `postgres://…` or `postgresql://…` → `PostgreSQL` /// - `mysql://…` → `MySQL` /// - `sqlite:…` (various path forms) → `SQLite` +/// - `duckdb:…` (various path forms) → `DuckDB` /// /// Engine is inferred from the scheme. Error messages never echo credentials. /// Use [`redact_dsn`] when including the original DSN string in any output. @@ -26,9 +27,11 @@ pub fn parse_dsn(dsn: &str) -> Result { parse_mysql_dsn(dsn) } else if dsn.starts_with("sqlite:") { parse_sqlite_dsn(dsn) + } else if dsn.starts_with("duckdb:") { + parse_duckdb_dsn(dsn) } else { Err(PlenumError::invalid_input( - "Unrecognized DSN scheme. Use postgres://, postgresql://, mysql://, or sqlite: prefix", + "Unrecognized DSN scheme. Use postgres://, postgresql://, mysql://, sqlite:, or duckdb: prefix", )) } } @@ -120,6 +123,33 @@ fn parse_sqlite_dsn(dsn: &str) -> Result { Ok(ConnectionConfig::sqlite(PathBuf::from(path_str))) } +fn parse_duckdb_dsn(dsn: &str) -> Result { + // Same pragmatic path forms as the sqlite: scheme: + // duckdb:///absolute/path.duckdb → /absolute/path.duckdb + // duckdb://relative/path.duckdb → relative/path.duckdb + // duckdb:/absolute/path.duckdb → /absolute/path.duckdb + // duckdb:relative/path.duckdb → relative/path.duckdb + // duckdb::memory: → :memory: (DuckDB in-memory database) + + let path_str: String = if let Some(p) = dsn.strip_prefix("duckdb:///") { + format!("/{p}") + } else if let Some(p) = dsn.strip_prefix("duckdb://") { + p.to_string() + } else if let Some(p) = dsn.strip_prefix("duckdb:/") { + format!("/{p}") + } else { + dsn.strip_prefix("duckdb:").expect("caller verified prefix").to_string() + }; + + if path_str.is_empty() { + return Err(PlenumError::invalid_input( + "DuckDB DSN missing file path (e.g. duckdb:///path/to/db.duckdb or duckdb::memory:)", + )); + } + + Ok(ConnectionConfig::duckdb(PathBuf::from(path_str))) +} + // ============================================================================ // Internal helpers // ============================================================================ @@ -370,6 +400,52 @@ mod tests { assert!(err.message().contains("missing file path"), "got: {}", err.message()); } + // ─── DuckDB ──────────────────────────────────────────────────────────────── + + #[test] + fn duckdb_triple_slash_absolute() { + let cfg = parse_dsn("duckdb:///tmp/test.duckdb").unwrap(); + assert_eq!(cfg.engine, DatabaseType::DuckDB); + assert_eq!(cfg.file, Some(PathBuf::from("/tmp/test.duckdb"))); + } + + #[test] + fn duckdb_single_slash_absolute() { + let cfg = parse_dsn("duckdb:/tmp/test.duckdb").unwrap(); + assert_eq!(cfg.file, Some(PathBuf::from("/tmp/test.duckdb"))); + } + + #[test] + fn duckdb_relative_path() { + let cfg = parse_dsn("duckdb:relative/path.duckdb").unwrap(); + assert_eq!(cfg.file, Some(PathBuf::from("relative/path.duckdb"))); + } + + #[test] + fn duckdb_double_slash_path() { + let cfg = parse_dsn("duckdb://./local.duckdb").unwrap(); + assert_eq!(cfg.file, Some(PathBuf::from("./local.duckdb"))); + } + + #[test] + fn duckdb_memory() { + let cfg = parse_dsn("duckdb::memory:").unwrap(); + assert_eq!(cfg.engine, DatabaseType::DuckDB); + assert_eq!(cfg.file, Some(PathBuf::from(":memory:"))); + } + + #[test] + fn duckdb_empty_path_fails() { + let err = parse_dsn("duckdb:").unwrap_err(); + assert!(err.message().contains("missing file path"), "got: {}", err.message()); + } + + #[test] + fn redact_duckdb_no_op() { + let dsn = "duckdb:///tmp/test.duckdb"; + assert_eq!(redact_dsn(dsn), dsn); + } + // ─── Redaction ───────────────────────────────────────────────────────────── #[test] diff --git a/src/engine/duckdb/mod.rs b/src/engine/duckdb/mod.rs new file mode 100644 index 0000000..47a6569 --- /dev/null +++ b/src/engine/duckdb/mod.rs @@ -0,0 +1,1498 @@ +//! `DuckDB` Database Engine Implementation +//! +//! This module implements the `DatabaseEngine` trait for `DuckDB` databases. +//! +//! # Features +//! - File-based connections (`/path/to/db.duckdb`) +//! - In-memory connections (`:memory:`) +//! - Schema introspection via `DuckDB` catalog functions (`duckdb_tables()`, +//! `duckdb_views()`, `duckdb_columns()`, `duckdb_constraints()`, `duckdb_indexes()`) +//! - Capability-enforced query execution +//! +//! # Implementation Notes +//! - Uses the `duckdb` crate (synchronous driver, rusqlite-style API) +//! - Connections open with `AccessMode::ReadOnly` — defense in depth at the +//! storage layer; writes are rejected by `DuckDB` itself even if the parser +//! is somehow bypassed. In-memory databases cannot be opened read-only +//! (there is nothing on disk to protect), so `:memory:` opens read-write +//! and relies on the capability parser. +//! - BLOB data is Base64-encoded for JSON safety +//! - Statement timeouts enforced via `InterruptHandle` (interrupt + timer thread) +//! - Row limits enforced in application code +//! - `DuckDB` supports schemas; introspection defaults to the `main` schema + +use duckdb::types::{TimeUnit, Value}; +use duckdb::{params_from_iter, AccessMode, Config, Connection}; +use std::time::{Duration, Instant}; + +use crate::capability::{strip_explain_prefix, validate_query}; +use crate::engine::{ + is_explain_query, Capabilities, ColumnInfo, ConnectionConfig, ConnectionInfo, DatabaseEngine, + DatabaseType, ExplainFormat, ExplainPlanNode, ForeignKeyInfo, IndexInfo, IndexSummary, + IntrospectOperation, IntrospectResult, QueryResult, TableFields, TableInfo, ViewInfo, +}; +use crate::error::{PlenumError, Result}; + +/// `DuckDB` database engine implementation +pub struct DuckDbEngine; + +/// Default schema used when no `--schema` is provided. +const DEFAULT_SCHEMA: &str = "main"; + +impl DatabaseEngine for DuckDbEngine { + async fn validate_connection(config: &ConnectionConfig) -> Result { + let file_path = extract_file_path(config)?; + let conn = open_connection(&file_path)?; + + // Get DuckDB version (e.g. "v1.5.4") + let version: String = + conn.query_row("SELECT version()", [], |row| row.get(0)).map_err(|e| { + PlenumError::connection_failed(format!("Failed to query DuckDB version: {e}")) + })?; + + let db_name = database_display_name(config); + + Ok(ConnectionInfo { + database_version: version.clone(), + server_info: format!("DuckDB {version}"), + connected_database: db_name, + user: "N/A".to_string(), // DuckDB has no user concept + }) + } + + async fn introspect( + config: &ConnectionConfig, + operation: &IntrospectOperation, + database: Option<&str>, + schema: Option<&str>, + ) -> Result { + let file_path = extract_file_path(config)?; + + // DuckDB is file-based: each file is one database. No database override. + if database.is_some() { + return Err(PlenumError::invalid_input( + "DuckDB does not support --database parameter (use a different connection config to target a different database file)" + )); + } + + let schema_name = schema.unwrap_or(DEFAULT_SCHEMA); + let conn = open_connection(&file_path)?; + + let result = match operation { + IntrospectOperation::ListDatabases => list_databases_duckdb(&conn)?, + IntrospectOperation::ListSchemas => list_schemas_duckdb(&conn)?, + IntrospectOperation::ListTables => list_tables_duckdb(&conn, schema_name)?, + IntrospectOperation::ListViews => list_views_duckdb(&conn, schema_name)?, + IntrospectOperation::ListIndexes { table } => { + list_indexes_duckdb(&conn, schema_name, table.as_deref())? + } + IntrospectOperation::TableDetails { name, fields } => { + get_table_details_duckdb(&conn, schema_name, name, fields)? + } + IntrospectOperation::ViewDetails { name } => { + get_view_details_duckdb(&conn, schema_name, name)? + } + }; + + Ok(result) + } + + async fn execute( + config: &ConnectionConfig, + query: &str, + params: &[serde_json::Value], + caps: &Capabilities, + ) -> Result { + // Validate query against capabilities before opening any connection + validate_query(query, caps, DatabaseType::DuckDB)?; + + let file_path = extract_file_path(config)?; + let conn = open_connection(&file_path)?; + + // Interrupt-based statement timeout: obtain a handle before the query + // starts, then spawn a thread that fires the interrupt after timeout_ms. + // DuckDB checks the interrupt flag during execution, cancelling the + // query server-side rather than just abandoning the wait. + if let Some(timeout_ms) = caps.timeout_ms { + let handle = conn.interrupt_handle(); + std::thread::spawn(move || { + std::thread::sleep(Duration::from_millis(timeout_ms)); + handle.interrupt(); + }); + } + + // Structured explain path: rewrite to EXPLAIN (FORMAT JSON), normalize. + if caps.explain_format == Some(ExplainFormat::Structured) { + if !is_explain_query(query) { + return Err(PlenumError::invalid_input( + "--explain-format structured requires an EXPLAIN statement; \ + non-EXPLAIN queries must omit this flag", + )); + } + let inner = strip_explain_prefix(query); + let start = Instant::now(); + let plan = execute_structured_explain_duckdb(&conn, &inner)?; + let elapsed = start.elapsed(); + return Ok(QueryResult { + columns: Vec::new(), + rows: Vec::new(), + rows_affected: None, + execution_ms: elapsed.as_millis() as u64, + rows_truncated: false, + truncated_by: None, + plan: Some(plan), + }); + } + + let start = Instant::now(); + let mut result = execute_query(&conn, query, params, caps)?; + let elapsed = start.elapsed(); + result.execution_ms = elapsed.as_millis() as u64; + + Ok(result) + } +} + +/// Validate the config targets `DuckDB` and extract the file path as a string. +fn extract_file_path(config: &ConnectionConfig) -> Result { + if config.engine != DatabaseType::DuckDB { + return Err(PlenumError::invalid_input(format!( + "Expected DuckDB engine, got {}", + config.engine + ))); + } + + let file_path = config + .file + .as_ref() + .ok_or_else(|| PlenumError::invalid_input("DuckDB requires 'file' parameter"))?; + + file_path.to_str().map(std::string::ToString::to_string).ok_or_else(|| { + PlenumError::invalid_input("DuckDB file path contains invalid UTF-8 characters") + }) +} + +/// Display name for the connected database (file name, or `:memory:`). +fn database_display_name(config: &ConnectionConfig) -> String { + config.file.as_ref().map_or_else( + || "unknown".to_string(), + |p| { + p.file_name() + .and_then(|n| n.to_str()) + .unwrap_or_else(|| p.to_str().unwrap_or("unknown")) + .to_string() + }, + ) +} + +/// Open a `DuckDB` connection. +/// +/// File-backed databases open with `AccessMode::ReadOnly` (defense in depth — +/// `DuckDB` itself rejects writes even if the capability parser were bypassed). +/// `:memory:` databases cannot be opened read-only, so they open read-write; +/// the capability parser remains the enforcement boundary there, and an +/// in-memory database holds no pre-existing data to protect. +fn open_connection(path: &str) -> Result { + if path == ":memory:" { + return Connection::open_in_memory().map_err(|e| { + PlenumError::connection_failed(format!("Failed to open DuckDB database: {e}")) + }); + } + + let config = Config::default().access_mode(AccessMode::ReadOnly).map_err(|e| { + PlenumError::engine_error("duckdb", format!("Failed to configure read-only mode: {e}")) + })?; + + Connection::open_with_flags(path, config) + .map_err(|e| PlenumError::connection_failed(format!("Failed to open DuckDB database: {e}"))) +} + +/// Returns true when a duckdb error was caused by the interrupt handle firing. +fn is_duckdb_interrupt(e: &duckdb::Error) -> bool { + e.to_string().to_uppercase().contains("INTERRUPT") +} + +/// Run a single-column string query and collect the results. +fn query_string_list(conn: &Connection, sql: &str, params: &[&str]) -> Result> { + let mut stmt = conn.prepare(sql).map_err(|e| { + PlenumError::engine_error("duckdb", format!("Failed to prepare query: {e}")) + })?; + + let values: Vec = stmt + .query_map(params_from_iter(params.iter()), |row| row.get(0)) + .map_err(|e| PlenumError::engine_error("duckdb", format!("Failed to execute query: {e}")))? + .collect::, _>>() + .map_err(|e| PlenumError::engine_error("duckdb", format!("Failed to collect rows: {e}")))?; + + Ok(values) +} + +/// List attached databases (excludes `DuckDB` internal catalogs). +fn list_databases_duckdb(conn: &Connection) -> Result { + let databases = query_string_list( + conn, + "SELECT database_name FROM duckdb_databases() WHERE NOT internal ORDER BY database_name", + &[], + )?; + Ok(IntrospectResult::DatabaseList { databases }) +} + +/// List schemas in the connected database. +/// +/// Filters on `database_name = current_database()` rather than the `internal` +/// flag: `DuckDB` marks the built-in `main` schema as internal, but it is the +/// default location for user tables and must be listed. +fn list_schemas_duckdb(conn: &Connection) -> Result { + let schemas = query_string_list( + conn, + "SELECT schema_name FROM duckdb_schemas() + WHERE database_name = current_database() + ORDER BY schema_name", + &[], + )?; + Ok(IntrospectResult::SchemaList { schemas }) +} + +/// List all tables in a schema (excludes `DuckDB` internal tables) +fn list_tables_duckdb(conn: &Connection, schema: &str) -> Result { + let tables = query_string_list( + conn, + "SELECT table_name FROM duckdb_tables() + WHERE NOT internal AND schema_name = ? + ORDER BY table_name", + &[schema], + )?; + Ok(IntrospectResult::TableList { tables }) +} + +/// List all views in a schema (excludes `DuckDB` internal views) +fn list_views_duckdb(conn: &Connection, schema: &str) -> Result { + let views = query_string_list( + conn, + "SELECT view_name FROM duckdb_views() + WHERE NOT internal AND schema_name = ? + ORDER BY view_name", + &[schema], + )?; + Ok(IntrospectResult::ViewList { views }) +} + +/// Parse a `DuckDB` expression list rendered as text (e.g. `[col_a, col_b]`) +/// into individual column names. +fn parse_bracketed_list(raw: &str) -> Vec { + raw.trim() + .trim_start_matches('[') + .trim_end_matches(']') + .split(',') + .map(|s| s.trim().trim_matches('\'').trim_matches('"').to_string()) + .filter(|s| !s.is_empty()) + .collect() +} + +/// List all indexes (optionally filtered by table) +fn list_indexes_duckdb( + conn: &Connection, + schema: &str, + table_filter: Option<&str>, +) -> Result { + let (sql, params): (&str, Vec<&str>) = if let Some(table) = table_filter { + ( + "SELECT index_name, table_name, is_unique, CAST(expressions AS VARCHAR) + FROM duckdb_indexes() + WHERE schema_name = ? AND table_name = ? + ORDER BY index_name", + vec![schema, table], + ) + } else { + ( + "SELECT index_name, table_name, is_unique, CAST(expressions AS VARCHAR) + FROM duckdb_indexes() + WHERE schema_name = ? + ORDER BY index_name", + vec![schema], + ) + }; + + let mut stmt = conn.prepare(sql).map_err(|e| { + PlenumError::engine_error("duckdb", format!("Failed to query indexes: {e}")) + })?; + + let indexes: Vec = stmt + .query_map(params_from_iter(params.iter()), |row| { + let name: String = row.get(0)?; + let table: String = row.get(1)?; + let unique: bool = row.get(2)?; + let expressions: String = row.get(3)?; + Ok(IndexSummary { name, table, unique, columns: parse_bracketed_list(&expressions) }) + }) + .map_err(|e| { + PlenumError::engine_error("duckdb", format!("Failed to fetch index data: {e}")) + })? + .collect::, _>>() + .map_err(|e| { + PlenumError::engine_error("duckdb", format!("Failed to collect index data: {e}")) + })?; + + Ok(IntrospectResult::IndexList { indexes }) +} + +/// Get full table details with conditional field retrieval +fn get_table_details_duckdb( + conn: &Connection, + schema: &str, + table_name: &str, + fields: &TableFields, +) -> Result { + let full_table = introspect_table(conn, schema, table_name)?; + + let table = TableInfo { + name: full_table.name, + schema: full_table.schema, + columns: if fields.columns { full_table.columns } else { Vec::new() }, + primary_key: if fields.primary_key { full_table.primary_key } else { None }, + foreign_keys: if fields.foreign_keys { full_table.foreign_keys } else { Vec::new() }, + indexes: if fields.indexes { full_table.indexes } else { Vec::new() }, + comment: full_table.comment, + row_estimate: full_table.row_estimate, + }; + + Ok(IntrospectResult::TableDetails { table }) +} + +/// Get view details including definition and columns +fn get_view_details_duckdb( + conn: &Connection, + schema: &str, + view_name: &str, +) -> Result { + let mut def_stmt = conn + .prepare( + "SELECT sql FROM duckdb_views() + WHERE schema_name = ? AND view_name = ?", + ) + .map_err(|e| { + PlenumError::engine_error("duckdb", format!("Failed to prepare view query: {e}")) + })?; + + let definition: Option = def_stmt + .query_row(params_from_iter([schema, view_name].iter()), |row| row.get(0)) + .map_err(|e| { + if matches!(e, duckdb::Error::QueryReturnedNoRows) { + PlenumError::invalid_input(format!("View '{view_name}' not found")) + } else { + PlenumError::engine_error("duckdb", format!("Failed to query view definition: {e}")) + } + })?; + + let columns = get_columns(conn, schema, view_name)?; + + let view = ViewInfo { + name: view_name.to_string(), + schema: Some(schema.to_string()), + definition, + columns, + }; + + Ok(IntrospectResult::ViewDetails { view }) +} + +/// Get column info for a table or view via `duckdb_columns()` +fn get_columns(conn: &Connection, schema: &str, table_name: &str) -> Result> { + let mut stmt = conn + .prepare( + "SELECT column_name, data_type, is_nullable, column_default, comment + FROM duckdb_columns() + WHERE schema_name = ? AND table_name = ? + ORDER BY column_index", + ) + .map_err(|e| { + PlenumError::engine_error("duckdb", format!("Failed to prepare column query: {e}")) + })?; + + let columns: Vec = stmt + .query_map(params_from_iter([schema, table_name].iter()), |row| { + Ok(ColumnInfo { + name: row.get(0)?, + data_type: row.get(1)?, + nullable: row.get(2)?, + default: row.get::<_, Option>(3)?, + comment: row.get::<_, Option>(4)?, + }) + }) + .map_err(|e| PlenumError::engine_error("duckdb", format!("Failed to query columns: {e}")))? + .collect::, _>>() + .map_err(|e| { + PlenumError::engine_error("duckdb", format!("Failed to collect columns: {e}")) + })?; + + Ok(columns) +} + +/// Introspect a single table and return `TableInfo` +fn introspect_table(conn: &Connection, schema: &str, table_name: &str) -> Result { + // Verify table exists and fetch comment + row estimate in one pass + let mut check_stmt = conn + .prepare( + "SELECT comment, estimated_size FROM duckdb_tables() + WHERE NOT internal AND schema_name = ? AND table_name = ?", + ) + .map_err(|e| { + PlenumError::engine_error("duckdb", format!("Failed to check table existence: {e}")) + })?; + + let (comment, row_estimate): (Option, Option) = check_stmt + .query_row(params_from_iter([schema, table_name].iter()), |row| { + Ok((row.get::<_, Option>(0)?, row.get::<_, Option>(1)?)) + }) + .map_err(|e| { + if matches!(e, duckdb::Error::QueryReturnedNoRows) { + PlenumError::invalid_input(format!("Table '{table_name}' not found")) + } else { + PlenumError::engine_error("duckdb", format!("Failed to query table: {e}")) + } + })?; + + let columns = get_columns(conn, schema, table_name)?; + + // Primary key from duckdb_constraints() + let mut pk_stmt = conn + .prepare( + "SELECT array_to_string(constraint_column_names, ',') + FROM duckdb_constraints() + WHERE schema_name = ? AND table_name = ? AND constraint_type = 'PRIMARY KEY'", + ) + .map_err(|e| { + PlenumError::engine_error("duckdb", format!("Failed to prepare pk query: {e}")) + })?; + + let pk_raw: Option = pk_stmt + .query_row(params_from_iter([schema, table_name].iter()), |row| row.get(0)) + .map(Some) + .or_else(|e| { + if matches!(e, duckdb::Error::QueryReturnedNoRows) { + Ok(None) + } else { + Err(PlenumError::engine_error( + "duckdb", + format!("Failed to query primary key: {e}"), + )) + } + })?; + + let primary_key = pk_raw.map(|raw| { + raw.split(',').map(|s| s.trim().to_string()).filter(|s| !s.is_empty()).collect::>() + }); + + // Foreign keys from duckdb_constraints() + let mut fk_stmt = conn + .prepare( + "SELECT array_to_string(constraint_column_names, ','), + referenced_table, + array_to_string(referenced_column_names, ',') + FROM duckdb_constraints() + WHERE schema_name = ? AND table_name = ? AND constraint_type = 'FOREIGN KEY' + ORDER BY constraint_index", + ) + .map_err(|e| { + PlenumError::engine_error("duckdb", format!("Failed to prepare fk query: {e}")) + })?; + + let fk_rows: Vec<(String, String, String)> = fk_stmt + .query_map(params_from_iter([schema, table_name].iter()), |row| { + Ok((row.get(0)?, row.get(1)?, row.get(2)?)) + }) + .map_err(|e| { + PlenumError::engine_error("duckdb", format!("Failed to query foreign keys: {e}")) + })? + .collect::, _>>() + .map_err(|e| { + PlenumError::engine_error("duckdb", format!("Failed to collect foreign keys: {e}")) + })?; + + let foreign_keys: Vec = fk_rows + .into_iter() + .enumerate() + .map(|(i, (cols, ref_table, ref_cols))| ForeignKeyInfo { + name: format!("fk_{table_name}_{i}"), + columns: cols.split(',').map(|s| s.trim().to_string()).collect(), + referenced_table: ref_table, + referenced_columns: ref_cols.split(',').map(|s| s.trim().to_string()).collect(), + }) + .collect(); + + // Indexes from duckdb_indexes() + let IntrospectResult::IndexList { indexes: index_summaries } = + list_indexes_duckdb(conn, schema, Some(table_name))? + else { + return Err(PlenumError::engine_error("duckdb", "Unexpected index list result")); + }; + + let indexes: Vec = index_summaries + .into_iter() + .map(|s| IndexInfo { name: s.name, columns: s.columns, unique: s.unique }) + .collect(); + + Ok(TableInfo { + name: table_name.to_string(), + schema: Some(schema.to_string()), + columns, + primary_key, + foreign_keys, + indexes, + comment, + row_estimate, + }) +} + +/// Run `EXPLAIN (FORMAT JSON)` on `inner_sql` and normalize the result into an +/// `ExplainPlanNode` tree. +/// +/// `DuckDB` returns rows of `(explain_key, explain_value)` where the value is a +/// JSON array of plan nodes: `[{"name": ..., "extra_info": {...}, "children": [...]}]`. +fn execute_structured_explain_duckdb( + conn: &Connection, + inner_sql: &str, +) -> Result { + let sql = format!("EXPLAIN (FORMAT JSON) {inner_sql}"); + + let mut stmt = conn.prepare(&sql).map_err(|e| { + PlenumError::query_failed(format!("Failed to prepare EXPLAIN (FORMAT JSON): {e}")) + })?; + + let json_text: String = stmt.query_row([], |row| row.get(1)).map_err(|e| { + PlenumError::query_failed(format!("Failed to execute EXPLAIN (FORMAT JSON): {e}")) + })?; + + let parsed: serde_json::Value = serde_json::from_str(&json_text).map_err(|e| { + PlenumError::query_failed(format!("Failed to parse DuckDB EXPLAIN JSON: {e}")) + })?; + + fn build_node(node: &serde_json::Value) -> ExplainPlanNode { + let node_type = + node.get("name").and_then(|v| v.as_str()).unwrap_or("UNKNOWN").trim().to_string(); + + let extra = node.get("extra_info"); + let relation = extra + .and_then(|e| e.get("Table")) + .and_then(|v| v.as_str()) + .map(|s| s.trim().to_string()); + let estimated_rows = extra + .and_then(|e| e.get("Estimated Cardinality")) + .and_then(|v| v.as_str().map_or_else(|| v.as_f64(), |s| s.trim().parse::().ok())); + + let children = node + .get("children") + .and_then(|v| v.as_array()) + .map(|kids| kids.iter().map(build_node).collect()) + .unwrap_or_default(); + + ExplainPlanNode { node_type, relation, estimated_rows, estimated_cost: None, children } + } + + let top_children: Vec = match &parsed { + serde_json::Value::Array(nodes) => nodes.iter().map(build_node).collect(), + other => vec![build_node(other)], + }; + + Ok(ExplainPlanNode { + node_type: "QUERY PLAN".to_string(), + relation: None, + estimated_rows: None, + estimated_cost: None, + children: top_children, + }) +} + +/// Convert a JSON parameter value to a `duckdb` native value for binding +fn json_to_duckdb_value(val: &serde_json::Value) -> Value { + match val { + serde_json::Value::Null => Value::Null, + serde_json::Value::Bool(b) => Value::Boolean(*b), + serde_json::Value::Number(n) => { + if let Some(i) = n.as_i64() { + Value::BigInt(i) + } else { + Value::Double(n.as_f64().unwrap_or(0.0)) + } + } + serde_json::Value::String(s) => Value::Text(s.clone()), + v => Value::Text(v.to_string()), + } +} + +/// Execute query and return `QueryResult` +fn execute_query( + conn: &Connection, + query: &str, + params: &[serde_json::Value], + caps: &Capabilities, +) -> Result { + let mut stmt = conn + .prepare(query) + .map_err(|e| PlenumError::query_failed(format!("Failed to prepare query: {e}")))?; + + let duckdb_params: Vec = params.iter().map(json_to_duckdb_value).collect(); + + let mut rows = stmt.query(params_from_iter(duckdb_params.iter())).map_err(|e| { + if is_duckdb_interrupt(&e) { + PlenumError::query_timeout("Query interrupted by DuckDB server-side timeout") + } else { + PlenumError::query_failed(format!("Failed to execute query: {e}")) + } + })?; + + // Column names are only available after execution in the duckdb crate. + let column_names: Vec = + rows.as_ref().map(duckdb::Statement::column_names).unwrap_or_default(); + + let offset = caps.offset.unwrap_or(0); + let max = caps.max_rows; + let mut pos = 0usize; + let mut rows_data: Vec> = Vec::new(); + let mut rows_truncated = false; + + loop { + let next = rows.next().map_err(|e| { + if is_duckdb_interrupt(&e) { + PlenumError::query_timeout( + "Query interrupted by DuckDB server-side timeout during row fetch", + ) + } else { + PlenumError::query_failed(format!("Failed to fetch row: {e}")) + } + })?; + let Some(row) = next else { break }; + + // Skip offset rows + if pos < offset { + pos += 1; + continue; + } + + // Probe one row past max_rows to detect truncation + if let Some(m) = max { + if rows_data.len() >= m { + rows_truncated = true; + break; + } + } + + let mut values = Vec::with_capacity(column_names.len()); + for idx in 0..column_names.len() { + let value_ref = row.get_ref(idx).map_err(|e| { + PlenumError::query_failed(format!("Failed to read column {idx}: {e}")) + })?; + values.push(duckdb_value_to_json(&value_ref.to_owned())); + } + rows_data.push(values); + pos += 1; + } + + Ok(QueryResult { + columns: column_names, + rows: rows_data, + rows_affected: None, + execution_ms: 0, + rows_truncated, + truncated_by: None, + plan: None, + }) +} + +/// Format a `DuckDB` timestamp/time value (count of `unit` since the epoch / +/// midnight) as an ISO-8601 string, falling back to the raw integer if the +/// value is out of chrono's representable range. +fn format_timestamp(unit: TimeUnit, value: i64) -> serde_json::Value { + let micros = unit.to_micros(value); + chrono::DateTime::from_timestamp_micros(micros).map_or_else( + || serde_json::Value::Number(micros.into()), + |dt| serde_json::Value::String(dt.naive_utc().format("%Y-%m-%d %H:%M:%S%.6f").to_string()), + ) +} + +fn format_time(unit: TimeUnit, value: i64) -> serde_json::Value { + let micros = unit.to_micros(value); + let raw = serde_json::Value::Number(micros.into()); + // TIME values are non-negative micros since midnight; a negative value is + // out of range and falls back to the raw integer. + let Ok(secs) = u32::try_from(micros / 1_000_000) else { return raw }; + let Ok(sub_micros) = u32::try_from(micros % 1_000_000) else { return raw }; + chrono::NaiveTime::from_num_seconds_from_midnight_opt(secs, sub_micros * 1000).map_or_else( + || serde_json::Value::Number(micros.into()), + |t| serde_json::Value::String(t.format("%H:%M:%S%.6f").to_string()), + ) +} + +fn format_date(days_since_epoch: i32) -> serde_json::Value { + chrono::DateTime::from_timestamp(i64::from(days_since_epoch) * 86_400, 0).map_or_else( + || serde_json::Value::Number(days_since_epoch.into()), + |dt| serde_json::Value::String(dt.date_naive().format("%Y-%m-%d").to_string()), + ) +} + +/// Convert an f64 to JSON, mapping NaN/Infinity to null +fn f64_to_json(f: f64) -> serde_json::Value { + serde_json::Number::from_f64(f).map_or(serde_json::Value::Null, serde_json::Value::Number) +} + +/// Convert an owned `DuckDB` value to a JSON value. +/// +/// Scalar types map to their natural JSON equivalents. Values that JSON cannot +/// represent natively are stringified deterministically: +/// - `HUGEINT` / `UHUGEINT` and `DECIMAL` → string (preserves precision) +/// - `TIMESTAMP` / `DATE` / `TIME` → ISO-8601 string +/// - `BLOB` → Base64 string +/// - `INTERVAL` → object with `months` / `days` / `nanos` +/// - Nested types (`LIST`, `ARRAY`, `STRUCT`, `MAP`, `UNION`, `ENUM`) convert +/// recursively to JSON arrays / objects. +fn duckdb_value_to_json(value: &Value) -> serde_json::Value { + use serde_json::Value as Json; + + match value { + Value::Null => Json::Null, + Value::Boolean(b) => Json::Bool(*b), + Value::TinyInt(i) => Json::Number((*i).into()), + Value::SmallInt(i) => Json::Number((*i).into()), + Value::Int(i) => Json::Number((*i).into()), + Value::BigInt(i) => Json::Number((*i).into()), + // HUGEINT exceeds JSON's i64 range; preserve precision as a string + Value::HugeInt(i) => Json::String(i.to_string()), + Value::UTinyInt(i) => Json::Number((*i).into()), + Value::USmallInt(i) => Json::Number((*i).into()), + Value::UInt(i) => Json::Number((*i).into()), + Value::UBigInt(i) => Json::Number((*i).into()), + Value::Float(f) => f64_to_json(f64::from(*f)), + Value::Double(f) => f64_to_json(*f), + // DECIMAL preserves exact precision as a string + Value::Decimal(d) => Json::String(d.to_string()), + Value::Timestamp(unit, v) => format_timestamp(*unit, *v), + Value::Text(s) | Value::Enum(s) => Json::String(s.clone()), + Value::Blob(b) => { + use base64::Engine; + Json::String(base64::engine::general_purpose::STANDARD.encode(b)) + } + Value::Date32(d) => format_date(*d), + Value::Time64(unit, v) => format_time(*unit, *v), + Value::Interval { months, days, nanos } => serde_json::json!({ + "months": months, + "days": days, + "nanos": nanos, + }), + Value::List(items) | Value::Array(items) => { + Json::Array(items.iter().map(duckdb_value_to_json).collect()) + } + Value::Struct(map) => { + let obj: serde_json::Map = + map.iter().map(|(k, v)| (k.clone(), duckdb_value_to_json(v))).collect(); + Json::Object(obj) + } + Value::Map(map) => { + let obj: serde_json::Map = map + .iter() + .map(|(k, v)| { + let key = match k { + Value::Text(s) | Value::Enum(s) => s.clone(), + other => match duckdb_value_to_json(other) { + Json::String(s) => s, + j => j.to_string(), + }, + }; + (key, duckdb_value_to_json(v)) + }) + .collect(); + Json::Object(obj) + } + Value::Union(inner) => duckdb_value_to_json(inner), + } +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::engine::DatabaseType; + use std::path::PathBuf; + use std::sync::atomic::{AtomicU64, Ordering}; + + static FIXTURE_COUNTER: AtomicU64 = AtomicU64::new(0); + + fn fixture_path(tag: &str) -> PathBuf { + let id = FIXTURE_COUNTER.fetch_add(1, Ordering::SeqCst); + let pid = std::process::id(); + std::env::temp_dir().join(format!("plenum_duckdb_{tag}_{pid}_{id}.duckdb")) + } + + #[tokio::test] + async fn test_validate_connection_memory() { + let config = ConnectionConfig::duckdb(":memory:".into()); + let result = DuckDbEngine::validate_connection(&config).await; + assert!(result.is_ok(), "validate failed: {:?}", result.err()); + + let info = result.unwrap(); + assert!(info.database_version.starts_with('v'), "unexpected: {}", info.database_version); + assert!(info.server_info.contains("DuckDB")); + assert_eq!(info.connected_database, ":memory:"); + assert_eq!(info.user, "N/A"); + } + + #[tokio::test] + async fn test_validate_connection_wrong_engine() { + let mut config = ConnectionConfig::duckdb(":memory:".into()); + config.engine = DatabaseType::SQLite; + + let result = DuckDbEngine::validate_connection(&config).await; + assert!(result.is_err()); + assert!(result.unwrap_err().message().contains("Expected DuckDB engine")); + } + + #[tokio::test] + async fn test_validate_connection_missing_file() { + let config = ConnectionConfig { + engine: DatabaseType::DuckDB, + file: None, + tls: None, + host: None, + port: None, + user: None, + password: None, + database: None, + }; + + let result = DuckDbEngine::validate_connection(&config).await; + assert!(result.is_err()); + assert!(result.unwrap_err().message().contains("DuckDB requires 'file' parameter")); + } + + #[tokio::test] + async fn test_validate_connection_nonexistent_file() { + let config = + ConnectionConfig::duckdb(std::env::temp_dir().join("plenum_no_such_file.duckdb")); + let result = DuckDbEngine::validate_connection(&config).await; + assert!(result.is_err()); + assert_eq!(result.unwrap_err().error_code(), "CONNECTION_FAILED"); + } + + #[tokio::test] + async fn test_introspect_schema() { + let temp_file = fixture_path("introspect"); + let _ = std::fs::remove_file(&temp_file); + + { + let conn = Connection::open(&temp_file).expect("Failed to create temp database"); + conn.execute_batch( + "CREATE TABLE users ( + id INTEGER PRIMARY KEY, + name TEXT NOT NULL, + email TEXT + )", + ) + .expect("Failed to create table"); + } + + let config = ConnectionConfig::duckdb(temp_file.clone()); + let result = DuckDbEngine::introspect( + &config, + &IntrospectOperation::TableDetails { + name: "users".to_string(), + fields: TableFields::all(), + }, + None, + None, + ) + .await; + assert!(result.is_ok(), "introspect failed: {:?}", result.err()); + + let IntrospectResult::TableDetails { table } = result.unwrap() else { + panic!("Expected TableDetails result") + }; + + assert_eq!(table.name, "users"); + assert_eq!(table.schema.as_deref(), Some("main")); + assert_eq!(table.columns.len(), 3); + + let pk = table.primary_key.as_ref().expect("primary key missing"); + assert_eq!(pk, &vec!["id".to_string()]); + + // NOT NULL detection + let name_col = table.columns.iter().find(|c| c.name == "name").unwrap(); + assert!(!name_col.nullable); + let email_col = table.columns.iter().find(|c| c.name == "email").unwrap(); + assert!(email_col.nullable); + + let _ = std::fs::remove_file(&temp_file); + } + + #[tokio::test] + async fn test_introspect_table_not_found() { + let temp_file = fixture_path("notfound"); + let _ = std::fs::remove_file(&temp_file); + { + let conn = Connection::open(&temp_file).expect("create"); + conn.execute_batch("CREATE TABLE t (id INTEGER)").unwrap(); + } + + let config = ConnectionConfig::duckdb(temp_file.clone()); + let result = DuckDbEngine::introspect( + &config, + &IntrospectOperation::TableDetails { + name: "missing".to_string(), + fields: TableFields::all(), + }, + None, + None, + ) + .await; + assert!(result.is_err()); + assert!(result.unwrap_err().message().contains("not found")); + let _ = std::fs::remove_file(&temp_file); + } + + #[tokio::test] + async fn test_introspect_list_schemas() { + let temp_file = fixture_path("schemas"); + let _ = std::fs::remove_file(&temp_file); + { + let conn = Connection::open(&temp_file).expect("create"); + conn.execute_batch("CREATE SCHEMA analytics; CREATE TABLE analytics.t (id INTEGER)") + .unwrap(); + } + + let config = ConnectionConfig::duckdb(temp_file.clone()); + let result = + DuckDbEngine::introspect(&config, &IntrospectOperation::ListSchemas, None, None) + .await + .expect("ListSchemas failed"); + + let IntrospectResult::SchemaList { schemas } = result else { + panic!("Expected SchemaList") + }; + assert!(schemas.contains(&"main".to_string())); + assert!(schemas.contains(&"analytics".to_string())); + + // Schema-scoped table listing + let result = DuckDbEngine::introspect( + &config, + &IntrospectOperation::ListTables, + None, + Some("analytics"), + ) + .await + .expect("ListTables failed"); + let IntrospectResult::TableList { tables } = result else { panic!("Expected TableList") }; + assert_eq!(tables, vec!["t".to_string()]); + + let _ = std::fs::remove_file(&temp_file); + } + + #[tokio::test] + async fn test_introspect_database_override_rejected() { + let config = ConnectionConfig::duckdb(":memory:".into()); + let result = DuckDbEngine::introspect( + &config, + &IntrospectOperation::ListTables, + Some("other"), + None, + ) + .await; + assert!(result.is_err()); + assert!(result.unwrap_err().message().contains("--database")); + } + + #[tokio::test] + async fn test_execute_select_query() { + let temp_file = fixture_path("select"); + let _ = std::fs::remove_file(&temp_file); + + { + let conn = Connection::open(&temp_file).expect("Failed to create temp database"); + conn.execute_batch( + "CREATE TABLE users (id INTEGER PRIMARY KEY, name TEXT); + INSERT INTO users VALUES (1, 'Alice')", + ) + .expect("seed"); + } + + let config = ConnectionConfig::duckdb(temp_file.clone()); + let caps = Capabilities::default(); + let result = DuckDbEngine::execute(&config, "SELECT * FROM users", &[], &caps).await; + assert!(result.is_ok(), "select failed: {:?}", result.err()); + + let query_result = result.unwrap(); + assert_eq!(query_result.columns, vec!["id".to_string(), "name".to_string()]); + assert_eq!(query_result.rows.len(), 1); + assert_eq!(query_result.rows[0][1], serde_json::json!("Alice")); + + let _ = std::fs::remove_file(&temp_file); + } + + #[tokio::test] + async fn test_execute_insert_rejected() { + let config = ConnectionConfig::duckdb(":memory:".into()); + let caps = Capabilities::default(); + let result = + DuckDbEngine::execute(&config, "INSERT INTO users VALUES (1, 'Bob')", &[], &caps).await; + + assert!(result.is_err()); + let err = result.unwrap_err(); + assert_eq!(err.error_code(), "CAPABILITY_VIOLATION"); + assert!(err.message().contains("Plenum is read-only")); + } + + #[tokio::test] + async fn test_execute_ddl_rejected() { + let config = ConnectionConfig::duckdb(":memory:".into()); + let caps = Capabilities::default(); + for sql in [ + "CREATE TABLE t (id INTEGER)", + "DROP TABLE t", + "ALTER TABLE t ADD COLUMN c INTEGER", + "UPDATE t SET id = 2", + "DELETE FROM t", + "ATTACH ':memory:' AS other", + "COPY t TO 'out.csv'", + ] { + let result = DuckDbEngine::execute(&config, sql, &[], &caps).await; + assert!(result.is_err(), "expected rejection for: {sql}"); + assert_eq!(result.unwrap_err().error_code(), "CAPABILITY_VIOLATION", "sql: {sql}"); + } + } + + #[tokio::test] + async fn test_execute_max_rows_limit() { + let temp_file = fixture_path("maxrows"); + let _ = std::fs::remove_file(&temp_file); + + { + let conn = Connection::open(&temp_file).expect("create"); + conn.execute_batch("CREATE TABLE nums AS SELECT range AS n FROM range(10)") + .expect("seed"); + } + + let config = ConnectionConfig::duckdb(temp_file.clone()); + let caps = Capabilities { max_rows: Some(5), ..Capabilities::default() }; + let result = + DuckDbEngine::execute(&config, "SELECT * FROM nums ORDER BY n", &[], &caps).await; + + assert!(result.is_ok(), "query failed: {:?}", result.err()); + let query_result = result.unwrap(); + assert_eq!(query_result.rows.len(), 5); + assert!(query_result.rows_truncated); + + let _ = std::fs::remove_file(&temp_file); + } + + #[tokio::test] + async fn test_execute_offset_pagination() { + let temp_file = fixture_path("offset"); + let _ = std::fs::remove_file(&temp_file); + { + let conn = Connection::open(&temp_file).expect("create"); + conn.execute_batch("CREATE TABLE nums AS SELECT range AS n FROM range(10)").unwrap(); + } + + let config = ConnectionConfig::duckdb(temp_file.clone()); + let caps = Capabilities { max_rows: Some(3), offset: Some(4), ..Capabilities::default() }; + let result = + DuckDbEngine::execute(&config, "SELECT n FROM nums ORDER BY n", &[], &caps).await; + let qr = result.expect("query failed"); + assert_eq!(qr.rows.len(), 3); + assert_eq!(qr.rows[0][0], serde_json::json!(4)); + + let _ = std::fs::remove_file(&temp_file); + } + + #[tokio::test] + async fn test_execute_all_data_types() { + let temp_file = fixture_path("types"); + let _ = std::fs::remove_file(&temp_file); + + { + let conn = Connection::open(&temp_file).expect("create"); + conn.execute_batch( + "CREATE TABLE test_types ( + c_bool BOOLEAN, + c_int INTEGER, + c_bigint BIGINT, + c_hugeint HUGEINT, + c_double DOUBLE, + c_decimal DECIMAL(18,4), + c_text VARCHAR, + c_blob BLOB, + c_date DATE, + c_time TIME, + c_timestamp TIMESTAMP, + c_list INTEGER[], + c_struct STRUCT(a INTEGER, b VARCHAR), + c_null VARCHAR + ); + INSERT INTO test_types VALUES ( + TRUE, + 42, + 9223372036854775807, + 170141183460469231731687303715884105727, + 3.5, + 12345.6789, + 'café résumé 🚀', + '\\xDE\\xAD\\xBE\\xEF'::BLOB, + DATE '2024-01-15', + TIME '13:45:30', + TIMESTAMP '2024-01-15 13:45:30', + [1, 2, 3], + {a: 7, b: 'x'}, + NULL + )", + ) + .expect("seed"); + } + + let config = ConnectionConfig::duckdb(temp_file.clone()); + let caps = Capabilities::default(); + let result = DuckDbEngine::execute(&config, "SELECT * FROM test_types", &[], &caps).await; + + assert!(result.is_ok(), "query failed: {:?}", result.err()); + let qr = result.unwrap(); + assert_eq!(qr.rows.len(), 1); + let row = &qr.rows[0]; + + assert_eq!(row[0], serde_json::json!(true)); + assert_eq!(row[1], serde_json::json!(42)); + assert_eq!(row[2], serde_json::json!(9_223_372_036_854_775_807_i64)); + // HUGEINT → string (exceeds JSON i64 range) + assert_eq!(row[3], serde_json::json!("170141183460469231731687303715884105727")); + assert_eq!(row[4], serde_json::json!(3.5)); + // DECIMAL → string (preserves precision) + assert_eq!(row[5], serde_json::json!("12345.6789")); + assert_eq!(row[6], serde_json::json!("café résumé 🚀")); + // BLOB → Base64 + assert_eq!(row[7], serde_json::json!("3q2+7w==")); + assert_eq!(row[8], serde_json::json!("2024-01-15")); + assert!(row[9].as_str().unwrap().starts_with("13:45:30")); + assert!(row[10].as_str().unwrap().starts_with("2024-01-15 13:45:30")); + assert_eq!(row[11], serde_json::json!([1, 2, 3])); + assert_eq!(row[12], serde_json::json!({"a": 7, "b": "x"})); + assert_eq!(row[13], serde_json::Value::Null); + + let _ = std::fs::remove_file(&temp_file); + } + + #[tokio::test] + async fn test_introspect_foreign_keys() { + let temp_file = fixture_path("fk"); + let _ = std::fs::remove_file(&temp_file); + + { + let conn = Connection::open(&temp_file).expect("create"); + conn.execute_batch( + "CREATE TABLE users (id INTEGER PRIMARY KEY, name TEXT); + CREATE TABLE posts ( + id INTEGER PRIMARY KEY, + user_id INTEGER, + FOREIGN KEY (user_id) REFERENCES users(id) + )", + ) + .expect("seed"); + } + + let config = ConnectionConfig::duckdb(temp_file.clone()); + let result = DuckDbEngine::introspect( + &config, + &IntrospectOperation::TableDetails { + name: "posts".to_string(), + fields: TableFields::all(), + }, + None, + None, + ) + .await; + assert!(result.is_ok(), "introspect failed: {:?}", result.err()); + + let IntrospectResult::TableDetails { table } = result.unwrap() else { + panic!("Expected TableDetails result") + }; + + assert!(!table.foreign_keys.is_empty(), "expected foreign keys"); + let fk = &table.foreign_keys[0]; + assert_eq!(fk.referenced_table, "users"); + assert_eq!(fk.columns, vec!["user_id"]); + assert_eq!(fk.referenced_columns, vec!["id"]); + + let _ = std::fs::remove_file(&temp_file); + } + + #[tokio::test] + async fn test_introspect_indexes_and_views() { + let temp_file = fixture_path("idx"); + let _ = std::fs::remove_file(&temp_file); + + { + let conn = Connection::open(&temp_file).expect("create"); + conn.execute_batch( + "CREATE TABLE users (id INTEGER PRIMARY KEY, email TEXT); + CREATE INDEX idx_email ON users(email); + CREATE UNIQUE INDEX idx_email_unique ON users(email); + CREATE VIEW v_users AS SELECT id FROM users", + ) + .expect("seed"); + } + + let config = ConnectionConfig::duckdb(temp_file.clone()); + + // ListIndexes + let result = DuckDbEngine::introspect( + &config, + &IntrospectOperation::ListIndexes { table: Some("users".to_string()) }, + None, + None, + ) + .await + .expect("ListIndexes failed"); + let IntrospectResult::IndexList { indexes } = result else { panic!("Expected IndexList") }; + assert!(indexes.iter().any(|i| i.name == "idx_email" && !i.unique)); + let uniq = indexes.iter().find(|i| i.name == "idx_email_unique").expect("unique idx"); + assert!(uniq.unique); + assert_eq!(uniq.columns, vec!["email"]); + + // ListViews + ViewDetails + let result = DuckDbEngine::introspect(&config, &IntrospectOperation::ListViews, None, None) + .await + .expect("ListViews failed"); + let IntrospectResult::ViewList { views } = result else { panic!("Expected ViewList") }; + assert_eq!(views, vec!["v_users".to_string()]); + + let result = DuckDbEngine::introspect( + &config, + &IntrospectOperation::ViewDetails { name: "v_users".to_string() }, + None, + None, + ) + .await + .expect("ViewDetails failed"); + let IntrospectResult::ViewDetails { view } = result else { panic!("Expected ViewDetails") }; + assert_eq!(view.name, "v_users"); + assert!(view.definition.is_some()); + assert_eq!(view.columns.len(), 1); + + let _ = std::fs::remove_file(&temp_file); + } + + #[tokio::test] + async fn test_introspect_comments_and_row_estimate() { + let temp_file = fixture_path("comments"); + let _ = std::fs::remove_file(&temp_file); + + { + let conn = Connection::open(&temp_file).expect("create"); + conn.execute_batch( + "CREATE TABLE items (id INTEGER PRIMARY KEY, label TEXT); + COMMENT ON TABLE items IS 'inventory items'; + COMMENT ON COLUMN items.label IS 'display label'; + INSERT INTO items SELECT range, 'x' FROM range(5)", + ) + .expect("seed"); + } + + let config = ConnectionConfig::duckdb(temp_file.clone()); + let result = DuckDbEngine::introspect( + &config, + &IntrospectOperation::TableDetails { + name: "items".to_string(), + fields: TableFields::all(), + }, + None, + None, + ) + .await + .expect("introspect failed"); + + let IntrospectResult::TableDetails { table } = result else { + panic!("Expected TableDetails") + }; + + assert_eq!(table.comment.as_deref(), Some("inventory items")); + let label = table.columns.iter().find(|c| c.name == "label").unwrap(); + assert_eq!(label.comment.as_deref(), Some("display label")); + assert_eq!(table.row_estimate, Some(5)); + + let _ = std::fs::remove_file(&temp_file); + } + + /// Prove that writes fail at the `DuckDB` storage layer independently of + /// the parser: a connection opened with `AccessMode::ReadOnly` must reject + /// direct DML/DDL without going through `validate_query`. + #[test] + fn test_duckdb_session_read_only_enforcement() { + let temp_file = fixture_path("session_ro"); + let _ = std::fs::remove_file(&temp_file); + + { + let conn = Connection::open(&temp_file).expect("create"); + conn.execute_batch("CREATE TABLE t (id INTEGER)").expect("seed"); + } + + let conn = open_connection(temp_file.to_str().unwrap()) + .expect("Failed to open read-only connection"); + + let insert_result = conn.execute("INSERT INTO t VALUES (1)", []); + assert!(insert_result.is_err(), "INSERT must be rejected at the DuckDB storage layer"); + + let create_result = conn.execute("CREATE TABLE t2 (id INTEGER)", []); + assert!( + create_result.is_err(), + "CREATE TABLE must be rejected at the DuckDB storage layer" + ); + + let _ = std::fs::remove_file(&temp_file); + } + + #[tokio::test] + async fn test_execute_server_side_timeout_interrupt() { + // A large cross-join aggregate runs for many seconds; with a 50ms + // timeout the interrupt thread fires mid-execution and DuckDB returns + // an INTERRUPT error, surfaced as QUERY_TIMEOUT. + let config = ConnectionConfig::duckdb(":memory:".into()); + let caps = Capabilities { timeout_ms: Some(50), ..Capabilities::default() }; + let sql = "SELECT max(a.range * b.range + a.range) \ + FROM range(200000) a, range(200000) b"; + + let result = DuckDbEngine::execute(&config, sql, &[], &caps).await; + + assert!(result.is_err(), "Expected a timeout error but got Ok"); + let err = result.unwrap_err(); + assert_eq!( + err.error_code(), + "QUERY_TIMEOUT", + "Expected QUERY_TIMEOUT, got {}: {}", + err.error_code(), + err.message() + ); + } + + #[tokio::test] + async fn test_execute_structured_explain() { + let temp_file = fixture_path("explain"); + let _ = std::fs::remove_file(&temp_file); + { + let conn = Connection::open(&temp_file).expect("create"); + conn.execute_batch("CREATE TABLE t AS SELECT range AS n FROM range(100)").unwrap(); + } + + let config = ConnectionConfig::duckdb(temp_file.clone()); + let caps = Capabilities { + explain_format: Some(ExplainFormat::Structured), + ..Capabilities::default() + }; + let result = + DuckDbEngine::execute(&config, "EXPLAIN SELECT * FROM t WHERE n > 5", &[], &caps).await; + assert!(result.is_ok(), "structured explain failed: {:?}", result.err()); + let qr = result.unwrap(); + let plan = qr.plan.expect("plan missing"); + assert_eq!(plan.node_type, "QUERY PLAN"); + assert!(!plan.children.is_empty(), "plan should have children"); + + // Structured explain on a non-EXPLAIN query is rejected + let result = DuckDbEngine::execute(&config, "SELECT 1", &[], &caps).await; + assert!(result.is_err()); + + let _ = std::fs::remove_file(&temp_file); + } + + #[tokio::test] + async fn test_execute_native_explain_passthrough() { + let config = ConnectionConfig::duckdb(":memory:".into()); + let caps = Capabilities::default(); + let result = DuckDbEngine::execute(&config, "EXPLAIN SELECT 1", &[], &caps).await; + assert!(result.is_ok(), "native explain failed: {:?}", result.err()); + let qr = result.unwrap(); + assert!(qr.plan.is_none()); + assert!(!qr.rows.is_empty()); + } + + // ========================================================================= + // Parameterized query tests + // ========================================================================= + + #[tokio::test] + async fn test_execute_bound_params() { + let temp_file = fixture_path("params"); + let _ = std::fs::remove_file(&temp_file); + { + let conn = Connection::open(&temp_file).expect("create"); + conn.execute_batch( + "CREATE TABLE t (id INTEGER, name TEXT, score DOUBLE); + INSERT INTO t VALUES (1, 'alice', 9.5), (2, 'bob', 7.0), (3, 'charlie', 8.2)", + ) + .unwrap(); + } + let config = ConnectionConfig::duckdb(temp_file.clone()); + + // Integer param + let params = vec![serde_json::json!(1)]; + let qr = DuckDbEngine::execute( + &config, + "SELECT name FROM t WHERE id = ?", + ¶ms, + &Capabilities::default(), + ) + .await + .expect("bound integer param"); + assert_eq!(qr.rows.len(), 1); + assert_eq!(qr.rows[0][0], serde_json::json!("alice")); + + // Text param + let params = vec![serde_json::json!("bob")]; + let qr = DuckDbEngine::execute( + &config, + "SELECT id FROM t WHERE name = ?", + ¶ms, + &Capabilities::default(), + ) + .await + .expect("bound text param"); + assert_eq!(qr.rows.len(), 1); + assert_eq!(qr.rows[0][0], serde_json::json!(2)); + + // Multiple float params + let params = vec![serde_json::json!(7.5), serde_json::json!(9.0)]; + let qr = DuckDbEngine::execute( + &config, + "SELECT name FROM t WHERE score >= ? AND score <= ? ORDER BY score", + ¶ms, + &Capabilities::default(), + ) + .await + .expect("multiple bound params"); + assert_eq!(qr.rows.len(), 1); + assert_eq!(qr.rows[0][0], serde_json::json!("charlie")); + + let _ = std::fs::remove_file(&temp_file); + } + + #[tokio::test] + async fn test_execute_write_still_rejected_with_params() { + let config = ConnectionConfig::duckdb(":memory:".into()); + let params = vec![serde_json::json!(42), serde_json::json!("evil")]; + let caps = Capabilities::default(); + let result = + DuckDbEngine::execute(&config, "INSERT INTO t VALUES (?, ?)", ¶ms, &caps).await; + assert!(result.is_err(), "write must be rejected even with params"); + assert!(result.unwrap_err().message().contains("Plenum is read-only")); + } + + #[tokio::test] + async fn test_execute_show_describe_summarize_allowed() { + let temp_file = fixture_path("show"); + let _ = std::fs::remove_file(&temp_file); + { + let conn = Connection::open(&temp_file).expect("create"); + conn.execute_batch("CREATE TABLE t AS SELECT range AS n FROM range(10)").unwrap(); + } + let config = ConnectionConfig::duckdb(temp_file.clone()); + let caps = Capabilities::default(); + + for sql in ["SHOW TABLES", "DESCRIBE t", "SUMMARIZE t", "PRAGMA table_info('t')"] { + let result = DuckDbEngine::execute(&config, sql, &[], &caps).await; + assert!(result.is_ok(), "expected {sql} to be allowed: {:?}", result.err()); + } + + let _ = std::fs::remove_file(&temp_file); + } +} diff --git a/src/engine/mod.rs b/src/engine/mod.rs index a8b348e..4ff04a5 100644 --- a/src/engine/mod.rs +++ b/src/engine/mod.rs @@ -74,6 +74,10 @@ pub mod postgres; // Phase 4 (in progress) #[cfg(feature = "mysql")] pub mod mysql; +// DuckDB engine (REF-290) +#[cfg(feature = "duckdb")] +pub mod duckdb; + /// Supported database engine types #[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] #[serde(rename_all = "lowercase")] @@ -84,6 +88,8 @@ pub enum DatabaseType { MySQL, /// `SQLite` database SQLite, + /// `DuckDB` database + DuckDB, } impl DatabaseType { @@ -94,6 +100,7 @@ impl DatabaseType { Self::Postgres => "postgres", Self::MySQL => "mysql", Self::SQLite => "sqlite", + Self::DuckDB => "duckdb", } } } @@ -200,6 +207,21 @@ impl ConnectionConfig { tls: None, } } + + /// Create a new `DuckDB` connection config + #[must_use] + pub fn duckdb(file: PathBuf) -> Self { + Self { + engine: DatabaseType::DuckDB, + host: None, + port: None, + user: None, + password: None, + database: None, + file: Some(file), + tls: None, + } + } } /// Connection information returned after successful connection validation @@ -781,6 +803,7 @@ mod tests { assert_eq!(serde_json::to_string(&DatabaseType::Postgres).unwrap(), r#""postgres""#); assert_eq!(serde_json::to_string(&DatabaseType::MySQL).unwrap(), r#""mysql""#); assert_eq!(serde_json::to_string(&DatabaseType::SQLite).unwrap(), r#""sqlite""#); + assert_eq!(serde_json::to_string(&DatabaseType::DuckDB).unwrap(), r#""duckdb""#); } #[test] @@ -811,6 +834,11 @@ mod tests { assert_eq!(sqlite_config.engine, DatabaseType::SQLite); assert!(sqlite_config.file.is_some()); assert!(sqlite_config.tls.is_none()); + + let duckdb_config = ConnectionConfig::duckdb(PathBuf::from("/tmp/test.duckdb")); + assert_eq!(duckdb_config.engine, DatabaseType::DuckDB); + assert!(duckdb_config.file.is_some()); + assert!(duckdb_config.tls.is_none()); } #[test] diff --git a/src/main.rs b/src/main.rs index bba86d6..ec2d7c0 100644 --- a/src/main.rs +++ b/src/main.rs @@ -21,6 +21,8 @@ use plenum::{ }; // Import database engines +#[cfg(feature = "duckdb")] +use plenum::engine::duckdb::DuckDbEngine; #[cfg(feature = "mysql")] use plenum::engine::mysql::MySqlEngine; #[cfg(feature = "postgres")] @@ -923,6 +925,13 @@ async fn handle_connect_test( DatabaseType::MySQL => Err(PlenumError::invalid_input( "MySQL engine not enabled. Build with --features mysql to enable MySQL support.", )), + + #[cfg(feature = "duckdb")] + DatabaseType::DuckDB => DuckDbEngine::validate_connection(&config).await, + #[cfg(not(feature = "duckdb"))] + DatabaseType::DuckDB => Err(PlenumError::invalid_input( + "DuckDB engine not enabled. Build with --features duckdb to enable DuckDB support.", + )), }; let elapsed_ms = start.elapsed().as_millis() as u64; @@ -980,7 +989,7 @@ async fn interactive_connect_picker( config.port.unwrap_or(0) ) } - DatabaseType::SQLite => { + DatabaseType::SQLite | DatabaseType::DuckDB => { config.file.as_ref().and_then(|f| f.to_str()).unwrap_or("?").to_string() } } @@ -1020,7 +1029,7 @@ async fn interactive_connect_wizard( eprintln!("\n=== Create New Database Connection ===\n"); // Prompt for engine - let engine_choices = vec!["postgres", "mysql", "sqlite"]; + let engine_choices = vec!["postgres", "mysql", "sqlite", "duckdb"]; let engine_idx = Select::new() .with_prompt("Select database engine") .items(&engine_choices) @@ -1064,13 +1073,17 @@ async fn interactive_connect_wizard( ConnectionConfig::mysql(host, port, user, password, database) } } - DatabaseType::SQLite => { + DatabaseType::SQLite | DatabaseType::DuckDB => { let file: String = Input::new() .with_prompt("Database file path") .interact_text() .map_err(|e| PlenumError::invalid_input(format!("Input failed: {e}")))?; - ConnectionConfig::sqlite(PathBuf::from(file)) + if engine == DatabaseType::DuckDB { + ConnectionConfig::duckdb(PathBuf::from(file)) + } else { + ConnectionConfig::sqlite(PathBuf::from(file)) + } } }; @@ -1171,11 +1184,11 @@ async fn non_interactive_connect( } // Indirect sources are only meaningful for engines that use passwords. - if engine_type == DatabaseType::SQLite + if matches!(engine_type, DatabaseType::SQLite | DatabaseType::DuckDB) && (password_env.is_some() || password_command.is_some() || keychain_entry.is_some()) { return Err(PlenumError::invalid_input( - "--password-env, --password-command, and --keychain-service are not applicable to sqlite (no authentication)", + "--password-env, --password-command, and --keychain-service are not applicable to file-based engines (no authentication)", )); } @@ -1219,10 +1232,15 @@ async fn non_interactive_connect( tls, } } - DatabaseType::SQLite => { - let file = - file.ok_or_else(|| PlenumError::invalid_input("--file is required for sqlite"))?; - let mut cfg = ConnectionConfig::sqlite(file); + DatabaseType::SQLite | DatabaseType::DuckDB => { + let file = file.ok_or_else(|| { + PlenumError::invalid_input(format!("--file is required for {engine_str}")) + })?; + let mut cfg = if engine_type == DatabaseType::DuckDB { + ConnectionConfig::duckdb(file) + } else { + ConnectionConfig::sqlite(file) + }; cfg.tls = tls; cfg } @@ -1488,6 +1506,21 @@ async fn handle_introspect( DatabaseType::MySQL => Err(PlenumError::invalid_input( "MySQL engine not enabled. Build with --features mysql to enable MySQL support.", )), + + #[cfg(feature = "duckdb")] + DatabaseType::DuckDB => { + DuckDbEngine::introspect( + &config, + &operation, + target_database.as_deref(), + schema.as_deref(), + ) + .await + } + #[cfg(not(feature = "duckdb"))] + DatabaseType::DuckDB => Err(PlenumError::invalid_input( + "DuckDB engine not enabled. Build with --features duckdb to enable DuckDB support.", + )), }; match introspect_result { @@ -1693,6 +1726,17 @@ async fn handle_query( "MySQL engine not enabled. Build with --features mysql to enable MySQL support." )) } + + #[cfg(feature = "duckdb")] + DatabaseType::DuckDB => { + DuckDbEngine::execute(&config, &sql_text, ¶ms, &capabilities).await + } + #[cfg(not(feature = "duckdb"))] + DatabaseType::DuckDB => { + Err(PlenumError::invalid_input( + "DuckDB engine not enabled. Build with --features duckdb to enable DuckDB support." + )) + } }; match execute_result { @@ -1920,6 +1964,11 @@ fn build_connection_config( file.ok_or_else(|| PlenumError::invalid_input("--file is required for sqlite"))?; ConnectionConfig::sqlite(file) } + DatabaseType::DuckDB => { + let file = + file.ok_or_else(|| PlenumError::invalid_input("--file is required for duckdb"))?; + ConnectionConfig::duckdb(file) + } }; config.tls = tls; @@ -1932,8 +1981,9 @@ fn parse_engine(engine: &str) -> Result { "postgres" => Ok(DatabaseType::Postgres), "mysql" => Ok(DatabaseType::MySQL), "sqlite" => Ok(DatabaseType::SQLite), + "duckdb" => Ok(DatabaseType::DuckDB), _ => Err(PlenumError::invalid_input(format!( - "Invalid engine '{engine}'. Must be postgres, mysql, or sqlite" + "Invalid engine '{engine}'. Must be postgres, mysql, sqlite, or duckdb" ))), } } diff --git a/src/mcp.rs b/src/mcp.rs index d27f37d..c48d3bb 100644 --- a/src/mcp.rs +++ b/src/mcp.rs @@ -49,6 +49,8 @@ use std::path::PathBuf; use crate::{parse_dsn, redact_dsn, Capabilities, ConnectionConfig, DatabaseEngine, DatabaseType}; // Import database engines +#[cfg(feature = "duckdb")] +use crate::engine::duckdb::DuckDbEngine; #[cfg(feature = "mysql")] use crate::engine::mysql::MySqlEngine; #[cfg(feature = "postgres")] @@ -263,7 +265,7 @@ fn handle_list_tools() -> Result { }, "engine": { "type": "string", - "enum": ["postgres", "mysql", "sqlite"], + "enum": ["postgres", "mysql", "sqlite", "duckdb"], "description": "DISCOURAGED: Database engine type for explicit one-off connections. Only use if no saved connection exists. If omitted along with 'connection', auto-resolves project's default connection (RECOMMENDED)." }, "host": { @@ -288,7 +290,7 @@ fn handle_list_tools() -> Result { }, "file": { "type": "string", - "description": "DISCOURAGED: SQLite database file path. Only for one-off sqlite explicit connections or as override. Prefer using saved connections." + "description": "DISCOURAGED: SQLite/DuckDB database file path. Only for one-off sqlite/duckdb explicit connections or as override. Prefer using saved connections." }, "list_databases": { "type": "boolean", @@ -373,7 +375,7 @@ fn handle_list_tools() -> Result { }, "engine": { "type": "string", - "enum": ["postgres", "mysql", "sqlite"], + "enum": ["postgres", "mysql", "sqlite", "duckdb"], "description": "DISCOURAGED: Database engine type for explicit one-off connections. Only use if no saved connection exists. Valid values: 'postgres', 'mysql', 'sqlite'. If omitted along with 'connection', auto-resolves project's default connection (RECOMMENDED)." }, "host": { @@ -398,7 +400,7 @@ fn handle_list_tools() -> Result { }, "file": { "type": "string", - "description": "DISCOURAGED: File path to SQLite database file. Only for one-off sqlite explicit connections. Can be relative or absolute path. Example: './app.db', '/var/lib/data.db'. Prefer using saved connections." + "description": "DISCOURAGED: File path to SQLite/DuckDB database file. Only for one-off sqlite/duckdb explicit connections. Can be relative or absolute path. Example: './app.db', '/var/lib/data.duckdb'. Prefer using saved connections." }, "max_rows": { "type": "number", @@ -441,7 +443,7 @@ fn handle_list_tools() -> Result { }, "engine": { "type": "string", - "enum": ["postgres", "mysql", "sqlite"], + "enum": ["postgres", "mysql", "sqlite", "duckdb"], "description": "DISCOURAGED: Database engine type for explicit one-off connection tests. Only use if no saved connection exists." }, "host": { @@ -466,7 +468,7 @@ fn handle_list_tools() -> Result { }, "file": { "type": "string", - "description": "DISCOURAGED: SQLite database file path. Only for explicit one-off tests." + "description": "DISCOURAGED: SQLite/DuckDB database file path. Only for explicit one-off tests." } } } @@ -529,6 +531,15 @@ async fn tool_connect(args: &Value) -> Result { DatabaseType::MySQL => { return Err(anyhow!("MySQL engine not enabled. Build with --features mysql")); } + + #[cfg(feature = "duckdb")] + DatabaseType::DuckDB => DuckDbEngine::validate_connection(&config) + .await + .map_err(|e| anyhow!("DuckDB connection test failed: {e}"))?, + #[cfg(not(feature = "duckdb"))] + DatabaseType::DuckDB => { + return Err(anyhow!("DuckDB engine not enabled. Build with --features duckdb")); + } }; CallToolResult::success(connection_info) @@ -592,6 +603,15 @@ async fn tool_introspect(args: &Value) -> Result { DatabaseType::MySQL => { return Err(anyhow!("MySQL engine not enabled. Build with --features mysql")); } + + #[cfg(feature = "duckdb")] + DatabaseType::DuckDB => DuckDbEngine::introspect(&config, &operation, database, schema) + .await + .map_err(|e| anyhow!("DuckDB introspection failed: {e}"))?, + #[cfg(not(feature = "duckdb"))] + DatabaseType::DuckDB => { + return Err(anyhow!("DuckDB engine not enabled. Build with --features duckdb")); + } }; CallToolResult::success(result) @@ -755,7 +775,8 @@ fn build_connection_config_from_args(args: &Value, engine_str: &str) -> Result DatabaseType::Postgres, "mysql" => DatabaseType::MySQL, "sqlite" => DatabaseType::SQLite, - _ => return Err(anyhow!("Invalid engine. Must be postgres, mysql, or sqlite")), + "duckdb" => DatabaseType::DuckDB, + _ => return Err(anyhow!("Invalid engine. Must be postgres, mysql, sqlite, or duckdb")), }; match engine_type { @@ -793,6 +814,12 @@ fn build_connection_config_from_args(args: &Value, engine_str: &str) -> Result { + let file_str = args["file"] + .as_str() + .ok_or_else(|| anyhow!("Missing required field for duckdb: file"))?; + Ok(ConnectionConfig::duckdb(PathBuf::from(file_str))) + } } } @@ -836,6 +863,7 @@ fn resolve_connection_from_args(args: &Value) -> Result<(ConnectionConfig, bool) "postgres" => DatabaseType::Postgres, "mysql" => DatabaseType::MySQL, "sqlite" => DatabaseType::SQLite, + "duckdb" => DatabaseType::DuckDB, _ => return Err(anyhow!("Invalid engine: {eng}")), }; } @@ -912,6 +940,15 @@ async fn validate_connection(config: &ConnectionConfig) -> Result { Err(anyhow!("MySQL engine not enabled. Build with --features mysql")) } + + #[cfg(feature = "duckdb")] + DatabaseType::DuckDB => DuckDbEngine::validate_connection(config) + .await + .map_err(|e| anyhow!("DuckDB connection failed: {e}")), + #[cfg(not(feature = "duckdb"))] + DatabaseType::DuckDB => { + Err(anyhow!("DuckDB engine not enabled. Build with --features duckdb")) + } } } @@ -951,5 +988,14 @@ async fn execute_query( DatabaseType::MySQL => { Err(anyhow!("MySQL engine not enabled. Build with --features mysql")) } + + #[cfg(feature = "duckdb")] + DatabaseType::DuckDB => DuckDbEngine::execute(config, sql, &[], capabilities) + .await + .map_err(|e| anyhow!("DuckDB query failed: {e}")), + #[cfg(not(feature = "duckdb"))] + DatabaseType::DuckDB => { + Err(anyhow!("DuckDB engine not enabled. Build with --features duckdb")) + } } } From f26221f0c18c5353e0bdd2dc5fee0afc204acc8d Mon Sep 17 00:00:00 2001 From: therecluse26 Date: Sun, 19 Jul 2026 22:09:56 -0400 Subject: [PATCH 2/4] test: add DuckDB offline parity suite (REF-290) 40 integration tests mirroring the SQLite parity matrix (REF-278): connect, introspection (composite PK/FK, unique indexes, views, schemas, native types), allowed reads (SELECT/EXPLAIN/SHOW/DESCRIBE/ SUMMARIZE/PRAGMA/transactions), denied writes with state-unchanged verification, max_rows/offset/timeout safety, and JSON envelope determinism. Runs offline - no Docker required. Co-Authored-By: Claude Fable 5 Co-Authored-By: Paperclip --- tests/duckdb_parity.rs | 1083 ++++++++++++++++++++++++++++++++++++++++ 1 file changed, 1083 insertions(+) create mode 100644 tests/duckdb_parity.rs diff --git a/tests/duckdb_parity.rs b/tests/duckdb_parity.rs new file mode 100644 index 0000000..a40e74f --- /dev/null +++ b/tests/duckdb_parity.rs @@ -0,0 +1,1083 @@ +//! `DuckDB` offline parity suite — REF-290. +//! +//! Brings the `DuckDB` test coverage to parity with the `SQLite` offline suite +//! (REF-278) and the live-DB coverage matrix (`MySQL` / `PostgreSQL`). The +//! fixture dataset mirrors the logical dataset from the live seed scripts: +//! `type_matrix`, customers, orders, `order_items`, `bulk_rows`, `v_order_totals`. +//! +//! All tests run offline — no Docker required. Plain `cargo test` includes +//! them all. Uses the `DuckDbEngine` API directly; no CLI binary is spawned. +//! +//! Coverage matrix: +//! connect — valid path; nonexistent file → normalized error envelope +//! introspect — tables, columns + native types, PK, composite FK, indexes, +//! views, schemas, comments, row estimates; stable JSON shape +//! query allowed — SELECT, EXPLAIN, EXPLAIN (FORMAT JSON), SHOW, DESCRIBE, +//! SUMMARIZE, PRAGMA allowlist, transaction control +//! query denied — INSERT / UPDATE / DELETE / CREATE / DROP / ALTER / COPY / +//! ATTACH → `CAPABILITY_VIOLATION` before execution, then +//! re-query to prove DB state unchanged +//! safety — `max_rows` truncation + `rows_truncated` flag; `timeout_ms` +//! (interrupt handle) tested for fast and long queries +//! envelope — `QueryResult` / `IntrospectResult` serialize to valid JSON; +//! deterministic with `execution_ms` excluded + +#![cfg(feature = "duckdb")] + +use plenum::engine::duckdb::DuckDbEngine; +use plenum::engine::{IntrospectOperation, IntrospectResult, TableFields}; +use plenum::{Capabilities, ConnectionConfig, DatabaseEngine}; +use std::path::PathBuf; +use std::sync::atomic::{AtomicU64, Ordering}; + +// ============================================================================ +// Fixture helpers +// ============================================================================ + +static FIXTURE_COUNTER: AtomicU64 = AtomicU64::new(0); + +fn fixture_path(tag: &str) -> PathBuf { + let id = FIXTURE_COUNTER.fetch_add(1, Ordering::SeqCst); + let pid = std::process::id(); + std::env::temp_dir().join(format!("plenum_duckdb_parity_{tag}_{pid}_{id}.duckdb")) +} + +/// Build the full parity fixture dataset into a fresh temp file. +/// +/// Mirrors the logical dataset from the `MySQL` / `PostgreSQL` seed scripts +/// and the `SQLite` parity fixture (REF-278), using `DuckDB` native types: +/// - `type_matrix` — one column per interesting `DuckDB` type +/// - `customers` — simple PK + UNIQUE index on email, emoji in data +/// - `orders` — composite PK, FK → customers +/// - `order_items` — composite 3-col PK, composite FK → orders, index on sku +/// - `bulk_rows` — 1 500 deterministic rows for `max_rows` tests +/// - `v_order_totals` — view over orders + `order_items` +fn build_parity_fixture() -> PathBuf { + use duckdb::Connection; + + let path = fixture_path("fixture"); + let _ = std::fs::remove_file(&path); + + let conn = Connection::open(&path).expect("create fixture DB"); + + // ------------------------------------------------------------------ + // type_matrix — one column per interesting DuckDB type. Mirrors the + // logical columns from the live seeds (INT, DOUBLE, DECIMAL, TEXT, BLOB, + // nullable TEXT, DATE / TIME / TIMESTAMP as native types) plus + // DuckDB-specific HUGEINT, BOOLEAN, and LIST. + // ------------------------------------------------------------------ + conn.execute_batch( + "CREATE TABLE type_matrix ( + id INTEGER PRIMARY KEY, + c_integer BIGINT, + c_double DOUBLE, + c_decimal DECIMAL(18,4), + c_hugeint HUGEINT, + c_bool BOOLEAN, + c_text VARCHAR, + c_blob BLOB, + c_null_col VARCHAR, + c_date DATE, + c_time TIME, + c_datetime TIMESTAMP, + c_list INTEGER[] + ); + + -- Row 1: boundary / positive values, emoji string, BLOB bytes + INSERT INTO type_matrix VALUES ( + 1, + 9223372036854775807, + 2.123456789012345, + 12345678.9999, + 170141183460469231731687303715884105727, + TRUE, + 'café résumé 🚀', + '\\xDE\\xAD\\xBE\\xEF'::BLOB, + NULL, + DATE '2024-01-15', + TIME '13:45:30', + TIMESTAMP '2024-01-15 13:45:30', + [1, 2, 3] + ); + + -- Row 2: negative / small values + INSERT INTO type_matrix VALUES ( + 2, + -9223372036854775807, + -6.62607015, + -0.0001, + -170141183460469231731687303715884105727, + FALSE, + 'plain ascii', + '\\x00\\x01\\x02\\x03'::BLOB, + NULL, + DATE '1999-12-31', + TIME '00:00:00', + TIMESTAMP '1999-12-31 23:59:59', + [] + ); + + -- Row 3: all-NULL except id + INSERT INTO type_matrix (id) VALUES (3);", + ) + .expect("create + seed type_matrix"); + + // ------------------------------------------------------------------ + // customers — PK on id, UNIQUE index on email, emoji in data + // ------------------------------------------------------------------ + conn.execute_batch( + "CREATE TABLE customers ( + id INTEGER NOT NULL, + name VARCHAR NOT NULL, + email VARCHAR NOT NULL, + PRIMARY KEY (id) + ); + CREATE UNIQUE INDEX uq_customers_email ON customers(email); + INSERT INTO customers (id, name, email) VALUES + (1, 'Ada Lovelace', 'ada@example.com'), + (2, 'Grace Hopper 🌟', 'grace@example.com'), + (3, 'Annie Easley', 'annie@example.com');", + ) + .expect("create + seed customers"); + + // ------------------------------------------------------------------ + // orders — composite PK (customer_id, order_no), FK → customers + // ------------------------------------------------------------------ + conn.execute_batch( + "CREATE TABLE orders ( + customer_id INTEGER NOT NULL, + order_no INTEGER NOT NULL, + status VARCHAR NOT NULL DEFAULT 'pending', + placed_at TIMESTAMP NOT NULL, + PRIMARY KEY (customer_id, order_no), + FOREIGN KEY (customer_id) REFERENCES customers(id) + ); + INSERT INTO orders (customer_id, order_no, status, placed_at) VALUES + (1, 1, 'shipped', TIMESTAMP '2024-02-01 09:00:00'), + (1, 2, 'pending', TIMESTAMP '2024-02-03 10:30:00'), + (2, 1, 'cancelled', TIMESTAMP '2024-02-05 16:45:00');", + ) + .expect("create + seed orders"); + + // ------------------------------------------------------------------ + // order_items — 3-col composite PK, composite FK → orders, index on sku + // ------------------------------------------------------------------ + conn.execute_batch( + "CREATE TABLE order_items ( + customer_id INTEGER NOT NULL, + order_no INTEGER NOT NULL, + line_no INTEGER NOT NULL, + sku VARCHAR NOT NULL, + qty INTEGER NOT NULL, + unit_price DECIMAL(10,2) NOT NULL, + PRIMARY KEY (customer_id, order_no, line_no), + FOREIGN KEY (customer_id, order_no) + REFERENCES orders(customer_id, order_no) + ); + CREATE INDEX idx_order_items_sku ON order_items(sku); + INSERT INTO order_items + (customer_id, order_no, line_no, sku, qty, unit_price) + VALUES + (1, 1, 1, 'SKU-0001', 2, 19.99), + (1, 1, 2, 'SKU-0002', 1, 5.00), + (1, 2, 1, 'SKU-0003', 4, 2.50), + (2, 1, 1, 'SKU-0001', 1, 19.99);", + ) + .expect("create + seed order_items"); + + // ------------------------------------------------------------------ + // bulk_rows — 1 500 deterministic rows for max_rows truncation tests + // ------------------------------------------------------------------ + conn.execute_batch( + "CREATE TABLE bulk_rows AS + SELECT CAST(range + 1 AS INTEGER) AS n, + printf('row-%04d', range + 1) AS label + FROM range(1500);", + ) + .expect("create + seed bulk_rows"); + + // ------------------------------------------------------------------ + // v_order_totals — view over orders + order_items + // ------------------------------------------------------------------ + conn.execute_batch( + "CREATE VIEW v_order_totals AS + SELECT o.customer_id, + o.order_no, + o.status, + SUM(i.qty * i.unit_price) AS total + FROM orders o + JOIN order_items i + ON i.customer_id = o.customer_id + AND i.order_no = o.order_no + GROUP BY o.customer_id, o.order_no, o.status;", + ) + .expect("create v_order_totals view"); + + path +} + +fn cleanup(path: &PathBuf) { + let _ = std::fs::remove_file(path); +} + +/// Return the value of column `name` from `row`, panicking with a useful +/// message if the column is absent. +fn get_col<'a>(cols: &[String], row: &'a [serde_json::Value], name: &str) -> &'a serde_json::Value { + let idx = cols + .iter() + .position(|c| c == name) + .unwrap_or_else(|| panic!("column '{name}' not found in {cols:?}")); + &row[idx] +} + +// ============================================================================ +// Connect +// ============================================================================ + +#[tokio::test] +async fn parity_connect_valid_path() { + let path = build_parity_fixture(); + let config = ConnectionConfig::duckdb(path.clone()); + let result = DuckDbEngine::validate_connection(&config).await; + assert!(result.is_ok(), "validate_connection failed: {:?}", result.err()); + let info = result.unwrap(); + assert!(!info.database_version.is_empty(), "database_version must not be empty"); + assert!(info.server_info.contains("DuckDB"), "server_info must mention DuckDB"); + assert!(!info.connected_database.is_empty(), "connected_database must not be empty"); + cleanup(&path); +} + +#[tokio::test] +async fn parity_connect_nonexistent_file() { + let path = PathBuf::from("/nonexistent/plenum_parity_test.duckdb"); + let config = ConnectionConfig::duckdb(path); + let result = DuckDbEngine::validate_connection(&config).await; + assert!(result.is_err(), "expected connection failure for nonexistent path"); + let err = result.unwrap_err(); + assert_eq!( + err.error_code(), + "CONNECTION_FAILED", + "nonexistent file must produce CONNECTION_FAILED, got: {}", + err.error_code() + ); +} + +// ============================================================================ +// Introspect — tables +// ============================================================================ + +#[tokio::test] +async fn parity_introspect_list_tables() { + let path = build_parity_fixture(); + let config = ConnectionConfig::duckdb(path.clone()); + let result = + DuckDbEngine::introspect(&config, &IntrospectOperation::ListTables, None, None).await; + assert!(result.is_ok(), "ListTables failed: {:?}", result.err()); + let IntrospectResult::TableList { tables } = result.unwrap() else { + panic!("Expected TableList variant"); + }; + for expected in &["type_matrix", "customers", "orders", "order_items", "bulk_rows"] { + assert!( + tables.iter().any(|t| t == expected), + "table '{expected}' missing from ListTables result; got: {tables:?}" + ); + } + cleanup(&path); +} + +#[tokio::test] +async fn parity_introspect_type_matrix_columns_and_types() { + let path = build_parity_fixture(); + let config = ConnectionConfig::duckdb(path.clone()); + let result = DuckDbEngine::introspect( + &config, + &IntrospectOperation::TableDetails { + name: "type_matrix".to_string(), + fields: TableFields::all(), + }, + None, + None, + ) + .await; + assert!(result.is_ok(), "TableDetails(type_matrix) failed: {:?}", result.err()); + let IntrospectResult::TableDetails { table } = result.unwrap() else { + panic!("Expected TableDetails variant"); + }; + + assert_eq!(table.name, "type_matrix"); + assert_eq!(table.schema.as_deref(), Some("main")); + + let col_type: std::collections::HashMap<&str, &str> = + table.columns.iter().map(|c| (c.name.as_str(), c.data_type.as_str())).collect(); + + // Verify native DuckDB types are reported verbatim + assert_eq!(col_type.get("c_integer").copied(), Some("BIGINT")); + assert_eq!(col_type.get("c_double").copied(), Some("DOUBLE")); + assert_eq!(col_type.get("c_decimal").copied(), Some("DECIMAL(18,4)")); + assert_eq!(col_type.get("c_hugeint").copied(), Some("HUGEINT")); + assert_eq!(col_type.get("c_bool").copied(), Some("BOOLEAN")); + assert_eq!(col_type.get("c_text").copied(), Some("VARCHAR")); + assert_eq!(col_type.get("c_blob").copied(), Some("BLOB")); + assert_eq!(col_type.get("c_date").copied(), Some("DATE")); + assert_eq!(col_type.get("c_datetime").copied(), Some("TIMESTAMP")); + assert_eq!(col_type.get("c_list").copied(), Some("INTEGER[]")); + + // PK must be reported + assert_eq!( + table.primary_key.as_deref(), + Some(["id".to_string()].as_slice()), + "type_matrix PK must be [id]" + ); + + // c_null_col has no NOT NULL constraint → nullable + let null_col = + table.columns.iter().find(|c| c.name == "c_null_col").expect("c_null_col column"); + assert!(null_col.nullable, "c_null_col must be nullable"); + + cleanup(&path); +} + +#[tokio::test] +async fn parity_introspect_customers_pk_and_unique_index() { + let path = build_parity_fixture(); + let config = ConnectionConfig::duckdb(path.clone()); + let result = DuckDbEngine::introspect( + &config, + &IntrospectOperation::TableDetails { + name: "customers".to_string(), + fields: TableFields::all(), + }, + None, + None, + ) + .await + .expect("TableDetails(customers) failed"); + let IntrospectResult::TableDetails { table } = result else { panic!("Expected TableDetails") }; + + assert_eq!( + table.primary_key.as_deref(), + Some(["id".to_string()].as_slice()), + "customers PK must be [id]" + ); + + // email must have a UNIQUE index + let email_idx = table.indexes.iter().find(|i| i.columns.contains(&"email".to_string())); + assert!(email_idx.is_some(), "expected a UNIQUE index on email; indexes: {:?}", table.indexes); + assert!(email_idx.unwrap().unique, "email index must be unique"); + + cleanup(&path); +} + +#[tokio::test] +async fn parity_introspect_orders_composite_pk_and_fk() { + let path = build_parity_fixture(); + let config = ConnectionConfig::duckdb(path.clone()); + let result = DuckDbEngine::introspect( + &config, + &IntrospectOperation::TableDetails { + name: "orders".to_string(), + fields: TableFields::all(), + }, + None, + None, + ) + .await + .expect("TableDetails(orders) failed"); + let IntrospectResult::TableDetails { table } = result else { panic!("Expected TableDetails") }; + + let pk = table.primary_key.as_ref().expect("orders must have a PK"); + assert!( + pk.contains(&"customer_id".to_string()) && pk.contains(&"order_no".to_string()), + "composite PK must include customer_id and order_no; got {pk:?}" + ); + + let fk = table + .foreign_keys + .iter() + .find(|fk| fk.referenced_table == "customers") + .expect("expected FK referencing customers"); + assert!( + fk.columns.contains(&"customer_id".to_string()), + "FK must include customer_id; got {:?}", + fk.columns + ); + + cleanup(&path); +} + +#[tokio::test] +async fn parity_introspect_order_items_composite_fk_and_index() { + let path = build_parity_fixture(); + let config = ConnectionConfig::duckdb(path.clone()); + let result = DuckDbEngine::introspect( + &config, + &IntrospectOperation::TableDetails { + name: "order_items".to_string(), + fields: TableFields::all(), + }, + None, + None, + ) + .await + .expect("TableDetails(order_items) failed"); + let IntrospectResult::TableDetails { table } = result else { panic!("Expected TableDetails") }; + + // 3-column composite PK + let pk = table.primary_key.as_ref().expect("order_items must have a PK"); + assert_eq!(pk.len(), 3, "composite PK must have 3 columns; got {pk:?}"); + for col in &["customer_id", "order_no", "line_no"] { + assert!(pk.contains(&(*col).to_string()), "composite PK must contain {col}"); + } + + // Composite FK to orders + let fk = table + .foreign_keys + .iter() + .find(|fk| fk.referenced_table == "orders") + .expect("expected composite FK referencing orders"); + assert_eq!(fk.columns.len(), 2, "composite FK must have 2 columns; got {:?}", fk.columns); + + // Index on sku + assert!( + table.indexes.iter().any(|i| i.columns.contains(&"sku".to_string())), + "expected an index on sku; indexes: {:?}", + table.indexes + ); + + cleanup(&path); +} + +#[tokio::test] +async fn parity_introspect_list_views() { + let path = build_parity_fixture(); + let config = ConnectionConfig::duckdb(path.clone()); + let result = + DuckDbEngine::introspect(&config, &IntrospectOperation::ListViews, None, None).await; + assert!(result.is_ok(), "ListViews failed: {:?}", result.err()); + let IntrospectResult::ViewList { views } = result.unwrap() else { + panic!("Expected ViewList variant"); + }; + assert!( + views.contains(&"v_order_totals".to_string()), + "v_order_totals missing from view list; got: {views:?}" + ); + cleanup(&path); +} + +#[tokio::test] +async fn parity_introspect_view_details() { + let path = build_parity_fixture(); + let config = ConnectionConfig::duckdb(path.clone()); + let result = DuckDbEngine::introspect( + &config, + &IntrospectOperation::ViewDetails { name: "v_order_totals".to_string() }, + None, + None, + ) + .await; + assert!(result.is_ok(), "ViewDetails failed: {:?}", result.err()); + let IntrospectResult::ViewDetails { view } = result.unwrap() else { + panic!("Expected ViewDetails variant"); + }; + assert_eq!(view.name, "v_order_totals"); + assert!(view.definition.is_some(), "view definition must be present"); + assert!( + view.definition.as_ref().unwrap().contains("order_items"), + "view definition must reference order_items" + ); + assert!(!view.columns.is_empty(), "view must report its columns"); + cleanup(&path); +} + +#[tokio::test] +async fn parity_introspect_list_indexes_for_table() { + let path = build_parity_fixture(); + let config = ConnectionConfig::duckdb(path.clone()); + let result = DuckDbEngine::introspect( + &config, + &IntrospectOperation::ListIndexes { table: Some("order_items".to_string()) }, + None, + None, + ) + .await; + assert!(result.is_ok(), "ListIndexes(order_items) failed: {:?}", result.err()); + let IntrospectResult::IndexList { indexes } = result.unwrap() else { + panic!("Expected IndexList variant"); + }; + assert!( + indexes.iter().any(|i| i.columns.contains(&"sku".to_string())), + "expected sku index in ListIndexes result; got: {indexes:?}" + ); + cleanup(&path); +} + +#[tokio::test] +async fn parity_introspect_list_schemas_and_databases() { + // DuckDB supports schemas (unlike SQLite) — ListSchemas and ListDatabases + // are first-class operations. + let path = build_parity_fixture(); + let config = ConnectionConfig::duckdb(path.clone()); + + let result = DuckDbEngine::introspect(&config, &IntrospectOperation::ListSchemas, None, None) + .await + .expect("ListSchemas failed"); + let IntrospectResult::SchemaList { schemas } = result else { panic!("Expected SchemaList") }; + assert!(schemas.contains(&"main".to_string()), "main schema must be listed; got {schemas:?}"); + + let result = DuckDbEngine::introspect(&config, &IntrospectOperation::ListDatabases, None, None) + .await + .expect("ListDatabases failed"); + let IntrospectResult::DatabaseList { databases } = result else { + panic!("Expected DatabaseList") + }; + assert!(!databases.is_empty(), "at least the connected database must be listed"); + + cleanup(&path); +} + +#[tokio::test] +async fn parity_introspect_stable_json_shape() { + // Successive introspections must produce identical JSON (determinism). + let path = build_parity_fixture(); + let config = ConnectionConfig::duckdb(path.clone()); + + let r1 = DuckDbEngine::introspect(&config, &IntrospectOperation::ListTables, None, None) + .await + .unwrap(); + let r2 = DuckDbEngine::introspect(&config, &IntrospectOperation::ListTables, None, None) + .await + .unwrap(); + + let j1 = serde_json::to_string(&r1).expect("serialize r1"); + let j2 = serde_json::to_string(&r2).expect("serialize r2"); + assert_eq!(j1, j2, "successive introspect results must be identical"); + + cleanup(&path); +} + +// ============================================================================ +// Query — allowed operations +// ============================================================================ + +#[tokio::test] +async fn parity_query_select_type_matrix_numeric_null_blob() { + // Verify numeric types, emoji string, NULLs, BLOBs, HUGEINT, DECIMAL, and + // LIST values survive the round-trip. + let path = build_parity_fixture(); + let config = ConnectionConfig::duckdb(path.clone()); + let caps = Capabilities::default(); + let result = DuckDbEngine::execute( + &config, + "SELECT c_integer, c_double, c_decimal, c_hugeint, c_bool, c_text, c_blob, \ + c_null_col, c_list \ + FROM type_matrix ORDER BY id", + &[], + &caps, + ) + .await; + assert!(result.is_ok(), "SELECT type_matrix failed: {:?}", result.err()); + let qr = result.unwrap(); + assert_eq!(qr.rows.len(), 3, "expected 3 rows"); + assert!(qr.rows_affected.is_none(), "SELECT must not set rows_affected"); + + // Row 1: non-null values + let r1 = &qr.rows[0]; + assert!(get_col(&qr.columns, r1, "c_integer").is_number(), "c_integer must be a number"); + assert!(get_col(&qr.columns, r1, "c_double").is_number(), "c_double must be a number"); + // DECIMAL and HUGEINT are stringified to preserve precision + assert_eq!(get_col(&qr.columns, r1, "c_decimal"), &serde_json::json!("12345678.9999")); + assert_eq!( + get_col(&qr.columns, r1, "c_hugeint"), + &serde_json::json!("170141183460469231731687303715884105727") + ); + assert_eq!(get_col(&qr.columns, r1, "c_bool"), &serde_json::json!(true)); + let text = get_col(&qr.columns, r1, "c_text").as_str().expect("c_text must be string"); + assert!(text.contains('🚀'), "emoji must survive round-trip; got: {text:?}"); + // BLOB must be base64-encoded + assert_eq!(get_col(&qr.columns, r1, "c_blob"), &serde_json::json!("3q2+7w==")); + assert!(get_col(&qr.columns, r1, "c_null_col").is_null(), "c_null_col must be JSON null"); + assert_eq!(get_col(&qr.columns, r1, "c_list"), &serde_json::json!([1, 2, 3])); + + // Row 3: all-NULL + let r3 = &qr.rows[2]; + for col in ["c_integer", "c_double", "c_decimal", "c_hugeint", "c_text", "c_blob", "c_list"] { + assert!(get_col(&qr.columns, r3, col).is_null(), "row-3 {col} must be null"); + } + + cleanup(&path); +} + +#[tokio::test] +async fn parity_query_select_customers_emoji_survives() { + let path = build_parity_fixture(); + let config = ConnectionConfig::duckdb(path.clone()); + let caps = Capabilities::default(); + let qr = + DuckDbEngine::execute(&config, "SELECT id, name FROM customers ORDER BY id", &[], &caps) + .await + .expect("SELECT customers failed"); + assert_eq!(qr.rows.len(), 3); + let name2 = get_col(&qr.columns, &qr.rows[1], "name").as_str().unwrap(); + assert!(name2.contains('🌟'), "emoji in customer name must survive; got: {name2:?}"); + cleanup(&path); +} + +#[tokio::test] +async fn parity_query_select_view() { + let path = build_parity_fixture(); + let config = ConnectionConfig::duckdb(path.clone()); + let caps = Capabilities::default(); + let result = DuckDbEngine::execute( + &config, + "SELECT customer_id, order_no, total \ + FROM v_order_totals ORDER BY customer_id, order_no", + &[], + &caps, + ) + .await; + assert!(result.is_ok(), "SELECT from view failed: {:?}", result.err()); + assert!(!result.unwrap().rows.is_empty(), "v_order_totals must return rows"); + cleanup(&path); +} + +#[tokio::test] +async fn parity_query_explain_select_allowed() { + let path = build_parity_fixture(); + let config = ConnectionConfig::duckdb(path.clone()); + let caps = Capabilities::default(); + let result = + DuckDbEngine::execute(&config, "EXPLAIN SELECT * FROM customers", &[], &caps).await; + assert!(result.is_ok(), "EXPLAIN SELECT must be allowed: {:?}", result.err()); + assert!(!result.unwrap().rows.is_empty(), "EXPLAIN must return plan rows"); + cleanup(&path); +} + +#[tokio::test] +async fn parity_query_structured_explain() { + // DuckDB structured explain uses EXPLAIN (FORMAT JSON) and normalizes to + // the engine-stable ExplainPlanNode tree. + let path = build_parity_fixture(); + let config = ConnectionConfig::duckdb(path.clone()); + let caps = Capabilities { + explain_format: Some(plenum::ExplainFormat::Structured), + ..Capabilities::default() + }; + let result = + DuckDbEngine::execute(&config, "EXPLAIN SELECT * FROM customers WHERE id = 1", &[], &caps) + .await; + assert!(result.is_ok(), "structured EXPLAIN failed: {:?}", result.err()); + let qr = result.unwrap(); + let plan = qr.plan.expect("structured explain must populate plan"); + assert_eq!(plan.node_type, "QUERY PLAN"); + assert!(!plan.children.is_empty(), "plan must have child nodes"); + cleanup(&path); +} + +#[tokio::test] +async fn parity_query_show_describe_summarize_allowed() { + let path = build_parity_fixture(); + let config = ConnectionConfig::duckdb(path.clone()); + let caps = Capabilities::default(); + + for sql in ["SHOW TABLES", "DESCRIBE customers", "SUMMARIZE customers"] { + let result = DuckDbEngine::execute(&config, sql, &[], &caps).await; + assert!(result.is_ok(), "'{sql}' must be allowed: {:?}", result.err()); + assert!(!result.unwrap().rows.is_empty(), "'{sql}' must return rows"); + } + cleanup(&path); +} + +#[tokio::test] +async fn parity_query_pragma_table_info_allowed() { + let path = build_parity_fixture(); + let config = ConnectionConfig::duckdb(path.clone()); + let caps = Capabilities::default(); + let result = DuckDbEngine::execute(&config, "PRAGMA table_info('customers')", &[], &caps).await; + assert!(result.is_ok(), "PRAGMA table_info must be allowed: {:?}", result.err()); + let qr = result.unwrap(); + assert!(!qr.rows.is_empty(), "PRAGMA table_info must return column rows"); + cleanup(&path); +} + +#[tokio::test] +async fn parity_query_pragma_database_list_allowed() { + let path = build_parity_fixture(); + let config = ConnectionConfig::duckdb(path.clone()); + let caps = Capabilities::default(); + let result = DuckDbEngine::execute(&config, "PRAGMA database_list", &[], &caps).await; + assert!(result.is_ok(), "PRAGMA database_list must be allowed: {:?}", result.err()); + cleanup(&path); +} + +#[tokio::test] +async fn parity_query_pragma_setter_rejected() { + let path = build_parity_fixture(); + let config = ConnectionConfig::duckdb(path.clone()); + let caps = Capabilities::default(); + let result = DuckDbEngine::execute(&config, "PRAGMA memory_limit='1GB'", &[], &caps).await; + assert!(result.is_err(), "setter PRAGMA must be rejected"); + assert_eq!(result.unwrap_err().error_code(), "CAPABILITY_VIOLATION"); + cleanup(&path); +} + +/// Verify that a transaction control statement is NOT rejected as a +/// `CAPABILITY_VIOLATION` by the Plenum capability checker. `DuckDB`-level +/// errors (e.g. "no transaction is active" on ROLLBACK without BEGIN) are +/// acceptable here; only `CAPABILITY_VIOLATION` is forbidden. +async fn assert_not_capability_violation(config: &ConnectionConfig, sql: &str) { + let result = DuckDbEngine::execute(config, sql, &[], &Capabilities::default()).await; + if let Err(ref err) = result { + assert_ne!( + err.error_code(), + "CAPABILITY_VIOLATION", + "'{sql}' must not be rejected as CAPABILITY_VIOLATION; got error_code={}", + err.error_code() + ); + } +} + +#[tokio::test] +async fn parity_query_transaction_begin_not_capability_violation() { + let path = build_parity_fixture(); + let config = ConnectionConfig::duckdb(path.clone()); + let result = DuckDbEngine::execute(&config, "BEGIN", &[], &Capabilities::default()).await; + assert!( + result.is_ok(), + "BEGIN must succeed on a read-only DuckDB connection: {:?}", + result.err() + ); + cleanup(&path); +} + +#[tokio::test] +async fn parity_query_transaction_control_not_capability_violations() { + let path = build_parity_fixture(); + let config = ConnectionConfig::duckdb(path.clone()); + assert_not_capability_violation(&config, "COMMIT").await; + assert_not_capability_violation(&config, "ROLLBACK").await; + cleanup(&path); +} + +// ============================================================================ +// Query — denied (CAPABILITY_VIOLATION) +// ============================================================================ + +/// Assert that `denied_sql` is rejected with `CAPABILITY_VIOLATION` and that +/// a follow-up `verify_sql` returns exactly `expected_rows`, proving the DB +/// was not mutated. +async fn assert_capability_violation_and_state_unchanged( + config: &ConnectionConfig, + denied_sql: &str, + verify_sql: &str, + expected_rows: usize, +) { + let caps = Capabilities::default(); + + let err_result = DuckDbEngine::execute(config, denied_sql, &[], &caps).await; + assert!(err_result.is_err(), "expected an error for: {denied_sql}"); + let err = err_result.unwrap_err(); + assert_eq!( + err.error_code(), + "CAPABILITY_VIOLATION", + "'{denied_sql}' must produce CAPABILITY_VIOLATION; got: {}", + err.error_code() + ); + + // Re-query to prove state is unchanged + let verify = DuckDbEngine::execute(config, verify_sql, &[], &caps) + .await + .unwrap_or_else(|e| panic!("verify SELECT failed after denied write: {e:?}")); + assert_eq!( + verify.rows.len(), + expected_rows, + "DB state must be unchanged after denied '{denied_sql}': \ + expected {expected_rows} rows from '{verify_sql}'" + ); +} + +#[tokio::test] +async fn parity_denied_insert_capability_violation() { + let path = build_parity_fixture(); + let config = ConnectionConfig::duckdb(path.clone()); + assert_capability_violation_and_state_unchanged( + &config, + "INSERT INTO customers (id, name, email) VALUES (99, 'Hacker', 'h@example.com')", + "SELECT id FROM customers ORDER BY id", + 3, + ) + .await; + cleanup(&path); +} + +#[tokio::test] +async fn parity_denied_update_capability_violation() { + let path = build_parity_fixture(); + let config = ConnectionConfig::duckdb(path.clone()); + assert_capability_violation_and_state_unchanged( + &config, + "UPDATE customers SET name = 'Hacker' WHERE id = 1", + "SELECT name FROM customers WHERE id = 1", + 1, + ) + .await; + // Double-check the actual value was not changed + let qr = DuckDbEngine::execute( + &config, + "SELECT name FROM customers WHERE id = 1", + &[], + &Capabilities::default(), + ) + .await + .unwrap(); + assert_eq!( + qr.rows[0][0].as_str(), + Some("Ada Lovelace"), + "UPDATE must not have mutated the row" + ); + cleanup(&path); +} + +#[tokio::test] +async fn parity_denied_delete_capability_violation() { + let path = build_parity_fixture(); + let config = ConnectionConfig::duckdb(path.clone()); + assert_capability_violation_and_state_unchanged( + &config, + "DELETE FROM customers WHERE id = 1", + "SELECT id FROM customers ORDER BY id", + 3, + ) + .await; + cleanup(&path); +} + +#[tokio::test] +async fn parity_denied_create_capability_violation() { + let path = build_parity_fixture(); + let config = ConnectionConfig::duckdb(path.clone()); + assert_capability_violation_and_state_unchanged( + &config, + "CREATE TABLE hacker (id INTEGER PRIMARY KEY)", + "SELECT table_name FROM duckdb_tables() WHERE table_name = 'hacker'", + 0, + ) + .await; + cleanup(&path); +} + +#[tokio::test] +async fn parity_denied_drop_capability_violation() { + let path = build_parity_fixture(); + let config = ConnectionConfig::duckdb(path.clone()); + assert_capability_violation_and_state_unchanged( + &config, + "DROP TABLE customers", + "SELECT id FROM customers ORDER BY id", + 3, + ) + .await; + cleanup(&path); +} + +#[tokio::test] +async fn parity_denied_alter_capability_violation() { + let path = build_parity_fixture(); + let config = ConnectionConfig::duckdb(path.clone()); + assert_capability_violation_and_state_unchanged( + &config, + "ALTER TABLE customers ADD COLUMN phone VARCHAR", + "SELECT id FROM customers ORDER BY id", + 3, + ) + .await; + cleanup(&path); +} + +#[tokio::test] +async fn parity_denied_copy_and_attach_capability_violation() { + // DuckDB-specific write/escape surfaces: COPY writes to the filesystem, + // ATTACH mutates catalog state, INSTALL/LOAD pull in extensions. + let path = build_parity_fixture(); + let config = ConnectionConfig::duckdb(path.clone()); + let caps = Capabilities::default(); + + for sql in [ + "COPY customers TO '/tmp/plenum_parity_leak.csv'", + "ATTACH ':memory:' AS other", + "DETACH other", + "INSTALL httpfs", + "LOAD httpfs", + "EXPORT DATABASE '/tmp/plenum_parity_leak'", + ] { + let result = DuckDbEngine::execute(&config, sql, &[], &caps).await; + assert!(result.is_err(), "'{sql}' must be rejected"); + assert_eq!( + result.unwrap_err().error_code(), + "CAPABILITY_VIOLATION", + "'{sql}' must produce CAPABILITY_VIOLATION" + ); + } + cleanup(&path); +} + +// ============================================================================ +// Safety — max_rows truncation and timeout_ms +// ============================================================================ + +#[tokio::test] +async fn parity_safety_max_rows_truncates_bulk_table() { + // bulk_rows has 1 500 rows; max_rows=100 must truncate and set the flag. + let path = build_parity_fixture(); + let config = ConnectionConfig::duckdb(path.clone()); + let caps = Capabilities { max_rows: Some(100), ..Capabilities::default() }; + let qr = + DuckDbEngine::execute(&config, "SELECT n, label FROM bulk_rows ORDER BY n", &[], &caps) + .await + .expect("SELECT bulk_rows failed"); + assert_eq!(qr.rows.len(), 100, "max_rows=100 must limit result to 100 rows"); + assert!(qr.rows_truncated, "rows_truncated must be true when max_rows fires"); + // Verify ordering: first row is n=1 + assert_eq!(qr.rows[0][0], serde_json::json!(1), "first row must be n=1"); + cleanup(&path); +} + +#[tokio::test] +async fn parity_safety_rows_returned_matches_max_rows() { + let path = build_parity_fixture(); + let config = ConnectionConfig::duckdb(path.clone()); + let caps = Capabilities { max_rows: Some(50), ..Capabilities::default() }; + let qr = DuckDbEngine::execute(&config, "SELECT n FROM bulk_rows ORDER BY n", &[], &caps) + .await + .expect("SELECT bulk_rows failed"); + assert_eq!(qr.rows.len(), 50, "rows returned must equal max_rows when truncated"); + assert!(qr.rows_truncated, "rows_truncated must be set"); + cleanup(&path); +} + +#[tokio::test] +async fn parity_safety_no_truncation_when_under_max_rows() { + let path = build_parity_fixture(); + let config = ConnectionConfig::duckdb(path.clone()); + let caps = Capabilities { max_rows: Some(5000), ..Capabilities::default() }; + let qr = DuckDbEngine::execute(&config, "SELECT n FROM bulk_rows ORDER BY n", &[], &caps) + .await + .expect("SELECT bulk_rows failed"); + assert_eq!(qr.rows.len(), 1500, "all 1500 rows must be returned when max_rows=5000"); + assert!(!qr.rows_truncated, "rows_truncated must be false when not truncated"); + cleanup(&path); +} + +#[tokio::test] +async fn parity_safety_offset_pagination() { + let path = build_parity_fixture(); + let config = ConnectionConfig::duckdb(path.clone()); + let caps = Capabilities { max_rows: Some(10), offset: Some(100), ..Capabilities::default() }; + let qr = DuckDbEngine::execute(&config, "SELECT n FROM bulk_rows ORDER BY n", &[], &caps) + .await + .expect("SELECT bulk_rows failed"); + assert_eq!(qr.rows.len(), 10, "offset+max_rows must return one page"); + assert_eq!(qr.rows[0][0], serde_json::json!(101), "first row after offset=100 must be n=101"); + cleanup(&path); +} + +#[tokio::test] +async fn parity_safety_timeout_ms_completes_fast_query() { + // A simple query must finish well within a 5 s window. + let path = build_parity_fixture(); + let config = ConnectionConfig::duckdb(path.clone()); + let caps = Capabilities { timeout_ms: Some(5000), ..Capabilities::default() }; + let result = DuckDbEngine::execute(&config, "SELECT count(*) FROM customers", &[], &caps).await; + assert!(result.is_ok(), "fast query must complete within 5 s timeout: {:?}", result.err()); + cleanup(&path); +} + +#[tokio::test] +async fn parity_safety_timeout_ms_interrupts_long_query() { + // A large cross-join aggregate takes many seconds. With timeout_ms=50 the + // interrupt thread fires mid-execution, causing DuckDB to return an + // INTERRUPT error → QUERY_TIMEOUT. Uses :memory: — no fixture file needed. + let config = ConnectionConfig::duckdb(":memory:".into()); + let caps = Capabilities { timeout_ms: Some(50), ..Capabilities::default() }; + let sql = "SELECT max(a.range * b.range + a.range) \ + FROM range(200000) a, range(200000) b"; + let result = DuckDbEngine::execute(&config, sql, &[], &caps).await; + assert!(result.is_err(), "long-running query must be interrupted by timeout"); + assert_eq!( + result.unwrap_err().error_code(), + "QUERY_TIMEOUT", + "interrupted query must surface as QUERY_TIMEOUT" + ); +} + +// ============================================================================ +// Envelope — JSON shape and determinism +// ============================================================================ + +#[tokio::test] +async fn parity_envelope_query_result_json_shape() { + let path = build_parity_fixture(); + let config = ConnectionConfig::duckdb(path.clone()); + let caps = Capabilities::default(); + let qr = DuckDbEngine::execute( + &config, + "SELECT id, name FROM customers ORDER BY id LIMIT 1", + &[], + &caps, + ) + .await + .expect("SELECT customers failed"); + + let json = serde_json::to_value(&qr).expect("QueryResult must serialize to JSON"); + assert!( + json.get("columns").is_some_and(serde_json::Value::is_array), + "must have 'columns' array" + ); + assert!(json.get("rows").is_some_and(serde_json::Value::is_array), "must have 'rows' array"); + assert!(json.get("execution_ms").is_some(), "must have 'execution_ms'"); + assert!(json.get("rows_affected").is_none(), "SELECT must not have 'rows_affected'"); + cleanup(&path); +} + +#[tokio::test] +async fn parity_envelope_introspect_result_json_shape() { + let path = build_parity_fixture(); + let config = ConnectionConfig::duckdb(path.clone()); + let result = DuckDbEngine::introspect(&config, &IntrospectOperation::ListTables, None, None) + .await + .expect("ListTables failed"); + + let json = serde_json::to_value(&result).expect("IntrospectResult must serialize to JSON"); + assert!(json.get("type").is_some(), "IntrospectResult must have 'type' tag"); + assert!( + json.get("tables").is_some_and(serde_json::Value::is_array), + "TableList must have 'tables' array" + ); + cleanup(&path); +} + +#[tokio::test] +async fn parity_envelope_deterministic_excluding_execution_ms() { + // Identical queries must produce identical row data (execution_ms is timing, excluded). + let path = build_parity_fixture(); + let config = ConnectionConfig::duckdb(path.clone()); + let caps = Capabilities::default(); + let sql = "SELECT id, name, email FROM customers ORDER BY id"; + + let r1 = DuckDbEngine::execute(&config, sql, &[], &caps).await.expect("execute 1"); + let r2 = DuckDbEngine::execute(&config, sql, &[], &caps).await.expect("execute 2"); + + assert_eq!(r1.columns, r2.columns, "columns must be identical"); + assert_eq!(r1.rows, r2.rows, "rows must be identical"); + assert_eq!(r1.rows_truncated, r2.rows_truncated, "rows_truncated must be identical"); + // execution_ms is timing and intentionally excluded from the comparison + cleanup(&path); +} + +#[tokio::test] +async fn parity_envelope_error_has_code_and_message() { + // Errors must expose a non-empty error_code and message. + let config = ConnectionConfig::duckdb(PathBuf::from("/does/not/exist.duckdb")); + let err = DuckDbEngine::validate_connection(&config).await.unwrap_err(); + assert!(!err.error_code().is_empty(), "error must have a non-empty error_code"); + assert!(!err.message().is_empty(), "error must have a non-empty message"); +} From d67605d4412ada77513b89feb5a974563af0a7b9 Mon Sep 17 00:00:00 2001 From: therecluse26 Date: Sun, 19 Jul 2026 22:09:56 -0400 Subject: [PATCH 3/4] docs: document DuckDB as a first-class engine (REF-290) Co-Authored-By: Claude Fable 5 Co-Authored-By: Paperclip --- CLAUDE.md | 1 + README.md | 22 +++++++++++++--------- 2 files changed, 14 insertions(+), 9 deletions(-) diff --git a/CLAUDE.md b/CLAUDE.md index 0970f9c..d462053 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -67,6 +67,7 @@ The MVP MUST support: - PostgreSQL - MySQL (primary target) - SQLite +- DuckDB All engines are first-class and equally constrained. diff --git a/README.md b/README.md index 2a6b2c0..174468f 100644 --- a/README.md +++ b/README.md @@ -19,10 +19,10 @@ Plenum is exposed via a local MCP (Model Context Protocol) server, making it sea ## Key Features - **Agent-First Design**: JSON-only output, no interactive UX, deterministic behavior -- **Vendor-Specific SQL**: No query abstraction layer - PostgreSQL SQL ≠ MySQL SQL ≠ SQLite SQL +- **Vendor-Specific SQL**: No query abstraction layer - PostgreSQL SQL ≠ MySQL SQL ≠ SQLite SQL ≠ DuckDB SQL - **Strictly Read-Only**: All write and DDL operations are rejected - guaranteed safe for AI agents - **Stateless Execution**: No persistent connections, no caching, no implicit state -- **Three Database Engines**: PostgreSQL, MySQL, and SQLite support (first-class, equally constrained) +- **Four Database Engines**: PostgreSQL, MySQL, SQLite, and DuckDB support (first-class, equally constrained) ## Installation @@ -55,7 +55,7 @@ Manage database connection configurations (interactive or non-interactive). | `--list` | — | List saved connections for the project as JSON (no secrets emitted) | | `--name ` | `"default"` | Connection name | | `--project-path ` | current directory | Project path for connection lookup | -| `--engine ` | — | Database engine: `postgres`, `mysql`, or `sqlite` | +| `--engine ` | — | Database engine: `postgres`, `mysql`, `sqlite`, or `duckdb` | | `--host ` | — | Hostname (postgres/mysql) | | `--port ` | — | Port (postgres/mysql) | | `--user ` | — | Username (postgres/mysql) | @@ -65,7 +65,7 @@ Manage database connection configurations (interactive or non-interactive). | `--keychain-service ` | — | OS keychain service name (pair with `--keychain-account`) | | `--keychain-account ` | — | OS keychain account name (pair with `--keychain-service`) | | `--database ` | — | Database name (postgres/mysql) | -| `--file ` | — | SQLite file path | +| `--file ` | — | SQLite/DuckDB file path | | `--save ` | — | Save location: `local` (`.plenum/config.json`) or `global` (`~/.config/plenum/connections.json`) | | `--ssl-mode ` | — | TLS/SSL mode: `disable`, `require`, `verify-ca`, or `verify-full` (postgres/mysql) | | `--ssl-ca ` | — | PEM CA certificate for TLS verification (required for `verify-ca`/`verify-full`) | @@ -102,6 +102,9 @@ plenum connect --name vault --engine postgres --host localhost \ # Save a SQLite connection plenum connect --name dev --engine sqlite --file ./dev.db --save local +# Save a DuckDB connection +plenum connect --name analytics --engine duckdb --file ./analytics.duckdb --save local + # Test a connection without saving plenum connect --engine postgres --host localhost --user dev \ --password-env DEV_DB_PASSWORD --database mydb --test @@ -151,16 +154,16 @@ Inspect database schema and return structured JSON. | Flag | Default | Description | |------|---------|-------------| -| `--dsn ` | — | One-off connection URL (mutually exclusive with `--name` and explicit flags). Accepted schemes: `postgres://`, `postgresql://`, `mysql://`, `sqlite:` | +| `--dsn ` | — | One-off connection URL (mutually exclusive with `--name` and explicit flags). Accepted schemes: `postgres://`, `postgresql://`, `mysql://`, `sqlite:`, `duckdb:` | | `--name ` | `"default"` | Named connection from the saved registry | | `--project-path ` | current directory | Project path for connection lookup | -| `--engine ` | — | Engine override: `postgres`, `mysql`, or `sqlite` | +| `--engine ` | — | Engine override: `postgres`, `mysql`, `sqlite`, or `duckdb` | | `--host ` | — | Host override | | `--port ` | — | Port override | | `--user ` | — | Username override | | `--password ` | — | Password override | | `--database ` | — | Database override | -| `--file ` | — | SQLite file override | +| `--file ` | — | SQLite/DuckDB file override | | `--ssl-mode ` | — | TLS/SSL mode: `disable`, `require`, `verify-ca`, or `verify-full` (postgres/mysql) | | `--ssl-ca ` | — | PEM CA certificate (required for `verify-ca`/`verify-full`) | | `--ssl-cert ` | — | PEM client certificate for mTLS (pair with `--ssl-key`) | @@ -171,14 +174,14 @@ Inspect database schema and return structured JSON. | Flag | Default | Description | |------|---------|-------------| | `--list-databases` | — | List all databases (requires a wildcard/no-database connection) | -| `--list-schemas` | — | List all schemas (PostgreSQL only) | +| `--list-schemas` | — | List all schemas (PostgreSQL/DuckDB) | | `--list-tables` | — | List all table names | | `--list-views` | — | List all view names | | `--list-indexes [TABLE]` | — | List all indexes, optionally filtered to a single table | | `--table ` | — | Return full details for a specific table | | `--view ` | — | Return details for a specific view | | `--target-database ` | — | Switch to a different database before introspecting | -| `--schema ` | — | Filter results to a specific schema (PostgreSQL/MySQL only) | +| `--schema ` | — | Filter results to a specific schema (PostgreSQL/MySQL/DuckDB) | | `--diff-against ` | — | Structural schema diff against another named connection. Mutually exclusive with all other operation flags. Returns tables/views added, removed, and changed (columns, indexes, foreign keys, primary keys) | | `--diff-against-project-path ` | current project | Project path for the `--diff-against` connection (for cross-project comparison) | @@ -411,6 +414,7 @@ Plenum uses native, engine-specific drivers (NOT sqlx): - **PostgreSQL**: `tokio-postgres` - **MySQL**: `mysql_async` - **SQLite**: `rusqlite` +- **DuckDB**: `duckdb` This ensures maximum isolation between engines and preserves vendor-specific behavior. From 5136eba2da1d086a71f94b05212b41c6bc7751bf Mon Sep 17 00:00:00 2001 From: therecluse26 Date: Sun, 19 Jul 2026 22:28:32 -0400 Subject: [PATCH 4/4] review: fix stale MCP engine description, document InterruptHandle drop safety (REF-290) - MCP query tool's engine description now lists duckdb alongside the enum - Timeout timer thread comment now cites the duckdb crate contract that interrupt() after connection drop is a no-op (verified against crate source) Co-Authored-By: Paperclip --- src/engine/duckdb/mod.rs | 7 +++++++ src/mcp.rs | 2 +- 2 files changed, 8 insertions(+), 1 deletion(-) diff --git a/src/engine/duckdb/mod.rs b/src/engine/duckdb/mod.rs index 47a6569..3ef9037 100644 --- a/src/engine/duckdb/mod.rs +++ b/src/engine/duckdb/mod.rs @@ -113,6 +113,13 @@ impl DatabaseEngine for DuckDbEngine { // starts, then spawn a thread that fires the interrupt after timeout_ms. // DuckDB checks the interrupt flag during execution, cancelling the // query server-side rather than just abandoning the wait. + // + // The timer thread is detached and may outlive the connection: if the + // query finishes early, `handle.interrupt()` fires against an + // already-dropped `Connection`. This is safe by documented crate + // contract — `InterruptHandle` holds a mutex-guarded connection + // pointer that is nulled when the connection drops, making a late + // `interrupt()` a no-op (duckdb crate, `InterruptHandle::interrupt`). if let Some(timeout_ms) = caps.timeout_ms { let handle = conn.interrupt_handle(); std::thread::spawn(move || { diff --git a/src/mcp.rs b/src/mcp.rs index c48d3bb..449395d 100644 --- a/src/mcp.rs +++ b/src/mcp.rs @@ -376,7 +376,7 @@ fn handle_list_tools() -> Result { "engine": { "type": "string", "enum": ["postgres", "mysql", "sqlite", "duckdb"], - "description": "DISCOURAGED: Database engine type for explicit one-off connections. Only use if no saved connection exists. Valid values: 'postgres', 'mysql', 'sqlite'. If omitted along with 'connection', auto-resolves project's default connection (RECOMMENDED)." + "description": "DISCOURAGED: Database engine type for explicit one-off connections. Only use if no saved connection exists. Valid values: 'postgres', 'mysql', 'sqlite', 'duckdb'. If omitted along with 'connection', auto-resolves project's default connection (RECOMMENDED)." }, "host": { "type": "string",