From 1fdbba0eb26a1e62c550b7e6d5b0719986562c96 Mon Sep 17 00:00:00 2001 From: Kevin Longe <273747162+kvlonge@users.noreply.github.com> Date: Thu, 12 Nov 2026 12:00:26 +0000 Subject: [PATCH] feat(motherduck): connect through a separate DuckDB 1.5 process (install sqlload[motherduck]), as MotherDuck doesn't accept DuckDB 2.0 clients yet A motherduck target now runs its SQL on Python's duckdb package (1.4.1 to 1.5.6, which MotherDuck accepts) in a process of its own, so the two DuckDB versions never share a process and every other target stays on the bundled DuckDB 2.0. SQLLoad sends it newline-delimited JSON requests over a pipe, one cursor per table connection, and hands it each batch as a temporary Parquet file it inserts with read_parquet inside the batch's transaction. The token goes over the pipe, never in the command line, and is redacted from everything the process prints. The sqlload package gains a motherduck extra (duckdb>=1.4.1,<=1.5.6) and runs the process on its own Python; sqlload-cli and other setups name one with SQLLOAD_MOTHERDUCK_PYTHON. Everything specific to it sits behind the DuckDB adapter's Bridge connection and adapters/motherduck_bridge; the MotherDuck page lists the steps to remove it once MotherDuck accepts DuckDB 2.0. Tested offline through a stand-in for MotherDuck (every write strategy and portable type compared with DuckDB, an incremental rerun, the process ending mid-load, a rejected value, setup errors) and live against a MotherDuck account (test_motherduck_live.py, run with SQLLOAD_TEST_MOTHERDUCK_TOKEN). --- README.md | 2 +- .../src/adapters/duckdb/_helpers/blocking.rs | 11 +- .../duckdb/_helpers/motherduck_connection.rs | 7 +- .../duckdb/_helpers/shared_destination.rs | 9 +- crates/sqlload/src/adapters/duckdb/errors.rs | 2 + .../src/adapters/duckdb/main/_append_batch.rs | 23 +- .../adapters/duckdb/main/_describe_table.rs | 34 +- .../duckdb/main/_execute_statements.rs | 12 +- .../adapters/duckdb/main/_open_destination.rs | 2 +- .../src/adapters/duckdb/main/_open_duckdb.rs | 2 +- .../adapters/duckdb/main/_open_motherduck.rs | 23 +- .../adapters/duckdb/main/_own_connection.rs | 20 +- .../src/adapters/duckdb/main/_query_text.rs | 12 +- .../src/adapters/duckdb/main/_query_texts.rs | 10 +- .../src/adapters/duckdb/main/_table_shape.rs | 65 ++-- crates/sqlload/src/adapters/duckdb/models.rs | 15 +- .../src/adapters/duckdb/tests/helpers.rs | 23 +- .../adapters/duckdb/tests/test_motherduck.rs | 15 +- crates/sqlload/src/adapters/mod.rs | 2 + .../motherduck_bridge/_helpers/bridge.py | 107 ++++++ .../motherduck_bridge/_helpers/mod.rs | 1 + .../motherduck_bridge/_helpers/process.rs | 177 +++++++++ .../adapters/motherduck_bridge/constants.rs | 16 + .../src/adapters/motherduck_bridge/errors.rs | 13 + .../motherduck_bridge/main/_append_batch.rs | 43 +++ .../main/_fork_connection.rs | 15 + .../motherduck_bridge/main/_open_bridge.rs | 48 +++ .../motherduck_bridge/main/_run_statement.rs | 46 +++ .../adapters/motherduck_bridge/main/mod.rs | 4 + .../src/adapters/motherduck_bridge/mod.rs | 7 + .../src/adapters/motherduck_bridge/models.rs | 45 +++ packages/sqlload/README.md | 2 +- packages/sqlload/pyproject.toml | 6 + packages/sqlload/python/sqlload/_cli.py | 4 + packages/sqlload/tests/test_motherduck.py | 357 ++++++++++++++++++ .../sqlload/tests/test_motherduck_live.py | 177 +++++++++ website/public/index.html | 2 +- .../src/content/docs/docs/configuration.mdx | 4 +- website/src/content/docs/docs/index.mdx | 2 +- .../content/docs/docs/supported-systems.mdx | 4 +- .../content/docs/docs/targets/motherduck.mdx | 90 +++-- 41 files changed, 1343 insertions(+), 116 deletions(-) create mode 100644 crates/sqlload/src/adapters/motherduck_bridge/_helpers/bridge.py create mode 100644 crates/sqlload/src/adapters/motherduck_bridge/_helpers/mod.rs create mode 100644 crates/sqlload/src/adapters/motherduck_bridge/_helpers/process.rs create mode 100644 crates/sqlload/src/adapters/motherduck_bridge/constants.rs create mode 100644 crates/sqlload/src/adapters/motherduck_bridge/errors.rs create mode 100644 crates/sqlload/src/adapters/motherduck_bridge/main/_append_batch.rs create mode 100644 crates/sqlload/src/adapters/motherduck_bridge/main/_fork_connection.rs create mode 100644 crates/sqlload/src/adapters/motherduck_bridge/main/_open_bridge.rs create mode 100644 crates/sqlload/src/adapters/motherduck_bridge/main/_run_statement.rs create mode 100644 crates/sqlload/src/adapters/motherduck_bridge/main/mod.rs create mode 100644 crates/sqlload/src/adapters/motherduck_bridge/mod.rs create mode 100644 crates/sqlload/src/adapters/motherduck_bridge/models.rs create mode 100644 packages/sqlload/tests/test_motherduck.py create mode 100644 packages/sqlload/tests/test_motherduck_live.py diff --git a/README.md b/README.md index a6f69fd9..4af89a26 100644 --- a/README.md +++ b/README.md @@ -94,7 +94,7 @@ versions and how to run them yourself. ¹ In builds with the `odbc` feature. -² Once MotherDuck accepts DuckDB 2.0 clients. +² From 0.34.0 (not released yet), through a separate DuckDB 1.5 process: install `sqlload[motherduck]`. See [Supported systems](https://sqlload.io/docs/supported-systems/) for every combination and write strategy. diff --git a/crates/sqlload/src/adapters/duckdb/_helpers/blocking.rs b/crates/sqlload/src/adapters/duckdb/_helpers/blocking.rs index 3aeb5efe..492e08a3 100644 --- a/crates/sqlload/src/adapters/duckdb/_helpers/blocking.rs +++ b/crates/sqlload/src/adapters/duckdb/_helpers/blocking.rs @@ -1,12 +1,17 @@ use crate::adapters::duckdb::errors::DuckDbError; -use crate::adapters::duckdb::models::DuckDbDestination; +use crate::adapters::duckdb::models::{DuckDbConnection, DuckDbDestination}; -/// Run `work` with the connection on a blocking thread, since DuckDB calls block. +/// Run `work` with the connection on a blocking thread, since DuckDB calls block. Only SQLLoad's own DuckDB runs it: a MotherDuck connection is refused. pub(crate) async fn with_connection( destination: &DuckDbDestination, work: impl FnOnce(&duckdb::Connection) -> Result + Send + 'static, ) -> Result { - let connection = destination.connection.clone(); + let DuckDbConnection::Local(connection) = &destination.connection else { + return Err(DuckDbError::MotherDuck( + "this step needs SQLLoad's own DuckDB, and MotherDuck runs on the bridge's".to_owned(), + )); + }; + let connection = connection.clone(); tokio::task::spawn_blocking(move || { let guard = connection .lock() diff --git a/crates/sqlload/src/adapters/duckdb/_helpers/motherduck_connection.rs b/crates/sqlload/src/adapters/duckdb/_helpers/motherduck_connection.rs index a9fdefb5..a6361d67 100644 --- a/crates/sqlload/src/adapters/duckdb/_helpers/motherduck_connection.rs +++ b/crates/sqlload/src/adapters/duckdb/_helpers/motherduck_connection.rs @@ -4,6 +4,7 @@ use crate::adapters::duckdb::constants::{ EXTENSION_EXCEPTION, MOTHERDUCK_TOKEN_VARIABLES, MOTHERDUCK_UNSUPPORTED_DUCKDB, }; use crate::adapters::duckdb::errors::DuckDbError; +use crate::adapters::motherduck_bridge::constants::{PYTHON_VARIABLE, SUPPORTED_DUCKDB}; use crate::auth::models::Secret; use crate::project::constants::MOTHERDUCK_PREFIX; use crate::project::main::motherduck_path::motherduck_path; @@ -65,7 +66,7 @@ pub(in crate::adapters::duckdb) fn connection_string(path: &str, token: &str) -> format!("{path}{separator}motherduck_token={}", query_value(token)) } -/// Why `shown` did not open, from DuckDB's `error` with `token` taken out: MotherDuck's own words when its extension refuses this DuckDB version. +/// Why `shown` did not open, from the bridge's `error` with `token` taken out: MotherDuck's own words when its extension refuses the bridge's DuckDB version. pub(in crate::adapters::duckdb) fn refused(error: &str, token: &str, shown: &str) -> DuckDbError { let error = error .replace(&query_value(token), "") @@ -81,12 +82,12 @@ pub(in crate::adapters::duckdb) fn refused(error: &str, token: &str, shown: &str .collect::>() .join(" "); DuckDbError::MotherDuck(format!( - "cannot open {shown}: MotherDuck doesn't accept the DuckDB version this SQLLoad bundles yet, so no SQLLoad release can connect to it until it does. MotherDuck says: {said}" + "cannot open {shown}: MotherDuck doesn't accept the version of Python's duckdb package that SQLLoad connects through: install {SUPPORTED_DUCKDB} there (pip install 'sqlload[motherduck]'), or set {PYTHON_VARIABLE} to a Python that has it. MotherDuck says: {said}" )) } /// `value` percent-encoded for a connection string's query, keeping only unreserved characters, so a token holding `&`, `=` or `#` reaches MotherDuck whole. -fn query_value(value: &str) -> String { +pub(in crate::adapters::duckdb) fn query_value(value: &str) -> String { let mut encoded = String::with_capacity(value.len()); for byte in value.bytes() { match byte { diff --git a/crates/sqlload/src/adapters/duckdb/_helpers/shared_destination.rs b/crates/sqlload/src/adapters/duckdb/_helpers/shared_destination.rs index 64e5bf5e..9e822e0d 100644 --- a/crates/sqlload/src/adapters/duckdb/_helpers/shared_destination.rs +++ b/crates/sqlload/src/adapters/duckdb/_helpers/shared_destination.rs @@ -3,17 +3,16 @@ use std::sync::{Arc, Mutex}; use crate::adapters::duckdb::errors::DuckDbError; -use crate::adapters::duckdb::models::DuckDbDestination; +use crate::adapters::duckdb::models::{DuckDbConnection, DuckDbDestination}; -/// `connection`, ready to take Arrow batches and to be shared by the load's tasks; `motherduck` when it is a MotherDuck database. +/// `connection`, ready to take Arrow batches and to be shared by the load's tasks. pub(in crate::adapters::duckdb) fn shared_destination( connection: duckdb::Connection, - motherduck: bool, ) -> Result { connection.register_table_function::("arrow")?; Ok(DuckDbDestination { - connection: Arc::new(Mutex::new(connection)), + connection: DuckDbConnection::Local(Arc::new(Mutex::new(connection))), ddl: Arc::default(), - motherduck, + motherduck: false, }) } diff --git a/crates/sqlload/src/adapters/duckdb/errors.rs b/crates/sqlload/src/adapters/duckdb/errors.rs index 2f7c3432..28370a5f 100644 --- a/crates/sqlload/src/adapters/duckdb/errors.rs +++ b/crates/sqlload/src/adapters/duckdb/errors.rs @@ -5,6 +5,8 @@ pub enum DuckDbError { Engine(#[from] duckdb::Error), #[error("MotherDuck: {0}")] MotherDuck(String), + #[error("MotherDuck: {0}")] + Bridge(#[from] crate::adapters::motherduck_bridge::errors::BridgeError), #[error("DuckDB task failed: {0}")] Task(String), } diff --git a/crates/sqlload/src/adapters/duckdb/main/_append_batch.rs b/crates/sqlload/src/adapters/duckdb/main/_append_batch.rs index 7d39d232..f4bc8033 100644 --- a/crates/sqlload/src/adapters/duckdb/main/_append_batch.rs +++ b/crates/sqlload/src/adapters/duckdb/main/_append_batch.rs @@ -8,23 +8,28 @@ use duckdb::vtab::ArrowBatchRegistration; use crate::adapters::contract::models::TableName; use crate::adapters::duckdb::_helpers::blocking::with_connection; use crate::adapters::duckdb::errors::DuckDbError; -use crate::adapters::duckdb::models::DuckDbDestination; +use crate::adapters::duckdb::models::{DuckDbConnection, DuckDbDestination}; +use crate::adapters::motherduck_bridge::main::_append_batch::append_batch as bridge_append; -/// Insert `batch` into `table` by column name, inside the connection's open transaction if it has one: columns the batch lacks get their `DEFAULT`, and values are converted to the column's type, so bad JSON or UUIDs fail here. +/// Insert `batch` into `table` by column name, inside the connection's open transaction if it has one: columns the batch lacks get their `DEFAULT`, and values are converted to the column's type, so bad JSON or UUIDs fail here. MotherDuck's bridge reads the batch from a Parquet file instead. pub(in crate::adapters) async fn append_batch( destination: &DuckDbDestination, table: &TableName, batch: RecordBatch, ) -> Result<(), DuckDbError> { let target = DuckDbSql.render_qualified_name(table); + let columns: Vec = batch + .schema() + .fields() + .iter() + .map(|field| DuckDbSql.render_identifier(field.name())) + .collect(); + let list = columns.join(", "); + if let DuckDbConnection::Bridge(bridge) = &destination.connection { + let insert = format!("INSERT INTO {target} ({list}) SELECT {list} FROM read_parquet(?)"); + return Ok(bridge_append(bridge, insert, batch).await?); + } with_connection(destination, move |connection| { - let fields = batch.schema(); - let columns: Vec = fields - .fields() - .iter() - .map(|field| DuckDbSql.render_identifier(field.name())) - .collect(); - let list = columns.join(", "); connection .prepare(&format!( "INSERT INTO {target} ({list}) SELECT {list} FROM system.main.arrow(?)" diff --git a/crates/sqlload/src/adapters/duckdb/main/_describe_table.rs b/crates/sqlload/src/adapters/duckdb/main/_describe_table.rs index d08fccd2..e5d8d4f5 100644 --- a/crates/sqlload/src/adapters/duckdb/main/_describe_table.rs +++ b/crates/sqlload/src/adapters/duckdb/main/_describe_table.rs @@ -5,7 +5,8 @@ use crate::adapters::contract::types::SqlAdapter; use crate::adapters::duckdb::_helpers::blocking::with_connection; use crate::adapters::duckdb::constants::DESCRIBE_TABLE_SQL; use crate::adapters::duckdb::errors::DuckDbError; -use crate::adapters::duckdb::models::{DuckDbDestination, DuckDbSql}; +use crate::adapters::duckdb::models::{DuckDbConnection, DuckDbDestination, DuckDbSql}; +use crate::adapters::motherduck_bridge::main::_run_statement::run_statement; use crate::column_types::models::LiveColumnType; use crate::schema_plan::models::LiveColumn; @@ -16,6 +17,14 @@ pub(in crate::adapters) async fn describe_table( ) -> Result>, DuckDbError> { let schema = table.schema.clone(); let name = table.table.clone(); + if let DuckDbConnection::Bridge(bridge) = &destination.connection { + let rows = run_statement(bridge, DESCRIBE_TABLE_SQL.to_owned(), vec![schema, name]).await?; + let columns = rows + .iter() + .map(|row| bridged_column(row)) + .collect::, _>>()?; + return Ok((!columns.is_empty()).then_some(columns)); + } let columns = with_connection(destination, move |connection| { let mut statement = connection.prepare(DESCRIBE_TABLE_SQL)?; let rows = statement.query_map([schema, name], |row| { @@ -35,3 +44,26 @@ pub(in crate::adapters) async fn describe_table( .await?; Ok((!columns.is_empty()).then_some(columns)) } + +/// A column from a bridged row of `DESCRIBE_TABLE_SQL`: its name, type, precision, scale and length, as text. +fn bridged_column(row: &[Option]) -> Result { + let value = |index: usize| row.get(index).cloned().flatten(); + let number = |index: usize| { + value(index) + .map(|text| text.parse::()) + .transpose() + .map_err(|error| { + DuckDbError::MotherDuck(format!("a column's size is not a number: {error}")) + }) + }; + Ok(LiveColumn { + name: value(0).unwrap_or_default(), + column_type: DuckDbSql.live_type(&LiveColumnType { + data_type: value(1).unwrap_or_default(), + precision: number(2)?, + scale: number(3)?, + char_length: number(4)?, + }), + fraction_digits: None, + }) +} diff --git a/crates/sqlload/src/adapters/duckdb/main/_execute_statements.rs b/crates/sqlload/src/adapters/duckdb/main/_execute_statements.rs index e78117c8..ae256a6f 100644 --- a/crates/sqlload/src/adapters/duckdb/main/_execute_statements.rs +++ b/crates/sqlload/src/adapters/duckdb/main/_execute_statements.rs @@ -2,13 +2,23 @@ use crate::adapters::duckdb::_helpers::blocking::with_connection; use crate::adapters::duckdb::errors::DuckDbError; -use crate::adapters::duckdb::models::DuckDbDestination; +use crate::adapters::duckdb::models::{DuckDbConnection, DuckDbDestination}; +use crate::adapters::motherduck_bridge::main::_run_statement::run_statement; /// Run each statement in order, stopping at the first failure (an open transaction is rolled back). pub(in crate::adapters) async fn execute_statements( destination: &DuckDbDestination, statements: Vec, ) -> Result<(), DuckDbError> { + if let DuckDbConnection::Bridge(bridge) = &destination.connection { + for statement in statements { + if let Err(error) = run_statement(bridge, statement, Vec::new()).await { + let _ = run_statement(bridge, "ROLLBACK".to_owned(), Vec::new()).await; + return Err(error.into()); + } + } + return Ok(()); + } with_connection(destination, move |connection| { for statement in &statements { if let Err(error) = connection.execute_batch(statement) { diff --git a/crates/sqlload/src/adapters/duckdb/main/_open_destination.rs b/crates/sqlload/src/adapters/duckdb/main/_open_destination.rs index 1952d51b..ccedbdca 100644 --- a/crates/sqlload/src/adapters/duckdb/main/_open_destination.rs +++ b/crates/sqlload/src/adapters/duckdb/main/_open_destination.rs @@ -13,7 +13,7 @@ pub(in crate::adapters) fn open_destination( ) -> Result { match connection { ConnectionConfig::DuckDb { database } => open_duckdb(database), - ConnectionConfig::MotherDuck { .. } => open_motherduck(connection, target_database), + ConnectionConfig::MotherDuck { .. } => open_motherduck(connection, target_database, None), other => Err(DuckDbError::Task(format!( "a {} connection is not a DuckDB one", other.kind() diff --git a/crates/sqlload/src/adapters/duckdb/main/_open_duckdb.rs b/crates/sqlload/src/adapters/duckdb/main/_open_duckdb.rs index 375082a0..4347da06 100644 --- a/crates/sqlload/src/adapters/duckdb/main/_open_duckdb.rs +++ b/crates/sqlload/src/adapters/duckdb/main/_open_duckdb.rs @@ -16,5 +16,5 @@ pub(in crate::adapters) fn open_duckdb(path: &str) -> Result duckdb::Connection::open_in_memory()?, path => duckdb::Connection::open(path)?, }; - shared_destination(connection, false) + shared_destination(connection) } diff --git a/crates/sqlload/src/adapters/duckdb/main/_open_motherduck.rs b/crates/sqlload/src/adapters/duckdb/main/_open_motherduck.rs index 114c10be..7b4ef09e 100644 --- a/crates/sqlload/src/adapters/duckdb/main/_open_motherduck.rs +++ b/crates/sqlload/src/adapters/duckdb/main/_open_motherduck.rs @@ -1,17 +1,20 @@ //! Open a MotherDuck database. +use std::sync::Arc; + use crate::adapters::duckdb::_helpers::motherduck_connection::{ - chosen_token, connection_string, environment_variable, path_text, refused, + chosen_token, connection_string, environment_variable, path_text, query_value, refused, }; -use crate::adapters::duckdb::_helpers::shared_destination::shared_destination; use crate::adapters::duckdb::errors::DuckDbError; -use crate::adapters::duckdb::models::DuckDbDestination; +use crate::adapters::duckdb::models::{DuckDbConnection, DuckDbDestination}; +use crate::adapters::motherduck_bridge::main::_open_bridge::open_bridge; use crate::project::models::ConnectionConfig; -/// Open the MotherDuck database of `connection` (`target_database` when the target names one) through DuckDB's `md:` connection, signed in with its token, else the token in the `motherduck_token` or `MOTHERDUCK_TOKEN` variable. Without a token MotherDuck would wait for a browser sign-in, so it is refused instead. +/// Open the MotherDuck database of `connection` (`target_database` when the target names one) through DuckDB's `md:` connection on the MotherDuck bridge, signed in with its token, else the token in the `motherduck_token` or `MOTHERDUCK_TOKEN` variable. Without a token MotherDuck would wait for a browser sign-in, so it is refused instead. `python` runs the bridge; `None` picks it as the bridge does. pub(in crate::adapters) fn open_motherduck( connection: &ConnectionConfig, target_database: Option<&str>, + python: Option<&str>, ) -> Result { let ConnectionConfig::MotherDuck { database, token } = connection else { return Err(DuckDbError::MotherDuck(format!( @@ -25,7 +28,15 @@ pub(in crate::adapters) fn open_motherduck( "{shown} needs an access token: set token in the connection, or the motherduck_token or MOTHERDUCK_TOKEN environment variable" )) })?; - let connection = duckdb::Connection::open(connection_string(&shown, &token)) + let config = serde_json::json!({ + "custom_user_agent": format!("sqlload/{}", env!("CARGO_PKG_VERSION")), + }); + let hidden = vec![query_value(&token), token.clone()]; + let (bridge, _) = open_bridge(&connection_string(&shown, &token), config, (python, hidden)) .map_err(|error| refused(&error.to_string(), &token, &shown))?; - shared_destination(connection, true) + Ok(DuckDbDestination { + connection: DuckDbConnection::Bridge(bridge), + ddl: Arc::default(), + motherduck: true, + }) } diff --git a/crates/sqlload/src/adapters/duckdb/main/_own_connection.rs b/crates/sqlload/src/adapters/duckdb/main/_own_connection.rs index f1a57508..f5f6f712 100644 --- a/crates/sqlload/src/adapters/duckdb/main/_own_connection.rs +++ b/crates/sqlload/src/adapters/duckdb/main/_own_connection.rs @@ -3,19 +3,25 @@ use std::sync::{Arc, Mutex}; use crate::adapters::duckdb::errors::DuckDbError; -use crate::adapters::duckdb::models::DuckDbDestination; +use crate::adapters::duckdb::models::{DuckDbConnection, DuckDbDestination}; +use crate::adapters::motherduck_bridge::main::_fork_connection::fork_connection; /// A new connection to the same database, so one table's transaction never mixes with another table's loading at the same time. pub(in crate::adapters) fn own_connection( destination: &DuckDbDestination, ) -> Result { - let connection = destination - .connection - .lock() - .map_err(|_| DuckDbError::Task("the connection lock is poisoned".to_owned()))? - .try_clone()?; + let connection = match &destination.connection { + DuckDbConnection::Local(connection) => { + let connection = connection + .lock() + .map_err(|_| DuckDbError::Task("the connection lock is poisoned".to_owned()))? + .try_clone()?; + DuckDbConnection::Local(Arc::new(Mutex::new(connection))) + } + DuckDbConnection::Bridge(bridge) => DuckDbConnection::Bridge(fork_connection(bridge)), + }; Ok(DuckDbDestination { - connection: Arc::new(Mutex::new(connection)), + connection, ddl: destination.ddl.clone(), motherduck: destination.motherduck, }) diff --git a/crates/sqlload/src/adapters/duckdb/main/_query_text.rs b/crates/sqlload/src/adapters/duckdb/main/_query_text.rs index 23ebae15..5abf9ec7 100644 --- a/crates/sqlload/src/adapters/duckdb/main/_query_text.rs +++ b/crates/sqlload/src/adapters/duckdb/main/_query_text.rs @@ -2,13 +2,23 @@ use crate::adapters::duckdb::_helpers::blocking::with_connection; use crate::adapters::duckdb::errors::DuckDbError; -use crate::adapters::duckdb::models::DuckDbDestination; +use crate::adapters::duckdb::models::{DuckDbConnection, DuckDbDestination}; +use crate::adapters::motherduck_bridge::main::_run_statement::run_statement; /// The first column of the first row of `query`, as text; `None` for SQL `NULL`. pub(in crate::adapters) async fn query_text( destination: &DuckDbDestination, query: String, ) -> Result, DuckDbError> { + if let DuckDbConnection::Bridge(bridge) = &destination.connection { + let rows = run_statement(bridge, query, Vec::new()).await?; + let first = rows + .into_iter() + .next() + .and_then(|row| row.into_iter().next()); + return first + .ok_or_else(|| DuckDbError::MotherDuck("the query returned no rows".to_owned())); + } with_connection(destination, move |connection| { Ok(connection.query_row(&query, [], |row| row.get::<_, Option>(0))?) }) diff --git a/crates/sqlload/src/adapters/duckdb/main/_query_texts.rs b/crates/sqlload/src/adapters/duckdb/main/_query_texts.rs index 2eaafe66..c4e9a75d 100644 --- a/crates/sqlload/src/adapters/duckdb/main/_query_texts.rs +++ b/crates/sqlload/src/adapters/duckdb/main/_query_texts.rs @@ -2,13 +2,21 @@ use crate::adapters::duckdb::_helpers::blocking::with_connection; use crate::adapters::duckdb::errors::DuckDbError; -use crate::adapters::duckdb::models::DuckDbDestination; +use crate::adapters::duckdb::models::{DuckDbConnection, DuckDbDestination}; +use crate::adapters::motherduck_bridge::main::_run_statement::run_statement; /// The first column of every row of `query`, as text; SQL `NULL`s are skipped. pub(in crate::adapters) async fn query_texts( destination: &DuckDbDestination, query: String, ) -> Result, DuckDbError> { + if let DuckDbConnection::Bridge(bridge) = &destination.connection { + let rows = run_statement(bridge, query, Vec::new()).await?; + return Ok(rows + .into_iter() + .filter_map(|row| row.into_iter().next().flatten()) + .collect()); + } with_connection(destination, move |connection| { let mut statement = connection.prepare(&query)?; let rows = statement.query_map([], |row| row.get::<_, Option>(0))?; diff --git a/crates/sqlload/src/adapters/duckdb/main/_table_shape.rs b/crates/sqlload/src/adapters/duckdb/main/_table_shape.rs index f9d4bcfa..1425cbba 100644 --- a/crates/sqlload/src/adapters/duckdb/main/_table_shape.rs +++ b/crates/sqlload/src/adapters/duckdb/main/_table_shape.rs @@ -5,7 +5,8 @@ use crate::adapters::contract::types::SqlAdapter; use crate::adapters::duckdb::_helpers::blocking::with_connection; use crate::adapters::duckdb::constants::{DEFINITION_QUERY, INDEXES_QUERY}; use crate::adapters::duckdb::errors::DuckDbError; -use crate::adapters::duckdb::models::{DuckDbDestination, DuckDbSql}; +use crate::adapters::duckdb::models::{DuckDbConnection, DuckDbDestination, DuckDbSql}; +use crate::adapters::motherduck_bridge::main::_run_statement::run_statement; /// `table`'s columns and constraints as DuckDB would create them, and its indexes' statements on `table` by its full name (DuckDB stores them on `schema.table`, which MotherDuck's other attached databases can make ambiguous); `None` when there is no such table. pub(in crate::adapters) async fn table_shape( @@ -13,29 +14,45 @@ pub(in crate::adapters) async fn table_shape( table: &TableName, ) -> Result, DuckDbError> { let (schema, name) = (table.schema.clone(), table.table.clone()); - let named = table.clone(); - with_connection(destination, move |connection| { - let found = |query: &str| -> Result, DuckDbError> { - let mut statement = connection.prepare(query)?; - let rows = statement.query_map([&schema, &name], |row| row.get::<_, String>(0))?; - Ok(rows.collect::>()?) - }; - let Some(definition) = found(DEFINITION_QUERY)? - .first() - .and_then(|sql| columns_part(sql)) - else { - return Ok(None); - }; - let qualified = DuckDbSql.render_qualified_name(&named); - Ok(Some(TableShape { - definition, - indexes: found(INDEXES_QUERY)? - .iter() - .map(|sql| on_table(sql, &qualified)) - .collect(), - })) - }) - .await + let (definitions, indexes) = match &destination.connection { + DuckDbConnection::Bridge(bridge) => { + let found = |query: &str| { + run_statement(bridge, query.to_owned(), vec![schema.clone(), name.clone()]) + }; + let first = |rows: Vec>>| -> Vec { + rows.into_iter() + .filter_map(|row| row.into_iter().next().flatten()) + .collect() + }; + ( + first(found(DEFINITION_QUERY).await?), + first(found(INDEXES_QUERY).await?), + ) + } + DuckDbConnection::Local(_) => { + with_connection(destination, move |connection| { + let found = |query: &str| -> Result, DuckDbError> { + let mut statement = connection.prepare(query)?; + let rows = + statement.query_map([&schema, &name], |row| row.get::<_, String>(0))?; + Ok(rows.collect::>()?) + }; + Ok((found(DEFINITION_QUERY)?, found(INDEXES_QUERY)?)) + }) + .await? + } + }; + let Some(definition) = definitions.first().and_then(|sql| columns_part(sql)) else { + return Ok(None); + }; + let qualified = DuckDbSql.render_qualified_name(table); + Ok(Some(TableShape { + definition, + indexes: indexes + .iter() + .map(|sql| on_table(sql, &qualified)) + .collect(), + })) } /// The `(...)` part of a `CREATE TABLE` statement: from the first parenthesis outside a quoted name. diff --git a/crates/sqlload/src/adapters/duckdb/models.rs b/crates/sqlload/src/adapters/duckdb/models.rs index bf0c8d25..80f5f4be 100644 --- a/crates/sqlload/src/adapters/duckdb/models.rs +++ b/crates/sqlload/src/adapters/duckdb/models.rs @@ -1,17 +1,30 @@ #[cfg(feature = "duckdb")] use std::sync::{Arc, Mutex}; +#[cfg(feature = "duckdb")] +use crate::adapters::motherduck_bridge::models::BridgeConnection; + /// An open DuckDB database shared by the load's tasks. #[cfg(feature = "duckdb")] #[derive(Debug, Clone)] pub struct DuckDbDestination { - pub connection: Arc>, + pub connection: DuckDbConnection, /// Held while a load's tables change the schema, one at a time: DuckDB fails concurrent catalog changes with write-write conflicts. pub ddl: Arc>, /// A MotherDuck database (`md:`), which only takes Arrow batches: it never reads files itself. pub motherduck: bool, } +/// Where a destination's SQL runs. +#[cfg(feature = "duckdb")] +#[derive(Debug, Clone)] +pub enum DuckDbConnection { + /// The DuckDB SQLLoad bundles. + Local(Arc>), + /// MotherDuck, through the bridge's DuckDB (`adapters::motherduck_bridge`), as MotherDuck doesn't accept SQLLoad's DuckDB yet. + Bridge(BridgeConnection), +} + /// DuckDB's SQL dialect. #[derive(Debug, Clone, Copy, Default)] pub struct DuckDbSql; diff --git a/crates/sqlload/src/adapters/duckdb/tests/helpers.rs b/crates/sqlload/src/adapters/duckdb/tests/helpers.rs index 6da9e262..2edf5929 100644 --- a/crates/sqlload/src/adapters/duckdb/tests/helpers.rs +++ b/crates/sqlload/src/adapters/duckdb/tests/helpers.rs @@ -1,6 +1,19 @@ +use std::sync::{Arc, Mutex}; + +use crate::adapters::duckdb::models::{DuckDbConnection, DuckDbDestination}; use crate::column_types::models::ColumnType; use crate::schema_plan::models::{ColumnOrigin, DesiredColumn}; +/// The connection of a destination on SQLLoad's own DuckDB. +pub(crate) fn local( + destination: &DuckDbDestination, +) -> Result>, String> { + let DuckDbConnection::Local(connection) = &destination.connection else { + return Err("the destination is MotherDuck's".to_owned()); + }; + Ok(connection.clone()) +} + /// A column read from the source, as a load writes it. pub(crate) fn column(name: &str, column_type: ColumnType) -> DesiredColumn { DesiredColumn { @@ -83,11 +96,9 @@ pub(crate) fn unloaded_row_loaded( /// An unload of two rows written as one Parquet file in a local folder (no file at all when not `with_files`), said to hold `reported` rows, loaded by DuckDB reading it itself straight into the table; then any open batch rolled back, as a load's close does. What the load returned (`failed` for an error), and the rows the table holds afterwards. pub(crate) fn pulled_unloaded(reported: u64, with_files: bool) -> Result<(String, u64), String> { - use std::sync::{Arc, Mutex}; - use crate::adapters::contract::models::{TableName, UnloadedBatch, WriteTarget}; use crate::adapters::contract::types::TableWriter; - use crate::adapters::duckdb::models::{DuckDbDestination, DuckDbWriter}; + use crate::adapters::duckdb::models::DuckDbWriter; use crate::postgres_source::models::{ReadColumn, WireType}; use crate::snowflake_source::models::UnloadedFiles; use crate::store_paths::models::{Location, StoreKind}; @@ -104,7 +115,7 @@ pub(crate) fn pulled_unloaded(reported: u64, with_files: bool) -> Result<(String .map_err(failed)?; let mut writer = DuckDbWriter { destination: DuckDbDestination { - connection: Arc::new(Mutex::new(connection)), + connection: DuckDbConnection::Local(Arc::new(Mutex::new(connection))), ddl: Arc::new(tokio::sync::Mutex::new(())), motherduck: false, }, @@ -161,9 +172,7 @@ pub(crate) fn pulled_unloaded(reported: u64, with_files: bool) -> Result<(String writer.rollback_batch().await; outcome }); - let held: i64 = writer - .destination - .connection + let held: i64 = local(&writer.destination)? .lock() .map_err(|error| error.to_string())? .query_row("SELECT count(*) FROM loaded", [], |row| row.get(0)) diff --git a/crates/sqlload/src/adapters/duckdb/tests/test_motherduck.rs b/crates/sqlload/src/adapters/duckdb/tests/test_motherduck.rs index 8d2bf80a..beeaf8f2 100644 --- a/crates/sqlload/src/adapters/duckdb/tests/test_motherduck.rs +++ b/crates/sqlload/src/adapters/duckdb/tests/test_motherduck.rs @@ -11,6 +11,7 @@ use crate::adapters::duckdb::main::_open_duckdb::open_duckdb; use crate::adapters::duckdb::main::_open_motherduck::open_motherduck; use crate::adapters::duckdb::main::_table_shape::table_shape; use crate::adapters::duckdb::models::DuckDbSql; +use crate::adapters::duckdb::tests::helpers::local; use crate::adapters::duckdb::tests::test_types::{ AttachedDatabaseTestCase, ConnectionStringTestCase, DuckDbPathTestCase, IndexReplayTestCase, MotherDuckOpenTestCase, MotherDuckPathTestCase, MotherDuckRefusalTestCase, @@ -22,7 +23,7 @@ use crate::loader::_helpers::resolve::source_target; use crate::project::main::parse_project::parse_project; use crate::project::models::ConnectionConfig; -/// What DuckDB 2.0.0 reports when MotherDuck's extension refuses to load on it. +/// What DuckDB 2.0.0 reports when MotherDuck's extension refuses to load on it, as Python's duckdb package 2.0.0 would in the bridge. const UNSUPPORTED_VERSION: &str = "Invalid Input Error: Initialization function \"motherduck_duckdb_cpp_init\" from file \"/home/u/.duckdb/extensions/v2.0.0/linux_amd64/motherduck.duckdb_extension\" threw an exception: \"\nYour DuckDB version (v2.0.0) is not yet supported by MotherDuck. The latest supported version is v1.5.6. Please downgrade to use MotherDuck.\n\nSee https://motherduck.com/docs/getting-started/connect-query-from-duckdb-cli for more information.\""; /// The refusal of a database setting that holds a token. @@ -178,7 +179,7 @@ fn given_motherduck_does_not_open_when_reporting_why_then_its_own_words_are_rela description: "the extension refusing this DuckDB version is relayed on one line", error: UNSUPPORTED_VERSION.to_owned(), token: "tok.en-1", - expected_message: "MotherDuck: cannot open md:shop: MotherDuck doesn't accept the DuckDB version this SQLLoad bundles yet, so no SQLLoad release can connect to it until it does. MotherDuck says: Your DuckDB version (v2.0.0) is not yet supported by MotherDuck. The latest supported version is v1.5.6. Please downgrade to use MotherDuck. See https://motherduck.com/docs/getting-started/connect-query-from-duckdb-cli for more information.", + expected_message: "MotherDuck: cannot open md:shop: MotherDuck doesn't accept the version of Python's duckdb package that SQLLoad connects through: install duckdb 1.4.1 to 1.5.6 there (pip install 'sqlload[motherduck]'), or set SQLLOAD_MOTHERDUCK_PYTHON to a Python that has it. MotherDuck says: Your DuckDB version (v2.0.0) is not yet supported by MotherDuck. The latest supported version is v1.5.6. Please downgrade to use MotherDuck. See https://motherduck.com/docs/getting-started/connect-query-from-duckdb-cli for more information.", }, MotherDuckRefusalTestCase { description: "another error that quotes the connection string loses the token", @@ -231,7 +232,7 @@ fn given_a_motherduck_connection_when_opening_then_the_targets_database_wins_and database: test_case.connection_database.map(str::to_owned), token: Some(Secret(String::new())), }; - let error = open_motherduck(&connection, test_case.target_database) + let error = open_motherduck(&connection, test_case.target_database, None) .err() .map(|error| error.to_string()); assert_eq!( @@ -400,10 +401,8 @@ fn given_other_attached_databases_when_reading_the_schema_and_staging_then_only_ DuckDbSql.render_qualified_name(&table) ); { - let connection = destination - .connection - .lock() - .map_err(|error| error.to_string())?; + let connection = local(&destination)?; + let connection = connection.lock().map_err(|error| error.to_string())?; connection .execute_batch("CREATE SCHEMA raw; CREATE TABLE raw.orders (id INTEGER);") .and_then(|()| connection.execute_batch(test_case.setup)) @@ -453,7 +452,7 @@ fn given_indexes_and_a_database_named_like_the_schema_when_a_rebuild_recreates_t schema: "raw".to_owned(), table: "orders".to_owned(), }; - let connection = destination.connection.clone(); + let connection = local(&destination)?; connection .lock() .map_err(|error| error.to_string())? diff --git a/crates/sqlload/src/adapters/mod.rs b/crates/sqlload/src/adapters/mod.rs index 329152c8..5cbf3261 100644 --- a/crates/sqlload/src/adapters/mod.rs +++ b/crates/sqlload/src/adapters/mod.rs @@ -7,5 +7,7 @@ pub mod databricks; pub mod dispatch; pub mod duckdb; pub mod files; +#[cfg(feature = "duckdb")] +pub mod motherduck_bridge; pub mod postgres; pub mod snowflake; diff --git a/crates/sqlload/src/adapters/motherduck_bridge/_helpers/bridge.py b/crates/sqlload/src/adapters/motherduck_bridge/_helpers/bridge.py new file mode 100644 index 00000000..799b1a2c --- /dev/null +++ b/crates/sqlload/src/adapters/motherduck_bridge/_helpers/bridge.py @@ -0,0 +1,107 @@ +"""SQLLoad's MotherDuck bridge: SQL run on PyPI's duckdb (1.4.1 to 1.5.6, which MotherDuck accepts) for a SQLLoad process, whose bundled DuckDB MotherDuck doesn't accept yet. + +SQLLoad starts it as `python -c ` and sends one JSON request per line on stdin; each reply is one JSON line on stdout, tagged with its request's id: + + {"id": 1, "op": "open", "database": "md:shop?motherduck_token=...", "config": {...}} + -> {"id": 1, "ok": true, "duckdb": "1.5.6", "python": "3.12.3"} + {"id": 2, "op": "run", "cursor": 3, "sql": "SELECT ...", "params": ["raw"]} + -> {"id": 2, "ok": true, "rows": [["id", "BIGINT"], ...]} + {"op": "close", "cursor": 3} + -> no reply + +A failure replies {"id": n, "ok": false, "error": "...", "type": "CatalogException"}. Each cursor is a connection of its own to +the opened database (DuckDB's `cursor()`), created on its first request and run on its own thread in request order, so one +table's transaction never mixes with another's. The bridge exits when stdin closes, which rolls back any open transaction. +""" + +import sys + +# `python -c` puts the working directory first on the path, where a stray duckdb.py would shadow the real package. +if sys.path and sys.path[0] == "": + del sys.path[0] + +import json +import math +import os +import queue +import signal +import threading + + +def plain(value): + """`value` as JSON can carry it: numbers, text, booleans and NULL as they are, lists item by item, anything else as its text.""" + if value is None or isinstance(value, (bool, int, str)): + return value + if isinstance(value, float): + return value if math.isfinite(value) else str(value) + if isinstance(value, (list, tuple)): + return [plain(item) for item in value] + return str(value) + + +def main(): + # Replies go to a private copy of stdout; anything else that prints (DuckDB, an extension) lands on stderr instead. + replies = os.fdopen(os.dup(1), "wb", buffering=0) + os.dup2(2, 1) + sys.stdout = sys.stderr + # Ctrl-C reaches the whole process group; SQLLoad ends, its pipe closes, and that ends the bridge. + signal.signal(signal.SIGINT, signal.SIG_IGN) + lock = threading.Lock() + + def reply(message): + line = json.dumps(message, ensure_ascii=True, allow_nan=False).encode("ascii") + b"\n" + with lock: + replies.write(line) + + def failed(request, error): + reply({"id": request.get("id"), "ok": False, "error": str(error), "type": type(error).__name__}) + + database = None + cursors = {} + + def serve(work): + cursor = None + while True: + request = work.get() + if request is None: + break + try: + if cursor is None: + cursor = database.cursor() + cursor.execute(request["sql"], request.get("params") or None) + rows = cursor.fetchall() if cursor.description else [] + reply({"id": request["id"], "ok": True, "rows": [[plain(value) for value in row] for row in rows]}) + except BaseException as error: # noqa: BLE001 - every failure goes back to SQLLoad as its reply, or it would wait forever + failed(request, error) + if cursor is not None: + cursor.close() + + for line in sys.stdin.buffer: + request = json.loads(line) + op = request.get("op") + if op == "open": + try: + import duckdb + + database = duckdb.connect(request["database"], config=request.get("config") or {}) + reply({"id": request["id"], "ok": True, "duckdb": duckdb.__version__, "python": sys.version.split()[0]}) + except Exception as error: # noqa: BLE001 + failed(request, error) + elif op == "run": + work = cursors.get(request["cursor"]) + if work is None: + work = cursors[request["cursor"]] = queue.Queue() + threading.Thread(target=serve, args=(work,), daemon=True).start() + work.put(request) + elif op == "close": + work = cursors.pop(request["cursor"], None) + if work is not None: + work.put(None) + else: + failed(request, ValueError(f"unknown op {op!r}")) + # stdin closed: SQLLoad is done or gone. Statements still running are abandoned, and open transactions roll back. + replies.flush() + os._exit(0) + + +main() diff --git a/crates/sqlload/src/adapters/motherduck_bridge/_helpers/mod.rs b/crates/sqlload/src/adapters/motherduck_bridge/_helpers/mod.rs new file mode 100644 index 00000000..3c433d35 --- /dev/null +++ b/crates/sqlload/src/adapters/motherduck_bridge/_helpers/mod.rs @@ -0,0 +1 @@ +pub(crate) mod process; diff --git a/crates/sqlload/src/adapters/motherduck_bridge/_helpers/process.rs b/crates/sqlload/src/adapters/motherduck_bridge/_helpers/process.rs new file mode 100644 index 00000000..df1d82bd --- /dev/null +++ b/crates/sqlload/src/adapters/motherduck_bridge/_helpers/process.rs @@ -0,0 +1,177 @@ +//! The bridge process: starting it, sending it requests, and routing its replies. + +use std::io::{BufRead, BufReader, Read, Write}; +use std::process::{Command, Stdio}; +use std::sync::atomic::{AtomicU64, Ordering}; +use std::sync::mpsc::{Receiver, channel}; +use std::sync::{Arc, Mutex, PoisonError}; + +use crate::adapters::motherduck_bridge::constants::{ + BRIDGE_SCRIPT, PYTHON_VARIABLE, STDERR_TAIL_BYTES, SUPPORTED_DUCKDB, +}; +use crate::adapters::motherduck_bridge::errors::BridgeError; +use crate::adapters::motherduck_bridge::models::{BridgeProcess, Replies}; + +/// Start the bridge on `python`: a writer thread feeding its stdin (closing it, and reaping the process, once every handle is gone), a reader thread routing its replies and, once stdout and stderr have both closed or a reply is unreadable, failing what still waits, and one keeping its stderr's tail with each of `hidden` taken out. +pub(crate) fn start(python: &str, hidden: Vec) -> Result, BridgeError> { + let mut child = Command::new(python) + .args(["-c", BRIDGE_SCRIPT]) + .stdin(Stdio::piped()) + .stdout(Stdio::piped()) + .stderr(Stdio::piped()) + .spawn() + .map_err(|error| { + BridgeError::Setup(format!( + "MotherDuck connects through Python's duckdb package ({SUPPORTED_DUCKDB}), and {python} cannot run ({error}): install sqlload[motherduck] with pip or uv, or set {PYTHON_VARIABLE} to a Python that has {SUPPORTED_DUCKDB}" + )) + })?; + let (Some(mut stdin), Some(stdout), Some(stderr)) = + (child.stdin.take(), child.stdout.take(), child.stderr.take()) + else { + return Err(BridgeError::Setup(format!( + "{python} started without its pipes" + ))); + }; + let (lines, outgoing) = channel::(); + let replies = Arc::new(Mutex::new(Replies::default())); + let tail = Arc::new(Mutex::new(String::new())); + std::thread::spawn(move || { + for line in outgoing { + if stdin + .write_all(format!("{line}\n").as_bytes()) + .and_then(|()| stdin.flush()) + .is_err() + { + break; + } + } + drop(stdin); + let _ = child.wait(); + }); + let kept = tail.clone(); + let reading_stderr = std::thread::spawn(move || keep_tail(stderr, &kept, &hidden)); + let routed = replies.clone(); + let (shown, kept) = (python.to_owned(), tail.clone()); + std::thread::spawn(move || match route_replies(stdout, &routed) { + Some(broken) => fail( + &routed, + format!("the MotherDuck bridge on {shown} {broken}"), + ), + None => { + let _ = reading_stderr.join(); + stop(&routed, &shown, &kept); + } + }); + Ok(Arc::new(BridgeProcess { + python: python.to_owned(), + lines, + replies, + next_request: AtomicU64::new(1), + next_cursor: AtomicU64::new(1), + stderr: tail, + })) +} + +/// Send `request` with an id of its own; the receiver gets its reply, or why the bridge stopped first. +pub(crate) fn send( + process: &BridgeProcess, + mut request: serde_json::Value, +) -> Result>, BridgeError> { + let id = process.next_request.fetch_add(1, Ordering::Relaxed); + request["id"] = id.into(); + let (reply, received) = channel(); + let mut replies = process + .replies + .lock() + .unwrap_or_else(PoisonError::into_inner); + if let Some(stopped) = &replies.stopped { + return Err(BridgeError::Stopped(stopped.clone())); + } + replies.waiting.insert(id, reply); + drop(replies); + if process.lines.send(request.to_string()).is_err() { + return Err(BridgeError::Stopped(format!( + "the MotherDuck bridge on {} is closed", + process.python + ))); + } + Ok(received) +} + +/// `reply`'s failure, as its message and its Python error type; `None` when it succeeded. +pub(crate) fn failure(reply: &serde_json::Value) -> Option<(String, String)> { + if reply["ok"].as_bool() == Some(true) { + return None; + } + let text = |key: &str| reply[key].as_str().unwrap_or_default().to_owned(); + Some((text("error"), text("type"))) +} + +/// Hand each reply line to the request waiting for it, until stdout ends (`None`) or a reply cannot be read, which says why (the bridge may still run, but no waiting request would ever get its reply). +fn route_replies(stdout: impl Read, replies: &Mutex) -> Option { + for line in BufReader::new(stdout).lines() { + let line = match line { + Ok(line) => line, + Err(error) => return Some(format!("cannot be read: {error}")), + }; + let reply = match serde_json::from_str::(&line) { + Ok(reply) => reply, + Err(error) => return Some(format!("replied with something other than JSON ({error})")), + }; + let waiting = reply["id"].as_u64().and_then(|id| { + replies + .lock() + .unwrap_or_else(PoisonError::into_inner) + .waiting + .remove(&id) + }); + if let Some(waiting) = waiting { + let _ = waiting.send(Ok(reply)); + } + } + None +} + +/// Fail every waiting and later request with why the bridge stopped, quoting the tail of its stderr. +fn stop(replies: &Mutex, python: &str, tail: &Mutex) { + let said = tail + .lock() + .unwrap_or_else(PoisonError::into_inner) + .trim() + .to_owned(); + let stopped = match said.is_empty() { + true => format!("the MotherDuck bridge on {python} stopped"), + false => format!("the MotherDuck bridge on {python} stopped: {said}"), + }; + fail(replies, stopped); +} + +/// Fail every waiting and later request with `stopped`. +fn fail(replies: &Mutex, stopped: String) { + let mut replies = replies.lock().unwrap_or_else(PoisonError::into_inner); + for (_, waiting) in replies.waiting.drain() { + let _ = waiting.send(Err(BridgeError::Stopped(stopped.clone()))); + } + replies.stopped = Some(stopped); +} + +/// Keep the last `STDERR_TAIL_BYTES` of `stderr`, which failures quote, with each of `hidden` taken out. +fn keep_tail(stderr: impl Read, tail: &Mutex, hidden: &[String]) { + for line in BufReader::new(stderr).lines() { + let Ok(mut line) = line else { break }; + for secret in hidden.iter().filter(|secret| !secret.is_empty()) { + line = line.replace(secret.as_str(), ""); + } + tracing::debug!(line, "MotherDuck bridge"); + let mut kept = tail.lock().unwrap_or_else(PoisonError::into_inner); + kept.push_str(&line); + kept.push('\n'); + if kept.len() > STDERR_TAIL_BYTES { + let mut cut = kept.len() - STDERR_TAIL_BYTES; + while !kept.is_char_boundary(cut) { + cut += 1; + } + kept.replace_range(..cut, ""); + } + } +} diff --git a/crates/sqlload/src/adapters/motherduck_bridge/constants.rs b/crates/sqlload/src/adapters/motherduck_bridge/constants.rs new file mode 100644 index 00000000..0799253f --- /dev/null +++ b/crates/sqlload/src/adapters/motherduck_bridge/constants.rs @@ -0,0 +1,16 @@ +/// The bridge's Python program, run with `python -c`. +pub const BRIDGE_SCRIPT: &str = include_str!("_helpers/bridge.py"); +/// The environment variable naming the Python the bridge runs on; the `sqlload` Python package sets it to its own interpreter unless it is set already. +pub const PYTHON_VARIABLE: &str = "SQLLOAD_MOTHERDUCK_PYTHON"; +/// The Python the bridge runs on when `PYTHON_VARIABLE` is unset: the first on `PATH`. +#[cfg(not(windows))] +pub const DEFAULT_PYTHON: &str = "python3"; +/// The Python the bridge runs on when `PYTHON_VARIABLE` is unset: the first on `PATH`. +#[cfg(windows)] +pub const DEFAULT_PYTHON: &str = "python"; +/// The duckdb package versions MotherDuck accepts, as the bridge's install hints name them. +pub const SUPPORTED_DUCKDB: &str = "duckdb 1.4.1 to 1.5.6"; +/// How many bytes of the bridge's last stderr output a failure quotes. +pub const STDERR_TAIL_BYTES: usize = 2048; +/// The error types the bridge replies with when its Python has no duckdb package. +pub const MISSING_PACKAGE_ERRORS: &[&str] = &["ModuleNotFoundError", "ImportError"]; diff --git a/crates/sqlload/src/adapters/motherduck_bridge/errors.rs b/crates/sqlload/src/adapters/motherduck_bridge/errors.rs new file mode 100644 index 00000000..d12113b1 --- /dev/null +++ b/crates/sqlload/src/adapters/motherduck_bridge/errors.rs @@ -0,0 +1,13 @@ +/// Why a MotherDuck bridge request failed. +#[derive(Debug, thiserror::Error)] +pub enum BridgeError { + /// The bridge could not start, or its Python cannot run it. + #[error("{0}")] + Setup(String), + /// The bridge process ended; any request still waiting fails with this. + #[error("{0}")] + Stopped(String), + /// DuckDB refused the request, in its own words. + #[error("{0}")] + Engine(String), +} diff --git a/crates/sqlload/src/adapters/motherduck_bridge/main/_append_batch.rs b/crates/sqlload/src/adapters/motherduck_bridge/main/_append_batch.rs new file mode 100644 index 00000000..44ce7fac --- /dev/null +++ b/crates/sqlload/src/adapters/motherduck_bridge/main/_append_batch.rs @@ -0,0 +1,43 @@ +//! Hand an Arrow batch to the bridge. + +use arrow_array::RecordBatch; +use parquet::arrow::ArrowWriter; + +use crate::adapters::motherduck_bridge::errors::BridgeError; +use crate::adapters::motherduck_bridge::main::_run_statement::run_statement; +use crate::adapters::motherduck_bridge::models::BridgeConnection; + +/// Run `insert`, whose one `?` is the path of a Parquet file holding `batch` (such as `INSERT INTO t (a) SELECT a FROM read_parquet(?)`), on `connection`. The file is a temporary one on this machine, which DuckDB reads there and removed once the statement ends: the bridge's DuckDB is another version than SQLLoad's, so Arrow cannot cross between them in memory. +pub(in crate::adapters) async fn append_batch( + connection: &BridgeConnection, + insert: String, + batch: RecordBatch, +) -> Result<(), BridgeError> { + let file = tokio::task::spawn_blocking(move || -> Result<_, String> { + let file = tempfile::Builder::new() + .prefix("sqlload-motherduck-") + .suffix(".parquet") + .tempfile() + .map_err(|error| error.to_string())?; + let mut writer = ArrowWriter::try_new( + file.reopen().map_err(|error| error.to_string())?, + batch.schema(), + None, + ) + .map_err(|error| error.to_string())?; + writer.write(&batch).map_err(|error| error.to_string())?; + writer.close().map_err(|error| error.to_string())?; + Ok(file) + }) + .await + .map_err(|error| BridgeError::Setup(error.to_string()))? + .map_err(|error| { + BridgeError::Setup(format!( + "cannot write a batch for the MotherDuck bridge: {error}" + )) + })?; + let path = file.path().display().to_string(); + run_statement(connection, insert, vec![path]).await?; + drop(file); + Ok(()) +} diff --git a/crates/sqlload/src/adapters/motherduck_bridge/main/_fork_connection.rs b/crates/sqlload/src/adapters/motherduck_bridge/main/_fork_connection.rs new file mode 100644 index 00000000..9e73a37f --- /dev/null +++ b/crates/sqlload/src/adapters/motherduck_bridge/main/_fork_connection.rs @@ -0,0 +1,15 @@ +//! Give a load another connection to the bridge's database. + +use std::sync::Arc; +use std::sync::atomic::Ordering; + +use crate::adapters::motherduck_bridge::models::{BridgeConnection, BridgeCursor}; + +/// A new connection to the database `connection` opened, on the same bridge: its own DuckDB cursor, so its transactions never mix with another connection's. +pub(in crate::adapters) fn fork_connection(connection: &BridgeConnection) -> BridgeConnection { + let process = connection.cursor.process.clone(); + let id = process.next_cursor.fetch_add(1, Ordering::Relaxed); + BridgeConnection { + cursor: Arc::new(BridgeCursor { id, process }), + } +} diff --git a/crates/sqlload/src/adapters/motherduck_bridge/main/_open_bridge.rs b/crates/sqlload/src/adapters/motherduck_bridge/main/_open_bridge.rs new file mode 100644 index 00000000..62f0e486 --- /dev/null +++ b/crates/sqlload/src/adapters/motherduck_bridge/main/_open_bridge.rs @@ -0,0 +1,48 @@ +//! Start the bridge and open a database on it. + +use std::sync::Arc; +use std::sync::atomic::Ordering; + +use crate::adapters::motherduck_bridge::_helpers::process::{failure, send, start}; +use crate::adapters::motherduck_bridge::constants::{ + DEFAULT_PYTHON, MISSING_PACKAGE_ERRORS, PYTHON_VARIABLE, SUPPORTED_DUCKDB, +}; +use crate::adapters::motherduck_bridge::errors::BridgeError; +use crate::adapters::motherduck_bridge::models::{BridgeConnection, BridgeCursor}; + +/// Start the bridge on `python` (else the Python `SQLLOAD_MOTHERDUCK_PYTHON` names, else `python3` on `PATH`) and open `database` there with DuckDB's `config` settings: a DuckDB path or connection string, such as `md:shop?motherduck_token=...`, which never appears in the process's arguments. Each of `hidden` (such as the token, as written and as encoded) is taken out of what the bridge prints. Returns a connection to it and the version of the duckdb package the bridge runs. Blocks until the database opens. +pub(in crate::adapters) fn open_bridge( + database: &str, + config: serde_json::Value, + (python, hidden): (Option<&str>, Vec), +) -> Result<(BridgeConnection, String), BridgeError> { + let python = match python { + Some(python) => python.to_owned(), + None => match std::env::var(PYTHON_VARIABLE) { + Ok(python) if !python.is_empty() => python, + Ok(_) | Err(std::env::VarError::NotPresent) => DEFAULT_PYTHON.to_owned(), + Err(std::env::VarError::NotUnicode(_)) => { + return Err(BridgeError::Setup(format!( + "{PYTHON_VARIABLE} is not valid Unicode" + ))); + } + }, + }; + let process = start(&python, hidden)?; + let open = serde_json::json!({"op": "open", "database": database, "config": config}); + let reply = send(&process, open)?.recv().map_err(|_| { + BridgeError::Stopped(format!("the MotherDuck bridge on {python} stopped")) + })??; + if let Some((error, kind)) = failure(&reply) { + return Err(match MISSING_PACKAGE_ERRORS.contains(&kind.as_str()) { + true => BridgeError::Setup(format!( + "MotherDuck connects through Python's duckdb package ({SUPPORTED_DUCKDB}), which {python} lacks ({error}): install sqlload[motherduck] with pip or uv, or set {PYTHON_VARIABLE} to a Python that has it" + )), + false => BridgeError::Engine(error), + }); + } + let version = reply["duckdb"].as_str().unwrap_or_default().to_owned(); + let id = process.next_cursor.fetch_add(1, Ordering::Relaxed); + let cursor = Arc::new(BridgeCursor { id, process }); + Ok((BridgeConnection { cursor }, version)) +} diff --git a/crates/sqlload/src/adapters/motherduck_bridge/main/_run_statement.rs b/crates/sqlload/src/adapters/motherduck_bridge/main/_run_statement.rs new file mode 100644 index 00000000..08a6b604 --- /dev/null +++ b/crates/sqlload/src/adapters/motherduck_bridge/main/_run_statement.rs @@ -0,0 +1,46 @@ +//! Run SQL on a bridge connection. + +use crate::adapters::motherduck_bridge::_helpers::process::{failure, send}; +use crate::adapters::motherduck_bridge::errors::BridgeError; +use crate::adapters::motherduck_bridge::models::BridgeConnection; + +/// Run `sql` (one statement or several, separated by `;`) with text `params` bound to its `?`s, after every earlier statement on `connection`, and return the last statement's rows: each value as text (numbers and booleans as their SQL text), `None` for `NULL`. +pub(in crate::adapters) async fn run_statement( + connection: &BridgeConnection, + sql: String, + params: Vec, +) -> Result>>, BridgeError> { + let cursor = &connection.cursor; + let request = + serde_json::json!({"op": "run", "cursor": cursor.id, "sql": sql, "params": params}); + let received = send(&cursor.process, request)?; + let python = cursor.process.python.clone(); + let reply = tokio::task::spawn_blocking(move || received.recv()) + .await + .map_err(|error| BridgeError::Stopped(error.to_string()))? + .map_err(|_| { + BridgeError::Stopped(format!("the MotherDuck bridge on {python} stopped")) + })??; + if let Some((error, _)) = failure(&reply) { + return Err(BridgeError::Engine(error)); + } + let rows = reply["rows"].as_array().cloned().unwrap_or_default(); + Ok(rows + .iter() + .map(|row| { + let values = row.as_array().map(Vec::as_slice).unwrap_or_default(); + values.iter().map(text).collect() + }) + .collect()) +} + +/// A reply value as text: `None` for `NULL`, booleans as SQL spells them. +fn text(value: &serde_json::Value) -> Option { + match value { + serde_json::Value::Null => None, + serde_json::Value::String(text) => Some(text.clone()), + serde_json::Value::Bool(true) => Some("true".to_owned()), + serde_json::Value::Bool(false) => Some("false".to_owned()), + other => Some(other.to_string()), + } +} diff --git a/crates/sqlload/src/adapters/motherduck_bridge/main/mod.rs b/crates/sqlload/src/adapters/motherduck_bridge/main/mod.rs new file mode 100644 index 00000000..1322dca7 --- /dev/null +++ b/crates/sqlload/src/adapters/motherduck_bridge/main/mod.rs @@ -0,0 +1,4 @@ +pub(in crate::adapters) mod _append_batch; +pub(in crate::adapters) mod _fork_connection; +pub(in crate::adapters) mod _open_bridge; +pub(in crate::adapters) mod _run_statement; diff --git a/crates/sqlload/src/adapters/motherduck_bridge/mod.rs b/crates/sqlload/src/adapters/motherduck_bridge/mod.rs new file mode 100644 index 00000000..e523812a --- /dev/null +++ b/crates/sqlload/src/adapters/motherduck_bridge/mod.rs @@ -0,0 +1,7 @@ +//! The MotherDuck bridge: SQL and batches run on Python's `duckdb` package in a separate process, as MotherDuck doesn't accept SQLLoad's own DuckDB yet; `website/src/content/docs/docs/targets/motherduck.mdx` says how to remove it. + +pub(crate) mod _helpers; +pub mod constants; +pub mod errors; +pub mod main; +pub mod models; diff --git a/crates/sqlload/src/adapters/motherduck_bridge/models.rs b/crates/sqlload/src/adapters/motherduck_bridge/models.rs new file mode 100644 index 00000000..273110ae --- /dev/null +++ b/crates/sqlload/src/adapters/motherduck_bridge/models.rs @@ -0,0 +1,45 @@ +use std::collections::HashMap; +use std::sync::atomic::AtomicU64; +use std::sync::mpsc::Sender; +use std::sync::{Arc, Mutex}; + +use crate::adapters::motherduck_bridge::errors::BridgeError; + +/// A connection to the database the bridge opened: one DuckDB cursor of the bridge process, shared by its clones and closed when the last one drops. +#[derive(Debug, Clone)] +pub struct BridgeConnection { + pub cursor: Arc, +} + +/// One cursor of the bridge process, by its id; the bridge creates it on its first statement and runs its statements in order. +#[derive(Debug)] +pub struct BridgeCursor { + pub id: u64, + pub process: Arc, +} + +impl Drop for BridgeCursor { + fn drop(&mut self) { + let close = serde_json::json!({"op": "close", "cursor": self.id}); + let _ = self.process.lines.send(close.to_string()); + } +} + +/// A running bridge process: the lines going to its stdin, the requests waiting for its replies, and the tail of its stderr. The process ends when the last handle drops, which closes its stdin. +#[derive(Debug)] +pub struct BridgeProcess { + /// The Python it runs on, as messages name it. + pub python: String, + pub lines: Sender, + pub replies: Arc>, + pub next_request: AtomicU64, + pub next_cursor: AtomicU64, + pub stderr: Arc>, +} + +/// The requests waiting for a reply, by id, each with where its reply goes; once the process has ended, why, which every later request fails with. +#[derive(Debug, Default)] +pub struct Replies { + pub waiting: HashMap>>, + pub stopped: Option, +} diff --git a/packages/sqlload/README.md b/packages/sqlload/README.md index a6f69fd9..4af89a26 100644 --- a/packages/sqlload/README.md +++ b/packages/sqlload/README.md @@ -94,7 +94,7 @@ versions and how to run them yourself. ¹ In builds with the `odbc` feature. -² Once MotherDuck accepts DuckDB 2.0 clients. +² From 0.34.0 (not released yet), through a separate DuckDB 1.5 process: install `sqlload[motherduck]`. See [Supported systems](https://sqlload.io/docs/supported-systems/) for every combination and write strategy. diff --git a/packages/sqlload/pyproject.toml b/packages/sqlload/pyproject.toml index a40b11cb..991934c6 100644 --- a/packages/sqlload/pyproject.toml +++ b/packages/sqlload/pyproject.toml @@ -20,6 +20,12 @@ classifiers = [ # orjson writes gathered dict rows as JSON bytes, which Rust parses without the GIL. dependencies = ["orjson>=3.10"] +[project.optional-dependencies] +# MotherDuck targets: MotherDuck accepts DuckDB 1.4.1 to 1.5.6 (raise the bound as MotherDuck does), not the DuckDB +# 2.0 SQLLoad bundles, so SQLLoad runs their SQL on this package in a separate Python process. Remove once MotherDuck +# accepts SQLLoad's DuckDB. +motherduck = ["duckdb>=1.4.1,<=1.5.6"] + [project.scripts] sqlload = "sqlload._cli:main" diff --git a/packages/sqlload/python/sqlload/_cli.py b/packages/sqlload/python/sqlload/_cli.py index fd9aa517..37ac4f76 100644 --- a/packages/sqlload/python/sqlload/_cli.py +++ b/packages/sqlload/python/sqlload/_cli.py @@ -1,6 +1,7 @@ """The `sqlload` console script: the native command line, run in this Python process.""" import contextlib +import os import signal import sys import threading @@ -13,6 +14,9 @@ def main() -> None: if threading.current_thread() is threading.main_thread(): # Python's handler only sets a flag, which the native command never checks while it runs; Ctrl-C ends it as it ends the binary. signal.signal(signal.SIGINT, signal.SIG_DFL) + # A MotherDuck target runs its SQL on this Python's duckdb package (the sqlload[motherduck] extra), in a process of its own. + if not os.environ.get("SQLLOAD_MOTHERDUCK_PYTHON") and sys.executable: + os.environ["SQLLOAD_MOTHERDUCK_PYTHON"] = sys.executable sys.stdout.flush() sys.stderr.flush() args = ["sqlload", *sys.argv[1:]] diff --git a/packages/sqlload/tests/test_motherduck.py b/packages/sqlload/tests/test_motherduck.py new file mode 100644 index 00000000..6e22b18e --- /dev/null +++ b/packages/sqlload/tests/test_motherduck.py @@ -0,0 +1,357 @@ +"""MotherDuck targets through the bridge, offline: `md:` databases opened by a stand-in for MotherDuck that keeps each one in a local DuckDB file, beside the same tables loaded into DuckDB itself.""" + +import json +import os +import subprocess +import sys +import textwrap +from datetime import date, datetime, time, timezone +from decimal import Decimal +from pathlib import Path +from typing import Any + +import duckdb +import pytest + +pytestmark = pytest.mark.skipif(os.name == "nt", reason="the stand-in Python is a shell script") + +PROJECT = """\ +python_paths = ["sources"] + +[connections.wh] +kind = "duckdb" +database = "wh.duckdb" + +[connections.md] +kind = "motherduck" +database = "shop" +token = "${ENV:SHOP_MD_TOKEN}" + +[targets.local] +connection = "wh" +loader_schema = "raw" + +[targets.cloud] +connection = "md" +loader_schema = "raw" +""" + +#: The token the project signs in with: `&`, `=` and `#` must reach MotherDuck whole. +TOKEN = "tok&en=#1" + +#: Stands in for MotherDuck in the bridge's Python: `md:?motherduck_token=...` opens `.duckdb` in MD_FOLDER +#: after checking the token, with another database attached as MotherDuck attaches the account's others. MD_REFUSE makes +#: it refuse as MotherDuck's extension does on a DuckDB version it doesn't support; MD_DIE_ON ends the bridge when a +#: statement holds that text; MD_LOG records every statement. +STAND_IN = r''' +import os +import sys +import urllib.parse + +import duckdb + +_connect = duckdb.connect +FOLDER = os.environ["MD_FOLDER"] + + +def log(sql): + if os.environ.get("MD_LOG"): + with open(os.environ["MD_LOG"], "a") as out: + out.write(" ".join(sql.split()) + "\n") + + +class Cursor: + def __init__(self, inner): + self.inner = inner + + def execute(self, sql, params=None): + log(sql) + if os.environ.get("MD_DIE_ON") and os.environ["MD_DIE_ON"] in sql: + sys.stderr.write(f"the stand-in for MotherDuck ended the bridge, signed in as {os.environ['MD_EXPECTED_TOKEN']}\n") + sys.stderr.flush() + os._exit(3) + return self.inner.execute(sql, params) + + def __getattr__(self, name): + return getattr(self.inner, name) + + +class Database: + def __init__(self, inner): + self.inner = inner + + def cursor(self): + return Cursor(self.inner.cursor()) + + +def connect(database=":memory:", *args, config=None, **kwargs): + if not str(database).startswith("md:"): + return _connect(database, *args, config=config, **kwargs) + log(f"python: {sys.executable}") + name, _, query = database[3:].partition("?") + token = dict(urllib.parse.parse_qsl(query)).get("motherduck_token") + if token != os.environ["MD_EXPECTED_TOKEN"]: + raise duckdb.InvalidInputException("Invalid Input Error: the token was not the one expected") + if os.environ.get("MD_REFUSE"): + raise duckdb.InvalidInputException( + 'Invalid Input Error: Initialization function "motherduck_duckdb_cpp_init" from file "motherduck.duckdb_extension" ' + 'threw an exception: "\nYour DuckDB version (v2.0.0) is not yet supported by MotherDuck. The latest supported ' + f'version is v1.5.6. Please downgrade to use MotherDuck. {database}\n"' + ) + inner = _connect(os.path.join(FOLDER, f"{name}.duckdb"), config=config or {}) + inner.execute(f"ATTACH '{os.path.join(FOLDER, 'other.duckdb')}' AS other") + inner.execute("CREATE SCHEMA IF NOT EXISTS other.raw; CREATE TABLE IF NOT EXISTS other.raw.cloud__cloud_rows (stray INTEGER)") + return Database(inner) + + +duckdb.connect = connect +''' + +#: Each table yields the rows in data/.json; a row whose `raise` is set raises it after the rows before it. +MODULE = ''' +import json +from datetime import date, datetime, time, timezone +from decimal import Decimal +from pathlib import Path +from uuid import UUID + +import sqlload as sl + +HERE = Path(__file__).resolve().parent.parent + + +def rows(ctx): + # local_rows and cloud_rows share data/rows.json: table names are unique across a project. + for row in json.loads((HERE / "data" / f"{ctx.table.removeprefix('local_').removeprefix('cloud_')}.json").read_text()): + if "raise" in row: + raise RuntimeError(row["raise"]) + yield row + + +def typed(ctx): + yield { + "flag": True, "small": 7, "big": 9007199254740993, "amount": Decimal("12.50"), "ratio": 0.25, + "label": "café ☕", "code": "ab", "day": date(2026, 11, 12), "at_time": time(13, 45, 30, 123456), + "at_utc": datetime(2026, 11, 12, 9, 30, 0, 654321, tzinfo=timezone.utc), "at_local": datetime(2026, 11, 12, 9, 30, 1), + "doc": {"k": [1, "two"]}, "uid": "0f8fad5b-d9cb-469f-a165-70867728950e", "tags": ["a", "b"], + "empty": None, + } + + +TYPED = dict( + description="Every portable type.", + columns={ + "flag": "bool", "small": "smallint", "big": "int", "amount": "decimal(10,2)", "ratio": "float", "label": "text", + "code": "varchar(4)", "day": "date", "at_time": "time", "at_utc": "timestamp_tz", "at_local": "timestamp_ntz", + "doc": "json", "uid": "uuid", "blob": "binary", "tags": "array(text)", "empty": "text", + }, +) + +local = sl.source("local", destination="local") +cloud = sl.source("cloud", destination="cloud") + +{tables} +''' + + +def make_project(tmp_path: Path, tables: str) -> Path: + for folder in ("sources", "data", "md", "bin", "shim", "home"): + (tmp_path / folder).mkdir() + (tmp_path / "sqlload.toml").write_text(PROJECT) + (tmp_path / "sources" / "shop.py").write_text(MODULE.replace("{tables}", textwrap.dedent(tables))) + (tmp_path / "shim" / "sitecustomize.py").write_text(STAND_IN) + python = tmp_path / "bin" / "python" + python.write_text(f'#!/bin/sh\nPYTHONPATH="{tmp_path / "shim"}" exec "{sys.executable}" "$@"\n') + python.chmod(0o755) + return tmp_path + + +def give(project: Path, table: str, rows: Any) -> None: + (project / "data" / f"{table}.json").write_text(json.dumps(rows)) + + +def sqlload(project: Path, *args: str, **env: str) -> subprocess.CompletedProcess[str]: + home = project / "home" + environment = { + **os.environ, + "HOME": str(home), + "XDG_CONFIG_HOME": str(home / ".config"), + "XDG_CACHE_HOME": str(home / ".cache"), + "SHOP_MD_TOKEN": TOKEN, + "MD_EXPECTED_TOKEN": TOKEN, + "MD_FOLDER": str(project / "md"), + "SQLLOAD_MOTHERDUCK_PYTHON": str(project / "bin" / "python"), + } + for name in ("MOTHERDUCK_TOKEN", "motherduck_token", "PYTHONPATH"): + environment.pop(name, None) + environment.update(env) + return subprocess.run( + [sys.executable, "-c", "from sqlload._cli import main; main()", *args], cwd=project, env=environment, capture_output=True, text=True + ) + + +def load(project: Path, *args: str, **env: str) -> subprocess.CompletedProcess[str]: + done = sqlload(project, "load", *args, **env) + assert done.returncode == 0, done.stdout + done.stderr + return done + + +def query(path: Path, sql: str) -> list[tuple[Any, ...]]: + with duckdb.connect(str(path), read_only=True) as connection: + return connection.execute(sql).fetchall() + + +def cloud(project: Path, sql: str) -> list[tuple[Any, ...]]: + return query(project / "md" / "shop.duckdb", sql) + + +def local(project: Path, sql: str) -> list[tuple[Any, ...]]: + return query(project / "wh.duckdb", sql) + + +@pytest.mark.parametrize( + ("settings", "first", "second", "expected"), + [ + ("write_strategy='append'", [{"id": 1, "v": "a"}, {"id": 2, "v": "b"}], [{"id": 3, "v": "c"}], [(1, "a"), (2, "b"), (3, "c")]), + ( + "write_strategy='merge', unique_key='id'", + [{"id": 1, "v": "a"}, {"id": 2, "v": "b"}], + [{"id": 2, "v": "B"}, {"id": 3, "v": "c"}], + [(1, "a"), (2, "B"), (3, "c")], + ), + ("write_strategy='replace'", [{"id": 1, "v": "a"}, {"id": 2, "v": "b"}], [{"id": 3, "v": "c"}], [(3, "c")]), + ( + "write_strategy='delete_insert', cursor_column='day'", + [{"id": 1, "v": "a", "day": "2026-01-01T00:00:00Z"}, {"id": 2, "v": "b", "day": "2026-01-02T00:00:00Z"}], + [ + {"id": 1, "v": "stale", "day": "2026-01-01T00:00:00Z"}, + {"id": 2, "v": "B", "day": "2026-01-02T00:00:00Z"}, + {"id": 3, "v": "c", "day": "2026-01-03T00:00:00Z"}, + ], + [(1, "a"), (2, "B"), (3, "c")], + ), + ], + ids=["append", "merge", "replace", "delete_insert"], +) +def test_given_a_table_when_loaded_twice_into_motherduck_then_its_strategy_applies_as_on_duckdb( + tmp_path: Path, settings: str, first: list[dict[str, Any]], second: list[dict[str, Any]], expected: list[tuple[Any, ...]] +) -> None: + project = make_project(tmp_path, f"local.table(description='Rows.', name='local_rows', {settings})(rows)\ncloud.table(description='Rows.', name='cloud_rows', {settings})(rows)\n") + give(project, "rows", first) + load(project) + give(project, "rows", second) + load(project) + assert cloud(project, "SELECT id, v FROM raw.cloud__cloud_rows ORDER BY id") == expected + assert local(project, "SELECT id, v FROM raw.local__local_rows ORDER BY id") == expected + assert cloud(project, "SELECT count(*) FROM information_schema.tables WHERE table_name LIKE '%sqlload_staging%'") == [(0,)] + assert query(project / "md" / "other.duckdb", "SELECT column_name FROM information_schema.columns WHERE table_name = 'cloud__cloud_rows'") == [("stray",)] + + +def test_given_every_portable_type_when_loaded_then_motherduck_holds_what_duckdb_holds(tmp_path: Path) -> None: + project = make_project(tmp_path, "local.table(name='local_typed', **TYPED)(typed)\ncloud.table(name='cloud_typed', **TYPED)(typed)\n") + load(project) + columns = "SELECT column_name, data_type FROM information_schema.columns WHERE table_name = '{}' ORDER BY ordinal_position" + cloud_columns = cloud(project, columns.format("cloud__cloud_typed")) + assert cloud_columns == local(project, columns.format("local__local_typed")) + assert dict(cloud_columns)["at_utc"] == "TIMESTAMP WITH TIME ZONE" + values = "SET TimeZone = 'UTC'; SELECT CAST(COLUMNS(* EXCLUDE (_sqlload_load_id, _sqlload_loaded_at)) AS VARCHAR) FROM raw.{}" + loaded = cloud(project, values.format("cloud__cloud_typed")) + assert loaded == local(project, values.format("local__local_typed")) + assert dict(zip([name for name, _ in cloud_columns], loaded[0], strict=False)) == { + "flag": "true", + "small": "7", + "big": "9007199254740993", + "amount": "12.50", + "ratio": "0.25", + "label": "café ☕", + "code": "ab", + "day": "2026-11-12", + "at_time": "13:45:30.123456", + "at_utc": "2026-11-12 09:30:00.654321+00", + "at_local": "2026-11-12 09:30:01", + "doc": '{"k":[1,"two"]}', + "uid": "0f8fad5b-d9cb-469f-a165-70867728950e", + "blob": None, + "tags": "[a, b]", + "empty": None, + } + + +def test_given_an_incremental_table_when_loaded_again_and_rerun_then_only_new_rows_arrive(tmp_path: Path) -> None: + project = make_project(tmp_path, "cloud.table(description='Rows.', name='rows', write_strategy='append', cursor_column='at')(rows)\n") + give(project, "rows", [{"id": 1, "at": "2026-01-01T10:00:00Z"}, {"id": 2, "at": "2026-01-02T10:00:00Z"}]) + load(project) + give(project, "rows", [{"id": 2, "at": "2026-01-02T10:00:00Z"}, {"id": 3, "at": "2026-01-03T10:00:00Z", "note": "new column"}]) + load(project) + load(project) + assert cloud(project, "SELECT id, note FROM raw.cloud__rows ORDER BY id") == [(1, None), (2, None), (3, "new column")] + + +def test_given_the_bridge_ends_mid_load_when_loading_then_the_table_fails_keeps_its_rows_and_quotes_it_without_the_token(tmp_path: Path) -> None: + project = make_project(tmp_path, "cloud.table(description='Rows.', name='rows', write_strategy='merge', unique_key='id')(rows)\n") + give(project, "rows", [{"id": 1, "v": "a"}]) + load(project) + give(project, "rows", [{"id": 1, "v": "A"}, {"id": 2, "v": "b"}]) + done = sqlload(project, "load", MD_DIE_ON="read_parquet") + assert done.returncode == 1, done.stdout + done.stderr + assert "the MotherDuck bridge on" in done.stdout + done.stderr + assert "the stand-in for MotherDuck ended the bridge, signed in as " in done.stdout + done.stderr + assert TOKEN not in done.stdout + done.stderr + assert cloud(project, "SELECT id, v FROM raw.cloud__rows ORDER BY id") == [(1, "a")] + load(project) + assert cloud(project, "SELECT id, v FROM raw.cloud__rows ORDER BY id") == [(1, "A"), (2, "b")] + + +def test_given_a_value_motherduck_rejects_when_loading_then_its_error_is_relayed_and_nothing_commits(tmp_path: Path) -> None: + project = make_project(tmp_path, "cloud.table(description='Rows.', name='rows', write_strategy='append', columns={'doc': 'json'})(rows)\n") + give(project, "rows", [{"id": 1, "doc": "{}"}]) + load(project) + give(project, "rows", [{"id": 2, "doc": "{}"}, {"id": 3, "doc": "{not json"}]) + done = sqlload(project, "load") + assert done.returncode == 1, done.stdout + done.stderr + assert "MotherDuck: Conversion Error" in done.stdout + done.stderr + assert cloud(project, "SELECT id FROM raw.cloud__rows ORDER BY id") == [(1,)] + + +@pytest.mark.parametrize( + ("env", "expected"), + [ + ( + {"SQLLOAD_MOTHERDUCK_PYTHON": "/no/such/python"}, + "MotherDuck connects through Python's duckdb package (duckdb 1.4.1 to 1.5.6), and /no/such/python cannot run", + ), + ( + {"MD_REFUSE": "1"}, + "MotherDuck doesn't accept the version of Python's duckdb package that SQLLoad connects through: install duckdb 1.4.1 to 1.5.6 there (pip install 'sqlload[motherduck]'), or set SQLLOAD_MOTHERDUCK_PYTHON to a Python that has it. MotherDuck says: Your DuckDB version (v2.0.0) is not yet supported by MotherDuck.", + ), + ({"MD_EXPECTED_TOKEN": "another"}, "MotherDuck: cannot open md:shop: Invalid Input Error: the token was not the one expected"), + ], + ids=["no-python", "refused-version", "refused-token"], +) +def test_given_the_bridge_cannot_open_motherduck_when_debugging_then_it_says_why_without_the_token( + tmp_path: Path, env: dict[str, str], expected: str +) -> None: + project = make_project(tmp_path, "") + done = sqlload(project, "debug", **env) + output = done.stdout + done.stderr + assert done.returncode == 1, output + assert expected in output + assert "tok" not in output.replace("token", "") + + +def test_given_a_python_without_duckdb_when_debugging_then_it_says_to_install_the_extra(tmp_path: Path) -> None: + project = make_project(tmp_path, "") + (project / "shim" / "sitecustomize.py").write_text("import sys\nsys.modules['duckdb'] = None\n") + done = sqlload(project, "debug") + assert done.returncode == 1, done.stdout + done.stderr + assert "lacks (" in done.stdout + done.stderr + assert "install sqlload[motherduck] with pip or uv" in done.stdout + done.stderr + + +def test_given_no_python_is_named_when_debugging_then_motherduck_runs_on_this_packages_python(tmp_path: Path) -> None: + project = make_project(tmp_path, "") + log = project / "statements.log" + done = sqlload(project, "debug", SQLLOAD_MOTHERDUCK_PYTHON="", PYTHONPATH=str(project / "shim"), MD_LOG=str(log)) + assert done.returncode == 0, done.stdout + done.stderr + assert "MotherDuck shop, DuckDB v" in done.stdout + assert f"python: {sys.executable}" in log.read_text().splitlines() diff --git a/packages/sqlload/tests/test_motherduck_live.py b/packages/sqlload/tests/test_motherduck_live.py new file mode 100644 index 00000000..9d1400e6 --- /dev/null +++ b/packages/sqlload/tests/test_motherduck_live.py @@ -0,0 +1,177 @@ +"""MotherDuck targets against a live MotherDuck account, on duckdb 1.5.6. Runs only with SQLLOAD_TEST_MOTHERDUCK_TOKEN set; each test creates +a database named sqlload_test_ and drops it afterwards.""" + +import json +import os +import shutil +import subprocess +import sys +import textwrap +import uuid +from collections.abc import Iterator +from pathlib import Path +from typing import Any + +import pytest + +TOKEN_VARIABLE = "SQLLOAD_TEST_MOTHERDUCK_TOKEN" + +pytestmark = pytest.mark.skipif(not os.environ.get(TOKEN_VARIABLE), reason=f"{TOKEN_VARIABLE} is not set") + +PROJECT = """\ +python_paths = ["sources"] + +[connections.md] +kind = "motherduck" +database = "{database}" +token = "${{ENV:{variable}}}" + +[targets.cloud] +connection = "md" +loader_schema = "raw" +""" + +MODULE = ''' +import json +from pathlib import Path + +import sqlload as sl + +HERE = Path(__file__).resolve().parent.parent +cloud = sl.source("cloud", destination="cloud") + + +def rows(ctx): + yield from json.loads((HERE / "data" / f"{ctx.table}.json").read_text()) + + +cloud.table(description="Appended by cursor.", name="events", write_strategy="append", cursor_column="at")(rows) +cloud.table(description="Merged by id.", name="customers", write_strategy="merge", unique_key="id")(rows) +cloud.table(description="Replaced by cursor window.", name="days", write_strategy="delete_insert", cursor_column="day")(rows) +cloud.table(description="Replaced whole.", name="snapshot", write_strategy="replace")(rows) +cloud.table( + description="Every portable type.", + name="typed", + columns={ + "flag": "bool", "small": "smallint", "big": "int", "amount": "decimal(10,2)", "ratio": "float", "label": "text", + "day": "date", "at_utc": "timestamp_tz", "at_local": "timestamp_ntz", "doc": "json", "uid": "uuid", "tags": "array(text)", + }, +)(rows) +''' + +#: Runs SQL on MotherDuck from the duckdb 1.5.6 environment: the database and statements as JSON on stdin, the last rows as JSON on stdout. +ADMIN = """ +import json, sys, urllib.parse +import duckdb +request = json.load(sys.stdin) +connection = duckdb.connect("md:?motherduck_token=" + urllib.parse.quote(request["token"], safe="")) +rows = [] +for sql in request["sql"]: + rows = connection.execute(sql).fetchall() +print(json.dumps(rows, default=str)) +""" + + +@pytest.fixture(scope="module") +def python(tmp_path_factory: pytest.TempPathFactory) -> str: + """A Python with duckdb 1.5.6, the newest version MotherDuck accepts.""" + folder = tmp_path_factory.mktemp("duckdb15") + uv = shutil.which("uv") + assert uv, "the live MotherDuck tests need uv to build a duckdb 1.5.6 environment" + subprocess.run([uv, "venv", "--quiet", "--python", "3.12", str(folder / "venv")], check=True) + executable = folder / "venv" / ("Scripts/python.exe" if os.name == "nt" else "bin/python") + subprocess.run([uv, "pip", "install", "--quiet", "--python", str(executable), "duckdb==1.5.6"], check=True) + return str(executable) + + +def admin(python: str, *sql: str) -> list[list[Any]]: + request = json.dumps({"token": os.environ[TOKEN_VARIABLE], "sql": list(sql)}) + done = subprocess.run([python, "-I", "-c", ADMIN], input=request, capture_output=True, text=True) + assert done.returncode == 0, done.stderr.replace(os.environ[TOKEN_VARIABLE], "") + return json.loads(done.stdout) + + +@pytest.fixture +def database(python: str) -> Iterator[str]: + name = f"sqlload_test_{uuid.uuid4().hex[:12]}" + admin(python, f"CREATE DATABASE {name}") + try: + yield name + finally: + admin(python, f"DROP DATABASE IF EXISTS {name}") + + +def sqlload(project: Path, python: str, *args: str) -> subprocess.CompletedProcess[str]: + environment = {**os.environ, "SQLLOAD_MOTHERDUCK_PYTHON": python} + for name in ("MOTHERDUCK_TOKEN", "motherduck_token"): + environment.pop(name, None) + done = subprocess.run( + [sys.executable, "-c", "from sqlload._cli import main; main()", *args], cwd=project, env=environment, capture_output=True, text=True + ) + assert os.environ[TOKEN_VARIABLE] not in done.stdout + done.stderr + return done + + +def load(project: Path, python: str) -> str: + done = sqlload(project, python, "load") + assert done.returncode == 0, done.stdout + done.stderr + return done.stdout + + +def give(project: Path, rows: dict[str, list[dict[str, Any]]]) -> None: + for table, table_rows in rows.items(): + (project / "data" / f"{table}.json").write_text(json.dumps(table_rows)) + + +def test_given_a_live_account_when_loading_twice_and_rerunning_then_every_strategy_and_type_lands( + tmp_path: Path, python: str, database: str +) -> None: + for folder in ("sources", "data"): + (tmp_path / folder).mkdir() + (tmp_path / "sqlload.toml").write_text(PROJECT.format(database=database, variable=TOKEN_VARIABLE)) + (tmp_path / "sources" / "shop.py").write_text(textwrap.dedent(MODULE)) + typed = { + "flag": True, "small": 7, "big": 9007199254740993, "amount": "12.50", "ratio": 0.25, "label": "café ☕", "day": "2026-11-12", + "at_utc": "2026-11-12T09:30:00.654321Z", "at_local": "2026-11-12T09:30:01", "doc": {"k": [1, "two"]}, + "uid": "0f8fad5b-d9cb-469f-a165-70867728950e", "tags": ["a", "b"], + } + give(tmp_path, { + "events": [{"id": 1, "at": "2026-01-01T10:00:00Z"}, {"id": 2, "at": "2026-01-02T10:00:00Z"}], + "customers": [{"id": 1, "name": "a"}, {"id": 2, "name": "b"}], + "days": [{"id": 1, "v": "a", "day": "2026-01-01T00:00:00Z"}, {"id": 2, "v": "b", "day": "2026-01-02T00:00:00Z"}], + "snapshot": [{"id": 1}, {"id": 2}], + "typed": [typed], + }) + debug = sqlload(tmp_path, python, "debug") + assert debug.returncode == 0, debug.stdout + debug.stderr + assert f"MotherDuck {database}, DuckDB v1.5.6" in debug.stdout + load(tmp_path, python) + give(tmp_path, { + "events": [{"id": 2, "at": "2026-01-02T10:00:00Z"}, {"id": 3, "at": "2026-01-03T10:00:00Z", "note": "new column"}], + "customers": [{"id": 2, "name": "B"}, {"id": 3, "name": "c"}], + "days": [{"id": 2, "v": "B", "day": "2026-01-02T00:00:00Z"}, {"id": 3, "v": "c", "day": "2026-01-03T00:00:00Z"}], + "snapshot": [{"id": 3}], + }) + load(tmp_path, python) + load(tmp_path, python) + tables = f"{database}.raw" + assert admin( + python, + f"SELECT id, note FROM {tables}.cloud__events ORDER BY id", + ) == [[1, None], [2, None], [3, "new column"]] + assert admin(python, f"SELECT id, name FROM {tables}.cloud__customers ORDER BY id") == [[1, "a"], [2, "B"], [3, "c"]] + assert admin(python, f"SELECT id, v FROM {tables}.cloud__days ORDER BY id") == [[1, "a"], [2, "B"], [3, "c"]] + assert admin(python, f"SELECT id FROM {tables}.cloud__snapshot ORDER BY id") == [[3]] + assert admin( + python, + "SET TimeZone = 'UTC'", + f"SELECT flag, small, big, CAST(amount AS VARCHAR), ratio, label, CAST(day AS VARCHAR), CAST(at_utc AS VARCHAR), " + f"CAST(at_local AS VARCHAR), CAST(doc AS VARCHAR), CAST(uid AS VARCHAR), tags FROM {tables}.cloud__typed", + ) == [[ + True, 7, 9007199254740993, "12.50", 0.25, "café ☕", "2026-11-12", "2026-11-12 09:30:00.654321+00", "2026-11-12 09:30:01", + '{"k":[1,"two"]}', "0f8fad5b-d9cb-469f-a165-70867728950e", ["a", "b"], + ]] + assert admin( + python, + f"SELECT count(*) FROM duckdb_tables() WHERE database_name = '{database}' AND table_name LIKE '%sqlload_staging%'", + ) == [[0]] diff --git a/website/public/index.html b/website/public/index.html index d8783680..6d199529 100644 --- a/website/public/index.html +++ b/website/public/index.html @@ -293,7 +293,7 @@

Start in a minute

ODBC databases¹✓
-

¹ In builds with the odbc feature.
² Once MotherDuck accepts DuckDB 2.0 clients.
See the connector matrix for every combination and write strategy.

+

¹ In builds with the odbc feature.
² From 0.34.0, through a separate DuckDB 1.5 process: install sqlload[motherduck].
See the connector matrix for every combination and write strategy.

  • Linux and macOS; Python only for Python sources.
  • Free and open source, Apache-2.0.
    • diff --git a/website/src/content/docs/docs/configuration.mdx b/website/src/content/docs/docs/configuration.mdx index d0f11a9a..662975b1 100644 --- a/website/src/content/docs/docs/configuration.mdx +++ b/website/src/content/docs/docs/configuration.mdx @@ -104,8 +104,8 @@ See [ClickHouse](/docs/targets/clickhouse/). ### `motherduck` -Available as a beta from 0.33.0. MotherDuck doesn't accept DuckDB 2.0 clients yet, so this -connection can't connect until it does; see [MotherDuck](/docs/targets/motherduck/). +Available as a beta from 0.33.0. From 0.34.0 (not released yet) it connects through a separate +DuckDB 1.5 process: install `sqlload[motherduck]`; see [MotherDuck](/docs/targets/motherduck/). | Key | Meaning | |-----|---------| diff --git a/website/src/content/docs/docs/index.mdx b/website/src/content/docs/docs/index.mdx index 0dd2f5c4..789b84e7 100644 --- a/website/src/content/docs/docs/index.mdx +++ b/website/src/content/docs/docs/index.mdx @@ -24,7 +24,7 @@ SQLLoad loads data into your warehouse, and shows you how before it changes anyt ¹ In builds with the `odbc` feature. -² Once MotherDuck accepts DuckDB 2.0 clients. +² From 0.34.0 (not released yet), through a separate DuckDB 1.5 process: install `sqlload[motherduck]`. See the [connector matrix](/docs/supported-systems/) for every combination and write strategy. diff --git a/website/src/content/docs/docs/supported-systems.mdx b/website/src/content/docs/docs/supported-systems.mdx index f5c209a8..3a627578 100644 --- a/website/src/content/docs/docs/supported-systems.mdx +++ b/website/src/content/docs/docs/supported-systems.mdx @@ -30,7 +30,7 @@ description: "What SQLLoad reads and writes, by which route, and how each is tes | ClickHouse (preview) | `append`, `merge`, `delete_insert`, `replace` | Arrow over HTTP into a staging table, published by swapping in a whole table or whole partitions; Parquet in S3 read by ClickHouse with `s3()`. See [ClickHouse](/docs/targets/clickhouse/). | | Postgres | `append`, `merge`, `delete_insert`, `replace` | Binary `COPY` into a temporary staging table, then one transaction per commit. See [Postgres](/docs/targets/postgres/). | | DuckDB | `append`, `merge`, `delete_insert`, `replace` | Arrow appends; Parquet files read by DuckDB with `read_parquet`. | -| MotherDuck (beta, from 0.33.0) | `append`, `merge`, `delete_insert`, `replace` | Arrow appends through DuckDB's `md:` connection; files are read by SQLLoad. MotherDuck doesn't accept DuckDB 2.0 clients yet, so it can't connect until it does. See [MotherDuck](/docs/targets/motherduck/). | +| MotherDuck (beta, from 0.33.0) | `append`, `merge`, `delete_insert`, `replace` | From 0.34.0 (not released yet), batches inserted by a separate DuckDB 1.5 process (`sqlload[motherduck]`), as MotherDuck doesn't accept DuckDB 2.0 clients yet; files are read by SQLLoad. See [MotherDuck](/docs/targets/motherduck/). | | Parquet or CSV files in S3 | `append`, `delete_insert`, `replace` | Parquet written by SQLLoad, zstd-compressed; or, from 0.27.0, CSV: plain, gzip or zstd. | | Parquet or CSV files in a local folder | As for S3; from 0.27.0, `replace` swaps the folder in one rename | As for S3 | @@ -63,7 +63,7 @@ them into DuckDB. | Postgres | Real Postgres in containers, as a source and as a target. `postgres_cdc` on Postgres 15, 16 and 17, plus live loads into DuckDB, Snowflake and Databricks. See [Postgres](/docs/targets/postgres/#tested). | | ClickHouse | Real ClickHouse 25.3, 25.8 and 26.10 in containers, with an S3-compatible store for the files it reads itself. See [ClickHouse](/docs/targets/clickhouse/#tested). | | DuckDB | Real, in process. | -| MotherDuck | Not tested against a live MotherDuck account. Offline tests cover the connection's `database` and token settings, the refusals (no token, MotherDuck's version message), the Arrow-only route, and loads naming only the current database. MotherDuck's real extension was checked refusing DuckDB 2.0.0. | +| MotherDuck | Live against a MotherDuck account on DuckDB 1.5.6: every write strategy, an incremental load and rerun, a new column and every portable type. Offline, every change loads through the separate process into a stand-in for MotherDuck and compares with DuckDB. See [MotherDuck](/docs/targets/motherduck/#tested). | | ODBC | psqlODBC against real Postgres in containers, into DuckDB. | | BigQuery source | A fake Storage Read API in tests (parallel streams, a broken stream resumed, cursor filters), and live loads from a sandbox project into DuckDB: tables, views, query tables, and integer, timestamp, text, `NUMERIC` and `BIGNUMERIC` cursors. | | S3 | A real S3-compatible store in containers; benchmarks against AWS S3. | diff --git a/website/src/content/docs/docs/targets/motherduck.mdx b/website/src/content/docs/docs/targets/motherduck.mdx index 2d394d1c..7760f968 100644 --- a/website/src/content/docs/docs/targets/motherduck.mdx +++ b/website/src/content/docs/docs/targets/motherduck.mdx @@ -1,28 +1,40 @@ --- title: "MotherDuck" -description: "Load into MotherDuck (beta): DuckDB's md: connection, with rows appended as Arrow." +description: "Load into MotherDuck (beta) through a separate DuckDB 1.5 process, with the same SQL and write strategies as DuckDB." --- -Available as a beta from 0.33.0. A MotherDuck target loads into a MotherDuck database through -DuckDB's `md:` connection, with the same SQL, types and write strategies as a -[DuckDB target](/docs/targets/duckdb/). It's in beta, and it hasn't been tested against a live -MotherDuck account. +Available as a beta from 0.33.0; working against a live MotherDuck account from 0.34.0 (not released yet). +A MotherDuck target loads into a MotherDuck database with the same SQL, types and write strategies as a +[DuckDB target](/docs/targets/duckdb/). -## DuckDB 2.0 isn't supported by MotherDuck yet +## Install -SQLLoad bundles DuckDB 2.0. MotherDuck doesn't accept DuckDB 2.0 clients yet: its extension -supports DuckDB up to 1.5.6 and refuses to load on 2.0. Until MotherDuck supports DuckDB 2.0, a -MotherDuck target can't connect. `sqlload debug`, `plan` and `load` report MotherDuck's own -message: +SQLLoad bundles DuckDB 2.0, and MotherDuck doesn't accept DuckDB 2.0 clients yet: it supports +DuckDB 1.4.1 to 1.5.6 ([duckdb-versions.json](https://motherduck.com/docs/duckdb-versions.json)). +So SQLLoad sends a MotherDuck target's SQL to a separate Python process running Python's `duckdb` +package at a version MotherDuck accepts. The two DuckDB versions never share a process, and every +other target still uses SQLLoad's own DuckDB. + +Install the `motherduck` extra, which adds `duckdb>=1.4.1,<=1.5.6`: + +```bash +uv tool install 'sqlload[motherduck]' +# or, in a virtual environment +pip install 'sqlload[motherduck]' +``` + +The `sqlload` command runs that process on its own Python. With `sqlload-cli`, or to use another +Python, set `SQLLOAD_MOTHERDUCK_PYTHON` to a Python that has `duckdb` 1.4.1 to 1.5.6; without it, +`sqlload-cli` uses `python3` on `PATH` (`python` on Windows). + +If the Python can't run, lacks `duckdb`, or has a version MotherDuck refuses, `sqlload debug`, +`plan` and `load` say so and how to fix it: ```console $ sqlload debug -fail target shop MotherDuck: cannot open md:shop: MotherDuck doesn't accept the DuckDB version this SQLLoad bundles yet, so no SQLLoad release can connect to it until it does. MotherDuck says: Your DuckDB version (v2.0.0) is not yet supported by MotherDuck. … +fail target shop MotherDuck: cannot open md:shop: MotherDuck doesn't accept the version of Python's duckdb package that SQLLoad connects through: install duckdb 1.4.1 to 1.5.6 there (pip install 'sqlload[motherduck]'), or set SQLLOAD_MOTHERDUCK_PYTHON to a Python that has it. MotherDuck says: Your DuckDB version (v2.0.0) is not yet supported by MotherDuck. … ``` -MotherDuck lists the client versions it supports in -[duckdb-versions.json](https://motherduck.com/docs/duckdb-versions.json). - ## Connection ```toml @@ -42,24 +54,44 @@ loader_schema = "raw" | `token` | A MotherDuck access token. Use `${ENV:NAME}`. Default: the `motherduck_token` or `MOTHERDUCK_TOKEN` environment variable. | A target's `database` overrides the connection's, so one connection can serve, say, a `shop_dev` -and a `shop` database. MotherDuck attaches the account's other databases too, so SQLLoad names every -table with its database, and a target that names no database is refused. Without a token SQLLoad refuses the connection rather than wait for -MotherDuck's browser sign-in. A `duckdb` connection refuses an `md:` path; use -`kind = "motherduck"`. The first connection installs MotherDuck's DuckDB extension into `~/.duckdb`, -which needs the network once. +and a `shop` database. The database must exist. MotherDuck attaches the account's other databases +too, so SQLLoad names every table with its database, and a target that names no database is refused. +Without a token SQLLoad refuses the connection rather than wait for MotherDuck's browser sign-in. +The token reaches the separate process over a pipe, never in its command line, and is taken out of +every error. A `duckdb` connection refuses an `md:` path; use `kind = "motherduck"`. The first +connection installs MotherDuck's DuckDB extension into `~/.duckdb`, which needs the network once. ## How data arrives -Every row reaches MotherDuck as Arrow batches appended inside the batch's transaction, as on DuckDB. -Unlike DuckDB, MotherDuck never reads files itself: files sources, including Parquet in S3 or a -local folder, and a Snowflake source's unloaded files are read by SQLLoad and appended as Arrow. -`plan` names that route for every table. +SQLLoad writes each batch to a temporary Parquet file on this machine, and the separate process +inserts it with `read_parquet` inside the batch's transaction, then the file is removed. MotherDuck +never reads your source files itself: files sources, including Parquet in S3 or a local folder, and a +Snowflake source's unloaded files are read by SQLLoad. `plan` names that route for every table. + +`append`, `merge`, `delete_insert` and `replace` run the same statements as on DuckDB. Each table +loads on its own connection in the separate process, so one table's transaction never mixes with +another's. If the process ends mid-load, the open batch isn't committed, the table fails, and the next +load picks up from the table's committed rows. + +## Tested + +- Live, against a MotherDuck account on DuckDB 1.5.6: create, load, every write strategy, an + incremental load and a rerun that adds nothing, a new column, every portable type, and dropping + the test databases afterwards. +- Offline, on every change: every write strategy and portable type loaded through the separate + process into a stand-in for MotherDuck and compared with the same load into DuckDB, an incremental + rerun, the process ending mid-load, a value MotherDuck rejects, and the setup errors. -`append`, `merge`, `delete_insert` and `replace` run the same statements as on DuckDB. +## Removing the separate process -## Limits +Once MotherDuck accepts the DuckDB version SQLLoad bundles, the separate process is no longer needed. +Everything specific to it is behind one interface, the `Bridge` connection of the DuckDB adapter: -- Not tested against a live MotherDuck account. The tests cover the connection settings, the - refusals and the route; the statements are DuckDB's, tested on DuckDB. -- Each table loads on its own connection, as on DuckDB. Whether MotherDuck runs concurrent schema - changes and transactions across those connections exactly as local DuckDB does is untested. +1. Delete `crates/sqlload/src/adapters/motherduck_bridge/` and its `mod` line in `adapters/mod.rs`. +2. In `adapters/duckdb`, remove `DuckDbConnection::Bridge` (keeping `Local` as the connection), the + `Bridge` branches in `main/`, the `Bridge` error, and open MotherDuck with + `duckdb::Connection::open(connection_string(...))` as 0.33.0 did. +3. Remove the `motherduck` extra from `packages/sqlload/pyproject.toml`, the + `SQLLOAD_MOTHERDUCK_PYTHON` line in `packages/sqlload/python/sqlload/_cli.py`, and + `packages/sqlload/tests/test_motherduck.py`. +4. Rewrite the Install section of this page.