Support reading data from S3 using native_datafusion Parquet scanner

[Kontinuation]

What is the problem the feature request solves?

The native_datafusion parquet scanner throws the following error when reading parquet files on S3:

org.apache.comet.CometNativeException: External: Generic S3 error: Error performing PUT http://169.254.169.254/latest/api/token in 1.79000475s, after 10 retries, max_retries: 10, retry_timeout: 180s  - HTTP error: error sending request
	at org.apache.comet.Native.executePlan(Native Method)
	at org.apache.comet.CometExecIterator.$anonfun$getNextBatch$2(CometExecIterator.scala:150)
	at org.apache.comet.CometExecIterator.$anonfun$getNextBatch$2$adapted(CometExecIterator.scala:149)
	at org.apache.comet.vector.NativeUtil.getNextBatch(NativeUtil.scala:157)
	at org.apache.comet.CometExecIterator.$anonfun$getNextBatch$1(CometExecIterator.scala:149)
	at org.apache.comet.Tracing$.withTrace(Tracing.scala:31)
	at org.apache.comet.CometExecIterator.getNextBatch(CometExecIterator.scala:147)
	at org.apache.comet.CometExecIterator.hasNext(CometExecIterator.scala:170)
	at scala.collection.Iterator$$anon$11.hasNext(Iterator.scala:491)
	at scala.collection.Iterator$$anon$10.hasNext(Iterator.scala:460)
	at org.apache.spark.sql.execution.SparkPlan.$anonfun$getByteArrayRdd$1(SparkPlan.scala:388)
	at org.apache.spark.rdd.RDD.$anonfun$mapPartitionsInternal$2(RDD.scala:893)
	at org.apache.spark.rdd.RDD.$anonfun$mapPartitionsInternal$2$adapted(RDD.scala:893)
	at org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:52)
	at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:367)
	at org.apache.spark.rdd.RDD.iterator(RDD.scala:331)
	at org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:93)
	at org.apache.spark.TaskContext.runTaskWithListeners(TaskContext.scala:166)
	at org.apache.spark.scheduler.Task.run(Task.scala:141)
	at org.apache.spark.executor.Executor$TaskRunner.$anonfun$run$4(Executor.scala:620)
	at org.apache.spark.util.SparkErrorUtils.tryWithSafeFinally(SparkErrorUtils.scala:64)
	at org.apache.spark.util.SparkErrorUtils.tryWithSafeFinally$(SparkErrorUtils.scala:61)
	at org.apache.spark.util.Utils$.tryWithSafeFinally(Utils.scala:94)
	at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:623)
	at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1136)
	at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:635)
	at java.base/java.lang.Thread.run(Thread.java:840)

The error message suggests that the S3 client is retrieving token on EC2 instance, even though I am not running comet on EC2.

It is also quite common to configure credentials for accessing AWS S3 by specifying Hadoop configurations, for instance:

spark = (SparkSession.builder
  .config("spark.hadoop.fs.s3a.aws.credentials.provider", "org.apache.hadoop.fs.s3a.AnonymousAWSCredentialsProvider")
  .getOrCreate())

Currently there's no mechanism for using credentials configured as Hadoop S3 configurations when reading S3 using the native datafusion parquet reader. We'd better build some compatibility layer to recognize Hadoop S3 configurations in native parquet reader for compatibility with Vanilla Spark.

Describe the potential solution

No response

Additional context

No response

[Kontinuation]

There are 2 potential approaches:

Approach 1: Translating hadoop configurations to datafusion object store configurations.

We'll use native implementations of DataFusion object stores to access cloud storages, but we have to setup credentials and other configurations using the config values retrieved from Hadoop configuration. For instance, we can construct credential configs for object_store using the values of spark.hadoop.fs.s3a.access.key and spark.hadoop.fs.s3a.secret.key in Hadoop config.

A concrete example of this approach is Apache Gluten's ConfigExtractor. It uses Velox's native storage accessors and maps Hadoop configurations to storage accessors' configurations.

This approach does not guarantee 100% compatibility with Vanilla Spark, as not all configurations are supported by Rust's object store implementations and translations for some Hadoop configurations are not possible.

Approach 2: Call Hadoop FileSystem API using JNI.

This approach provides better compatibility with Vanilla Spark, as we are using the original Hadoop FileSystem API implementations for accessing cloud storages. However, this approach adds additional JNI interface and has higher overall complexity.

I suggest that we implement approach 1 first as it is straightforward and works for many useful use cases, then considering approach 2 for cases that approach 1 cannot handle.

[parthchandra]

I've been thinking of Approach 2 as the way to go but not perhaps for the entire HDFS api. I haven't thought it all the way thru but I will take a look at it once some of the major complex type issues are addressed.
Meanwhile if you want to go ahead with Approach 1 then at least users wishing to access S3 will get unblocked somewhat.