[HUDI-5968] Fix global index duplicate and handle custom payload when update partition #8490

xushiyan · 2023-04-18T14:00:54Z

Change Logs

When using global index (bloom or simple), and update partition is set to true. There is a chance where record is in p1 at the beginning, and later updated to p2, when updating to p3 and compaction not yet happened, global index joined both old versions of the record in p1 and p2, and tagged 2 records to insert to p3. This sort of duplicates will reside in the dataset and won't be reconciled unless manually dedup the table.

When records are inserted into new partitions, existing logic does not honor custom payload, which should be handled by record merger API.

Impact

Global index will load fileslice to perform merge and tagging, which slows down the whole process if a lot partition updates happen.

Risk level

High.

End to end testing and UT.

Documentation Update

New config hoodie.global.index.reconcile.parallelism

Contributor's checklist

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

xushiyan · 2023-04-18T16:39:33Z

More code clean up needed

nsivabalan · 2023-04-25T23:35:32Z

so, #8344 is not valid anymore ?

hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/HoodieMergedReadHandle.java

hudi-client/hudi-client-common/src/main/java/org/apache/hudi/index/HoodieIndexUtils.java

nsivabalan · 2023-04-26T00:20:42Z

do you think we should do the snapshot read only when updatePartitionPath is set to true and avoid when its set to false.
I am inclined towards leave it uniform(and not have two code paths). but just bringing up a point if we feel we really need to avoid the overhead if not required.

hudi-common/src/test/java/org/apache/hudi/common/testutils/RawTripTestPayload.java

...ent/hudi-client-common/src/main/java/org/apache/hudi/index/bloom/HoodieGlobalBloomIndex.java

nsivabalan · 2023-05-03T00:38:59Z

hudi-client/hudi-client-common/src/main/java/org/apache/hudi/index/HoodieIndexUtils.java

+    HoodieData<HoodieRecord<R>> newRecords = taggedHoodieRecords.filter(p -> !p.getRight().isPresent()).map(Pair::getLeft);
+    // the records tagged to existing base files
+    HoodieData<HoodieRecord<R>> updatingRecords = taggedHoodieRecords.filter(p -> p.getRight().isPresent()).map(Pair::getLeft)
+        .distinctWithKey(HoodieRecord::getRecordKey, config.getGlobalIndexReconcileParallelism());


I see we are doing distinctWithKey here. So, we assume that records may not be duplicated at all?
what happens if there are duplicates already. for eg, some one ingested same batch of data w/ bulk_insert. may be we need to revisit overall end to end flow for this scenario of how our global index will work.
but trying to think through how it might surface after this fix?

we may not require to fix anything as such. but wanted to see what will be outcome.

the tagged records at this point will contain dups in case of last write updated partition and inserted a new record to new partition, and compaction has not happened yet. The first look up will still get 2 records due to join only with base files.

hudi-client/hudi-client-common/src/main/java/org/apache/hudi/index/HoodieIndexUtils.java

hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/HoodieMergedReadHandle.java

nsivabalan · 2023-05-03T01:24:24Z

hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/HoodieMergedReadHandle.java

+
+  private Option<FileSlice> getLatestFileSlice() {
+    if (nonEmpty(instantTime)
+        && hoodieTable.getMetaClient().getCommitsTimeline().filterCompletedInstants().lastInstant().isPresent()) {


can we move these checks to constructor.

hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/HoodieMergedReadHandle.java

hudi-client/hudi-client-common/src/main/java/org/apache/hudi/index/HoodieIndexUtils.java

...tasource/hudi-spark/src/test/scala/org/apache/hudi/functional/TestMORDataSourceStorage.scala

hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/HoodieMergedReadHandle.java

hudi-client/hudi-client-common/src/main/java/org/apache/hudi/index/HoodieIndexUtils.java

hudi-common/src/main/java/org/apache/hudi/common/util/SpillableMapUtils.java

hudi-common/src/main/java/org/apache/hudi/common/model/HoodieAvroIndexedRecord.java

nsivabalan · 2023-05-04T14:15:57Z

hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/common/model/HoodieSparkRecord.java

  }

  /**
   * Utility method to convert bytes to HoodieRecord using schema and payload class.
   */
  private static HoodieRecord<InternalRow> convertToHoodieSparkRecord(StructType structType, HoodieSparkRecord record, Pair<String, String> recordKeyPartitionPathFieldPair,
-      boolean withOperationField, Option<String> partitionName) {
+      boolean withOperationField, Option<String> partitionName, Option<StructType> structTypeWithoutMetaFields) {


can you enhance the java doc on when and how to use this method. for eg, when should the last arg be set?
or should we introduce overloaded method.

nsivabalan · 2023-05-04T14:20:05Z

MergeHandle needs some thorough testing. can you file a follow up ticket for that

...t/hudi-client-common/src/main/java/org/apache/hudi/index/simple/HoodieGlobalSimpleIndex.java

nsivabalan

Once CI is green, we can go ahead!

hudi-bot · 2023-05-05T12:56:57Z

CI report:

d64221f Azure: SUCCESS

Bot commands

@hudi-bot supports the following commands:

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

… update partition (apache#8490)

xushiyan · 2023-05-17T08:42:43Z

hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/HoodieMergedReadHandle.java

+        String key = record.getRecordKey();
+        if (deltaRecordMap.containsKey(key)) {
+          deltaRecordKeys.remove(key);
+          Option<Pair<HoodieRecord, Schema>> mergeResult = recordMerger


this merge result needs to be wrapped back to the original payload so that caller won't have to do it. fixed in #8736

xushiyan · 2023-05-17T08:47:02Z

hudi-client/hudi-client-common/src/main/java/org/apache/hudi/index/HoodieIndexUtils.java

+          if (incoming.getData() instanceof EmptyHoodieRecordPayload) {
+            // incoming is a delete: force tag the incoming to the old partition
+            return Collections.singletonList(getTaggedRecord(incoming, Option.of(existing.getCurrentLocation()))).iterator();


this needs to use isDelete() api to check and incoming's key need to be overwritten to the existing's key. fixed in #8736

danny0405 · 2023-06-12T13:22:39Z

hudi-client/hudi-client-common/src/main/java/org/apache/hudi/index/HoodieIndexUtils.java

+    HoodieRecord incomingWithMetaFields = incomingPrepended
+        .wrapIntoHoodieRecordPayloadWithParams(writeSchema, config.getProps(), Option.empty(), config.allowOperationMetadataField(), Option.empty(), false, Option.empty());
+    Option<Pair<HoodieRecord, Schema>> mergeResult = config.getRecordMerger()
+        .merge(existing, existingSchema, incomingWithMetaFields, writeSchemaWithMetaFields, config.getProps());


The record merger is instantiated for each time, will cause unnecessary onverhead.

good catch! saw it's been fixed now

loukey-lj · 2023-11-22T02:23:17Z

@xushiyan @nsivabalan https://issues.apache.org/jira/browse/HUDI-7131

xushiyan marked this pull request as draft April 18, 2023 16:39

xushiyan force-pushed the HUDI-5968-fix-global-index-dup-2 branch 2 times, most recently from dbe76d5 to df20633 Compare April 24, 2023 05:28

xushiyan marked this pull request as ready for review April 24, 2023 05:39

xushiyan force-pushed the HUDI-5968-fix-global-index-dup-2 branch from 991afee to f5ebd44 Compare April 24, 2023 05:48

vinothchandar added release-0.14.0 priority:critical production down; pipelines stalled; Need help asap. labels Apr 25, 2023

nsivabalan requested changes Apr 26, 2023

View reviewed changes

codope mentioned this pull request Apr 27, 2023

[SUPPORT] There are duplicate values in HUDI MOR table for different partition and not updating values in same partition for GLOBAL_BLOOM #5869

Closed

This was referenced Apr 28, 2023

[SUPPORT]Duplicate records in MOR #6591

Closed

Duplicate data in MOR table Hudi #8178

Open

xushiyan and others added 6 commits April 30, 2023 23:52

[HUDI-5968] Fix global index duplicate when update partition

dc53ee5

use read handle

65bd4a4

fix style

f7a764b

update config name and tests

3ffac9b

fix style

f363372

fix ut

85dd094

xushiyan force-pushed the HUDI-5968-fix-global-index-dup-2 branch from f5ebd44 to 85dd094 Compare May 2, 2023 08:48

xushiyan added 3 commits May 2, 2023 16:51

fix style

2e2995c

revert pom changes

4cf5141

fix ut

8ee4e9f

xushiyan commented May 2, 2023

View reviewed changes

hudi-common/src/test/java/org/apache/hudi/common/testutils/RawTripTestPayload.java Show resolved Hide resolved

xushiyan requested a review from nsivabalan May 2, 2023 17:54

nsivabalan requested changes May 3, 2023

View reviewed changes

xushiyan added 3 commits May 4, 2023 02:35

fix merger api and tests (api to be cleaned up)

2d71996

fix avro api usage

2f0c75a

(temp) add cache logic

3d7d1f6

fix style

7575e66

apache deleted a comment from hudi-bot May 4, 2023

nsivabalan reviewed May 4, 2023

View reviewed changes

clean up api usage and add UT

2fa5dfe

nsivabalan reviewed May 4, 2023

View reviewed changes

nsivabalan reviewed May 5, 2023

View reviewed changes

...t/hudi-client-common/src/main/java/org/apache/hudi/index/simple/HoodieGlobalSimpleIndex.java Outdated Show resolved Hide resolved

add more readhandle UTs

d64221f

nsivabalan approved these changes May 5, 2023

View reviewed changes

xushiyan merged commit cabcb2b into apache:master May 5, 2023
17 checks passed

xushiyan deleted the HUDI-5968-fix-global-index-dup-2 branch May 5, 2023 13:28

yihua pushed a commit to yihua/hudi that referenced this pull request May 15, 2023

[HUDI-5968] Fix global index duplicate and handle custom payload when…

39e775e

… update partition (apache#8490)

yihua pushed a commit to yihua/hudi that referenced this pull request May 15, 2023

[HUDI-5968] Fix global index duplicate and handle custom payload when…

cac608b

… update partition (apache#8490)

xushiyan commented May 17, 2023

View reviewed changes

ad1happy2go mentioned this pull request May 31, 2023

[SUPPORT] Duplicate Data in MOR Partitioned Table #8835

Open

danny0405 reviewed Jun 12, 2023

View reviewed changes

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[HUDI-5968] Fix global index duplicate and handle custom payload when update partition #8490

[HUDI-5968] Fix global index duplicate and handle custom payload when update partition #8490

xushiyan commented Apr 18, 2023

xushiyan commented Apr 18, 2023

nsivabalan commented Apr 25, 2023

nsivabalan commented Apr 26, 2023

nsivabalan May 3, 2023

nsivabalan May 3, 2023

xushiyan May 3, 2023

nsivabalan May 3, 2023

nsivabalan May 4, 2023

nsivabalan commented May 4, 2023

nsivabalan left a comment

hudi-bot commented May 5, 2023

xushiyan May 17, 2023

xushiyan May 17, 2023

danny0405 Jun 12, 2023

xushiyan Jun 24, 2023

loukey-lj commented Nov 22, 2023

[HUDI-5968] Fix global index duplicate and handle custom payload when update partition #8490

[HUDI-5968] Fix global index duplicate and handle custom payload when update partition #8490

Conversation

xushiyan commented Apr 18, 2023

Change Logs

Impact

Risk level

Documentation Update

Contributor's checklist

xushiyan commented Apr 18, 2023

nsivabalan commented Apr 25, 2023

nsivabalan commented Apr 26, 2023

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

nsivabalan commented May 4, 2023

nsivabalan left a comment

Choose a reason for hiding this comment

hudi-bot commented May 5, 2023

CI report:

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

loukey-lj commented Nov 22, 2023