[HUDI-3069] compact improve #4400
Conversation
vinothchandar
left a comment
vinothchandar
left a comment
There was a problem hiding this comment.
Have a clarification on the first fix. Could you add some UTs for this?
| .getLatestFileSlices(partitionPath) | ||
| .filter(slice -> !fgIdsInPendingCompactionAndClustering.contains(slice.getFileGroupId())) | ||
| .map(s -> { | ||
| // We can think that the latest data is in the latest delta log file, so we sort it from large |
There was a problem hiding this comment.
I think you are assuming the later writes in the log always overwrites the earlier ones? this is not true always.
There was a problem hiding this comment.
You're right, but in most cases, the new data is often in the latest delta log, so we sort it from large to small according to the instance time. The program will avoid updating the data in the externalspillablemap to save compact time. What do you think
There was a problem hiding this comment.
Have a clarification on the first fix. Could you add some UTs for this?
OK, I'll try to add some UTs
There was a problem hiding this comment.
I think you are assuming the later writes in the log always overwrites the earlier ones? this is not true always.
In the compact plan generation phase, I just changed the order of reading delta log files. In the internal production environment, I have used this method for a month, and no data exceptions have occurred. Now, I don't know how I should test this place. Can you give me some suggestions
There was a problem hiding this comment.
In addition, I changed the reading order of deltalog to avoid data rewriting to the greatest extent. Houdierecordpayload#precombine will still execute and select the correct data.
| HoodieOperation operation = choosePrev ? oldRecord.getOperation() : hoodieRecord.getOperation(); | ||
| records.put(key, new HoodieRecord<>(new HoodieKey(key, hoodieRecord.getPartitionPath()), combinedValue, operation)); | ||
| // If combinedValue is oldValue, no need rePut oldRecord | ||
| if (!combinedValue.equals(oldValue)) { |
There was a problem hiding this comment.
This feels like a valid optimization.
|
@yihua : Can you follow up on the review please. |
Co-authored-by: [email protected] <loukey_7821>
… 0 (apache#4823) Co-authored-by: Hui An <[email protected]>
* [HUDI-3389] fix ColumnarArrayData ClassCastException issue * [HUDI-3389] remove MapColumnVector.java, RowColumnVector.java, and add test case for array<int> field
…pache#4837) * [HUDI-3446] Supports batch Reader in BootstrapOperator#loadRecords
…he#3887) Co-authored-by: yuezhang <[email protected]>
* Fixing restore with metadata enabled * Fixing test failures
…d remove duplicates (apache#4845) Co-authored-by: Hui An <[email protected]>
…n operations are present using a config. (apache#4212) Co-authored-by: sivabalan <[email protected]>
…ot be reused (apache#4861) * Before the patch, the flink streaming reader caches the meta client thus the archived timeline, when fetching the instant details from the reused timeline, the exception throws * Add a method in HoodieTableMetaClient to return a fresh new archived timeline each time
…che#4808) Co-authored-by: yuzhaojing <[email protected]>
…e#4898) Co-authored-by: 苏承祥 <[email protected]>
…field does not have diff data type (apache#4852)
ParquetColumnarRowSplitReader#batchSize is 2048, so Changing MINI_BATCH_SIZE to 2048 will reduce memory cache.
…e Index (apache#4840) Co-authored-by: guanziyue <[email protected]>
…ypedProperties.java (apache#4920)
…he#4844) Co-authored-by: Wenning Ding <[email protected]>
…he#4809) Co-authored-by: yuzhaojing <[email protected]>
…eMetadataTableValidator (apache#4878)
* Use iterator to void eager materialization to be memory friendly
…h the latest schema (apache#4000) Co-authored-by: yuzhaojing <[email protected]>
|
I found that after modifying the reading order of the delta log, HoodieRecordPayload#preCombine may have some problems when compacting (when the orderingVal of the two data is the same, the latest submitted data will not be selected). Later I will submit a new pr separately to fix this issue. So this pr just optimizes the code。 |
|
Very sorry, my fault, there was a problem with the merge. I will split it into two PRs and resubmit. |
19 participants
Brief change log
I found that when the compact plan is generated, the delta log files under each filegroup are arranged in the natural order of instant time. in the majority of cases,We can think that the latest data is in the latest delta log file, so we sort it from large to small according to the instance time, which can largely avoid rewriting the data in the compact process, and then optimize the compact time.
In addition, when reading the delta log file, we compare the data in the external spillablemap with the delta log data. If oldrecord is selected, there is no need to rewrite the data in the external spillablemap. Rewriting data will waste a lot of resources when data is spill to disk
This pull request is already covered by existing tests, such as (please describe tests).
Committer checklist
[*] Has a corresponding JIRA in PR title & commit()
[*] Commit message is descriptive of the change
CI is green
Necessary doc changes done or have another open PR
For large changes, please consider breaking it into sub-tasks under an umbrella JIRA.