+ Send + 'static,
+{
+ tokio::task::spawn_blocking(work)
+ .await
+ .map_err(|error| DriverError::Protocol(format!("DuckDB work stopped: {error}")))?
+}
+
+/// Hand a run's events to `sink` while `produce` makes them on the
+/// connection the call holds. A sink that goes stops the run.
+pub(crate) async fn stream_into(
+ call: &mut DuckCall,
+ request: Option<&StopRequest>,
+ sink: &mut dyn ResultSink,
+ produce: P,
+) -> Result
+where
+ P: FnOnce(&Link, &mut Events) -> Result + Send + 'static,
+{
+ let (sender, mut receiver) = tokio::sync::mpsc::channel(2);
+ let running = call.run(request, move |link| produce(link, &mut Events::new(sender)));
+ let forwarding = async move {
+ while let Some(event) = receiver.recv().await {
+ if sink.accept(event).await == SinkFlow::ConsumerGone {
+ return true;
+ }
+ }
+ false
+ };
+ let (outcome, gone) = tokio::join!(running, forwarding);
+ let mut outcome = outcome?;
+ outcome.consumer_gone |= gone;
+ Ok(outcome)
+}
+
+#[async_trait]
+impl Connection for DuckDbConnection {
+ fn server_info(&self) -> &ServerInfo {
+ &self.server_info
+ }
+
+ fn dialect(&self) -> Arc {
+ self.dialect.clone()
+ }
+
+ fn read_only(&self) -> bool {
+ self.settings.read_only
+ }
+
+ fn link_state(&self) -> LinkState {
+ self.link.state()
+ }
+
+ fn lost(&self) -> WaitForCancellationFutureOwned {
+ self.link.lost()
+ }
+
+ /// With every slot busy, the probe asks on a connection of its own.
+ async fn probe(&self, options: CallOptions) -> Result<(), DriverError> {
+ self.ensure_open()?;
+ let call = match self.pool.timeout_get(&probe_timeouts(&self.liveness)).await {
+ Ok(object) => DuckCall::pooled(object),
+ Err(_) => self.own_call().await?,
+ };
+ let stop = Arc::new(DuckStop::new());
+ stop.arm(call.stop_handle());
+ let probed = self
+ .calls
+ .run_read(options, call, stop, |mut call, request| async move {
+ let result = call.run(Some(&request), |link| batch(link, "SELECT 1")).await;
+ (call, result)
+ })
+ .await;
+ let probed = match probed {
+ Err(DriverError::Timeout { .. }) => Err(DriverError::Timeout {
+ phase: TimeoutPhase::Probe,
+ server_cancelled: false,
+ }),
+ other => other,
+ };
+ self.watch(probed).await
+ }
+
+ async fn list_schemas(&self, options: CallOptions) -> Result, DriverError> {
+ self.read(options, catalog::list_schemas).await
+ }
+
+ async fn list_objects(&self, scope: &ObjectScope, options: CallOptions) -> Result, DriverError> {
+ let scope = scope.clone();
+ self.read(options, move |link, catalog| {
+ catalog::list_objects(link, catalog, &scope)
+ })
+ .await
+ }
+
+ async fn fetch_columns(&self, table: &TableRef, options: CallOptions) -> Result, DriverError> {
+ let table = table.clone();
+ self.read(options, move |link, catalog| {
+ catalog::fetch_columns(link, catalog, &table)
+ })
+ .await
+ }
+
+ async fn fetch_indexes(&self, table: &TableRef, options: CallOptions) -> Result, DriverError> {
+ let table = table.clone();
+ self.read(options, move |link, catalog| {
+ catalog::fetch_indexes(link, catalog, &table)
+ })
+ .await
+ }
+
+ async fn fetch_foreign_keys(
+ &self,
+ table: &TableRef,
+ options: CallOptions,
+ ) -> Result, DriverError> {
+ let table = table.clone();
+ self.read(options, move |link, catalog| {
+ catalog::fetch_foreign_keys(link, catalog, &table)
+ })
+ .await
+ }
+
+ async fn fetch_outbound_foreign_keys(
+ &self,
+ table: &TableRef,
+ options: CallOptions,
+ ) -> Result, DriverError> {
+ let table = table.clone();
+ self.read(options, move |link, catalog| {
+ catalog::outbound_foreign_keys(link, catalog, &table)
+ })
+ .await
+ }
+
+ async fn fetch_inbound_foreign_keys(
+ &self,
+ table: &TableRef,
+ options: CallOptions,
+ ) -> Result, DriverError> {
+ let table = table.clone();
+ self.read(options, move |link, catalog| {
+ catalog::inbound_foreign_keys(link, catalog, &table)
+ })
+ .await
+ }
+
+ async fn fetch_table_definition(
+ &self,
+ table: &TableRef,
+ options: CallOptions,
+ ) -> Result {
+ let table = table.clone();
+ // One transaction, so the parts come from the same moment.
+ self.read(options, move |link, catalog| {
+ batch(link, "BEGIN TRANSACTION READ ONLY")?;
+ let definition = catalog::fetch_table_definition(link, catalog, &table);
+ link.hold_stops();
+ batch(link, "ROLLBACK")?;
+ definition
+ })
+ .await
+ }
+
+ async fn row_identity(&self, table: &TableRef, options: CallOptions) -> Result {
+ let table = table.clone();
+ self.read(options, move |link, catalog| {
+ catalog::row_identity(link, catalog, &table)
+ })
+ .await
+ }
+
+ async fn table_estimate(&self, table: &TableRef, options: CallOptions) -> Result {
+ let table = table.clone();
+ self.read(options, move |link, catalog| {
+ catalog::table_estimate(link, catalog, &table)
+ })
+ .await
+ }
+
+ async fn query_bound(
+ &self,
+ statement: &Statement,
+ limit: RowLimit,
+ options: CallOptions,
+ ) -> Result {
+ let statement = statement.clone();
+ self.read(options, move |link, _| collect(link, &statement, Some(limit)))
+ .await
+ }
+
+ async fn count_rows(&self, statement: &Statement, options: CallOptions) -> Result {
+ let statement = statement.clone();
+ self.read(options, move |link, _| count(link, &statement)).await
+ }
+
+ /// Every row, on a connection of the read's own, inside a read-only
+ /// transaction, so user SQL that writes is refused by DuckDB itself.
+ async fn stream_read(
+ &self,
+ read: &ReadStatement,
+ sink: Box,
+ options: CallOptions,
+ ) -> Result {
+ self.ensure_open()?;
+ if let ReadStatement::User(sql) = read
+ && controls_transaction(sql.as_str())
+ {
+ return Err(DriverError::ReadOnly(ReadOnlyRefusal::ClientGuard));
+ }
+ let call = self.own_call().await?;
+ let stop = Arc::new(DuckStop::new());
+ stop.arm(call.stop_handle());
+ let read = read.clone();
+ let streamed = self
+ .calls
+ .run_read(options, call, stop, move |mut call, request| async move {
+ let mut sink = sink;
+ let result = stream_into(&mut call, Some(&request), sink.as_mut(), move |link, events| {
+ batch(link, "BEGIN TRANSACTION READ ONLY")?;
+ let streamed = match &read {
+ ReadStatement::Generated(statement) => run_bound(link, statement, events),
+ ReadStatement::User(sql) => run_script(link, sql.as_str(), None, events, &|| false),
+ };
+ link.hold_stops();
+ let _ = batch(link, "ROLLBACK");
+ streamed
+ })
+ .await;
+ (call, result)
+ })
+ .await;
+ let streamed = self.watch(streamed).await?;
+ // The sink stopped reading, so the read did not run to its end.
+ if streamed.consumer_gone {
+ return Err(DriverError::Cancelled);
+ }
+ Ok(StreamSummary::new(streamed.rows))
+ }
+
+ async fn write_batch(&self, batch: &WriteBatch, options: CallOptions) -> Result {
+ if self.settings.read_only {
+ return Err(DriverError::ReadOnly(ReadOnlyRefusal::ClientGuard));
+ }
+ if batch.is_empty() {
+ return Ok(WriteReport::default());
+ }
+ batch.foreign_keys.require_enforced()?;
+ self.ensure_open()?;
+ // A schema change runs as long as it takes: one stopped halfway
+ // by the statement timeout would only roll back.
+ let options = match batch.kind {
+ WriteKind::RowChanges => self.statement_bounded(options),
+ WriteKind::SchemaChanges => options,
+ };
+ let ledger = Arc::new(WriteLedger::new(WriteAtomicity::Transactional));
+ let recorded = ledger.clone();
+ let stop = Arc::new(DuckStop::new());
+ let armed = stop.clone();
+ let pool = self.pool.clone();
+ let timeouts = pool.timeouts();
+ let batch = batch.clone();
+ let written = self
+ .calls
+ .run_write(
+ options,
+ DuckCall::new(),
+ stop,
+ ledger,
+ move |mut call, request| async move {
+ if let Err(error) = call.acquire(&pool, &timeouts, &request).await {
+ return (call, Err(error));
+ }
+ armed.arm(call.stop_handle());
+ let stopping = request.clone();
+ let outcome = call
+ .run(Some(&request), move |link| {
+ Ok(write_batch::run(link, &batch, &recorded, &stopping))
+ })
+ .await;
+ let result = match outcome {
+ Ok((result, reusable)) => {
+ if !reusable {
+ call.taint();
+ }
+ result
+ }
+ Err(error) => Err(error),
+ };
+ (call, result)
+ },
+ )
+ .await;
+ self.watch(written).await
+ }
+
+ /// DuckDB commits on the local file, so a COMMIT never goes out over
+ /// a link that can fail, and there is no token to ask about.
+ async fn resolve_commit(
+ &self,
+ _token: &CommitToken,
+ _options: CallOptions,
+ ) -> Result {
+ self.ensure_open()?;
+ Ok(CommitResolution::Unresolvable)
+ }
+
+ async fn open_editor_session(
+ &self,
+ options: EditorSessionOptions,
+ cancel: CancellationToken,
+ ) -> Result, DriverError> {
+ let session = DuckDbEditorSession::open(self.instance()?, options, &self.liveness, cancel).await?;
+ Ok(Arc::new(session))
+ }
+
+ async fn close(&self) -> Result<(), DriverError> {
+ self.closed.store(true, Ordering::SeqCst);
+ self.calls.shutdown().await;
+ self.pool.close();
+ let instance = self.instance.lock().unwrap_or_else(PoisonError::into_inner).take();
+ // DuckDB closes the file as the last connection to it goes, which
+ // checkpoints it: not on an async thread.
+ if let Some(instance) = instance {
+ let _ = blocking(move || {
+ drop(instance);
+ Ok(())
+ })
+ .await;
+ }
+ Ok(())
+ }
+}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+
+ /// A call given up after a stop that did not settle synthesizes a
+ /// lost connection, which leaves a healthy database's link live.
+ #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+ async fn a_lost_call_on_a_healthy_database_keeps_the_link() {
+ let directory = tempfile::tempdir().unwrap();
+ let endpoint = FileEndpoint::create_new(directory.path().join("live.duckdb")).unwrap();
+ let liveness = tablepro_test_support::conformance::test_liveness();
+ let connection = DuckDbConnection::open(
+ &endpoint,
+ WorkspaceSettings::workspace(false),
+ liveness,
+ CancellationToken::new(),
+ )
+ .await
+ .unwrap();
+
+ let lost: Result<(), DriverError> = Err(DriverError::ConnectionLost {
+ during: LossPhase::Statement,
+ });
+ let _ = connection.watch(lost).await;
+
+ assert_eq!(connection.link_state(), LinkState::Live);
+ connection.close().await.unwrap();
+ }
+
+ #[test]
+ fn the_version_reads_as_its_parts() {
+ let info = server_info("v1.5.6");
+
+ assert_eq!(info, ServerInfo::new("DuckDB", "v1.5.6", 1, 5, 6));
+ }
+}
diff --git a/linux/crates/drivers/duckdb/src/create_file.rs b/linux/crates/drivers/duckdb/src/create_file.rs
new file mode 100644
index 0000000000..11faaae0dc
--- /dev/null
+++ b/linux/crates/drivers/duckdb/src/create_file.rs
@@ -0,0 +1,254 @@
+use std::ffi::OsStr;
+use std::fs::{File, OpenOptions};
+use std::io::ErrorKind;
+use std::os::unix::fs::PermissionsExt;
+use std::path::{Path, PathBuf};
+
+use rustix::fs::FlockOperation;
+use rustix::io::Errno;
+use tablepro_core::{BusyReason, DriverError};
+use tokio_util::sync::CancellationToken;
+
+use crate::errors::map_error;
+use crate::instance::file_error;
+
+/// The write-ahead log DuckDB keeps beside a database. One left from the
+/// database being replaced would be replayed into the new one.
+const WAL_SUFFIX: &str = ".wal";
+
+/// Put an empty database at `path`, replacing whatever is there.
+///
+/// The save dialog has already asked before replacing a file. An empty
+/// file is no DuckDB database, so the new one is written beside the
+/// target first, then renamed over it, which is atomic: a reader sees the
+/// old file or the new one, never half of one.
+///
+/// `given_up` is the connect's stop, read again once the new database is
+/// built: building it takes a while, and a connect given up meanwhile
+/// leaves the old file where it is.
+pub(crate) fn create_or_replace(path: &Path, given_up: &CancellationToken) -> Result<(), DriverError> {
+ let parent = match path.parent() {
+ Some(parent) if !parent.as_os_str().is_empty() => parent,
+ _ => {
+ return Err(DriverError::FileNotFound {
+ path: path.to_path_buf(),
+ });
+ }
+ };
+ let Some(name) = path.file_name() else {
+ return Err(DriverError::NotADatabase {
+ path: path.to_path_buf(),
+ });
+ };
+ if !parent.is_dir() {
+ return Err(DriverError::FileNotFound {
+ path: path.to_path_buf(),
+ });
+ }
+ let temporary = beside(parent, name, "tmp");
+ let placed = build_empty(&temporary, path).and_then(|()| match given_up.is_cancelled() {
+ true => Err(DriverError::Cancelled),
+ false => place(&temporary, path, parent, name),
+ });
+ if placed.is_err() {
+ // Nothing half-built is left beside the user's files.
+ let _ = std::fs::remove_file(&temporary);
+ let _ = std::fs::remove_file(wal_of(&temporary));
+ return placed;
+ }
+ // The rename is durable only once the directory that records it is.
+ File::open(parent)
+ .and_then(|directory| directory.sync_all())
+ .map_err(|error| file_error(error.kind(), path))
+}
+
+/// A hidden name beside the user's file, unique to this call.
+fn beside(parent: &Path, name: &OsStr, suffix: &str) -> PathBuf {
+ parent.join(format!(
+ ".{}.{}.{suffix}",
+ name.to_string_lossy(),
+ uuid::Uuid::new_v4().simple()
+ ))
+}
+
+/// An empty database, checkpointed into the file so no log goes with
+/// it, readable by its owner only.
+fn build_empty(temporary: &Path, target: &Path) -> Result<(), DriverError> {
+ let connection = duckdb::Connection::open(temporary).map_err(map_error)?;
+ connection.execute_batch("CHECKPOINT").map_err(map_error)?;
+ connection.close().map_err(|(_, error)| map_error(error))?;
+ std::fs::set_permissions(temporary, std::fs::Permissions::from_mode(0o600))
+ .and_then(|()| File::open(temporary)?.sync_all())
+ .map_err(|error| file_error(error.kind(), target))
+}
+
+/// Rename the new database over `path`. The old file's log is moved
+/// aside first and put back if the rename fails, so a database that
+/// stays keeps the changes only its log holds.
+fn place(temporary: &Path, path: &Path, parent: &Path, name: &OsStr) -> Result<(), DriverError> {
+ let _held = lock_replaced(path)?;
+ let wal = wal_of(path);
+ let aside = beside(parent, name, "wal.tmp");
+ let moved = match std::fs::rename(&wal, &aside) {
+ Ok(()) => true,
+ Err(error) if error.kind() == ErrorKind::NotFound => false,
+ Err(error) => return Err(file_error(error.kind(), path)),
+ };
+ if let Err(error) = std::fs::rename(temporary, path) {
+ if moved && let Err(restoring) = std::fs::rename(&aside, &wal) {
+ tracing::warn!(error = %restoring, aside = %aside.display(), "could not put a database's log back");
+ }
+ return Err(file_error(error.kind(), path));
+ }
+ if moved && let Err(error) = std::fs::remove_file(&aside) {
+ tracing::warn!(%error, aside = %aside.display(), "could not remove a replaced database's log");
+ }
+ Ok(())
+}
+
+/// Lock the file being replaced as DuckDB locks a database it opens, and
+/// hold the lock through the rename. Another process with the database
+/// open holds that lock, and would go on writing a log beside the new
+/// file under the same name.
+fn lock_replaced(path: &Path) -> Result, DriverError> {
+ match std::fs::metadata(path) {
+ Ok(metadata) if metadata.is_file() => {}
+ // Nothing there to lock, or something the rename itself refuses.
+ Ok(_) => return Ok(None),
+ Err(error) if error.kind() == ErrorKind::NotFound => return Ok(None),
+ Err(error) => return Err(file_error(error.kind(), path)),
+ }
+ // A POSIX lock belongs to the process, and closing any descriptor on
+ // the file drops all of them, so a file this process holds open is
+ // not opened again here.
+ if tablepro_core::held_by_this_process(path) {
+ return Err(DriverError::Busy(BusyReason::FileLocked));
+ }
+ let (file, lock) = match OpenOptions::new().read(true).write(true).open(path) {
+ Ok(file) => (file, FlockOperation::NonBlockingLockExclusive),
+ // A write lock needs the file open for writing. A shared lock
+ // still meets the writer's.
+ Err(error) if error.kind() == ErrorKind::PermissionDenied || error.kind() == ErrorKind::ReadOnlyFilesystem => (
+ File::open(path).map_err(|error| file_error(error.kind(), path))?,
+ FlockOperation::NonBlockingLockShared,
+ ),
+ Err(error) => return Err(file_error(error.kind(), path)),
+ };
+ match rustix::fs::fcntl_lock(&file, lock) {
+ Ok(()) => Ok(Some(file)),
+ Err(errno) if errno == Errno::AGAIN || errno == Errno::ACCESS => Err(DriverError::Busy(BusyReason::FileLocked)),
+ // A file system without locks: DuckDB could not have locked it
+ // either.
+ Err(_) => Ok(Some(file)),
+ }
+}
+
+fn wal_of(path: &Path) -> PathBuf {
+ let mut wal = path.as_os_str().to_owned();
+ wal.push(WAL_SUFFIX);
+ PathBuf::from(wal)
+}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+
+ #[test]
+ fn an_existing_file_and_its_log_are_replaced_by_an_empty_private_database() {
+ let directory = tempfile::tempdir().unwrap();
+ let path = directory.path().join("data.duckdb");
+ std::fs::write(&path, b"old contents").unwrap();
+ std::fs::write(wal_of(&path), b"old log").unwrap();
+
+ create_or_replace(&path, &CancellationToken::new()).unwrap();
+
+ let metadata = std::fs::metadata(&path).unwrap();
+ assert_eq!(metadata.permissions().mode() & 0o777, 0o600);
+ assert!(
+ !wal_of(&path).exists(),
+ "the old log would be replayed into the new database"
+ );
+ let leftovers: Vec<_> = std::fs::read_dir(directory.path())
+ .unwrap()
+ .map(|entry| entry.unwrap().file_name())
+ .collect();
+ assert_eq!(leftovers, [std::ffi::OsString::from("data.duckdb")]);
+ let reopened = duckdb::Connection::open(&path).unwrap();
+ let tables: i64 = reopened
+ .query_row("SELECT count(*) FROM duckdb_tables()", [], |row| row.get(0))
+ .unwrap();
+ assert_eq!(tables, 0);
+ }
+
+ #[test]
+ fn the_log_of_a_database_that_stays_is_kept() {
+ let directory = tempfile::tempdir().unwrap();
+ // A directory cannot be renamed over, so the database stays.
+ let path = directory.path().join("data.duckdb");
+ std::fs::create_dir(&path).unwrap();
+ std::fs::write(path.join("inside"), b"keeps the directory from being replaced").unwrap();
+ std::fs::write(wal_of(&path), b"committed changes").unwrap();
+
+ assert!(create_or_replace(&path, &CancellationToken::new()).is_err());
+
+ assert_eq!(std::fs::read(wal_of(&path)).unwrap(), b"committed changes");
+ let mut leftovers: Vec<_> = std::fs::read_dir(directory.path())
+ .unwrap()
+ .map(|entry| entry.unwrap().file_name())
+ .collect();
+ leftovers.sort();
+ assert_eq!(
+ leftovers,
+ [
+ std::ffi::OsString::from("data.duckdb"),
+ std::ffi::OsString::from("data.duckdb.wal")
+ ]
+ );
+ }
+
+ #[test]
+ fn a_file_this_process_holds_is_not_replaced() {
+ let directory = tempfile::tempdir().unwrap();
+ let path = directory.path().join("held.duckdb");
+ std::fs::write(&path, b"open elsewhere in this process").unwrap();
+ let open = File::open(&path).unwrap();
+
+ assert_eq!(
+ create_or_replace(&path, &CancellationToken::new()),
+ Err(DriverError::Busy(BusyReason::FileLocked))
+ );
+ assert_eq!(std::fs::read(&path).unwrap(), b"open elsewhere in this process");
+ drop(open);
+ }
+
+ /// A connect given up while the new database was built leaves the
+ /// old file, and nothing half-built beside it.
+ #[test]
+ fn a_create_given_up_while_building_keeps_the_old_file() {
+ let directory = tempfile::tempdir().unwrap();
+ let path = directory.path().join("data.duckdb");
+ std::fs::write(&path, b"what the user had").unwrap();
+ let given_up = CancellationToken::new();
+ given_up.cancel();
+
+ assert_eq!(create_or_replace(&path, &given_up), Err(DriverError::Cancelled));
+
+ assert_eq!(std::fs::read(&path).unwrap(), b"what the user had");
+ let left: Vec<_> = std::fs::read_dir(directory.path())
+ .unwrap()
+ .map(|entry| entry.unwrap().file_name())
+ .collect();
+ assert_eq!(left, [std::ffi::OsString::from("data.duckdb")]);
+ }
+
+ #[test]
+ fn a_missing_directory_is_file_not_found() {
+ let directory = tempfile::tempdir().unwrap();
+ let path = directory.path().join("missing").join("data.duckdb");
+
+ assert_eq!(
+ create_or_replace(&path, &CancellationToken::new()),
+ Err(DriverError::FileNotFound { path })
+ );
+ }
+}
diff --git a/linux/crates/drivers/duckdb/src/ddl.rs b/linux/crates/drivers/duckdb/src/ddl.rs
new file mode 100644
index 0000000000..c42691b13b
--- /dev/null
+++ b/linux/crates/drivers/duckdb/src/ddl.rs
@@ -0,0 +1,437 @@
+//! Structure changes on DuckDB.
+//!
+//! DuckDB takes one action per ALTER TABLE, and a table something else
+//! depends on, an index made with CREATE INDEX or another table's
+//! foreign key, takes only ADD COLUMN and a column's default: every
+//! other change refuses with a dependency error. A column refuses a new
+//! type while a key, a CHECK, an index or a generated column uses it.
+//! The plans offer only what DuckDB 1.5 takes, measured against it in
+//! the driver's DDL tests.
+
+use tablepro_core::column::{ColumnKind, ColumnType};
+use tablepro_core::ddl::ddl_sql::{column_list, create_index_sql, default_sql, foreign_key_sql};
+use tablepro_core::ddl::{
+ BuildDdlError, ColumnAttributes, ColumnCapabilities, ColumnDefinition, DdlCapabilities, DdlDialect, DdlOperation,
+ DdlPlan, DraftColumn, Permission, ReferentialAction, RefusalReason, TableChanges, TableDefinition, TableDraft,
+};
+use tablepro_core::dialect::SqlDialect;
+use tablepro_core::meta::{IndexInfo, TableRef};
+use tablepro_core::value::Value;
+
+use crate::dialect::DuckDbDialect;
+
+static CAPABILITIES: DdlCapabilities = DdlCapabilities {
+ create_table: true,
+ rename_table: true,
+ add_column: true,
+ drop_column: true,
+ rename_column: true,
+ alter_column_type: true,
+ alter_column_nullability: true,
+ alter_column_default: true,
+ // A key can be added to a table that has none; DuckDB drops or
+ // moves none.
+ alter_primary_key: true,
+ alter_auto_increment: false,
+ auto_increment_outside_primary_key: false,
+ create_index: true,
+ drop_index: true,
+ expression_index_keys: true,
+ // DuckDB takes DESC in CREATE INDEX and keeps the key ascending.
+ descending_index_keys: false,
+ partial_indexes: false,
+ add_foreign_key: false,
+ drop_foreign_key: false,
+ column_comments: true,
+ // CASCADE drops no table whose key another table references.
+ drop_table_cascade: false,
+ column_defaults: true,
+ required_columns: true,
+ declared_column_order: true,
+};
+
+const TYPE_SUGGESTIONS: &[&str] = &[
+ "VARCHAR",
+ "INTEGER",
+ "BIGINT",
+ "DOUBLE",
+ "DECIMAL(18,3)",
+ "BOOLEAN",
+ "DATE",
+ "TIMESTAMP",
+ "TIMESTAMPTZ",
+ "TIME",
+ "INTERVAL",
+ "UUID",
+ "BLOB",
+ "JSON",
+ "HUGEINT",
+];
+
+/// What holds the table in place: the indexes made on it and the keys
+/// of other tables that reference it.
+fn dependents(table: &TableDefinition) -> Vec {
+ let created_indexes = table.indexes.iter().filter(|index| {
+ let constraint = table
+ .primary_key
+ .as_ref()
+ .is_some_and(|key| key.name.as_deref() == Some(index.name.as_str()))
+ || index.primary;
+ let created = table
+ .source
+ .as_ref()
+ .is_some_and(|source| source.dependents.iter().any(|dependent| dependent.name == index.name));
+ created && !constraint
+ });
+ created_indexes
+ .map(|index| index.name.clone())
+ .chain(table.inbound_foreign_keys.iter().map(|key| key.name.clone()))
+ .collect()
+}
+
+/// What uses one column so that DuckDB will not change its type: a key
+/// or index over it, a foreign key from it, a generated column on it.
+fn type_users(table: &TableDefinition, column: &ColumnDefinition) -> Vec {
+ let name = column.name();
+ let indexes = table
+ .indexes
+ .iter()
+ .filter(|index| index.keys.iter().any(|key| key.column_name() == Some(name)))
+ .map(|index| index.name.clone());
+ let keys = table
+ .foreign_keys
+ .iter()
+ .filter(|key| key.columns.iter().any(|column| column == name))
+ .map(|key| key.name.clone());
+ let generated = table
+ .columns
+ .iter()
+ .filter(|other| {
+ other
+ .attributes
+ .generated
+ .as_ref()
+ .is_some_and(|generated| mentions(generated.expression.as_sql(), name))
+ })
+ .map(|other| other.name().to_owned());
+ indexes.chain(keys).chain(generated).collect()
+}
+
+/// Whether an expression names the column, as a word or quoted.
+fn mentions(expression: &str, column: &str) -> bool {
+ let quoted = format!("\"{}\"", column.replace('"', "\"\""));
+ expression.contains("ed)
+ || expression
+ .split(|c: char| !(c.is_alphanumeric() || c == '_'))
+ .any(|word| word.eq_ignore_ascii_case(column))
+}
+
+impl DdlDialect for DuckDbDialect {
+ fn capabilities(&self) -> &DdlCapabilities {
+ &CAPABILITIES
+ }
+
+ fn sql_dialect(&self) -> &dyn SqlDialect {
+ self
+ }
+
+ /// A table something depends on can be neither renamed nor given a
+ /// key, and a key goes only on a table with none.
+ fn table_capabilities(&self, table: &TableDefinition) -> DdlCapabilities {
+ let free = dependents(table).is_empty();
+ DdlCapabilities {
+ rename_table: free,
+ alter_primary_key: free && table.primary_key.is_none(),
+ ..CAPABILITIES
+ }
+ }
+
+ fn column_capabilities(&self, table: &TableDefinition, column: &ColumnDefinition) -> ColumnCapabilities {
+ let mut permissions = ColumnCapabilities::standard(&self.table_capabilities(table), table, column);
+ let held = dependents(table);
+ if !held.is_empty() {
+ let in_use = || Permission::Refused(RefusalReason::ColumnInUse(held.clone()));
+ permissions.rename = permissions.rename.and(in_use);
+ permissions.drop = permissions.drop.and(in_use);
+ permissions.alter_type = permissions.alter_type.and(in_use);
+ permissions.alter_nullability = permissions.alter_nullability.and(in_use);
+ }
+ let users = type_users(table, column);
+ if !users.is_empty() {
+ let in_use = || Permission::Refused(RefusalReason::ColumnInUse(users.clone()));
+ permissions.alter_type = permissions.alter_type.and(in_use);
+ permissions.drop = permissions.drop.and(in_use);
+ }
+ if let Some(check) = &column.attributes.check {
+ let checked = || Permission::Refused(RefusalReason::ConflictingAttribute("CHECK constraint".to_owned()));
+ permissions.alter_type = permissions.alter_type.and(checked);
+ // DuckDB drops a CHECK on the column alone with it, and
+ // refuses to drop a column a CHECK shares with another.
+ let shared = table
+ .columns
+ .iter()
+ .any(|other| other.name() != column.name() && mentions(check.as_sql(), other.name()));
+ if shared {
+ permissions.drop = permissions.drop.and(checked);
+ }
+ }
+ let referencing: Vec = table
+ .foreign_keys
+ .iter()
+ .filter(|key| key.columns.iter().any(|name| name == column.name()))
+ .map(|key| key.name.clone())
+ .collect();
+ if !referencing.is_empty() {
+ let in_use = || Permission::Refused(RefusalReason::ColumnInUse(referencing.clone()));
+ permissions.rename = permissions.rename.and(in_use);
+ }
+ permissions
+ }
+
+ /// DuckDB drops an index made with CREATE INDEX, and no constraint:
+ /// a UNIQUE key's index goes only with its table.
+ fn index_removal(&self, table: &TableDefinition, index: &IndexInfo) -> Permission {
+ let created = table
+ .source
+ .as_ref()
+ .is_some_and(|source| source.dependents.iter().any(|dependent| dependent.name == index.name));
+ Permission::unless(created, RefusalReason::EngineUnsupported)
+ }
+
+ fn type_suggestions(&self) -> &'static [&'static str] {
+ TYPE_SUGGESTIONS
+ }
+
+ fn default_column_type(&self) -> &'static str {
+ "VARCHAR"
+ }
+
+ fn referential_actions(&self) -> &'static [ReferentialAction] {
+ &[ReferentialAction::NoAction]
+ }
+
+ /// DuckDB keeps a foreign key's table by name alone, in the schema
+ /// of the table that holds the key.
+ fn references_other_schemas(&self) -> bool {
+ false
+ }
+
+ fn plan_create_table(&self, draft: &TableDraft) -> Result {
+ draft.check_create(self)?;
+ let table = TableRef::new(draft.table.schema.clone(), draft.table.name.trim());
+ let mut definitions = draft
+ .columns
+ .iter()
+ .map(|column| self.column_definition(column, true))
+ .collect::, _>>()?;
+ let key: Vec = draft.primary_key().into_iter().map(str::to_owned).collect();
+ if !key.is_empty() {
+ definitions.push(format!("PRIMARY KEY ({})", column_list(self, &key)));
+ }
+ definitions.extend(draft.foreign_keys.iter().map(|key| foreign_key_sql(self, &table, key)));
+ let qualified = self.qualify(&table);
+ let mut statements = vec![format!(
+ "CREATE TABLE {qualified} (\n {}\n)",
+ definitions.join(",\n ")
+ )];
+ for column in &draft.columns {
+ if let Some(comment) = column.comment_text() {
+ statements.push(self.comment_on(&qualified, &column.name, Some(comment)));
+ }
+ }
+ statements.extend(
+ draft
+ .indexes
+ .iter()
+ .filter(|index| !index.primary)
+ .map(|index| create_index_sql(self, &table, index)),
+ );
+ Ok(DdlPlan::from_sql(statements))
+ }
+
+ fn plan_alter_table(&self, original: &TableDefinition, changes: &TableChanges) -> Result {
+ changes.check(self, original)?;
+ let narrowed = self.table_capabilities(original);
+ if changes.renamed_table.is_some() && !narrowed.rename_table {
+ return Err(refused("", DdlOperation::RenameTable, dependents(original)));
+ }
+ let mut statements = Vec::new();
+ if let Some(name) = &changes.renamed_table {
+ statements.push(format!(
+ "ALTER TABLE {} RENAME TO {}",
+ self.qualify(&original.table),
+ self.quote_identifier(name)
+ ));
+ }
+ let table = changes.target(&original.table);
+ let qualified = self.qualify(&table);
+ let alter = |action: String| format!("ALTER TABLE {qualified} {action}");
+ for index in &changes.dropped_indexes {
+ let index = TableRef::new(table.schema.clone(), index.name.clone());
+ statements.push(format!("DROP INDEX {}", self.qualify(&index)));
+ }
+ for column in &changes.dropped_columns {
+ statements.push(alter(format!("DROP COLUMN {}", self.quote_identifier(column.name()))));
+ }
+ for rename in changes.ordered_renames(original) {
+ statements.push(alter(format!(
+ "RENAME COLUMN {} TO {}",
+ self.quote_identifier(&rename.from),
+ self.quote_identifier(&rename.to)
+ )));
+ }
+ for alteration in &changes.altered_columns {
+ let draft = &alteration.draft;
+ let name = self.quote_identifier(&draft.name);
+ let changed = alteration.changed;
+ if changed.column_type {
+ let resolved = draft.column_type.resolve(self, &draft.name)?;
+ // DuckDB drops a column's collation with its old type
+ // unless the new one names it again.
+ let collation = collation_clause(Some(&alteration.original.attributes), &resolved.column_type);
+ statements.push(alter(format!("ALTER COLUMN {name} TYPE {}{collation}", resolved.sql)));
+ }
+ if changed.nullability {
+ let action = if draft.nullable {
+ "DROP NOT NULL"
+ } else {
+ "SET NOT NULL"
+ };
+ statements.push(alter(format!("ALTER COLUMN {name} {action}")));
+ }
+ if changed.default {
+ let column_type = draft.column_type.column_type(self, &draft.name)?;
+ statements.push(alter(
+ match default_sql(self, &draft.name, &draft.default, &column_type)? {
+ Some(default) => format!("ALTER COLUMN {name} SET DEFAULT {default}"),
+ None => format!("ALTER COLUMN {name} DROP DEFAULT"),
+ },
+ ));
+ }
+ if changed.comment {
+ statements.push(self.comment_on(&qualified, &draft.name, draft.comment_text()));
+ }
+ }
+ let held = dependents(original);
+ for column in &changes.added_columns {
+ // DuckDB adds no column with a constraint, so NOT NULL is set
+ // once it is there, which a table with dependents refuses.
+ if !column.nullable && !held.is_empty() {
+ return Err(refused(&column.name, DdlOperation::AddColumn, held));
+ }
+ statements.push(alter(format!("ADD COLUMN {}", self.column_definition(column, false)?)));
+ if !column.nullable {
+ statements.push(alter(format!(
+ "ALTER COLUMN {} SET NOT NULL",
+ self.quote_identifier(&column.name)
+ )));
+ }
+ if let Some(comment) = column.comment_text() {
+ statements.push(self.comment_on(&qualified, &column.name, Some(comment)));
+ }
+ }
+ if let Some(change) = &changes.primary_key {
+ if change.from.is_some() || !narrowed.alter_primary_key {
+ let column = change.to.first().map_or("", String::as_str);
+ return Err(refused(column, DdlOperation::AlterPrimaryKey, held));
+ }
+ if !change.to.is_empty() {
+ statements.push(alter(format!("ADD PRIMARY KEY ({})", column_list(self, &change.to))));
+ }
+ }
+ statements.extend(
+ changes
+ .added_indexes
+ .iter()
+ .map(|index| create_index_sql(self, &table, index)),
+ );
+ DdlPlan::for_changes(changes, statements)
+ }
+
+ /// DuckDB's own CREATE TABLE text, which keeps its CHECK and UNIQUE
+ /// constraints and its generated columns, then the indexes made on
+ /// the table and the column comments, which that text leaves out.
+ fn create_table_script(&self, definition: &TableDefinition) -> Result {
+ let source = definition.source.as_ref().ok_or(BuildDdlError::NoTableSource)?;
+ let mut statements = vec![source.create_statement.trim().trim_end_matches(';').to_owned()];
+ statements.extend(
+ source
+ .dependents
+ .iter()
+ .map(|dependent| dependent.sql.trim().trim_end_matches(';').to_owned()),
+ );
+ for column in &definition.columns {
+ if let Some(comment) = column.info.comment.as_deref().filter(|text| !text.is_empty()) {
+ statements.push(self.comment_on(&self.portable_name(&definition.table), column.name(), Some(comment)));
+ }
+ }
+ Ok(DdlPlan::from_sql(statements))
+ }
+}
+
+/// ` COLLATE ` for a text column the catalogue read with one, as
+/// DuckDB's own text spelled it.
+fn collation_clause(attributes: Option<&ColumnAttributes>, column_type: &ColumnType) -> String {
+ attributes
+ .and_then(|attributes| attributes.collation.as_ref())
+ .filter(|_| matches!(column_type.kind, ColumnKind::Text { .. }))
+ .map(|collation| format!(" COLLATE {}", collation.name))
+ .unwrap_or_default()
+}
+
+fn refused(column: &str, operation: DdlOperation, users: Vec) -> BuildDdlError {
+ BuildDdlError::Refused {
+ column: column.to_owned(),
+ operation,
+ reason: match users.is_empty() {
+ true => RefusalReason::EngineUnsupported,
+ false => RefusalReason::ColumnInUse(users),
+ },
+ }
+}
+
+impl DuckDbDialect {
+ /// A column as CREATE TABLE declares it, or as ADD COLUMN does when
+ /// `constraints` is off. A column read from the catalogue keeps its
+ /// generation expression and its CHECK.
+ fn column_definition(&self, column: &DraftColumn, constraints: bool) -> Result {
+ let resolved = column.column_type.resolve(self, &column.name)?;
+ let attributes = column.original.as_ref().map(|original| &original.attributes);
+ let mut parts = vec![
+ self.quote_identifier(&column.name),
+ format!(
+ "{}{}",
+ resolved.sql,
+ collation_clause(attributes, &resolved.column_type)
+ ),
+ ];
+ if let Some(generated) = attributes.and_then(|attributes| attributes.generated.as_ref()) {
+ parts.push(format!(
+ "GENERATED ALWAYS AS ({}) VIRTUAL",
+ generated.expression.as_sql()
+ ));
+ } else if let Some(default) = default_sql(self, &column.name, &column.default, &resolved.column_type)? {
+ parts.push(format!("DEFAULT {default}"));
+ }
+ if constraints {
+ if !column.nullable {
+ parts.push("NOT NULL".to_owned());
+ }
+ if let Some(check) = attributes.and_then(|attributes| attributes.check.as_ref()) {
+ parts.push(format!("CHECK ({})", check.as_sql()));
+ }
+ }
+ Ok(parts.join(" "))
+ }
+
+ /// `table` as the statement names it: qualified for a change, portable
+ /// in a script the user copies to replay elsewhere.
+ fn comment_on(&self, table: &str, column: &str, comment: Option<&str>) -> String {
+ let target = format!("{table}.{}", self.quote_identifier(column));
+ let text = comment.map(|text| Value::Text(text.to_owned()));
+ match text.and_then(|text| SqlDialect::literal(self, &text, &ColumnType::untyped_text("VARCHAR")).ok()) {
+ Some(literal) => format!("COMMENT ON COLUMN {target} IS {literal}"),
+ None => format!("COMMENT ON COLUMN {target} IS NULL"),
+ }
+ }
+}
diff --git a/linux/crates/drivers/duckdb/src/decode/bits.rs b/linux/crates/drivers/duckdb/src/decode/bits.rs
new file mode 100644
index 0000000000..914ed3813d
--- /dev/null
+++ b/linux/crates/drivers/duckdb/src/decode/bits.rs
@@ -0,0 +1,49 @@
+use tablepro_core::value::BitString;
+
+/// A BIT as DuckDB stores it: a byte that says how many bits of the next
+/// byte are padding, then the bits, first bit highest.
+pub(crate) fn bit_string(stored: &[u8]) -> Option {
+ let (&padding, data) = stored.split_first()?;
+ if padding > 7 || (data.is_empty() && padding != 0) {
+ return None;
+ }
+ let padding = u32::from(padding);
+ let bit_len = u32::try_from(data.len()).ok()?.checked_mul(8)?.checked_sub(padding)?;
+ let mut packed = vec![0u8; usize::try_from(bit_len.div_ceil(8)).ok()?];
+ for index in 0..bit_len {
+ let source = index + padding;
+ let byte = data.get(usize::try_from(source / 8).ok()?)?;
+ if byte & (0x80 >> (source % 8)) != 0 {
+ let target = packed.get_mut(usize::try_from(index / 8).ok()?)?;
+ *target |= 0x80 >> (index % 8);
+ }
+ }
+ BitString::from_bytes(bit_len, packed).ok()
+}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+
+ #[test]
+ fn padding_bits_are_dropped_from_the_front() {
+ // '10101': three bits of padding, then 10101.
+ let bits = bit_string(&[3, 0b1111_0101]).unwrap();
+
+ assert_eq!(bits.to_string(), "10101");
+ assert_eq!(bits.bit_len(), 5);
+ }
+
+ #[test]
+ fn a_bit_string_can_span_bytes() {
+ let bits = bit_string(&[6, 0b1111_1110, 0b1010_1010]).unwrap();
+
+ assert_eq!(bits.to_string(), "1010101010");
+ }
+
+ #[test]
+ fn a_malformed_bit_string_is_refused() {
+ assert_eq!(bit_string(&[]), None);
+ assert_eq!(bit_string(&[9, 0]), None);
+ }
+}
diff --git a/linux/crates/drivers/duckdb/src/decode/mod.rs b/linux/crates/drivers/duckdb/src/decode/mod.rs
new file mode 100644
index 0000000000..939968efbf
--- /dev/null
+++ b/linux/crates/drivers/duckdb/src/decode/mod.rs
@@ -0,0 +1,293 @@
+//! Values off DuckDB's results.
+//!
+//! DuckDB hands a result over as Arrow arrays, whose types say less
+//! than DuckDB's own: HUGEINT, UHUGEINT and DECIMAL(38,0) arrive alike,
+//! as do VARCHAR, JSON and UUID. Each value is read by the column's
+//! DuckDB type, and one that Arrow cannot carry whole reads as
+//! unreadable rather than as a wrong value.
+
+mod bits;
+mod nested;
+mod text;
+
+use chrono::{DateTime, FixedOffset, NaiveDate, NaiveDateTime, Offset, Utc};
+use duckdb::arrow::array::{Array, AsArray};
+use duckdb::arrow::datatypes::{
+ DataType, Date32Type, Decimal32Type, Decimal64Type, Decimal128Type, Float32Type, Float64Type, Int8Type, Int16Type,
+ Int32Type, Int64Type, IntervalMonthDayNanoType, Time64MicrosecondType, Time64NanosecondType, TimeUnit,
+ TimestampMicrosecondType, TimestampMillisecondType, TimestampNanosecondType, TimestampSecondType, UInt8Type,
+ UInt16Type, UInt32Type, UInt64Type,
+};
+use tablepro_core::column::IntegerWidth;
+use tablepro_core::value::{
+ JsonText, OffsetTimestamp, OtherValue, SqlDecimal, SqlInterval, SqlTime, Temporal, UndecodableReason,
+ UndecodedValue, Value, WideInteger,
+};
+
+use crate::duck_type::DuckType;
+
+pub(crate) use self::bits::bit_string;
+pub(crate) use self::text::interval_text;
+
+/// DuckDB's infinite DATE, one either side of every finite one.
+const DATE_INFINITY: i32 = i32::MAX;
+/// DuckDB's infinite TIMESTAMP, in every unit.
+const TIMESTAMP_INFINITY: i64 = i64::MAX;
+const DAYS_TO_UNIX_EPOCH_FROM_CE: i32 = 719_163;
+
+/// The value at `row` of a column of type `duck`.
+pub(crate) fn value_at(array: &dyn Array, duck: &DuckType, row: usize) -> Value {
+ if array.is_null(row) {
+ return Value::Null;
+ }
+ if duck.is_nested() {
+ return match nested::text(array, duck, row) {
+ Some(text) => Value::Other(Box::new(OtherValue {
+ type_name: duck.name(),
+ text,
+ })),
+ None => unreadable(duck, UndecodableReason::UnsupportedType),
+ };
+ }
+ scalar(array, duck, row).unwrap_or_else(|reason| unreadable(duck, reason))
+}
+
+pub(crate) fn unreadable(duck: &DuckType, reason: UndecodableReason) -> Value {
+ Value::Undecodable(Box::new(UndecodedValue {
+ type_name: duck.name(),
+ reason,
+ }))
+}
+
+type Read = Result;
+
+fn scalar(array: &dyn Array, duck: &DuckType, row: usize) -> Read {
+ Ok(match duck {
+ DuckType::Boolean => Value::Bool(array.as_boolean_opt().ok_or(mismatch())?.value(row)),
+ DuckType::Integer { width, signed } => integer(array, *width, *signed, row)?,
+ DuckType::Float => Value::Float32(primitive::(array, row)?),
+ DuckType::Double => Value::Float64(primitive::(array, row)?),
+ DuckType::Decimal { scale, .. } => Value::Decimal(SqlDecimal::from_scaled_i128(
+ decimal_mantissa(array, row)?,
+ u32::from(*scale),
+ )),
+ DuckType::Varchar | DuckType::Enum => Value::Text(string(array, row)?.to_owned()),
+ DuckType::Json => {
+ let text = string(array, row)?.to_owned();
+ match JsonText::parse(text.clone()) {
+ Ok(json) => Value::Json(json),
+ Err(_) => Value::Text(text),
+ }
+ }
+ DuckType::Blob | DuckType::Geometry => Value::Bytes(bytes(array, row)?.to_vec()),
+ DuckType::Uuid => Value::Uuid(uuid(array, row)?),
+ DuckType::Bit => Value::Bits(bit_string(bytes(array, row)?).ok_or(UndecodableReason::InvalidEncoding)?),
+ DuckType::BigNum => big_number(array, row)?,
+ DuckType::Date => Value::Date(date(primitive::(array, row)?)?),
+ DuckType::Time => Value::Time(time_of_day(array, row)?),
+ DuckType::TimeNs => Value::Time(time_of_day(array, row)?),
+ DuckType::Timestamp(_) => Value::Timestamp(timestamp(array, row)?),
+ DuckType::TimestampTz => Value::TimestampTz(match timestamp(array, row)? {
+ Temporal::Finite(utc) => Temporal::Finite(OffsetTimestamp::from_datetime(
+ utc.and_utc().with_timezone(&utc_offset()),
+ )),
+ Temporal::Infinity => Temporal::Infinity,
+ Temporal::NegInfinity => Temporal::NegInfinity,
+ }),
+ DuckType::Interval => {
+ let interval = primitive::(array, row)?;
+ Value::Interval(SqlInterval {
+ months: interval.months,
+ days: interval.days,
+ microseconds: interval_micros(interval.nanoseconds).ok_or(UndecodableReason::OutOfRange)?,
+ })
+ }
+ DuckType::Null => Value::Null,
+ // Arrow carries a TIME WITH TIME ZONE as its time alone, so the
+ // offset would be lost and the value read as another one.
+ DuckType::TimeTz | DuckType::Variant | DuckType::Other(_) => return Err(UndecodableReason::UnsupportedType),
+ DuckType::List(_) | DuckType::Array(..) | DuckType::Struct(_) | DuckType::Map(..) | DuckType::Union(_) => {
+ return Err(UndecodableReason::UnsupportedType);
+ }
+ })
+}
+
+/// The microseconds of an interval DuckDB handed over as nanoseconds.
+/// DuckDB multiplies its microseconds by 1,000 in an `i64` on the way to
+/// Arrow, which wraps for a span past about 292 years; a wrapped value is
+/// no whole number of microseconds, and one that is cannot be told apart.
+pub(crate) fn interval_micros(nanoseconds: i64) -> Option {
+ (nanoseconds % 1_000 == 0).then_some(nanoseconds / 1_000)
+}
+
+fn mismatch() -> UndecodableReason {
+ UndecodableReason::InvalidEncoding
+}
+
+fn primitive(array: &dyn Array, row: usize) -> Read {
+ array
+ .as_primitive_opt::()
+ .map(|values| values.value(row))
+ .ok_or(mismatch())
+}
+
+fn integer(array: &dyn Array, width: IntegerWidth, signed: bool, row: usize) -> Read {
+ Ok(match (width, signed) {
+ (IntegerWidth::W8, true) => Value::Int(i64::from(primitive::(array, row)?)),
+ (IntegerWidth::W16, true) => Value::Int(i64::from(primitive::(array, row)?)),
+ (IntegerWidth::W24 | IntegerWidth::W32, true) => Value::Int(i64::from(primitive::(array, row)?)),
+ (IntegerWidth::W64, true) => Value::Int(primitive::(array, row)?),
+ (IntegerWidth::W8, false) => Value::UInt(u64::from(primitive::(array, row)?)),
+ (IntegerWidth::W16, false) => Value::UInt(u64::from(primitive::(array, row)?)),
+ (IntegerWidth::W24 | IntegerWidth::W32, false) => Value::UInt(u64::from(primitive::(array, row)?)),
+ (IntegerWidth::W64, false) => Value::UInt(primitive::(array, row)?),
+ (IntegerWidth::W128 | IntegerWidth::W256, true) => {
+ Value::WideInt(WideInteger::from_i128(primitive::(array, row)?))
+ }
+ // A UHUGEINT travels in a 128-bit decimal's bits, so its sign bit is
+ // the top bit of the number.
+ (IntegerWidth::W128 | IntegerWidth::W256, false) => Value::WideInt(WideInteger::from_u128(
+ primitive::(array, row)?.cast_unsigned(),
+ )),
+ })
+}
+
+/// A decimal's digits, at whatever width DuckDB sent them.
+fn decimal_mantissa(array: &dyn Array, row: usize) -> Read {
+ match array.data_type() {
+ DataType::Decimal32(..) => primitive::(array, row).map(i128::from),
+ DataType::Decimal64(..) => primitive::(array, row).map(i128::from),
+ _ => primitive::(array, row),
+ }
+}
+
+/// A string column's text, whichever Arrow layout carries it.
+pub(crate) fn string(array: &dyn Array, row: usize) -> Read<&str> {
+ if let Some(values) = array.as_string_opt::() {
+ return Ok(values.value(row));
+ }
+ if let Some(values) = array.as_string_opt::() {
+ return Ok(values.value(row));
+ }
+ if let Some(values) = array.as_string_view_opt() {
+ return Ok(values.value(row));
+ }
+ dictionary_label(array, row)
+}
+
+/// An ENUM's label: its dictionary entry for the row's key.
+fn dictionary_label(array: &dyn Array, row: usize) -> Read<&str> {
+ let (key, values) = if let Some(dictionary) = array.as_dictionary_opt::() {
+ (usize::from(dictionary.keys().value(row)), dictionary.values())
+ } else if let Some(dictionary) = array.as_dictionary_opt::() {
+ (usize::from(dictionary.keys().value(row)), dictionary.values())
+ } else if let Some(dictionary) = array.as_dictionary_opt::() {
+ let key = usize::try_from(dictionary.keys().value(row)).map_err(|_| mismatch())?;
+ (key, dictionary.values())
+ } else {
+ return Err(mismatch());
+ };
+ if key >= values.len() {
+ return Err(UndecodableReason::OutOfRange);
+ }
+ string(values.as_ref(), key)
+}
+
+pub(crate) fn bytes(array: &dyn Array, row: usize) -> Read<&[u8]> {
+ if let Some(values) = array.as_binary_opt::() {
+ return Ok(values.value(row));
+ }
+ if let Some(values) = array.as_binary_opt::() {
+ return Ok(values.value(row));
+ }
+ if let Some(values) = array.as_binary_view_opt() {
+ return Ok(values.value(row));
+ }
+ array
+ .as_fixed_size_binary_opt()
+ .map(|values| values.value(row))
+ .ok_or(mismatch())
+}
+
+fn uuid(array: &dyn Array, row: usize) -> Read {
+ if let Some(values) = array.as_fixed_size_binary_opt() {
+ return uuid::Uuid::from_slice(values.value(row)).map_err(|_| mismatch());
+ }
+ uuid::Uuid::parse_str(string(array, row)?).map_err(|_| mismatch())
+}
+
+/// A BIGNUM as its digits, where DuckDB sent them as text, and as an
+/// integer where it fits one.
+fn big_number(array: &dyn Array, row: usize) -> Read {
+ let text = string(array, row).map_err(|_| UndecodableReason::UnsupportedType)?;
+ Ok(match text.parse::() {
+ Ok(number) => Value::WideInt(number),
+ Err(_) => Value::Text(text.to_owned()),
+ })
+}
+
+pub(crate) fn date(days: i32) -> Read> {
+ match days {
+ DATE_INFINITY => Ok(Temporal::Infinity),
+ days if days == -DATE_INFINITY => Ok(Temporal::NegInfinity),
+ days => days
+ .checked_add(DAYS_TO_UNIX_EPOCH_FROM_CE)
+ .and_then(NaiveDate::from_num_days_from_ce_opt)
+ .map(Temporal::Finite)
+ .ok_or(UndecodableReason::OutOfRange),
+ }
+}
+
+/// TIME and TIME_NS: DuckDB's TIME runs to 24:00:00 inclusive.
+pub(crate) fn time_of_day(array: &dyn Array, row: usize) -> Read {
+ let nanos = match array.data_type() {
+ DataType::Time64(TimeUnit::Nanosecond) => i128::from(primitive::(array, row)?),
+ _ => i128::from(primitive::(array, row)?) * 1_000,
+ };
+ time_from_nanos(nanos)
+}
+
+pub(crate) fn time_from_nanos(nanos: i128) -> Read {
+ let seconds = nanos.div_euclid(1_000_000_000);
+ let fraction = u32::try_from(nanos.rem_euclid(1_000_000_000)).map_err(|_| UndecodableReason::OutOfRange)?;
+ let hours = u32::try_from(seconds / 3_600).map_err(|_| UndecodableReason::OutOfRange)?;
+ let minutes = u8::try_from(seconds / 60 % 60).map_err(|_| UndecodableReason::OutOfRange)?;
+ let whole = u8::try_from(seconds % 60).map_err(|_| UndecodableReason::OutOfRange)?;
+ SqlTime::new(false, hours, minutes, whole, fraction).map_err(|_| UndecodableReason::OutOfRange)
+}
+
+/// A timestamp in whatever unit it counts, as the wall time it holds.
+pub(crate) fn timestamp(array: &dyn Array, row: usize) -> Read> {
+ let (count, unit) = match array.data_type() {
+ DataType::Timestamp(TimeUnit::Second, _) => (primitive::(array, row)?, TimeUnit::Second),
+ DataType::Timestamp(TimeUnit::Millisecond, _) => (
+ primitive::(array, row)?,
+ TimeUnit::Millisecond,
+ ),
+ DataType::Timestamp(TimeUnit::Nanosecond, _) => {
+ (primitive::(array, row)?, TimeUnit::Nanosecond)
+ }
+ _ => (
+ primitive::(array, row)?,
+ TimeUnit::Microsecond,
+ ),
+ };
+ match count {
+ TIMESTAMP_INFINITY => return Ok(Temporal::Infinity),
+ count if count == -TIMESTAMP_INFINITY => return Ok(Temporal::NegInfinity),
+ _ => {}
+ }
+ let stamp = match unit {
+ TimeUnit::Second => DateTime::from_timestamp(count, 0),
+ TimeUnit::Millisecond => DateTime::from_timestamp_millis(count),
+ TimeUnit::Nanosecond => Some(DateTime::from_timestamp_nanos(count)),
+ TimeUnit::Microsecond => DateTime::from_timestamp_micros(count),
+ };
+ stamp
+ .map(|stamp| Temporal::Finite(stamp.naive_utc()))
+ .ok_or(UndecodableReason::OutOfRange)
+}
+
+fn utc_offset() -> FixedOffset {
+ Utc.fix()
+}
diff --git a/linux/crates/drivers/duckdb/src/decode/nested.rs b/linux/crates/drivers/duckdb/src/decode/nested.rs
new file mode 100644
index 0000000000..e4a78edaf3
--- /dev/null
+++ b/linux/crates/drivers/duckdb/src/decode/nested.rs
@@ -0,0 +1,100 @@
+//! LIST, ARRAY, STRUCT, MAP and UNION values as DuckDB's own text.
+
+use duckdb::arrow::array::{Array, AsArray};
+
+use super::text::{escaped, scalar_text};
+use crate::duck_type::DuckType;
+
+/// The value at `row` as DuckDB prints it, or `None` where a part of it
+/// cannot be read off Arrow.
+pub(super) fn text(array: &dyn Array, duck: &DuckType, row: usize) -> Option {
+ match duck {
+ DuckType::List(element) => list(array, element, row),
+ DuckType::Array(element, _) => {
+ let values = array.as_fixed_size_list_opt()?;
+ let width = usize::try_from(values.value_length()).ok()?;
+ let start = values.value_offset(row);
+ let start = usize::try_from(start).ok()?;
+ items(values.values().as_ref(), element, start..start + width)
+ }
+ DuckType::Struct(fields) => {
+ let values = array.as_struct_opt()?;
+ let mut parts = Vec::with_capacity(fields.len());
+ for (index, (name, field)) in fields.iter().enumerate() {
+ let column = values.columns().get(index)?;
+ parts.push(format!(
+ "{}: {}",
+ escaped(name, true),
+ part(column.as_ref(), field, row)?
+ ));
+ }
+ Some(format!("{{{}}}", parts.join(", ")))
+ }
+ DuckType::Map(key, value) => {
+ let values = array.as_map_opt()?;
+ let offsets = values.value_offsets();
+ let (start, end) = (
+ usize::try_from(*offsets.get(row)?).ok()?,
+ usize::try_from(*offsets.get(row + 1)?).ok()?,
+ );
+ let mut parts = Vec::with_capacity(end - start);
+ for entry in start..end {
+ parts.push(format!(
+ "{}={}",
+ part(values.keys().as_ref(), key, entry)?,
+ part(values.values().as_ref(), value, entry)?
+ ));
+ }
+ Some(format!("{{{}}}", parts.join(", ")))
+ }
+ DuckType::Union(members) => {
+ let values = array.as_union_opt()?;
+ let tag = values.type_id(row);
+ let (_, member) = members.get(usize::try_from(tag).ok()?)?;
+ let child = values.child(tag);
+ let at = values.value_offset(row);
+ match child.is_null(at) {
+ true => Some("NULL".to_owned()),
+ false => whole(child.as_ref(), member, at),
+ }
+ }
+ scalar => scalar_text(array, scalar, row),
+ }
+}
+
+fn list(array: &dyn Array, element: &DuckType, row: usize) -> Option {
+ if let Some(values) = array.as_list_opt::() {
+ let offsets = values.value_offsets();
+ let range = usize::try_from(*offsets.get(row)?).ok()?..usize::try_from(*offsets.get(row + 1)?).ok()?;
+ return items(values.values().as_ref(), element, range);
+ }
+ let values = array.as_list_opt::()?;
+ let offsets = values.value_offsets();
+ let range = usize::try_from(*offsets.get(row)?).ok()?..usize::try_from(*offsets.get(row + 1)?).ok()?;
+ items(values.values().as_ref(), element, range)
+}
+
+fn items(values: &dyn Array, element: &DuckType, range: std::ops::Range) -> Option {
+ let parts: Option> = range.map(|index| part(values, element, index)).collect();
+ Some(format!("[{}]", parts?.join(", ")))
+}
+
+/// One member of a nested value: NULL, a nested value as it prints, or
+/// a scalar escaped as DuckDB escapes one there.
+fn part(values: &dyn Array, duck: &DuckType, row: usize) -> Option {
+ if values.is_null(row) {
+ return Some("NULL".to_owned());
+ }
+ match duck.is_nested() {
+ true => text(values, duck, row),
+ false => Some(escaped(&scalar_text(values, duck, row)?, false)),
+ }
+}
+
+/// A union prints its member's own text, unescaped.
+fn whole(values: &dyn Array, duck: &DuckType, row: usize) -> Option {
+ match duck.is_nested() {
+ true => text(values, duck, row),
+ false => scalar_text(values, duck, row),
+ }
+}
diff --git a/linux/crates/drivers/duckdb/src/decode/text.rs b/linux/crates/drivers/duckdb/src/decode/text.rs
new file mode 100644
index 0000000000..7fd62fcfdf
--- /dev/null
+++ b/linux/crates/drivers/duckdb/src/decode/text.rs
@@ -0,0 +1,261 @@
+//! A value inside a nested one, spelled as DuckDB's CAST to VARCHAR
+//! spells it, so a list in an editor run reads as it does in a table tab.
+
+use chrono::NaiveDateTime;
+use duckdb::arrow::array::{Array, AsArray};
+use duckdb::arrow::datatypes::{Float32Type, Float64Type, IntervalMonthDayNanoType};
+use tablepro_core::value::{SqlDecimal, Temporal};
+
+use super::{bit_string, bytes, date, decimal_mantissa, integer, string, time_of_day, timestamp, uuid};
+use crate::duck_type::DuckType;
+
+/// A scalar's text, or `None` for one that cannot be read off Arrow.
+pub(super) fn scalar_text(array: &dyn Array, duck: &DuckType, row: usize) -> Option {
+ Some(match duck {
+ DuckType::Boolean => array.as_boolean_opt()?.value(row).to_string(),
+ DuckType::Integer { width, signed } => {
+ tablepro_core::export::value_to_text(&integer(array, *width, *signed, row).ok()?)?
+ }
+ DuckType::Float => float_text(f64::from(array.as_primitive_opt::()?.value(row)), true),
+ DuckType::Double => float_text(array.as_primitive_opt::()?.value(row), false),
+ DuckType::Decimal { scale, .. } => decimal_text(decimal_mantissa(array, row).ok()?, *scale),
+ DuckType::Varchar | DuckType::Enum | DuckType::Json => string(array, row).ok()?.to_owned(),
+ DuckType::Blob => blob_text(bytes(array, row).ok()?),
+ DuckType::Uuid => uuid(array, row).ok()?.to_string(),
+ DuckType::Bit => bit_string(bytes(array, row).ok()?)?.to_string(),
+ DuckType::Date => match date(
+ array
+ .as_primitive_opt::()?
+ .value(row),
+ )
+ .ok()?
+ {
+ Temporal::Finite(day) => crate::encode::date_text(day),
+ Temporal::Infinity => "infinity".to_owned(),
+ Temporal::NegInfinity => "-infinity".to_owned(),
+ },
+ DuckType::Time | DuckType::TimeNs => time_of_day(array, row).ok()?.format(None),
+ DuckType::Timestamp(_) => stamp_text(timestamp(array, row).ok()?, ""),
+ // Without the ICU extension DuckDB keeps no time zone but UTC.
+ DuckType::TimestampTz => stamp_text(timestamp(array, row).ok()?, "+00"),
+ DuckType::Interval => {
+ let interval = array.as_primitive_opt::()?.value(row);
+ interval_text(
+ interval.months,
+ interval.days,
+ super::interval_micros(interval.nanoseconds)?,
+ )
+ }
+ _ => return None,
+ })
+}
+
+/// Whether DuckDB quotes this text inside a nested value: an empty or
+/// padded string, the word null, or one holding a character that would
+/// read as part of the nesting.
+pub(super) fn needs_quotes(text: &str, key: bool) -> bool {
+ let bytes = text.as_bytes();
+ let space = |byte: &u8| matches!(byte, b' ' | b'\t' | b'\n' | 0x0b | 0x0c | b'\r');
+ key || bytes.is_empty()
+ || bytes.first().is_some_and(space)
+ || (bytes.len() >= 2 && bytes.last().is_some_and(space))
+ || text.eq_ignore_ascii_case("null")
+ || bytes.iter().any(|byte| {
+ matches!(
+ byte,
+ b'"' | b'\'' | b'(' | b')' | b',' | b':' | b'=' | b'[' | b']' | b'{' | b'}'
+ )
+ })
+}
+
+/// A string as DuckDB writes it inside a list, a struct or a map.
+pub(super) fn escaped(text: &str, key: bool) -> String {
+ if !needs_quotes(text, key) {
+ return text.to_owned();
+ }
+ let mut quoted = String::with_capacity(text.len() + 2);
+ quoted.push('\'');
+ for character in text.chars() {
+ if matches!(character, '\'' | '\\') {
+ quoted.push('\\');
+ }
+ quoted.push(character);
+ }
+ quoted.push('\'');
+ quoted
+}
+
+/// A double as DuckDB prints one: the shortest digits that read back
+/// as the same number, in plain notation from 1e-4 up to 1e16, with a
+/// `.0` on a whole number, and in exponent notation outside that.
+pub(super) fn float_text(value: f64, single: bool) -> String {
+ if value.is_nan() {
+ return "nan".to_owned();
+ }
+ if value.is_infinite() {
+ return if value < 0.0 { "-inf" } else { "inf" }.to_owned();
+ }
+ // A FLOAT's value came from an f32, so it narrows back exactly.
+ let scientific = match single {
+ true => format!("{:e}", value as f32),
+ false => format!("{value:e}"),
+ };
+ let (mantissa, exponent) = scientific.split_once('e').unwrap_or((&scientific, "0"));
+ let exponent: i32 = exponent.parse().unwrap_or(0);
+ let negative = mantissa.starts_with('-');
+ let digits: String = mantissa.chars().filter(char::is_ascii_digit).collect();
+ let sign = if negative { "-" } else { "" };
+ if (-4..16).contains(&exponent) {
+ let point = exponent + 1;
+ let body = if point <= 0 {
+ format!("0.{}{digits}", "0".repeat(point.unsigned_abs() as usize))
+ } else {
+ let point = point.unsigned_abs() as usize;
+ let padded = format!("{digits:0 first.to_owned(),
+ false => format!("{first}.{rest}"),
+ };
+ let exponent_sign = if exponent < 0 { '-' } else { '+' };
+ format!("{sign}{mantissa}e{exponent_sign}{:02}", exponent.unsigned_abs())
+}
+
+fn decimal_text(mantissa: i128, scale: u8) -> String {
+ let text = SqlDecimal::from_scaled_i128(mantissa, u32::from(scale)).to_string();
+ let present = text.split_once('.').map_or(0, |(_, fraction)| fraction.len());
+ let missing = usize::from(scale).saturating_sub(present);
+ match (missing, present) {
+ (0, _) => text,
+ (missing, 0) => format!("{text}.{}", "0".repeat(missing)),
+ (missing, _) => format!("{text}{}", "0".repeat(missing)),
+ }
+}
+
+/// Printable ASCII as itself, every other byte and the quote and
+/// backslash characters as `\xHH`.
+fn blob_text(bytes: &[u8]) -> String {
+ let mut text = String::with_capacity(bytes.len());
+ for byte in bytes {
+ if (32..=126).contains(byte) && !matches!(byte, b'\\' | b'\'' | b'"') {
+ text.push(char::from(*byte));
+ } else {
+ text.push_str(&format!("\\x{byte:02X}"));
+ }
+ }
+ text
+}
+
+fn stamp_text(stamp: Temporal