[SPARK-33094][SQL] Make ORC format propagate Hadoop config from DS options to underlying HDFS file system

[MaxGekk]

What changes were proposed in this pull request?

Propagate ORC options to Hadoop configs in Hive OrcFileFormat and in the regular ORC datasource.

Why are the changes needed?

There is a bug that when running:

spark.read.format("orc").options(conf).load(path)

The underlying file system will not receive the conf options.

Does this PR introduce any user-facing change?

Yes

How was this patch tested?

Added UT to OrcSourceSuite.

[MaxGekk]

This is similar changes to #29971. @HyukjinKwon @yuningzh-db @dongjoon-hyun Please, review this PR.

[MaxGekk]

I ran the test from this PR (just changed format) on Parquet V1/V2, ORC v2, CSV v1/v2, JSON v1/v2 and on Text v1/v2. Everywhere it passed.

[SparkQA]

Kubernetes integration test starting
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/34160/

[SparkQA]

Kubernetes integration test status failure
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/34160/

[MaxGekk]

I have looked at build failures, it seems they are not related to the changes - some failures while downloading artefacts.

[SparkQA]

Test build #129554 has finished for PR 29976 at commit ef6d7f9.

• This patch passes all tests.
• This patch merges cleanly.
• This patch adds no public classes.

[dongjoon-hyun]

+1, LGTM. Thank you, @MaxGekk and @HyukjinKwon .

The K8s IT test failure is irrelevant to this one.

- Test basic decommissioning *** FAILED ***

Merged to master for Apache Spark 3.1.0

[MaxGekk]

@dongjoon-hyun Can this be merged to branch-3.0 (maybe 2.4 too) since it can be considered as a bug fix?
Here is a real use case, a customer tries to read files in Azure Data Lake:

def hadoopConf1() = Map[String, String](
  s"fs.adl.oauth2.access.token.provider.type" -> "ClientCredential",
  s"fs.adl.oauth2.client.id" -> dbutils.secrets.get(scope = "...", key = "..."),
  s"fs.adl.oauth2.credential" -> dbutils.secrets.get(scope = "...", key = "..."),
  s"fs.adl.oauth2.refresh.url" -> s"https://login.microsoftonline.com/.../oauth2/token")
val df = spark.read.format("...").options(hadoopConf1).load("adl://....azuredatalakestore.net/foldersp1/...")

but gets the following exception because the settings above are not propagated to the filesystem:

java.lang.IllegalArgumentException: No value for fs.adl.oauth2.access.token.provider found in conf file.
	at ....adl.AdlFileSystem.getNonEmptyVal(AdlFileSystem.java:820)
	at ....adl.AdlFileSystem.getCustomAccessTokenProvider(AdlFileSystem.java:220)
	at ....adl.AdlFileSystem.getAccessTokenProvider(AdlFileSystem.java:257)
	at ....adl.AdlFileSystem.initialize(AdlFileSystem.java:164)
	at org.apache.hadoop.fs.FileSystem.createFileSystem(FileSystem.java:2669)

[MaxGekk]

@HyukjinKwon You merged similar fix for avro to branch-3.0 in #29971 . WDYT should I open a PR with the changes for branch-3.0?

[HyukjinKwon]

So it can fix a bug right? sure let's open a PR to port back.

[dongjoon-hyun]

+1 for backporting, @MaxGekk and @HyukjinKwon .

[MaxGekk]

Here is the backport to branch-3.0: #29985

[MaxGekk]

Regarding to #29976 (comment) , I could put the test to a common trait and test all built-in datasources including Avro, ORC, LibSVM, CSV and so on. Let me know if you think it makes sense for improving test coverage. cc @gatorsmile @cloud-fan

[MaxGekk]

Here is the PR #30067 with common test.