lanyuanxiaoyao/hudi
1
0
Fork 0
Code Activity

[MINOR] Sync HUDI-1196 to FlinkWriteHelper (#2415)

Browse Source
This commit is contained in:
Trevor
2021-01-09 07:57:56 +08:00
committed by GitHub
parent 17df517b81
commit c151147819
1 changed files with 2 additions and 4 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/table/action/commit/FlinkWriteHelper.java
View File

@@ -59,10 +59,8 @@ public class FlinkWriteHelper<T extends HoodieRecordPayload,R> extends AbstractW
return keyedRecords.values().stream().map(x -> x.stream().map(Pair::getRight).reduce((rec1, rec2) -> { return keyedRecords.values().stream().map(x -> x.stream().map(Pair::getRight).reduce((rec1, rec2) -> {
@SuppressWarnings("unchecked") @SuppressWarnings("unchecked")
T reducedData = (T) rec1.getData().preCombine(rec2.getData()); T reducedData = (T) rec1.getData().preCombine(rec2.getData());
// we cannot allow the user to change the key or partitionPath, since that will affect HoodieKey reducedKey = rec1.getData().equals(reducedData) ? rec1.getKey() : rec2.getKey();
// everything return new HoodieRecord<T>(reducedKey, reducedData);
// so pick it from one of the records.
return new HoodieRecord<T>(rec1.getKey(), reducedData);
}).orElse(null)).filter(Objects::nonNull).collect(Collectors.toList()); }).orElse(null)).filter(Objects::nonNull).collect(Collectors.toList());
} }
} }