chore: upgrade datafusion and and some other crates (#219)

* chore: upgrade datafusion and and some other crates

* Exclude odbc from PR checks

* Install ODBC & sqlite packages

---------

Co-authored-by: Phillip LeBlanc <phillip@leblanc.tech>

Author: xhwhis

.github/workflows/pr.yaml

@@ -85,7 +85,7 @@ jobs:
             google-cloud-sdk imagemagick \
             libmagickcore-dev libmagickwand-dev libmagic-dev ant ant-optional kubectl \
             mercurial apt-transport-https mono-complete libmysqlclient \
-            unixodbc-dev yarn chrpath libssl-dev libxft-dev \
+            yarn chrpath libssl-dev libxft-dev \
             libfreetype6 libfreetype6-dev libfontconfig1 libfontconfig1-dev \
             snmp pollinate libpq-dev postgresql-client powershell ruby-full \
             sphinxsearch subversion mongodb-org azure-cli microsoft-edge-stable \
@@ -100,5 +100,10 @@ jobs:
           echo "some packages purged"
           df -h
 
+      - name: Install ODBC & Sqlite
+        run: |
+          sudo apt-get install -y unixodbc-dev
+          sudo apt-get install -y libsqlite3-dev
+
       - name: Run tests
         run: make test

Cargo.toml

@@ -13,19 +13,19 @@ arrow-flight = { version = "53", optional = true, features = [
   "tls",
 ] }
 arrow-odbc = { version = "=14.0.1", optional = true }
-async-stream = { version = "0.3.6", optional = true }
+async-stream = { version = "0.3", optional = true }
 async-trait = "0.1"
-bb8 = { version = "0.8", optional = true }
-bb8-postgres = { version = "0.8", optional = true }
-bigdecimal = "0.4.6"
+bb8 = { version = "0.9", optional = true }
+bb8-postgres = { version = "0.9", optional = true }
+bigdecimal = "0.4"
 byteorder = "1.5.0"
-chrono = "0.4.38"
+chrono = "0.4"
 dashmap = "6.1.0"
-datafusion = { version = "43", default-features = false }
-datafusion-federation = { version = "0.3.1", features = [
+datafusion = { version = "44", default-features = false }
+datafusion-federation = { version = "=0.3.4", features = [
   "sql",
 ], optional = true }
-datafusion-proto = { version = "43", optional = true }
+datafusion-proto = { version = "44", optional = true }
 duckdb = { version = "1.1.1", features = [
   "bundled",
   "r2d2",
@@ -38,20 +38,22 @@ fallible-iterator = "0.3.0"
 fundu = "2.0.1"
 futures = "0.3"
 geo-types = "0.7"
-itertools = "0.13.0"
-mysql_async = { version = "0.34", features = [
+itertools = "0.14.0"
+mysql_async = { version = "0.35", features = [
   "native-tls-tls",
   "chrono",
+  "time",
+  "bigdecimal",
 ], optional = true }
-native-tls = { version = "0.2.12", optional = true }
+native-tls = { version = "0.2", optional = true }
 num-bigint = "0.4"
-odbc-api = { version = "10.0.0", optional = true }
+odbc-api = { version = "10.1", optional = true }
 pem = { version = "3.0.4", optional = true }
 postgres-native-tls = { version = "0.5.0", optional = true }
 prost = { version = "0.13", optional = true }
-r2d2 = { version = "0.8.10", optional = true }
-rusqlite = { version = "0.32.1", optional = true }
-sea-query = { version = "0.32.0", features = [
+r2d2 = { version = "0.8", optional = true }
+rusqlite = { version = "0.32", optional = true }
+sea-query = { version = "0.32", features = [
   "backend-sqlite",
   "backend-postgres",
   "postgres-array",
@@ -63,11 +65,11 @@ sea-query = { version = "0.32.0", features = [
 secrecy = "0.8.0"
 serde = { version = "1.0", optional = true }
 serde_json = "1.0"
-sha2 = "0.10.8"
-snafu = "0.8.5"
-time = "0.3.36"
-tokio = { version = "1.41", features = ["macros", "fs"] }
-tokio-postgres = { version = "0.7.12", features = [
+sha2 = "0.10"
+snafu = "0.8"
+time = "0.3"
+tokio = { version = "1.43", features = ["macros", "fs"] }
+tokio-postgres = { version = "0.7", features = [
   "with-chrono-0_4",
   "with-uuid-1",
   "with-serde_json-1",
@@ -78,23 +80,23 @@ tonic = { version = "0.12", optional = true, features = [
   "tls-native-roots",
   "tls-webpki-roots",
 ] }
-tracing = "0.1.40"
+tracing = "0.1"
 trust-dns-resolver = "0.23.2"
 url = "2.5.4"
-uuid = { version = "1.11.0", optional = true }
+uuid = { version = "1.11", optional = true }
 
 [dev-dependencies]
 anyhow = "1.0"
 bollard = "0.18.1"
 geozero = { version = "0.14.0", features = ["with-wkb"] }
-insta = { version = "1.41.1", features = ["filters"] }
+insta = { version = "1.42.0", features = ["filters"] }
 prost = { version = "0.13" }
-rand = "0.8.5"
-reqwest = "0.12.9"
-rstest = "0.23.0"
-test-log = { version = "0.2.16", features = ["trace"] }
-tokio-stream = { version = "0.1.16", features = ["net"] }
-tracing-subscriber = { version = "0.3.18", features = ["env-filter"] }
+rand = "0.8"
+reqwest = "0.12"
+rstest = "0.24.0"
+test-log = { version = "0.2", features = ["trace"] }
+tokio-stream = { version = "0.1", features = ["net"] }
+tracing-subscriber = { version = "0.3", features = ["env-filter"] }
 
 [features]
 duckdb = [
@@ -130,3 +132,8 @@ postgres = [
 postgres-federation = ["postgres", "federation"]
 sqlite = ["dep:rusqlite", "dep:tokio-rusqlite"]
 sqlite-federation = ["sqlite", "federation"]
+
+[[example]]
+name = "odbc_sqlite"
+path = "examples/odbc_sqlite.rs"
+required-features = ["sqlite", "odbc"]

Makefile

@@ -3,7 +3,7 @@ all:
 
 .PHONY: test
 test:
-	cargo test --all-features
+	cargo test --features duckdb-federation,flight,mysql-federation,postgres-federation,sqlite-federation
 
 .PHONY: lint
 lint:

src/flight/exec.rs

@@ -36,10 +36,9 @@ use datafusion::common::Result;
 use datafusion::common::{project_schema, DataFusionError};
 use datafusion::execution::{SendableRecordBatchStream, TaskContext};
 use datafusion::physical_expr::{EquivalenceProperties, Partitioning};
+use datafusion::physical_plan::execution_plan::{Boundedness, EmissionType};
 use datafusion::physical_plan::stream::RecordBatchStreamAdapter;
-use datafusion::physical_plan::{
-    DisplayAs, DisplayFormatType, ExecutionMode, ExecutionPlan, PlanProperties,
-};
+use datafusion::physical_plan::{DisplayAs, DisplayFormatType, ExecutionPlan, PlanProperties};
 use futures::{StreamExt, TryStreamExt};
 use serde::{Deserialize, Serialize};
 use tonic::metadata::{AsciiMetadataKey, MetadataMap};
@@ -84,13 +83,16 @@ impl FlightExec {
 impl From<FlightConfig> for FlightExec {
     fn from(config: FlightConfig) -> Self {
         let exec_mode = if config.properties.unbounded_streams {
-            ExecutionMode::Unbounded
+            Boundedness::Unbounded {
+                requires_infinite_memory: false,
+            }
         } else {
-            ExecutionMode::Bounded
+            Boundedness::Bounded
         };
         let plan_properties = PlanProperties::new(
             EquivalenceProperties::new(config.schema.clone()),
             Partitioning::UnknownPartitioning(config.partitions.len()),
+            EmissionType::Incremental,
             exec_mode,
         );
         let mut mm = MetadataMap::new();

src/sql/arrow_sql_gen/mysql.rs

@@ -568,7 +568,8 @@ pub fn map_column_to_data_type(
         | ColumnType::MYSQL_TYPE_LONG_BLOB
         | ColumnType::MYSQL_TYPE_TINY_BLOB
         | ColumnType::MYSQL_TYPE_MEDIUM_BLOB
-        | ColumnType::MYSQL_TYPE_GEOMETRY => {
+        | ColumnType::MYSQL_TYPE_GEOMETRY
+        | ColumnType::MYSQL_TYPE_VECTOR => {
             unimplemented!("Unsupported column type {:?}", column_type)
         }
     }

src/sql/sql_provider_datafusion/mod.rs

@@ -12,6 +12,7 @@ use crate::sql::db_connection_pool::{
 use async_trait::async_trait;
 use datafusion::{
     catalog::Session,
+    physical_plan::execution_plan::{Boundedness, EmissionType},
     sql::unparser::dialect::{
         DefaultDialect, Dialect, MySqlDialect, PostgreSqlDialect, SqliteDialect,
     },
@@ -32,8 +33,8 @@ use datafusion::{
     },
     physical_expr::EquivalenceProperties,
     physical_plan::{
-        stream::RecordBatchStreamAdapter, DisplayAs, DisplayFormatType, ExecutionMode,
-        ExecutionPlan, Partitioning, PlanProperties, SendableRecordBatchStream,
+        stream::RecordBatchStreamAdapter, DisplayAs, DisplayFormatType, ExecutionPlan,
+        Partitioning, PlanProperties, SendableRecordBatchStream,
     },
     sql::{unparser::Unparser, TableReference},
 };
@@ -295,7 +296,8 @@ impl<T, P> SqlExec<T, P> {
             properties: PlanProperties::new(
                 EquivalenceProperties::new(projected_schema),
                 Partitioning::UnknownPartitioning(1),
-                ExecutionMode::Bounded,
+                EmissionType::Incremental,
+                Boundedness::Bounded,
             ),
         })
     }

src/sqlite/sqlite_interval.rs

@@ -240,6 +240,7 @@ impl SQLiteIntervalVisitor {
             over: None,
             within_group: Vec::new(),
             parameters: ast::FunctionArguments::None,
+            uses_odbc_syntax: false,
         });
 
         Expr::Cast {
@@ -409,6 +410,7 @@ mod test {
                 over: None,
                 within_group: Vec::new(),
                 parameters: ast::FunctionArguments::None,
+                uses_odbc_syntax: false,
             })),
             data_type: ast::DataType::Text,
             format: None,
@@ -458,6 +460,7 @@ mod test {
                 over: None,
                 within_group: Vec::new(),
                 parameters: ast::FunctionArguments::None,
+                uses_odbc_syntax: false,
             })),
             data_type: ast::DataType::Text,
             format: None,

src/util/test.rs

@@ -1,6 +1,7 @@
 use std::{any::Any, sync::Arc};
 
 use datafusion::arrow::{array::RecordBatch, datatypes::SchemaRef};
+use datafusion::physical_plan::execution_plan::{Boundedness, EmissionType};
 use datafusion::{
     common::Statistics,
     error::{DataFusionError, Result},
@@ -9,7 +10,7 @@ use datafusion::{
     physical_plan::{
         common,
         stream::{RecordBatchReceiverStream, RecordBatchStreamAdapter},
-        DisplayAs, DisplayFormatType, ExecutionMode, ExecutionPlan, Partitioning, PlanProperties,
+        DisplayAs, DisplayFormatType, ExecutionPlan, Partitioning, PlanProperties,
     },
 };
 
@@ -59,7 +60,8 @@ impl MockExec {
         PlanProperties::new(
             eq_properties,
             Partitioning::UnknownPartitioning(1),
-            ExecutionMode::Bounded,
+            EmissionType::Incremental,
+            Boundedness::Bounded,
         )
     }
 }