[HUDI-6085] Eliminate cleaning tasks for flink mor table if online async compaciton is disabled

[zhuanshenbsj1]

Change Logs

MOR table in upsert scenario, not need to clean when online async compaction is turned off.

In scenarios with a large number of files and partitions, turning off clean will improve performance

Impact

Describe any public API or user-facing feature change or any performance impact.

Risk level (write none, low medium or high below)

If medium or high, explain what verification was done to mitigate the risks.

Documentation Update

Describe any necessary documentation update if there is any new feature, config, or user-facing change

• The config description must be updated if new configs are added or the default value of the configs are changed
• Any new feature or user-facing change requires updating the Hudi website. Please create a Jira ticket, attach the
  ticket number here and follow the instruction to make
  changes to the website.

Contributor's checklist

• Read through contributor's guide
• Change Logs and Impact were stated clearly
• Adequate tests were added if applicable
• CI passed

[danny0405]

So what component is responsible for data cleaning then?

[zhuanshenbsj1]

Offline compaction do clean，Consistent with clustering.

// Append mode
  if (OptionsResolver.isAppendMode(conf)) {
    DataStream<Object> pipeline = Pipelines.append(conf, rowType, dataStream, context.isBounded());
    if (OptionsResolver.needsAsyncClustering(conf)) {
      return Pipelines.cluster(conf, rowType, pipeline);
    } else {
      return Pipelines.dummySink(pipeline);// If async-clustering=true,  no clean operator.
    }
  }

[danny0405]

The cleaning can take effect finally right? Because the table would get compacted for sometime anyway.

[zhuanshenbsj1]

Offline clustering/compaction will do clean by default , add --clean-async-enabled config will close it.

[zhuanshenbsj1]

Current online async-clean and offline clean use the same configuration ：clean.async.enabled. Add a new configuration clean.offline.enable，making the configuration description clearer.

[danny0405]

The spark offline compaction Job does not take care of cleaning, could you make it clrear how user can handle the cleaning when they use the flink streaming ingestion and spark offline compaction.

[zhuanshenbsj1]

Adjust the cleaning operation in SparkRDDWriteClient#cluster/compact, when ASYNC_CLEAN is true will do asynchronous clean in prewrite， otherwise will do synchronous clean in autoCleanOnCommit().

[danny0405]

We need to think through the e2e use case for flink streaming ingestion and spark offline compaction, we should add cleaning for spark compaction and clustering first which is a block change of this patch.

[danny0405]

We will not schedule cleaning task if async cleaning is disabled:

hudi/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/CleanFunction.java

Line 72 in 67ae0c8

if (conf.getBoolean(FlinkOptions.CLEAN_ASYNC_ENABLED) && isCleaning) {

, so should be fine?

[zhuanshenbsj1]

If CLEAN_ASYNC_ENABLED = true，a schedule will still be executed. I think cluster and compact should be consistent here(If the cow table async cluster is closed, there will be no clean operator). And now both Spark and Flink will clean up when executing offline jobs, unless forcibly closed. What do you think?

[hudi-bot]

CI report:

• 9edae81 Azure: SUCCESS

Bot commands

@hudi-bot supports the following commands:

• @hudi-bot run azure re-run the last Azure build

[danny0405]

Do not introduce unrelated changes in one patch, if we want to add cleaning procedure for Spark compaction and cleaning, fire another PR and let's discuss with it.

[zhuanshenbsj1]

Removed Spark related modifications and migrated to a new PR. #8505

[yihua]

@zhuanshenbsj1 @danny0405 is this PR still relevant?

[danny0405]

cc @zhuanshenbsj1 I think we can cloase it, because we already have #8505 .