duckdb/appender/
mod.rs

1use super::{AppenderParams, Connection, Result, ValueRef, ffi};
2use std::{ffi::c_void, fmt, os::raw::c_char};
3
4use crate::{
5    Error,
6    error::{error_from_appender_code, result_from_duckdb_appender},
7    types::{ToSql, ToSqlOutput, to_duckdb_hugeint, to_duckdb_uhugeint, value_ref_from_value},
8};
9
10/// Appender for fast import data
11///
12/// # Error Reporting
13///
14/// DuckDB buffers appended rows and checks constraints (`NOT NULL`, foreign
15/// keys, ...) when the buffered data is flushed to the table. That can happen
16/// through an explicit [`flush`](Appender::flush), through DuckDB's internal
17/// appender flush while appending many rows, or when the appender is dropped.
18/// As a result, [`append_row`](Appender::append_row) may return `Ok` for a row
19/// that later fails a constraint check.
20///
21/// Always call `flush` explicitly and check its result before dropping the
22/// appender. Dropping an appender flushes any remaining rows, but errors
23/// raised during that implicit flush are discarded because `Drop` cannot
24/// report them.
25///
26/// # Thread Safety
27///
28/// `Appender` is `Sync` but not `Send`:
29/// - `Sync` because every method that touches the underlying DuckDB appender
30///   takes `&mut self`, so no mutation is reachable through a shared
31///   `&Appender`. This is in the spirit of [`std::sync::Exclusive`].
32/// - Not `Send` because it holds a reference to `Connection`, which is `!Sync`.
33///
34/// To use an appender in another thread, move the `Connection` to that thread
35/// and create the appender there.
36///
37/// See [DuckDB concurrency documentation](https://duckdb.org/docs/stable/connect/concurrency.html) for more details.
38///
39/// [`std::sync::Exclusive`]: https://doc.rust-lang.org/std/sync/struct.Exclusive.html
40///
41/// # Wide Tables (Many Columns)
42///
43/// Array literals `[value; N]` are supported for tables with up to 32 columns.
44///
45/// ```rust,ignore
46/// appender.append_row([0; 32])?;
47/// appender.append_row([1, 2, 3, 4, 5])?;
48/// ```
49///
50/// For tables with more than 32 columns, use one of these alternatives:
51///
52/// ## 1. Slice approach - convert values to `&dyn ToSql`
53///
54/// ```rust,ignore
55/// let values: Vec<i32> = vec![0; 100];
56/// let params: Vec<&dyn ToSql> = values.iter().map(|v| v as &dyn ToSql).collect();
57/// appender.append_row(params.as_slice())?;
58/// ```
59///
60/// ## 2. `params!` macro - write values explicitly
61///
62/// ```rust,ignore
63/// appender.append_row(params![v1, v2, v3, ..., v50])?;
64/// ```
65///
66/// ## 3. `appender_params_from_iter` - pass an iterator directly
67///
68/// ```rust,ignore
69/// use duckdb::appender_params_from_iter;
70/// let values: Vec<i32> = vec![0; 100];
71/// appender.append_row(appender_params_from_iter(values))?;
72/// ```
73///
74/// All three methods can be used interchangeably and mixed in the same appender.
75pub struct Appender<'conn> {
76    conn: &'conn Connection,
77    app: ffi::duckdb_appender,
78}
79
80// SAFETY: every method that touches the appender pointer takes `&mut self`.
81// The only shared access exposed by the public API (the `Debug` impl) reads
82// `Connection`'s immutable `path` field without touching the appender pointer
83// or the connection's `RefCell`. No mutation is reachable through a shared
84// `&Appender`, so sharing one across threads is sound. This is in the spirit
85// of `std::sync::Exclusive`, with an additional read-only `Debug` view. `Send`
86// stays disallowed because `&Connection: !Send` (`Connection` is `!Sync`).
87unsafe impl Sync for Appender<'_> {}
88
89#[cfg(feature = "appender-arrow")]
90mod arrow;
91
92impl Appender<'_> {
93    /// Append multiple rows from Iterator
94    ///
95    /// ## Example
96    ///
97    /// ```rust,no_run
98    /// # use duckdb::{Connection, Result, params};
99    /// fn insert_rows(conn: &Connection) -> Result<()> {
100    ///     let mut app = conn.appender("foo")?;
101    ///     app.append_rows([[1, 2], [3, 4], [5, 6], [7, 8], [9, 10]])?;
102    ///     Ok(())
103    /// }
104    /// ```
105    ///
106    /// # Failure
107    ///
108    /// Returns `Err` if a row cannot be appended.
109    #[inline]
110    pub fn append_rows<P, I>(&mut self, rows: I) -> Result<()>
111    where
112        I: IntoIterator<Item = P>,
113        P: AppenderParams,
114    {
115        for row in rows {
116            self.append_row(row)?;
117        }
118        Ok(())
119    }
120
121    /// Append one row
122    ///
123    /// ## Example
124    ///
125    /// ```rust,no_run
126    /// # use duckdb::{Connection, Result, params};
127    /// fn insert_row(conn: &Connection) -> Result<()> {
128    ///     let mut app = conn.appender("foo")?;
129    ///     app.append_row([1, 2])?;
130    ///     Ok(())
131    /// }
132    /// ```
133    ///
134    /// # Failure
135    ///
136    /// Returns `Err` if the row cannot be appended.
137    #[inline]
138    pub fn append_row<P: AppenderParams>(&mut self, params: P) -> Result<()> {
139        // AppenderParams normalizes arrays, tuples, slices, and iterators into
140        // append_parameter_row, which validates up-front, then ends the row
141        // after binding.
142        params.__bind_in(self)
143    }
144
145    #[inline]
146    pub(crate) fn append_parameter_row<P>(&mut self, params: P) -> Result<()>
147    where
148        P: IntoIterator,
149        P::Item: ToSql,
150    {
151        let params = params.into_iter().collect::<Vec<_>>();
152        let values = params
153            .iter()
154            .map(ToSql::to_sql)
155            .collect::<Result<Vec<ToSqlOutput<'_>>>>()?;
156
157        self.validate_parameter_values(&values)?;
158
159        if let Err(err) = self.bind_parameter_values(&values) {
160            // A failed bind can leave DuckDB with a partial row. End the row
161            // only to let DuckDB mark the appender invalid, then preserve the
162            // original bind error; the cleanup failure is usually less useful.
163            if !self.app.is_null() {
164                let rc = unsafe { ffi::duckdb_appender_end_row(self.app) };
165                if rc != ffi::DuckDBSuccess {
166                    unsafe {
167                        ffi::duckdb_appender_destroy(&mut self.app);
168                    }
169                }
170            }
171            return Err(err);
172        }
173        let rc = unsafe { ffi::duckdb_appender_end_row(self.app) };
174        result_from_duckdb_appender(rc, &mut self.app)
175    }
176
177    fn validate_parameter_values(&self, values: &[ToSqlOutput<'_>]) -> Result<()> {
178        for value in values {
179            let value = to_value_ref(value)?;
180            validate_appender_value_ref(value)?;
181        }
182        Ok(())
183    }
184
185    fn bind_parameter_values(&mut self, values: &[ToSqlOutput<'_>]) -> Result<()> {
186        for value in values {
187            self.bind_parameter(value)?;
188        }
189        Ok(())
190    }
191
192    fn bind_parameter(&mut self, value: &ToSqlOutput<'_>) -> Result<()> {
193        let ptr = self.app;
194        let value = to_value_ref(value)?;
195        // TODO: append more
196        let rc = match value {
197            ValueRef::Null => unsafe { ffi::duckdb_append_null(ptr) },
198            ValueRef::Boolean(i) => unsafe { ffi::duckdb_append_bool(ptr, i) },
199            ValueRef::TinyInt(i) => unsafe { ffi::duckdb_append_int8(ptr, i) },
200            ValueRef::SmallInt(i) => unsafe { ffi::duckdb_append_int16(ptr, i) },
201            ValueRef::Int(i) => unsafe { ffi::duckdb_append_int32(ptr, i) },
202            ValueRef::BigInt(i) => unsafe { ffi::duckdb_append_int64(ptr, i) },
203            ValueRef::UTinyInt(i) => unsafe { ffi::duckdb_append_uint8(ptr, i) },
204            ValueRef::USmallInt(i) => unsafe { ffi::duckdb_append_uint16(ptr, i) },
205            ValueRef::UInt(i) => unsafe { ffi::duckdb_append_uint32(ptr, i) },
206            ValueRef::UBigInt(i) => unsafe { ffi::duckdb_append_uint64(ptr, i) },
207            ValueRef::HugeInt(i) => unsafe { ffi::duckdb_append_hugeint(ptr, to_duckdb_hugeint(i)) },
208            ValueRef::UHugeInt(i) => unsafe { ffi::duckdb_append_uhugeint(ptr, to_duckdb_uhugeint(i)) },
209
210            ValueRef::Float(r) => unsafe { ffi::duckdb_append_float(ptr, r) },
211            ValueRef::Double(r) => unsafe { ffi::duckdb_append_double(ptr, r) },
212            ValueRef::Text(s) => unsafe {
213                ffi::duckdb_append_varchar_length(ptr, s.as_ptr() as *const c_char, s.len() as u64)
214            },
215            ValueRef::Timestamp(u, i) => unsafe {
216                ffi::duckdb_append_timestamp(ptr, ffi::duckdb_timestamp { micros: u.to_micros(i) })
217            },
218            ValueRef::Blob(b) | ValueRef::Geometry(b) => unsafe {
219                ffi::duckdb_append_blob(ptr, b.as_ptr() as *const c_void, b.len() as u64)
220            },
221            ValueRef::Date32(d) => unsafe { ffi::duckdb_append_date(ptr, ffi::duckdb_date { days: d }) },
222            ValueRef::Time64(u, v) => unsafe {
223                ffi::duckdb_append_time(ptr, ffi::duckdb_time { micros: u.to_micros(v) })
224            },
225            ValueRef::Interval { months, days, nanos } => unsafe {
226                ffi::duckdb_append_interval(
227                    ptr,
228                    ffi::duckdb_interval {
229                        months,
230                        days,
231                        micros: nanos / 1000,
232                    },
233                )
234            },
235            ValueRef::Decimal(d) => unsafe {
236                let decimal = crate::types::to_duckdb_decimal(d);
237                append_decimal(ptr, decimal)?
238            },
239            _ => {
240                return Err(Error::ToSqlConversionFailure(
241                    appending_unsupported_value(value.data_type()).into(),
242                ));
243            }
244        };
245        if rc != 0 {
246            return Err(error_from_appender_code(rc, self.app));
247        }
248        Ok(())
249    }
250
251    #[inline]
252    pub(super) fn new(conn: &Connection, app: ffi::duckdb_appender) -> Appender<'_> {
253        Appender { conn, app }
254    }
255
256    /// Flush buffered rows into the table.
257    ///
258    /// Constraint violations (e.g. `NOT NULL`) in previously appended rows are
259    /// reported when buffered rows are flushed. This method is the explicit way
260    /// to observe those errors, although DuckDB may also surface them from an
261    /// internal appender flush during append. Call this explicitly and check the
262    /// result before dropping the appender: the implicit flush on drop discards
263    /// any error.
264    ///
265    /// # Failure
266    ///
267    /// Returns `Err` if the buffered rows cannot be written, e.g. because a
268    /// row violates a table constraint. The appender is invalidated and the
269    /// buffered rows are lost.
270    #[inline]
271    pub fn flush(&mut self) -> Result<()> {
272        unsafe {
273            let res = ffi::duckdb_appender_flush(self.app);
274            result_from_duckdb_appender(res, &mut self.app)
275        }
276    }
277
278    /// Add a column to the appender's active column list.
279    ///
280    /// When columns are added, only those columns need values during append.
281    /// Other columns will use their DEFAULT value (or NULL if no default).
282    ///
283    /// This flushes any pending data before modifying the column list.
284    #[inline]
285    pub fn add_column(&mut self, name: &str) -> Result<()> {
286        let c_name = std::ffi::CString::new(name)?;
287        let rc = unsafe { ffi::duckdb_appender_add_column(self.app, c_name.as_ptr() as *const c_char) };
288        result_from_duckdb_appender(rc, &mut self.app)
289    }
290
291    /// Clear the appender's active column list.
292    ///
293    /// After clearing, all columns become active again and values must be
294    /// provided for every column during append.
295    ///
296    /// This flushes any pending data before clearing.
297    #[inline]
298    pub fn clear_columns(&mut self) -> Result<()> {
299        let rc = unsafe { ffi::duckdb_appender_clear_columns(self.app) };
300        result_from_duckdb_appender(rc, &mut self.app)
301    }
302}
303
304impl Drop for Appender<'_> {
305    fn drop(&mut self) {
306        if !self.app.is_null() {
307            unsafe {
308                ffi::duckdb_appender_destroy(&mut self.app);
309            }
310        }
311    }
312}
313
314impl fmt::Debug for Appender<'_> {
315    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
316        f.debug_struct("Appender").field("conn", self.conn).finish()
317    }
318}
319
320fn appending_unsupported_value(value_type: impl fmt::Display) -> String {
321    format!("appending {value_type} values is not yet supported")
322}
323
324unsafe fn append_decimal(ptr: ffi::duckdb_appender, decimal: ffi::duckdb_decimal) -> Result<ffi::duckdb_state> {
325    let mut value = unsafe { ffi::duckdb_create_decimal(decimal) };
326    if value.is_null() {
327        return Err(Error::AppendError);
328    }
329    let res = unsafe { ffi::duckdb_append_value(ptr, value) };
330    unsafe {
331        ffi::duckdb_destroy_value(&mut value);
332    }
333    Ok(res)
334}
335
336fn to_value_ref<'value, 'output>(value: &'value ToSqlOutput<'output>) -> Result<ValueRef<'value>>
337where
338    'output: 'value,
339{
340    match *value {
341        ToSqlOutput::Borrowed(v) => Ok(v),
342        ToSqlOutput::Owned(ref v) => value_ref_from_value(v, appending_unsupported_value),
343    }
344}
345
346fn validate_appender_value_ref(value: ValueRef<'_>) -> Result<()> {
347    match value {
348        ValueRef::Null
349        | ValueRef::Boolean(_)
350        | ValueRef::TinyInt(_)
351        | ValueRef::SmallInt(_)
352        | ValueRef::Int(_)
353        | ValueRef::BigInt(_)
354        | ValueRef::HugeInt(_)
355        | ValueRef::UHugeInt(_)
356        | ValueRef::UTinyInt(_)
357        | ValueRef::USmallInt(_)
358        | ValueRef::UInt(_)
359        | ValueRef::UBigInt(_)
360        | ValueRef::Float(_)
361        | ValueRef::Double(_)
362        | ValueRef::Text(_)
363        | ValueRef::Timestamp(_, _)
364        | ValueRef::Blob(_)
365        | ValueRef::Geometry(_)
366        | ValueRef::Date32(_)
367        | ValueRef::Time64(_, _)
368        | ValueRef::Interval { .. }
369        | ValueRef::Decimal(_) => Ok(()),
370        _ => Err(Error::ToSqlConversionFailure(
371            appending_unsupported_value(value.data_type()).into(),
372        )),
373    }
374}
375
376#[cfg(test)]
377mod test {
378    use super::Appender;
379    use crate::{
380        Connection, Error, Result, params,
381        types::{Decimal, Value},
382    };
383
384    #[test]
385    fn appender_is_sync() {
386        fn assert_sync<T: Sync>() {}
387
388        assert_sync::<Appender<'_>>();
389    }
390
391    #[test]
392    fn test_append_one_row() -> Result<()> {
393        let db = Connection::open_in_memory()?;
394        db.execute_batch("CREATE TABLE foo(x INTEGER)")?;
395
396        {
397            let mut app = db.appender("foo")?;
398            app.append_row([42])?;
399        }
400
401        let val = db.query_row("SELECT x FROM foo", [], |row| <(i32,)>::try_from(row))?;
402        assert_eq!(val, (42,));
403        Ok(())
404    }
405
406    #[test]
407    fn test_append_geometry_value_as_wkb_blob() -> Result<()> {
408        let db = Connection::open_in_memory()?;
409        db.execute_batch("CREATE TABLE foo(wkb BLOB)")?;
410
411        let wkb: Vec<u8> = db.query_row("SELECT ST_AsWKB('POINT EMPTY'::GEOMETRY)", [], |row| row.get(0))?;
412        let mut app = db.appender("foo")?;
413        app.append_row([Value::Geometry(wkb.clone())])?;
414        app.flush()?;
415
416        let got: Vec<u8> = db.query_row("SELECT wkb FROM foo", [], |row| row.get(0))?;
417        assert_eq!(got, wkb);
418        Ok(())
419    }
420
421    #[test]
422    fn test_append_geometry_value_to_geometry_column_fails() -> Result<()> {
423        let db = Connection::open_in_memory()?;
424        db.execute_batch("CREATE TABLE foo(geom GEOMETRY)")?;
425
426        let wkb: Vec<u8> = db.query_row("SELECT ST_AsWKB('POINT EMPTY'::GEOMETRY)", [], |row| row.get(0))?;
427        let mut app = db.appender("foo")?;
428        let err = app.append_row([Value::Geometry(wkb)]).unwrap_err();
429        let message = err.to_string();
430        assert!(
431            message.contains("BLOB") && message.contains("GEOMETRY"),
432            "expected BLOB/GEOMETRY type mismatch, got: {message}"
433        );
434        Ok(())
435    }
436
437    #[test]
438    fn test_appender_variant_columns_delegate_to_duckdb() -> Result<()> {
439        let db = Connection::open_in_memory()?;
440        db.execute_batch("CREATE TABLE foo(id INTEGER, variant_col VARIANT)")?;
441
442        let mut app = db.appender("foo")?;
443        app.append_row(params![1, "hello"])?;
444        app.flush()?;
445
446        let mut app = db.appender_with_columns("foo", &["variant_col"])?;
447        app.append_row(params!["column_subset"])?;
448        app.flush()?;
449
450        let mut app = db.appender_with_columns("foo", &["id"])?;
451        app.add_column("variant_col")?;
452        app.append_row(params![3, "added_column"])?;
453        app.flush()?;
454
455        let mut app = db.appender_with_columns("foo", &["id"])?;
456        app.clear_columns()?;
457        app.append_row(params![4, "cleared_columns"])?;
458        app.flush()?;
459
460        let count: i64 = db.query_row("SELECT COUNT(*) FROM foo WHERE variant_col IS NOT NULL", [], |row| {
461            row.get(0)
462        })?;
463        assert_eq!(count, 4);
464        Ok(())
465    }
466
467    #[test]
468    fn test_append_unsupported_container_type_returns_error() -> Result<()> {
469        use arrow::{array::ListArray, datatypes::Int32Type};
470
471        use crate::{
472            ToSql,
473            types::{ListType, ToSqlOutput, Value, ValueRef},
474        };
475
476        struct OwnedList;
477        impl ToSql for OwnedList {
478            fn to_sql(&self) -> Result<ToSqlOutput<'_>> {
479                Ok(ToSqlOutput::Owned(Value::List(vec![Value::Int(1), Value::Int(2)])))
480            }
481        }
482
483        struct BorrowedList(ListArray);
484        impl BorrowedList {
485            fn new() -> Self {
486                Self(ListArray::from_iter_primitive::<Int32Type, _, _>(vec![Some(vec![
487                    Some(1),
488                    Some(2),
489                ])]))
490            }
491        }
492        impl ToSql for BorrowedList {
493            fn to_sql(&self) -> Result<ToSqlOutput<'_>> {
494                Ok(ToSqlOutput::Borrowed(ValueRef::List(ListType::Regular(&self.0), 0)))
495            }
496        }
497
498        fn assert_unsupported_list_error(err: Error) {
499            match err {
500                Error::ToSqlConversionFailure(e) => {
501                    assert!(
502                        e.to_string().contains("appending List values is not yet supported"),
503                        "unexpected message: {e}"
504                    );
505                }
506                other => panic!("expected ToSqlConversionFailure, got {other:?}"),
507            }
508        }
509
510        let db = Connection::open_in_memory()?;
511        db.execute_batch("CREATE TABLE foo(id INTEGER, name TEXT)")?;
512
513        let list = OwnedList;
514        let mut app = db.appender("foo")?;
515        app.append_row(params![10, "before"])?;
516        app.append_row(params![11, "also before"])?;
517        let err = app.append_row(params![1, list]).unwrap_err();
518        assert_unsupported_list_error(err);
519
520        let borrowed_list = BorrowedList::new();
521        let err = app.append_row(params![3, borrowed_list]).unwrap_err();
522        assert_unsupported_list_error(err);
523        app.append_row(params![2, "ok"])?;
524        app.flush()?;
525
526        let rows = db
527            .prepare("SELECT id, name FROM foo ORDER BY id")?
528            .query_map([], |row| Ok((row.get::<_, i32>(0)?, row.get::<_, String>(1)?)))?
529            .collect::<Result<Vec<_>>>()?;
530        assert_eq!(
531            rows,
532            vec![
533                (2, "ok".to_string()),
534                (10, "before".to_string()),
535                (11, "also before".to_string())
536            ]
537        );
538        let count: i32 = db.query_row("SELECT COUNT(*) FROM foo", [], |row| row.get(0))?;
539        assert_eq!(count, 3);
540        Ok(())
541    }
542
543    #[test]
544    fn test_append_bind_failure_preserves_bind_error() -> Result<()> {
545        let db = Connection::open_in_memory()?;
546        db.execute_batch("CREATE TABLE foo(id INTEGER, value UUID)")?;
547
548        let mut app = db.appender("foo")?;
549        let err = app.append_row(params![1, 2]).unwrap_err();
550
551        match err {
552            Error::DuckDBFailure(_, Some(msg)) => {
553                assert_eq!(
554                    msg,
555                    "Failed to cast value: Unimplemented type for cast (INTEGER -> UUID)"
556                );
557            }
558            other => panic!("expected DuckDBFailure from appender bind, got {other:?}"),
559        }
560        assert!(app.app.is_null());
561        Ok(())
562    }
563
564    #[test]
565    fn test_append_rows() -> Result<()> {
566        let db = Connection::open_in_memory()?;
567        db.execute_batch("CREATE TABLE foo(x INTEGER, y INTEGER)")?;
568
569        {
570            let mut app = db.appender("foo")?;
571            app.append_rows([[1, 2], [3, 4], [5, 6], [7, 8], [9, 10]])?;
572        }
573
574        let val = db.query_row("SELECT sum(x), sum(y) FROM foo", [], |row| <(i32, i32)>::try_from(row))?;
575        assert_eq!(val, (25, 30));
576        Ok(())
577    }
578
579    #[test]
580    fn test_append_u128_as_uhugeint() -> Result<()> {
581        let db = Connection::open_in_memory()?;
582        db.execute("CREATE TABLE t(v UHUGEINT)", [])?;
583
584        let mut appender = db.appender("t")?;
585        appender.append_row([0u128])?;
586        appender.append_row([u128::MAX])?;
587        drop(appender);
588
589        let values: Vec<u128> = db
590            .prepare("SELECT v FROM t ORDER BY v")?
591            .query_map([], |row| row.get(0))?
592            .collect::<Result<Vec<_>>>()?;
593
594        assert_eq!(values, vec![0, u128::MAX]);
595
596        Ok(())
597    }
598
599    #[cfg(feature = "uuid")]
600    #[test]
601    fn test_append_uuid() -> Result<()> {
602        use uuid::Uuid;
603
604        let db = Connection::open_in_memory()?;
605        db.execute_batch("CREATE TABLE foo(x UUID)")?;
606
607        let id = Uuid::new_v4();
608        {
609            let mut app = db.appender("foo")?;
610            app.append_row([id])?;
611        }
612
613        let val = db.query_row("SELECT x FROM foo", [], |row| <(Uuid,)>::try_from(row))?;
614        assert_eq!(val, (id,));
615        Ok(())
616    }
617
618    #[test]
619    fn test_append_string_as_ts_row() -> Result<()> {
620        let db = Connection::open_in_memory()?;
621        db.execute_batch("CREATE TABLE foo(x TIMESTAMP)")?;
622
623        {
624            let mut app = db.appender("foo")?;
625            app.append_row(["2022-04-09 15:56:37.544"])?;
626        }
627
628        let val = db.query_row("SELECT x FROM foo", [], |row| <(i64,)>::try_from(row))?;
629        assert_eq!(val, (1649519797544000,));
630        Ok(())
631    }
632
633    #[test]
634    fn test_append_timestamp() -> Result<()> {
635        use std::time::Duration;
636        let db = Connection::open_in_memory()?;
637        db.execute_batch("CREATE TABLE foo(x TIMESTAMP)")?;
638
639        let d = Duration::from_secs(1);
640        {
641            let mut app = db.appender("foo")?;
642            app.append_row([d])?;
643        }
644
645        let val = db.query_row("SELECT x FROM foo where x=?", [d], |row| <(i32,)>::try_from(row))?;
646        assert_eq!(val, (d.as_micros() as i32,));
647        Ok(())
648    }
649
650    #[test]
651    #[cfg(feature = "chrono")]
652    fn test_append_datetime() -> Result<()> {
653        use chrono::{NaiveDate, NaiveDateTime};
654
655        let db = Connection::open_in_memory()?;
656        db.execute_batch("CREATE TABLE foo(x DATE, y TIMESTAMP)")?;
657
658        let date = NaiveDate::from_ymd_opt(2024, 6, 5).unwrap();
659        let timestamp = date.and_hms_opt(18, 26, 53).unwrap();
660        {
661            let mut app = db.appender("foo")?;
662            app.append_row(params![date, timestamp])?;
663        }
664        let (date2, timestamp2) = db.query_row("SELECT x, y FROM foo", [], |row| {
665            Ok((row.get::<_, NaiveDate>(0)?, row.get::<_, NaiveDateTime>(1)?))
666        })?;
667        assert_eq!(date, date2);
668        assert_eq!(timestamp, timestamp2);
669        Ok(())
670    }
671
672    #[test]
673    #[cfg(feature = "chrono")]
674    fn test_append_struct_with_params() -> Result<()> {
675        use chrono::NaiveDate;
676
677        struct Person {
678            first_name: String,
679            last_name: String,
680            dob: NaiveDate,
681        }
682
683        let db = Connection::open_in_memory()?;
684
685        db.execute_batch("CREATE TABLE foo(first_name VARCHAR, last_name VARCHAR, dob DATE);")?;
686
687        let person1 = Person {
688            first_name: String::from("John"),
689            last_name: String::from("Smith"),
690            dob: NaiveDate::from_ymd_opt(1970, 1, 1).unwrap(),
691        };
692
693        let person2 = Person {
694            first_name: String::from("Jane"),
695            last_name: String::from("Smith"),
696            dob: NaiveDate::from_ymd_opt(1975, 1, 1).unwrap(),
697        };
698
699        // Use params! to extract struct fields
700        {
701            let persons = vec![&person1, &person2];
702            let mut app = db.appender("foo")?;
703            for p in &persons {
704                app.append_row(params![&p.first_name, &p.last_name, p.dob])?;
705            }
706        }
707
708        let count: i64 = db.query_row("SELECT count(*) FROM foo", [], |row| row.get(0))?;
709        assert_eq!(count, 2);
710
711        Ok(())
712    }
713
714    #[test]
715    fn test_appender_error() -> Result<()> {
716        let conn = Connection::open_in_memory()?;
717        conn.execute(
718            r"CREATE TABLE foo (
719            foobar TEXT,
720            foobar_int INT,
721            foobar_split TEXT[] AS (split(trim(foobar), ','))
722            );",
723            [],
724        )?;
725        let mut appender = conn.appender("foo")?;
726        match appender.append_row(params!["foo"]) {
727            Err(Error::DuckDBFailure(.., Some(msg))) => {
728                assert_eq!(msg, "Call to EndRow before all columns have been appended to!")
729            }
730            Err(err) => panic!("unexpected error: {err:?}"),
731            Ok(_) => panic!("expected an error but got Ok"),
732        }
733        Ok(())
734    }
735
736    #[test]
737    fn test_appender_foreign_key_constraint() -> Result<()> {
738        let conn = Connection::open_in_memory()?;
739        conn.execute_batch(
740            r"
741            CREATE TABLE parent (id INTEGER PRIMARY KEY);
742            CREATE TABLE child (
743                id INTEGER,
744                parent_id INTEGER,
745                FOREIGN KEY (parent_id) REFERENCES parent(id)
746            );",
747        )?;
748        conn.execute("INSERT INTO parent VALUES (1)", [])?;
749
750        let mut appender = conn.appender("child")?;
751        appender.append_row(params![1, 999])?; // Invalid parent_id
752
753        // Foreign key constraint should be checked during flush
754        match appender.flush() {
755            Err(Error::DuckDBFailure(_, Some(msg))) => {
756                assert_eq!(
757                    msg,
758                    "Failed to append: Violates foreign key constraint because key \"id: 999\" does not exist in the referenced table"
759                );
760            }
761            Err(e) => panic!("Expected foreign key constraint error, got: {e:?}"),
762            Ok(_) => panic!("Expected foreign key constraint error, but flush succeeded"),
763        }
764
765        Ok(())
766    }
767
768    #[test]
769    fn test_appender_defaults_and_column_switching() -> Result<()> {
770        let db = Connection::open_in_memory()?;
771        db.execute_batch("CREATE TABLE foo(a INT DEFAULT 99, b INT, c INT DEFAULT 7)")?;
772
773        // Only provide column b; a and c should use their defaults
774        {
775            let mut app = db.appender_with_columns("foo", &["b"])?;
776            app.append_row([Some(1)])?;
777            app.append_row([Option::<i32>::None])?;
778        }
779
780        // Switch to a different active column set, then back to full width
781        {
782            let mut app = db.appender("foo")?;
783            app.add_column("c")?;
784            app.add_column("a")?;
785            app.append_row([10, 1])?; // set c and a; b gets NULL
786
787            app.clear_columns()?; // revert to all columns
788            app.append_row([2, 3, 4])?;
789        }
790
791        let rows: Vec<(i32, Option<i32>, i32)> = db
792            .prepare("SELECT a, b, c FROM foo ORDER BY a, b NULLS LAST")?
793            .query_map([], |row| Ok((row.get(0)?, row.get(1)?, row.get(2)?)))?
794            .collect::<Result<_>>()?;
795
796        assert_eq!(
797            rows,
798            vec![
799                (1, None, 10),    // add_column path; b NULL, c set
800                (2, Some(3), 4),  // clear_columns path; all provided
801                (99, Some(1), 7), // defaults applied for a and c
802                (99, None, 7)     // default + NULL
803            ]
804        );
805
806        Ok(())
807    }
808
809    #[test]
810    fn test_appender_with_columns_sequence_default() -> Result<()> {
811        let db = Connection::open_in_memory()?;
812        db.execute_batch(
813            "CREATE SEQUENCE seq START 1;
814             CREATE TABLE foo(id INTEGER DEFAULT nextval('seq'), name TEXT)",
815        )?;
816
817        {
818            let mut app = db.appender_with_columns("foo", &["name"])?;
819            app.append_row(["Alice"])?;
820            app.append_row(["Bob"])?;
821            app.append_row(["Charlie"])?;
822        }
823
824        let rows: Vec<(i32, String)> = db
825            .prepare("SELECT id, name FROM foo ORDER BY id")?
826            .query_map([], |row| Ok((row.get(0)?, row.get(1)?)))?
827            .collect::<Result<_>>()?;
828
829        assert_eq!(
830            rows,
831            vec![(1, "Alice".into()), (2, "Bob".into()), (3, "Charlie".into())]
832        );
833
834        Ok(())
835    }
836
837    #[test]
838    fn test_appender_with_columns_to_db_schema() -> Result<()> {
839        let db = Connection::open_in_memory()?;
840        db.execute_batch(
841            "CREATE SCHEMA s;
842             CREATE TABLE s.foo(a INTEGER DEFAULT 5, b INTEGER)",
843        )?;
844
845        {
846            let mut app = db.appender_with_columns_to_db("foo", "s", &["b"])?;
847            app.append_row([7])?;
848        }
849
850        let (a, b): (i32, i32) = db.query_row("SELECT a, b FROM s.foo", [], |row| Ok((row.get(0)?, row.get(1)?)))?;
851        assert_eq!((a, b), (5, 7));
852        Ok(())
853    }
854
855    #[test]
856    fn test_appender_with_columns_to_catalog_and_db() -> Result<()> {
857        let db = Connection::open_in_memory()?;
858        db.execute_batch(
859            "CREATE SCHEMA s;
860             CREATE TABLE s.bar(a INTEGER DEFAULT 11, b INTEGER)",
861        )?;
862
863        {
864            // Default in-memory catalog is "memory"
865            let mut app = db.appender_with_columns_to_catalog_and_db("bar", "memory", "s", &["b"])?;
866            app.append_row([9])?;
867        }
868
869        let (a, b): (i32, i32) = db.query_row("SELECT a, b FROM s.bar", [], |row| Ok((row.get(0)?, row.get(1)?)))?;
870        assert_eq!((a, b), (11, 9));
871        Ok(())
872    }
873
874    #[test]
875    fn test_appender_decimal() -> Result<()> {
876        let d1 = Decimal::new(20, 4, 11344)?;
877        let d2 = Decimal::new(20, 3, 12312)?;
878        let d3 = Decimal::new(20, 5, -98765)?;
879
880        let conn = Connection::open_in_memory()?;
881        conn.execute_batch("CREATE TABLE decimals (value DECIMAL(20, 10));")?;
882
883        let mut appender = conn.appender("decimals")?;
884        appender.append_row(params![d1])?;
885        appender.append_row(params![d2])?;
886        appender.append_row(params![d3])?;
887        appender.flush()?;
888
889        let results: Vec<Decimal> = conn
890            .prepare("SELECT value FROM decimals ORDER BY value ASC")?
891            .query_map([], |row| row.get(0))?
892            .collect::<Result<Vec<Decimal>>>()?;
893
894        assert_eq!(
895            results,
896            vec![
897                Decimal::new(20, 10, -9876500000)?,
898                Decimal::new(20, 10, 11344000000)?,
899                Decimal::new(20, 10, 123120000000)?,
900            ]
901        );
902
903        Ok(())
904    }
905
906    #[test]
907    fn test_appender_decimal_hugeint_upper_bits() -> Result<()> {
908        let negative = Decimal::new(28, 10, -7922816251426433759354395033_i128)?;
909        let positive = Decimal::new(28, 10, 7922816251426433759354395033_i128)?;
910
911        let conn = Connection::open_in_memory()?;
912        conn.execute_batch("CREATE TABLE decimals (value DECIMAL(28, 10));")?;
913
914        let mut appender = conn.appender("decimals")?;
915        appender.append_row(params![negative])?;
916        appender.append_row(params![positive])?;
917        appender.flush()?;
918
919        let results: Vec<Decimal> = conn
920            .prepare("SELECT value FROM decimals ORDER BY value ASC")?
921            .query_map([], |row| row.get(0))?
922            .collect::<Result<Vec<Decimal>>>()?;
923
924        assert_eq!(results, vec![negative, positive]);
925        Ok(())
926    }
927
928    #[test]
929    fn test_appender_decimal_boundary_values() -> Result<()> {
930        let conn = Connection::open_in_memory()?;
931        conn.execute_batch("CREATE TABLE decimals (value DECIMAL(29, 0));")?;
932
933        let zero = Decimal::new(29, 0, 0)?;
934        let max = Decimal::new(29, 0, 79_228_162_514_264_337_593_543_950_335)?;
935
936        let mut appender = conn.appender("decimals")?;
937        appender.append_row(params![zero])?;
938        appender.append_row(params![max])?;
939        appender.flush()?;
940
941        let results: Vec<Decimal> = conn
942            .prepare("SELECT value FROM decimals ORDER BY value ASC")?
943            .query_map([], |row| row.get(0))?
944            .collect::<Result<Vec<Decimal>>>()?;
945
946        assert_eq!(results, vec![zero, max]);
947        Ok(())
948    }
949
950    #[test]
951    fn test_appender_native_decimal() -> Result<()> {
952        let conn = Connection::open_in_memory()?;
953        conn.execute_batch("CREATE TABLE decimals (value DECIMAL(38, 2));")?;
954
955        let direct = Decimal::new(38, 2, 12_345_678_901_234_567_890_123_456_789_012)?;
956        let owned = Decimal::new(38, 2, -12_345_678_901_234_567_890_123_456_789_012)?;
957
958        let mut appender = conn.appender("decimals")?;
959        appender.append_row(params![direct])?;
960        appender.append_row(params![Value::Decimal(owned)])?;
961        appender.flush()?;
962
963        let results: Vec<Decimal> = conn
964            .prepare("SELECT value FROM decimals ORDER BY value ASC")?
965            .query_map([], |row| row.get(0))?
966            .collect::<Result<Vec<Decimal>>>()?;
967
968        assert_eq!(results, vec![owned, direct]);
969        Ok(())
970    }
971}
duckdb/appender/mod.rs

duckdb/appender/
mod.rs
