You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@paimon.apache.org by "discivigour (via GitHub)" <gi...@apache.org> on 2026/05/13 11:41:51 UTC
[PR] [core] Support manifest sort feature when commit [paimon]
discivigour opened a new pull request, #7842:
URL: https://github.com/apache/paimon/pull/7842
### Purpose
### Tests
--
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] Support manifest sort feature when commit [paimon]
Posted by "JingsongLi (via GitHub)" <gi...@apache.org>.
JingsongLi commented on PR #7842:
URL: https://github.com/apache/paimon/pull/7842#issuecomment-4611857388
Thanks for the clarification. I understand these are intentional design choices, but I still think we need to make the contract explicit and add regression coverage, otherwise this will be hard to reason about later.
1. About `manifest-sort.max-rewrite-size`: if files outside the threshold may still be merged/deleted, then the current option name/description is misleading. It says this is the *maximum total size of manifest files to rewrite in a single sort rewrite pass* and that sections exceeding the limit are skipped. However, the current code can still process the remaining part later through `rewriteSubSegments` when `hasDefaultCompactMeta` is true. So either the code should enforce the cap strictly, or the documentation should say that the cap only limits the sorted rewrite portion and default/minor cleanup may still happen beyond it.
I am also still worried about ordering. Appending `remainingSection` to the tail means the remaining key range from the current section is emitted after later sections. If global manifest order by sort key is not a required invariant, please document that explicitly. If it is an invariant, we should keep the remaining files in their original position/order. A test that hits the budget in the middle of a section and verifies the final manifest order would settle this.
2. About `manifest.merge-min-count`: saying the parameters can still be used normally is not enough for me to verify the behavior. In the old minor merge path, the trailing candidates are kept unchanged when their count is below `manifest.merge-min-count`. In the sort path, a non-empty `defaultCompactionMap` can trigger rewrite even when the number of small manifests is below that threshold. If the intended behavior is different from the old manifest merge, that is fine, but then it should be documented as a semantic change and covered by a test. Otherwise we should preserve the old gate.
3. About boundary equality: reducing the number of SortedRuns is a reasonable optimization, but the current class comment says the intervals do not overlap. With closed partition ranges, `[1, 3]` and `[3, 5]` overlap at `3`; they do not just touch in an open-interval sense. If the intended invariant allows boundary-touching intervals in one run/section, please update the comments to state that explicitly and add a boundary test. Otherwise future readers will assume the stricter non-overlap invariant and may build incorrect logic on top of it.
So my ask is not necessarily to reject the design, but to make the semantics precise and backed by tests for these three edge cases.
--
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] Support manifest sort feature when commit [paimon]
Posted by "leaves12138 (via GitHub)" <gi...@apache.org>.
leaves12138 commented on code in PR #7842:
URL: https://github.com/apache/paimon/pull/7842#discussion_r3263390733
##########
paimon-core/src/main/java/org/apache/paimon/operation/ManifestFileSorter.java:
##########
@@ -0,0 +1,785 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.operation;
+
+import org.apache.paimon.CoreOptions;
+import org.apache.paimon.data.BinaryRow;
+import org.apache.paimon.io.RollingFileWriter;
+import org.apache.paimon.manifest.FileEntry;
+import org.apache.paimon.manifest.ManifestEntry;
+import org.apache.paimon.manifest.ManifestFile;
+import org.apache.paimon.manifest.ManifestFileMeta;
+import org.apache.paimon.operation.ManifestFileMerger.FullCompactionReadResult;
+import org.apache.paimon.partition.PartitionPredicate;
+import org.apache.paimon.types.DataType;
+import org.apache.paimon.types.DecimalType;
+import org.apache.paimon.types.RowType;
+import org.apache.paimon.utils.Filter;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.annotation.Nullable;
+
+import java.util.ArrayList;
+import java.util.Comparator;
+import java.util.HashSet;
+import java.util.Iterator;
+import java.util.LinkedHashMap;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.PriorityQueue;
+import java.util.Set;
+import java.util.function.Function;
+
+import static java.util.Collections.singletonList;
+import static org.apache.paimon.utils.ManifestReadThreadPool.sequentialBatchedExecute;
+
+/** Manifest file sorter that sorts and rewrites manifest files by a configured partition field. */
+public class ManifestFileSorter {
+
+ private static final Logger LOG = LoggerFactory.getLogger(ManifestFileSorter.class);
+
+ /**
+ * Try to sort-rewrite the merged manifest list by a configured partition field. If the sort
+ * field cannot be resolved or the delta file size is below the full compaction threshold, the
+ * input is returned as-is.
+ */
+ static Optional<List<ManifestFileMeta>> trySortRewrite(
+ List<ManifestFileMeta> input,
+ List<ManifestFileMeta> newFilesForAbort,
+ ManifestFile manifestFile,
+ RowType partitionType,
+ CoreOptions options)
+ throws Exception {
+ // Extract configuration from options
+ long suggestedMetaSize = options.manifestTargetSize().getBytes();
+ Integer manifestReadParallelism = options.scanManifestParallelism();
+ String sortPartitionField = options.manifestSortPartitionField();
+ // Step 1: Resolve sort field.
+ String sortField = resolveSortField(sortPartitionField, partitionType);
+ if (sortField == null) {
+ throw new IllegalArgumentException(
+ "Cannot resolve sort field for manifest sort rewrite.");
+ }
+ int sortFieldIndex = partitionType.getFieldNames().indexOf(sortField);
+ DataType sortFieldType = partitionType.getTypeAt(sortFieldIndex);
+
+ // Step 2: Classify manifests into defaultCompaction and LSM.
+ ClassifyResult classified =
+ classifyManifests(
+ input,
+ suggestedMetaSize,
+ manifestFile,
+ partitionType,
+ manifestReadParallelism);
+ Map<ManifestFileMeta, boolean[]> defaultCompactionMap =
+ classified.defaultCompactionManifests;
+ List<ManifestFileMeta> lsmFiles = classified.lsmFiles;
+ Set<FileEntry.Identifier> deleteEntries = classified.deleteEntries;
+
+ // Step 3: Build LSM Tree and assign levels (only for lsmFiles).
+ List<ManifestSortedRun> levelRuns =
+ lsmFiles.isEmpty()
+ ? new ArrayList<>()
+ : buildLevelSortedRuns(lsmFiles, sortFieldIndex, sortFieldType);
+
+ // Step 4: Pick runs to compact.
+ ManifestPickStrategy pickStrategy =
+ new ManifestPickStrategy(
+ options.maxSizeAmplificationPercent(), options.sortedRunSizeRatio());
+ List<ManifestSortedRun> pickedRuns = pickStrategy.pick(levelRuns);
+
+ if (pickedRuns.isEmpty() && defaultCompactionMap.isEmpty()) {
+ LOG.debug(
+ "Manifest sort rewrite skipped: no runs picked and no defaultCompaction files.");
+ return Optional.empty();
+ }
+
+ LOG.info(
+ "Manifest sort rewrite: input={} files, lsm={} runs, picked={} runs, "
+ + "defaultCompaction={} files.",
+ input.size(),
+ levelRuns.size(),
+ pickedRuns.size(),
+ defaultCompactionMap.size());
+
+ Set<ManifestSortedRun> pickedSet = new HashSet<>(pickedRuns);
+ List<ManifestFileMeta> reusedFiles = new ArrayList<>();
+ for (ManifestSortedRun run : levelRuns) {
+ if (!pickedSet.contains(run)) {
+ reusedFiles.addAll(run.files());
+ }
+ }
+ List<ManifestFileMeta> result = new ArrayList<>(reusedFiles);
+
+ // Step 5: Split picked files into sections, sort and rewrite each.
+ List<ManifestFileMeta> pickedFiles = new ArrayList<>();
+ for (ManifestSortedRun run : pickedRuns) {
+ pickedFiles.addAll(run.files());
+ }
+ pickedFiles.addAll(defaultCompactionMap.keySet());
+
+ List<Section> sections =
+ splitIntoSections(pickedFiles, sortFieldIndex, sortFieldType, defaultCompactionMap);
+ sections = mergeSmallAdjacentSections(sections, suggestedMetaSize);
+
+ rewriteSections(
+ sections,
+ defaultCompactionMap,
+ manifestFile,
+ sortFieldIndex,
+ sortFieldType,
+ deleteEntries,
+ suggestedMetaSize,
+ options.manifestSortMaxRewriteSize(),
+ result,
+ newFilesForAbort,
+ manifestReadParallelism);
+
+ LOG.info(
+ "Manifest sort rewrite completed: sections={}, newFiles={}, resultFiles={}.",
+ sections.size(),
+ newFilesForAbort.size(),
+ result.size());
+ return Optional.of(result);
+ }
+
+ /**
+ * Classify manifest files into default-compaction group and LSM group.
+ *
+ * <p>When full compaction is triggered (totalDeltaFileSize >= threshold), files that must
+ * change or overlap with delete partitions go into defaultCompactionManifests; the rest stay as
+ * lsmFiles.
+ *
+ * <p>When full compaction is NOT triggered, adjacent small manifests whose cumulative size
+ * reaches suggestedMetaSize are grouped into defaultCompactionManifests (minor-style pick).
+ */
+ private static ClassifyResult classifyManifests(
+ List<ManifestFileMeta> input,
+ long suggestedMetaSize,
+ ManifestFile manifestFile,
+ RowType partitionType,
+ @Nullable Integer manifestReadParallelism) {
+ Map<ManifestFileMeta, boolean[]> defaultCompactionManifests = new LinkedHashMap<>();
+ List<ManifestFileMeta> lsmFiles = new LinkedList<>(input);
+ Set<FileEntry.Identifier> deleteEntries =
+ FileEntry.readDeletedEntries(manifestFile, input, manifestReadParallelism);
+
+ PartitionPredicate predicate;
+ if (deleteEntries.isEmpty()) {
+ predicate = PartitionPredicate.ALWAYS_FALSE;
+ } else {
+ if (partitionType.getFieldCount() > 0) {
+ Set<BinaryRow> deletePartitions =
+ ManifestFileMerger.computeDeletePartitions(deleteEntries);
+ predicate = PartitionPredicate.fromMultiple(partitionType, deletePartitions);
+ } else {
+ predicate = PartitionPredicate.ALWAYS_TRUE;
+ }
+ }
+
+ Iterator<ManifestFileMeta> iterator = lsmFiles.iterator();
+ while (iterator.hasNext()) {
+ ManifestFileMeta file = iterator.next();
+ boolean small = file.fileSize() < suggestedMetaSize;
+ boolean inDeleteRange =
+ predicate != null
+ && predicate.test(
+ file.numAddedFiles() + file.numDeletedFiles(),
+ file.partitionStats().minValues(),
+ file.partitionStats().maxValues(),
+ file.partitionStats().nullCounts());
+ if (small || inDeleteRange) {
+ iterator.remove();
+ defaultCompactionManifests.put(file, new boolean[] {small, inDeleteRange});
+ }
+ }
+ return new ClassifyResult(defaultCompactionManifests, lsmFiles, deleteEntries);
+ }
+
+ /**
+ * Build level-sorted runs from a list of manifest files. Sorts files by min partition value,
+ * greedy-scans to build non-overlapping SortedRuns, then assigns levels by totalSize (Top-4
+ * largest to level 1~4, rest to level 0).
+ */
+ static List<ManifestSortedRun> buildLevelSortedRuns(
+ List<ManifestFileMeta> input, int sortFieldIndex, DataType sortFieldType) {
+ // Step 1: Sort by min value (if equal, then by max value)
+ input.sort(
+ (a, b) -> {
+ int cmp =
+ compareField(
+ a.partitionStats().minValues(),
+ b.partitionStats().minValues(),
+ sortFieldIndex,
+ sortFieldType);
+ if (cmp != 0) {
+ return cmp;
+ }
+ return compareField(
+ a.partitionStats().maxValues(),
+ b.partitionStats().maxValues(),
+ sortFieldIndex,
+ sortFieldType);
+ });
+
+ // Step 2: Interval graph coloring algorithm - assign files to runs
+ // Use priority queue to track runs by their max values
+ PriorityQueue<List<ManifestFileMeta>> runs =
+ new PriorityQueue<>(
+ (r1, r2) -> {
+ ManifestFileMeta last1 = r1.get(r1.size() - 1);
+ ManifestFileMeta last2 = r2.get(r2.size() - 1);
+ return compareField(
+ last1.partitionStats().maxValues(),
+ last2.partitionStats().maxValues(),
+ sortFieldIndex,
+ sortFieldType);
+ });
+
+ for (ManifestFileMeta file : input) {
+ boolean addedToExisting = false;
+
+ // Try to find a run where current file's min >= run's max
+ if (!runs.isEmpty()) {
+ List<ManifestFileMeta> earliestRun = runs.peek();
+ ManifestFileMeta last = earliestRun.get(earliestRun.size() - 1);
+
+ if (compareField(
+ file.partitionStats().minValues(),
+ last.partitionStats().maxValues(),
+ sortFieldIndex,
+ sortFieldType)
+ >= 0) {
+ // Current file can be added to this run
+ runs.poll();
+ earliestRun.add(file);
+ runs.offer(earliestRun);
+ addedToExisting = true;
+ }
+ }
+
+ if (!addedToExisting) {
Review Comment:
Do not use boolean addedToExisting.
Just
List<ManifestFileMeta> earliestRun = runs.pool();
if (earliestRun == null) {
do something
} else if (compare(xxx) > 0) {
do something
} else {
do something
}
It makes this more pretty
--
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] Support manifest sort feature when commit [paimon]
Posted by "discivigour (via GitHub)" <gi...@apache.org>.
discivigour commented on PR #7842:
URL: https://github.com/apache/paimon/pull/7842#issuecomment-4493949954
> LGTM. Please check the two questions: 1、If sorted run has files overlap, is it correct? 2、If delete comes, how to deal with it. Nothing else to me.
1. If the file endpoints in SortedRun overlap, it does not affect the subsequent reconstruction of the LSM Tree. In addition, when users filter through partitions, they will only read 1-2 more files.
2. If a delete entry comes, it will first determine whether achieve manifestFullCompactionThresholdSize.If achieved, will eliminate all the delete; If not reach, the meta related to the delete entry partition will be retained and not participate in the sorting to prevent the order of add and delete from being disrupted
--
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] Support manifest sort feature when commit [paimon]
Posted by "leaves12138 (via GitHub)" <gi...@apache.org>.
leaves12138 commented on code in PR #7842:
URL: https://github.com/apache/paimon/pull/7842#discussion_r3263362636
##########
paimon-core/src/main/java/org/apache/paimon/operation/ManifestFileSorter.java:
##########
@@ -0,0 +1,785 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.operation;
+
+import org.apache.paimon.CoreOptions;
+import org.apache.paimon.data.BinaryRow;
+import org.apache.paimon.io.RollingFileWriter;
+import org.apache.paimon.manifest.FileEntry;
+import org.apache.paimon.manifest.ManifestEntry;
+import org.apache.paimon.manifest.ManifestFile;
+import org.apache.paimon.manifest.ManifestFileMeta;
+import org.apache.paimon.operation.ManifestFileMerger.FullCompactionReadResult;
+import org.apache.paimon.partition.PartitionPredicate;
+import org.apache.paimon.types.DataType;
+import org.apache.paimon.types.DecimalType;
+import org.apache.paimon.types.RowType;
+import org.apache.paimon.utils.Filter;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.annotation.Nullable;
+
+import java.util.ArrayList;
+import java.util.Comparator;
+import java.util.HashSet;
+import java.util.Iterator;
+import java.util.LinkedHashMap;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.PriorityQueue;
+import java.util.Set;
+import java.util.function.Function;
+
+import static java.util.Collections.singletonList;
+import static org.apache.paimon.utils.ManifestReadThreadPool.sequentialBatchedExecute;
+
+/** Manifest file sorter that sorts and rewrites manifest files by a configured partition field. */
+public class ManifestFileSorter {
+
+ private static final Logger LOG = LoggerFactory.getLogger(ManifestFileSorter.class);
+
+ /**
+ * Try to sort-rewrite the merged manifest list by a configured partition field. If the sort
+ * field cannot be resolved or the delta file size is below the full compaction threshold, the
+ * input is returned as-is.
+ */
+ static Optional<List<ManifestFileMeta>> trySortRewrite(
+ List<ManifestFileMeta> input,
+ List<ManifestFileMeta> newFilesForAbort,
+ ManifestFile manifestFile,
+ RowType partitionType,
+ CoreOptions options)
+ throws Exception {
+ // Extract configuration from options
+ long suggestedMetaSize = options.manifestTargetSize().getBytes();
+ Integer manifestReadParallelism = options.scanManifestParallelism();
+ String sortPartitionField = options.manifestSortPartitionField();
+ // Step 1: Resolve sort field.
+ String sortField = resolveSortField(sortPartitionField, partitionType);
+ if (sortField == null) {
+ throw new IllegalArgumentException(
+ "Cannot resolve sort field for manifest sort rewrite.");
+ }
+ int sortFieldIndex = partitionType.getFieldNames().indexOf(sortField);
+ DataType sortFieldType = partitionType.getTypeAt(sortFieldIndex);
+
+ // Step 2: Classify manifests into defaultCompaction and LSM.
+ ClassifyResult classified =
+ classifyManifests(
+ input,
+ suggestedMetaSize,
+ manifestFile,
+ partitionType,
+ manifestReadParallelism);
+ Map<ManifestFileMeta, boolean[]> defaultCompactionMap =
+ classified.defaultCompactionManifests;
+ List<ManifestFileMeta> lsmFiles = classified.lsmFiles;
+ Set<FileEntry.Identifier> deleteEntries = classified.deleteEntries;
+
+ // Step 3: Build LSM Tree and assign levels (only for lsmFiles).
+ List<ManifestSortedRun> levelRuns =
+ lsmFiles.isEmpty()
+ ? new ArrayList<>()
+ : buildLevelSortedRuns(lsmFiles, sortFieldIndex, sortFieldType);
+
+ // Step 4: Pick runs to compact.
+ ManifestPickStrategy pickStrategy =
+ new ManifestPickStrategy(
+ options.maxSizeAmplificationPercent(), options.sortedRunSizeRatio());
+ List<ManifestSortedRun> pickedRuns = pickStrategy.pick(levelRuns);
+
+ if (pickedRuns.isEmpty() && defaultCompactionMap.isEmpty()) {
+ LOG.debug(
+ "Manifest sort rewrite skipped: no runs picked and no defaultCompaction files.");
+ return Optional.empty();
+ }
+
+ LOG.info(
+ "Manifest sort rewrite: input={} files, lsm={} runs, picked={} runs, "
+ + "defaultCompaction={} files.",
+ input.size(),
+ levelRuns.size(),
+ pickedRuns.size(),
+ defaultCompactionMap.size());
+
+ Set<ManifestSortedRun> pickedSet = new HashSet<>(pickedRuns);
+ List<ManifestFileMeta> reusedFiles = new ArrayList<>();
+ for (ManifestSortedRun run : levelRuns) {
+ if (!pickedSet.contains(run)) {
+ reusedFiles.addAll(run.files());
+ }
+ }
+ List<ManifestFileMeta> result = new ArrayList<>(reusedFiles);
+
+ // Step 5: Split picked files into sections, sort and rewrite each.
+ List<ManifestFileMeta> pickedFiles = new ArrayList<>();
+ for (ManifestSortedRun run : pickedRuns) {
+ pickedFiles.addAll(run.files());
+ }
+ pickedFiles.addAll(defaultCompactionMap.keySet());
+
+ List<Section> sections =
+ splitIntoSections(pickedFiles, sortFieldIndex, sortFieldType, defaultCompactionMap);
+ sections = mergeSmallAdjacentSections(sections, suggestedMetaSize);
+
+ rewriteSections(
+ sections,
+ defaultCompactionMap,
+ manifestFile,
+ sortFieldIndex,
+ sortFieldType,
+ deleteEntries,
+ suggestedMetaSize,
+ options.manifestSortMaxRewriteSize(),
+ result,
+ newFilesForAbort,
+ manifestReadParallelism);
+
+ LOG.info(
+ "Manifest sort rewrite completed: sections={}, newFiles={}, resultFiles={}.",
+ sections.size(),
+ newFilesForAbort.size(),
+ result.size());
+ return Optional.of(result);
+ }
+
+ /**
+ * Classify manifest files into default-compaction group and LSM group.
+ *
+ * <p>When full compaction is triggered (totalDeltaFileSize >= threshold), files that must
+ * change or overlap with delete partitions go into defaultCompactionManifests; the rest stay as
+ * lsmFiles.
+ *
+ * <p>When full compaction is NOT triggered, adjacent small manifests whose cumulative size
+ * reaches suggestedMetaSize are grouped into defaultCompactionManifests (minor-style pick).
+ */
+ private static ClassifyResult classifyManifests(
+ List<ManifestFileMeta> input,
+ long suggestedMetaSize,
+ ManifestFile manifestFile,
+ RowType partitionType,
+ @Nullable Integer manifestReadParallelism) {
+ Map<ManifestFileMeta, boolean[]> defaultCompactionManifests = new LinkedHashMap<>();
+ List<ManifestFileMeta> lsmFiles = new LinkedList<>(input);
+ Set<FileEntry.Identifier> deleteEntries =
+ FileEntry.readDeletedEntries(manifestFile, input, manifestReadParallelism);
+
+ PartitionPredicate predicate;
+ if (deleteEntries.isEmpty()) {
+ predicate = PartitionPredicate.ALWAYS_FALSE;
+ } else {
+ if (partitionType.getFieldCount() > 0) {
+ Set<BinaryRow> deletePartitions =
+ ManifestFileMerger.computeDeletePartitions(deleteEntries);
+ predicate = PartitionPredicate.fromMultiple(partitionType, deletePartitions);
+ } else {
+ predicate = PartitionPredicate.ALWAYS_TRUE;
+ }
+ }
+
+ Iterator<ManifestFileMeta> iterator = lsmFiles.iterator();
+ while (iterator.hasNext()) {
+ ManifestFileMeta file = iterator.next();
+ boolean small = file.fileSize() < suggestedMetaSize;
+ boolean inDeleteRange =
+ predicate != null
+ && predicate.test(
+ file.numAddedFiles() + file.numDeletedFiles(),
+ file.partitionStats().minValues(),
+ file.partitionStats().maxValues(),
+ file.partitionStats().nullCounts());
+ if (small || inDeleteRange) {
+ iterator.remove();
+ defaultCompactionManifests.put(file, new boolean[] {small, inDeleteRange});
+ }
+ }
+ return new ClassifyResult(defaultCompactionManifests, lsmFiles, deleteEntries);
+ }
+
+ /**
+ * Build level-sorted runs from a list of manifest files. Sorts files by min partition value,
+ * greedy-scans to build non-overlapping SortedRuns, then assigns levels by totalSize (Top-4
+ * largest to level 1~4, rest to level 0).
+ */
+ static List<ManifestSortedRun> buildLevelSortedRuns(
+ List<ManifestFileMeta> input, int sortFieldIndex, DataType sortFieldType) {
+ // Step 1: Sort by min value (if equal, then by max value)
+ input.sort(
+ (a, b) -> {
+ int cmp =
+ compareField(
+ a.partitionStats().minValues(),
+ b.partitionStats().minValues(),
+ sortFieldIndex,
+ sortFieldType);
+ if (cmp != 0) {
+ return cmp;
+ }
+ return compareField(
+ a.partitionStats().maxValues(),
+ b.partitionStats().maxValues(),
+ sortFieldIndex,
+ sortFieldType);
+ });
+
+ // Step 2: Interval graph coloring algorithm - assign files to runs
+ // Use priority queue to track runs by their max values
+ PriorityQueue<List<ManifestFileMeta>> runs =
+ new PriorityQueue<>(
+ (r1, r2) -> {
+ ManifestFileMeta last1 = r1.get(r1.size() - 1);
+ ManifestFileMeta last2 = r2.get(r2.size() - 1);
+ return compareField(
+ last1.partitionStats().maxValues(),
+ last2.partitionStats().maxValues(),
+ sortFieldIndex,
+ sortFieldType);
+ });
+
+ for (ManifestFileMeta file : input) {
+ boolean addedToExisting = false;
+
+ // Try to find a run where current file's min >= run's max
+ if (!runs.isEmpty()) {
+ List<ManifestFileMeta> earliestRun = runs.peek();
+ ManifestFileMeta last = earliestRun.get(earliestRun.size() - 1);
+
+ if (compareField(
+ file.partitionStats().minValues(),
+ last.partitionStats().maxValues(),
+ sortFieldIndex,
+ sortFieldType)
+ >= 0) {
+ // Current file can be added to this run
+ runs.poll();
+ earliestRun.add(file);
+ runs.offer(earliestRun);
+ addedToExisting = true;
+ }
+ }
+
+ if (!addedToExisting) {
+ // Create a new run
+ List<ManifestFileMeta> newRun = new ArrayList<>();
+ newRun.add(file);
+ runs.offer(newRun);
+ }
+ }
+
+ // Step 3: Convert to ManifestSortedRun list
+ List<ManifestSortedRun> result = new ArrayList<>();
+ while (!runs.isEmpty()) {
+ result.add(ManifestSortedRun.fromSorted(runs.poll()));
+ }
+
+ // Step 4: Sort by totalSize and assign levels
+ result.sort(Comparator.comparingLong(ManifestSortedRun::totalSize));
+ int n = result.size();
+ int maxLevel = 4;
+ for (int i = 0; i < n; i++) {
+ if (i >= n - maxLevel) {
+ result.get(i).setLevel(i - (n - maxLevel) + 1);
+ } else {
+ result.get(i).setLevel(0);
+ }
+ }
+ return result;
+ }
+
+ /**
+ * Split picked files into sections. Files with overlapping sort-key intervals go into the same
+ * section. Each section is built with pre-computed totalSize and hasDefaultCompactMeta.
+ */
+ static List<Section> splitIntoSections(
+ List<ManifestFileMeta> pickedFiles,
+ int sortFieldIndex,
+ DataType sortFieldType,
+ Map<ManifestFileMeta, boolean[]> defaultCompactionMap) {
+ pickedFiles.sort(
+ (a, b) -> {
+ int cmp =
+ compareField(
+ a.partitionStats().minValues(),
+ b.partitionStats().minValues(),
+ sortFieldIndex,
+ sortFieldType);
+ if (cmp != 0) {
+ return cmp;
+ }
+ return compareField(
+ a.partitionStats().maxValues(),
+ b.partitionStats().maxValues(),
+ sortFieldIndex,
+ sortFieldType);
+ });
+
+ List<Section> sections = new ArrayList<>();
+ List<ManifestFileMeta> currentFiles = new ArrayList<>();
+ long currentTotalSize = 0;
+ boolean currentHasDefault = false;
+ ManifestFileMeta first = pickedFiles.get(0);
+ currentFiles.add(first);
+ currentTotalSize += first.fileSize();
+ currentHasDefault = defaultCompactionMap.containsKey(first);
+ BinaryRow sectionMaxBound = first.partitionStats().maxValues();
+
+ for (int i = 1; i < pickedFiles.size(); i++) {
+ ManifestFileMeta file = pickedFiles.get(i);
+ if (compareField(
+ file.partitionStats().minValues(),
+ sectionMaxBound,
+ sortFieldIndex,
+ sortFieldType)
+ >= 0) {
+ sections.add(new Section(currentFiles, currentTotalSize, currentHasDefault));
+ currentFiles = new ArrayList<>();
+ currentTotalSize = 0;
+ currentFiles.add(file);
+ currentTotalSize += file.fileSize();
+ currentHasDefault = defaultCompactionMap.containsKey(file);
+ sectionMaxBound = file.partitionStats().maxValues();
+ } else {
+ currentFiles.add(file);
+ currentTotalSize += file.fileSize();
+ if (!currentHasDefault && defaultCompactionMap.containsKey(file)) {
+ currentHasDefault = true;
+ }
+ if (compareField(
+ file.partitionStats().maxValues(),
+ sectionMaxBound,
+ sortFieldIndex,
+ sortFieldType)
+ > 0) {
+ sectionMaxBound = file.partitionStats().maxValues();
+ }
+ }
+ }
+ sections.add(new Section(currentFiles, currentTotalSize, currentHasDefault));
+ return sections;
+ }
+
+ /**
+ * Merge small adjacent sections to avoid producing too many small rewrite batches. If either
+ * the pending section or the current section total size is smaller than half of {@code
+ * suggestedMetaSize}, they are combined into a single section.
+ */
+ private static List<Section> mergeSmallAdjacentSections(
+ List<Section> sections, long suggestedMetaSize) {
+ List<Section> merged = new ArrayList<>();
+ Section pending = null;
+
+ for (Section section : sections) {
+ if (pending == null) {
+ pending = section;
+ } else {
+ if (pending.totalSize < suggestedMetaSize
+ || section.totalSize < suggestedMetaSize) {
+ pending = Section.merge(pending, section);
+ } else {
+ merged.add(pending);
+ pending = section;
+ }
+ }
+ }
+ if (pending != null) {
+ merged.add(pending);
+ }
+ return merged;
+ }
+
+ /**
+ * Iterate over sections, decide whether to rewrite each section fully or partially based on the
+ * maxRewriteSize threshold and whether the section contains defaultCompaction files.
+ *
+ * <p>Within threshold: read all metas, sort and rewrite the entire section. Exceeds threshold
+ * but contains defaultCompaction files: only rewrite sub-segments around those files. Exceeds
+ * threshold with no defaultCompaction files: skip (keep as-is).
+ */
+ private static void rewriteSections(
+ List<Section> sections,
+ Map<ManifestFileMeta, boolean[]> defaultCompactionMap,
+ ManifestFile manifestFile,
+ int sortFieldIndex,
+ DataType sortFieldType,
+ Set<FileEntry.Identifier> deleteEntries,
+ long suggestedMetaSize,
+ long maxRewriteSize,
+ List<ManifestFileMeta> result,
+ List<ManifestFileMeta> sortNewFiles,
+ @Nullable Integer manifestReadParallelism)
+ throws Exception {
+ long processedSize = 0;
+ boolean reachedLimit = false;
+
+ for (int i = 0; i < sections.size(); i++) {
+ Section section = sections.get(i);
+ if (section.files.size() == 1) {
+ sortAndRewriteSection(
+ section.files,
+ manifestFile,
+ sortFieldIndex,
+ sortFieldType,
+ deleteEntries,
+ defaultCompactionMap,
+ result,
+ sortNewFiles,
+ manifestReadParallelism);
+ continue;
+ }
+
+ if (processedSize + section.totalSize <= maxRewriteSize) {
+ processedSize += section.totalSize;
+ sortAndRewriteSection(
+ section.files,
+ manifestFile,
+ sortFieldIndex,
+ sortFieldType,
+ deleteEntries,
+ defaultCompactionMap,
+ result,
+ sortNewFiles,
+ manifestReadParallelism);
+ } else if (!reachedLimit) {
+ // First time exceeding threshold without defaultCompaction:
+ // partial rewrite within remaining budget.
+ long rewriteTotalSize = maxRewriteSize - processedSize;
+ processedSize += section.totalSize;
+ // Split section into two parts: files within budget and remaining files
+ List<ManifestFileMeta> rewriteFiles = new ArrayList<>();
+ List<ManifestFileMeta> remainingFiles = new ArrayList<>();
+ long rewriteSize = 0;
+ long remainingSize = 0;
+ boolean remainingHasDefault = false;
+
+ for (ManifestFileMeta file : section.files) {
+ if (rewriteSize + file.fileSize() <= rewriteTotalSize) {
+ rewriteFiles.add(file);
+ rewriteSize += file.fileSize();
+ } else {
+ remainingFiles.add(file);
+ remainingSize += file.fileSize();
+ if (defaultCompactionMap.containsKey(file)) {
+ remainingHasDefault = true;
+ }
+ }
+ }
+
+ sortAndRewriteSection(
+ rewriteFiles,
+ manifestFile,
+ sortFieldIndex,
+ sortFieldType,
+ deleteEntries,
+ defaultCompactionMap,
+ result,
+ sortNewFiles,
+ manifestReadParallelism);
+
+ // Create new section for remaining files and append to sections list
+ if (!remainingFiles.isEmpty()) {
+ Section remainingSection =
+ new Section(remainingFiles, remainingSize, remainingHasDefault);
+ // Append remaining section to the end of sections list
+ sections.add(remainingSection);
+ }
+ reachedLimit = true;
+ } else if (section.hasDefaultCompactMeta) {
+ rewriteSubSegments(
+ section.files,
+ defaultCompactionMap,
+ manifestFile,
+ sortFieldIndex,
+ sortFieldType,
+ deleteEntries,
+ suggestedMetaSize,
+ result,
+ sortNewFiles,
+ manifestReadParallelism);
+ } else {
+ result.addAll(section.files);
+ }
+ }
+ }
+
+ /** Rewrite sub-segments within a section that exceeds the rewrite threshold. */
+ private static void rewriteSubSegments(
+ List<ManifestFileMeta> section,
+ Map<ManifestFileMeta, boolean[]> defaultCompactionMap,
+ ManifestFile manifestFile,
+ int sortFieldIndex,
+ DataType sortFieldType,
+ @Nullable Set<FileEntry.Identifier> deleteEntries,
+ long manifestTargetSize,
+ List<ManifestFileMeta> result,
+ List<ManifestFileMeta> sortNewFiles,
+ @Nullable Integer manifestReadParallelism)
+ throws Exception {
+ List<ManifestFileMeta> subSegment = new ArrayList<>();
+ long subSegmentSize = 0;
+ for (ManifestFileMeta m : section) {
+ subSegmentSize += m.fileSize();
+ subSegment.add(m);
+
+ if (subSegmentSize >= manifestTargetSize) {
+ sortAndRewriteSection(
+ subSegment,
+ manifestFile,
+ sortFieldIndex,
+ sortFieldType,
+ deleteEntries,
+ defaultCompactionMap,
+ result,
+ sortNewFiles,
+ manifestReadParallelism);
+ subSegment.clear();
+ subSegmentSize = 0;
+ }
+ }
+ // Flush remaining sub-segment
+ if (!subSegment.isEmpty()) {
+ sortAndRewriteSection(
+ subSegment,
+ manifestFile,
+ sortFieldIndex,
+ sortFieldType,
+ deleteEntries,
+ defaultCompactionMap,
+ result,
+ sortNewFiles,
+ manifestReadParallelism);
+ }
+ }
+
+ /**
+ * Read all entries from a section's manifest files, sort them in memory by the specified
+ * partition field, filter out DELETE entries and cancelled ADD entries, then write surviving
+ * entries to new manifest files via the rolling writer.
+ *
+ * <p>All files participate in sorting, enabling full sort across the entire section.
+ *
+ * <p>Reading is parallelized via {@code sequentialBatchedExecute} following the same pattern as
+ * {@link ManifestFileMerger#tryFullCompaction}.
+ */
+ private static void sortAndRewriteSection(
+ List<ManifestFileMeta> section,
+ ManifestFile manifestFile,
+ int sortFieldIndex,
+ DataType sortFieldType,
+ Set<FileEntry.Identifier> deletedIdentifiers,
+ Map<ManifestFileMeta, boolean[]> defaultCompactionMap,
+ List<ManifestFileMeta> result,
+ List<ManifestFileMeta> sortNewFiles,
+ @Nullable Integer manifestReadParallelism)
+ throws Exception {
+ if (section.size() == 1
+ && (!defaultCompactionMap.containsKey(section.get(0))
+ || !defaultCompactionMap.get(section.get(0))[1])) {
+ result.add(section.get(0));
+ return;
+ }
+ // Parallel read: each meta is read independently
+ Function<ManifestFileMeta, List<FullCompactionReadResult>> reader =
+ meta -> singletonList(readForSortRewrite(meta, manifestFile, deletedIdentifiers));
+
+ List<ManifestEntry> entriesToRewrite = new ArrayList<>();
+ for (FullCompactionReadResult readResult :
+ sequentialBatchedExecute(reader, section, manifestReadParallelism)) {
+ entriesToRewrite.addAll(readResult.entries);
+ }
+
+ if (!entriesToRewrite.isEmpty()) {
+ entriesToRewrite.sort((a, b) -> compareSortKey(a, b, sortFieldIndex, sortFieldType));
+
+ RollingFileWriter<ManifestEntry, ManifestFileMeta> writer =
+ manifestFile.createRollingWriter();
+ Exception exception = null;
+ try {
+ writer.write(entriesToRewrite);
+ } catch (Exception e) {
+ exception = e;
+ } finally {
+ if (exception != null) {
+ writer.abort();
+ throw exception;
+ }
+ writer.close();
+ }
+ List<ManifestFileMeta> sorted = writer.result();
+ result.addAll(sorted);
+ sortNewFiles.addAll(sorted);
+ }
+ }
+
+ /**
+ * Compare two {@link ManifestEntry}s by the composite key {@code (sort-field, fileName)}.
+ * {@code fileName} is used as the tie-breaker so that all entries sharing the same sort-field
+ * value AND the same data file are emitted contiguously.
+ */
+ static int compareSortKey(
+ ManifestEntry a, ManifestEntry b, int sortFieldIndex, DataType sortFieldType) {
+ int c = compareField(a.partition(), b.partition(), sortFieldIndex, sortFieldType);
+ if (c != 0) {
+ return c;
+ }
+ // ADD before DELETE
+ int kindCmp = a.kind().compareTo(b.kind());
+ if (kindCmp != 0) {
+ return kindCmp;
+ }
+ return a.file().fileName().compareTo(b.file().fileName());
+ }
+
+ /**
+ * Compares the value at field {@code k} of two {@link BinaryRow}s according to {@code type}.
+ */
+ static int compareField(BinaryRow a, BinaryRow b, int k, DataType type) {
Review Comment:
Why not use CodeGenUtils.newRecordComparator?
--
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] Support manifest sort feature when commit [paimon]
Posted by "discivigour (via GitHub)" <gi...@apache.org>.
discivigour commented on code in PR #7842:
URL: https://github.com/apache/paimon/pull/7842#discussion_r3264555497
##########
paimon-core/src/main/java/org/apache/paimon/operation/ManifestFileSorter.java:
##########
@@ -0,0 +1,785 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.operation;
+
+import org.apache.paimon.CoreOptions;
+import org.apache.paimon.data.BinaryRow;
+import org.apache.paimon.io.RollingFileWriter;
+import org.apache.paimon.manifest.FileEntry;
+import org.apache.paimon.manifest.ManifestEntry;
+import org.apache.paimon.manifest.ManifestFile;
+import org.apache.paimon.manifest.ManifestFileMeta;
+import org.apache.paimon.operation.ManifestFileMerger.FullCompactionReadResult;
+import org.apache.paimon.partition.PartitionPredicate;
+import org.apache.paimon.types.DataType;
+import org.apache.paimon.types.DecimalType;
+import org.apache.paimon.types.RowType;
+import org.apache.paimon.utils.Filter;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.annotation.Nullable;
+
+import java.util.ArrayList;
+import java.util.Comparator;
+import java.util.HashSet;
+import java.util.Iterator;
+import java.util.LinkedHashMap;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.PriorityQueue;
+import java.util.Set;
+import java.util.function.Function;
+
+import static java.util.Collections.singletonList;
+import static org.apache.paimon.utils.ManifestReadThreadPool.sequentialBatchedExecute;
+
+/** Manifest file sorter that sorts and rewrites manifest files by a configured partition field. */
+public class ManifestFileSorter {
+
+ private static final Logger LOG = LoggerFactory.getLogger(ManifestFileSorter.class);
+
+ /**
+ * Try to sort-rewrite the merged manifest list by a configured partition field. If the sort
+ * field cannot be resolved or the delta file size is below the full compaction threshold, the
+ * input is returned as-is.
+ */
+ static Optional<List<ManifestFileMeta>> trySortRewrite(
+ List<ManifestFileMeta> input,
+ List<ManifestFileMeta> newFilesForAbort,
+ ManifestFile manifestFile,
+ RowType partitionType,
+ CoreOptions options)
+ throws Exception {
+ // Extract configuration from options
+ long suggestedMetaSize = options.manifestTargetSize().getBytes();
+ Integer manifestReadParallelism = options.scanManifestParallelism();
+ String sortPartitionField = options.manifestSortPartitionField();
+ // Step 1: Resolve sort field.
+ String sortField = resolveSortField(sortPartitionField, partitionType);
+ if (sortField == null) {
+ throw new IllegalArgumentException(
+ "Cannot resolve sort field for manifest sort rewrite.");
+ }
+ int sortFieldIndex = partitionType.getFieldNames().indexOf(sortField);
+ DataType sortFieldType = partitionType.getTypeAt(sortFieldIndex);
+
+ // Step 2: Classify manifests into defaultCompaction and LSM.
+ ClassifyResult classified =
+ classifyManifests(
+ input,
+ suggestedMetaSize,
+ manifestFile,
+ partitionType,
+ manifestReadParallelism);
+ Map<ManifestFileMeta, boolean[]> defaultCompactionMap =
+ classified.defaultCompactionManifests;
+ List<ManifestFileMeta> lsmFiles = classified.lsmFiles;
+ Set<FileEntry.Identifier> deleteEntries = classified.deleteEntries;
+
+ // Step 3: Build LSM Tree and assign levels (only for lsmFiles).
+ List<ManifestSortedRun> levelRuns =
+ lsmFiles.isEmpty()
+ ? new ArrayList<>()
+ : buildLevelSortedRuns(lsmFiles, sortFieldIndex, sortFieldType);
+
+ // Step 4: Pick runs to compact.
+ ManifestPickStrategy pickStrategy =
+ new ManifestPickStrategy(
+ options.maxSizeAmplificationPercent(), options.sortedRunSizeRatio());
+ List<ManifestSortedRun> pickedRuns = pickStrategy.pick(levelRuns);
+
+ if (pickedRuns.isEmpty() && defaultCompactionMap.isEmpty()) {
+ LOG.debug(
+ "Manifest sort rewrite skipped: no runs picked and no defaultCompaction files.");
+ return Optional.empty();
+ }
+
+ LOG.info(
+ "Manifest sort rewrite: input={} files, lsm={} runs, picked={} runs, "
+ + "defaultCompaction={} files.",
+ input.size(),
+ levelRuns.size(),
+ pickedRuns.size(),
+ defaultCompactionMap.size());
+
+ Set<ManifestSortedRun> pickedSet = new HashSet<>(pickedRuns);
+ List<ManifestFileMeta> reusedFiles = new ArrayList<>();
+ for (ManifestSortedRun run : levelRuns) {
+ if (!pickedSet.contains(run)) {
+ reusedFiles.addAll(run.files());
+ }
+ }
+ List<ManifestFileMeta> result = new ArrayList<>(reusedFiles);
+
+ // Step 5: Split picked files into sections, sort and rewrite each.
+ List<ManifestFileMeta> pickedFiles = new ArrayList<>();
+ for (ManifestSortedRun run : pickedRuns) {
+ pickedFiles.addAll(run.files());
+ }
+ pickedFiles.addAll(defaultCompactionMap.keySet());
+
+ List<Section> sections =
+ splitIntoSections(pickedFiles, sortFieldIndex, sortFieldType, defaultCompactionMap);
+ sections = mergeSmallAdjacentSections(sections, suggestedMetaSize);
+
+ rewriteSections(
+ sections,
+ defaultCompactionMap,
+ manifestFile,
+ sortFieldIndex,
+ sortFieldType,
+ deleteEntries,
+ suggestedMetaSize,
+ options.manifestSortMaxRewriteSize(),
+ result,
+ newFilesForAbort,
+ manifestReadParallelism);
+
+ LOG.info(
+ "Manifest sort rewrite completed: sections={}, newFiles={}, resultFiles={}.",
+ sections.size(),
+ newFilesForAbort.size(),
+ result.size());
+ return Optional.of(result);
+ }
+
+ /**
+ * Classify manifest files into default-compaction group and LSM group.
+ *
+ * <p>When full compaction is triggered (totalDeltaFileSize >= threshold), files that must
+ * change or overlap with delete partitions go into defaultCompactionManifests; the rest stay as
+ * lsmFiles.
+ *
+ * <p>When full compaction is NOT triggered, adjacent small manifests whose cumulative size
+ * reaches suggestedMetaSize are grouped into defaultCompactionManifests (minor-style pick).
+ */
+ private static ClassifyResult classifyManifests(
+ List<ManifestFileMeta> input,
+ long suggestedMetaSize,
+ ManifestFile manifestFile,
+ RowType partitionType,
+ @Nullable Integer manifestReadParallelism) {
+ Map<ManifestFileMeta, boolean[]> defaultCompactionManifests = new LinkedHashMap<>();
+ List<ManifestFileMeta> lsmFiles = new LinkedList<>(input);
+ Set<FileEntry.Identifier> deleteEntries =
+ FileEntry.readDeletedEntries(manifestFile, input, manifestReadParallelism);
+
+ PartitionPredicate predicate;
+ if (deleteEntries.isEmpty()) {
+ predicate = PartitionPredicate.ALWAYS_FALSE;
+ } else {
+ if (partitionType.getFieldCount() > 0) {
+ Set<BinaryRow> deletePartitions =
+ ManifestFileMerger.computeDeletePartitions(deleteEntries);
+ predicate = PartitionPredicate.fromMultiple(partitionType, deletePartitions);
+ } else {
+ predicate = PartitionPredicate.ALWAYS_TRUE;
+ }
+ }
+
+ Iterator<ManifestFileMeta> iterator = lsmFiles.iterator();
+ while (iterator.hasNext()) {
+ ManifestFileMeta file = iterator.next();
+ boolean small = file.fileSize() < suggestedMetaSize;
+ boolean inDeleteRange =
+ predicate != null
+ && predicate.test(
+ file.numAddedFiles() + file.numDeletedFiles(),
+ file.partitionStats().minValues(),
+ file.partitionStats().maxValues(),
+ file.partitionStats().nullCounts());
+ if (small || inDeleteRange) {
+ iterator.remove();
+ defaultCompactionManifests.put(file, new boolean[] {small, inDeleteRange});
+ }
+ }
+ return new ClassifyResult(defaultCompactionManifests, lsmFiles, deleteEntries);
+ }
+
+ /**
+ * Build level-sorted runs from a list of manifest files. Sorts files by min partition value,
+ * greedy-scans to build non-overlapping SortedRuns, then assigns levels by totalSize (Top-4
+ * largest to level 1~4, rest to level 0).
+ */
+ static List<ManifestSortedRun> buildLevelSortedRuns(
+ List<ManifestFileMeta> input, int sortFieldIndex, DataType sortFieldType) {
+ // Step 1: Sort by min value (if equal, then by max value)
+ input.sort(
+ (a, b) -> {
+ int cmp =
+ compareField(
+ a.partitionStats().minValues(),
+ b.partitionStats().minValues(),
+ sortFieldIndex,
+ sortFieldType);
+ if (cmp != 0) {
+ return cmp;
+ }
+ return compareField(
+ a.partitionStats().maxValues(),
+ b.partitionStats().maxValues(),
+ sortFieldIndex,
+ sortFieldType);
+ });
+
+ // Step 2: Interval graph coloring algorithm - assign files to runs
+ // Use priority queue to track runs by their max values
+ PriorityQueue<List<ManifestFileMeta>> runs =
+ new PriorityQueue<>(
+ (r1, r2) -> {
+ ManifestFileMeta last1 = r1.get(r1.size() - 1);
+ ManifestFileMeta last2 = r2.get(r2.size() - 1);
+ return compareField(
+ last1.partitionStats().maxValues(),
+ last2.partitionStats().maxValues(),
+ sortFieldIndex,
+ sortFieldType);
+ });
+
+ for (ManifestFileMeta file : input) {
+ boolean addedToExisting = false;
+
+ // Try to find a run where current file's min >= run's max
+ if (!runs.isEmpty()) {
+ List<ManifestFileMeta> earliestRun = runs.peek();
+ ManifestFileMeta last = earliestRun.get(earliestRun.size() - 1);
+
+ if (compareField(
+ file.partitionStats().minValues(),
+ last.partitionStats().maxValues(),
+ sortFieldIndex,
+ sortFieldType)
+ >= 0) {
+ // Current file can be added to this run
+ runs.poll();
+ earliestRun.add(file);
+ runs.offer(earliestRun);
+ addedToExisting = true;
+ }
+ }
+
+ if (!addedToExisting) {
+ // Create a new run
+ List<ManifestFileMeta> newRun = new ArrayList<>();
+ newRun.add(file);
+ runs.offer(newRun);
+ }
+ }
+
+ // Step 3: Convert to ManifestSortedRun list
+ List<ManifestSortedRun> result = new ArrayList<>();
+ while (!runs.isEmpty()) {
+ result.add(ManifestSortedRun.fromSorted(runs.poll()));
+ }
+
+ // Step 4: Sort by totalSize and assign levels
+ result.sort(Comparator.comparingLong(ManifestSortedRun::totalSize));
+ int n = result.size();
+ int maxLevel = 4;
+ for (int i = 0; i < n; i++) {
+ if (i >= n - maxLevel) {
+ result.get(i).setLevel(i - (n - maxLevel) + 1);
+ } else {
+ result.get(i).setLevel(0);
+ }
+ }
+ return result;
+ }
+
+ /**
+ * Split picked files into sections. Files with overlapping sort-key intervals go into the same
+ * section. Each section is built with pre-computed totalSize and hasDefaultCompactMeta.
+ */
+ static List<Section> splitIntoSections(
+ List<ManifestFileMeta> pickedFiles,
+ int sortFieldIndex,
+ DataType sortFieldType,
+ Map<ManifestFileMeta, boolean[]> defaultCompactionMap) {
+ pickedFiles.sort(
+ (a, b) -> {
+ int cmp =
+ compareField(
+ a.partitionStats().minValues(),
+ b.partitionStats().minValues(),
+ sortFieldIndex,
+ sortFieldType);
+ if (cmp != 0) {
+ return cmp;
+ }
+ return compareField(
+ a.partitionStats().maxValues(),
+ b.partitionStats().maxValues(),
+ sortFieldIndex,
+ sortFieldType);
+ });
+
+ List<Section> sections = new ArrayList<>();
+ List<ManifestFileMeta> currentFiles = new ArrayList<>();
+ long currentTotalSize = 0;
+ boolean currentHasDefault = false;
+ ManifestFileMeta first = pickedFiles.get(0);
+ currentFiles.add(first);
+ currentTotalSize += first.fileSize();
+ currentHasDefault = defaultCompactionMap.containsKey(first);
+ BinaryRow sectionMaxBound = first.partitionStats().maxValues();
+
+ for (int i = 1; i < pickedFiles.size(); i++) {
+ ManifestFileMeta file = pickedFiles.get(i);
+ if (compareField(
+ file.partitionStats().minValues(),
+ sectionMaxBound,
+ sortFieldIndex,
+ sortFieldType)
+ >= 0) {
+ sections.add(new Section(currentFiles, currentTotalSize, currentHasDefault));
+ currentFiles = new ArrayList<>();
+ currentTotalSize = 0;
+ currentFiles.add(file);
+ currentTotalSize += file.fileSize();
+ currentHasDefault = defaultCompactionMap.containsKey(file);
+ sectionMaxBound = file.partitionStats().maxValues();
+ } else {
+ currentFiles.add(file);
+ currentTotalSize += file.fileSize();
+ if (!currentHasDefault && defaultCompactionMap.containsKey(file)) {
+ currentHasDefault = true;
+ }
+ if (compareField(
+ file.partitionStats().maxValues(),
+ sectionMaxBound,
+ sortFieldIndex,
+ sortFieldType)
+ > 0) {
+ sectionMaxBound = file.partitionStats().maxValues();
+ }
+ }
+ }
+ sections.add(new Section(currentFiles, currentTotalSize, currentHasDefault));
+ return sections;
+ }
+
+ /**
+ * Merge small adjacent sections to avoid producing too many small rewrite batches. If either
+ * the pending section or the current section total size is smaller than half of {@code
+ * suggestedMetaSize}, they are combined into a single section.
+ */
+ private static List<Section> mergeSmallAdjacentSections(
+ List<Section> sections, long suggestedMetaSize) {
+ List<Section> merged = new ArrayList<>();
+ Section pending = null;
+
+ for (Section section : sections) {
+ if (pending == null) {
+ pending = section;
+ } else {
+ if (pending.totalSize < suggestedMetaSize
+ || section.totalSize < suggestedMetaSize) {
+ pending = Section.merge(pending, section);
+ } else {
+ merged.add(pending);
+ pending = section;
+ }
+ }
+ }
+ if (pending != null) {
+ merged.add(pending);
+ }
+ return merged;
+ }
+
+ /**
+ * Iterate over sections, decide whether to rewrite each section fully or partially based on the
+ * maxRewriteSize threshold and whether the section contains defaultCompaction files.
+ *
+ * <p>Within threshold: read all metas, sort and rewrite the entire section. Exceeds threshold
+ * but contains defaultCompaction files: only rewrite sub-segments around those files. Exceeds
+ * threshold with no defaultCompaction files: skip (keep as-is).
+ */
+ private static void rewriteSections(
+ List<Section> sections,
+ Map<ManifestFileMeta, boolean[]> defaultCompactionMap,
+ ManifestFile manifestFile,
+ int sortFieldIndex,
+ DataType sortFieldType,
+ Set<FileEntry.Identifier> deleteEntries,
+ long suggestedMetaSize,
+ long maxRewriteSize,
+ List<ManifestFileMeta> result,
+ List<ManifestFileMeta> sortNewFiles,
+ @Nullable Integer manifestReadParallelism)
+ throws Exception {
+ long processedSize = 0;
+ boolean reachedLimit = false;
+
+ for (int i = 0; i < sections.size(); i++) {
+ Section section = sections.get(i);
+ if (section.files.size() == 1) {
+ sortAndRewriteSection(
+ section.files,
+ manifestFile,
+ sortFieldIndex,
+ sortFieldType,
+ deleteEntries,
+ defaultCompactionMap,
+ result,
+ sortNewFiles,
+ manifestReadParallelism);
+ continue;
+ }
+
+ if (processedSize + section.totalSize <= maxRewriteSize) {
+ processedSize += section.totalSize;
+ sortAndRewriteSection(
+ section.files,
+ manifestFile,
+ sortFieldIndex,
+ sortFieldType,
+ deleteEntries,
+ defaultCompactionMap,
+ result,
+ sortNewFiles,
+ manifestReadParallelism);
+ } else if (!reachedLimit) {
+ // First time exceeding threshold without defaultCompaction:
+ // partial rewrite within remaining budget.
+ long rewriteTotalSize = maxRewriteSize - processedSize;
+ processedSize += section.totalSize;
+ // Split section into two parts: files within budget and remaining files
+ List<ManifestFileMeta> rewriteFiles = new ArrayList<>();
+ List<ManifestFileMeta> remainingFiles = new ArrayList<>();
+ long rewriteSize = 0;
+ long remainingSize = 0;
+ boolean remainingHasDefault = false;
+
+ for (ManifestFileMeta file : section.files) {
+ if (rewriteSize + file.fileSize() <= rewriteTotalSize) {
+ rewriteFiles.add(file);
+ rewriteSize += file.fileSize();
+ } else {
+ remainingFiles.add(file);
+ remainingSize += file.fileSize();
+ if (defaultCompactionMap.containsKey(file)) {
+ remainingHasDefault = true;
+ }
+ }
+ }
+
+ sortAndRewriteSection(
+ rewriteFiles,
+ manifestFile,
+ sortFieldIndex,
+ sortFieldType,
+ deleteEntries,
+ defaultCompactionMap,
+ result,
+ sortNewFiles,
+ manifestReadParallelism);
+
+ // Create new section for remaining files and append to sections list
+ if (!remainingFiles.isEmpty()) {
+ Section remainingSection =
+ new Section(remainingFiles, remainingSize, remainingHasDefault);
+ // Append remaining section to the end of sections list
+ sections.add(remainingSection);
+ }
+ reachedLimit = true;
+ } else if (section.hasDefaultCompactMeta) {
+ rewriteSubSegments(
+ section.files,
+ defaultCompactionMap,
+ manifestFile,
+ sortFieldIndex,
+ sortFieldType,
+ deleteEntries,
+ suggestedMetaSize,
+ result,
+ sortNewFiles,
+ manifestReadParallelism);
+ } else {
+ result.addAll(section.files);
+ }
+ }
+ }
+
+ /** Rewrite sub-segments within a section that exceeds the rewrite threshold. */
+ private static void rewriteSubSegments(
+ List<ManifestFileMeta> section,
+ Map<ManifestFileMeta, boolean[]> defaultCompactionMap,
+ ManifestFile manifestFile,
+ int sortFieldIndex,
+ DataType sortFieldType,
+ @Nullable Set<FileEntry.Identifier> deleteEntries,
+ long manifestTargetSize,
+ List<ManifestFileMeta> result,
+ List<ManifestFileMeta> sortNewFiles,
+ @Nullable Integer manifestReadParallelism)
+ throws Exception {
+ List<ManifestFileMeta> subSegment = new ArrayList<>();
+ long subSegmentSize = 0;
+ for (ManifestFileMeta m : section) {
+ subSegmentSize += m.fileSize();
+ subSegment.add(m);
+
+ if (subSegmentSize >= manifestTargetSize) {
+ sortAndRewriteSection(
+ subSegment,
+ manifestFile,
+ sortFieldIndex,
+ sortFieldType,
+ deleteEntries,
+ defaultCompactionMap,
+ result,
+ sortNewFiles,
+ manifestReadParallelism);
+ subSegment.clear();
+ subSegmentSize = 0;
+ }
+ }
+ // Flush remaining sub-segment
+ if (!subSegment.isEmpty()) {
+ sortAndRewriteSection(
+ subSegment,
+ manifestFile,
+ sortFieldIndex,
+ sortFieldType,
+ deleteEntries,
+ defaultCompactionMap,
+ result,
+ sortNewFiles,
+ manifestReadParallelism);
+ }
+ }
+
+ /**
+ * Read all entries from a section's manifest files, sort them in memory by the specified
+ * partition field, filter out DELETE entries and cancelled ADD entries, then write surviving
+ * entries to new manifest files via the rolling writer.
+ *
+ * <p>All files participate in sorting, enabling full sort across the entire section.
+ *
+ * <p>Reading is parallelized via {@code sequentialBatchedExecute} following the same pattern as
+ * {@link ManifestFileMerger#tryFullCompaction}.
+ */
+ private static void sortAndRewriteSection(
+ List<ManifestFileMeta> section,
+ ManifestFile manifestFile,
+ int sortFieldIndex,
+ DataType sortFieldType,
+ Set<FileEntry.Identifier> deletedIdentifiers,
+ Map<ManifestFileMeta, boolean[]> defaultCompactionMap,
+ List<ManifestFileMeta> result,
+ List<ManifestFileMeta> sortNewFiles,
+ @Nullable Integer manifestReadParallelism)
+ throws Exception {
+ if (section.size() == 1
+ && (!defaultCompactionMap.containsKey(section.get(0))
+ || !defaultCompactionMap.get(section.get(0))[1])) {
+ result.add(section.get(0));
+ return;
+ }
+ // Parallel read: each meta is read independently
+ Function<ManifestFileMeta, List<FullCompactionReadResult>> reader =
+ meta -> singletonList(readForSortRewrite(meta, manifestFile, deletedIdentifiers));
+
+ List<ManifestEntry> entriesToRewrite = new ArrayList<>();
+ for (FullCompactionReadResult readResult :
+ sequentialBatchedExecute(reader, section, manifestReadParallelism)) {
+ entriesToRewrite.addAll(readResult.entries);
+ }
+
+ if (!entriesToRewrite.isEmpty()) {
+ entriesToRewrite.sort((a, b) -> compareSortKey(a, b, sortFieldIndex, sortFieldType));
+
+ RollingFileWriter<ManifestEntry, ManifestFileMeta> writer =
+ manifestFile.createRollingWriter();
+ Exception exception = null;
+ try {
+ writer.write(entriesToRewrite);
+ } catch (Exception e) {
+ exception = e;
+ } finally {
+ if (exception != null) {
+ writer.abort();
+ throw exception;
+ }
+ writer.close();
+ }
+ List<ManifestFileMeta> sorted = writer.result();
+ result.addAll(sorted);
+ sortNewFiles.addAll(sorted);
+ }
+ }
+
+ /**
+ * Compare two {@link ManifestEntry}s by the composite key {@code (sort-field, fileName)}.
+ * {@code fileName} is used as the tie-breaker so that all entries sharing the same sort-field
+ * value AND the same data file are emitted contiguously.
+ */
+ static int compareSortKey(
+ ManifestEntry a, ManifestEntry b, int sortFieldIndex, DataType sortFieldType) {
+ int c = compareField(a.partition(), b.partition(), sortFieldIndex, sortFieldType);
+ if (c != 0) {
+ return c;
+ }
+ // ADD before DELETE
+ int kindCmp = a.kind().compareTo(b.kind());
+ if (kindCmp != 0) {
+ return kindCmp;
+ }
+ return a.file().fileName().compareTo(b.file().fileName());
+ }
+
+ /**
+ * Compares the value at field {@code k} of two {@link BinaryRow}s according to {@code type}.
+ */
+ static int compareField(BinaryRow a, BinaryRow b, int k, DataType type) {
Review Comment:
Nice point.
##########
paimon-core/src/main/java/org/apache/paimon/operation/ManifestFileSorter.java:
##########
@@ -0,0 +1,785 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.operation;
+
+import org.apache.paimon.CoreOptions;
+import org.apache.paimon.data.BinaryRow;
+import org.apache.paimon.io.RollingFileWriter;
+import org.apache.paimon.manifest.FileEntry;
+import org.apache.paimon.manifest.ManifestEntry;
+import org.apache.paimon.manifest.ManifestFile;
+import org.apache.paimon.manifest.ManifestFileMeta;
+import org.apache.paimon.operation.ManifestFileMerger.FullCompactionReadResult;
+import org.apache.paimon.partition.PartitionPredicate;
+import org.apache.paimon.types.DataType;
+import org.apache.paimon.types.DecimalType;
+import org.apache.paimon.types.RowType;
+import org.apache.paimon.utils.Filter;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.annotation.Nullable;
+
+import java.util.ArrayList;
+import java.util.Comparator;
+import java.util.HashSet;
+import java.util.Iterator;
+import java.util.LinkedHashMap;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.PriorityQueue;
+import java.util.Set;
+import java.util.function.Function;
+
+import static java.util.Collections.singletonList;
+import static org.apache.paimon.utils.ManifestReadThreadPool.sequentialBatchedExecute;
+
+/** Manifest file sorter that sorts and rewrites manifest files by a configured partition field. */
+public class ManifestFileSorter {
+
+ private static final Logger LOG = LoggerFactory.getLogger(ManifestFileSorter.class);
+
+ /**
+ * Try to sort-rewrite the merged manifest list by a configured partition field. If the sort
+ * field cannot be resolved or the delta file size is below the full compaction threshold, the
+ * input is returned as-is.
+ */
+ static Optional<List<ManifestFileMeta>> trySortRewrite(
+ List<ManifestFileMeta> input,
+ List<ManifestFileMeta> newFilesForAbort,
+ ManifestFile manifestFile,
+ RowType partitionType,
+ CoreOptions options)
+ throws Exception {
+ // Extract configuration from options
+ long suggestedMetaSize = options.manifestTargetSize().getBytes();
+ Integer manifestReadParallelism = options.scanManifestParallelism();
+ String sortPartitionField = options.manifestSortPartitionField();
+ // Step 1: Resolve sort field.
+ String sortField = resolveSortField(sortPartitionField, partitionType);
+ if (sortField == null) {
+ throw new IllegalArgumentException(
+ "Cannot resolve sort field for manifest sort rewrite.");
+ }
+ int sortFieldIndex = partitionType.getFieldNames().indexOf(sortField);
+ DataType sortFieldType = partitionType.getTypeAt(sortFieldIndex);
+
+ // Step 2: Classify manifests into defaultCompaction and LSM.
+ ClassifyResult classified =
+ classifyManifests(
+ input,
+ suggestedMetaSize,
+ manifestFile,
+ partitionType,
+ manifestReadParallelism);
+ Map<ManifestFileMeta, boolean[]> defaultCompactionMap =
+ classified.defaultCompactionManifests;
+ List<ManifestFileMeta> lsmFiles = classified.lsmFiles;
+ Set<FileEntry.Identifier> deleteEntries = classified.deleteEntries;
+
+ // Step 3: Build LSM Tree and assign levels (only for lsmFiles).
+ List<ManifestSortedRun> levelRuns =
+ lsmFiles.isEmpty()
+ ? new ArrayList<>()
+ : buildLevelSortedRuns(lsmFiles, sortFieldIndex, sortFieldType);
+
+ // Step 4: Pick runs to compact.
+ ManifestPickStrategy pickStrategy =
+ new ManifestPickStrategy(
+ options.maxSizeAmplificationPercent(), options.sortedRunSizeRatio());
+ List<ManifestSortedRun> pickedRuns = pickStrategy.pick(levelRuns);
+
+ if (pickedRuns.isEmpty() && defaultCompactionMap.isEmpty()) {
+ LOG.debug(
+ "Manifest sort rewrite skipped: no runs picked and no defaultCompaction files.");
+ return Optional.empty();
+ }
+
+ LOG.info(
+ "Manifest sort rewrite: input={} files, lsm={} runs, picked={} runs, "
+ + "defaultCompaction={} files.",
+ input.size(),
+ levelRuns.size(),
+ pickedRuns.size(),
+ defaultCompactionMap.size());
+
+ Set<ManifestSortedRun> pickedSet = new HashSet<>(pickedRuns);
+ List<ManifestFileMeta> reusedFiles = new ArrayList<>();
+ for (ManifestSortedRun run : levelRuns) {
+ if (!pickedSet.contains(run)) {
+ reusedFiles.addAll(run.files());
+ }
+ }
+ List<ManifestFileMeta> result = new ArrayList<>(reusedFiles);
+
+ // Step 5: Split picked files into sections, sort and rewrite each.
+ List<ManifestFileMeta> pickedFiles = new ArrayList<>();
+ for (ManifestSortedRun run : pickedRuns) {
+ pickedFiles.addAll(run.files());
+ }
+ pickedFiles.addAll(defaultCompactionMap.keySet());
+
+ List<Section> sections =
+ splitIntoSections(pickedFiles, sortFieldIndex, sortFieldType, defaultCompactionMap);
+ sections = mergeSmallAdjacentSections(sections, suggestedMetaSize);
+
+ rewriteSections(
+ sections,
+ defaultCompactionMap,
+ manifestFile,
+ sortFieldIndex,
+ sortFieldType,
+ deleteEntries,
+ suggestedMetaSize,
+ options.manifestSortMaxRewriteSize(),
+ result,
+ newFilesForAbort,
+ manifestReadParallelism);
+
+ LOG.info(
+ "Manifest sort rewrite completed: sections={}, newFiles={}, resultFiles={}.",
+ sections.size(),
+ newFilesForAbort.size(),
+ result.size());
+ return Optional.of(result);
+ }
+
+ /**
+ * Classify manifest files into default-compaction group and LSM group.
+ *
+ * <p>When full compaction is triggered (totalDeltaFileSize >= threshold), files that must
+ * change or overlap with delete partitions go into defaultCompactionManifests; the rest stay as
+ * lsmFiles.
+ *
+ * <p>When full compaction is NOT triggered, adjacent small manifests whose cumulative size
+ * reaches suggestedMetaSize are grouped into defaultCompactionManifests (minor-style pick).
+ */
+ private static ClassifyResult classifyManifests(
+ List<ManifestFileMeta> input,
+ long suggestedMetaSize,
+ ManifestFile manifestFile,
+ RowType partitionType,
+ @Nullable Integer manifestReadParallelism) {
+ Map<ManifestFileMeta, boolean[]> defaultCompactionManifests = new LinkedHashMap<>();
+ List<ManifestFileMeta> lsmFiles = new LinkedList<>(input);
+ Set<FileEntry.Identifier> deleteEntries =
+ FileEntry.readDeletedEntries(manifestFile, input, manifestReadParallelism);
+
+ PartitionPredicate predicate;
+ if (deleteEntries.isEmpty()) {
+ predicate = PartitionPredicate.ALWAYS_FALSE;
+ } else {
+ if (partitionType.getFieldCount() > 0) {
+ Set<BinaryRow> deletePartitions =
+ ManifestFileMerger.computeDeletePartitions(deleteEntries);
+ predicate = PartitionPredicate.fromMultiple(partitionType, deletePartitions);
+ } else {
+ predicate = PartitionPredicate.ALWAYS_TRUE;
+ }
+ }
+
+ Iterator<ManifestFileMeta> iterator = lsmFiles.iterator();
+ while (iterator.hasNext()) {
+ ManifestFileMeta file = iterator.next();
+ boolean small = file.fileSize() < suggestedMetaSize;
+ boolean inDeleteRange =
+ predicate != null
+ && predicate.test(
+ file.numAddedFiles() + file.numDeletedFiles(),
+ file.partitionStats().minValues(),
+ file.partitionStats().maxValues(),
+ file.partitionStats().nullCounts());
+ if (small || inDeleteRange) {
+ iterator.remove();
+ defaultCompactionManifests.put(file, new boolean[] {small, inDeleteRange});
+ }
+ }
+ return new ClassifyResult(defaultCompactionManifests, lsmFiles, deleteEntries);
+ }
+
+ /**
+ * Build level-sorted runs from a list of manifest files. Sorts files by min partition value,
+ * greedy-scans to build non-overlapping SortedRuns, then assigns levels by totalSize (Top-4
+ * largest to level 1~4, rest to level 0).
+ */
+ static List<ManifestSortedRun> buildLevelSortedRuns(
+ List<ManifestFileMeta> input, int sortFieldIndex, DataType sortFieldType) {
+ // Step 1: Sort by min value (if equal, then by max value)
+ input.sort(
+ (a, b) -> {
+ int cmp =
+ compareField(
+ a.partitionStats().minValues(),
+ b.partitionStats().minValues(),
+ sortFieldIndex,
+ sortFieldType);
+ if (cmp != 0) {
+ return cmp;
+ }
+ return compareField(
+ a.partitionStats().maxValues(),
+ b.partitionStats().maxValues(),
+ sortFieldIndex,
+ sortFieldType);
+ });
+
+ // Step 2: Interval graph coloring algorithm - assign files to runs
+ // Use priority queue to track runs by their max values
+ PriorityQueue<List<ManifestFileMeta>> runs =
+ new PriorityQueue<>(
+ (r1, r2) -> {
+ ManifestFileMeta last1 = r1.get(r1.size() - 1);
+ ManifestFileMeta last2 = r2.get(r2.size() - 1);
+ return compareField(
+ last1.partitionStats().maxValues(),
+ last2.partitionStats().maxValues(),
+ sortFieldIndex,
+ sortFieldType);
+ });
+
+ for (ManifestFileMeta file : input) {
+ boolean addedToExisting = false;
+
+ // Try to find a run where current file's min >= run's max
+ if (!runs.isEmpty()) {
+ List<ManifestFileMeta> earliestRun = runs.peek();
+ ManifestFileMeta last = earliestRun.get(earliestRun.size() - 1);
+
+ if (compareField(
+ file.partitionStats().minValues(),
+ last.partitionStats().maxValues(),
+ sortFieldIndex,
+ sortFieldType)
+ >= 0) {
+ // Current file can be added to this run
+ runs.poll();
+ earliestRun.add(file);
+ runs.offer(earliestRun);
+ addedToExisting = true;
+ }
+ }
+
+ if (!addedToExisting) {
Review Comment:
changed.
--
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] Support manifest sort feature when commit [paimon]
Posted by "JingsongLi (via GitHub)" <gi...@apache.org>.
JingsongLi commented on PR #7842:
URL: https://github.com/apache/paimon/pull/7842#issuecomment-4613175775
Thanks for adding the tests and comments. The clarification for `manifest-sort.max-rewrite-size` and boundary equality helps.
I still think point 2 is not fully addressed. Your explanation about `rewriteSubSegments` is correct for the overflow/fallback path, but the normal under-budget path does not go through that tail check. In `rewriteSections`, when `processedSize + section.totalSize <= maxRewriteSize`, it directly calls `sortAndRewriteSection(section, ...)` without checking `suggestedMinMetaCount`. Since minor sort compaction puts all files with `fileSize < suggestedMetaSize` into `defaultCompactionMap`, two small manifest files can be rewritten into one new manifest even when `manifest.merge-min-count = 3` and full compaction is not triggered. In the old minor merge path, that trailing candidate group would be kept unchanged.
I verified this locally with a small regression case: two small ADD-only manifests, `manifest.target-file-size = 10MB`, `manifest.merge-min-count = 3`, and `manifest.full-compaction-threshold-size = Long.MAX_VALUE`. With `manifest-sort.enabled = true`, the result is one newly written manifest instead of the original two manifests.
So either the sort path should preserve the old minor merge gate before rewriting small under-budget sections, or the option documentation/tests should explicitly state this semantic change for `manifest.merge-min-count` under `manifest-sort.enabled`.
--
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] Support manifest sort feature when commit [paimon]
Posted by "leaves12138 (via GitHub)" <gi...@apache.org>.
leaves12138 commented on PR #7842:
URL: https://github.com/apache/paimon/pull/7842#issuecomment-4488349870
One more thing:
Can you add test for ManifestFileSorter and ManifestPickStrategy?
--
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] Support manifest sort feature when commit [paimon]
Posted by "leaves12138 (via GitHub)" <gi...@apache.org>.
leaves12138 commented on code in PR #7842:
URL: https://github.com/apache/paimon/pull/7842#discussion_r3263390733
##########
paimon-core/src/main/java/org/apache/paimon/operation/ManifestFileSorter.java:
##########
@@ -0,0 +1,785 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.operation;
+
+import org.apache.paimon.CoreOptions;
+import org.apache.paimon.data.BinaryRow;
+import org.apache.paimon.io.RollingFileWriter;
+import org.apache.paimon.manifest.FileEntry;
+import org.apache.paimon.manifest.ManifestEntry;
+import org.apache.paimon.manifest.ManifestFile;
+import org.apache.paimon.manifest.ManifestFileMeta;
+import org.apache.paimon.operation.ManifestFileMerger.FullCompactionReadResult;
+import org.apache.paimon.partition.PartitionPredicate;
+import org.apache.paimon.types.DataType;
+import org.apache.paimon.types.DecimalType;
+import org.apache.paimon.types.RowType;
+import org.apache.paimon.utils.Filter;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.annotation.Nullable;
+
+import java.util.ArrayList;
+import java.util.Comparator;
+import java.util.HashSet;
+import java.util.Iterator;
+import java.util.LinkedHashMap;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.PriorityQueue;
+import java.util.Set;
+import java.util.function.Function;
+
+import static java.util.Collections.singletonList;
+import static org.apache.paimon.utils.ManifestReadThreadPool.sequentialBatchedExecute;
+
+/** Manifest file sorter that sorts and rewrites manifest files by a configured partition field. */
+public class ManifestFileSorter {
+
+ private static final Logger LOG = LoggerFactory.getLogger(ManifestFileSorter.class);
+
+ /**
+ * Try to sort-rewrite the merged manifest list by a configured partition field. If the sort
+ * field cannot be resolved or the delta file size is below the full compaction threshold, the
+ * input is returned as-is.
+ */
+ static Optional<List<ManifestFileMeta>> trySortRewrite(
+ List<ManifestFileMeta> input,
+ List<ManifestFileMeta> newFilesForAbort,
+ ManifestFile manifestFile,
+ RowType partitionType,
+ CoreOptions options)
+ throws Exception {
+ // Extract configuration from options
+ long suggestedMetaSize = options.manifestTargetSize().getBytes();
+ Integer manifestReadParallelism = options.scanManifestParallelism();
+ String sortPartitionField = options.manifestSortPartitionField();
+ // Step 1: Resolve sort field.
+ String sortField = resolveSortField(sortPartitionField, partitionType);
+ if (sortField == null) {
+ throw new IllegalArgumentException(
+ "Cannot resolve sort field for manifest sort rewrite.");
+ }
+ int sortFieldIndex = partitionType.getFieldNames().indexOf(sortField);
+ DataType sortFieldType = partitionType.getTypeAt(sortFieldIndex);
+
+ // Step 2: Classify manifests into defaultCompaction and LSM.
+ ClassifyResult classified =
+ classifyManifests(
+ input,
+ suggestedMetaSize,
+ manifestFile,
+ partitionType,
+ manifestReadParallelism);
+ Map<ManifestFileMeta, boolean[]> defaultCompactionMap =
+ classified.defaultCompactionManifests;
+ List<ManifestFileMeta> lsmFiles = classified.lsmFiles;
+ Set<FileEntry.Identifier> deleteEntries = classified.deleteEntries;
+
+ // Step 3: Build LSM Tree and assign levels (only for lsmFiles).
+ List<ManifestSortedRun> levelRuns =
+ lsmFiles.isEmpty()
+ ? new ArrayList<>()
+ : buildLevelSortedRuns(lsmFiles, sortFieldIndex, sortFieldType);
+
+ // Step 4: Pick runs to compact.
+ ManifestPickStrategy pickStrategy =
+ new ManifestPickStrategy(
+ options.maxSizeAmplificationPercent(), options.sortedRunSizeRatio());
+ List<ManifestSortedRun> pickedRuns = pickStrategy.pick(levelRuns);
+
+ if (pickedRuns.isEmpty() && defaultCompactionMap.isEmpty()) {
+ LOG.debug(
+ "Manifest sort rewrite skipped: no runs picked and no defaultCompaction files.");
+ return Optional.empty();
+ }
+
+ LOG.info(
+ "Manifest sort rewrite: input={} files, lsm={} runs, picked={} runs, "
+ + "defaultCompaction={} files.",
+ input.size(),
+ levelRuns.size(),
+ pickedRuns.size(),
+ defaultCompactionMap.size());
+
+ Set<ManifestSortedRun> pickedSet = new HashSet<>(pickedRuns);
+ List<ManifestFileMeta> reusedFiles = new ArrayList<>();
+ for (ManifestSortedRun run : levelRuns) {
+ if (!pickedSet.contains(run)) {
+ reusedFiles.addAll(run.files());
+ }
+ }
+ List<ManifestFileMeta> result = new ArrayList<>(reusedFiles);
+
+ // Step 5: Split picked files into sections, sort and rewrite each.
+ List<ManifestFileMeta> pickedFiles = new ArrayList<>();
+ for (ManifestSortedRun run : pickedRuns) {
+ pickedFiles.addAll(run.files());
+ }
+ pickedFiles.addAll(defaultCompactionMap.keySet());
+
+ List<Section> sections =
+ splitIntoSections(pickedFiles, sortFieldIndex, sortFieldType, defaultCompactionMap);
+ sections = mergeSmallAdjacentSections(sections, suggestedMetaSize);
+
+ rewriteSections(
+ sections,
+ defaultCompactionMap,
+ manifestFile,
+ sortFieldIndex,
+ sortFieldType,
+ deleteEntries,
+ suggestedMetaSize,
+ options.manifestSortMaxRewriteSize(),
+ result,
+ newFilesForAbort,
+ manifestReadParallelism);
+
+ LOG.info(
+ "Manifest sort rewrite completed: sections={}, newFiles={}, resultFiles={}.",
+ sections.size(),
+ newFilesForAbort.size(),
+ result.size());
+ return Optional.of(result);
+ }
+
+ /**
+ * Classify manifest files into default-compaction group and LSM group.
+ *
+ * <p>When full compaction is triggered (totalDeltaFileSize >= threshold), files that must
+ * change or overlap with delete partitions go into defaultCompactionManifests; the rest stay as
+ * lsmFiles.
+ *
+ * <p>When full compaction is NOT triggered, adjacent small manifests whose cumulative size
+ * reaches suggestedMetaSize are grouped into defaultCompactionManifests (minor-style pick).
+ */
+ private static ClassifyResult classifyManifests(
+ List<ManifestFileMeta> input,
+ long suggestedMetaSize,
+ ManifestFile manifestFile,
+ RowType partitionType,
+ @Nullable Integer manifestReadParallelism) {
+ Map<ManifestFileMeta, boolean[]> defaultCompactionManifests = new LinkedHashMap<>();
+ List<ManifestFileMeta> lsmFiles = new LinkedList<>(input);
+ Set<FileEntry.Identifier> deleteEntries =
+ FileEntry.readDeletedEntries(manifestFile, input, manifestReadParallelism);
+
+ PartitionPredicate predicate;
+ if (deleteEntries.isEmpty()) {
+ predicate = PartitionPredicate.ALWAYS_FALSE;
+ } else {
+ if (partitionType.getFieldCount() > 0) {
+ Set<BinaryRow> deletePartitions =
+ ManifestFileMerger.computeDeletePartitions(deleteEntries);
+ predicate = PartitionPredicate.fromMultiple(partitionType, deletePartitions);
+ } else {
+ predicate = PartitionPredicate.ALWAYS_TRUE;
+ }
+ }
+
+ Iterator<ManifestFileMeta> iterator = lsmFiles.iterator();
+ while (iterator.hasNext()) {
+ ManifestFileMeta file = iterator.next();
+ boolean small = file.fileSize() < suggestedMetaSize;
+ boolean inDeleteRange =
+ predicate != null
+ && predicate.test(
+ file.numAddedFiles() + file.numDeletedFiles(),
+ file.partitionStats().minValues(),
+ file.partitionStats().maxValues(),
+ file.partitionStats().nullCounts());
+ if (small || inDeleteRange) {
+ iterator.remove();
+ defaultCompactionManifests.put(file, new boolean[] {small, inDeleteRange});
+ }
+ }
+ return new ClassifyResult(defaultCompactionManifests, lsmFiles, deleteEntries);
+ }
+
+ /**
+ * Build level-sorted runs from a list of manifest files. Sorts files by min partition value,
+ * greedy-scans to build non-overlapping SortedRuns, then assigns levels by totalSize (Top-4
+ * largest to level 1~4, rest to level 0).
+ */
+ static List<ManifestSortedRun> buildLevelSortedRuns(
+ List<ManifestFileMeta> input, int sortFieldIndex, DataType sortFieldType) {
+ // Step 1: Sort by min value (if equal, then by max value)
+ input.sort(
+ (a, b) -> {
+ int cmp =
+ compareField(
+ a.partitionStats().minValues(),
+ b.partitionStats().minValues(),
+ sortFieldIndex,
+ sortFieldType);
+ if (cmp != 0) {
+ return cmp;
+ }
+ return compareField(
+ a.partitionStats().maxValues(),
+ b.partitionStats().maxValues(),
+ sortFieldIndex,
+ sortFieldType);
+ });
+
+ // Step 2: Interval graph coloring algorithm - assign files to runs
+ // Use priority queue to track runs by their max values
+ PriorityQueue<List<ManifestFileMeta>> runs =
+ new PriorityQueue<>(
+ (r1, r2) -> {
+ ManifestFileMeta last1 = r1.get(r1.size() - 1);
+ ManifestFileMeta last2 = r2.get(r2.size() - 1);
+ return compareField(
+ last1.partitionStats().maxValues(),
+ last2.partitionStats().maxValues(),
+ sortFieldIndex,
+ sortFieldType);
+ });
+
+ for (ManifestFileMeta file : input) {
+ boolean addedToExisting = false;
+
+ // Try to find a run where current file's min >= run's max
+ if (!runs.isEmpty()) {
+ List<ManifestFileMeta> earliestRun = runs.peek();
+ ManifestFileMeta last = earliestRun.get(earliestRun.size() - 1);
+
+ if (compareField(
+ file.partitionStats().minValues(),
+ last.partitionStats().maxValues(),
+ sortFieldIndex,
+ sortFieldType)
+ >= 0) {
+ // Current file can be added to this run
+ runs.poll();
+ earliestRun.add(file);
+ runs.offer(earliestRun);
+ addedToExisting = true;
+ }
+ }
+
+ if (!addedToExisting) {
Review Comment:
Do not use boolean addedToExisting.
Just
List<ManifestFileMeta> earliestRun = runs.peek();
if (earliestRun == null) {
do something
} else if (compare(xxx) > 0) {
do something
} else {
do something
}
It makes this more pretty
--
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] Support manifest sort feature when commit [paimon]
Posted by "leaves12138 (via GitHub)" <gi...@apache.org>.
leaves12138 commented on code in PR #7842:
URL: https://github.com/apache/paimon/pull/7842#discussion_r3263406025
##########
paimon-core/src/main/java/org/apache/paimon/operation/ManifestFileSorter.java:
##########
@@ -0,0 +1,785 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.operation;
+
+import org.apache.paimon.CoreOptions;
+import org.apache.paimon.data.BinaryRow;
+import org.apache.paimon.io.RollingFileWriter;
+import org.apache.paimon.manifest.FileEntry;
+import org.apache.paimon.manifest.ManifestEntry;
+import org.apache.paimon.manifest.ManifestFile;
+import org.apache.paimon.manifest.ManifestFileMeta;
+import org.apache.paimon.operation.ManifestFileMerger.FullCompactionReadResult;
+import org.apache.paimon.partition.PartitionPredicate;
+import org.apache.paimon.types.DataType;
+import org.apache.paimon.types.DecimalType;
+import org.apache.paimon.types.RowType;
+import org.apache.paimon.utils.Filter;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.annotation.Nullable;
+
+import java.util.ArrayList;
+import java.util.Comparator;
+import java.util.HashSet;
+import java.util.Iterator;
+import java.util.LinkedHashMap;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.PriorityQueue;
+import java.util.Set;
+import java.util.function.Function;
+
+import static java.util.Collections.singletonList;
+import static org.apache.paimon.utils.ManifestReadThreadPool.sequentialBatchedExecute;
+
+/** Manifest file sorter that sorts and rewrites manifest files by a configured partition field. */
+public class ManifestFileSorter {
+
+ private static final Logger LOG = LoggerFactory.getLogger(ManifestFileSorter.class);
+
+ /**
+ * Try to sort-rewrite the merged manifest list by a configured partition field. If the sort
+ * field cannot be resolved or the delta file size is below the full compaction threshold, the
+ * input is returned as-is.
+ */
+ static Optional<List<ManifestFileMeta>> trySortRewrite(
+ List<ManifestFileMeta> input,
+ List<ManifestFileMeta> newFilesForAbort,
+ ManifestFile manifestFile,
+ RowType partitionType,
+ CoreOptions options)
+ throws Exception {
+ // Extract configuration from options
+ long suggestedMetaSize = options.manifestTargetSize().getBytes();
+ Integer manifestReadParallelism = options.scanManifestParallelism();
+ String sortPartitionField = options.manifestSortPartitionField();
+ // Step 1: Resolve sort field.
+ String sortField = resolveSortField(sortPartitionField, partitionType);
+ if (sortField == null) {
+ throw new IllegalArgumentException(
+ "Cannot resolve sort field for manifest sort rewrite.");
+ }
+ int sortFieldIndex = partitionType.getFieldNames().indexOf(sortField);
+ DataType sortFieldType = partitionType.getTypeAt(sortFieldIndex);
+
+ // Step 2: Classify manifests into defaultCompaction and LSM.
+ ClassifyResult classified =
+ classifyManifests(
+ input,
+ suggestedMetaSize,
+ manifestFile,
+ partitionType,
+ manifestReadParallelism);
+ Map<ManifestFileMeta, boolean[]> defaultCompactionMap =
+ classified.defaultCompactionManifests;
+ List<ManifestFileMeta> lsmFiles = classified.lsmFiles;
+ Set<FileEntry.Identifier> deleteEntries = classified.deleteEntries;
+
+ // Step 3: Build LSM Tree and assign levels (only for lsmFiles).
+ List<ManifestSortedRun> levelRuns =
+ lsmFiles.isEmpty()
+ ? new ArrayList<>()
+ : buildLevelSortedRuns(lsmFiles, sortFieldIndex, sortFieldType);
+
+ // Step 4: Pick runs to compact.
+ ManifestPickStrategy pickStrategy =
+ new ManifestPickStrategy(
+ options.maxSizeAmplificationPercent(), options.sortedRunSizeRatio());
+ List<ManifestSortedRun> pickedRuns = pickStrategy.pick(levelRuns);
+
+ if (pickedRuns.isEmpty() && defaultCompactionMap.isEmpty()) {
+ LOG.debug(
+ "Manifest sort rewrite skipped: no runs picked and no defaultCompaction files.");
+ return Optional.empty();
+ }
+
+ LOG.info(
+ "Manifest sort rewrite: input={} files, lsm={} runs, picked={} runs, "
+ + "defaultCompaction={} files.",
+ input.size(),
+ levelRuns.size(),
+ pickedRuns.size(),
+ defaultCompactionMap.size());
+
+ Set<ManifestSortedRun> pickedSet = new HashSet<>(pickedRuns);
+ List<ManifestFileMeta> reusedFiles = new ArrayList<>();
+ for (ManifestSortedRun run : levelRuns) {
+ if (!pickedSet.contains(run)) {
+ reusedFiles.addAll(run.files());
+ }
+ }
+ List<ManifestFileMeta> result = new ArrayList<>(reusedFiles);
+
+ // Step 5: Split picked files into sections, sort and rewrite each.
+ List<ManifestFileMeta> pickedFiles = new ArrayList<>();
+ for (ManifestSortedRun run : pickedRuns) {
+ pickedFiles.addAll(run.files());
+ }
+ pickedFiles.addAll(defaultCompactionMap.keySet());
+
+ List<Section> sections =
+ splitIntoSections(pickedFiles, sortFieldIndex, sortFieldType, defaultCompactionMap);
+ sections = mergeSmallAdjacentSections(sections, suggestedMetaSize);
+
+ rewriteSections(
+ sections,
+ defaultCompactionMap,
+ manifestFile,
+ sortFieldIndex,
+ sortFieldType,
+ deleteEntries,
+ suggestedMetaSize,
+ options.manifestSortMaxRewriteSize(),
+ result,
+ newFilesForAbort,
+ manifestReadParallelism);
+
+ LOG.info(
+ "Manifest sort rewrite completed: sections={}, newFiles={}, resultFiles={}.",
+ sections.size(),
+ newFilesForAbort.size(),
+ result.size());
+ return Optional.of(result);
+ }
+
+ /**
+ * Classify manifest files into default-compaction group and LSM group.
+ *
+ * <p>When full compaction is triggered (totalDeltaFileSize >= threshold), files that must
+ * change or overlap with delete partitions go into defaultCompactionManifests; the rest stay as
+ * lsmFiles.
+ *
+ * <p>When full compaction is NOT triggered, adjacent small manifests whose cumulative size
+ * reaches suggestedMetaSize are grouped into defaultCompactionManifests (minor-style pick).
+ */
+ private static ClassifyResult classifyManifests(
+ List<ManifestFileMeta> input,
+ long suggestedMetaSize,
+ ManifestFile manifestFile,
+ RowType partitionType,
+ @Nullable Integer manifestReadParallelism) {
+ Map<ManifestFileMeta, boolean[]> defaultCompactionManifests = new LinkedHashMap<>();
+ List<ManifestFileMeta> lsmFiles = new LinkedList<>(input);
+ Set<FileEntry.Identifier> deleteEntries =
+ FileEntry.readDeletedEntries(manifestFile, input, manifestReadParallelism);
+
+ PartitionPredicate predicate;
+ if (deleteEntries.isEmpty()) {
+ predicate = PartitionPredicate.ALWAYS_FALSE;
+ } else {
+ if (partitionType.getFieldCount() > 0) {
+ Set<BinaryRow> deletePartitions =
+ ManifestFileMerger.computeDeletePartitions(deleteEntries);
+ predicate = PartitionPredicate.fromMultiple(partitionType, deletePartitions);
+ } else {
+ predicate = PartitionPredicate.ALWAYS_TRUE;
+ }
+ }
+
+ Iterator<ManifestFileMeta> iterator = lsmFiles.iterator();
+ while (iterator.hasNext()) {
+ ManifestFileMeta file = iterator.next();
+ boolean small = file.fileSize() < suggestedMetaSize;
+ boolean inDeleteRange =
+ predicate != null
+ && predicate.test(
+ file.numAddedFiles() + file.numDeletedFiles(),
+ file.partitionStats().minValues(),
+ file.partitionStats().maxValues(),
+ file.partitionStats().nullCounts());
+ if (small || inDeleteRange) {
+ iterator.remove();
+ defaultCompactionManifests.put(file, new boolean[] {small, inDeleteRange});
+ }
+ }
+ return new ClassifyResult(defaultCompactionManifests, lsmFiles, deleteEntries);
+ }
+
+ /**
+ * Build level-sorted runs from a list of manifest files. Sorts files by min partition value,
+ * greedy-scans to build non-overlapping SortedRuns, then assigns levels by totalSize (Top-4
+ * largest to level 1~4, rest to level 0).
+ */
+ static List<ManifestSortedRun> buildLevelSortedRuns(
+ List<ManifestFileMeta> input, int sortFieldIndex, DataType sortFieldType) {
+ // Step 1: Sort by min value (if equal, then by max value)
+ input.sort(
+ (a, b) -> {
+ int cmp =
+ compareField(
+ a.partitionStats().minValues(),
+ b.partitionStats().minValues(),
+ sortFieldIndex,
+ sortFieldType);
+ if (cmp != 0) {
+ return cmp;
+ }
+ return compareField(
+ a.partitionStats().maxValues(),
+ b.partitionStats().maxValues(),
+ sortFieldIndex,
+ sortFieldType);
+ });
+
+ // Step 2: Interval graph coloring algorithm - assign files to runs
+ // Use priority queue to track runs by their max values
+ PriorityQueue<List<ManifestFileMeta>> runs =
+ new PriorityQueue<>(
+ (r1, r2) -> {
+ ManifestFileMeta last1 = r1.get(r1.size() - 1);
+ ManifestFileMeta last2 = r2.get(r2.size() - 1);
+ return compareField(
+ last1.partitionStats().maxValues(),
+ last2.partitionStats().maxValues(),
+ sortFieldIndex,
+ sortFieldType);
+ });
+
+ for (ManifestFileMeta file : input) {
+ boolean addedToExisting = false;
+
+ // Try to find a run where current file's min >= run's max
+ if (!runs.isEmpty()) {
+ List<ManifestFileMeta> earliestRun = runs.peek();
+ ManifestFileMeta last = earliestRun.get(earliestRun.size() - 1);
+
+ if (compareField(
+ file.partitionStats().minValues(),
+ last.partitionStats().maxValues(),
+ sortFieldIndex,
+ sortFieldType)
+ >= 0) {
Review Comment:
There is overlap in one run if "equals".
--
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] Support manifest sort feature when commit [paimon]
Posted by "leaves12138 (via GitHub)" <gi...@apache.org>.
leaves12138 commented on code in PR #7842:
URL: https://github.com/apache/paimon/pull/7842#discussion_r3271314884
##########
paimon-core/src/main/java/org/apache/paimon/operation/ManifestFileSorter.java:
##########
@@ -0,0 +1,740 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.operation;
+
+import org.apache.paimon.CoreOptions;
+import org.apache.paimon.codegen.CodeGenUtils;
+import org.apache.paimon.codegen.RecordComparator;
+import org.apache.paimon.data.BinaryRow;
+import org.apache.paimon.io.RollingFileWriter;
+import org.apache.paimon.manifest.FileEntry;
+import org.apache.paimon.manifest.ManifestEntry;
+import org.apache.paimon.manifest.ManifestFile;
+import org.apache.paimon.manifest.ManifestFileMeta;
+import org.apache.paimon.operation.ManifestFileMerger.FullCompactionReadResult;
+import org.apache.paimon.partition.PartitionPredicate;
+import org.apache.paimon.types.RowType;
+import org.apache.paimon.utils.Filter;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.annotation.Nullable;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.Comparator;
+import java.util.HashSet;
+import java.util.Iterator;
+import java.util.LinkedHashMap;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.PriorityQueue;
+import java.util.Set;
+import java.util.function.Function;
+
+import static java.util.Collections.singletonList;
+import static org.apache.paimon.utils.ManifestReadThreadPool.sequentialBatchedExecute;
+
+/** Manifest file sorter that sorts and rewrites manifest files by a configured partition field. */
+public class ManifestFileSorter {
+
+ private static final Logger LOG = LoggerFactory.getLogger(ManifestFileSorter.class);
+
+ /**
+ * Try to sort-rewrite the merged manifest list by a configured partition field. If the sort
+ * field cannot be resolved, the input is returned as-is.
+ */
+ static Optional<List<ManifestFileMeta>> trySortRewrite(
+ List<ManifestFileMeta> input,
+ List<ManifestFileMeta> newFilesForAbort,
+ ManifestFile manifestFile,
+ RowType partitionType,
+ CoreOptions options)
+ throws Exception {
+ // Extract configuration from options
+ long suggestedMetaSize = options.manifestTargetSize().getBytes();
+ Integer manifestReadParallelism = options.scanManifestParallelism();
+ String sortPartitionField = options.manifestSortPartitionField();
+ long manifestFullCompactionThresholdSize =
+ options.manifestFullCompactionThresholdSize().getBytes();
+ // Step 1: Resolve sort field.
+ String sortField = resolveSortField(sortPartitionField, partitionType);
+ if (sortField == null) {
+ throw new IllegalArgumentException(
+ "Cannot resolve sort field for manifest sort rewrite.");
+ }
+ int sortFieldIndex = partitionType.getFieldNames().indexOf(sortField);
+ RecordComparator fieldComparator =
+ CodeGenUtils.newRecordComparator(
+ partitionType.getFieldTypes(), new int[] {sortFieldIndex});
+
+ // Step 2: Classify manifests into defaultCompaction and LSM.
+ List<ManifestFileMeta> result = new ArrayList<>();
+ ClassifyResult classified =
+ classifyManifests(
+ input,
+ result,
+ suggestedMetaSize,
+ manifestFile,
+ partitionType,
+ manifestFullCompactionThresholdSize,
+ manifestReadParallelism);
+ Map<ManifestFileMeta, Boolean> defaultCompactionMap = classified.defaultCompactionManifests;
+ List<ManifestFileMeta> lsmFiles = classified.lsmFiles;
+ Set<FileEntry.Identifier> deleteEntries = classified.deleteEntries;
+
+ // Step 3: Build LSM Tree and assign levels (only for lsmFiles).
+ List<ManifestSortedRun> levelRuns =
+ lsmFiles.isEmpty()
+ ? new ArrayList<>()
+ : buildLevelSortedRuns(lsmFiles, fieldComparator);
+
+ // Step 4: Pick runs to compact.
+ ManifestPickStrategy pickStrategy =
+ new ManifestPickStrategy(
+ options.maxSizeAmplificationPercent(), options.sortedRunSizeRatio());
+ List<ManifestSortedRun> pickedRuns = pickStrategy.pick(levelRuns);
+
+ if (pickedRuns.isEmpty() && defaultCompactionMap.isEmpty()) {
+ LOG.debug(
+ "Manifest sort rewrite skipped: no runs picked and no defaultCompaction files.");
+ return Optional.empty();
+ }
+
+ LOG.info(
+ "Manifest sort rewrite: input={} files, lsm={} runs, picked={} runs, "
+ + "defaultCompaction={} files.",
+ input.size(),
+ levelRuns.size(),
+ pickedRuns.size(),
+ defaultCompactionMap.size());
+
+ Set<ManifestSortedRun> pickedSet = new HashSet<>(pickedRuns);
+ List<ManifestFileMeta> reusedFiles = new ArrayList<>();
+ for (ManifestSortedRun run : levelRuns) {
+ if (!pickedSet.contains(run)) {
+ reusedFiles.addAll(run.files());
+ }
+ }
+ result.addAll(reusedFiles);
+
+ // Step 5: Split picked files into sections, sort and rewrite each.
+ List<ManifestFileMeta> pickedFiles = new ArrayList<>();
+ for (ManifestSortedRun run : pickedRuns) {
+ pickedFiles.addAll(run.files());
+ }
+ pickedFiles.addAll(defaultCompactionMap.keySet());
+
+ List<Section> sections =
+ splitIntoSections(pickedFiles, fieldComparator, defaultCompactionMap);
+ sections = mergeSmallAdjacentSections(sections, suggestedMetaSize);
+
+ rewriteSections(
+ sections,
+ defaultCompactionMap,
+ manifestFile,
+ fieldComparator,
+ deleteEntries,
+ suggestedMetaSize,
+ options.manifestMergeMinCount(),
+ options.manifestSortMaxRewriteSize(),
+ result,
+ newFilesForAbort,
+ manifestReadParallelism);
+
+ LOG.info(
+ "Manifest sort rewrite completed: sections={}, newFiles={}, resultFiles={}.",
+ sections.size(),
+ newFilesForAbort.size(),
+ result.size());
+ return Optional.of(result);
+ }
+
+ /**
+ * Classify manifest files into default-compaction group and LSM group.
+ *
+ * <p>Full compaction: small files and files overlapping delete partitions go into
+ * defaultCompactionManifests; the rest stay as lsmFiles.
+ *
+ * <p>Non-full compaction: delete-overlapping files go to result, small files go to
+ * defaultCompactionManifests for minor-style merge.
+ */
+ private static ClassifyResult classifyManifests(
+ List<ManifestFileMeta> input,
+ List<ManifestFileMeta> result,
+ long suggestedMetaSize,
+ ManifestFile manifestFile,
+ RowType partitionType,
+ long sizeTrigger,
+ @Nullable Integer manifestReadParallelism) {
+ // Calculate total size of files that need compaction to determine full-compaction trigger
+ Filter<ManifestFileMeta> mustChange =
+ file -> file.numDeletedFiles() > 0 || file.fileSize() < suggestedMetaSize;
+ long totalDeltaFileSize = 0;
+ for (ManifestFileMeta file : input) {
+ if (mustChange.test(file)) {
+ totalDeltaFileSize += file.fileSize();
+ }
+ }
+ boolean removeAllDelete = totalDeltaFileSize >= sizeTrigger;
Review Comment:
Rename to triggerFullCompact
--
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] Support manifest sort feature when commit [paimon]
Posted by "discivigour (via GitHub)" <gi...@apache.org>.
discivigour commented on PR #7842:
URL: https://github.com/apache/paimon/pull/7842#issuecomment-4485799665
> Thanks for the update. I took another pass over the latest revision, and I think there are still a few issues that should be fixed before merging.
>
> 1. The boundary condition for interval overlap still looks wrong.
> In `ManifestFileSorter.buildLevelSortedRuns`, a file is appended to an existing run when `file.min >= last.max`. In `splitIntoSections`, a new section is also started when `file.min >= sectionMaxBound`.
> However, partition stats represent closed intervals. For example, `[1, 3]` and `[3, 5]` still overlap at partition value `3`. A sorted run is documented as containing non-overlapping intervals, so this case should not be placed into the same run. Similarly, sections are used as overlap-connected rewrite units, so this case should not be split into different sections either. I think both checks should use `> 0`, not `>= 0`, and we should add a test for the `max == min` boundary case.
> 2. `manifest-sort.enabled` currently bypasses the original manifest compaction trigger/gate.
> `ManifestFileMerger.merge` directly enters `ManifestFileSorter.trySortRewrite` and returns from that path when manifest sort is enabled. This means the original `manifest.full-compaction-threshold-size` and `manifest.merge-min-count` behavior is no longer applied in the same way. Inside `classifyManifests`, files are classified only by `fileSize < targetSize` or delete-range overlap, so small manifests / delete manifests can trigger sort rewrite more aggressively than the existing merge logic.
> If this is intentional, I think the new semantics should be documented very clearly. Otherwise, the sort path should preserve the existing full/minor compaction gates, especially `manifest.merge-min-count` for minor manifest merging.
> 3. The partial rewrite path for `manifest-sort.max-rewrite-size` can break the output order.
> When the first section exceeds the rewrite budget, `rewriteSections` splits it into `rewriteFiles` and `remainingFiles`, rewrites the first part, and appends the remaining section to the end of the `sections` list. If there are later sections with larger key ranges, the remaining part of the current section will be emitted after them, which can produce an order like `0..10, 20..30, 10..20`.
> To keep the manifest list sorted, I think we should either skip the whole section once the budget is exceeded, or keep the remaining section at the current position/order instead of appending it to the tail. This also needs a regression test.
> 4. Test coverage is still missing some important edge cases.
> The new tests cover large overlapping ranges and delete elimination, but I do not see coverage for boundary-touching intervals (`max == min`), `manifest.merge-min-count` / full threshold behavior under `manifest-sort.enabled`, `manifest-sort.max-rewrite-size` preserving global output order, or null partition values. The switch to `RecordComparator` is a good improvement, but a null partition test would make this safer.
Thanks for your comment.
1. I designed it this way to ensure that the minimum number of Sorted runs is built to reduce the burden of sorting.
2. I have introduced manifestFullCompactionThresholdSize to to reduce the phenomenon of "one delete entry causing large-scale file rewriting.
3. I think it is not a problem.
4. I will add more tests.
--
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] Support manifest sort feature when commit [paimon]
Posted by "discivigour (via GitHub)" <gi...@apache.org>.
discivigour commented on code in PR #7842:
URL: https://github.com/apache/paimon/pull/7842#discussion_r3273214943
##########
paimon-core/src/main/java/org/apache/paimon/operation/ManifestFileSorter.java:
##########
@@ -0,0 +1,740 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.operation;
+
+import org.apache.paimon.CoreOptions;
+import org.apache.paimon.codegen.CodeGenUtils;
+import org.apache.paimon.codegen.RecordComparator;
+import org.apache.paimon.data.BinaryRow;
+import org.apache.paimon.io.RollingFileWriter;
+import org.apache.paimon.manifest.FileEntry;
+import org.apache.paimon.manifest.ManifestEntry;
+import org.apache.paimon.manifest.ManifestFile;
+import org.apache.paimon.manifest.ManifestFileMeta;
+import org.apache.paimon.operation.ManifestFileMerger.FullCompactionReadResult;
+import org.apache.paimon.partition.PartitionPredicate;
+import org.apache.paimon.types.RowType;
+import org.apache.paimon.utils.Filter;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.annotation.Nullable;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.Comparator;
+import java.util.HashSet;
+import java.util.Iterator;
+import java.util.LinkedHashMap;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.PriorityQueue;
+import java.util.Set;
+import java.util.function.Function;
+
+import static java.util.Collections.singletonList;
+import static org.apache.paimon.utils.ManifestReadThreadPool.sequentialBatchedExecute;
+
+/** Manifest file sorter that sorts and rewrites manifest files by a configured partition field. */
+public class ManifestFileSorter {
+
+ private static final Logger LOG = LoggerFactory.getLogger(ManifestFileSorter.class);
+
+ /**
+ * Try to sort-rewrite the merged manifest list by a configured partition field. If the sort
+ * field cannot be resolved, the input is returned as-is.
+ */
+ static Optional<List<ManifestFileMeta>> trySortRewrite(
+ List<ManifestFileMeta> input,
+ List<ManifestFileMeta> newFilesForAbort,
+ ManifestFile manifestFile,
+ RowType partitionType,
+ CoreOptions options)
+ throws Exception {
+ // Extract configuration from options
+ long suggestedMetaSize = options.manifestTargetSize().getBytes();
+ Integer manifestReadParallelism = options.scanManifestParallelism();
+ String sortPartitionField = options.manifestSortPartitionField();
+ long manifestFullCompactionThresholdSize =
+ options.manifestFullCompactionThresholdSize().getBytes();
+ // Step 1: Resolve sort field.
+ String sortField = resolveSortField(sortPartitionField, partitionType);
+ if (sortField == null) {
+ throw new IllegalArgumentException(
+ "Cannot resolve sort field for manifest sort rewrite.");
+ }
+ int sortFieldIndex = partitionType.getFieldNames().indexOf(sortField);
+ RecordComparator fieldComparator =
+ CodeGenUtils.newRecordComparator(
+ partitionType.getFieldTypes(), new int[] {sortFieldIndex});
+
+ // Step 2: Classify manifests into defaultCompaction and LSM.
+ List<ManifestFileMeta> result = new ArrayList<>();
+ ClassifyResult classified =
+ classifyManifests(
+ input,
+ result,
+ suggestedMetaSize,
+ manifestFile,
+ partitionType,
+ manifestFullCompactionThresholdSize,
+ manifestReadParallelism);
+ Map<ManifestFileMeta, Boolean> defaultCompactionMap = classified.defaultCompactionManifests;
+ List<ManifestFileMeta> lsmFiles = classified.lsmFiles;
+ Set<FileEntry.Identifier> deleteEntries = classified.deleteEntries;
+
+ // Step 3: Build LSM Tree and assign levels (only for lsmFiles).
+ List<ManifestSortedRun> levelRuns =
+ lsmFiles.isEmpty()
+ ? new ArrayList<>()
+ : buildLevelSortedRuns(lsmFiles, fieldComparator);
+
+ // Step 4: Pick runs to compact.
+ ManifestPickStrategy pickStrategy =
+ new ManifestPickStrategy(
+ options.maxSizeAmplificationPercent(), options.sortedRunSizeRatio());
+ List<ManifestSortedRun> pickedRuns = pickStrategy.pick(levelRuns);
+
+ if (pickedRuns.isEmpty() && defaultCompactionMap.isEmpty()) {
+ LOG.debug(
+ "Manifest sort rewrite skipped: no runs picked and no defaultCompaction files.");
+ return Optional.empty();
+ }
+
+ LOG.info(
+ "Manifest sort rewrite: input={} files, lsm={} runs, picked={} runs, "
+ + "defaultCompaction={} files.",
+ input.size(),
+ levelRuns.size(),
+ pickedRuns.size(),
+ defaultCompactionMap.size());
+
+ Set<ManifestSortedRun> pickedSet = new HashSet<>(pickedRuns);
+ List<ManifestFileMeta> reusedFiles = new ArrayList<>();
+ for (ManifestSortedRun run : levelRuns) {
+ if (!pickedSet.contains(run)) {
+ reusedFiles.addAll(run.files());
+ }
+ }
+ result.addAll(reusedFiles);
+
+ // Step 5: Split picked files into sections, sort and rewrite each.
+ List<ManifestFileMeta> pickedFiles = new ArrayList<>();
+ for (ManifestSortedRun run : pickedRuns) {
+ pickedFiles.addAll(run.files());
+ }
+ pickedFiles.addAll(defaultCompactionMap.keySet());
+
+ List<Section> sections =
+ splitIntoSections(pickedFiles, fieldComparator, defaultCompactionMap);
+ sections = mergeSmallAdjacentSections(sections, suggestedMetaSize);
+
+ rewriteSections(
+ sections,
+ defaultCompactionMap,
+ manifestFile,
+ fieldComparator,
+ deleteEntries,
+ suggestedMetaSize,
+ options.manifestMergeMinCount(),
+ options.manifestSortMaxRewriteSize(),
+ result,
+ newFilesForAbort,
+ manifestReadParallelism);
+
+ LOG.info(
+ "Manifest sort rewrite completed: sections={}, newFiles={}, resultFiles={}.",
+ sections.size(),
+ newFilesForAbort.size(),
+ result.size());
+ return Optional.of(result);
+ }
+
+ /**
+ * Classify manifest files into default-compaction group and LSM group.
+ *
+ * <p>Full compaction: small files and files overlapping delete partitions go into
+ * defaultCompactionManifests; the rest stay as lsmFiles.
+ *
+ * <p>Non-full compaction: delete-overlapping files go to result, small files go to
+ * defaultCompactionManifests for minor-style merge.
+ */
+ private static ClassifyResult classifyManifests(
+ List<ManifestFileMeta> input,
+ List<ManifestFileMeta> result,
+ long suggestedMetaSize,
+ ManifestFile manifestFile,
+ RowType partitionType,
+ long sizeTrigger,
+ @Nullable Integer manifestReadParallelism) {
+ // Calculate total size of files that need compaction to determine full-compaction trigger
+ Filter<ManifestFileMeta> mustChange =
+ file -> file.numDeletedFiles() > 0 || file.fileSize() < suggestedMetaSize;
+ long totalDeltaFileSize = 0;
+ for (ManifestFileMeta file : input) {
+ if (mustChange.test(file)) {
+ totalDeltaFileSize += file.fileSize();
+ }
+ }
+ boolean removeAllDelete = totalDeltaFileSize >= sizeTrigger;
+
+ // Initialize classification containers and read delete entries
+ Map<ManifestFileMeta, Boolean> defaultCompactionManifests = new LinkedHashMap<>();
+ List<ManifestFileMeta> lsmFiles = new LinkedList<>(input);
+ Set<FileEntry.Identifier> deleteEntries =
+ FileEntry.readDeletedEntries(manifestFile, input, manifestReadParallelism);
Review Comment:
Thanks, I will split full compaction and minor compaction.
--
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] Support manifest sort feature when commit [paimon]
Posted by "leaves12138 (via GitHub)" <gi...@apache.org>.
leaves12138 commented on code in PR #7842:
URL: https://github.com/apache/paimon/pull/7842#discussion_r3271341713
##########
paimon-core/src/main/java/org/apache/paimon/operation/ManifestFileSorter.java:
##########
@@ -0,0 +1,740 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.operation;
+
+import org.apache.paimon.CoreOptions;
+import org.apache.paimon.codegen.CodeGenUtils;
+import org.apache.paimon.codegen.RecordComparator;
+import org.apache.paimon.data.BinaryRow;
+import org.apache.paimon.io.RollingFileWriter;
+import org.apache.paimon.manifest.FileEntry;
+import org.apache.paimon.manifest.ManifestEntry;
+import org.apache.paimon.manifest.ManifestFile;
+import org.apache.paimon.manifest.ManifestFileMeta;
+import org.apache.paimon.operation.ManifestFileMerger.FullCompactionReadResult;
+import org.apache.paimon.partition.PartitionPredicate;
+import org.apache.paimon.types.RowType;
+import org.apache.paimon.utils.Filter;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.annotation.Nullable;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.Comparator;
+import java.util.HashSet;
+import java.util.Iterator;
+import java.util.LinkedHashMap;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.PriorityQueue;
+import java.util.Set;
+import java.util.function.Function;
+
+import static java.util.Collections.singletonList;
+import static org.apache.paimon.utils.ManifestReadThreadPool.sequentialBatchedExecute;
+
+/** Manifest file sorter that sorts and rewrites manifest files by a configured partition field. */
+public class ManifestFileSorter {
+
+ private static final Logger LOG = LoggerFactory.getLogger(ManifestFileSorter.class);
+
+ /**
+ * Try to sort-rewrite the merged manifest list by a configured partition field. If the sort
+ * field cannot be resolved, the input is returned as-is.
+ */
+ static Optional<List<ManifestFileMeta>> trySortRewrite(
+ List<ManifestFileMeta> input,
+ List<ManifestFileMeta> newFilesForAbort,
+ ManifestFile manifestFile,
+ RowType partitionType,
+ CoreOptions options)
+ throws Exception {
+ // Extract configuration from options
+ long suggestedMetaSize = options.manifestTargetSize().getBytes();
+ Integer manifestReadParallelism = options.scanManifestParallelism();
+ String sortPartitionField = options.manifestSortPartitionField();
+ long manifestFullCompactionThresholdSize =
+ options.manifestFullCompactionThresholdSize().getBytes();
+ // Step 1: Resolve sort field.
+ String sortField = resolveSortField(sortPartitionField, partitionType);
+ if (sortField == null) {
+ throw new IllegalArgumentException(
+ "Cannot resolve sort field for manifest sort rewrite.");
+ }
+ int sortFieldIndex = partitionType.getFieldNames().indexOf(sortField);
+ RecordComparator fieldComparator =
+ CodeGenUtils.newRecordComparator(
+ partitionType.getFieldTypes(), new int[] {sortFieldIndex});
+
+ // Step 2: Classify manifests into defaultCompaction and LSM.
+ List<ManifestFileMeta> result = new ArrayList<>();
+ ClassifyResult classified =
+ classifyManifests(
+ input,
+ result,
+ suggestedMetaSize,
+ manifestFile,
+ partitionType,
+ manifestFullCompactionThresholdSize,
+ manifestReadParallelism);
+ Map<ManifestFileMeta, Boolean> defaultCompactionMap = classified.defaultCompactionManifests;
+ List<ManifestFileMeta> lsmFiles = classified.lsmFiles;
+ Set<FileEntry.Identifier> deleteEntries = classified.deleteEntries;
+
+ // Step 3: Build LSM Tree and assign levels (only for lsmFiles).
+ List<ManifestSortedRun> levelRuns =
+ lsmFiles.isEmpty()
+ ? new ArrayList<>()
+ : buildLevelSortedRuns(lsmFiles, fieldComparator);
+
+ // Step 4: Pick runs to compact.
+ ManifestPickStrategy pickStrategy =
+ new ManifestPickStrategy(
+ options.maxSizeAmplificationPercent(), options.sortedRunSizeRatio());
+ List<ManifestSortedRun> pickedRuns = pickStrategy.pick(levelRuns);
+
+ if (pickedRuns.isEmpty() && defaultCompactionMap.isEmpty()) {
+ LOG.debug(
+ "Manifest sort rewrite skipped: no runs picked and no defaultCompaction files.");
+ return Optional.empty();
+ }
+
+ LOG.info(
+ "Manifest sort rewrite: input={} files, lsm={} runs, picked={} runs, "
+ + "defaultCompaction={} files.",
+ input.size(),
+ levelRuns.size(),
+ pickedRuns.size(),
+ defaultCompactionMap.size());
+
+ Set<ManifestSortedRun> pickedSet = new HashSet<>(pickedRuns);
+ List<ManifestFileMeta> reusedFiles = new ArrayList<>();
+ for (ManifestSortedRun run : levelRuns) {
+ if (!pickedSet.contains(run)) {
+ reusedFiles.addAll(run.files());
+ }
+ }
+ result.addAll(reusedFiles);
+
+ // Step 5: Split picked files into sections, sort and rewrite each.
+ List<ManifestFileMeta> pickedFiles = new ArrayList<>();
+ for (ManifestSortedRun run : pickedRuns) {
+ pickedFiles.addAll(run.files());
+ }
+ pickedFiles.addAll(defaultCompactionMap.keySet());
+
+ List<Section> sections =
+ splitIntoSections(pickedFiles, fieldComparator, defaultCompactionMap);
+ sections = mergeSmallAdjacentSections(sections, suggestedMetaSize);
+
+ rewriteSections(
+ sections,
+ defaultCompactionMap,
+ manifestFile,
+ fieldComparator,
+ deleteEntries,
+ suggestedMetaSize,
+ options.manifestMergeMinCount(),
+ options.manifestSortMaxRewriteSize(),
+ result,
+ newFilesForAbort,
+ manifestReadParallelism);
+
+ LOG.info(
+ "Manifest sort rewrite completed: sections={}, newFiles={}, resultFiles={}.",
+ sections.size(),
+ newFilesForAbort.size(),
+ result.size());
+ return Optional.of(result);
+ }
+
+ /**
+ * Classify manifest files into default-compaction group and LSM group.
+ *
+ * <p>Full compaction: small files and files overlapping delete partitions go into
+ * defaultCompactionManifests; the rest stay as lsmFiles.
+ *
+ * <p>Non-full compaction: delete-overlapping files go to result, small files go to
+ * defaultCompactionManifests for minor-style merge.
+ */
+ private static ClassifyResult classifyManifests(
+ List<ManifestFileMeta> input,
+ List<ManifestFileMeta> result,
+ long suggestedMetaSize,
+ ManifestFile manifestFile,
+ RowType partitionType,
+ long sizeTrigger,
+ @Nullable Integer manifestReadParallelism) {
+ // Calculate total size of files that need compaction to determine full-compaction trigger
+ Filter<ManifestFileMeta> mustChange =
+ file -> file.numDeletedFiles() > 0 || file.fileSize() < suggestedMetaSize;
+ long totalDeltaFileSize = 0;
+ for (ManifestFileMeta file : input) {
+ if (mustChange.test(file)) {
+ totalDeltaFileSize += file.fileSize();
+ }
+ }
+ boolean removeAllDelete = totalDeltaFileSize >= sizeTrigger;
+
+ // Initialize classification containers and read delete entries
+ Map<ManifestFileMeta, Boolean> defaultCompactionManifests = new LinkedHashMap<>();
+ List<ManifestFileMeta> lsmFiles = new LinkedList<>(input);
+ Set<FileEntry.Identifier> deleteEntries =
+ FileEntry.readDeletedEntries(manifestFile, input, manifestReadParallelism);
Review Comment:
Why read delete every time? If not full compaction, we still need to read all the deletes?
--
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] Support manifest sort feature when commit [paimon]
Posted by "leaves12138 (via GitHub)" <gi...@apache.org>.
leaves12138 commented on code in PR #7842:
URL: https://github.com/apache/paimon/pull/7842#discussion_r3271348809
##########
paimon-core/src/main/java/org/apache/paimon/operation/ManifestFileSorter.java:
##########
@@ -0,0 +1,740 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.operation;
+
+import org.apache.paimon.CoreOptions;
+import org.apache.paimon.codegen.CodeGenUtils;
+import org.apache.paimon.codegen.RecordComparator;
+import org.apache.paimon.data.BinaryRow;
+import org.apache.paimon.io.RollingFileWriter;
+import org.apache.paimon.manifest.FileEntry;
+import org.apache.paimon.manifest.ManifestEntry;
+import org.apache.paimon.manifest.ManifestFile;
+import org.apache.paimon.manifest.ManifestFileMeta;
+import org.apache.paimon.operation.ManifestFileMerger.FullCompactionReadResult;
+import org.apache.paimon.partition.PartitionPredicate;
+import org.apache.paimon.types.RowType;
+import org.apache.paimon.utils.Filter;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.annotation.Nullable;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.Comparator;
+import java.util.HashSet;
+import java.util.Iterator;
+import java.util.LinkedHashMap;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.PriorityQueue;
+import java.util.Set;
+import java.util.function.Function;
+
+import static java.util.Collections.singletonList;
+import static org.apache.paimon.utils.ManifestReadThreadPool.sequentialBatchedExecute;
+
+/** Manifest file sorter that sorts and rewrites manifest files by a configured partition field. */
+public class ManifestFileSorter {
+
+ private static final Logger LOG = LoggerFactory.getLogger(ManifestFileSorter.class);
+
+ /**
+ * Try to sort-rewrite the merged manifest list by a configured partition field. If the sort
+ * field cannot be resolved, the input is returned as-is.
+ */
+ static Optional<List<ManifestFileMeta>> trySortRewrite(
+ List<ManifestFileMeta> input,
+ List<ManifestFileMeta> newFilesForAbort,
+ ManifestFile manifestFile,
+ RowType partitionType,
+ CoreOptions options)
+ throws Exception {
+ // Extract configuration from options
+ long suggestedMetaSize = options.manifestTargetSize().getBytes();
+ Integer manifestReadParallelism = options.scanManifestParallelism();
+ String sortPartitionField = options.manifestSortPartitionField();
+ long manifestFullCompactionThresholdSize =
+ options.manifestFullCompactionThresholdSize().getBytes();
+ // Step 1: Resolve sort field.
+ String sortField = resolveSortField(sortPartitionField, partitionType);
+ if (sortField == null) {
+ throw new IllegalArgumentException(
+ "Cannot resolve sort field for manifest sort rewrite.");
+ }
+ int sortFieldIndex = partitionType.getFieldNames().indexOf(sortField);
+ RecordComparator fieldComparator =
+ CodeGenUtils.newRecordComparator(
+ partitionType.getFieldTypes(), new int[] {sortFieldIndex});
+
+ // Step 2: Classify manifests into defaultCompaction and LSM.
+ List<ManifestFileMeta> result = new ArrayList<>();
+ ClassifyResult classified =
+ classifyManifests(
+ input,
+ result,
+ suggestedMetaSize,
+ manifestFile,
+ partitionType,
+ manifestFullCompactionThresholdSize,
+ manifestReadParallelism);
+ Map<ManifestFileMeta, Boolean> defaultCompactionMap = classified.defaultCompactionManifests;
+ List<ManifestFileMeta> lsmFiles = classified.lsmFiles;
+ Set<FileEntry.Identifier> deleteEntries = classified.deleteEntries;
+
+ // Step 3: Build LSM Tree and assign levels (only for lsmFiles).
+ List<ManifestSortedRun> levelRuns =
+ lsmFiles.isEmpty()
+ ? new ArrayList<>()
+ : buildLevelSortedRuns(lsmFiles, fieldComparator);
+
+ // Step 4: Pick runs to compact.
+ ManifestPickStrategy pickStrategy =
+ new ManifestPickStrategy(
+ options.maxSizeAmplificationPercent(), options.sortedRunSizeRatio());
+ List<ManifestSortedRun> pickedRuns = pickStrategy.pick(levelRuns);
+
+ if (pickedRuns.isEmpty() && defaultCompactionMap.isEmpty()) {
+ LOG.debug(
+ "Manifest sort rewrite skipped: no runs picked and no defaultCompaction files.");
+ return Optional.empty();
+ }
+
+ LOG.info(
+ "Manifest sort rewrite: input={} files, lsm={} runs, picked={} runs, "
+ + "defaultCompaction={} files.",
+ input.size(),
+ levelRuns.size(),
+ pickedRuns.size(),
+ defaultCompactionMap.size());
+
+ Set<ManifestSortedRun> pickedSet = new HashSet<>(pickedRuns);
+ List<ManifestFileMeta> reusedFiles = new ArrayList<>();
+ for (ManifestSortedRun run : levelRuns) {
+ if (!pickedSet.contains(run)) {
+ reusedFiles.addAll(run.files());
+ }
+ }
+ result.addAll(reusedFiles);
+
+ // Step 5: Split picked files into sections, sort and rewrite each.
+ List<ManifestFileMeta> pickedFiles = new ArrayList<>();
+ for (ManifestSortedRun run : pickedRuns) {
+ pickedFiles.addAll(run.files());
+ }
+ pickedFiles.addAll(defaultCompactionMap.keySet());
+
+ List<Section> sections =
+ splitIntoSections(pickedFiles, fieldComparator, defaultCompactionMap);
+ sections = mergeSmallAdjacentSections(sections, suggestedMetaSize);
+
+ rewriteSections(
+ sections,
+ defaultCompactionMap,
+ manifestFile,
+ fieldComparator,
+ deleteEntries,
+ suggestedMetaSize,
+ options.manifestMergeMinCount(),
+ options.manifestSortMaxRewriteSize(),
+ result,
+ newFilesForAbort,
+ manifestReadParallelism);
+
+ LOG.info(
+ "Manifest sort rewrite completed: sections={}, newFiles={}, resultFiles={}.",
+ sections.size(),
+ newFilesForAbort.size(),
+ result.size());
+ return Optional.of(result);
+ }
+
+ /**
+ * Classify manifest files into default-compaction group and LSM group.
+ *
+ * <p>Full compaction: small files and files overlapping delete partitions go into
+ * defaultCompactionManifests; the rest stay as lsmFiles.
+ *
+ * <p>Non-full compaction: delete-overlapping files go to result, small files go to
+ * defaultCompactionManifests for minor-style merge.
+ */
+ private static ClassifyResult classifyManifests(
+ List<ManifestFileMeta> input,
+ List<ManifestFileMeta> result,
+ long suggestedMetaSize,
+ ManifestFile manifestFile,
+ RowType partitionType,
+ long sizeTrigger,
+ @Nullable Integer manifestReadParallelism) {
+ // Calculate total size of files that need compaction to determine full-compaction trigger
+ Filter<ManifestFileMeta> mustChange =
+ file -> file.numDeletedFiles() > 0 || file.fileSize() < suggestedMetaSize;
+ long totalDeltaFileSize = 0;
+ for (ManifestFileMeta file : input) {
+ if (mustChange.test(file)) {
+ totalDeltaFileSize += file.fileSize();
+ }
+ }
+ boolean removeAllDelete = totalDeltaFileSize >= sizeTrigger;
+
+ // Initialize classification containers and read delete entries
+ Map<ManifestFileMeta, Boolean> defaultCompactionManifests = new LinkedHashMap<>();
+ List<ManifestFileMeta> lsmFiles = new LinkedList<>(input);
+ Set<FileEntry.Identifier> deleteEntries =
+ FileEntry.readDeletedEntries(manifestFile, input, manifestReadParallelism);
Review Comment:
Split full compaction and minor compaction, don't make it mixed. Refer to ManifestFileMerge.merge
--
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] Support manifest sort feature when commit [paimon]
Posted by "discivigour (via GitHub)" <gi...@apache.org>.
discivigour commented on PR #7842:
URL: https://github.com/apache/paimon/pull/7842#issuecomment-4611240352
> Thanks for the update. I took another pass over the latest revision and found a few issues that I think should be addressed before merging.
>
> 1. `manifest-sort.max-rewrite-size` can be exceeded and can reorder the remaining section.
> In `ManifestFileSorter.rewriteSections`, when a section is larger than the remaining budget, the code splits it into `rewriteFiles` and `remainingFiles`, rewrites `rewriteFiles`, then appends `remainingFiles` to the end of `sections`. If the remaining section contains default-compaction files, it can still be rewritten later through `rewriteSubSegments`, so the configured max rewrite size is no longer a hard cap. Also, appending the remaining range after later sections can produce an output order such as `0..10, 20..30, 10..20`. I think we should either skip the whole over-budget section, or keep the remaining files unchanged in the original position/order. This needs a regression test for the budget-hit path.
> 2. `manifest-sort.enabled` changes the existing minor manifest merge trigger semantics.
> Once manifest sort is enabled, `ManifestFileMerger.merge` bypasses the old full/minor compaction path and always enters `ManifestFileSorter.trySortCompaction`. The minor sort path classifies files by `fileSize < targetSize` and delete-range overlap, but it does not preserve the previous `manifest.merge-min-count` gate in the same way. This can rewrite small manifests more aggressively than users expect from existing manifest merge behavior. If this is intentional, the option documentation should state the changed semantics clearly; otherwise the sort path should preserve the old minor merge trigger.
> 3. Boundary-touching partition ranges are treated as non-overlapping.
> `buildLevelSortedRuns` appends a file to a run when `file.min >= last.max`, and `splitIntoSections` starts a new section when `file.min >= sectionMaxBound`. Manifest partition stats are closed intervals, so `[1, 3]` and `[3, 5]` still overlap at partition value `3`. This conflicts with the `ManifestAdjacentSortedRun` contract that intervals do not overlap on the sort field. I think these comparisons should use `> 0`, or the deliberate relaxed invariant should be documented very explicitly and covered by boundary tests.
Thanks for your comment.
1. The files within the manifest-sort.max-rewriting-size threshold are sorted, while those outside the threshold are merged into small files and deleted. This is intentional.
2. SortCompact has completely rewritten Manifest Merge, and the related parameters can still be used normally.
3. I designed it this way to ensure that the minimum number of Sorted runs is built to reduce the burden of sorting. The endpoints of intervals between sections do not overlap because there is no need to merge these sections
--
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] Support manifest sort feature when commit [paimon]
Posted by "discivigour (via GitHub)" <gi...@apache.org>.
discivigour commented on PR #7842:
URL: https://github.com/apache/paimon/pull/7842#issuecomment-4613753077
> Thanks for adding the tests and comments. The clarification for `manifest-sort.max-rewrite-size` and boundary equality helps.
>
> I still think point 2 is not fully addressed. Your explanation about `rewriteSubSegments` is correct for the overflow/fallback path, but the normal under-budget path does not go through that tail check. In `rewriteSections`, when `processedSize + section.totalSize <= maxRewriteSize`, it directly calls `sortAndRewriteSection(section, ...)` without checking `suggestedMinMetaCount`. Since minor sort compaction puts all files with `fileSize < suggestedMetaSize` into `defaultCompactionMap`, two small manifest files can be rewritten into one new manifest even when `manifest.merge-min-count = 3` and full compaction is not triggered. In the old minor merge path, that trailing candidate group would be kept unchanged.
>
> I verified this locally with a small regression case: two small ADD-only manifests, `manifest.target-file-size = 10MB`, `manifest.merge-min-count = 3`, and `manifest.full-compaction-threshold-size = Long.MAX_VALUE`. With `manifest-sort.enabled = true`, the result is one newly written manifest instead of the original two manifests.
>
> So either the sort path should preserve the old minor merge gate before rewriting small under-budget sections, or the option documentation/tests should explicitly state this semantic change for `manifest.merge-min-count` under `manifest-sort.enabled`.
ok, comments for the params added.
--
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] Support manifest sort feature when commit [paimon]
Posted by "JingsongLi (via GitHub)" <gi...@apache.org>.
JingsongLi merged PR #7842:
URL: https://github.com/apache/paimon/pull/7842
--
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] Support manifest sort feature when commit [paimon]
Posted by "discivigour (via GitHub)" <gi...@apache.org>.
discivigour commented on code in PR #7842:
URL: https://github.com/apache/paimon/pull/7842#discussion_r3264562509
##########
paimon-core/src/main/java/org/apache/paimon/operation/ManifestFileSorter.java:
##########
@@ -0,0 +1,785 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.operation;
+
+import org.apache.paimon.CoreOptions;
+import org.apache.paimon.data.BinaryRow;
+import org.apache.paimon.io.RollingFileWriter;
+import org.apache.paimon.manifest.FileEntry;
+import org.apache.paimon.manifest.ManifestEntry;
+import org.apache.paimon.manifest.ManifestFile;
+import org.apache.paimon.manifest.ManifestFileMeta;
+import org.apache.paimon.operation.ManifestFileMerger.FullCompactionReadResult;
+import org.apache.paimon.partition.PartitionPredicate;
+import org.apache.paimon.types.DataType;
+import org.apache.paimon.types.DecimalType;
+import org.apache.paimon.types.RowType;
+import org.apache.paimon.utils.Filter;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.annotation.Nullable;
+
+import java.util.ArrayList;
+import java.util.Comparator;
+import java.util.HashSet;
+import java.util.Iterator;
+import java.util.LinkedHashMap;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.PriorityQueue;
+import java.util.Set;
+import java.util.function.Function;
+
+import static java.util.Collections.singletonList;
+import static org.apache.paimon.utils.ManifestReadThreadPool.sequentialBatchedExecute;
+
+/** Manifest file sorter that sorts and rewrites manifest files by a configured partition field. */
+public class ManifestFileSorter {
+
+ private static final Logger LOG = LoggerFactory.getLogger(ManifestFileSorter.class);
+
+ /**
+ * Try to sort-rewrite the merged manifest list by a configured partition field. If the sort
+ * field cannot be resolved or the delta file size is below the full compaction threshold, the
+ * input is returned as-is.
+ */
+ static Optional<List<ManifestFileMeta>> trySortRewrite(
+ List<ManifestFileMeta> input,
+ List<ManifestFileMeta> newFilesForAbort,
+ ManifestFile manifestFile,
+ RowType partitionType,
+ CoreOptions options)
+ throws Exception {
+ // Extract configuration from options
+ long suggestedMetaSize = options.manifestTargetSize().getBytes();
+ Integer manifestReadParallelism = options.scanManifestParallelism();
+ String sortPartitionField = options.manifestSortPartitionField();
+ // Step 1: Resolve sort field.
+ String sortField = resolveSortField(sortPartitionField, partitionType);
+ if (sortField == null) {
+ throw new IllegalArgumentException(
+ "Cannot resolve sort field for manifest sort rewrite.");
+ }
+ int sortFieldIndex = partitionType.getFieldNames().indexOf(sortField);
+ DataType sortFieldType = partitionType.getTypeAt(sortFieldIndex);
+
+ // Step 2: Classify manifests into defaultCompaction and LSM.
+ ClassifyResult classified =
+ classifyManifests(
+ input,
+ suggestedMetaSize,
+ manifestFile,
+ partitionType,
+ manifestReadParallelism);
+ Map<ManifestFileMeta, boolean[]> defaultCompactionMap =
+ classified.defaultCompactionManifests;
+ List<ManifestFileMeta> lsmFiles = classified.lsmFiles;
+ Set<FileEntry.Identifier> deleteEntries = classified.deleteEntries;
+
+ // Step 3: Build LSM Tree and assign levels (only for lsmFiles).
+ List<ManifestSortedRun> levelRuns =
+ lsmFiles.isEmpty()
+ ? new ArrayList<>()
+ : buildLevelSortedRuns(lsmFiles, sortFieldIndex, sortFieldType);
+
+ // Step 4: Pick runs to compact.
+ ManifestPickStrategy pickStrategy =
+ new ManifestPickStrategy(
+ options.maxSizeAmplificationPercent(), options.sortedRunSizeRatio());
+ List<ManifestSortedRun> pickedRuns = pickStrategy.pick(levelRuns);
+
+ if (pickedRuns.isEmpty() && defaultCompactionMap.isEmpty()) {
+ LOG.debug(
+ "Manifest sort rewrite skipped: no runs picked and no defaultCompaction files.");
+ return Optional.empty();
+ }
+
+ LOG.info(
+ "Manifest sort rewrite: input={} files, lsm={} runs, picked={} runs, "
+ + "defaultCompaction={} files.",
+ input.size(),
+ levelRuns.size(),
+ pickedRuns.size(),
+ defaultCompactionMap.size());
+
+ Set<ManifestSortedRun> pickedSet = new HashSet<>(pickedRuns);
+ List<ManifestFileMeta> reusedFiles = new ArrayList<>();
+ for (ManifestSortedRun run : levelRuns) {
+ if (!pickedSet.contains(run)) {
+ reusedFiles.addAll(run.files());
+ }
+ }
+ List<ManifestFileMeta> result = new ArrayList<>(reusedFiles);
+
+ // Step 5: Split picked files into sections, sort and rewrite each.
+ List<ManifestFileMeta> pickedFiles = new ArrayList<>();
+ for (ManifestSortedRun run : pickedRuns) {
+ pickedFiles.addAll(run.files());
+ }
+ pickedFiles.addAll(defaultCompactionMap.keySet());
+
+ List<Section> sections =
+ splitIntoSections(pickedFiles, sortFieldIndex, sortFieldType, defaultCompactionMap);
+ sections = mergeSmallAdjacentSections(sections, suggestedMetaSize);
+
+ rewriteSections(
+ sections,
+ defaultCompactionMap,
+ manifestFile,
+ sortFieldIndex,
+ sortFieldType,
+ deleteEntries,
+ suggestedMetaSize,
+ options.manifestSortMaxRewriteSize(),
+ result,
+ newFilesForAbort,
+ manifestReadParallelism);
+
+ LOG.info(
+ "Manifest sort rewrite completed: sections={}, newFiles={}, resultFiles={}.",
+ sections.size(),
+ newFilesForAbort.size(),
+ result.size());
+ return Optional.of(result);
+ }
+
+ /**
+ * Classify manifest files into default-compaction group and LSM group.
+ *
+ * <p>When full compaction is triggered (totalDeltaFileSize >= threshold), files that must
+ * change or overlap with delete partitions go into defaultCompactionManifests; the rest stay as
+ * lsmFiles.
+ *
+ * <p>When full compaction is NOT triggered, adjacent small manifests whose cumulative size
+ * reaches suggestedMetaSize are grouped into defaultCompactionManifests (minor-style pick).
+ */
+ private static ClassifyResult classifyManifests(
+ List<ManifestFileMeta> input,
+ long suggestedMetaSize,
+ ManifestFile manifestFile,
+ RowType partitionType,
+ @Nullable Integer manifestReadParallelism) {
+ Map<ManifestFileMeta, boolean[]> defaultCompactionManifests = new LinkedHashMap<>();
+ List<ManifestFileMeta> lsmFiles = new LinkedList<>(input);
+ Set<FileEntry.Identifier> deleteEntries =
+ FileEntry.readDeletedEntries(manifestFile, input, manifestReadParallelism);
+
+ PartitionPredicate predicate;
+ if (deleteEntries.isEmpty()) {
+ predicate = PartitionPredicate.ALWAYS_FALSE;
+ } else {
+ if (partitionType.getFieldCount() > 0) {
+ Set<BinaryRow> deletePartitions =
+ ManifestFileMerger.computeDeletePartitions(deleteEntries);
+ predicate = PartitionPredicate.fromMultiple(partitionType, deletePartitions);
+ } else {
+ predicate = PartitionPredicate.ALWAYS_TRUE;
+ }
+ }
+
+ Iterator<ManifestFileMeta> iterator = lsmFiles.iterator();
+ while (iterator.hasNext()) {
+ ManifestFileMeta file = iterator.next();
+ boolean small = file.fileSize() < suggestedMetaSize;
+ boolean inDeleteRange =
+ predicate != null
+ && predicate.test(
+ file.numAddedFiles() + file.numDeletedFiles(),
+ file.partitionStats().minValues(),
+ file.partitionStats().maxValues(),
+ file.partitionStats().nullCounts());
+ if (small || inDeleteRange) {
+ iterator.remove();
+ defaultCompactionManifests.put(file, new boolean[] {small, inDeleteRange});
+ }
+ }
+ return new ClassifyResult(defaultCompactionManifests, lsmFiles, deleteEntries);
+ }
+
+ /**
+ * Build level-sorted runs from a list of manifest files. Sorts files by min partition value,
+ * greedy-scans to build non-overlapping SortedRuns, then assigns levels by totalSize (Top-4
+ * largest to level 1~4, rest to level 0).
+ */
+ static List<ManifestSortedRun> buildLevelSortedRuns(
+ List<ManifestFileMeta> input, int sortFieldIndex, DataType sortFieldType) {
+ // Step 1: Sort by min value (if equal, then by max value)
+ input.sort(
+ (a, b) -> {
+ int cmp =
+ compareField(
+ a.partitionStats().minValues(),
+ b.partitionStats().minValues(),
+ sortFieldIndex,
+ sortFieldType);
+ if (cmp != 0) {
+ return cmp;
+ }
+ return compareField(
+ a.partitionStats().maxValues(),
+ b.partitionStats().maxValues(),
+ sortFieldIndex,
+ sortFieldType);
+ });
+
+ // Step 2: Interval graph coloring algorithm - assign files to runs
+ // Use priority queue to track runs by their max values
+ PriorityQueue<List<ManifestFileMeta>> runs =
+ new PriorityQueue<>(
+ (r1, r2) -> {
+ ManifestFileMeta last1 = r1.get(r1.size() - 1);
+ ManifestFileMeta last2 = r2.get(r2.size() - 1);
+ return compareField(
+ last1.partitionStats().maxValues(),
+ last2.partitionStats().maxValues(),
+ sortFieldIndex,
+ sortFieldType);
+ });
+
+ for (ManifestFileMeta file : input) {
+ boolean addedToExisting = false;
+
+ // Try to find a run where current file's min >= run's max
+ if (!runs.isEmpty()) {
+ List<ManifestFileMeta> earliestRun = runs.peek();
+ ManifestFileMeta last = earliestRun.get(earliestRun.size() - 1);
+
+ if (compareField(
+ file.partitionStats().minValues(),
+ last.partitionStats().maxValues(),
+ sortFieldIndex,
+ sortFieldType)
+ >= 0) {
Review Comment:
I designed it this way to ensure that the minimum number of Sorted runs is built to reduce the burden of sorting.
--
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] Support manifest sort feature when commit [paimon]
Posted by "discivigour (via GitHub)" <gi...@apache.org>.
discivigour commented on PR #7842:
URL: https://github.com/apache/paimon/pull/7842#issuecomment-4612194705
> Thanks for the clarification. I understand these are intentional design choices, but I still think we need to make the contract explicit and add regression coverage, otherwise this will be hard to reason about later.
>
> 1. About `manifest-sort.max-rewrite-size`: if files outside the threshold may still be merged/deleted, then the current option name/description is misleading. It says this is the _maximum total size of manifest files to rewrite in a single sort rewrite pass_ and that sections exceeding the limit are skipped. However, the current code can still process the remaining part later through `rewriteSubSegments` when `hasDefaultCompactMeta` is true. So either the code should enforce the cap strictly, or the documentation should say that the cap only limits the sorted rewrite portion and default/minor cleanup may still happen beyond it.
> I am also still worried about ordering. Appending `remainingSection` to the tail means the remaining key range from the current section is emitted after later sections. If global manifest order by sort key is not a required invariant, please document that explicitly. If it is an invariant, we should keep the remaining files in their original position/order. A test that hits the budget in the middle of a section and verifies the final manifest order would settle this.
> 2. About `manifest.merge-min-count`: saying the parameters can still be used normally is not enough for me to verify the behavior. In the old minor merge path, the trailing candidates are kept unchanged when their count is below `manifest.merge-min-count`. In the sort path, a non-empty `defaultCompactionMap` can trigger rewrite even when the number of small manifests is below that threshold. If the intended behavior is different from the old manifest merge, that is fine, but then it should be documented as a semantic change and covered by a test. Otherwise we should preserve the old gate.
> 3. About boundary equality: reducing the number of SortedRuns is a reasonable optimization, but the current class comment says the intervals do not overlap. With closed partition ranges, `[1, 3]` and `[3, 5]` overlap at `3`; they do not just touch in an open-interval sense. If the intended invariant allows boundary-touching intervals in one run/section, please update the comments to state that explicitly and add a boundary test. Otherwise future readers will assume the stricter non-overlap invariant and may build incorrect logic on top of it.
>
> So my ask is not necessarily to reject the design, but to make the semantics precise and backed by tests for these three edge cases.
1. add test and comment
2. It is triggered when the small file is smaller than suggestedMinMetaCount because `!ctx.deleteEntries.isEmpty()` means full compact, suggestedMinMetaCount only in minor compact when used.
3. add test and comment
--
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