[HUDI-2144]Bug-Fix:Offline clustering(HoodieClusteringJob) will cause insert action losing data

[zhangyue19921010]

What is the purpose of the pull request

Please read https://issues.apache.org/jira/projects/HUDI/issues/HUDI-2144 for details

Brief change log

Modify UpsertPartitioner.java

Verify this pull request

(Please pick either of the following options)

This pull request is a trivial rework / code cleanup without any test coverage.

(or)

This pull request is already covered by existing tests, such as (please describe tests).

(or)

This change added tests and can be verified as follows:

(example:)

• Added integration tests for end-to-end.
• Added HoodieClientWriteTest to verify the change.
• Manually verified the change by running a job locally.

Committer checklist

• Has a corresponding JIRA in PR title & commit

• Commit message is descriptive of the change

• CI is green

• Necessary doc changes done or have another open PR

• For large changes, please consider breaking it into sub-tasks under an umbrella JIRA.

[hudi-bot]

CI report:

• 9d3e143 Azure: SUCCESS

Bot commands

@hudi-bot supports the following commands:

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

[codecov-commenter]

Codecov Report

Merging #3240 (9d3e143) into master (9908204) will increase coverage by 0.06%.
The diff coverage is 100.00%.

@@             Coverage Diff              @@
##             master    #3240      +/-   ##
============================================
+ Coverage     47.61%   47.67%   +0.06%     
- Complexity     5487     5517      +30     
============================================
  Files           924      929       +5     
  Lines         41206    41303      +97     
  Branches       4133     4144      +11     
============================================
+ Hits          19619    19691      +72     
- Misses        19844    19864      +20     
- Partials       1743     1748       +5     

Flag	Coverage Δ
hudicli	39.97% <ø> (ø)
hudiclient	34.57% <100.00%> (-0.02%)	⬇️
hudicommon	48.57% <ø> (+<0.01%)	⬆️
hudiflink	60.03% <ø> (+0.44%)	⬆️
hudihadoopmr	51.29% <ø> (ø)
hudisparkdatasource	67.32% <ø> (-0.01%)	⬇️
hudisync	54.51% <ø> (+0.03%)	⬆️
huditimelineservice	64.07% <ø> (ø)
hudiutilities	58.60% <ø> (+0.01%)	⬆️

Flags with carried forward coverage won't be shown. Click here to find out more.

Impacted Files	Coverage Δ
...he/hudi/table/action/commit/UpsertPartitioner.java	96.10% <100.00%> (ø)
...n/java/org/apache/hudi/index/SparkHoodieIndex.java	56.52% <0.00%> (-30.15%)	⬇️
...java/org/apache/hudi/table/HoodieTableFactory.java	84.61% <0.00%> (-7.06%)	⬇️
...n/java/org/apache/hudi/common/model/FileSlice.java	73.80% <0.00%> (-2.39%)	⬇️
...hudi/utilities/sources/helpers/KafkaOffsetGen.java	87.68% <0.00%> (-0.65%)	⬇️
.../org/apache/hudi/common/model/HoodieFileGroup.java	83.92% <0.00%> (-0.56%)	⬇️
...in/java/org/apache/hudi/table/HoodieTableSink.java	10.52% <0.00%> (ø)
.../org/apache/hudi/streamer/FlinkStreamerConfig.java	0.00% <0.00%> (ø)
.../org/apache/hudi/streamer/HoodieFlinkStreamer.java	0.00% <0.00%> (ø)
...apache/hudi/utilities/sources/AvroKafkaSource.java	0.00% <0.00%> (ø)
... and 31 more

---

Continue to review full report at Codecov.

Legend - Click here to learn more
Δ = absolute <relative> (impact), ø = not affected, ? = missing data
Powered by Codecov. Last update 9908204...9d3e143. Read the comment docs.

[leesf]

when clustering plan contains the small files, the new insert should not get into small files, so the new insert get into file slice2 is strange.

[zhangyue19921010]

Hi @leesf Thanks for your review. Yes, when clustering plan contains the small files, the new insert should not get into small files only works when users set this.config.isClusteringEnabled() during insert job linked here:

hudi/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/table/action/commit/UpsertPartitioner.java

Line 149 in 650c445

if (this.config.isClusteringEnabled()) {

what I means is that this limitation is not good enough because others job like HoodieClusteringJob also can generate cluster plan and the insert job mentioned above can't be aware of it and can't use filterSmallFilesInClustering func to filter cluster-related small files.

[vinothchandar]

@satishkotha @codope Can one of you please help review?

[codope]

If the clustering is not enabled, can the pendingClusteringFileGroupsId set be non-empty? Would be great if we can add a test to replicate the scenario.

[zhangyue19921010]

Hi @codope Thanks for your view.

As I mentioned above. If the clustering is not enabled during insert but there is an offline clustering plan job is running for this table, the pendingClusteringFileGroupsId set could be non-empty too.

I just add a UT named testUpsertPartitionerWithSmallFileHandlingAndClusteringPlan to prove it. I set withClusteringConfig(HoodieClusteringConfig.newBuilder().withInlineClustering(false).withAsyncClustering(false).build()) during ingestion and create a clustering plan manually out of this ingestion scope.

As you you can see, If the condition is this.config.isClusteringEnabled() only, there will be two small slices can be inserted including file slice 1 in pending clustering plan.

If the condition is !pendingClusteringFileGroupsId.isEmpty(), there will be only one small slice can be used and file slice 1 is filtered.

[codope]

Got it. Thanks @zhangyue19921010

[satishkotha]

Good find. We can probably remove config.isClusteringEnabled() check below and only rely on pendingClusteringFileGroupsId?

cc: @lw309637554 any reason to base this on config originally in #2275?

[zhangyue19921010]

Hi @satishkotha Thanks for your view. Only rely on pendingClusteringFileGroupsId is much better :) changed.

[lw309637554]

@satishkotha @zhangyue19921010
At first we have two config for clustering. If set ASYNC_CLUSTERING_ENABLE_OPT_KEY will be ok.
public boolean isAsyncClusteringEnabled() {
return Boolean.parseBoolean(props.getProperty(HoodieClusteringConfig.ASYNC_CLUSTERING_ENABLE_OPT_KEY));
}

public boolean isClusteringEnabled() {
// TODO: future support async clustering
return inlineClusteringEnabled() || isAsyncClusteringEnabled();
}

[lw309637554]

@satishkotha @zhangyue19921010
Use "if (!pendingClusteringFileGroupsId.isEmpty())" will improve ease of use.
Another need to modify. But if this will bring performance loss? @satishkotha

" private JavaRDD<HoodieRecord> clusteringHandleUpdate(JavaRDD<HoodieRecord> inputRecordsRDD) {
if (config.isClusteringEnabled()) {"

[codope]

@vinothchandar @satishkotha It looks good to me and I've verified the patch. Can you guys take a pass and land it?