Adds MotherDuck support to DuckDB native fdw. (#511)

* Adds MotherDuck support to duckdb native fdw. 

MotherDuck is remote DuckDB instance that gets ATTACHed to the local one so we're able to honor the LIMIT TO / EXCEPT table list arguments to `IMPORT FOREIGN SCHEMA` by pushing them down to the query agains the remote information_schema.

MotherDuck doesn't use a DuckDB secret block, instead it relies on env vars (not relevant here) or configuration settings. Despite what the docs say, passing those settings as url params to the first argument to ATTACH doesn't seem to work any longer.

* Fixed code formatting issue.

* Fix code formatting issues and a bug the crept in while cleaning up the
code.

* change duckdb_fdw version number

---------

Co-authored-by: Bo Lu <[email protected]>

Author: scottynomad

wrappers/src/fdw/duckdb_fdw/README.md

@@ -10,6 +10,6 @@ This is a foreign data wrapper for [DuckDB](https://duckdb.org/). It is develope
 
 | Version | Date       | Notes                                                |
 | ------- | ---------- | ---------------------------------------------------- |
+| 0.1.2   | 2025-10-16 | Add MotherDuck support                              |
 | 0.1.1   | 2025-08-15 | Replace execute_batch() with execute()               |
 | 0.1.0   | 2024-10-31 | Initial version                                      |
-

wrappers/src/fdw/duckdb_fdw/duckdb_fdw.rs

@@ -9,7 +9,7 @@ use supabase_wrappers::prelude::*;
 use super::{mapper, server_type::ServerType, DuckdbFdwError, DuckdbFdwResult};
 
 #[wrappers_fdw(
-    version = "0.1.1",
+    version = "0.1.2",
     author = "Supabase",
     website = "https://github.com/supabase/wrappers/tree/main/wrappers/src/fdw/duckdb_fdw",
     error_type = "DuckdbFdwError"
@@ -28,13 +28,8 @@ impl DuckdbFdw {
 
     fn init_duckdb(&self) -> DuckdbFdwResult<()> {
         let sql_batch = String::default()
-            + if self.svr_type.is_iceberg() { "install iceberg;load iceberg;" } else { "" }
-            // security tips: https://duckdb.org/docs/stable/operations_manual/securing_duckdb/overview
-            + "
-                set disabled_filesystems = 'LocalFileSystem';
-                set allow_community_extensions = false;
-                set lock_configuration = true;
-            "
+            + self.svr_type.get_duckdb_extension_sql()
+            + &self.svr_type.get_settings_sql(&self.svr_opts)
             + &self.svr_type.get_create_secret_sql(&self.svr_opts)
             + &self.svr_type.get_attach_sql(&self.svr_opts)?;
 
@@ -291,6 +286,47 @@ impl ForeignDataWrapper<DuckdbFdwError> for DuckdbFdw {
                 .iter()
                 .map(|t| (t.clone(), t.replace(".", "_")))
                 .collect()
+        } else if self.svr_type.is_sql_like() {
+            let db_name = self.svr_type.as_str();
+            let schema = import_stmt.remote_schema;
+
+            let table_list = import_stmt
+                .table_list
+                .iter()
+                .map(|name| format!("'{}'", name.replace("'", "''")))
+                .collect::<Vec<_>>()
+                .join(",");
+            let table_filter = match import_stmt.list_type {
+                ImportSchemaType::FdwImportSchemaAll => "".to_string(),
+                ImportSchemaType::FdwImportSchemaLimitTo => {
+                    format!("and table_name in ({table_list})")
+                }
+                ImportSchemaType::FdwImportSchemaExcept => {
+                    format!("and table_name not in ({table_list})")
+                }
+            };
+            let query = format!(
+                "
+                    select table_name
+                    from information_schema.tables
+                    where table_catalog = ?
+                      and table_schema = ?
+                    {table_filter}
+                    order by table_name
+                "
+            );
+            let mut stmt = self.conn.prepare(&query)?;
+            let tables = stmt
+                .query_map([db_name, &schema], |row| row.get::<_, String>(0))?
+                .collect::<Result<Vec<_>, _>>()?;
+
+            // NOTE: We don't do any munging of the postgres table names because
+            // the pg code will ignore CREATE FOREIGN TABLE statements that don't target
+            // the table names specified in the IMPORT FOREIGN SCHEMA statement.
+            tables
+                .iter()
+                .map(|t| (format!("{db_name}.{schema}.{t}"), t.clone()))
+                .collect()
         } else {
             let tables = require_option_or("tables", &import_stmt.options, "")
                 .split(",")

wrappers/src/fdw/duckdb_fdw/server_type.rs

@@ -19,6 +19,9 @@ pub(super) enum ServerType {
     R2Catalog,
     Polaris,
     Lakekeeper,
+
+    // SQL-like Remotes
+    MotherDuck,
 }
 
 impl ServerType {
@@ -32,6 +35,7 @@ impl ServerType {
             "r2_catalog" => Self::R2Catalog,
             "polaris" => Self::Polaris,
             "lakekeeper" => Self::Lakekeeper,
+            "md" => Self::MotherDuck,
             _ => return Err(DuckdbFdwError::InvalidServerType(svr_type.to_owned())),
         };
         Ok(ret)
@@ -46,6 +50,7 @@ impl ServerType {
             Self::R2Catalog => "r2_catalog",
             Self::Polaris => "polaris",
             Self::Lakekeeper => "lakekeeper",
+            Self::MotherDuck => "md",
         }
     }
 
@@ -56,6 +61,20 @@ impl ServerType {
         )
     }
 
+    pub(super) fn is_sql_like(&self) -> bool {
+        matches!(self, Self::MotherDuck)
+    }
+
+    pub(super) fn get_duckdb_extension_sql(&self) -> &'static str {
+        match self {
+            Self::Iceberg | Self::S3Tables | Self::R2Catalog | Self::Polaris | Self::Lakekeeper => {
+                "install iceberg;load iceberg;"
+            }
+            Self::MotherDuck => "install md;load md;",
+            _ => "",
+        }
+    }
+
     fn allowed_secret_params(&self) -> Vec<&'static str> {
         match self {
             // ref: https://duckdb.org/docs/stable/core_extensions/httpfs/s3api.html#overview-of-s3-secret-parameters
@@ -81,6 +100,7 @@ impl ServerType {
                 "oauth2_scope",
                 "oauth2_server_uri",
             ],
+            Self::MotherDuck => vec![],
         }
     }
 
@@ -123,6 +143,8 @@ impl ServerType {
             Self::R2Catalog | Self::Polaris | Self::Lakekeeper => {
                 vec![("iceberg", self.allowed_secret_params())]
             }
+
+            _ => vec![],
         };
 
         let mut ret = String::default();
@@ -134,6 +156,44 @@ impl ServerType {
         ret
     }
 
+    pub(super) fn get_settings_sql(&self, svr_opts: &ServerOptions) -> String {
+        let settings: Vec<(&str, String)> = match self {
+            Self::MotherDuck => {
+                let token = if svr_opts.contains_key("vault_motherduck_token") {
+                    get_vault_secret(svr_opts.get("vault_motherduck_token").unwrap()).ok_or_else(
+                        || OptionsError::OptionNameNotFound("vault_motherduck_token".to_string()),
+                    )
+                } else {
+                    require_option("motherduck_token", svr_opts).map(|s| s.to_string())
+                }
+                .expect("motherduck_token is required");
+
+                let sanitized_token = format!("'{}'", token.replace("'", "''"));
+                vec![
+                    ("motherduck_token", sanitized_token),
+                    ("motherduck_attach_mode", "'single'".to_string()),
+                    ("allow_community_extensions", "false".to_string()),
+                    // Has the same effect as below, disables the local filesystem and locks the config.
+                    ("motherduck_saas_mode", "true".to_string()),
+                ]
+            }
+            _ => {
+                // security tips: https://duckdb.org/docs/stable/operations_manual/securing_duckdb/overview
+                vec![
+                    ("disabled_filesystems", "'LocalFileSystem'".to_string()),
+                    ("allow_community_extensions", "false".to_string()),
+                    ("lock_configuration", "true".to_string()),
+                ]
+            }
+        };
+        let mut ret = String::default();
+        for (key, value) in settings {
+            ret.push_str(&format!("set {key}={value};"));
+        }
+
+        ret
+    }
+
     pub(super) fn get_attach_sql(&self, svr_opts: &ServerOptions) -> DuckdbFdwResult<String> {
         let ret = match self {
             Self::S3Tables => {
@@ -143,7 +203,7 @@ impl ServerType {
                     "
                     attach '{arn}' as {db_name} (
                         type iceberg,
-                        endpoint_type s3_tables  
+                        endpoint_type s3_tables
                     );"
                 )
             }
@@ -160,6 +220,11 @@ impl ServerType {
                     );"
                 )
             }
+            Self::MotherDuck => {
+                let database = require_option("database", svr_opts)?.replace("'", "''");
+                let db_name = self.as_str();
+                format!("attach 'md:{database}' as {db_name};")
+            }
             _ => String::default(),
         };
         Ok(ret)

wrappers/src/fdw/duckdb_fdw/tests.rs

@@ -50,6 +50,18 @@ mod tests {
                 &[],
             )
             .unwrap();
+            c.update(
+                r#"CREATE SERVER duckdb_server_motherduck
+                    FOREIGN DATA WRAPPER duckdb_wrapper
+                    OPTIONS (
+                        type 'md',
+                        database 'my_db',
+                        motherduck_token 'my_token'
+                    )"#,
+                None,
+                &[],
+            )
+            .unwrap();
             c.update(r#"CREATE SCHEMA IF NOT EXISTS duckdb"#, None, &[])
                 .unwrap();
             c.update(
@@ -71,7 +83,6 @@ mod tests {
                 &[],
             )
             .unwrap();
-
             let results = c
                 .select(
                     "SELECT * FROM duckdb.s3_0_test_data order by name",