Added CaseInsensitiveSchemaDataSourceExec physical optimizer rule (#74)

* merge

* Fix typo

* CaseInsensitiveSchema optimizer rule

* CaseInsensitiveSchema optimizer rule

* Remove case sensitive table provider

* Fix tests

Author: osipovartem

Cargo.lock

@@ -44,15 +44,15 @@ datafusion-expr = { version = "50.0.0" }
 datafusion-functions-json = { git = "https://github.com/Embucket/datafusion-functions-json.git", rev = "439cbd2282504c3ffaf262f1ffdb530a0fb1a151" }
 datafusion-macros = { version = "50.0.0" }
 datafusion-physical-plan = { version = "50.0.0" }
-datafusion_iceberg = { git = "https://github.com/Embucket/iceberg-rust.git", rev = "b61b904066672f4753eea1105a1e7c0e760785be" }
+datafusion_iceberg = { git = "https://github.com/Embucket/iceberg-rust.git", rev = "df5ad5825bdb7c65fa3d5ad420e025dfbb51afff" }
 futures = { version = "0.3" }
 http = "1.2"
 http-body-util = "0.1.0"
 iceberg = { git = "https://github.com/apache/iceberg-rust.git", rev="7a5ad1fcaf00d4638857812bab788105f6c60573"}
-iceberg-rest-catalog = { git = "https://github.com/Embucket/iceberg-rust.git", rev = "b61b904066672f4753eea1105a1e7c0e760785be" }
-iceberg-rust = { git = "https://github.com/Embucket/iceberg-rust.git", rev = "b61b904066672f4753eea1105a1e7c0e760785be" }
-iceberg-rust-spec = { git = "https://github.com/Embucket/iceberg-rust.git", rev = "b61b904066672f4753eea1105a1e7c0e760785be" }
-iceberg-s3tables-catalog = { git = "https://github.com/Embucket/iceberg-rust.git", rev = "b61b904066672f4753eea1105a1e7c0e760785be" }
+iceberg-rest-catalog = { git = "https://github.com/Embucket/iceberg-rust.git", rev = "df5ad5825bdb7c65fa3d5ad420e025dfbb51afff" }
+iceberg-rust = { git = "https://github.com/Embucket/iceberg-rust.git", rev = "df5ad5825bdb7c65fa3d5ad420e025dfbb51afff" }
+iceberg-rust-spec = { git = "https://github.com/Embucket/iceberg-rust.git", rev = "df5ad5825bdb7c65fa3d5ad420e025dfbb51afff" }
+iceberg-s3tables-catalog = { git = "https://github.com/Embucket/iceberg-rust.git", rev = "df5ad5825bdb7c65fa3d5ad420e025dfbb51afff" }
 indexmap = "2.7.1"
 jsonwebtoken = "9.3.1"
 lazy_static = { version = "1.5" }

Cargo.toml

@@ -1,4 +1,3 @@
-use crate::snowflake_table::CaseInsensitiveTable;
 use crate::{block_in_new_runtime, error};
 use async_trait::async_trait;
 use catalog_metastore::error as metastore_error;
@@ -95,10 +94,8 @@ impl SchemaProvider for EmbucketSchema {
                 .await
                 .map_err(|e| DataFusionError::External(Box::new(e)))?;
                 let tabular = IcebergTabular::Table(iceberg_table);
-
-                let table_provider: Arc<dyn TableProvider> = Arc::new(CaseInsensitiveTable::new(
-                    Arc::new(IcebergDataFusionTable::new(tabular, None, None, None)),
-                ));
+                let table_provider: Arc<dyn TableProvider> =
+                    Arc::new(IcebergDataFusionTable::new(tabular, None, None, None));
                 Ok(Some(table_provider))
             }
             Ok(None) => Ok(None),

crates/catalog/src/catalogs/embucket/schema.rs

@@ -10,7 +10,6 @@ pub mod df_error;
 pub mod error;
 pub mod information_schema;
 pub mod schema;
-pub mod snowflake_table;
 pub mod table;
 pub mod utils;
 

crates/catalog/src/lib.rs

@@ -1,4 +1,3 @@
-use crate::snowflake_table::CaseInsensitiveTable;
 use crate::table::{CachingTable, IcebergTableBuilder};
 use crate::{block_in_new_runtime, error};
 use async_trait::async_trait;
@@ -116,9 +115,8 @@ impl SchemaProvider for CachingSchema {
                     .await
                     .context(error::IcebergSnafu)?;
                 let tabular = IcebergTabular::Table(iceberg_table);
-                let table_provider: Arc<dyn TableProvider> = Arc::new(CaseInsensitiveTable::new(
-                    Arc::new(DataFusionTable::new(tabular, None, None, None)),
-                ));
+                let table_provider: Arc<dyn TableProvider> =
+                    Arc::new(DataFusionTable::new(tabular, None, None, None));
                 Ok(table_provider)
             })
             .map_err(|err| DataFusionError::External(Box::new(err)))?

crates/catalog/src/schema.rs

@@ -1,3 +1,4 @@
+use datafusion::arrow::datatypes::Schema;
 use datafusion::catalog::{SchemaProvider, TableProvider};
 use datafusion_common::Result as DataFusionResult;
 use futures::stream::{self, StreamExt};
@@ -47,3 +48,17 @@ pub async fn fetch_table_providers(
 
     Ok(tables)
 }
+
+#[must_use]
+pub fn normalize_schema_case(schema: &Schema) -> Schema {
+    let fields = schema
+        .fields()
+        .iter()
+        .map(|field| {
+            let mut cloned = field.as_ref().clone();
+            cloned.set_name(field.name().to_ascii_lowercase());
+            cloned
+        })
+        .collect::<Vec<_>>();
+    Schema::new(fields)
+}