You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@hudi.apache.org by "cshuo (via GitHub)" <gi...@apache.org> on 2026/09/28 09:53:07 UTC
[PR] perf(flink): stream deduplication for sorted LSM write batches [hudi]
cshuo opened a new pull request, #20124:
URL: https://github.com/apache/hudi/pull/20124
### Describe the issue this Pull Request addresses
Closes #20123.
Flink LSM write buckets are already sorted by record key before pre-combine, but deduplication still collects the entire batch into a `LinkedHashMap<RecordKey, List<Record>>`. Stream adjacent equal-key records to remove this extra batch retention.
### Summary and Changelog
- Add `FlinkWriteHelper.deduplicateSortedRecords(..., isSortedByRecordKey, ...)`: reduce sorted input lazily with one lookahead record, and delegate unsorted input to the existing implementation.
- Reuse `reduceRecords()` to preserve the merger, left-to-right reduction order, delete handling, and empty-merge-result behavior.
- Pass the sorted-input flag from `StreamWriteFunction` after bucket sorting; callers that have not sorted their input retain the grouped path.
- Add coverage for lazy iteration, empty input, unsorted non-adjacent duplicates, event-time/commit-time ordering, ties, deletes/reinserts, partial updates, custom merger order, and COW/MOR writes with default and LSM layouts.
Validation: 13 targeted tests passed (9 helper tests and 4 COW/MOR write cases), with Checkstyle and `git diff --check` passing.
```bash
mvn -o -Pflink2.2 -pl hudi-flink-datasource/hudi-flink -am \
'-Dtest=TestFlinkWriteHelper,TestWriteCopyOnWrite#testDeduplicationWithInterleavedKeys,TestWriteMergeOnRead#testDeduplicationWithInterleavedKeys' \
-Dsurefire.failIfNoSpecifiedTests=false \
-DskipITs -DskipSparkTests -DskipScalaTests test
```
### Impact
For sorted LSM batches with pre-combine enabled, deduplication retains a constant number of record states instead of O(N) batch references and grouping containers. The sorting buffer remains allocated, and the number of merge operations is unchanged. No new configuration or storage-format changes. Throughput and peak-memory improvements have not been benchmarked.
### Risk Level
Low. The streaming path requires contiguous equal record keys and records that remain valid as the iterator advances. It is selected only after the existing LSM bucket sort; unsorted callers use the old implementation. Differential tests cover merge semantics, and COW/MOR tests cover write-path integration.
### Documentation Update
Added method Javadoc documenting sorted-input preconditions and unsorted fallback. No user-facing documentation changes are required.
### Contributor's checklist
- [ ] Read through [contributor's guide](https://hudi.apache.org/contribute/how-to-contribute)
- [x] Enough context is provided in the sections above
- [x] Adequate tests were added if applicable
--
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] perf(flink): stream deduplication for sorted LSM write batches [hudi]
Posted by "hudi-bot (via GitHub)" <gi...@apache.org>.
hudi-bot commented on PR #20124:
URL: https://github.com/apache/hudi/pull/20124#issuecomment-5868008028
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "77a5850b6c5a3b2e359648df1590d5b30dd94c51",
"status" : "PENDING",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=17332",
"triggerID" : "77a5850b6c5a3b2e359648df1590d5b30dd94c51",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 77a5850b6c5a3b2e359648df1590d5b30dd94c51 Azure: [PENDING](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=17332)
<details>
<summary>Bot commands</summary>
@hudi-bot supports the following commands:
- `@hudi-bot run azure` re-run the last Azure build
</details>
--
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] perf(flink): stream deduplication for sorted LSM write batches [hudi]
Posted by "codecov-commenter (via GitHub)" <gi...@apache.org>.
codecov-commenter commented on PR #20124:
URL: https://github.com/apache/hudi/pull/20124#issuecomment-5867952717
## [Codecov](https://app.codecov.io/gh/apache/hudi/pull/20124?dropdown=coverage&src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) Report
:white_check_mark: All modified and coverable lines are covered by tests.
:white_check_mark: Project coverage is 80.40%. Comparing base ([`ead194c`](https://app.codecov.io/gh/apache/hudi/commit/ead194cce54829fae9113689a2f28772d0d76045?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)) to head ([`77a5850`](https://app.codecov.io/gh/apache/hudi/commit/77a5850b6c5a3b2e359648df1590d5b30dd94c51?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)).
:warning: Report is 2 commits behind head on master.
<details><summary>Additional details and impacted files</summary>
```diff
@@ Coverage Diff @@
## master #20124 +/- ##
=========================================
Coverage 80.40% 80.40%
- Complexity 34874 34890 +16
=========================================
Files 2546 2546
Lines 142889 142940 +51
Branches 17374 17389 +15
=========================================
+ Hits 114895 114937 +42
- Misses 20082 20089 +7
- Partials 7912 7914 +2
```
| [Components](https://app.codecov.io/gh/apache/hudi/pull/20124/components?src=pr&el=components&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Coverage Δ | |
|---|---|---|
| [hudi-common](https://app.codecov.io/gh/apache/hudi/pull/20124/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `83.97% <ø> (ø)` | |
| [hudi-client](https://app.codecov.io/gh/apache/hudi/pull/20124/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `83.45% <ø> (-0.01%)` | :arrow_down: |
| [hudi-flink](https://app.codecov.io/gh/apache/hudi/pull/20124/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `85.74% <ø> (-0.01%)` | :arrow_down: |
| [hudi-spark-datasource](https://app.codecov.io/gh/apache/hudi/pull/20124/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `73.85% <ø> (ø)` | |
| [hudi-utilities](https://app.codecov.io/gh/apache/hudi/pull/20124/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `78.17% <ø> (ø)` | |
| [hudi-cli](https://app.codecov.io/gh/apache/hudi/pull/20124/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `70.05% <ø> (ø)` | |
| [hudi-hadoop](https://app.codecov.io/gh/apache/hudi/pull/20124/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `70.99% <ø> (ø)` | |
| [hudi-sync](https://app.codecov.io/gh/apache/hudi/pull/20124/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `76.02% <ø> (ø)` | |
| [hudi-io](https://app.codecov.io/gh/apache/hudi/pull/20124/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `81.50% <ø> (ø)` | |
| [hudi-timeline-service](https://app.codecov.io/gh/apache/hudi/pull/20124/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `83.06% <ø> (ø)` | |
| [hudi-cloud](https://app.codecov.io/gh/apache/hudi/pull/20124/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `81.00% <ø> (ø)` | |
| [hudi-kafka-connect](https://app.codecov.io/gh/apache/hudi/pull/20124/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `53.20% <ø> (ø)` | |
| [Flag](https://app.codecov.io/gh/apache/hudi/pull/20124/flags?src=pr&el=flags&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Coverage Δ | |
|---|---|---|
| [hadoop-mr-java-client](https://app.codecov.io/gh/apache/hudi/pull/20124/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `43.97% <ø> (-0.01%)` | :arrow_down: |
Flags with carried forward coverage won't be shown. [Click here](https://docs.codecov.io/docs/carryforward-flags?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#carryforward-flags-in-the-pull-request-comment) to find out more.
| [Files with missing lines](https://app.codecov.io/gh/apache/hudi/pull/20124?dropdown=coverage&src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Coverage Δ | |
|---|---|---|
| [...che/hudi/table/action/commit/FlinkWriteHelper.java](https://app.codecov.io/gh/apache/hudi/pull/20124?src=pr&el=tree&filepath=hudi-client%2Fhudi-flink-client%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fhudi%2Ftable%2Faction%2Fcommit%2FFlinkWriteHelper.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-aHVkaS1jbGllbnQvaHVkaS1mbGluay1jbGllbnQvc3JjL21haW4vamF2YS9vcmcvYXBhY2hlL2h1ZGkvdGFibGUvYWN0aW9uL2NvbW1pdC9GbGlua1dyaXRlSGVscGVyLmphdmE=) | `77.27% <ø> (ø)` | |
| [...java/org/apache/hudi/sink/StreamWriteFunction.java](https://app.codecov.io/gh/apache/hudi/pull/20124?src=pr&el=tree&filepath=hudi-flink-datasource%2Fhudi-flink%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fhudi%2Fsink%2FStreamWriteFunction.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-aHVkaS1mbGluay1kYXRhc291cmNlL2h1ZGktZmxpbmsvc3JjL21haW4vamF2YS9vcmcvYXBhY2hlL2h1ZGkvc2luay9TdHJlYW1Xcml0ZUZ1bmN0aW9uLmphdmE=) | `80.65% <ø> (-0.21%)` | :arrow_down: |
... and [1 file with indirect coverage changes](https://app.codecov.io/gh/apache/hudi/pull/20124/indirect-changes?src=pr&el=tree-more&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)
</details>
<details><summary> :rocket: New features to boost your workflow: </summary>
- :snowflake: [Test Analytics](https://docs.codecov.com/docs/test-analytics): Detect flaky tests, report on failures, and find test suite problems.
- :package: [JS Bundle Analysis](https://docs.codecov.com/docs/javascript-bundle-analysis): Save yourself from yourself by tracking and limiting bundle sizes in JS merges.
</details>
--
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] perf(flink): stream deduplication for sorted LSM write batches [hudi]
Posted by "danny0405 (via GitHub)" <gi...@apache.org>.
danny0405 merged PR #20124:
URL: https://github.com/apache/hudi/pull/20124
--
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] perf(flink): stream deduplication for sorted LSM write batches [hudi]
Posted by "hudi-bot (via GitHub)" <gi...@apache.org>.
hudi-bot commented on PR #20124:
URL: https://github.com/apache/hudi/pull/20124#issuecomment-5867694021
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "77a5850b6c5a3b2e359648df1590d5b30dd94c51",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "77a5850b6c5a3b2e359648df1590d5b30dd94c51",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 77a5850b6c5a3b2e359648df1590d5b30dd94c51 UNKNOWN
<details>
<summary>Bot commands</summary>
@hudi-bot supports the following commands:
- `@hudi-bot run azure` re-run the last Azure build
</details>
--
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] perf(flink): stream deduplication for sorted LSM write batches [hudi]
Posted by "hudi-agent (via GitHub)" <gi...@apache.org>.
hudi-agent commented on code in PR #20124:
URL: https://github.com/apache/hudi/pull/20124#discussion_r4127111823
##########
hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/StreamWriteFunction.java:
##########
@@ -560,7 +560,7 @@ protected List<WriteStatus> writeRecords(
rowItr, rowData -> recordConverter.convert(rowData, rowDataBucket.getBucketInfo()));
List<WriteStatus> statuses = writeFunction.write(
- deduplicateRecordsIfNeeded(recordItr), rowDataBucket.getBucketInfo(), instant);
+ deduplicateRecordsIfNeeded(recordItr, recordKeyComputer != null), rowDataBucket.getBucketInfo(), instant);
Review Comment:
🤖 Not introduced here, but relevant to the order this path preserves: `RowDataBucket.sort()` uses Flink's `QuickSort`, which keeps arrival order only below 13 records, and neither LSM comparator breaks ties by arrival order. In larger buckets, equal-key records can be reordered before dedup, so COMMIT_TIME_ORDERING or event-time ties could keep an older record. Would a follow-up with an arrival-order tie-breaker, plus a >13-record interleaved test, make sense? @danny0405 FYI.
<sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag quality.</i></sub>
--
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] perf(flink): stream deduplication for sorted LSM write batches [hudi]
Posted by "hudi-bot (via GitHub)" <gi...@apache.org>.
hudi-bot commented on PR #20124:
URL: https://github.com/apache/hudi/pull/20124#issuecomment-5869268537
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "77a5850b6c5a3b2e359648df1590d5b30dd94c51",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=17332",
"triggerID" : "77a5850b6c5a3b2e359648df1590d5b30dd94c51",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 77a5850b6c5a3b2e359648df1590d5b30dd94c51 Azure: [SUCCESS](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=17332)
<details>
<summary>Bot commands</summary>
@hudi-bot supports the following commands:
- `@hudi-bot run azure` re-run the last Azure build
</details>
--
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org