What changes were proposed in this pull request?

For current data source insert sql commit protocol, it have below problems:
case a: both job A and job B write data into partitioned table TBL with different statistic partition, it will have conflict since they use same temp location ${table_location}/_temporary/0/....,
when job A has finished and then it will clean this temp location, then cause job B’s temp data is cleared. Then it will cause job B to fail to write data.
case b: for current dynamic partition insert, if we kill a job writing data, will remain data under table location in the staging dir under table path.
case c: If we use a dynamic partition insert to insert a new table with a huge number of partitions, we need to move partition path one by one, for this case, we can just rename stagingdir path to table path to make it more quicker. But if we want to do this, we need to make staging dir can be customized and should not use the staging path under table location.

In this approach, we plan to do two thing:

1. Make staging dir can be customized like hive

   • When we terminate a job doing dynamic partition insert, it will remain staging dir under table location . If we make the staging dir can be customized like hive(such as define staging dir as /tmp/spark/.stagingdir ) can avoid remaining such staging dir and data under table path.

   • If we define staging dir using a tmp location, not under table location, when we use dynamic partition insert to write data to a new table, we can just rename staging dir to target table location to avoid moving partition dir one by one. It’s more quickly.

   • If we can customize staging dir, we can implement a new commit protocol to enable use of staging dir to write non-partitioned table and static partition insert use staging dir and won’t increase FS operation.

2. New SQL Commit protocol supports staging dir and won’t increase FS operation.

For current static partition insert in v1, it have step:

1. Task attempts firstly write files under the intermediate path, e.g. /path/to/outputPath/_temporary/{appAttemptId}/_temporary/{taskAttemptId}/{part_spec_path}/xxx.parquet.

2. Then task commit file to /path/to/outputPath/_temporary/{appId}/_temporary/{taskId}/{part_spec_path}/xxx.parquet.

3. Job commit move file to /path/to/outputPath/{part_spec_path}/xxx.parquet.

For current dynamic partition insert in v1, it have step:

1. Task attempts firstly write files under the intermediate path, e.g. /path/to/outputPath/.spark-staging-{jobId}/_temporary/{appAttemptId}/_temporary/{taskAttemptId}/a=1/b=1/xxx.parquet.
2. Then task commit file to /path/to/outputPath/.spark-staging-{jobId}/_temporary/{appId}/_temporary/{taskId}/a=1/b=1/xxx.parquet.

3. During Job commit move file to /path/to/outputPath/a=1/b=1/xxx.parquet.

In this new sql commit protocol SQLPathHadoopMapReduceCommitProtocol ,

for non-partition insert:

1. Task attempts firstly write files under the intermediate path, e.g. {staging_dir}/_temporary/{appAttemptId}/_temporary/{taskAttemptId}/xxx.parquet.
2. Then task commit file to {staging_dir}/_temporary/{appId}/_temporary/{taskId}/xxx.parquet.
3. When job commit, moving file to /path/to/outputPath/xxx.parquet.

for all static partition insert:

1. Task attempts firstly write files under the intermediate path, e.g. {staging_dir}/_temporary/{appAttemptId}/_temporary/{taskAttemptId}/{part_spec_path}/xxx.parquet.
2. Then task commit file to {staging_dir}/_temporary/{appId}/_temporary/{taskId}/{part_spec_path}/xxx.parquet.
3. When job commit, moving file to /path/to/outputPath/{part_spec_path}/xxx.parquet.

for dynamic partition insert, we implement it as:

1. Task attempts firstly write files under the intermediate path, e.g. {staging_dir}/_temporary/{appAttemptId}/_temporary/{taskAttemptId}/{part_spec_path}/xxx.parquet.
2. Then task commit file to {staging_dir}/_temporary/{appId}/_temporary/{taskId}/{part_spec_path}/xxx.parquet.
3. When job commit, move file to /{staging_dir}/{part_spec_path}/xxx.parquet.  Then, if we write to a empty table path, we directly rename   /{staging_dir} to /{table_path}, if not, we move partition dir one by one from  /{staging_dir}/{part_spec_path} to  /{table_location}/{part_spec_path}

The new sql commit protocal’s benefit is:
- Can support Insert into non-partitioned table form it self
- Can support Insert into partition table's statistic partition and read data from target partition.
- Can support Insert into different partition using statistic partition concurrently

These are all normal problems when we use data source API insert data

Why are the changes needed?

Provide a more flexible commit protocol and won't impact perf

Does this PR introduce any user-facing change?

User can set sql commit protocol to SQLPathHadoopMapReduceCommitProtocol to use a commit protocol with staging dir

How was this patch tested?

Added UT