Fix deltalake s3

Author: matthewmturner

crates/datafusion-app/src/extensions/deltalake.rs

@@ -19,7 +19,11 @@
 
 use crate::config::ExecutionConfig;
 use crate::extensions::{DftSessionStateBuilder, Extension};
-use deltalake::delta_datafusion::DeltaTableFactory;
+use datafusion::catalog::{Session, TableProviderFactory};
+use datafusion::common::DataFusionError;
+use datafusion::logical_expr::logical_plan::CreateExternalTable;
+use deltalake::table::builder::ensure_table_uri;
+use std::collections::HashMap;
 use std::sync::Arc;
 
 #[derive(Debug, Default)]
@@ -31,20 +35,66 @@ impl DeltaLakeExtension {
     }
 }
 
+/// A wrapper around delta-rs's table factory that normalizes option keys passed
+/// through DataFusion's `CREATE EXTERNAL TABLE ... OPTIONS (...)` clause.
+///
+/// DataFusion's `parse_options_map` prepends `"format."` to any option key that
+/// does not contain a dot (e.g. `'aws_endpoint' '...'` becomes `"format.aws_endpoint"`
+/// in `cmd.options`).  delta-rs's `S3ObjectStoreFactory` parses these keys via
+/// `AmazonS3ConfigKey::from_str`, which does not recognise the `"format."` prefix,
+/// so credentials and endpoint settings are silently ignored and the AWS SDK
+/// credential chain is invoked instead.
+///
+/// This factory strips the leading `"format."` prefix before forwarding the options
+/// to `open_table_with_storage_options`, allowing callers to supply S3 credentials
+/// directly in the DDL without needing ambient environment variables.
+#[derive(Debug, Default)]
+struct DeltaTableFactory {}
+
+#[async_trait::async_trait]
+impl TableProviderFactory for DeltaTableFactory {
+    async fn create(
+        &self,
+        _ctx: &dyn Session,
+        cmd: &CreateExternalTable,
+    ) -> datafusion::error::Result<Arc<dyn datafusion::catalog::TableProvider>> {
+        let table_url = ensure_table_uri(&cmd.location)
+            .map_err(|e| DataFusionError::External(Box::new(e)))?;
+
+        let provider = if cmd.options.is_empty() {
+            deltalake::open_table(table_url)
+                .await
+                .map_err(|e| DataFusionError::External(Box::new(e)))?
+        } else {
+            // DataFusion prepends "format." to option keys that don't contain a dot.
+            // Strip that prefix so delta-rs can recognise the keys.
+            let options: HashMap<String, String> = cmd
+                .options
+                .iter()
+                .map(|(k, v)| {
+                    let key = k
+                        .strip_prefix("format.")
+                        .map(str::to_string)
+                        .unwrap_or_else(|| k.clone());
+                    (key, v.clone())
+                })
+                .collect();
+            deltalake::open_table_with_storage_options(table_url, options)
+                .await
+                .map_err(|e| DataFusionError::External(Box::new(e)))?
+        };
+
+        Ok(Arc::new(provider))
+    }
+}
+
 #[async_trait::async_trait]
 impl Extension for DeltaLakeExtension {
     async fn register(
         &self,
         _config: ExecutionConfig,
         builder: &mut DftSessionStateBuilder,
     ) -> datafusion::common::Result<()> {
-        // Register S3 handlers if s3 feature is enabled
-        // This is required for Delta Lake to recognize s3:// URLs
-        #[cfg(feature = "s3")]
-        {
-            deltalake::aws::register_handlers(None);
-        }
-
         builder.add_table_factory("DELTATABLE", Arc::new(DeltaTableFactory {}));
         Ok(())
     }