feat(motherduck): connect through a separate DuckDB 1.5 process (install sqlload[motherduck]), as MotherDuck doesn't accept DuckDB 2.0 clients yet

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.

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<T: Send + 'static>(
     destination: &DuckDbDestination,
     work: impl FnOnce(&duckdb::Connection) -> Result<T, DuckDbError> + Send + 'static,
 ) -> Result<T, DuckDbError> {
-    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()

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), "<redacted>")
@@ -81,12 +82,12 @@ pub(in crate::adapters::duckdb) fn refused(error: &str, token: &str, shown: &str
         .collect::<Vec<_>>()
         .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 {

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<DuckDbDestination, DuckDbError> {
     connection.register_table_function::<duckdb::vtab::arrow::ArrowVTab>("arrow")?;
     Ok(DuckDbDestination {
-        connection: Arc::new(Mutex::new(connection)),
+        connection: DuckDbConnection::Local(Arc::new(Mutex::new(connection))),
         ddl: Arc::default(),
-        motherduck,
+        motherduck: false,
     })
 }

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),
 }

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<String> = 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<String> = 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(?)"

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<Option<Vec<LiveColumn>>, 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::<Result<Vec<_>, _>>()?;
+        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<String>]) -> Result<LiveColumn, DuckDbError> {
+    let value = |index: usize| row.get(index).cloned().flatten();
+    let number = |index: usize| {
+        value(index)
+            .map(|text| text.parse::<i64>())
+            .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,
+    })
+}

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<String>,
 ) -> 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) {

crates/sqlload/src/adapters/duckdb/main/_open_destination.rs

@@ -13,7 +13,7 @@ pub(in crate::adapters) fn open_destination(
 ) -> Result<DuckDbDestination, DuckDbError> {
     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()

crates/sqlload/src/adapters/duckdb/main/_open_duckdb.rs

@@ -16,5 +16,5 @@ pub(in crate::adapters) fn open_duckdb(path: &str) -> Result<DuckDbDestination,
         ":memory:" => duckdb::Connection::open_in_memory()?,
         path => duckdb::Connection::open(path)?,
     };
-    shared_destination(connection, false)
+    shared_destination(connection)
 }

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<DuckDbDestination, DuckDbError> {
     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,
+    })
 }

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<DuckDbDestination, DuckDbError> {
-    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,
     })

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<Option<String>, 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<String>>(0))?)
     })

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<Vec<String>, 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<String>>(0))?;