feat: Translate Hadoop S3A configurations to object_store configurations (apache#1817)

Author: Kontinuation

native/Cargo.lock

@@ -51,6 +51,8 @@ regex = "1.9.6"
 thiserror = "2"
 object_store = { version = "0.12.0", features = ["gcp", "azure", "aws", "http"] }
 url = "2.2"
+aws-config = "1.6.3"
+aws-credential-types = "1.2.3"
 
 [profile.release]
 debug = true

native/Cargo.toml

@@ -69,6 +69,8 @@ datafusion-comet-spark-expr = { workspace = true }
 datafusion-comet-proto = { workspace = true }
 object_store = { workspace = true }
 url = { workspace = true }
+aws-config = { workspace = true }
+aws-credential-types = { workspace = true }
 parking_lot = "0.12.3"
 datafusion-comet-objectstore-hdfs = { path = "../hdfs", optional = true, default-features = false, features = ["hdfs"] }
 

native/core/Cargo.toml

@@ -69,7 +69,7 @@ use datafusion_comet_spark_expr::{create_comet_physical_fun, create_negate_expr,
 use crate::execution::operators::ExecutionError::GeneralError;
 use crate::execution::shuffle::CompressionCodec;
 use crate::execution::spark_plan::SparkPlan;
-use crate::parquet::parquet_support::prepare_object_store;
+use crate::parquet::parquet_support::prepare_object_store_with_configs;
 use datafusion::common::scalar::ScalarStructBuilder;
 use datafusion::common::{
     tree_node::{Transformed, TransformedResult, TreeNode, TreeNodeRecursion, TreeNodeRewriter},
@@ -1156,8 +1156,17 @@ impl PhysicalPlanner {
                     .and_then(|f| f.partitioned_file.first())
                     .map(|f| f.file_path.clone())
                     .ok_or(GeneralError("Failed to locate file".to_string()))?;
-                let (object_store_url, _) =
-                    prepare_object_store(self.session_ctx.runtime_env(), one_file)?;
+
+                let object_store_options: HashMap<String, String> = scan
+                    .object_store_options
+                    .iter()
+                    .map(|(k, v)| (k.clone(), v.clone()))
+                    .collect();
+                let (object_store_url, _) = prepare_object_store_with_configs(
+                    self.session_ctx.runtime_env(),
+                    one_file,
+                    &object_store_options,
+                )?;
 
                 // Generate file groups
                 let mut file_groups: Vec<Vec<PartitionedFile>> =

native/core/src/execution/planner.rs

@@ -26,6 +26,8 @@ pub mod parquet_support;
 pub mod read;
 pub mod schema_adapter;
 
+mod objectstore;
+
 use std::task::Poll;
 use std::{boxed::Box, ptr::NonNull, sync::Arc};
 

native/core/src/parquet/mod.rs

@@ -0,0 +1,18 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+pub mod s3;