feat: add duckdb checks for unsupported column types

Cargo.toml

@@ -39,7 +39,7 @@ r2d2 = { version = "0.8.10", optional = true }
 rusqlite = { version = "0.31.0", optional = true }
 sea-query = { git = "https://github.com/spiceai/sea-query.git", rev = "213b6b876068f58159ebdd5852604a021afaebf9", features = ["backend-sqlite", "backend-postgres", "postgres-array", "with-rust_decimal", "with-bigdecimal", "with-time", "with-chrono"] }
 secrecy = "0.8.0"
-serde = { version = "1.0.209", optional = true }
+serde = { version = "1.0.209", features = ["derive"] }
 serde_json = "1.0.124"
 snafu = "0.8.3"
 time = "0.3.36"
@@ -80,7 +80,7 @@ insta = { version = "1.40.0", features = ["filters"] }
 mysql = ["dep:mysql_async", "dep:async-stream"]
 postgres = ["dep:tokio-postgres", "dep:uuid", "dep:postgres-native-tls", "dep:bb8", "dep:bb8-postgres", "dep:native-tls", "dep:pem", "dep:async-stream"]
 sqlite = ["dep:rusqlite", "dep:tokio-rusqlite"]
-duckdb = ["dep:duckdb", "dep:r2d2", "dep:uuid", "dep:dyn-clone", "dep:async-stream"]
+duckdb = ["dep:duckdb", "dep:r2d2", "dep:uuid", "dep:dyn-clone", "dep:async-stream", "dep:arrow-schema"]
 flight = [
   "dep:arrow-array",
   "dep:arrow-flight",
@@ -92,7 +92,6 @@ flight = [
   "dep:datafusion-physical-plan",
   "dep:datafusion-proto",
   "dep:prost",
-  "dep:serde",
   "dep:tonic",
 ]
 duckdb-federation = ["duckdb"]

src/duckdb.rs

@@ -1,14 +1,3 @@
-use crate::sql::db_connection_pool::{
-    self,
-    dbconnection::{
-        duckdbconn::{
-            flatten_table_function_name, is_table_function, DuckDBParameter, DuckDbConnection,
-        },
-        get_schema, DbConnection,
-    },
-    duckdbpool::DuckDbConnectionPool,
-    DbConnectionPool, DbInstanceKey, Mode,
-};
 use crate::sql::sql_provider_datafusion;
 use crate::util::{
     self,
@@ -17,6 +6,20 @@ use crate::util::{
     indexes::IndexType,
     on_conflict::{self, OnConflict},
 };
+use crate::{
+    sql::db_connection_pool::{
+        self,
+        dbconnection::{
+            duckdbconn::{
+                flatten_table_function_name, is_table_function, DuckDBParameter, DuckDbConnection,
+            },
+            get_schema, DbConnection,
+        },
+        duckdbpool::DuckDbConnectionPool,
+        DbConnectionPool, DbInstanceKey, Mode,
+    },
+    InvalidTypeAction,
+};
 use arrow::{array::RecordBatch, datatypes::SchemaRef};
 use async_trait::async_trait;
 use datafusion::{
@@ -120,6 +123,7 @@ type Result<T, E = Error> = std::result::Result<T, E>;
 pub struct DuckDBTableProviderFactory {
     access_mode: AccessMode,
     instances: Arc<Mutex<HashMap<DbInstanceKey, DuckDbConnectionPool>>>,
+    invalid_type_action: InvalidTypeAction,
 }
 
 const DUCKDB_DB_PATH_PARAM: &str = "open";
@@ -132,9 +136,16 @@ impl DuckDBTableProviderFactory {
         Self {
             access_mode,
             instances: Arc::new(Mutex::new(HashMap::new())),
+            invalid_type_action: InvalidTypeAction::Error,
         }
     }
 
+    #[must_use]
+    pub fn with_invalid_type_action(mut self, invalid_type_action: InvalidTypeAction) -> Self {
+        self.invalid_type_action = invalid_type_action;
+        self
+    }
+
     #[must_use]
     pub fn attach_databases(&self, options: &HashMap<String, String>) -> Vec<Arc<str>> {
         options
@@ -181,7 +192,9 @@ impl DuckDBTableProviderFactory {
             return Ok(instance.clone());
         }
 
-        let pool = DuckDbConnectionPool::new_memory().context(DbConnectionPoolSnafu)?;
+        let pool = DuckDbConnectionPool::new_memory()
+            .context(DbConnectionPoolSnafu)?
+            .with_invalid_type_action(self.invalid_type_action);
 
         instances.insert(key, pool.clone());
 
@@ -201,7 +214,8 @@ impl DuckDBTableProviderFactory {
         }
 
         let pool = DuckDbConnectionPool::new_file(&db_path, &self.access_mode)
-            .context(DbConnectionPoolSnafu)?;
+            .context(DbConnectionPoolSnafu)?
+            .with_invalid_type_action(self.invalid_type_action);
 
         instances.insert(key, pool.clone());
 

src/lib.rs

@@ -1,3 +1,4 @@
+use serde::{Deserialize, Serialize};
 use snafu::prelude::*;
 
 pub mod sql;
@@ -23,3 +24,12 @@ pub enum Error {
     #[snafu(display("Error reading file: {source}"))]
     FileReadError { source: std::io::Error },
 }
+
+#[derive(PartialEq, Eq, Clone, Copy, Default, Debug, Serialize, Deserialize)]
+#[serde(rename_all = "lowercase")]
+pub enum InvalidTypeAction {
+    #[default]
+    Error,
+    Warn,
+    Ignore,
+}

src/sql/db_connection_pool/dbconnection.rs

@@ -1,5 +1,8 @@
 use std::{any::Any, sync::Arc};
 
+#[cfg(feature = "duckdb")]
+use arrow_schema::DataType;
+
 use datafusion::{
     arrow::datatypes::SchemaRef, execution::SendableRecordBatchStream, sql::TableReference,
 };
@@ -25,6 +28,13 @@ pub enum Error {
     #[snafu(display("Unable to get schema: {source}"))]
     UnableToGetSchema { source: GenericError },
 
+    #[snafu(display("The field '{field_name}' has an unsupported data type: {data_type}"))]
+    #[cfg(feature = "duckdb")]
+    UnsupportedDataType {
+        data_type: DataType,
+        field_name: String,
+    },
+
     #[snafu(display("Unable to query arrow: {source}"))]
     UnableToQueryArrow { source: GenericError },