[HUDI-8518][HUDI-8741] Fix RLI and Secondary index with custom payload or merge mode #12525
Conversation
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
codope
commented
Dec 20, 2024
Comment on lines
+830
to
+841
| // For log records, we only need to process deletes. However, deletes may or may not be part of delete blocks (delete using custom merge mode). | ||
| // So, we need to process the log files to get the record keys that are deleted. We can then generate RLI records for those keys. | ||
| // 1. Get all merged record keys - any custom merger which handles delete outside delete block should not be present in merged keys. | ||
| // 2. Get all un-merged record keys - this will contain all valid and deleted keys, irrespective of delete block or merge mode. | ||
| // 3. Get deleted record keys - this will be the difference of merged and un-merged keys. | ||
| Set<String> mergedRecordKeys = getRecordKeys(logFilePaths, dataTableMetaClient, | ||
| finalWriterSchemaOpt, maxBufferSize, instantTime, true, true, true, engineType); | ||
| Set<String> unMergedRecordKeys = getRecordKeys(logFilePaths, dataTableMetaClient, | ||
| finalWriterSchemaOpt, maxBufferSize, instantTime, true, true, false, engineType); | ||
| Set<String> deletedRecordKeys = new HashSet<>(unMergedRecordKeys); | ||
| deletedRecordKeys.removeAll(mergedRecordKeys); | ||
| return deletedRecordKeys.stream().map(recordKey -> HoodieMetadataPayload.createRecordIndexDelete(recordKey)).collect(toList()).iterator(); |
There was a problem hiding this comment.
This is the logic to process deletes across all log files irrespective of merge mode. This is going to be a bit costly because we're doing both merged and unmerged log records scanning to figure out the deletes.
nsivabalan
requested changes
Dec 20, 2024
| finalWriterSchemaOpt, maxBufferSize, instantTime, true, true, true, engineType); | ||
| Set<String> unMergedRecordKeys = getRecordKeys(logFilePaths, dataTableMetaClient, | ||
| finalWriterSchemaOpt, maxBufferSize, instantTime, true, true, false, engineType); | ||
| Set<String> deletedRecordKeys = new HashSet<>(unMergedRecordKeys); |
There was a problem hiding this comment.
won't this also give us deleted records from a previous log file even if not touched in this commit of interest?
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
Change Logs
Impact
Index correctness with custom mergers.
Risk level (write none, low medium or high below)
medium
Documentation Update
Describe any necessary documentation update if there is any new feature, config, or user-facing change. If not, put "none".
ticket number here and follow the instruction to make
changes to the website.
Contributor's checklist