Add column-selective appender API with default support (#642)

Fixes #551

I did not end up using `duckdb_append_default_to_chunk` because:

- When limiting active columns, DuckDB automatically fills all the other
columns with their default values at flush time
- The Rust appender is row-oriented, not chunk-based
- `duckdb_append_default_to_chunk` is unstable

Author: mlafeldt

crates/duckdb/src/appender/mod.rs

@@ -204,6 +204,31 @@ impl Appender<'_> {
             result_from_duckdb_appender(res, &mut self.app)
         }
     }
+
+    /// Add a column to the appender's active column list.
+    ///
+    /// When columns are added, only those columns need values during append.
+    /// Other columns will use their DEFAULT value (or NULL if no default).
+    ///
+    /// This flushes any pending data before modifying the column list.
+    #[inline]
+    pub fn add_column(&mut self, name: &str) -> Result<()> {
+        let c_name = std::ffi::CString::new(name)?;
+        let rc = unsafe { ffi::duckdb_appender_add_column(self.app, c_name.as_ptr() as *const c_char) };
+        result_from_duckdb_appender(rc, &mut self.app)
+    }
+
+    /// Clear the appender's active column list.
+    ///
+    /// After clearing, all columns become active again and values must be
+    /// provided for every column during append.
+    ///
+    /// This flushes any pending data before clearing.
+    #[inline]
+    pub fn clear_columns(&mut self) -> Result<()> {
+        let rc = unsafe { ffi::duckdb_appender_clear_columns(self.app) };
+        result_from_duckdb_appender(rc, &mut self.app)
+    }
 }
 
 impl Drop for Appender<'_> {
@@ -424,4 +449,110 @@ mod test {
 
         Ok(())
     }
+
+    #[test]
+    fn test_appender_defaults_and_column_switching() -> Result<()> {
+        let db = Connection::open_in_memory()?;
+        db.execute_batch("CREATE TABLE foo(a INT DEFAULT 99, b INT, c INT DEFAULT 7)")?;
+
+        // Only provide column b; a and c should use their defaults
+        {
+            let mut app = db.appender_with_columns("foo", &["b"])?;
+            app.append_row([Some(1)])?;
+            app.append_row([Option::<i32>::None])?;
+        }
+
+        // Switch to a different active column set, then back to full width
+        {
+            let mut app = db.appender("foo")?;
+            app.add_column("c")?;
+            app.add_column("a")?;
+            app.append_row([10, 1])?; // set c and a; b gets NULL
+
+            app.clear_columns()?; // revert to all columns
+            app.append_row([2, 3, 4])?;
+        }
+
+        let rows: Vec<(i32, Option<i32>, i32)> = db
+            .prepare("SELECT a, b, c FROM foo ORDER BY a, b NULLS LAST")?
+            .query_map([], |row| Ok((row.get(0)?, row.get(1)?, row.get(2)?)))?
+            .collect::<Result<_>>()?;
+
+        assert_eq!(
+            rows,
+            vec![
+                (1, None, 10),    // add_column path; b NULL, c set
+                (2, Some(3), 4),  // clear_columns path; all provided
+                (99, Some(1), 7), // defaults applied for a and c
+                (99, None, 7)     // default + NULL
+            ]
+        );
+
+        Ok(())
+    }
+
+    #[test]
+    fn test_appender_with_columns_sequence_default() -> Result<()> {
+        let db = Connection::open_in_memory()?;
+        db.execute_batch(
+            "CREATE SEQUENCE seq START 1;
+             CREATE TABLE foo(id INTEGER DEFAULT nextval('seq'), name TEXT)",
+        )?;
+
+        {
+            let mut app = db.appender_with_columns("foo", &["name"])?;
+            app.append_row(["Alice"])?;
+            app.append_row(["Bob"])?;
+            app.append_row(["Charlie"])?;
+        }
+
+        let rows: Vec<(i32, String)> = db
+            .prepare("SELECT id, name FROM foo ORDER BY id")?
+            .query_map([], |row| Ok((row.get(0)?, row.get(1)?)))?
+            .collect::<Result<_>>()?;
+
+        assert_eq!(
+            rows,
+            vec![(1, "Alice".into()), (2, "Bob".into()), (3, "Charlie".into())]
+        );
+
+        Ok(())
+    }
+
+    #[test]
+    fn test_appender_with_columns_to_db_schema() -> Result<()> {
+        let db = Connection::open_in_memory()?;
+        db.execute_batch(
+            "CREATE SCHEMA s;
+             CREATE TABLE s.foo(a INTEGER DEFAULT 5, b INTEGER)",
+        )?;
+
+        {
+            let mut app = db.appender_with_columns_to_db("foo", "s", &["b"])?;
+            app.append_row([7])?;
+        }
+
+        let (a, b): (i32, i32) = db.query_row("SELECT a, b FROM s.foo", [], |row| Ok((row.get(0)?, row.get(1)?)))?;
+        assert_eq!((a, b), (5, 7));
+        Ok(())
+    }
+
+    #[test]
+    fn test_appender_with_columns_to_catalog_and_db() -> Result<()> {
+        let db = Connection::open_in_memory()?;
+        db.execute_batch(
+            "CREATE SCHEMA s;
+             CREATE TABLE s.bar(a INTEGER DEFAULT 11, b INTEGER)",
+        )?;
+
+        {
+            // Default in-memory catalog is "memory"
+            let mut app = db.appender_with_columns_to_catalog_and_db("bar", "memory", "s", &["b"])?;
+            app.append_row([9])?;
+        }
+
+        let (a, b): (i32, i32) = db.query_row("SELECT a, b FROM s.bar", [], |row| Ok((row.get(0)?, row.get(1)?)))?;
+        assert_eq!((a, b), (11, 9));
+        Ok(())
+    }
 }

crates/duckdb/src/inner_connection.rs

@@ -85,7 +85,7 @@ impl InnerConnection {
     }
 
     pub fn execute(&mut self, sql: &str) -> Result<()> {
-        let c_str = CString::new(sql).unwrap();
+        let c_str = CString::new(sql)?;
         unsafe {
             let mut out = mem::zeroed();
             let r = ffi::duckdb_query_arrow(self.con, c_str.as_ptr() as *const c_char, &mut out);
@@ -96,7 +96,7 @@ impl InnerConnection {
     }
 
     pub fn prepare<'a>(&mut self, conn: &'a Connection, sql: &str) -> Result<Statement<'a>> {
-        let c_str = CString::new(sql).unwrap();
+        let c_str = CString::new(sql)?;
 
         // Extract statements (handles both single and multi-statement queries)
         let mut extracted = ptr::null_mut();
@@ -162,8 +162,8 @@ impl InnerConnection {
 
     pub fn appender<'a>(&mut self, conn: &'a Connection, table: &str, schema: &str) -> Result<Appender<'a>> {
         let mut c_app: ffi::duckdb_appender = ptr::null_mut();
-        let c_table = CString::new(table).unwrap();
-        let c_schema = CString::new(schema).unwrap();
+        let c_table = CString::new(table)?;
+        let c_schema = CString::new(schema)?;
         let r = unsafe {
             ffi::duckdb_appender_create(
                 self.con,
@@ -184,9 +184,9 @@ impl InnerConnection {
         schema: &str,
     ) -> Result<Appender<'a>> {
         let mut c_app: ffi::duckdb_appender = ptr::null_mut();
-        let c_table = CString::new(table).unwrap();
-        let c_catalog = CString::new(catalog).unwrap();
-        let c_schema = CString::new(schema).unwrap();
+        let c_table = CString::new(table)?;
+        let c_catalog = CString::new(catalog)?;
+        let c_schema = CString::new(schema)?;
 
         let r = unsafe {
             ffi::duckdb_appender_create_ext(
@@ -201,6 +201,26 @@ impl InnerConnection {
         Ok(Appender::new(conn, c_app))
     }
 
+    pub fn appender_with_columns<'a>(
+        &mut self,
+        conn: &'a Connection,
+        table: &str,
+        schema: &str,
+        catalog: Option<&str>,
+        columns: &[&str],
+    ) -> Result<Appender<'a>> {
+        // The C API only supports narrowing columns after the appender is created.
+        // Create the appender first, then activate the requested column subset.
+        let mut appender = match catalog {
+            Some(catalog) => self.appender_to_catalog_and_db(conn, table, catalog, schema)?,
+            None => self.appender(conn, table, schema)?,
+        };
+        for column in columns {
+            appender.add_column(column)?;
+        }
+        Ok(appender)
+    }
+
     pub fn get_interrupt_handle(&self) -> Arc<InterruptHandle> {
         self.interrupt.clone()
     }

crates/duckdb/src/lib.rs

@@ -553,6 +553,57 @@ impl Connection {
             .appender_to_catalog_and_db(self, table, catalog, schema)
     }
 
+    /// Create an Appender that only provides values for specific columns.
+    ///
+    /// Columns not in the list will use their DEFAULT value, or NULL if no default.
+    /// This supports all types of DEFAULT expressions including non-deterministic
+    /// ones like `random()`, `current_timestamp`, or sequences.
+    ///
+    /// ## Example
+    ///
+    /// ```rust,no_run
+    /// # use duckdb::{Connection, Result};
+    /// fn insert_partial(conn: &Connection) -> Result<()> {
+    ///     // Table: CREATE TABLE foo(id INT DEFAULT nextval('seq'), name TEXT, created TIMESTAMP DEFAULT current_timestamp)
+    ///     let mut app = conn.appender_with_columns("foo", &["name"])?;
+    ///     // Only provide name; id and created use their defaults
+    ///     app.append_row(["Alice"])?;
+    ///     app.append_row(["Bob"])?;
+    ///     Ok(())
+    /// }
+    /// ```
+    ///
+    /// # Failure
+    ///
+    /// Will return `Err` if `table` does not exist or a column name is invalid.
+    pub fn appender_with_columns(&self, table: &str, columns: &[&str]) -> Result<Appender<'_>> {
+        self.appender_with_columns_to_db(table, &DatabaseName::Main.to_string(), columns)
+    }
+
+    /// Create an Appender that only provides values for specific columns, with schema.
+    ///
+    /// See [`appender_with_columns`](Connection::appender_with_columns) for details.
+    pub fn appender_with_columns_to_db(&self, table: &str, schema: &str, columns: &[&str]) -> Result<Appender<'_>> {
+        self.db
+            .borrow_mut()
+            .appender_with_columns(self, table, schema, None, columns)
+    }
+
+    /// Create an Appender that only provides values for specific columns, with catalog and schema.
+    ///
+    /// See [`appender_with_columns`](Connection::appender_with_columns) for details.
+    pub fn appender_with_columns_to_catalog_and_db(
+        &self,
+        table: &str,
+        catalog: &str,
+        schema: &str,
+        columns: &[&str],
+    ) -> Result<Appender<'_>> {
+        self.db
+            .borrow_mut()
+            .appender_with_columns(self, table, schema, Some(catalog), columns)
+    }
+
     /// Get a handle to interrupt long-running queries.
     ///
     /// ## Example