update datafusion for olympus; update SqlExec schema organized tables

Author: jwhittle933

Cargo.toml

@@ -8,12 +8,15 @@ license = "Apache-2.0"
 description = "Extend the capabilities of DataFusion to support additional data sources via implementations of the `TableProvider` trait."
 
 [dependencies]
-arrow = "52.2.0"
-arrow-array = { version = "52.2.0", optional = true }
-arrow-cast = { version = "52.2.0", optional = true }
-arrow-flight = { version = "52.2.0", optional = true, features = ["flight-sql-experimental", "tls"] }
-arrow-schema = { version = "52.2.0", optional = true, features = ["serde"] }
-arrow-json = "52.2.0"
+arrow = "53.0.0"
+arrow-array = { version = "53.0.0", optional = true }
+arrow-cast = { version = "53.0.0", optional = true }
+arrow-flight = { version = "53.0.0", optional = true, features = [
+  "flight-sql-experimental",
+  "tls",
+] }
+arrow-schema = { version = "53.0.0", optional = true, features = ["serde"] }
+arrow-json = "53.0.0"
 async-stream = { version = "0.3.5", optional = true }
 async-trait = "0.1.80"
 num-bigint = "0.4.4"
@@ -23,12 +26,12 @@ bigdecimal = "0.4.5"
 bigdecimal_0_3_0 = { package = "bigdecimal", version = "0.3.0" }
 byteorder = "1.5.0"
 chrono = "0.4.38"
-datafusion = "41.0.0"
-datafusion-expr = { version = "41.0.0", optional = true }
-datafusion-physical-expr = { version = "41.0.0", optional = true }
-datafusion-physical-plan = { version = "41.0.0", optional = true }
-datafusion-proto = { version = "41.0.0", optional = true }
-datafusion-federation = { version = "0.2.2", features = ["sql"] } 
+datafusion = "42.0.0"
+datafusion-expr = { version = "42.0.0", optional = true }
+datafusion-physical-expr = { version = "42.0.0", optional = true }
+datafusion-physical-plan = { version = "42.0.0", optional = true }
+datafusion-proto = { version = "42.0.0", optional = true }
+datafusion-federation = { version = "0.3.0", features = ["sql"] }
 duckdb = { version = "1", features = [
   "bundled",
   "r2d2",
@@ -38,18 +41,34 @@ duckdb = { version = "1", features = [
 ], optional = true }
 fallible-iterator = "0.3.0"
 futures = "0.3.30"
-mysql_async = { version = "0.34.1", features = ["native-tls-tls", "chrono"], optional = true }
-prost = { version = "0.12" , optional = true } # pinned for arrow-flight compat
+mysql_async = { version = "0.34.1", features = [
+  "native-tls-tls",
+  "chrono",
+], optional = true }
+prost = { version = "0.13.3", optional = true } # pinned for arrow-flight compat
 r2d2 = { version = "0.8.10", optional = true }
 rusqlite = { version = "0.31.0", optional = true }
-sea-query = { version = "0.31.0", features = ["backend-sqlite", "backend-postgres", "postgres-array", "with-rust_decimal", "with-bigdecimal", "with-time", "with-chrono"] }
+sea-query = { version = "0.31.0", 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_json = "1.0.124"
 snafu = "0.8.3"
 time = "0.3.36"
 tokio = { version = "1.38.0", features = ["macros", "fs"] }
-tokio-postgres = { version = "0.7.10", features = ["with-chrono-0_4", "with-uuid-1", "with-serde_json-1", "with-geo-types-0_7"], optional = true }
+tokio-postgres = { version = "0.7.10", features = [
+  "with-chrono-0_4",
+  "with-uuid-1",
+  "with-serde_json-1",
+  "with-geo-types-0_7",
+], optional = true }
 tracing = "0.1.40"
 uuid = { version = "1.9.1", optional = true }
 postgres-native-tls = { version = "0.5.0", optional = true }
@@ -79,7 +98,16 @@ arrow-schema = "52.2.0"
 
 [features]
 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"]
+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"]
 flight = [

src/sql/db_connection_pool/dbconnection/postgresconn.rs

@@ -86,11 +86,9 @@ impl<'a>
         &self,
         table_reference: &TableReference,
     ) -> Result<SchemaRef, super::Error> {
-        let rows = match self
-            .conn
-            .query(&format!("SELECT * FROM {table_reference} LIMIT 1"), &[])
-            .await
-        {
+        let q = &format!("SELECT * FROM {table_reference} LIMIT 1");
+
+        let rows = match self.conn.query(q.as_str(), &[]).await {
             Ok(rows) => rows,
             Err(e) => {
                 if let Some(error_source) = e.source() {

src/sql/sql_provider_datafusion/mod.rs

@@ -311,7 +311,24 @@ impl<T, P> SqlExec<T, P> {
     }
 
     fn table_name_escaped(&self) -> String {
-        self.ident_escaped(&self.table_reference.to_string())
+        match &self.table_reference {
+            TableReference::Bare { table } => self.ident_escaped(&table),
+            TableReference::Partial { schema, table } => format!(
+                "{}.{}",
+                self.ident_escaped(&schema),
+                self.ident_escaped(&table)
+            ),
+            TableReference::Full {
+                catalog,
+                schema,
+                table,
+            } => format!(
+                "{}.{}.{}",
+                self.ident_escaped(&catalog),
+                self.ident_escaped(&schema),
+                self.ident_escaped(&table)
+            ),
+        }
     }
 
     fn column_name_escaped(&self, column_name: &str) -> String {