[HUDI-6822] Fix deletes handling in hbase index when partition path is updated

[flashJd]

Change Logs

Similar to #9114, corner case when a record moves from 1 partition to another with partition path update set to true

Impact

hudi/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/index/hbase/SparkHoodieHBaseIndex.java

Lines 235 to 242 in eab00d5

	private <R> Function2<Integer, Iterator<HoodieRecord<R>>, Iterator<HoodieRecord<R>>> locationTagFunction(
	HoodieTableMetaClient metaClient) {
	// `multiGetBatchSize` is intended to be a batch per 100ms. To create a rate limiter that measures
	// operations per second, we need to multiply `multiGetBatchSize` by 10.
	Integer multiGetBatchSize = config.getHbaseIndexGetBatchSize();
	return (partitionNum, hoodieRecordIterator) -> {
	boolean updatePartitionPath = config.getHbaseIndexUpdatePartitionPath();

1. Hbase index tagLocation is executed concurrently in different spark partitions, so the index delete operation may happen after index add operation, cause the undefined behavior as codope said.
2. ReduceByKey can be a costly operation as [HUDI-6467] Fix deletes handling in rli when partition path is updated #9114 (comment) discussed
   so add a flag to ignore this index delete operations in partitionPathUpdate condition

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

N/A

Documentation Update

N/A

Contributor's checklist

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

[danny0405]

Why ingores the delete records for old partitions directly?

[flashJd]

Only ignore the deleted record in partition change scene, not all deleted record:
existing record(id1, par1), new record(id1, par2), if updatePartitionPath set to true, will emit delete record to par1 and add new record(id1, par2) to par2, in this scene, the delete record is no need to update index as the new record will override the index

[nsivabalan]

how do we handle deletes. i.e. if we get deletes for a record in partition p1, when it reaches metadata writer, we might just have 1 recordDelegate but theignore flag will not be set since we are not setting it in any of write handles? and so we should be good.

we are setting the ignore flag only in indexing code and specifically when indexing could reutrn two version of record delegate.

just wanted to confirm my understanding.

[flashJd]

yeah, it's right, we are setting the ignore flag only in indexing code and specifically when indexing could reutrn two version of record delegate.

[nsivabalan]

how do we handle deletes. i.e. if we get deletes for a record in partition p1, when it reaches metadata writer, we might just have 1 recordDelegate but theignore flag will not be set since we are not setting it in any of write handles? and so we should be good.

we are setting the ignore flag only in indexing code and specifically when indexing could reutrn two version of record delegate.

just wanted to confirm my understanding.

[hudi-bot]

CI report:

• fe66314 Azure: SUCCESS

Bot commands

@hudi-bot supports the following commands:

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