[SPARK-27194][SPARK-29302][SQL] Define a spark staging committer to resolve FileAlreadyExistingException

[turboFei]

What changes were proposed in this pull request?

For dynamic partition overwrite, its working dir is .spark-staging-{jobId}.
Task file name formatted part-$taskId-$jobId$ext(regardless task attempt Id).
Each task writes its output to:

• .spark-staging-{jobId}/partitionPath1/taskFileName1
• .spark-staging-{jobId}/partitionPath2/taskFileName2
• ...
• .spark-staging-{jobId}/partitionPathN/taskFileNameN

If speculation is enabled, there may be several tasks, which have same taskId and different attemptId, write to the same files concurrently.
For distributedFileSystem, it only allow one task to hold the lease to write a file, if two tasks want to write the same file, an exception like no lease on inode would be thrown.

Even speculation is not enabled, if a task aborted due to Executor OOM, its output would not be cleaned up.
Then a new task launched to write the same file, because parquet disallows overwriting, a FileAlreadyExistsException would be thrown, like.

Caused by: org.apache.hadoop.fs.FileAlreadyExistsException: /user/hive/warehouse/t2/.spark-staging-1f1efbfd-7e20-4e0f-a49c-a7fa3eae4cb1/part1=2/part2=2/part-00000-1f1efbfd-7e20-4e0f-a49c-a7fa3eae4cb1.c000.snappy.parquet for client 127.0.0.1 already exists
at org.apache.hadoop.hdfs.server.namenode.FSNamesystem.startFileInternal(FSNamesystem.java:2578)
at org.apache.hadoop.hdfs.server.namenode.FSNamesystem.startFileInt(FSNamesystem.java:2465)
at org.apache.hadoop.hdfs.server.namenode.FSNamesystem.startFile(FSNamesystem.java:2349)
at org.apache.hadoop.hdfs.server.namenode.NameNodeRpcServer.create(NameNodeRpcServer.java:624)
at org.apache.hadoop.hdfs.protocolPB.ClientNamenodeProtocolServerSideTranslatorPB.create(ClientNamenodeProtocolServerSideTranslatorPB.java:398)

It is a critical issue and would cause job failed.

In this Pr, we define a spark staging output committer to fix this issue:

1. set a working path under staging dir named partitionPath-attemptId.
2. after task completed, rename partitionPath-attemptId/fileName to partitionPath/fileName
3. leverage the OutputCommitCoordinator to coordinate the task commits

Why are the changes needed?

Without this PR, dynamic partition overwrite operation might fail.

Does this PR introduce any user-facing change?

No.

How was this patch tested?

Added UT.

[turboFei]

gentle ping @cloud-fan

Hi, we found a new solution to fix the issues when dynamic partition overwrite is enabled.

1. FileAlreadyExsitException when executor was crash
2. task conflicts with its speculation task.

In this PR, we define a new type of OutputCommitter and leverage the OutputCommitCoordinator to coordinate the task commits.
Could you kindly give some suggestions?

[turboFei]

aslo cc @Ngone51

[turboFei]

gentle ping @cloud-fan@dongjoon-hyun@xuanyuanking@WinkerDu@LuciferYang@advancedxy

[dongjoon-hyun]

ok to test

[SparkQA]

Test build #125132 has started for PR 28989 at commit b09a665.

[shaneknapp]

test this please

[SparkQA]

Test build #125143 has finished for PR 28989 at commit b09a665.

• This patch fails Spark unit tests.
• This patch merges cleanly.
• This patch adds the following public classes (experimental):
• class SparkStagingOutputCommitter(

[turboFei]

will try to fix it.

[SparkQA]

Test build #125196 has finished for PR 28989 at commit 0492977.

• This patch fails Spark unit tests.
• This patch merges cleanly.
• This patch adds the following public classes (experimental):
• class SparkStagingOutputCommitter(

[SparkQA]

Test build #125205 has finished for PR 28989 at commit bc7d2b5.

• This patch fails Spark unit tests.
• This patch merges cleanly.
• This patch adds no public classes.

[SparkQA]

Test build #125212 has finished for PR 28989 at commit a7a8d4b.

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

[SparkQA]

Test build #125275 has finished for PR 28989 at commit 0d722a9.

• This patch fails to generate documentation.
• This patch merges cleanly.
• This patch adds no public classes.

[turboFei]

close this,
I am wrong about that, I thought the taskAttemptContext.getTaskAttemptId.getId is same with the taskAttemptId of spark and it would at most create several(the largest task attempt number) staging partition dir.
But the taskAttemptContext.getTaskAttemptId.getId is also a uniq id, so this would also create multi staging partition dir for each task.

Prefer to #29000