jonvex commented on code in PR #13603:
URL: https://github.com/apache/hudi/pull/13603#discussion_r2240890598
##########
hudi-common/src/main/java/org/apache/hudi/metadata/HoodieTableMetadataUtil.java:
##########
@@ -1083,24 +1084,32 @@ public static Pair<Set<String>, Set<String>>
computeRevivedAndDeletedKeys(Set<St
* @return
*/
@VisibleForTesting
- public static HoodieData<HoodieRecord> reduceByKeys(HoodieData<HoodieRecord>
recordIndexRecords, int parallelism) {
- return recordIndexRecords.mapToPair(
- (SerializablePairFunction<HoodieRecord, HoodieKey, HoodieRecord>)
t -> Pair.of(t.getKey(), t))
- .reduceByKey((SerializableBiFunction<HoodieRecord, HoodieRecord,
HoodieRecord>) (record1, record2) -> {
- boolean isRecord1Deleted = record1.getData() instanceof
EmptyHoodieRecordPayload;
- boolean isRecord2Deleted = record2.getData() instanceof
EmptyHoodieRecordPayload;
- if (isRecord1Deleted && !isRecord2Deleted) {
- return record2;
- } else if (!isRecord1Deleted && isRecord2Deleted) {
- return record1;
- } else if (isRecord1Deleted && isRecord2Deleted) {
- // let's delete just 1 of them
- return record1;
- } else {
- throw new HoodieIOException("Two HoodieRecord updates to RLI is
seen for same record key " + record2.getRecordKey() + ", record 1 : "
- + record1.getData().toString() + ", record 2 : " +
record2.getData().toString());
- }
- }, parallelism).values();
+ public static HoodieData<HoodieRecord> reduceByKeys(HoodieData<HoodieRecord>
recordIndexRecords, int parallelism, boolean isPartitionedRLI) {
+ HoodiePairData<HoodieKey, HoodieRecord> recordIndexRecordsPair;
+ if (isPartitionedRLI) {
+ recordIndexRecordsPair = recordIndexRecords.mapToPair(r -> {
+ String recordPartitionPath = r.getData() instanceof
EmptyHoodieRecordPayloadWithPartition
Review Comment:
If we do that, then whenever a partition is updated, we will need to update
all the filegroups in the table. Because record key + data table partition path
can hash to any of the mdt filegroups. So using this way, the number of
filegroups updated in the mdt will be proportional to the number of partitions
updated in data table. This helps with large scale tables
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
