Repository: vigimite/aqueducts (language: rust)

Issue:
Feature: ODBC Destination
Currently only ODBC sources are supported but I want to add support for ODBC destinations too.

The implementation will be similar to the sources however it will also need to create the table in the specified DB during the execution if it doesn't exist. When registering this destination we need to check if the schema is compatible with the table definition.

Optionally I will add support for SQL migrations for the table schema


Discussion / hints:

Fix patch:
diff --git a/.github/workflows/build.yml b/.github/workflows/build.yml
index 53062c1..479aa74 100644
--- a/.github/workflows/build.yml
+++ b/.github/workflows/build.yml
@@ -12,6 +12,20 @@ jobs:
   test:
     runs-on: ubuntu-latest
 
+    services:
+      postgres:
+        image: postgres:15
+        env:
+          POSTGRES_USER: postgres
+          POSTGRES_PASSWORD: postgres
+        ports:
+          - 5432:5432
+        options: >-
+          --health-cmd "pg_isready -U postgres -d postgres"
+          --health-interval 10s
+          --health-timeout 5s
+          --health-retries 5
+
     steps:
       - name: Checkout repository
         uses: actions/checkout@v4
@@ -34,11 +48,19 @@ jobs:
           toolchain: stable
           components: rustfmt
 
+      - name: Setup test environment
+        run: |
+          sudo apt-get update && sudo apt-get install -y postgresql-client unixodbc odbc-postgresql
+          export CONTAINER_ID=$(docker ps --filter "name=postgres" --format "{{.ID}}")
+          docker cp ./examples/temp_readings_jan_2024.csv $CONTAINER_ID:/opt/
+          docker cp ./examples/temp_readings_feb_2024.csv $CONTAINER_ID:/opt/
+          PGPASSWORD=postgres psql -h localhost -U postgres -d postgres -f ./db/init.sql
+
       - name: Test aqueducts
         run: |
           cd aqueducts/core
           cargo fmt -- --check
-          cargo test
+          cargo test --workspace --all-features
 
   build:
     runs-on: ubuntu-latest
@@ -67,9 +89,9 @@ jobs:
       - name: Build aqueducts in release mode
         run: |
           cd aqueducts/core
-          cargo build --release
+          cargo build --workspace --release
 
       - name: Build aqueducts-cli in release mode
         run: |
           cd aqueducts-cli
-          cargo build --release
+          cargo build --workspace --release
diff --git a/Cargo.toml b/Cargo.toml
index 06cc8b3..5f161f0 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -1,6 +1,7 @@
 [workspace]
 members = [
   "aqueducts/core",
+  "aqueducts/odbc",
   "aqueducts/utils",
   "aqueducts/schema_gen",
   "aqueducts-cli"
@@ -21,6 +22,7 @@ license-file = "LICENSE"
 
 [workspace.dependencies]
 aqueducts = { path = "aqueducts/core", version = "0.3.2" }
+aqueducts-odbc = { path = "aqueducts/odbc", version = "0.3.2" }
 aqueducts-utils = { path = "aqueducts/utils", version = "0.3.2" }
 
 datafusion = "39"
diff --git a/aqueducts/core/Cargo.toml b/aqueducts/core/Cargo.toml
index abe6ff8..09084a2 100644
--- a/aqueducts/core/Cargo.toml
+++ b/aqueducts/core/Cargo.toml
@@ -16,7 +16,7 @@ default = []
 s3 = ["aqueducts-utils/s3", "object_store/aws"]
 gcs = ["aqueducts-utils/gcs", "object_store/gcp"]
 azure = ["aqueducts-utils/azure", "object_store/azure"]
-odbc = ["aqueducts-utils/odbc"]
+odbc = ["aqueducts-odbc"]
 schema_gen = ["schemars"]
 
 [dependencies]
@@ -37,6 +37,7 @@ regex.workspace = true
 derive-new.workspace = true
 chrono.workspace = true
 
+aqueducts-odbc = { workspace = true, optional = true }
 aqueducts-utils.workspace = true
 
 [dev-dependencies]
diff --git a/aqueducts/core/src/destinations/error.rs b/aqueducts/core/src/destinations/error.rs
index e53750d..7dede60 100644
--- a/aqueducts/core/src/destinations/error.rs
+++ b/aqueducts/core/src/destinations/error.rs
@@ -3,6 +3,11 @@ pub enum Error {
     #[error("Unknown parquet config key: {0}")]
     UnknownConfigOption(String),
 
+    // -- Modules
+    #[cfg(feature = "odbc")]
+    #[error("OdbcError({0})")]
+    OdbcError(#[from] aqueducts_odbc::error::Error),
+
     // -- External
     #[error("ArrowError({0})")]
     ArrowError(#[from] datafusion::arrow::error::ArrowError),
diff --git a/aqueducts/core/src/destinations/mod.rs b/aqueducts/core/src/destinations/mod.rs
index 4088b1b..8b8b577 100644
--- a/aqueducts/core/src/destinations/mod.rs
+++ b/aqueducts/core/src/destinations/mod.rs
@@ -6,6 +6,8 @@ use tracing::{info, instrument};
 
 pub mod delta;
 pub mod file;
+#[cfg(feature = "odbc")]
+pub mod odbc;
 
 pub(crate) mod error;
 pub(crate) type Result<T> = core::result::Result<T, error::Error>;
@@ -19,11 +21,17 @@ pub enum Destination {
     Delta(delta::DeltaDestination),
     /// A file output destination
     File(file::FileDestination),
+    #[cfg(feature = "odbc")]
+    /// An ODBC insert query to write to a DB table
+    Odbc(odbc::OdbcDestination),
 }
 
 /// Creates a `Destination`
 #[instrument(skip(ctx, destination), err)]
-pub async fn create_destination(ctx: Arc<SessionContext>, destination: &Destination) -> Result<()> {
+pub async fn register_destination(
+    ctx: Arc<SessionContext>,
+    destination: &Destination,
+) -> Result<()> {
     match destination {
         Destination::Delta(table_def) => {
             info!(
@@ -38,6 +46,15 @@ pub async fn create_destination(ctx: Arc<SessionContext>, destination: &Destinat
             register_object_store(ctx, &file_dest.location, &file_dest.storage_options)?;
             Ok(())
         }
+        #[cfg(feature = "odbc")]
+        Destination::Odbc(odbc_def) => {
+            aqueducts_odbc::register_odbc_destination(
+                odbc_def.connection_string.as_str(),
+                odbc_def.name.as_str(),
+            )
+            .await?;
+            Ok(())
+        }
     }
 }
 
@@ -58,6 +75,12 @@ pub async fn write_to_destination(destination: &Destination, data: DataFrame) ->
             info!("Writing data to file at location '{}'", file_def.location);
             file::write(file_def, data).await?;
 
+            Ok(())
+        }
+        #[cfg(feature = "odbc")]
+        Destination::Odbc(odbc_def) => {
+            odbc::write(odbc_def, data).await?;
+
             Ok(())
         }
     }
diff --git a/aqueducts/core/src/destinations/odbc.rs b/aqueducts/core/src/destinations/odbc.rs
new file mode 100644
index 0000000..66937bb
--- /dev/null
+++ b/aqueducts/core/src/destinations/odbc.rs
@@ -0,0 +1,36 @@
+use datafusion::dataframe::DataFrame;
+use serde::{Deserialize, Serialize};
+use std::sync::Arc;
+
+use super::Result;
+
+/// A file output destination
+#[derive(Debug, Clone, Serialize, Deserialize, derive_new::new)]
+#[cfg_attr(feature = "schema_gen", derive(schemars::JsonSchema))]
+pub struct OdbcDestination {
+    /// Name of the ODBC destination
+    pub name: String,
+
+    /// ODBC connection string
+    /// Please reference the respective database connection string syntax (e.g. <https://www.connectionstrings.com/postgresql-odbc-driver-psqlodbc/>)
+    pub connection_string: String,
+
+    /// batch size (rows) to use when inserting data
+    pub batch_size: usize,
+}
+
+pub(super) async fn write(odbc_def: &OdbcDestination, data: DataFrame) -> Result<()> {
+    let schema = data.schema().as_arrow().clone();
+    let batches = data.collect().await?;
+
+    aqueducts_odbc::write_arrow_batches(
+        odbc_def.connection_string.as_str(),
+        odbc_def.name.as_str(),
+        batches,
+        Arc::new(schema),
+        odbc_def.batch_size,
+    )
+    .await?;
+
+    Ok(())
+}
diff --git a/aqueducts/core/src/lib.rs b/aqueducts/core/src/lib.rs
index 583b4ac..0699554 100644
--- a/aqueducts/core/src/lib.rs
+++ b/aqueducts/core/src/lib.rs
@@ -194,7 +194,7 @@ pub async fn run_pipeline(aqueduct: Aqueduct, ctx: Option<SessionContext>) -> Re
     if let Some(destination) = &aqueduct.destination {
         let time = Instant::now();
 
-        create_destination(ctx.clone(), destination).await?;
+        register_destination(ctx.clone(), destination).await?;
 
         info!(
             "Created destination ... Elapsed time: {:.2?}",
diff --git a/aqueducts/core/src/sources/error.rs b/aqueducts/core/src/sources/error.rs
index 805cb41..276a49a 100644
--- a/aqueducts/core/src/sources/error.rs
+++ b/aqueducts/core/src/sources/error.rs
@@ -1,7 +1,10 @@
 #[derive(Debug, thiserror::Error)]
 pub enum Error {
-    #[error("To use ODBC type you need to enable the odbc feature")]
-    OdbcFeatureDisabled,
+    // -- Modules
+    #[cfg(feature = "odbc")]
+    #[error("OdbcError({0})")]
+    OdbcError(#[from] aqueducts_odbc::error::Error),
+
     // -- External
     #[error("ArrowError({0})")]
     ArrowError(#[from] datafusion::arrow::error::ArrowError),
diff --git a/aqueducts/core/src/sources/mod.rs b/aqueducts/core/src/sources/mod.rs
index 0837b21..22d1dff 100644
--- a/aqueducts/core/src/sources/mod.rs
+++ b/aqueducts/core/src/sources/mod.rs
@@ -29,6 +29,7 @@ pub enum Source {
     File(FileSource),
     /// A directory source
     Directory(DirSource),
+    #[cfg(feature = "odbc")]
     /// An ODBC source
     Odbc(OdbcSource),
 }
@@ -109,6 +110,7 @@ pub struct DirSource {
 }
 
 /// An ODBC source
+#[cfg(feature = "odbc")]
 #[derive(Debug, Clone, Serialize, Deserialize)]
 #[cfg_attr(feature = "schema_gen", derive(schemars::JsonSchema))]
 pub struct OdbcSource {
@@ -204,13 +206,13 @@ pub async fn register_source(ctx: Arc<SessionContext>, source: Source) -> Result
             register_dir_source(ctx, dir_source).await?
         }
         #[cfg(feature = "odbc")]
-        Source::Odbc(odbc_source) if cfg!(feature = "odbc") => {
+        Source::Odbc(odbc_source) => {
             info!(
                 "Registering ODBC source '{}' using query '{}'",
                 odbc_source.name, odbc_source.query
             );
 
-            aqueducts_utils::odbc::register_odbc_source(
+            aqueducts_odbc::register_odbc_source(
                 ctx,
                 odbc_source.connection_string.as_str(),
                 odbc_source.query.as_str(),
@@ -218,9 +220,6 @@ pub async fn register_source(ctx: Arc<SessionContext>, source: Source) -> Result
             )
             .await?
         }
-        Source::Odbc(_) => {
-            return Err(error::Error::OdbcFeatureDisabled);
-        }
     };
 
     Ok(())
diff --git a/aqueducts/odbc/Cargo.toml b/aqueducts/odbc/Cargo.toml
new file mode 100644
index 0000000..3f4b40f
--- /dev/null
+++ b/aqueducts/odbc/Cargo.toml
@@ -0,0 +1,25 @@
+[package]
+name = "aqueducts-odbc"
+authors.workspace = true
+edition.workspace = true
+description.workspace = true
+repository.workspace = true
+readme.workspace = true
+version.workspace = true
+homepage.workspace = true
+keywords.workspace = true
+categories.workspace = true
+license-file.workspace = true
+
+[dependencies]
+datafusion.workspace = true
+deltalake.workspace = true
+serde.workspace = true
+thiserror.workspace = true
+tracing.workspace = true
+
+arrow-odbc = { version = "11.2.0" }
+
+[dev-dependencies]
+tokio = { version = "1", features = ["full"] }
+tracing-test = "0.2"
diff --git a/aqueducts/odbc/src/error.rs b/aqueducts/odbc/src/error.rs
new file mode 100644
index 0000000..bdbfe94
--- /dev/null
+++ b/aqueducts/odbc/src/error.rs
@@ -0,0 +1,14 @@
+#[derive(Debug, thiserror::Error)]
+pub enum Error {
+    // -- External
+    #[error("ArrowError({0})")]
+    ArrowError(#[from] datafusion::arrow::error::ArrowError),
+    #[error("DataFusionError({0})")]
+    DataFusionError(#[from] datafusion::error::DataFusionError),
+    #[error("OdbcError({0})")]
+    OdbcError(#[from] arrow_odbc::Error),
+    #[error("OdbcApiError({0})")]
+    OdbcApiError(#[from] arrow_odbc::odbc_api::Error),
+    #[error("OdbcWriterError({0})")]
+    OdbcWriterError(#[from] arrow_odbc::WriterError),
+}
diff --git a/aqueducts/odbc/src/lib.rs b/aqueducts/odbc/src/lib.rs
new file mode 100644
index 0000000..dc3c693
--- /dev/null
+++ b/aqueducts/odbc/src/lib.rs
@@ -0,0 +1,249 @@
+use std::sync::Arc;
+
+use arrow_odbc::odbc_api::{ConnectionOptions, Environment};
+use arrow_odbc::{insert_into_table, OdbcReaderBuilder};
+use datafusion::arrow::datatypes::Schema;
+use datafusion::arrow::{array::RecordBatch, error::ArrowError};
+use datafusion::execution::context::SessionContext;
+use deltalake::arrow::array::RecordBatchIterator;
+
+pub mod error;
+
+pub type Result<T> = core::result::Result<T, error::Error>;
+
+/// Register a table via ODBC using [arrow-odbc](https://docs.rs/arrow-odbc)
+/// ```rust,ignore
+/// use datafusion::prelude::SessionContext;
+/// use arrow_odbc::odbc_api::ConnectionOptions;
+///
+/// let connection_string: &str = "\
+///     Driver={PostgreSQL Unicode};\
+///     Server=localhost;\
+///     UID=postgres;\
+///     PWD=postgres;\
+/// ";
+///
+/// // query to request data from the ODBC source
+/// // make sure to constrain this query to a dataset that is manageble in memory
+/// let query = "SELECT * FROM my_table WHERE date > '2024-01-01'";
+///
+/// let ctx = SessionContext::new();
+///
+/// register_odbc_source(&ctx, query, connection_string, "my_table_name").await.unwrap();
+///
+/// let df = ctx.sql("SELECT * FROM my_table_name").await.unwrap();
+/// df.show().await.unwrap();
+/// ```
+pub async fn register_odbc_source(
+    ctx: Arc<SessionContext>,
+    connection_string: &str,
+    query: &str,
+    source_name: &str,
+) -> Result<()> {
+    let odbc_environment = Environment::new().unwrap();
+
+    let connection = odbc_environment
+        .connect_with_connection_string(connection_string, ConnectionOptions::default())?;
+
+    let parameters = ();
+
+    let cursor = connection
+        .execute(query, parameters)?
+        .expect("SELECT statement must produce a cursor");
+
+    let reader = OdbcReaderBuilder::new().build(cursor)?;
+
+    let batches = reader
+        .into_iter()
+        .collect::<std::result::Result<Vec<RecordBatch>, ArrowError>>()?;
+
+    let df = ctx.read_batches(batches)?;
+    let schema: Arc<Schema> = Arc::new(df.schema().into());
+    let batch = deltalake::arrow::compute::concat_batches(&schema, df.collect().await?.iter())?;
+
+    ctx.register_batch(source_name, batch)?;
+
+    Ok(())
+}
+
+/// Checks if the provided table for the destination exists
+/// will try to query one record from the provided table name
+pub async fn register_odbc_destination(
+    connection_string: &str,
+    destination_name: &str,
+) -> Result<()> {
+    let odbc_environment = Environment::new().unwrap();
+
+    let connection = odbc_environment
+        .connect_with_connection_string(connection_string, ConnectionOptions::default())?;
+
+    let parameters = ();
+
+    let query = format!("SELECT * FROM {destination_name} LIMIT 1");
+    connection
+        .execute(query.as_str(), parameters)?
+        .expect("SELECT statement must produce a cursor");
+
+    Ok(())
+}
+
+/// Write arrow batches to a table via ODBC using [arrow-odbc](https://docs.rs/arrow-odbc)
+/// ```rust,ignore
+/// use datafusion::prelude::SessionContext;
+/// use arrow_odbc::odbc_api::ConnectionOptions;
+/// use std::sync::Arc;
+///
+/// let connection_string: &str = "\
+///     Driver={PostgreSQL Unicode};\
+///     Server=localhost;\
+///     UID=postgres;\
+///     PWD=postgres;\
+/// ";
+///
+/// let query = "SELECT * FROM my_table WHERE date > '2024-01-01'";
+/// let ctx = SessionContext::new();
+/// register_odbc_source(&ctx, query, connection_string, "my_table_name").await.unwrap();
+///
+/// //check if table exists
+/// register_odbc_destination(connection_string, "another_table").await.unwrap();
+///
+/// let df = ctx.sql("SELECT * FROM my_table_name").await.unwrap();
+/// let schema = df.schema().as_arrow().clone();
+/// let batches = df.collect().await.unwrap();
+///
+/// write_arrow_batches(connection_string, "another_table", batches, Arc::new(schema), 1000).await.unwrap();
+/// ```
+pub async fn write_arrow_batches(
+    connection_string: &str,
+    destination_name: &str,
+    batches: Vec<RecordBatch>,
+    schema: Arc<Schema>,
+    batch_size: usize,
+) -> Result<()> {
+    let odbc_environment = Environment::new().unwrap();
+
+    let connection = odbc_environment
+        .connect_with_connection_string(connection_string, ConnectionOptions::default())?;
+
+    let mut record_batch_iterator = RecordBatchIterator::new(batches.into_iter().map(Ok), schema);
+
+    insert_into_table(
+        &connection,
+        &mut record_batch_iterator,
+        destination_name,
+        batch_size,
+    )?;
+
+    Ok(())
+}
+
+#[cfg(test)]
+mod tests {
+    use datafusion::arrow::array::*;
+    use datafusion::{assert_batches_eq, prelude::*};
+    use std::sync::Arc;
+
+    use super::*;
+
+    #[tokio::test]
+    #[tracing_test::traced_test]
+    async fn test_register_odbc_source_ok() {
+        let connection_string: &str = "\
+            Driver={PostgreSQL Unicode};\
+            Server=localhost;\
+            UID=postgres;\
+            PWD=postgres;\
+        ";
+
+        let ctx = Arc::new(SessionContext::new());
+
+        register_odbc_source(
+            ctx.clone(),
+            connection_string,
+            "SELECT * FROM temp_readings WHERE timestamp::date BETWEEN '2024-01-01' AND '2024-01-31'",
+            "my_table",
+        )
+        .await
+        .unwrap();
+
+        let result = ctx
+            .sql("SELECT count(*) num_rows FROM my_table")
+            .await
+            .unwrap()
+            .collect()
+            .await
+            .unwrap();
+
+        assert_batches_eq!(
+            &[
+                "+----------+",
+                "| num_rows |",
+                "+----------+",
+                "| 1000     |",
+                "+----------+",
+            ],
+            result.as_slice()
+        );
+    }
+
+    #[tokio::test]
+    #[tracing_test::traced_test]
+    async fn test_register_odbc_destination_ok() {
+        let connection_string: &str = "\
+            Driver={PostgreSQL Unicode};\
+            Server=localhost;\
+            UID=postgres;\
+            PWD=postgres;\
+        ";
+
+        let result = register_odbc_destination(connection_string, "temp_readings_empty").await;
+
+        assert!(result.is_ok());
+    }
+
+    #[tokio::test]
+    #[tracing_test::traced_test]
+    async fn test_write_arrow_batches_ok() {
+        let connection_string: &str = "\
+            Driver={PostgreSQL Unicode};\
+            Server=localhost;\
+            UID=postgres;\
+            PWD=postgres;\
+        ";
+
+        let locations = (0..1000).collect::<Vec<i32>>();
+        let timestamps = (1704067200..1704068200).collect::<Vec<i64>>();
+        let temperatures = (0..1000).map(|i| i as f64).collect::<Vec<f64>>();
+        let humidity = (0..1000).map(|i| i as f64).collect::<Vec<f64>>();
+        let conditions = (0..1000)
+            .map(|i| format!("CONDITION_{i}"))
+            .collect::<Vec<String>>();
+
+        let a: ArrayRef = Arc::new(Int32Array::from(locations));
+        let b: ArrayRef = Arc::new(TimestampSecondArray::from(timestamps));
+        let c: ArrayRef = Arc::new(Float64Array::from(temperatures));
+        let d: ArrayRef = Arc::new(Float64Array::from(humidity));
+        let e: ArrayRef = Arc::new(StringArray::from(conditions));
+
+        let record_batch = RecordBatch::try_from_iter(vec![
+            ("location_id", a),
+            ("timestamp", b),
+            ("temperature_c", c),
+            ("humidity", d),
+            ("weather_condition", e),
+        ])
+        .unwrap();
+        let schema = record_batch.schema();
+
+        let result = write_arrow_batches(
+            connection_string,
+            "temp_readings_empty",
+            vec![record_batch],
+            schema,
+            100,
+        )
+        .await;
+
+        assert!(result.is_ok());
+    }
+}
diff --git a/aqueducts/utils/Cargo.toml b/aqueducts/utils/Cargo.toml
index e1fd546..63addd8 100644
--- a/aqueducts/utils/Cargo.toml
+++ b/aqueducts/utils/Cargo.toml
@@ -16,7 +16,6 @@ default = []
 s3 = ["deltalake/s3"]
 gcs = ["deltalake/gcs"]
 azure = ["deltalake/azure"]
-odbc = ["dep:arrow-odbc"]
 
 [dependencies]
 datafusion.workspace = true
@@ -24,9 +23,7 @@ deltalake.workspace = true
 url.workspace = true
 serde.workspace = true
 thiserror.workspace = true
-
-arrow-odbc = { version = "11.1.0", optional = true }
-tracing = "0.1"
+tracing.workspace = true
 
 [dev-dependencies]
 tokio = { version = "1", features = ["full"] }
diff --git a/aqueducts/utils/src/lib.rs b/aqueducts/utils/src/lib.rs
index 1610a52..9a6c093 100644
--- a/aqueducts/utils/src/lib.rs
+++ b/aqueducts/utils/src/lib.rs
@@ -122,124 +122,3 @@ pub mod store {
         }
     }
 }
-
-/// odbc functionality
-#[cfg(feature = "odbc")]
-pub mod odbc {
-    use std::sync::Arc;
-
-    use arrow_odbc::odbc_api::{ConnectionOptions, Environment};
-    use arrow_odbc::OdbcReaderBuilder;
-    use datafusion::arrow::datatypes::Schema;
-    use datafusion::arrow::{array::RecordBatch, error::ArrowError};
-    use datafusion::execution::context::SessionContext;
-
-    /// Register a table via ODBC using [arrow-odbc](https://docs.rs/arrow-odbc)
-    /// ```rust,ignore
-    /// use datafusion::prelude::SessionContext;
-    /// use arrow_odbc::odbc_api::ConnectionOptions;
-    ///
-    /// let connection_string: &str = "\
-    ///     Driver={PostgreSQL Unicode};\
-    ///     Server=localhost;\
-    ///     UID=postgres;\
-    ///     PWD=postgres;\
-    /// ";
-    ///
-    /// // query to request data from the ODBC source
-    /// // make sure to constrain this query to a dataset that is manageble in memory
-    /// let query = "SELECT * FROM my_table WHERE date > '2024-01-01'";
-    ///
-    /// let ctx = SessionContext::new();
-    ///
-    /// register_odbc_source(&ctx, query, connection_string, "my_table_name").await.unwrap();
-    ///
-    /// let df = ctx.sql("SELECT * FROM my_table_name").await.unwrap();
-    /// df.show().await.unwrap();
-    /// ```
-    pub async fn register_odbc_source(
-        ctx: Arc<SessionContext>,
-        connection_string: &str,
-        query: &str,
-        source_name: &str,
-    ) -> Result<(), ArrowError> {
-        let odbc_environment = Environment::new().unwrap();
-
-        let connection = odbc_environment
-            .connect_with_connection_string(connection_string, ConnectionOptions::default())
-            .expect("failed to connect to ODBC source");
-
-        let parameters = ();
-
-        let cursor = connection
-            .execute(query, parameters)
-            .expect("failed to execute SQL statement statement")
-            .expect("SELECT statement must produce a cursor");
-
-        let reader = OdbcReaderBuilder::new()
-            .with_max_bytes_per_batch(256 * 1024 * 1024)
-            .build(cursor)
-            .expect("failed to build OdbcReaderBuilder");
-
-        let batches = reader
-            .into_iter()
-            .collect::<Result<Vec<RecordBatch>, ArrowError>>()?;
-
-        let df = ctx.read_batches(batches)?;
-        let schema: Arc<Schema> = Arc::new(df.schema().into());
-        let batch = deltalake::arrow::compute::concat_batches(&schema, df.collect().await?.iter())?;
-
-        ctx.register_batch(source_name, batch)?;
-
-        Ok(())
-    }
-}
-
-#[cfg(all(test, feature = "odbc"))]
-mod odbc_tests {
-    use super::odbc::*;
-
-    use datafusion::{assert_batches_eq, prelude::*};
-    use std::sync::Arc;
-
-    #[tokio::test]
-    #[tracing_test::traced_test]
-    async fn test_register_odbc_source_ok() {
-        let connection_string: &str = "\
-            Driver={PostgreSQL Unicode};\
-            Server=localhost;\
-            UID=postgres;\
-            PWD=postgres;\
-        ";
-
-        let ctx = Arc::new(SessionContext::new());
-
-        register_odbc_source(
-            ctx.clone(),
-            connection_string,
-            "SELECT * FROM temp_readings WHERE timestamp::date BETWEEN '2024-01-01' AND '2024-01-31'",
-            "my_table",
-        )
-        .await
-        .unwrap();
-
-        let result = ctx
-            .sql("SELECT count(*) num_rows FROM my_table")
-            .await
-            .unwrap()
-            .collect()
-            .await
-            .unwrap();
-
-        assert_batches_eq!(
-            &[
-                "+----------+",
-                "| num_rows |",
-                "+----------+",
-                "| 1000     |",
-                "+----------+",
-            ],
-            result.as_slice()
-        );
-    }
-}
diff --git a/db/init.sql b/db/init.sql
index a794469..bba40b3 100644
--- a/db/init.sql
+++ b/db/init.sql
@@ -1,10 +1,29 @@
 CREATE TABLE temp_readings (
-    location_id INTEGER,
-    timestamp TIMESTAMP,
-    temperature_c FLOAT,
-    humidity FLOAT,
+    location_id       INTEGER,
+    timestamp         TIMESTAMP,
+    temperature_c     FLOAT,
+    humidity          FLOAT,
     weather_condition VARCHAR(50)
 );
 
-COPY temp_readings FROM '/data/temp_readings_jan_2024.csv' DELIMITER ',' CSV HEADER;
-COPY temp_readings FROM '/data/temp_readings_feb_2024.csv' DELIMITER ',' CSV HEADER;
+CREATE TABLE temp_readings_empty (
+    location_id       INTEGER,
+    timestamp         TIMESTAMP,
+    temperature_c     FLOAT,
+    humidity          FLOAT,
+    weather_condition VARCHAR(50)
+);
+
+CREATE TABLE temp_readings_aggregated (
+    date            DATE,
+    location_id     INTEGER,
+    min_temp_c      FLOAT,
+    min_humidity    FLOAT,
+    max_temp_c      FLOAT,
+    max_humidity    FLOAT,
+    avg_temp_c      FLOAT,
+    avg_humidity    FLOAT
+);
+
+COPY temp_readings FROM '/opt/temp_readings_jan_2024.csv' DELIMITER ',' CSV HEADER;
+COPY temp_readings FROM '/opt/temp_readings_feb_2024.csv' DELIMITER ',' CSV HEADER;
diff --git a/docker-compose.yml b/docker-compose.yml
index 5af974d..59aff83 100644
--- a/docker-compose.yml
+++ b/docker-compose.yml
@@ -12,5 +12,5 @@ services:
       - 5432:5432
     volumes:
       - ./db/init.sql:/docker-entrypoint-initdb.d/init.sql
-      - ./examples/temp_readings_jan_2024.csv:/data/temp_readings_jan_2024.csv
-      - ./examples/temp_readings_feb_2024.csv:/data/temp_readings_feb_2024.csv
+      - ./examples/temp_readings_jan_2024.csv:/opt/temp_readings_jan_2024.csv
+      - ./examples/temp_readings_feb_2024.csv:/opt/temp_readings_feb_2024.csv
diff --git a/docs/usage.md b/docs/usage.md
index 95c30b7..bd3d27c 100644
--- a/docs/usage.md
+++ b/docs/usage.md
@@ -209,7 +209,7 @@ Here are some examples on how to use the Aqueducts deserialization schema for YA
         ```yaml
     
         destination:
-          type: file
+          type: File
           name: results
           file_type:
             type: Csv
@@ -221,7 +221,7 @@ Here are some examples on how to use the Aqueducts deserialization schema for YA
 
         ```yaml
         destination:
-          type: delta
+          type: Delta
           name: example_output
           location: ${local_path}/examples/output_delta_example/${run_id}
           storage_options: {}
@@ -229,7 +229,7 @@ Here are some examples on how to use the Aqueducts deserialization schema for YA
 
           write_mode:
             # appends data to the table
-            operation: append
+            operation: Append
 
           # columns by which to partition the table
           partition_cols:
@@ -259,7 +259,7 @@ Here are some examples on how to use the Aqueducts deserialization schema for YA
 
         ```yaml
         destination:
-          type: delta
+          type: Delta
           name: example_output
           location: ${local_path}/examples/output_delta_example/${run_id}
           storage_options: {}
@@ -267,7 +267,7 @@ Here are some examples on how to use the Aqueducts deserialization schema for YA
 
           write_mode:
             # upserts using the date as the "primary" key
-            operation: upsert
+            operation: Upsert
             params: 
               - date
 
@@ -299,7 +299,7 @@ Here are some examples on how to use the Aqueducts deserialization schema for YA
 
         ```yaml
         destination:
-          type: delta
+          type: Delta
           name: example_output
           location: ${local_path}/examples/output_delta_example/${run_id}
           storage_options: {}
@@ -307,7 +307,7 @@ Here are some examples on how to use the Aqueducts deserialization schema for YA
 
           write_mode:
             # replaces using the date column to delete all data for that date
-            operation: replace
+            operation: Replace
             params: 
               - column: date
                 value: '2024-01-01'
@@ -335,3 +335,13 @@ Here are some examples on how to use the Aqueducts deserialization schema for YA
               nullable: true
               metadata: {}
         ```
+
+    === "ODBC destination"
+
+        ```yaml
+        destination:
+          type: Odbc
+          name: temp_readings_aggregated
+          connection_string: Driver={PostgreSQL Unicode};Server=localhost;UID=${user};PWD=${pass};
+          batch_size: 100
+        ```
diff --git a/examples/aqueduct_pipeline_odbc.yml b/examples/aqueduct_pipeline_odbc.yml
index f886abc..9dc4be5 100644
--- a/examples/aqueduct_pipeline_odbc.yml
+++ b/examples/aqueduct_pipeline_odbc.yml
@@ -44,7 +44,12 @@ stages:
       # print the entire result for this query to stdout
       show: 0
 
-  - name: union
-    query: >
-      SELECT count(*) FROM jan_aggregated
-    show: 0
+  - - name: union
+      query: >
+        SELECT * FROM jan_aggregated UNION ALL SELECT * FROM feb_aggregated
+
+destination:
+  type: Odbc
+  name: temp_readings_aggregated
+  connection_string: Driver={PostgreSQL Unicode};Server=localhost;UID=${user};PWD=${pass};
+  batch_size: 100


Test patch: