You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@paimon.apache.org by "leaves12138 (via GitHub)" <gi...@apache.org> on 2026/05/12 06:04:51 UTC
[PR] [codex] Add manifest sort compaction [paimon]
leaves12138 opened a new pull request, #7826:
URL: https://github.com/apache/paimon/pull/7826
## Summary
- add manifest-sort.enabled and manifest-sort.rewrite-manifest-count options
- add independent manifest sort compaction that groups ManifestFileMeta by partition-range sorted runs
- wire the sort compaction after existing manifest merge during snapshot/manifest compact commits
- add focused manifest metadata tests for partial target-run rewrite and delete-manifest skip behavior
## Validation
- mvn -pl paimon-core -am -DfailIfNoTests=false -Dtest=ManifestFileMetaTest test
- mvn -pl paimon-core -am -Pfast-build -DfailIfNoTests=false -Dtest=FileStoreCommitTest#testManifestCompact test
--
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: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [core] Add manifest sort compaction [paimon]
Posted by "discivigour (via GitHub)" <gi...@apache.org>.
discivigour commented on code in PR #7826:
URL: https://github.com/apache/paimon/pull/7826#discussion_r3226824300
##########
paimon-core/src/main/java/org/apache/paimon/operation/FileStoreCommitImpl.java:
##########
@@ -1164,6 +1153,87 @@ private boolean compactManifestOnce() {
return commitSnapshotImpl(newSnapshot, emptyList());
}
+ private List<ManifestFileMeta> mergeManifests(
+ List<ManifestFileMeta> input,
+ int suggestedMinMetaCount,
+ long manifestFullCompactionSize) {
+ List<ManifestFileMeta> newFilesForAbort = new ArrayList<>();
+ try {
+ return mergeManifests(
+ input, suggestedMinMetaCount, manifestFullCompactionSize, newFilesForAbort);
+ } catch (Throwable e) {
+ for (ManifestFileMeta manifest : newFilesForAbort) {
+ manifestFile.delete(manifest.fileName());
+ }
+ if (e instanceof Error) {
+ throw (Error) e;
+ }
+ throw new RuntimeException(e);
+ }
+ }
+
+ private List<ManifestFileMeta> mergeManifests(
+ List<ManifestFileMeta> input,
+ int suggestedMinMetaCount,
+ long manifestFullCompactionSize,
+ List<ManifestFileMeta> newFilesForAbort)
+ throws Exception {
+ List<ManifestFileMeta> merged =
+ ManifestFileMerger.merge(
+ input,
+ manifestFile,
+ options.manifestTargetSize().getBytes(),
+ suggestedMinMetaCount,
+ manifestFullCompactionSize,
+ partitionType,
+ options.scanManifestParallelism());
+ addCreatedManifestFiles(input, merged, newFilesForAbort);
+ if (options.manifestSortEnabled()) {
+ Optional<List<ManifestFileMeta>> sorted =
+ ManifestFileMerger.trySortCompaction(
Review Comment:
Perhaps we shouldn't perform manifest sort when the user commits.
--
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: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [core] Add manifest sort compaction [paimon]
Posted by "discivigour (via GitHub)" <gi...@apache.org>.
discivigour commented on code in PR #7826:
URL: https://github.com/apache/paimon/pull/7826#discussion_r3226790272
##########
paimon-core/src/main/java/org/apache/paimon/operation/ManifestFileMerger.java:
##########
@@ -155,6 +161,276 @@ private static void mergeCandidates(
}
}
+ public static Optional<List<ManifestFileMeta>> trySortCompaction(
+ List<ManifestFileMeta> input,
+ List<ManifestFileMeta> newFilesForAbort,
+ ManifestFile manifestFile,
+ RowType partitionType,
+ int rewriteManifestCount,
+ @Nullable Integer manifestReadParallelism)
+ throws Exception {
+ checkArgument(
+ rewriteManifestCount > 0,
+ "Manifest sort rewrite manifest count must be greater than 0.");
+
+ if (partitionType.getFieldCount() == 0 || input.size() <= 1) {
+ return Optional.empty();
+ }
+
+ // Sort compaction may move a rewritten group before non-rewritten manifests. Even though
+ // rewriteGroups keeps the input order inside each group, it cannot preserve ordering
+ // dependencies between a DELETE entry and an ADD entry in manifests outside the group.
+ for (ManifestFileMeta file : input) {
+ if (file.numDeletedFiles() > 0) {
+ return Optional.empty();
+ }
+ }
+
+ RecordComparator partitionComparator =
+ CodeGenUtils.newRecordComparator(partitionType.getFieldTypes());
+ List<ManifestFileMeta> compactCandidates = compactCandidates(input, partitionComparator);
+ if (compactCandidates.size() <= 1) {
+ return Optional.empty();
+ }
+
+ List<ManifestSortedRun> runs = partitionSortedRuns(compactCandidates, partitionComparator);
+ if (runs.size() <= 1) {
+ return Optional.empty();
+ }
+
+ List<ManifestSortedRun> pickedRuns = pickRuns(runs, rewriteManifestCount);
+ if (pickedRuns.isEmpty()) {
+ return Optional.empty();
+ }
+
+ List<ManifestFileMeta> pickedFiles = new ArrayList<>();
+ for (ManifestSortedRun run : pickedRuns) {
+ pickedFiles.addAll(run.files);
+ }
+ List<List<ManifestFileMeta>> rewriteGroups =
+ rewriteGroups(pickedFiles, input, partitionComparator);
+ LinkedHashSet<String> rewriteFileNames = new LinkedHashSet<>();
+ for (List<ManifestFileMeta> rewriteGroup : rewriteGroups) {
+ addFileNames(rewriteGroup, rewriteFileNames);
+ }
+ if (rewriteFileNames.size() <= 1) {
+ return Optional.empty();
+ }
+
+ LOG.info(
+ "Start Manifest File Sort Compaction: sortedRuns: {}, pickedRuns: {}, pickedFiles: {}",
+ runs.size(),
+ pickedRuns.size(),
+ pickedFiles.size());
+
+ int insertPos = -1;
+ for (int i = 0; i < input.size(); i++) {
+ ManifestFileMeta file = input.get(i);
+ if (rewriteFileNames.contains(file.fileName())) {
+ insertPos = i;
+ break;
+ }
+ }
+
+ List<ManifestFileMeta> rewritten = new ArrayList<>();
+ for (List<ManifestFileMeta> rewriteGroup : rewriteGroups) {
+ Map<FileEntry.Identifier, ManifestEntry> map = new LinkedHashMap<>();
Review Comment:
If 1KB and many 8MB are continuously picked, I think the sort write amplification is a bit large.
##########
paimon-core/src/main/java/org/apache/paimon/operation/ManifestFileMerger.java:
##########
@@ -155,6 +161,276 @@ private static void mergeCandidates(
}
}
+ public static Optional<List<ManifestFileMeta>> trySortCompaction(
+ List<ManifestFileMeta> input,
+ List<ManifestFileMeta> newFilesForAbort,
+ ManifestFile manifestFile,
+ RowType partitionType,
+ int rewriteManifestCount,
+ @Nullable Integer manifestReadParallelism)
+ throws Exception {
+ checkArgument(
+ rewriteManifestCount > 0,
+ "Manifest sort rewrite manifest count must be greater than 0.");
+
+ if (partitionType.getFieldCount() == 0 || input.size() <= 1) {
+ return Optional.empty();
+ }
+
+ // Sort compaction may move a rewritten group before non-rewritten manifests. Even though
+ // rewriteGroups keeps the input order inside each group, it cannot preserve ordering
+ // dependencies between a DELETE entry and an ADD entry in manifests outside the group.
+ for (ManifestFileMeta file : input) {
+ if (file.numDeletedFiles() > 0) {
+ return Optional.empty();
+ }
+ }
+
+ RecordComparator partitionComparator =
+ CodeGenUtils.newRecordComparator(partitionType.getFieldTypes());
+ List<ManifestFileMeta> compactCandidates = compactCandidates(input, partitionComparator);
+ if (compactCandidates.size() <= 1) {
+ return Optional.empty();
+ }
+
+ List<ManifestSortedRun> runs = partitionSortedRuns(compactCandidates, partitionComparator);
+ if (runs.size() <= 1) {
+ return Optional.empty();
+ }
+
+ List<ManifestSortedRun> pickedRuns = pickRuns(runs, rewriteManifestCount);
+ if (pickedRuns.isEmpty()) {
+ return Optional.empty();
+ }
+
+ List<ManifestFileMeta> pickedFiles = new ArrayList<>();
+ for (ManifestSortedRun run : pickedRuns) {
+ pickedFiles.addAll(run.files);
+ }
+ List<List<ManifestFileMeta>> rewriteGroups =
+ rewriteGroups(pickedFiles, input, partitionComparator);
+ LinkedHashSet<String> rewriteFileNames = new LinkedHashSet<>();
+ for (List<ManifestFileMeta> rewriteGroup : rewriteGroups) {
+ addFileNames(rewriteGroup, rewriteFileNames);
+ }
+ if (rewriteFileNames.size() <= 1) {
+ return Optional.empty();
+ }
+
+ LOG.info(
+ "Start Manifest File Sort Compaction: sortedRuns: {}, pickedRuns: {}, pickedFiles: {}",
+ runs.size(),
+ pickedRuns.size(),
+ pickedFiles.size());
+
+ int insertPos = -1;
+ for (int i = 0; i < input.size(); i++) {
+ ManifestFileMeta file = input.get(i);
+ if (rewriteFileNames.contains(file.fileName())) {
+ insertPos = i;
+ break;
+ }
+ }
+
+ List<ManifestFileMeta> rewritten = new ArrayList<>();
+ for (List<ManifestFileMeta> rewriteGroup : rewriteGroups) {
+ Map<FileEntry.Identifier, ManifestEntry> map = new LinkedHashMap<>();
Review Comment:
If 1KB and many 8MBs are continuously picked, I think the sort write amplification is a bit large.
--
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: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [core] Add manifest sort compaction [paimon]
Posted by "discivigour (via GitHub)" <gi...@apache.org>.
discivigour commented on code in PR #7826:
URL: https://github.com/apache/paimon/pull/7826#discussion_r3226824300
##########
paimon-core/src/main/java/org/apache/paimon/operation/FileStoreCommitImpl.java:
##########
@@ -1164,6 +1153,87 @@ private boolean compactManifestOnce() {
return commitSnapshotImpl(newSnapshot, emptyList());
}
+ private List<ManifestFileMeta> mergeManifests(
+ List<ManifestFileMeta> input,
+ int suggestedMinMetaCount,
+ long manifestFullCompactionSize) {
+ List<ManifestFileMeta> newFilesForAbort = new ArrayList<>();
+ try {
+ return mergeManifests(
+ input, suggestedMinMetaCount, manifestFullCompactionSize, newFilesForAbort);
+ } catch (Throwable e) {
+ for (ManifestFileMeta manifest : newFilesForAbort) {
+ manifestFile.delete(manifest.fileName());
+ }
+ if (e instanceof Error) {
+ throw (Error) e;
+ }
+ throw new RuntimeException(e);
+ }
+ }
+
+ private List<ManifestFileMeta> mergeManifests(
+ List<ManifestFileMeta> input,
+ int suggestedMinMetaCount,
+ long manifestFullCompactionSize,
+ List<ManifestFileMeta> newFilesForAbort)
+ throws Exception {
+ List<ManifestFileMeta> merged =
+ ManifestFileMerger.merge(
+ input,
+ manifestFile,
+ options.manifestTargetSize().getBytes(),
+ suggestedMinMetaCount,
+ manifestFullCompactionSize,
+ partitionType,
+ options.scanManifestParallelism());
+ addCreatedManifestFiles(input, merged, newFilesForAbort);
+ if (options.manifestSortEnabled()) {
+ Optional<List<ManifestFileMeta>> sorted =
+ ManifestFileMerger.trySortCompaction(
Review Comment:
Perhaps we shouldn't perform manifest sort when the user commits.
--
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: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [core] Add manifest sort compaction [paimon]
Posted by "discivigour (via GitHub)" <gi...@apache.org>.
discivigour commented on code in PR #7826:
URL: https://github.com/apache/paimon/pull/7826#discussion_r3226790272
##########
paimon-core/src/main/java/org/apache/paimon/operation/ManifestFileMerger.java:
##########
@@ -155,6 +161,276 @@ private static void mergeCandidates(
}
}
+ public static Optional<List<ManifestFileMeta>> trySortCompaction(
+ List<ManifestFileMeta> input,
+ List<ManifestFileMeta> newFilesForAbort,
+ ManifestFile manifestFile,
+ RowType partitionType,
+ int rewriteManifestCount,
+ @Nullable Integer manifestReadParallelism)
+ throws Exception {
+ checkArgument(
+ rewriteManifestCount > 0,
+ "Manifest sort rewrite manifest count must be greater than 0.");
+
+ if (partitionType.getFieldCount() == 0 || input.size() <= 1) {
+ return Optional.empty();
+ }
+
+ // Sort compaction may move a rewritten group before non-rewritten manifests. Even though
+ // rewriteGroups keeps the input order inside each group, it cannot preserve ordering
+ // dependencies between a DELETE entry and an ADD entry in manifests outside the group.
+ for (ManifestFileMeta file : input) {
+ if (file.numDeletedFiles() > 0) {
+ return Optional.empty();
+ }
+ }
+
+ RecordComparator partitionComparator =
+ CodeGenUtils.newRecordComparator(partitionType.getFieldTypes());
+ List<ManifestFileMeta> compactCandidates = compactCandidates(input, partitionComparator);
+ if (compactCandidates.size() <= 1) {
+ return Optional.empty();
+ }
+
+ List<ManifestSortedRun> runs = partitionSortedRuns(compactCandidates, partitionComparator);
+ if (runs.size() <= 1) {
+ return Optional.empty();
+ }
+
+ List<ManifestSortedRun> pickedRuns = pickRuns(runs, rewriteManifestCount);
+ if (pickedRuns.isEmpty()) {
+ return Optional.empty();
+ }
+
+ List<ManifestFileMeta> pickedFiles = new ArrayList<>();
+ for (ManifestSortedRun run : pickedRuns) {
+ pickedFiles.addAll(run.files);
+ }
+ List<List<ManifestFileMeta>> rewriteGroups =
+ rewriteGroups(pickedFiles, input, partitionComparator);
+ LinkedHashSet<String> rewriteFileNames = new LinkedHashSet<>();
+ for (List<ManifestFileMeta> rewriteGroup : rewriteGroups) {
+ addFileNames(rewriteGroup, rewriteFileNames);
+ }
+ if (rewriteFileNames.size() <= 1) {
+ return Optional.empty();
+ }
+
+ LOG.info(
+ "Start Manifest File Sort Compaction: sortedRuns: {}, pickedRuns: {}, pickedFiles: {}",
+ runs.size(),
+ pickedRuns.size(),
+ pickedFiles.size());
+
+ int insertPos = -1;
+ for (int i = 0; i < input.size(); i++) {
+ ManifestFileMeta file = input.get(i);
+ if (rewriteFileNames.contains(file.fileName())) {
+ insertPos = i;
+ break;
+ }
+ }
+
+ List<ManifestFileMeta> rewritten = new ArrayList<>();
+ for (List<ManifestFileMeta> rewriteGroup : rewriteGroups) {
+ Map<FileEntry.Identifier, ManifestEntry> map = new LinkedHashMap<>();
Review Comment:
If 1KB and many 8MBs are continuously picked, I think the sort write amplification is a bit large.
--
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: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [core] Add manifest sort compaction [paimon]
Posted by "discivigour (via GitHub)" <gi...@apache.org>.
discivigour commented on code in PR #7826:
URL: https://github.com/apache/paimon/pull/7826#discussion_r3226790272
##########
paimon-core/src/main/java/org/apache/paimon/operation/ManifestFileMerger.java:
##########
@@ -155,6 +161,276 @@ private static void mergeCandidates(
}
}
+ public static Optional<List<ManifestFileMeta>> trySortCompaction(
+ List<ManifestFileMeta> input,
+ List<ManifestFileMeta> newFilesForAbort,
+ ManifestFile manifestFile,
+ RowType partitionType,
+ int rewriteManifestCount,
+ @Nullable Integer manifestReadParallelism)
+ throws Exception {
+ checkArgument(
+ rewriteManifestCount > 0,
+ "Manifest sort rewrite manifest count must be greater than 0.");
+
+ if (partitionType.getFieldCount() == 0 || input.size() <= 1) {
+ return Optional.empty();
+ }
+
+ // Sort compaction may move a rewritten group before non-rewritten manifests. Even though
+ // rewriteGroups keeps the input order inside each group, it cannot preserve ordering
+ // dependencies between a DELETE entry and an ADD entry in manifests outside the group.
+ for (ManifestFileMeta file : input) {
+ if (file.numDeletedFiles() > 0) {
+ return Optional.empty();
+ }
+ }
+
+ RecordComparator partitionComparator =
+ CodeGenUtils.newRecordComparator(partitionType.getFieldTypes());
+ List<ManifestFileMeta> compactCandidates = compactCandidates(input, partitionComparator);
+ if (compactCandidates.size() <= 1) {
+ return Optional.empty();
+ }
+
+ List<ManifestSortedRun> runs = partitionSortedRuns(compactCandidates, partitionComparator);
+ if (runs.size() <= 1) {
+ return Optional.empty();
+ }
+
+ List<ManifestSortedRun> pickedRuns = pickRuns(runs, rewriteManifestCount);
+ if (pickedRuns.isEmpty()) {
+ return Optional.empty();
+ }
+
+ List<ManifestFileMeta> pickedFiles = new ArrayList<>();
+ for (ManifestSortedRun run : pickedRuns) {
+ pickedFiles.addAll(run.files);
+ }
+ List<List<ManifestFileMeta>> rewriteGroups =
+ rewriteGroups(pickedFiles, input, partitionComparator);
+ LinkedHashSet<String> rewriteFileNames = new LinkedHashSet<>();
+ for (List<ManifestFileMeta> rewriteGroup : rewriteGroups) {
+ addFileNames(rewriteGroup, rewriteFileNames);
+ }
+ if (rewriteFileNames.size() <= 1) {
+ return Optional.empty();
+ }
+
+ LOG.info(
+ "Start Manifest File Sort Compaction: sortedRuns: {}, pickedRuns: {}, pickedFiles: {}",
+ runs.size(),
+ pickedRuns.size(),
+ pickedFiles.size());
+
+ int insertPos = -1;
+ for (int i = 0; i < input.size(); i++) {
+ ManifestFileMeta file = input.get(i);
+ if (rewriteFileNames.contains(file.fileName())) {
+ insertPos = i;
+ break;
+ }
+ }
+
+ List<ManifestFileMeta> rewritten = new ArrayList<>();
+ for (List<ManifestFileMeta> rewriteGroup : rewriteGroups) {
+ Map<FileEntry.Identifier, ManifestEntry> map = new LinkedHashMap<>();
Review Comment:
If 1KB and 8MB are continuously picked, I think the sort write amplification is a bit large.
--
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: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [core] Add manifest sort compaction [paimon]
Posted by "JingsongLi (via GitHub)" <gi...@apache.org>.
JingsongLi closed pull request #7826: [core] Add manifest sort compaction
URL: https://github.com/apache/paimon/pull/7826
--
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: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org