[SUPPORT]BaseDatasetBulkInsertCommitActionExecutor execute WriteStatus donot persist cause data write hudi 4 times

[dongtingting]

Describe the problem you faced

There is a job use bulk insert insert overwrite a cow table. We find there are 4 stage run bulk insert write, data write 4 times and only the last stage data remain, other 3 stage writen data is finally remove when finalize write.

all of the four stage on red line do bulk insert write.

more details about the four stage:

This is because the four stage all use writestatus rdd, DatasetBulkInsertOverwriteCommitActionExecutor writeStatus rdd is not persist, this will case upstream rdd(bulk insert write) repeat running 4 times.

DatasetBulkInsertOverwriteCommitActionExecutor getPartitionToReplacedFileIds: use writestatus isEmpty DatasetBulkInsertOverwriteCommitActionExecutor getPartitionToReplacedFileIds: use writestatus distinct HoodieSparkSqlWriter.commitAndPerformPostOperations : use writestatus count HoodieSparkSqlWriter.commitAndPerformPostOperations :use writestatus collect

Upsert(do not use bulk insert) do not have this problem, because they persist writestatus. But BaseDatasetBulkInsertCommitActionExecutor do not persist writestatus. I think we should persist rdd at the beging of BaseDatasetBulkInsertCommitActionExecutor.buildHoodieWriteMetadata, does anyone agree?

To Reproduce

Steps to reproduce the behavior:

1. create a cow table test_table, using simple index

   create table if not exists test_table
   (
   id                                string     
   , name                             string    
   ,p_date                            string        comment '分区日期, yyyyMMdd'
   )USING hudi
    partitioned by (p_date)
     options (
     type='cow'
   );

2. insert overwrite table use bulk insert

   
   set hoodie.datasource.write.operation=BULK_INSERT;
   set hoodie.bulkinsert.shuffle.parallelism=200;

insert overwrite test_table partition (p_date = '20240806') select id, name, p_date from source table

3. check spark job task  log, you will find isEmpty, all of distinct , count , collect  stage task  log have create marker and create handle log.

**Expected behavior**

A clear and concise description of what you expected to happen.

**Environment Description**

* Hudi version : 0.14.0

* Spark version : 2.4

* Hadoop version : 2.6

[dongtingting]

@danny0405 @beyond1920 can you help me to confirm？

[danny0405]

@KnightChess maybe you can give some insights here, also cc @nsivabalan for visibility.

[beyond1920]

@dongtingting Good catch. Thanks for fire this issue. It seems writestatus rdd does not persist. I would like to check this problem today, would reply later.

[KnightChess]

@dongtingting nice catch. we can persist writestatus avoid this issue.

[dongtingting]

@dongtingting nice catch. we can persist writestatus avoid this issue.

thanks very munch for your relay。 I am glad to fix it, later i will create a pr to fix it.

[ad1happy2go]

@dongtingting Thanks a lot. Let us know when you have PR ready.

Creating tracking JIRA for the same - https://issues.apache.org/jira/browse/HUDI-8078

[dongtingting]

@dongtingting Thanks a lot. Let us know when you have PR ready.

Creating tracking JIRA for the same - https://issues.apache.org/jira/browse/HUDI-8078

sorry for late reply. i have create a pr to fix: https://github.com/apache/hudi/pull/11811/files. cc @KnightChess