diff --git a/paimon-common/src/main/java/org/apache/paimon/utils/ByteArrayKey.java b/paimon-common/src/main/java/org/apache/paimon/utils/ByteArrayKey.java index 09d9ded426a4..274e20abdc0a 100644 --- a/paimon-common/src/main/java/org/apache/paimon/utils/ByteArrayKey.java +++ b/paimon-common/src/main/java/org/apache/paimon/utils/ByteArrayKey.java @@ -47,8 +47,7 @@ byte[] bytes() { public boolean equals(Object obj) { return obj == this || (obj instanceof ByteArrayKey && Arrays.equals(bytes, ((ByteArrayKey) obj).bytes)) - || (obj instanceof ByteArrayLookupKey - && Arrays.equals(bytes, ((ByteArrayLookupKey) obj).bytes())); + || (obj instanceof ByteArrayLookupKey && obj.equals(this)); } @Override diff --git a/paimon-common/src/main/java/org/apache/paimon/utils/ByteArrayLookupKey.java b/paimon-common/src/main/java/org/apache/paimon/utils/ByteArrayLookupKey.java index aaa913ace7ee..023a6b686609 100644 --- a/paimon-common/src/main/java/org/apache/paimon/utils/ByteArrayLookupKey.java +++ b/paimon-common/src/main/java/org/apache/paimon/utils/ByteArrayLookupKey.java @@ -20,8 +20,6 @@ import javax.annotation.Nullable; -import java.util.Arrays; - import static org.apache.paimon.utils.Preconditions.checkArgument; /** @@ -33,6 +31,8 @@ public final class ByteArrayLookupKey { private @Nullable byte[] bytes; + private int offset; + private int length; private int hash; public ByteArrayLookupKey() {} @@ -43,12 +43,26 @@ public ByteArrayLookupKey(byte[] bytes) { public void reset(byte[] bytes) { checkArgument(bytes != null, "Byte array cannot be null."); + reset(bytes, 0, bytes.length); + } + + public void reset(byte[] bytes, int offset, int length) { + checkArgument(bytes != null, "Byte array cannot be null."); + checkArgument(offset >= 0 && length >= 0 && offset <= bytes.length - length); this.bytes = bytes; - this.hash = Arrays.hashCode(bytes); + this.offset = offset; + this.length = length; + int hash = 1; + for (int i = offset; i < offset + length; i++) { + hash = 31 * hash + bytes[i]; + } + this.hash = hash; } public void clear() { bytes = null; + offset = 0; + length = 0; hash = 0; } @@ -62,14 +76,38 @@ public boolean equals(Object obj) { return obj == this || (bytes != null && obj instanceof ByteArrayKey - && Arrays.equals(bytes, ((ByteArrayKey) obj).bytes())) + && equals(((ByteArrayKey) obj).bytes())) || (bytes != null && obj instanceof ByteArrayLookupKey - && Arrays.equals(bytes, ((ByteArrayLookupKey) obj).bytes)); + && equals((ByteArrayLookupKey) obj)); } @Override public int hashCode() { return hash; } + + private boolean equals(byte[] other) { + if (length != other.length) { + return false; + } + for (int i = 0; i < length; i++) { + if (bytes[offset + i] != other[i]) { + return false; + } + } + return true; + } + + private boolean equals(ByteArrayLookupKey other) { + if (other.bytes == null || length != other.length) { + return false; + } + for (int i = 0; i < length; i++) { + if (bytes[offset + i] != other.bytes[other.offset + i]) { + return false; + } + } + return true; + } } diff --git a/paimon-common/src/test/java/org/apache/paimon/utils/ByteArrayKeyTest.java b/paimon-common/src/test/java/org/apache/paimon/utils/ByteArrayKeyTest.java index 89c2db09d426..8248cfcb36cc 100644 --- a/paimon-common/src/test/java/org/apache/paimon/utils/ByteArrayKeyTest.java +++ b/paimon-common/src/test/java/org/apache/paimon/utils/ByteArrayKeyTest.java @@ -59,6 +59,23 @@ void testReusableMapLookup() { assertThat(lookup.hashCode()).isZero(); } + @Test + void testReusableSliceLookup() { + Map values = new HashMap<>(); + ByteArrayKey key = new ByteArrayKey(new byte[] {1, 2, 3}); + values.put(key, "value"); + ByteArrayLookupKey lookup = new ByteArrayLookupKey(); + + lookup.reset(new byte[] {9, 1, 2, 3, 8}, 1, 3); + assertThat(lookup).isEqualTo(key); + assertThat(key).isEqualTo(lookup); + assertThat(lookup.hashCode()).isEqualTo(key.hashCode()); + assertThat(values.get(lookup)).isEqualTo("value"); + + lookup.clear(); + assertThat(new ByteArrayLookupKey(new byte[] {1, 2, 3})).isNotEqualTo(lookup); + } + @Test void testLookupEqualityLifecycle() { ByteArrayLookupKey first = new ByteArrayLookupKey(new byte[] {1}); diff --git a/paimon-core/src/main/java/org/apache/paimon/manifest/DeletedRowIdSet.java b/paimon-core/src/main/java/org/apache/paimon/manifest/DeletedRowIdSet.java index cd7203764d9d..2e5e87883d51 100644 --- a/paimon-core/src/main/java/org/apache/paimon/manifest/DeletedRowIdSet.java +++ b/paimon-core/src/main/java/org/apache/paimon/manifest/DeletedRowIdSet.java @@ -23,7 +23,7 @@ import java.util.Arrays; /** Primitive set used by RowID compaction to avoid rebuilding file identifiers. */ -final class DeletedRowIdSet { +public final class DeletedRowIdSet { private static final long EMPTY = Long.MIN_VALUE; @@ -32,7 +32,7 @@ final class DeletedRowIdSet { private boolean containsMinValue; private @Nullable long[] sortedRowIds; - void add(long value) { + public void add(long value) { if (value == EMPTY) { if (!containsMinValue) { containsMinValue = true; @@ -56,7 +56,7 @@ void add(long value) { sortedRowIds = null; } - void addAll(DeletedRowIdSet other) { + public void addAll(DeletedRowIdSet other) { if (other.containsMinValue) { add(EMPTY); } @@ -67,7 +67,7 @@ void addAll(DeletedRowIdSet other) { } } - boolean contains(long value) { + public boolean contains(long value) { if (value == EMPTY) { return containsMinValue; } @@ -81,7 +81,7 @@ boolean contains(long value) { return false; } - boolean intersects(long minInclusive, long maxInclusive) { + public boolean intersects(long minInclusive, long maxInclusive) { if (minInclusive > maxInclusive) { return true; } @@ -93,12 +93,12 @@ boolean intersects(long minInclusive, long maxInclusive) { return position < values.length && values[position] <= maxInclusive; } - void prepareRangeIndex() { + public void prepareRangeIndex() { // Publish the immutable sorted snapshot before concurrent manifest planning starts. sortedRowIds(); } - void releaseRangeIndex() { + public void releaseRangeIndex() { sortedRowIds = null; } diff --git a/paimon-core/src/main/java/org/apache/paimon/manifest/FileEntry.java b/paimon-core/src/main/java/org/apache/paimon/manifest/FileEntry.java index 11f08cf6329e..8cc109b6a882 100644 --- a/paimon-core/src/main/java/org/apache/paimon/manifest/FileEntry.java +++ b/paimon-core/src/main/java/org/apache/paimon/manifest/FileEntry.java @@ -213,6 +213,19 @@ public ReusableIdentifier replaceWithPartition(ProjectedManifestEntry entry) { return appendEntryFields(entry); } + /** Replaces this encoding with an already serialized identifier. */ + public ReusableIdentifier replace(byte[] value, int offset, int valueLength) { + checkArgument(value != null, "Serialized identifier cannot be null."); + checkArgument( + offset >= 0 && valueLength >= 0 && offset <= value.length - valueLength, + "Identifier byte range is invalid."); + length = 0; + ensureCapacity(valueLength); + System.arraycopy(value, offset, bytes, 0, valueLength); + length = valueLength; + return this; + } + private ReusableIdentifier appendEntryFields(ProjectedManifestEntry entry) { putInt(entry.bucket()); ProjectedDataFileMeta file = entry.file(); diff --git a/paimon-core/src/main/java/org/apache/paimon/manifest/ManifestFile.java b/paimon-core/src/main/java/org/apache/paimon/manifest/ManifestFile.java index e91f84a2e65c..0088744dc5fa 100644 --- a/paimon-core/src/main/java/org/apache/paimon/manifest/ManifestFile.java +++ b/paimon-core/src/main/java/org/apache/paimon/manifest/ManifestFile.java @@ -212,6 +212,11 @@ public ManifestAvroReader scanAvroBlocks(String fileName, @Nullable Long fileSiz } } + /** Opens a low-allocation reader for the encoded manifest fields needed by run merge. */ + public ManifestAvroReader scanForRunMerge(String fileName, @Nullable Long fileSize) { + return scanAvroBlocks(fileName, fileSize); + } + @VisibleForTesting public long suggestedFileSize() { return suggestedFileSize; diff --git a/paimon-core/src/main/java/org/apache/paimon/operation/ManifestEntryRunMerge.java b/paimon-core/src/main/java/org/apache/paimon/operation/ManifestEntryRunMerge.java new file mode 100644 index 000000000000..1b9d4f8db6a1 --- /dev/null +++ b/paimon-core/src/main/java/org/apache/paimon/operation/ManifestEntryRunMerge.java @@ -0,0 +1,675 @@ +/* + * 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.data.BinaryArray; +import org.apache.paimon.data.BinaryRow; +import org.apache.paimon.data.GenericRow; +import org.apache.paimon.data.InternalRow; +import org.apache.paimon.format.SimpleStatsCollector; +import org.apache.paimon.io.DataFileMeta; +import org.apache.paimon.manifest.CompactFileIdentifierSet; +import org.apache.paimon.manifest.DeletedRowIdSet; +import org.apache.paimon.manifest.FileKind; +import org.apache.paimon.manifest.ManifestAvroReader; +import org.apache.paimon.manifest.ManifestAvroReader.RawBlock; +import org.apache.paimon.manifest.ManifestAvroReader.RowIterator; +import org.apache.paimon.manifest.ManifestAvroWriter.EncodedBlockMeta; +import org.apache.paimon.manifest.ManifestEntry; +import org.apache.paimon.manifest.ManifestFile; +import org.apache.paimon.manifest.ManifestFileMeta; +import org.apache.paimon.stats.SimpleStats; +import org.apache.paimon.stats.SimpleStatsConverter; +import org.apache.paimon.types.DataField; +import org.apache.paimon.types.RowType; +import org.apache.paimon.utils.Pair; + +import javax.annotation.Nullable; + +import java.util.ArrayList; +import java.util.Collections; +import java.util.List; +import java.util.function.Function; + +import static org.apache.paimon.utils.ManifestReadThreadPool.sequentialBatchedExecute; +import static org.apache.paimon.utils.Preconditions.checkState; + +/** Streaming merge of the naturally sorted runs in data-evolution manifest files. */ +final class ManifestEntryRunMerge { + + private static final int FRAGMENTED_RUN_THRESHOLD = 64; + private static final long MAX_IN_MEMORY_FRAGMENTED_ENTRIES = 25_000L; + private static final int MAX_STREAM_CURSORS = 128; + private static final int MAX_STREAM_READ_AMPLIFICATION = 8; + static final int KIND = 0; + static final int PARTITION = 1; + static final int BUCKET = 2; + static final int FILE = 3; + static final int FILE_NAME = 0; + static final int ROW_COUNT = 1; + static final int LEVEL = 2; + static final int SCHEMA_ID = 3; + static final int FIRST_ROW_ID = 4; + static final int MAX_SEQUENCE_NUMBER = 5; + static final int EXTRA_FILES = 6; + static final int EMBEDDED_FILE_INDEX = 7; + static final int EXTERNAL_PATH = 8; + static final int FILE_FIELD_COUNT = 9; + private static final String[] ENTRY_FILE_FIELD_NAMES = { + DataFileMeta.FILE_NAME, + DataFileMeta.ROW_COUNT, + DataFileMeta.LEVEL, + DataFileMeta.SCHEMA_ID, + DataFileMeta.FIRST_ROW_ID, + DataFileMeta.MAX_SEQUENCE_NUMBER, + DataFileMeta.EXTRA_FILES, + DataFileMeta.EMBEDDED_FILE_INDEX, + DataFileMeta.EXTERNAL_PATH + }; + private static final InternalRow.FieldGetter[] ENTRY_FILE_GETTERS = entryFileGetters(); + static final RowType ENTRY_LAYOUT = entryLayout(); + private static final int FULL_KIND = + ManifestEntry.MANIFEST_ROW_TYPE.getFieldIndex(ManifestEntry.KIND); + private static final int FULL_PARTITION = + ManifestEntry.MANIFEST_ROW_TYPE.getFieldIndex(ManifestEntry.PARTITION); + private static final int FULL_BUCKET = + ManifestEntry.MANIFEST_ROW_TYPE.getFieldIndex(ManifestEntry.BUCKET); + private static final int FULL_FILE = + ManifestEntry.MANIFEST_ROW_TYPE.getFieldIndex(ManifestEntry.FILE); + + private ManifestEntryRunMerge() {} + + private static InternalRow.FieldGetter[] entryFileGetters() { + InternalRow.FieldGetter[] getters = + new InternalRow.FieldGetter[ENTRY_FILE_FIELD_NAMES.length]; + for (int field = 0; field < getters.length; field++) { + int position = DataFileMeta.SCHEMA.getFieldIndex(ENTRY_FILE_FIELD_NAMES[field]); + getters[field] = + InternalRow.createFieldGetter( + DataFileMeta.SCHEMA.getTypeAt(position), position); + } + return getters; + } + + private static RowType entryLayout() { + List fields = new ArrayList<>(); + fields.add(ManifestEntry.MANIFEST_ROW_TYPE.getField(ManifestEntry.KIND)); + fields.add(ManifestEntry.MANIFEST_ROW_TYPE.getField(ManifestEntry.PARTITION)); + fields.add(ManifestEntry.MANIFEST_ROW_TYPE.getField(ManifestEntry.BUCKET)); + fields.add( + ManifestEntry.MANIFEST_ROW_TYPE + .getField(ManifestEntry.FILE) + .newType( + DataFileMeta.SCHEMA.project( + DataFileMeta.FILE_NAME, + DataFileMeta.ROW_COUNT, + DataFileMeta.LEVEL, + DataFileMeta.SCHEMA_ID, + DataFileMeta.FIRST_ROW_ID, + DataFileMeta.MAX_SEQUENCE_NUMBER, + DataFileMeta.EXTRA_FILES, + DataFileMeta.EMBEDDED_FILE_INDEX, + DataFileMeta.EXTERNAL_PATH))); + return new RowType(false, fields); + } + + static GenericRow projectEntryLayout( + GenericRow fullRow, GenericRow reuse, GenericRow reuseFile) { + reuse.setField(KIND, fullRow.getByte(FULL_KIND)); + reuse.setField(PARTITION, fullRow.getBinary(FULL_PARTITION)); + reuse.setField(BUCKET, fullRow.getInt(FULL_BUCKET)); + InternalRow fullFile = fullRow.getRow(FULL_FILE, DataFileMeta.SCHEMA.getFieldCount()); + for (int field = 0; field < ENTRY_FILE_GETTERS.length; field++) { + reuseFile.setField(field, ENTRY_FILE_GETTERS[field].getFieldOrNull(fullFile)); + } + reuse.setField(FILE, reuseFile); + return reuse; + } + + /** + * Returns null when the input is too fragmented for a bounded streaming merge. The caller must + * fall back to the spillable external sorter in that case. + */ + @Nullable + static List sortAndWriteFullEntries( + List section, + ManifestFileSorter.RowIdEntrySortKey sortKey, + RowType partitionType, + ManifestFile manifestFile, + List newFilesForAbort, + CompactFileIdentifierSet deletedIdentifiers, + DeletedRowIdSet deletedRowIds, + int maxNumFileHandles, + @Nullable Integer manifestReadParallelism) + throws Exception { + ManifestEntryRunMergeEntry.Filter filter = + new ManifestEntryRunMergeEntry.Filter(deletedIdentifiers, deletedRowIds, true); + ManifestEntryRunMergePlan plan = + discoverRuns( + section, + sortKey, + partitionType, + manifestFile, + filter, + maxNumFileHandles, + manifestReadParallelism); + if (plan == null) { + return null; + } + return plan.mergeToManifest(sortKey, manifestFile, filter, newFilesForAbort); + } + + /** + * Returns null when the input is too fragmented for a bounded streaming merge or primitive + * manifest reading is unavailable. The caller must fall back to the spillable external sorter. + */ + @Nullable + static Pair, List> sortAndWriteMinorEntries( + List section, + ManifestFileSorter.RowIdEntrySortKey sortKey, + RowType partitionType, + ManifestFile manifestFile, + List newFilesForAbort, + int maxNumFileHandles, + @Nullable Integer manifestReadParallelism) + throws Exception { + CompactFileIdentifierSet deletedIdentifiers = new CompactFileIdentifierSet(); + DeletedRowIdSet deletedRowIds = new DeletedRowIdSet(); + ManifestEntryRunMergeEntry.Filter.Minor filter = + new ManifestEntryRunMergeEntry.Filter.Minor( + deletedIdentifiers, deletedRowIds, true); + try { + ManifestEntryRunMergePlan plan; + try { + plan = + discoverRuns( + section, + sortKey, + partitionType, + manifestFile, + filter, + maxNumFileHandles, + manifestReadParallelism); + } finally { + deletedRowIds.releaseRangeIndex(); + } + if (plan == null) { + return null; + } + return plan.mergeMinorToManifest( + sortKey, + manifestFile, + filter, + deletedIdentifiers, + deletedRowIds, + newFilesForAbort); + } finally { + deletedIdentifiers.release(); + } + } + + @Nullable + private static ManifestEntryRunMergePlan discoverRuns( + List section, + ManifestFileSorter.RowIdEntrySortKey sortKey, + RowType partitionType, + ManifestFile manifestFile, + ManifestEntryRunMergeEntry.Filter filter, + int maxNumFileHandles, + @Nullable Integer manifestReadParallelism) + throws Exception { + ManifestEntryRunMergeEntry.PartitionDictionary partitions = + new ManifestEntryRunMergeEntry.PartitionDictionary(sortKey); + List sources = new ArrayList<>(); + int streamCursorCount = 0; + long inMemoryEntries = 0; + List discovered = new ArrayList<>(section.size()); + if (section.size() <= 1 + || (manifestReadParallelism != null && manifestReadParallelism <= 1)) { + for (ManifestFileMeta meta : section) { + Discovery.DiscoveredManifest manifest = + discoverManifestRuns(meta, manifestFile, partitionType, partitions, filter); + if (manifest.requiresExternalSort) { + return null; + } + discovered.add(manifest); + } + } else { + Function> reader = + meta -> { + try { + return Collections.singletonList( + discoverManifestRuns( + meta, manifestFile, partitionType, partitions, filter)); + } catch (Exception e) { + throw new RuntimeException( + "Failed to discover sorted Avro runs in " + meta.fileName(), e); + } + }; + for (Discovery.DiscoveredManifest manifest : + sequentialBatchedExecute(reader, section, manifestReadParallelism)) { + discovered.add(manifest); + } + } + for (int manifestIndex = 0; manifestIndex < section.size(); manifestIndex++) { + ManifestFileMeta meta = section.get(manifestIndex); + Discovery.DiscoveredManifest manifest = discovered.get(manifestIndex); + if (manifest.requiresExternalSort) { + return null; + } + if (manifest.fragmented) { + long entryCount = meta.numAddedFiles() + meta.numDeletedFiles(); + inMemoryEntries += entryCount; + if (inMemoryEntries > MAX_IN_MEMORY_FRAGMENTED_ENTRIES) { + return null; + } + sources.add(new ManifestEntryRunMergePlan.Source.FragmentedManifestSpec(meta)); + streamCursorCount++; + } else { + sources.addAll(manifest.runs); + streamCursorCount += manifest.runs.size(); + } + if (streamCursorCount > Math.min(MAX_STREAM_CURSORS, maxNumFileHandles)) { + return null; + } + } + partitions.finish(); + for (Discovery.DiscoveredManifest manifest : discovered) { + manifest.finishFiltering(filter); + manifest.updatePartitionRanks(partitions); + } + return new ManifestEntryRunMergePlan(sources, partitions); + } + + private static Discovery.DiscoveredManifest discoverManifestRuns( + ManifestFileMeta meta, + ManifestFile manifestFile, + RowType partitionType, + ManifestEntryRunMergeEntry.PartitionDictionary partitions, + ManifestEntryRunMergeEntry.Filter filter) + throws Exception { + try (ManifestAvroReader reader = + manifestFile.scanAvroBlocks(meta.fileName(), meta.fileSize())) { + return discoverManifestRuns(meta, reader, partitionType, partitions, filter); + } catch (UnsupportedOperationException unsupported) { + return Discovery.DiscoveredManifest.requiresExternalSort(); + } + } + + private static Discovery.DiscoveredManifest discoverManifestRuns( + ManifestFileMeta meta, + ManifestAvroReader reader, + RowType partitionType, + ManifestEntryRunMergeEntry.PartitionDictionary partitions, + ManifestEntryRunMergeEntry.Filter filter) + throws Exception { + SimpleStatsConverter partitionStatsConverter = new SimpleStatsConverter(partitionType); + List runs = new ArrayList<>(); + List blocks = new ArrayList<>(); + ManifestEntryRunMergeEntry.Key previous = new ManifestEntryRunMergeEntry.Key(); + ManifestEntryRunMergeEntry.Key current = new ManifestEntryRunMergeEntry.Key(); + boolean hasPrevious = false; + long runStart = 0; + long position = 0; + long entryCount = meta.numAddedFiles() + meta.numDeletedFiles(); + boolean fragmented = false; + while (reader.hasNext()) { + RawBlock rawBlock = reader.next(); + RowIterator rows = rawBlock.toRows(ENTRY_LAYOUT); + while (rows.hasNext()) { + GenericRow row = rows.next(); + current.replace(row, partitions); + filter.observe(row, current); + if (fragmented) { + position++; + continue; + } + if (rows.recordIndex() == 0) { + blocks.add( + new Discovery.BlockInfo( + rawBlock.blockOrdinal(), + position, + rawBlock.rawBlockCopySupported(), + current.stableCopy(), + partitionType)); + } + Discovery.BlockInfo block = blocks.get(blocks.size() - 1); + block.collectForSort(row, current, partitions, filter); + boolean inversion = + hasPrevious && compareDiscoveryKeys(previous, current, partitions) > 0; + if (inversion) { + if (rows.recordIndex() > 0) { + block.sorted = false; + } + runs.add( + new ManifestEntryRunMergePlan.Source.ManifestRunSpec( + meta, runStart, position, blocks)); + runStart = position; + if (runs.size() >= FRAGMENTED_RUN_THRESHOLD) { + if (entryCount > MAX_IN_MEMORY_FRAGMENTED_ENTRIES) { + return Discovery.DiscoveredManifest.requiresExternalSort(); + } + fragmented = true; + runs.clear(); + blocks.clear(); + position++; + continue; + } + } + position++; + if (rows.recordIndex() + 1 == rawBlock.recordCount()) { + ManifestEntryRunMergeEntry.Key stableLastKey = current.stableCopy(); + block.finishSort(position, stableLastKey, partitionStatsConverter); + previous.copyFrom(stableLastKey); + } else { + previous.copyFrom(current); + } + hasPrevious = true; + } + } + if (fragmented) { + return Discovery.DiscoveredManifest.fragmented(); + } + if (position > runStart) { + runs.add( + new ManifestEntryRunMergePlan.Source.ManifestRunSpec( + meta, runStart, position, blocks)); + } + if (exceedsStreamingReadAmplification(runs, blocks.size())) { + return entryCount > MAX_IN_MEMORY_FRAGMENTED_ENTRIES + ? Discovery.DiscoveredManifest.requiresExternalSort() + : Discovery.DiscoveredManifest.fragmented(); + } + return Discovery.DiscoveredManifest.runs(runs, blocks); + } + + private static boolean exceedsStreamingReadAmplification( + List runs, int blockCount) { + if (runs.size() <= 1 || blockCount == 0) { + return false; + } + + long prefixBlocksRead = 0; + for (ManifestEntryRunMergePlan.Source.ManifestRunSpec run : runs) { + prefixBlocksRead += run.prefixBlockCount(); + } + return prefixBlocksRead > (long) blockCount * MAX_STREAM_READ_AMPLIFICATION; + } + + private static int compareDiscoveryKeys( + ManifestEntryRunMergeEntry.Key left, + ManifestEntryRunMergeEntry.Key right, + ManifestEntryRunMergeEntry.PartitionDictionary partitions) { + return compareRemainingKeys( + left, right, partitions.compareIds(left.partitionId, right.partitionId)); + } + + static int compareMergeKeys( + ManifestEntryRunMergeEntry.Key left, ManifestEntryRunMergeEntry.Key right) { + return compareRemainingKeys( + left, right, Integer.compare(left.partitionRank, right.partitionRank)); + } + + private static int compareRemainingKeys( + ManifestEntryRunMergeEntry.Key left, + ManifestEntryRunMergeEntry.Key right, + int comparison) { + if (comparison == 0) { + comparison = Byte.compare(left.kind, right.kind); + } + if (comparison == 0) { + comparison = Long.compare(left.firstRowId, right.firstRowId); + } + if (comparison == 0) { + comparison = Long.compare(left.rangeEnd, right.rangeEnd); + } + if (comparison == 0) { + comparison = Long.compare(left.reverseSequence, right.reverseSequence); + } + if (comparison == 0) { + comparison = compareBytes(left, right); + } + return comparison; + } + + private static int compareBytes( + ManifestEntryRunMergeEntry.Key left, ManifestEntryRunMergeEntry.Key right) { + int minLength = Math.min(left.fileNameLength, right.fileNameLength); + for (int i = 0; i < minLength; i++) { + int leftByte = left.fileNameBytes[left.fileNameOffset + i] & 0xFF; + int rightByte = right.fileNameBytes[right.fileNameOffset + i] & 0xFF; + if (leftByte != rightByte) { + return leftByte - rightByte; + } + } + return left.fileNameLength - right.fileNameLength; + } + + /** Results and Avro block metadata collected while discovering natural manifest runs. */ + static final class Discovery { + + private Discovery() {} + + static final class DiscoveredManifest { + + final List runs; + final List blocks; + final boolean fragmented; + final boolean requiresExternalSort; + + DiscoveredManifest( + List runs, + List blocks, + boolean fragmented, + boolean requiresExternalSort) { + this.runs = runs; + this.blocks = blocks; + this.fragmented = fragmented; + this.requiresExternalSort = requiresExternalSort; + } + + static DiscoveredManifest runs( + List runs, + List blocks) { + return new DiscoveredManifest(runs, blocks, false, false); + } + + static DiscoveredManifest fragmented() { + return new DiscoveredManifest( + Collections.emptyList(), Collections.emptyList(), true, false); + } + + static DiscoveredManifest requiresExternalSort() { + return new DiscoveredManifest( + Collections.emptyList(), Collections.emptyList(), false, true); + } + + void updatePartitionRanks(ManifestEntryRunMergeEntry.PartitionDictionary partitions) { + for (BlockInfo block : blocks) { + block.updatePartitionRanks(partitions); + } + } + + void finishFiltering(ManifestEntryRunMergeEntry.Filter filter) { + for (BlockInfo block : blocks) { + block.finishFiltering(filter); + } + } + } + + static final class BlockInfo { + + final long ordinal; + final long start; + final @Nullable ManifestEntryRunMergeEntry.Key firstKey; + boolean eligible; + boolean sorted = true; + long end; + @Nullable ManifestEntryRunMergeEntry.Key lastKey; + long addedFiles; + long deletedFiles; + long schemaId = Long.MIN_VALUE; + int minBucket = Integer.MAX_VALUE; + int maxBucket = Integer.MIN_VALUE; + int minLevel = Integer.MAX_VALUE; + int maxLevel = Integer.MIN_VALUE; + long minRowId = Long.MAX_VALUE; + long maxRowId = Long.MIN_VALUE; + final boolean singleFieldSortedPartitionStats; + @Nullable SimpleStatsCollector partitionStats; + final RowType partitionType; + @Nullable BinaryRow nullPartition; + long nullPartitionCount; + @Nullable BinaryRow minNonNullPartition; + @Nullable BinaryRow maxNonNullPartition; + EncodedBlockMeta metadata; + + BlockInfo( + long ordinal, + long start, + boolean eligible, + ManifestEntryRunMergeEntry.Key firstKey, + RowType partitionType) { + this.ordinal = ordinal; + this.start = start; + this.eligible = eligible; + this.firstKey = firstKey; + this.partitionType = partitionType; + this.singleFieldSortedPartitionStats = + eligible && firstKey != null && partitionType.getFieldCount() == 1; + this.partitionStats = + eligible && firstKey != null && !singleFieldSortedPartitionStats + ? new SimpleStatsCollector(partitionType) + : null; + } + + void collectForSort( + GenericRow record, + ManifestEntryRunMergeEntry.Key key, + ManifestEntryRunMergeEntry.PartitionDictionary partitions, + ManifestEntryRunMergeEntry.Filter filter) { + if (!eligible) { + return; + } + if (!filter.copyable(record, key)) { + eligible = false; + releasePartitionStats(); + return; + } + collectEntryStats(record, key); + BinaryRow partition = partitions.partition(key.partitionId); + if (singleFieldSortedPartitionStats) { + if (partition.isNullAt(0)) { + nullPartition = partition; + nullPartitionCount++; + } else { + if (minNonNullPartition == null) { + minNonNullPartition = partition; + } + maxNonNullPartition = partition; + } + } else { + partitionStats.collect(partition); + } + } + + private void collectEntryStats(GenericRow record, ManifestEntryRunMergeEntry.Key key) { + InternalRow file = ManifestEntryRunMergeEntry.file(record); + if (key.kind == FileKind.ADD.toByteValue()) { + addedFiles++; + } else { + deletedFiles++; + } + schemaId = Math.max(schemaId, file.getLong(SCHEMA_ID)); + int bucket = record.getInt(BUCKET); + minBucket = Math.min(minBucket, bucket); + maxBucket = Math.max(maxBucket, bucket); + int level = file.getInt(LEVEL); + minLevel = Math.min(minLevel, level); + maxLevel = Math.max(maxLevel, level); + minRowId = Math.min(minRowId, key.firstRowId); + maxRowId = Math.max(maxRowId, key.rangeEnd); + } + + void finishSort( + long end, + ManifestEntryRunMergeEntry.Key lastKey, + SimpleStatsConverter partitionStatsConverter) { + this.end = end; + this.lastKey = lastKey; + if (eligible && sorted) { + SimpleStats encodedPartitionStats; + if (singleFieldSortedPartitionStats) { + BinaryRow min = + minNonNullPartition == null ? nullPartition : minNonNullPartition; + BinaryRow max = + maxNonNullPartition == null ? nullPartition : maxNonNullPartition; + checkState(min != null && max != null, "Manifest block has no partition."); + encodedPartitionStats = + new SimpleStats( + min, + max, + BinaryArray.fromLongArray(new Long[] {nullPartitionCount})); + } else { + checkState( + partitionStats != null, "Manifest block has no partition stats."); + encodedPartitionStats = + partitionStatsConverter.toBinaryAllMode(partitionStats.extract()); + } + metadata = + new EncodedBlockMeta( + addedFiles, + deletedFiles, + schemaId, + minBucket, + maxBucket, + minLevel, + maxLevel, + minRowId, + maxRowId, + encodedPartitionStats); + } + releasePartitionStats(); + } + + private void releasePartitionStats() { + partitionStats = null; + nullPartition = null; + minNonNullPartition = null; + maxNonNullPartition = null; + } + + boolean copyable(long runStart, long runEnd) { + return metadata != null && start >= runStart && end <= runEnd; + } + + void finishFiltering(ManifestEntryRunMergeEntry.Filter filter) { + if (metadata != null && !filter.copyableAfterDiscovery(minRowId, maxRowId)) { + metadata = null; + } + } + + void updatePartitionRanks(ManifestEntryRunMergeEntry.PartitionDictionary partitions) { + checkState(firstKey != null && lastKey != null, "Manifest block has no sort keys."); + firstKey.partitionRank = partitions.rank(firstKey.partitionId); + lastKey.partitionRank = partitions.rank(lastKey.partitionId); + } + } + } +} diff --git a/paimon-core/src/main/java/org/apache/paimon/operation/ManifestEntryRunMergeEntry.java b/paimon-core/src/main/java/org/apache/paimon/operation/ManifestEntryRunMergeEntry.java new file mode 100644 index 000000000000..5e8c318e9fda --- /dev/null +++ b/paimon-core/src/main/java/org/apache/paimon/operation/ManifestEntryRunMergeEntry.java @@ -0,0 +1,339 @@ +/* + * 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.data.BinaryRow; +import org.apache.paimon.data.BinaryString; +import org.apache.paimon.data.GenericRow; +import org.apache.paimon.data.InternalRow; +import org.apache.paimon.manifest.CompactFileIdentifierSet; +import org.apache.paimon.manifest.DeletedRowIdSet; +import org.apache.paimon.manifest.FileEntry.ReusableIdentifier; +import org.apache.paimon.manifest.FileKind; +import org.apache.paimon.manifest.ProjectedManifestEntry; +import org.apache.paimon.memory.MemorySegmentUtils; +import org.apache.paimon.utils.ByteArrayKey; +import org.apache.paimon.utils.ByteArrayLookupKey; +import org.apache.paimon.utils.SerializationUtils; + +import java.util.ArrayList; +import java.util.Arrays; +import java.util.List; +import java.util.Map; +import java.util.concurrent.ConcurrentHashMap; + +import static org.apache.paimon.utils.Preconditions.checkState; + +/** Entry-level state shared by manifest run discovery and merge execution. */ +final class ManifestEntryRunMergeEntry { + + private ManifestEntryRunMergeEntry() {} + + static final class Key { + + int partitionId; + int partitionRank; + byte kind; + boolean hasRowId; + long firstRowId; + long rangeEnd; + long reverseSequence; + byte[] fileNameBytes; + int fileNameOffset; + int fileNameLength; + + static Key viewOf(ProjectedManifestEntry entry, PartitionDictionary partitions) { + Key key = new Key(); + key.replace(entry, partitions); + return key; + } + + void replace(ProjectedManifestEntry entry, PartitionDictionary partitions) { + long firstRowId = entry.file().nonNullFirstRowId(); + this.partitionId = partitions.id(entry.partitionBytes()); + this.partitionRank = partitions.rank(partitionId); + this.kind = entry.kind().toByteValue(); + this.hasRowId = true; + this.firstRowId = firstRowId; + this.rangeEnd = firstRowId + entry.file().rowCount() - 1L; + this.reverseSequence = Long.MAX_VALUE - entry.file().maxSequenceNumber(); + this.fileNameBytes = entry.file().fileNameBinary().toBytes(); + this.fileNameOffset = 0; + this.fileNameLength = fileNameBytes.length; + } + + void replace(GenericRow record, PartitionDictionary partitions) { + InternalRow file = file(record); + checkState( + !file.isNullAt(ManifestEntryRunMerge.FIRST_ROW_ID), + "First row id should not be null."); + this.partitionId = partitions.id(record.getBinary(ManifestEntryRunMerge.PARTITION)); + this.partitionRank = partitions.rank(partitionId); + this.kind = record.getByte(ManifestEntryRunMerge.KIND); + this.hasRowId = true; + this.firstRowId = file.getLong(ManifestEntryRunMerge.FIRST_ROW_ID); + this.rangeEnd = firstRowId + file.getLong(ManifestEntryRunMerge.ROW_COUNT) - 1L; + this.reverseSequence = + Long.MAX_VALUE - file.getLong(ManifestEntryRunMerge.MAX_SEQUENCE_NUMBER); + BinaryString fileName = file.getString(ManifestEntryRunMerge.FILE_NAME); + this.fileNameBytes = + MemorySegmentUtils.copyToBytes( + fileName.getSegments(), + fileName.getOffset(), + fileName.getSizeInBytes()); + this.fileNameOffset = 0; + this.fileNameLength = fileNameBytes.length; + } + + void replaceForCompaction(GenericRow record) { + InternalRow file = file(record); + this.kind = record.getByte(ManifestEntryRunMerge.KIND); + this.hasRowId = !file.isNullAt(ManifestEntryRunMerge.FIRST_ROW_ID); + if (hasRowId) { + this.firstRowId = file.getLong(ManifestEntryRunMerge.FIRST_ROW_ID); + this.rangeEnd = firstRowId + file.getLong(ManifestEntryRunMerge.ROW_COUNT) - 1L; + } + } + + void copyFrom(Key key) { + this.partitionId = key.partitionId; + this.partitionRank = key.partitionRank; + this.kind = key.kind; + this.hasRowId = key.hasRowId; + this.firstRowId = key.firstRowId; + this.rangeEnd = key.rangeEnd; + this.reverseSequence = key.reverseSequence; + this.fileNameBytes = key.fileNameBytes; + this.fileNameOffset = key.fileNameOffset; + this.fileNameLength = key.fileNameLength; + } + + Key stableCopy() { + Key copy = new Key(); + copy.copyFrom(this); + copy.fileNameBytes = + Arrays.copyOfRange( + fileNameBytes, fileNameOffset, fileNameOffset + fileNameLength); + copy.fileNameOffset = 0; + return copy; + } + + void clear() { + fileNameBytes = null; + } + } + + /** Interns variable-width partition bytes once and assigns comparator-compatible ranks. */ + static final class PartitionDictionary { + + final ManifestFileSorter.RowIdEntrySortKey sortKey; + final Map ids = new ConcurrentHashMap<>(); + final ThreadLocal lookup = + ThreadLocal.withInitial(ByteArrayLookupKey::new); + volatile BinaryRow[] partitions = new BinaryRow[16]; + int partitionCount; + int[] ranks; + + PartitionDictionary(ManifestFileSorter.RowIdEntrySortKey sortKey) { + this.sortKey = sortKey; + } + + PartitionDictionary() { + this.sortKey = null; + } + + int id(byte[] bytes) { + return id(bytes, 0, bytes.length); + } + + int id(byte[] bytes, int offset, int length) { + ByteArrayLookupKey lookupKey = lookup.get(); + lookupKey.reset(bytes, offset, length); + try { + Integer existing = ids.get(lookupKey); + if (existing != null) { + return existing; + } + synchronized (this) { + existing = ids.get(lookupKey); + if (existing != null) { + return existing; + } + checkState(ranks == null, "Full manifest scan found an unknown partition."); + byte[] canonical = Arrays.copyOfRange(bytes, offset, offset + length); + int id = partitionCount; + if (id == partitions.length) { + partitions = Arrays.copyOf(partitions, partitions.length << 1); + } + partitions[id] = SerializationUtils.deserializeBinaryRow(canonical); + ids.put(new ByteArrayKey(canonical), id); + partitionCount = id + 1; + return id; + } + } finally { + lookupKey.clear(); + } + } + + int compareIds(int left, int right) { + checkState(sortKey != null, "Partition dictionary has no sort key."); + return sortKey.comparePartitions(partitions[left], partitions[right]); + } + + void finish() { + List order = new ArrayList<>(partitionCount); + for (int id = 0; id < partitionCount; id++) { + order.add(id); + } + order.sort((left, right) -> compareIds(left, right)); + ranks = new int[partitionCount]; + int rank = 0; + for (int position = 0; position < order.size(); position++) { + if (position > 0 && compareIds(order.get(position - 1), order.get(position)) != 0) { + rank++; + } + ranks[order.get(position)] = rank; + } + } + + int rank(int id) { + return ranks == null ? 0 : ranks[id]; + } + + BinaryRow partition(int id) { + return partitions[id]; + } + } + + static class Filter { + + final CompactFileIdentifierSet deletedIdentifiers; + final DeletedRowIdSet deletedRowIds; + final boolean useRowIdFilter; + final ThreadLocal identifier = + ThreadLocal.withInitial(IdentifierEncoder::new); + + Filter( + CompactFileIdentifierSet deletedIdentifiers, + DeletedRowIdSet deletedRowIds, + boolean useRowIdFilter) { + this.deletedIdentifiers = deletedIdentifiers; + this.deletedRowIds = deletedRowIds; + this.useRowIdFilter = useRowIdFilter; + } + + boolean include(ProjectedManifestEntry entry) { + return entry.isAdd() && !deletedIdentifiers.contains(entry); + } + + boolean include(GenericRow record, Key key) { + return key.kind == FileKind.ADD.toByteValue() && !isDeleted(record, key); + } + + boolean copyable(GenericRow record, Key key) { + return include(record, key); + } + + void observe(GenericRow record, Key key) {} + + boolean copyableAfterDiscovery(long minRowId, long maxRowId) { + return true; + } + + ReusableIdentifier identifier(GenericRow record) { + return identifier.get().replace(record); + } + + boolean isDeleted(GenericRow record, Key key) { + // RowID is only a cheap negative filter. The complete identifier remains the + // authoritative match, and is also sufficient for manifests which predate RowID. + if (useRowIdFilter) { + checkState(key.hasRowId, "First row id should not be null."); + if (!deletedRowIds.contains(key.firstRowId)) { + return false; + } + } + return deletedIdentifiers.contains(identifier(record)); + } + + static final class Minor extends Filter { + + Minor( + CompactFileIdentifierSet deletedIdentifiers, + DeletedRowIdSet deletedRowIds, + boolean useRowIdFilter) { + super(deletedIdentifiers, deletedRowIds, useRowIdFilter); + } + + @Override + boolean include(ProjectedManifestEntry entry) { + return true; + } + + @Override + boolean include(GenericRow record, Key key) { + return true; + } + + @Override + boolean copyable(GenericRow record, Key key) { + return key.kind == FileKind.ADD.toByteValue(); + } + + @Override + void observe(GenericRow record, Key key) { + if (key.kind != FileKind.DELETE.toByteValue()) { + return; + } + ReusableIdentifier reusable = identifier(record); + synchronized (this) { + deletedIdentifiers.add(reusable); + if (useRowIdFilter) { + checkState(key.hasRowId, "First row id should not be null."); + deletedRowIds.add(key.firstRowId); + } + } + } + + @Override + boolean copyableAfterDiscovery(long minRowId, long maxRowId) { + // A DELETE preserves the deleted ADD's globally unique first RowID. A range hit may + // be a false positive and only disables block copying; a miss proves the block has + // no deleted ADD. + return useRowIdFilter && !deletedRowIds.intersects(minRowId, maxRowId); + } + } + + private static final class IdentifierEncoder { + + final ProjectedManifestEntry entry = + ProjectedManifestEntry.Projection.create(ManifestEntryRunMerge.ENTRY_LAYOUT) + .createEntry(); + final ReusableIdentifier identifier = new ReusableIdentifier(); + + ReusableIdentifier replace(GenericRow record) { + return identifier.replaceWithPartition(entry.replace(record)); + } + } + } + + static InternalRow file(GenericRow record) { + return record.getRow(ManifestEntryRunMerge.FILE, ManifestEntryRunMerge.FILE_FIELD_COUNT); + } +} diff --git a/paimon-core/src/main/java/org/apache/paimon/operation/ManifestEntryRunMergePlan.java b/paimon-core/src/main/java/org/apache/paimon/operation/ManifestEntryRunMergePlan.java new file mode 100644 index 000000000000..eb7702b8aa5f --- /dev/null +++ b/paimon-core/src/main/java/org/apache/paimon/operation/ManifestEntryRunMergePlan.java @@ -0,0 +1,839 @@ +/* + * 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.data.BinaryRow; +import org.apache.paimon.data.GenericRow; +import org.apache.paimon.data.InternalRow; +import org.apache.paimon.data.serializer.InternalRowSerializer; +import org.apache.paimon.format.avro.AvroRawBlock; +import org.apache.paimon.manifest.CompactFileIdentifierSet; +import org.apache.paimon.manifest.DeletedRowIdSet; +import org.apache.paimon.manifest.FileEntry.ReusableIdentifier; +import org.apache.paimon.manifest.FileKind; +import org.apache.paimon.manifest.ManifestAvroReader; +import org.apache.paimon.manifest.ManifestAvroReader.RawBlock; +import org.apache.paimon.manifest.ManifestAvroReader.RowIterator; +import org.apache.paimon.manifest.ManifestAvroWriter; +import org.apache.paimon.manifest.ManifestAvroWriter.EncodedBlockMeta; +import org.apache.paimon.manifest.ManifestAvroWriter.EncodedEntry; +import org.apache.paimon.manifest.ManifestEntry; +import org.apache.paimon.manifest.ManifestFile; +import org.apache.paimon.manifest.ManifestFileMeta; +import org.apache.paimon.manifest.ProjectedManifestEntry; +import org.apache.paimon.utils.CloseableIterator; +import org.apache.paimon.utils.Pair; + +import javax.annotation.Nullable; + +import java.nio.ByteBuffer; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collections; +import java.util.List; + +import static org.apache.paimon.utils.Preconditions.checkState; + +final class ManifestEntryRunMergePlan { + + final List sources; + final ManifestEntryRunMergeEntry.PartitionDictionary partitions; + + ManifestEntryRunMergePlan( + List sources, ManifestEntryRunMergeEntry.PartitionDictionary partitions) { + this.sources = sources; + this.partitions = partitions; + } + + List mergeToManifest( + ManifestFileSorter.RowIdEntrySortKey sortKey, + ManifestFile manifestFile, + ManifestEntryRunMergeEntry.Filter filter, + List newFilesForAbort) + throws Exception { + List cursors = new ArrayList<>(sources.size()); + Exception failure = null; + try { + for (Source.Spec source : sources) { + Cursor cursor = source.open(manifestFile, sortKey, filter, partitions); + cursors.add(cursor); + cursor.advance(); + } + SelectionTree selectionTree = new SelectionTree(cursors); + if (selectionTree.winner() < 0) { + return Collections.emptyList(); + } + List files = writeSelected(selectionTree, manifestFile); + newFilesForAbort.addAll(files); + return files; + } catch (Exception e) { + failure = e; + throw e; + } finally { + try { + closeCursors(cursors); + } catch (Exception closeFailure) { + if (failure == null) { + throw closeFailure; + } + failure.addSuppressed(closeFailure); + } + } + } + + Pair, List> mergeMinorToManifest( + ManifestFileSorter.RowIdEntrySortKey sortKey, + ManifestFile manifestFile, + ManifestEntryRunMergeEntry.Filter filter, + CompactFileIdentifierSet deletedIdentifiers, + DeletedRowIdSet deletedRowIds, + List newFilesForAbort) + throws Exception { + List cursors = new ArrayList<>(sources.size()); + Exception failure = null; + try { + for (Source.Spec source : sources) { + Cursor cursor = source.open(manifestFile, sortKey, filter, partitions); + cursors.add(cursor); + cursor.advance(); + } + SelectionTree selectionTree = new SelectionTree(cursors); + if (selectionTree.winner() < 0) { + return Pair.of(Collections.emptyList(), Collections.emptyList()); + } + Pair, List> files = + writeMinorSelected( + selectionTree, manifestFile, deletedIdentifiers, deletedRowIds); + newFilesForAbort.addAll(files.getLeft()); + newFilesForAbort.addAll(files.getRight()); + return files; + } catch (Exception e) { + failure = e; + throw e; + } finally { + try { + closeCursors(cursors); + } catch (Exception closeFailure) { + if (failure == null) { + throw closeFailure; + } + failure.addSuppressed(closeFailure); + } + } + } + + static List writeSelected( + SelectionTree selectionTree, ManifestFile manifestFile) throws Exception { + ManifestAvroWriter writer = manifestFile.createAvroWriter(); + Exception failure = null; + try { + int winner; + while ((winner = selectionTree.winner()) >= 0) { + Cursor cursor = selectionTree.cursor(winner); + if (cursor.hasCopyableBlock() + && selectionTree.blockPrecedesOthers(winner, cursor.blockLastKey())) { + writer.writeEncodedBlock(cursor.encodedBlock(), cursor.blockMetadata()); + selectionTree.update(winner, cursor.advanceAfterBlock()); + continue; + } + cursor.materializeCurrent(); + writeCurrent(writer, cursor); + selectionTree.update(winner, cursor.advance()); + } + } catch (Exception e) { + failure = e; + } finally { + if (failure != null) { + writer.abort(); + throw failure; + } + writer.close(); + } + return writer.result(); + } + + private static Pair, List> writeMinorSelected( + SelectionTree selectionTree, + ManifestFile manifestFile, + CompactFileIdentifierSet deletedIdentifiers, + DeletedRowIdSet deletedRowIds) + throws Exception { + ManifestAvroWriter addWriter = manifestFile.createAvroWriter(); + ManifestAvroWriter deleteWriter = manifestFile.createAvroWriter(); + CompactFileIdentifierSet matchedEntries = new CompactFileIdentifierSet(); + CompactFileIdentifierSet emittedDeletes = new CompactFileIdentifierSet(); + Exception failure = null; + try { + int winner; + while ((winner = selectionTree.winner()) >= 0) { + Cursor cursor = selectionTree.cursor(winner); + if (cursor.hasCopyableBlock() + && selectionTree.blockPrecedesOthers(winner, cursor.blockLastKey())) { + addWriter.writeEncodedBlock(cursor.encodedBlock(), cursor.blockMetadata()); + selectionTree.update(winner, cursor.advanceAfterBlock()); + continue; + } + + cursor.materializeCurrent(); + if (cursor.key().kind == FileKind.ADD.toByteValue()) { + if (!deletedRowIds.contains(cursor.key().firstRowId)) { + writeCurrent(addWriter, cursor); + } else { + ReusableIdentifier identifier = cursor.identifier(); + if (deletedIdentifiers.contains(identifier)) { + matchedEntries.add(identifier); + } else { + writeCurrent(addWriter, cursor); + } + } + } else { + ReusableIdentifier identifier = cursor.identifier(); + if (!matchedEntries.contains(identifier) + && !emittedDeletes.contains(identifier)) { + emittedDeletes.add(identifier); + writeCurrent(deleteWriter, cursor); + } + } + selectionTree.update(winner, cursor.advance()); + } + addWriter.close(); + deleteWriter.close(); + } catch (Exception e) { + failure = e; + } finally { + matchedEntries.release(); + emittedDeletes.release(); + if (failure != null) { + addWriter.abort(); + deleteWriter.abort(); + throw failure; + } + } + return Pair.of(addWriter.result(), deleteWriter.result()); + } + + private static void writeCurrent(ManifestAvroWriter writer, Cursor cursor) throws Exception { + InternalRow decodedRow = cursor.decodedRow(); + if (decodedRow != null) { + writer.writeRow(decodedRow, cursor.metadata()); + return; + } + ByteBuffer encodedRecord = cursor.encodedRecord(); + if (encodedRecord == null) { + writer.write(cursor.current()); + } else { + writer.writeEncoded(encodedRecord, cursor.metadata()); + } + } + + static void closeCursors(List cursors) throws Exception { + Exception failure = null; + for (Cursor cursor : cursors) { + try { + cursor.close(); + } catch (Exception e) { + if (failure == null) { + failure = e; + } else { + failure.addSuppressed(e); + } + } + } + if (failure != null) { + throw failure; + } + } + + /** Describes the manifest inputs which become cursors when this plan starts executing. */ + static final class Source { + + private Source() {} + + interface Spec { + + Cursor open( + ManifestFile manifestFile, + ManifestFileSorter.RowIdEntrySortKey sortKey, + ManifestEntryRunMergeEntry.Filter filter, + ManifestEntryRunMergeEntry.PartitionDictionary partitions) + throws Exception; + } + + static final class ManifestRunSpec implements Spec { + + final ManifestFileMeta meta; + final long start; + final long end; + final List blocks; + + ManifestRunSpec( + ManifestFileMeta meta, + long start, + long end, + List blocks) { + this.meta = meta; + this.start = start; + this.end = end; + this.blocks = blocks; + } + + long prefixBlockCount() { + long lastBlockOrdinal = -1; + for (ManifestEntryRunMerge.Discovery.BlockInfo block : blocks) { + if (block.start >= end) { + break; + } + if (block.end > start) { + lastBlockOrdinal = block.ordinal; + } + } + checkState(lastBlockOrdinal >= 0, "Manifest run does not contain an Avro block."); + return lastBlockOrdinal + 1; + } + + @Override + public Cursor open( + ManifestFile manifestFile, + ManifestFileSorter.RowIdEntrySortKey sortKey, + ManifestEntryRunMergeEntry.Filter filter, + ManifestEntryRunMergeEntry.PartitionDictionary partitions) + throws Exception { + return new PrimitiveManifestRunCursor( + manifestFile, meta, start, end, blocks, filter, partitions); + } + } + + static final class FragmentedManifestSpec implements Spec { + + final ManifestFileMeta meta; + + FragmentedManifestSpec(ManifestFileMeta meta) { + this.meta = meta; + } + + @Override + public Cursor open( + ManifestFile manifestFile, + ManifestFileSorter.RowIdEntrySortKey sortKey, + ManifestEntryRunMergeEntry.Filter filter, + ManifestEntryRunMergeEntry.PartitionDictionary partitions) + throws Exception { + return new InMemoryManifestCursor(manifestFile, meta, sortKey, filter, partitions); + } + } + } + + interface Cursor extends AutoCloseable { + + boolean advance() throws Exception; + + boolean hasCurrent(); + + @Nullable + ProjectedManifestEntry current(); + + @Nullable + EncodedEntry metadata(); + + ManifestEntryRunMergeEntry.Key key(); + + @Nullable + ByteBuffer encodedRecord(); + + default @Nullable InternalRow decodedRow() { + return null; + } + + ReusableIdentifier identifier(); + + default boolean hasCopyableBlock() { + return false; + } + + default ManifestEntryRunMergeEntry.Key blockLastKey() { + throw new UnsupportedOperationException(); + } + + default AvroRawBlock encodedBlock() { + throw new UnsupportedOperationException(); + } + + default EncodedBlockMeta blockMetadata() { + throw new UnsupportedOperationException(); + } + + default boolean advanceAfterBlock() throws Exception { + throw new UnsupportedOperationException(); + } + + default void materializeCurrent() throws Exception {} + + @Override + void close() throws Exception; + } + + static final class PrimitiveManifestRunCursor implements Cursor { + + final ManifestAvroReader reader; + final boolean encodedRecordsCompatible; + final ManifestEntryRunMergeEntry.Filter filter; + final ManifestEntryRunMergeEntry.PartitionDictionary partitions; + final ManifestEntryRunMergeEntry.Key key = new ManifestEntryRunMergeEntry.Key(); + final EncodedEntry metadata = new EncodedEntry(); + final List blocks; + final long runStart; + final long runEnd; + int blockIndex; + long nextReaderBlockOrdinal; + long decodedRemaining; + boolean rawBlock; + boolean current; + @Nullable RawBlock currentRawBlock; + @Nullable RowIterator currentRows; + @Nullable GenericRow currentRow; + @Nullable GenericRow currentSourceRow; + @Nullable GenericRow compactRow; + @Nullable GenericRow compactFile; + @Nullable ManifestEntryRunMerge.Discovery.BlockInfo currentBlock; + boolean closed; + + PrimitiveManifestRunCursor( + ManifestFile manifestFile, + ManifestFileMeta meta, + long start, + long end, + List blocks, + ManifestEntryRunMergeEntry.Filter filter, + ManifestEntryRunMergeEntry.PartitionDictionary partitions) + throws Exception { + this.reader = manifestFile.scanAvroBlocks(meta.fileName(), meta.fileSize()); + this.encodedRecordsCompatible = reader.rawBlockCopySupported(); + if (!encodedRecordsCompatible) { + this.compactRow = + new GenericRow(ManifestEntryRunMerge.ENTRY_LAYOUT.getFieldCount()); + this.compactFile = new GenericRow(ManifestEntryRunMerge.FILE_FIELD_COUNT); + } + this.filter = filter; + this.partitions = partitions; + this.blocks = blocks; + this.runStart = start; + this.runEnd = end; + try { + while (blockIndex < blocks.size() && blocks.get(blockIndex).end <= start) { + blockIndex++; + } + checkState( + blockIndex < blocks.size(), + "Manifest run starts after the end of the file."); + } catch (Exception e) { + try { + reader.close(); + } catch (Exception closeFailure) { + e.addSuppressed(closeFailure); + } + throw e; + } + } + + @Override + public boolean advance() throws Exception { + current = false; + while (true) { + if (decodedRemaining == 0) { + if (!prepareNextBlock()) { + key.clear(); + close(); + return false; + } + if (rawBlock) { + return true; + } + } + checkState( + currentRows != null && currentRows.hasNext(), + "Manifest block ends before its discovered boundary."); + currentSourceRow = currentRows.next(); + currentRow = + encodedRecordsCompatible + ? currentSourceRow + : ManifestEntryRunMerge.projectEntryLayout( + currentSourceRow, compactRow, compactFile); + decodedRemaining--; + key.replace(currentRow, partitions); + if (filter.include(currentRow, key)) { + current = true; + InternalRow file = ManifestEntryRunMergeEntry.file(currentRow); + metadata.replace( + key.kind, + partitions.partition(key.partitionId), + currentRow.getInt(ManifestEntryRunMerge.BUCKET), + file.getInt(ManifestEntryRunMerge.LEVEL), + file.getLong(ManifestEntryRunMerge.SCHEMA_ID), + key.firstRowId, + file.getLong(ManifestEntryRunMerge.ROW_COUNT)); + return true; + } + } + } + + boolean prepareNextBlock() throws Exception { + rawBlock = false; + current = false; + currentRows = null; + currentRow = null; + currentSourceRow = null; + while (blockIndex < blocks.size()) { + ManifestEntryRunMerge.Discovery.BlockInfo info = blocks.get(blockIndex); + if (info.start >= runEnd) { + return false; + } + while (nextReaderBlockOrdinal < info.ordinal) { + checkState(reader.hasNext(), "Manifest block ordinal is missing."); + reader.next(); + nextReaderBlockOrdinal++; + } + checkState(reader.hasNext(), "Manifest run ends after the end of the file."); + currentRawBlock = reader.next(); + nextReaderBlockOrdinal++; + currentBlock = info; + if (info.copyable(runStart, runEnd)) { + rawBlock = true; + key.copyFrom(info.firstKey); + return true; + } + + long overlapStart = Math.max(runStart, info.start); + long overlapEnd = Math.min(runEnd, info.end); + long prefix = overlapStart - info.start; + currentRows = + currentRawBlock.toRows( + encodedRecordsCompatible + ? ManifestEntryRunMerge.ENTRY_LAYOUT + : ManifestEntry.MANIFEST_ROW_TYPE); + for (long i = 0; i < prefix; i++) { + checkState( + currentRows.hasNext(), + "Manifest run starts after the end of its block."); + currentRows.next(); + } + decodedRemaining = overlapEnd - overlapStart; + blockIndex++; + if (decodedRemaining > 0) { + return true; + } + } + return false; + } + + @Override + public boolean hasCurrent() { + return current || rawBlock; + } + + @Override + public ProjectedManifestEntry current() { + return null; + } + + @Override + public EncodedEntry metadata() { + return metadata; + } + + @Override + public ManifestEntryRunMergeEntry.Key key() { + return key; + } + + @Override + public ByteBuffer encodedRecord() { + return current && encodedRecordsCompatible ? currentRows.encodedRecord() : null; + } + + @Override + public InternalRow decodedRow() { + return current && !encodedRecordsCompatible ? currentSourceRow : null; + } + + @Override + public ReusableIdentifier identifier() { + checkState(current, "Manifest entry has not been materialized."); + return filter.identifier(currentRow); + } + + @Override + public boolean hasCopyableBlock() { + return rawBlock; + } + + @Override + public ManifestEntryRunMergeEntry.Key blockLastKey() { + return currentBlock.lastKey; + } + + @Override + public AvroRawBlock encodedBlock() { + return currentRawBlock.encodedBlock(); + } + + @Override + public EncodedBlockMeta blockMetadata() { + return currentBlock.metadata; + } + + @Override + public boolean advanceAfterBlock() throws Exception { + checkState(rawBlock, "There is no raw block to advance."); + rawBlock = false; + currentRawBlock = null; + blockIndex++; + return advance(); + } + + @Override + public void materializeCurrent() throws Exception { + if (!rawBlock) { + return; + } + rawBlock = false; + decodedRemaining = currentBlock.end - currentBlock.start; + checkState(decodedRemaining > 0, "Raw Avro block is empty."); + currentRows = + currentRawBlock.toRows( + encodedRecordsCompatible + ? ManifestEntryRunMerge.ENTRY_LAYOUT + : ManifestEntry.MANIFEST_ROW_TYPE); + checkState(currentRows.hasNext(), "Manifest block cannot be decompressed."); + currentSourceRow = currentRows.next(); + currentRow = + encodedRecordsCompatible + ? currentSourceRow + : ManifestEntryRunMerge.projectEntryLayout( + currentSourceRow, compactRow, compactFile); + decodedRemaining--; + key.replace(currentRow, partitions); + checkState( + filter.include(currentRow, key), + "Copyable manifest block contains a filtered entry."); + current = true; + InternalRow file = ManifestEntryRunMergeEntry.file(currentRow); + metadata.replace( + key.kind, + partitions.partition(key.partitionId), + currentRow.getInt(ManifestEntryRunMerge.BUCKET), + file.getInt(ManifestEntryRunMerge.LEVEL), + file.getLong(ManifestEntryRunMerge.SCHEMA_ID), + key.firstRowId, + file.getLong(ManifestEntryRunMerge.ROW_COUNT)); + blockIndex++; + } + + @Override + public void close() throws Exception { + if (closed) { + return; + } + closed = true; + current = false; + currentRawBlock = null; + currentRows = null; + currentRow = null; + currentSourceRow = null; + compactRow = null; + compactFile = null; + currentBlock = null; + rawBlock = false; + key.clear(); + reader.close(); + } + } + + static final class InMemoryManifestCursor implements Cursor { + + final List entries; + final ProjectedManifestEntry current = + ProjectedManifestEntry.fullProjection().createEntry(); + final ReusableIdentifier identifier = new ReusableIdentifier(); + int position = -1; + + InMemoryManifestCursor( + ManifestFile manifestFile, + ManifestFileMeta meta, + ManifestFileSorter.RowIdEntrySortKey sortKey, + ManifestEntryRunMergeEntry.Filter filter, + ManifestEntryRunMergeEntry.PartitionDictionary partitions) + throws Exception { + long entryCount = meta.numAddedFiles() + meta.numDeletedFiles(); + this.entries = new ArrayList<>((int) entryCount); + InternalRowSerializer serializer = + new InternalRowSerializer(ManifestEntry.MANIFEST_ROW_TYPE); + ProjectedManifestEntry view = ProjectedManifestEntry.fullProjection().createEntry(); + try (CloseableIterator iterator = + manifestFile.scan(meta.fileName(), ProjectedManifestEntry.fullProjection())) { + while (iterator.hasNext()) { + ProjectedManifestEntry entry = iterator.next(); + if (!filter.include(entry)) { + continue; + } + BinaryRow row = serializer.toBinaryRow(entry.fullRow()).copy(); + entries.add( + new StoredEntry( + row, + ManifestEntryRunMergeEntry.Key.viewOf( + view.replace(row), partitions))); + } + } + entries.sort( + (left, right) -> ManifestEntryRunMerge.compareMergeKeys(left.key, right.key)); + view.clear(); + } + + @Override + public boolean advance() { + position++; + if (position >= entries.size()) { + current.clear(); + return false; + } + StoredEntry stored = entries.get(position); + current.replace(stored.row); + return true; + } + + @Override + public boolean hasCurrent() { + return position >= 0 && position < entries.size(); + } + + @Override + public ProjectedManifestEntry current() { + return current; + } + + @Override + public EncodedEntry metadata() { + return null; + } + + @Override + public ManifestEntryRunMergeEntry.Key key() { + return entries.get(position).key; + } + + @Override + public ByteBuffer encodedRecord() { + return null; + } + + @Override + public ReusableIdentifier identifier() { + return identifier.replaceWithPartition(current); + } + + @Override + public void close() { + current.clear(); + identifier.release(); + entries.clear(); + position = -1; + } + } + + private static final class StoredEntry { + + final BinaryRow row; + final ManifestEntryRunMergeEntry.Key key; + + StoredEntry(BinaryRow row, ManifestEntryRunMergeEntry.Key key) { + this.row = row; + this.key = key; + } + } + + /** Fixed-size tournament tree which selects a cursor with one comparison per tree level. */ + private static final class SelectionTree { + + final List cursors; + final int leafBase; + final int[] winners; + + SelectionTree(List cursors) { + this.cursors = cursors; + int base = 1; + while (base < cursors.size()) { + base <<= 1; + } + this.leafBase = base; + this.winners = new int[leafBase << 1]; + Arrays.fill(winners, -1); + for (int cursor = 0; cursor < cursors.size(); cursor++) { + if (cursors.get(cursor).hasCurrent()) { + winners[leafBase + cursor] = cursor; + } + } + for (int node = leafBase - 1; node > 0; node--) { + winners[node] = select(winners[node << 1], winners[(node << 1) + 1]); + } + } + + int winner() { + return winners[1]; + } + + Cursor cursor(int index) { + return cursors.get(index); + } + + void update(int cursor, boolean hasCurrent) { + int node = leafBase + cursor; + winners[node] = hasCurrent ? cursor : -1; + while ((node >>= 1) > 0) { + winners[node] = select(winners[node << 1], winners[(node << 1) + 1]); + } + } + + int select(int left, int right) { + if (left < 0) { + return right; + } + if (right < 0) { + return left; + } + int comparison = + ManifestEntryRunMerge.compareMergeKeys( + cursors.get(left).key(), cursors.get(right).key()); + return comparison < 0 || (comparison == 0 && left < right) ? left : right; + } + + boolean blockPrecedesOthers(int cursor, ManifestEntryRunMergeEntry.Key blockLastKey) { + for (int other = 0; other < cursors.size(); other++) { + if (other == cursor || !cursors.get(other).hasCurrent()) { + continue; + } + int comparison = + ManifestEntryRunMerge.compareMergeKeys( + blockLastKey, cursors.get(other).key()); + if (comparison > 0 || (comparison == 0 && cursor > other)) { + return false; + } + } + return true; + } + } +} diff --git a/paimon-core/src/main/java/org/apache/paimon/operation/ManifestFileSorter.java b/paimon-core/src/main/java/org/apache/paimon/operation/ManifestFileSorter.java index 75eb9e7e6312..eda776751df8 100644 --- a/paimon-core/src/main/java/org/apache/paimon/operation/ManifestFileSorter.java +++ b/paimon-core/src/main/java/org/apache/paimon/operation/ManifestFileSorter.java @@ -27,6 +27,7 @@ import org.apache.paimon.data.InternalRow; import org.apache.paimon.disk.IOManager; import org.apache.paimon.manifest.CompactFileIdentifierSet; +import org.apache.paimon.manifest.DeletedRowIdSet; import org.apache.paimon.manifest.ManifestEntry; import org.apache.paimon.manifest.ManifestFile; import org.apache.paimon.manifest.ManifestFileMeta; @@ -70,9 +71,12 @@ public class ManifestFileSorter { /** Context object that carries shared state across compaction methods. */ static class CompactionContext { final boolean fullCompaction; + final boolean runMergeOptimizeEnabled; final ManifestSortKey sortKey; + final RowType partitionType; final ManifestEntryExternalSort.ExternalSortConfig externalSortConfig; final CompactFileIdentifierSet deleteEntries; + final DeletedRowIdSet deletedRowIds; /** * Manifest files that need unsorted compaction. * @@ -81,31 +85,37 @@ static class CompactionContext { *

Value: true if fullCompaction is true and the file overlaps with delete partitions. It * means the file needs to eliminate delete entries file */ - final Map compactWithoutSort; + final Map defaultCompactFiles; final List levelRuns; final List pickedRuns; CompactionContext( boolean fullCompaction, + boolean runMergeOptimizeEnabled, ManifestSortKey sortKey, + RowType partitionType, ManifestEntryExternalSort.ExternalSortConfig externalSortConfig, CompactFileIdentifierSet deleteEntries, - Map compactWithoutSort, + DeletedRowIdSet deletedRowIds, + Map defaultCompactFiles, List levelRuns, List pickedRuns) { this.fullCompaction = fullCompaction; + this.runMergeOptimizeEnabled = runMergeOptimizeEnabled; this.sortKey = sortKey; + this.partitionType = partitionType; this.externalSortConfig = externalSortConfig; this.deleteEntries = deleteEntries; - this.compactWithoutSort = compactWithoutSort; + this.deletedRowIds = deletedRowIds; + this.defaultCompactFiles = defaultCompactFiles; this.levelRuns = levelRuns; this.pickedRuns = pickedRuns; } /** Check whether the given manifest file is marked for unsorted compaction. */ - boolean isMarkedForUnsortedCompaction(ManifestFileMeta file) { - return compactWithoutSort.containsKey(file); + boolean isMarkedForDefaultCompaction(ManifestFileMeta file) { + return defaultCompactFiles.containsKey(file); } } @@ -113,6 +123,7 @@ boolean isMarkedForUnsortedCompaction(ManifestFileMeta file) { static class ClassifyResult { final List lsmFiles; final CompactFileIdentifierSet deleteEntries; + final DeletedRowIdSet deletedRowIds; /** * Manifest files that need unsorted compaction. * @@ -126,9 +137,11 @@ static class ClassifyResult { ClassifyResult( List lsmFiles, CompactFileIdentifierSet deleteEntries, + DeletedRowIdSet deletedRowIds, Map compactWithoutSort) { this.lsmFiles = lsmFiles; this.deleteEntries = deleteEntries; + this.deletedRowIds = deletedRowIds; this.compactWithoutSort = compactWithoutSort; } } @@ -136,10 +149,15 @@ static class ClassifyResult { /** Binary identifiers and partition values collected from DELETE entries. */ private static class DeletedEntryInfo { final CompactFileIdentifierSet identifiers; + final DeletedRowIdSet rowIds; final Set partitions; - private DeletedEntryInfo(CompactFileIdentifierSet identifiers, Set partitions) { + private DeletedEntryInfo( + CompactFileIdentifierSet identifiers, + DeletedRowIdSet rowIds, + Set partitions) { this.identifiers = identifiers; + this.rowIds = rowIds; this.partitions = partitions; } } @@ -160,6 +178,7 @@ static List trySortCompaction( @Nullable IOManager ioManager) throws Exception { String sortPartitionField = options.manifestSortPartitionField(); + boolean runMergeOptimizeEnabled = options.manifestMergeOptimizeEnabled(); long suggestedMetaSize = options.manifestTargetSize().getBytes(); int suggestedMinMetaCount = options.manifestMergeMinCount(); long fullCompactionThreshold = options.manifestFullCompactionThresholdSize().getBytes(); @@ -178,6 +197,7 @@ static List trySortCompaction( partitionType, sortPartitionField, options.dataEvolutionEnabled(), + runMergeOptimizeEnabled, suggestedMetaSize, suggestedMinMetaCount, fullCompactionThreshold, @@ -196,6 +216,7 @@ static List trySortCompaction( partitionType, sortPartitionField, options.dataEvolutionEnabled(), + runMergeOptimizeEnabled, suggestedMetaSize, suggestedMinMetaCount, maxRewriteSize, @@ -218,6 +239,7 @@ private static Optional> tryFullCompaction( RowType partitionType, String sortPartitionField, boolean dataEvolutionEnabled, + boolean runMergeOptimizeEnabled, long suggestedMetaSize, int suggestedMinMetaCount, long fullCompactionThreshold, @@ -240,6 +262,7 @@ private static Optional> tryFullCompaction( partitionType, sortPartitionField, dataEvolutionEnabled, + runMergeOptimizeEnabled, suggestedMetaSize, maxSizeAmplificationPercent, sortedRunSizeRatio, @@ -248,19 +271,19 @@ private static Optional> tryFullCompaction( List levelRuns = ctx.levelRuns; List pickedRuns = ctx.pickedRuns; - if (pickedRuns.isEmpty() && ctx.compactWithoutSort.isEmpty()) { + if (pickedRuns.isEmpty() && ctx.defaultCompactFiles.isEmpty()) { LOG.debug( - "Manifest sort full compact skipped: no runs picked and no compactWithoutSort files."); + "Manifest sort full compact skipped: no runs picked and no defaultCompactFiles."); return Optional.empty(); } LOG.info( "Manifest sort full compact: input={} files, lsm={} runs, picked={} runs, " - + "compactWithoutSort={} files.", + + "defaultCompactFiles={}.", input.size(), levelRuns.size(), pickedRuns.size(), - ctx.compactWithoutSort.size()); + ctx.defaultCompactFiles.size()); // Step 3: Collect reused files (not picked) and picked files Set pickedSet = new HashSet<>(pickedRuns); @@ -274,7 +297,7 @@ private static Optional> tryFullCompaction( for (ManifestAdjacentSortedRun run : pickedRuns) { pickedFiles.addAll(run.files()); } - pickedFiles.addAll(ctx.compactWithoutSort.keySet()); + pickedFiles.addAll(ctx.defaultCompactFiles.keySet()); // Step 4: Split into sections and merge small adjacent sections List

sections = splitIntoSections(pickedFiles, ctx); @@ -318,6 +341,7 @@ private static List tryMinorCompaction( RowType partitionType, String sortPartitionField, boolean dataEvolutionEnabled, + boolean runMergeOptimizeEnabled, long suggestedMetaSize, int suggestedMinMetaCount, long maxRewriteSize, @@ -335,6 +359,7 @@ private static List tryMinorCompaction( partitionType, sortPartitionField, dataEvolutionEnabled, + runMergeOptimizeEnabled, suggestedMetaSize, maxSizeAmplificationPercent, sortedRunSizeRatio, @@ -343,19 +368,19 @@ private static List tryMinorCompaction( List levelRuns = ctx.levelRuns; List pickedRuns = ctx.pickedRuns; - if (pickedRuns.isEmpty() && ctx.compactWithoutSort.isEmpty()) { + if (pickedRuns.isEmpty() && ctx.defaultCompactFiles.isEmpty()) { LOG.debug( - "Manifest sort minor compact skipped: no runs picked and no compactWithoutSort files."); + "Manifest sort minor compact skipped: no runs picked and no defaultCompactFiles."); return input; } LOG.info( "Manifest sort minor compact: input={} files, lsm={} runs, picked={} runs, " - + "compactWithoutSort={} files.", + + "defaultCompactFiles={}.", input.size(), levelRuns.size(), pickedRuns.size(), - ctx.compactWithoutSort.size()); + ctx.defaultCompactFiles.size()); // Step 2: Build fileName -> index mapping and initialize 2D result Map fileNameToIndex = new HashMap<>(); @@ -382,7 +407,7 @@ private static List tryMinorCompaction( for (ManifestAdjacentSortedRun run : pickedRuns) { pickedFiles.addAll(run.files()); } - pickedFiles.addAll(ctx.compactWithoutSort.keySet()); + pickedFiles.addAll(ctx.defaultCompactFiles.keySet()); // Step 4: Compute index range int minIdx = Integer.MAX_VALUE; @@ -444,26 +469,29 @@ private static CompactionContext prepareCompaction( RowType partitionType, String sortPartitionField, boolean dataEvolutionEnabled, + boolean runMergeOptimizeEnabled, long suggestedMetaSize, int maxSizeAmplificationPercent, int sortedRunSizeRatio, ManifestEntryExternalSort.ExternalSortConfig externalSortConfig, @Nullable Integer manifestReadParallelism) { + boolean rowIdSort = dataEvolutionEnabled && ManifestFileMeta.allContainsRowId(input); + boolean useRunMergeOptimize = rowIdSort && runMergeOptimizeEnabled; // Step 1: Resolve sort key. Data evolution tables prefer RowID ranges when available. - ManifestSortKey sortKey = - createSortKey(dataEvolutionEnabled, input, sortPartitionField, partitionType); + ManifestSortKey sortKey = createSortKey(rowIdSort, sortPartitionField, partitionType); // Step 2: Classify manifests into LSM files and collect delete entries. - ClassifyResult classifyResult = + ClassifyResult classification = classifyManifests( input, fullCompaction, manifestFile, partitionType, suggestedMetaSize, + useRunMergeOptimize, manifestReadParallelism); - List lsmFiles = classifyResult.lsmFiles; + List lsmFiles = classification.lsmFiles; // Step 3: Build level-sorted runs from LSM files based on partition order. List levelRuns = @@ -476,10 +504,13 @@ private static CompactionContext prepareCompaction( return new CompactionContext( fullCompaction, + useRunMergeOptimize, sortKey, + partitionType, externalSortConfig, - classifyResult.deleteEntries, - classifyResult.compactWithoutSort, + classification.deleteEntries, + classification.deletedRowIds, + classification.compactWithoutSort, levelRuns, pickedRuns); } @@ -499,12 +530,12 @@ static boolean reachesFullCompactionThreshold( * Classify manifest files into default-compaction group and LSM group. * *

Full compaction: small files and files overlapping delete partitions go into - * compactWithoutSort; the rest are returned as lsmFiles. + * defaultCompactFiles; the rest are returned as lsmFiles. * - *

Non-full compaction: small files go to compactWithoutSort for minor-style merge; the rest + *

Non-full compaction: small files go to defaultCompactFiles for minor-style merge; the rest * are returned as lsmFiles. * - * @return ClassifyResult containing lsmFiles, deleteEntries, and compactWithoutSort + * @return classification containing lsmFiles, deleteEntries, and defaultCompactFiles */ static ClassifyResult classifyManifests( List input, @@ -513,16 +544,37 @@ static ClassifyResult classifyManifests( RowType partitionType, long suggestedMetaSize, @Nullable Integer manifestReadParallelism) { + return classifyManifests( + input, + fullCompaction, + manifestFile, + partitionType, + suggestedMetaSize, + false, + manifestReadParallelism); + } + + private static ClassifyResult classifyManifests( + List input, + boolean fullCompaction, + ManifestFile manifestFile, + RowType partitionType, + long suggestedMetaSize, + boolean runMergeOptimizeEnabled, + @Nullable Integer manifestReadParallelism) { // Initialize classification containers and read delete entries - Map compactWithoutSort = new LinkedHashMap<>(); + Map defaultCompactFiles = new LinkedHashMap<>(); List lsmFiles = new LinkedList<>(input); CompactFileIdentifierSet classifiedDeleteEntries = new CompactFileIdentifierSet(); + DeletedRowIdSet deletedRowIds = new DeletedRowIdSet(); Set deletePartitions = Collections.emptySet(); PartitionPredicate predicate = null; if (fullCompaction) { DeletedEntryInfo deletedEntries = - readDeletedEntries(manifestFile, input, manifestReadParallelism); + readDeletedEntries( + manifestFile, input, runMergeOptimizeEnabled, manifestReadParallelism); classifiedDeleteEntries = deletedEntries.identifiers; + deletedRowIds = deletedEntries.rowIds; deletePartitions = deletedEntries.partitions; // Build partition predicate from delete entries for overlap detection. @@ -551,18 +603,21 @@ static ClassifyResult classifyManifests( file.partitionStats().nullCounts()); if (small || inDeleteRange) { iterator.remove(); - compactWithoutSort.put(file, inDeleteRange); + defaultCompactFiles.put(file, inDeleteRange); } } - return new ClassifyResult(lsmFiles, classifiedDeleteEntries, compactWithoutSort); + return new ClassifyResult( + lsmFiles, classifiedDeleteEntries, deletedRowIds, defaultCompactFiles); } private static DeletedEntryInfo readDeletedEntries( ManifestFile manifestFile, List manifestFiles, + boolean runMergeOptimizeEnabled, @Nullable Integer manifestReadParallelism) { CompactFileIdentifierSet identifiers = new CompactFileIdentifierSet(); + DeletedRowIdSet rowIds = new DeletedRowIdSet(); Set partitions = new HashSet<>(); List filesWithDeletes = new ArrayList<>(); for (ManifestFileMeta meta : manifestFiles) { @@ -574,12 +629,26 @@ private static DeletedEntryInfo readDeletedEntries( if (filesWithDeletes.size() <= 1 || (manifestReadParallelism != null && manifestReadParallelism <= 1)) { for (ManifestFileMeta meta : filesWithDeletes) { - collectDeletedEntries(meta, manifestFile, identifiers, partitions, false); + collectDeletedEntries( + meta, + manifestFile, + identifiers, + rowIds, + partitions, + runMergeOptimizeEnabled, + false); } } else { Function> reader = meta -> { - collectDeletedEntries(meta, manifestFile, identifiers, partitions, true); + collectDeletedEntries( + meta, + manifestFile, + identifiers, + rowIds, + partitions, + runMergeOptimizeEnabled, + true); return Collections.singletonList(Boolean.TRUE); }; for (Boolean ignored : @@ -587,14 +656,16 @@ private static DeletedEntryInfo readDeletedEntries( // Iteration waits for each bounded batch of parallel reads. } } - return new DeletedEntryInfo(identifiers, partitions); + return new DeletedEntryInfo(identifiers, rowIds, partitions); } private static void collectDeletedEntries( ManifestFileMeta meta, ManifestFile manifestFile, CompactFileIdentifierSet identifiers, + DeletedRowIdSet rowIds, Set partitions, + boolean runMergeOptimizeEnabled, boolean synchronize) { try (CloseableIterator entries = manifestFile.scan( @@ -608,10 +679,16 @@ private static void collectDeletedEntries( if (synchronize) { synchronized (identifiers) { identifiers.add(entry); + if (runMergeOptimizeEnabled) { + rowIds.add(entry.file().nonNullFirstRowId()); + } partitions.add(partition); } } else { identifiers.add(entry); + if (runMergeOptimizeEnabled) { + rowIds.add(entry.file().nonNullFirstRowId()); + } partitions.add(partition); } } @@ -696,7 +773,7 @@ static List buildLevelSortedRuns( /** * 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 hasUnsortedCompactMeta. + * section. Each section is built with pre-computed totalSize and hasDefaultCompactFile. */ static List

splitIntoSections( List pickedFiles, CompactionContext ctx) { @@ -717,7 +794,7 @@ static List
splitIntoSections( currentSectionFiles.add(first); currentSectionTotalSize += first.fileSize(); - boolean currentSectionHasUnsortedCompactMeta = ctx.isMarkedForUnsortedCompaction(first); + boolean currentSectionHasDefaultCompactFile = ctx.isMarkedForDefaultCompaction(first); ManifestFileMeta sectionMaxFile = first; for (int i = 1; i < pickedFiles.size(); i++) { @@ -729,20 +806,20 @@ static List
splitIntoSections( new Section( currentSectionFiles, currentSectionTotalSize, - currentSectionHasUnsortedCompactMeta)); + currentSectionHasDefaultCompactFile)); // start a new section currentSectionFiles = new ArrayList<>(); currentSectionTotalSize = 0; currentSectionFiles.add(file); currentSectionTotalSize += file.fileSize(); - currentSectionHasUnsortedCompactMeta = ctx.isMarkedForUnsortedCompaction(file); + currentSectionHasDefaultCompactFile = ctx.isMarkedForDefaultCompaction(file); sectionMaxFile = file; } else { currentSectionFiles.add(file); currentSectionTotalSize += file.fileSize(); - if (!currentSectionHasUnsortedCompactMeta - && ctx.isMarkedForUnsortedCompaction(file)) { - currentSectionHasUnsortedCompactMeta = true; + if (!currentSectionHasDefaultCompactFile + && ctx.isMarkedForDefaultCompaction(file)) { + currentSectionHasDefaultCompactFile = true; } if (sortKey.compareMax(file, sectionMaxFile) > 0) { sectionMaxFile = file; @@ -753,7 +830,7 @@ static List
splitIntoSections( new Section( currentSectionFiles, currentSectionTotalSize, - currentSectionHasUnsortedCompactMeta)); + currentSectionHasDefaultCompactFile)); return sections; } @@ -796,7 +873,7 @@ private static List
mergeSmallAdjacentSections( *
  • First overflow: The current section is split. The rewritable part is sorted and * rewritten. The remaining part is appended back to the sections queue for later * processing. - *
  • Subsequent overflows: If the section has files in compactWithoutSort (needs unsorted + *
  • Subsequent overflows: If the section has files in defaultCompactFiles (needs default * compaction), unsortedCompactSection is called to process it in smaller chunks. * Otherwise, the section is skipped. * @@ -907,9 +984,9 @@ private static Section splitSectionAndRewriteHead( List tailFiles = new ArrayList<>(); long headSize = 0; long tailSize = 0; - // Whether tail section has files in compactWithoutSort, if true, the section need to + // Whether the tail section has files in defaultCompactFiles. If so, the section needs to // be rewritten. - boolean tailHasUnsortedCompactMeta = false; + boolean tailHasDefaultCompactFile = false; for (ManifestFileMeta file : section.files) { // Rewrite budget is enforced at manifest-file granularity. Include the first file that @@ -921,8 +998,8 @@ private static Section splitSectionAndRewriteHead( } else { tailFiles.add(file); tailSize += file.fileSize(); - if (ctx.isMarkedForUnsortedCompaction(file)) { - tailHasUnsortedCompactMeta = true; + if (ctx.isMarkedForDefaultCompaction(file)) { + tailHasDefaultCompactFile = true; } } } @@ -932,7 +1009,7 @@ private static Section splitSectionAndRewriteHead( if (tailFiles.isEmpty()) { return null; } - return new Section(tailFiles, tailSize, tailHasUnsortedCompactMeta); + return new Section(tailFiles, tailSize, tailHasDefaultCompactFile); } /** @@ -950,7 +1027,7 @@ private static void rewriteSectionBeyondBudget( int suggestedMinMetaCount, @Nullable Integer manifestReadParallelism) throws Exception { - if (section.hasUnsortedCompactMeta) { + if (section.hasDefaultCompactFile) { unsortedCompactSection( section.files, output, @@ -970,8 +1047,8 @@ private static void rewriteSectionBeyondBudget( * *

    Semantics difference from old minor merge: In the old ManifestFileMerger path, the * trailing candidates are kept unchanged when their count is below manifest.merge-min-count. In - * this sort path, unsortedCompactSection is triggered when compactWithoutSort is non-empty, - * regardless of the manifest count. This is because files in compactWithoutSort either: + * this sort path, unsortedCompactSection is triggered when defaultCompactFiles is non-empty, + * regardless of the manifest count. This is because files in defaultCompactFiles either: * *

      *
    • Are small files needing consolidation @@ -1041,7 +1118,7 @@ private static void rewriteSection( @Nullable Integer manifestReadParallelism) throws Exception { // Skip rewrite for single file not in delete-range. - if (section.size() == 1 && !ctx.compactWithoutSort.getOrDefault(section.get(0), false)) { + if (section.size() == 1 && !ctx.defaultCompactFiles.getOrDefault(section.get(0), false)) { output.addUnchanged(section.get(0)); return; } @@ -1067,24 +1144,40 @@ private static void rewriteFull( ManifestFile manifestFile, @Nullable Integer manifestReadParallelism) throws Exception { - List sorted = - ManifestEntryExternalSort.sortAndWriteFullEntries( - section, - ctx.sortKey, - ctx.externalSortConfig, - manifestFile, - sortNewFiles, - ctx.deleteEntries, - manifestReadParallelism); + List sorted = null; + if (ctx.runMergeOptimizeEnabled) { + sorted = + ManifestEntryRunMerge.sortAndWriteFullEntries( + section, + (RowIdEntrySortKey) ctx.sortKey, + ctx.partitionType, + manifestFile, + sortNewFiles, + ctx.deleteEntries, + ctx.deletedRowIds, + ctx.externalSortConfig.maxNumFileHandles, + manifestReadParallelism); + } + if (sorted == null) { + sorted = + ManifestEntryExternalSort.sortAndWriteFullEntries( + section, + ctx.sortKey, + ctx.externalSortConfig, + manifestFile, + sortNewFiles, + ctx.deleteEntries, + manifestReadParallelism); + } if (!sorted.isEmpty()) { output.addSortedFiles(sorted); } } /** - * Minor compaction path: collect DELETE entries in memory while external-sorting all entries, - * then write surviving ADD entries from the sorted stream and remaining DELETE entries from - * memory. + * Minor compaction path: collect DELETE identities, merge the existing sorted runs, and write + * surviving ADD entries and unmatched DELETE entries separately. Falls back to external sort + * when the input is not suitable for run merge. */ private static void rewriteMinor( List section, @@ -1094,14 +1187,28 @@ private static void rewriteMinor( ManifestFile manifestFile, @Nullable Integer manifestReadParallelism) throws Exception { - Pair, List> sorted = - ManifestEntryExternalSort.sortAndWriteMinorEntries( - section, - ctx.sortKey, - ctx.externalSortConfig, - manifestFile, - sortNewFiles, - manifestReadParallelism); + Pair, List> sorted = null; + if (ctx.runMergeOptimizeEnabled) { + sorted = + ManifestEntryRunMerge.sortAndWriteMinorEntries( + section, + (RowIdEntrySortKey) ctx.sortKey, + ctx.partitionType, + manifestFile, + sortNewFiles, + ctx.externalSortConfig.maxNumFileHandles, + manifestReadParallelism); + } + if (sorted == null) { + sorted = + ManifestEntryExternalSort.sortAndWriteMinorEntries( + section, + ctx.sortKey, + ctx.externalSortConfig, + manifestFile, + sortNewFiles, + manifestReadParallelism); + } if (!sorted.getLeft().isEmpty()) { output.addSortedFiles(sorted.getLeft()); @@ -1126,7 +1233,15 @@ static ManifestSortKey createSortKey( List input, String sortPartitionField, RowType partitionType) { - if (dataEvolutionEnabled && ManifestFileMeta.allContainsRowId(input)) { + return createSortKey( + dataEvolutionEnabled && ManifestFileMeta.allContainsRowId(input), + sortPartitionField, + partitionType); + } + + private static ManifestSortKey createSortKey( + boolean rowIdSort, String sortPartitionField, RowType partitionType) { + if (rowIdSort) { // RowID sorting uses the configured partition field as the primary key when specified, // otherwise it uses the full partition row to preserve partition locality. It then // orders files by RowID. @@ -1205,6 +1320,11 @@ void replaceExternalSortRow( InternalRow binaryManifestRow(BinaryRow row); } + interface RowIdEntrySortKey extends ManifestSortKey { + + int comparePartitions(BinaryRow left, BinaryRow right); + } + private static class PartitionSortKey implements ManifestSortKey { private final RecordComparator fieldComparator; @@ -1276,7 +1396,7 @@ public InternalRow binaryManifestRow(BinaryRow row) { } } - private static class RowIdSortKey implements ManifestSortKey { + private static class RowIdSortKey implements RowIdEntrySortKey { @Nullable private final RecordComparator partitionComparator; private final InternalRow.FieldGetter[] partitionFieldGetters; @@ -1291,21 +1411,8 @@ private RowIdSortKey( this.partitionComparator = partitionComparator; this.partitionFieldGetters = createPartitionFieldGetters(partitionType, partitionSortFields); - - List fieldTypes = new ArrayList<>(); - for (int partitionSortField : partitionSortFields) { - fieldTypes.add(partitionType.getTypeAt(partitionSortField)); - } - // ADD must precede DELETE for the same partition. Minor compaction streams the sorted - // rows once and uses this ordering to eliminate a matching pair without retaining all - // ADD identifiers. - fieldTypes.add(DataTypes.TINYINT()); - fieldTypes.add(DataTypes.BIGINT()); - fieldTypes.add(DataTypes.BIGINT()); - fieldTypes.add(DataTypes.BIGINT()); - fieldTypes.add(DataTypes.STRING()); - fieldTypes.add(ManifestEntry.MANIFEST_ROW_TYPE); - this.externalSortRowType = DataTypes.ROW(fieldTypes.toArray(new DataType[0])); + this.externalSortRowType = + createRowIdExternalSortRowType(partitionType, partitionSortFields); this.sortFieldNum = externalSortRowType.getFieldCount() - 1; this.externalSortKeyFields = createSequentialFields(sortFieldNum); } @@ -1339,7 +1446,7 @@ public boolean isAfterMax(ManifestFileMeta file, ManifestFileMeta maxFile) { return c > 0; } } - return Long.compare(nonNullMinRowId(file), nonNullMaxRowId(maxFile)) > 0; + return nonNullMinRowId(file) > nonNullMaxRowId(maxFile); } @Override @@ -1376,6 +1483,11 @@ public InternalRow binaryManifestRow(BinaryRow row) { return row.getRow(sortFieldNum, ManifestEntry.MANIFEST_ROW_TYPE.getFieldCount()); } + @Override + public int comparePartitions(BinaryRow left, BinaryRow right) { + return partitionComparator == null ? 0 : partitionComparator.compare(left, right); + } + private int comparePartitionMin(ManifestFileMeta a, ManifestFileMeta b) { if (partitionComparator == null) { return 0; @@ -1415,6 +1527,26 @@ private static long rowIdRangeEnd(ManifestEntry entry) { } } + private static RowType createRowIdExternalSortRowType( + RowType partitionType, int[] partitionSortFields) { + List fieldTypes = new ArrayList<>(partitionSortFields.length + 6); + for (int partitionSortField : partitionSortFields) { + fieldTypes.add(partitionType.getTypeAt(partitionSortField)); + } + // ADD must precede DELETE for the same partition. Minor compaction streams the sorted rows + // once and uses this ordering to eliminate a matching pair without retaining all ADD + // identifiers. + Collections.addAll( + fieldTypes, + DataTypes.TINYINT(), + DataTypes.BIGINT(), + DataTypes.BIGINT(), + DataTypes.BIGINT(), + DataTypes.STRING(), + ManifestEntry.MANIFEST_ROW_TYPE); + return DataTypes.ROW(fieldTypes.toArray(new DataType[0])); + } + private static int[] createSequentialFields(int fieldCount) { int[] fields = new int[fieldCount]; for (int i = 0; i < fieldCount; i++) { @@ -1533,12 +1665,12 @@ public void addDeleteFiles(List files) { static class Section { final List files; final long totalSize; - final boolean hasUnsortedCompactMeta; + final boolean hasDefaultCompactFile; - Section(List files, long totalSize, boolean hasUnsortedCompactMeta) { + Section(List files, long totalSize, boolean hasDefaultCompactFile) { this.files = files; this.totalSize = totalSize; - this.hasUnsortedCompactMeta = hasUnsortedCompactMeta; + this.hasDefaultCompactFile = hasDefaultCompactFile; } /** Create a merged section from two sections. */ @@ -1548,7 +1680,7 @@ static Section merge(Section a, Section b) { return new Section( merged, a.totalSize + b.totalSize, - a.hasUnsortedCompactMeta || b.hasUnsortedCompactMeta); + a.hasDefaultCompactFile || b.hasDefaultCompactFile); } } } diff --git a/paimon-core/src/test/java/org/apache/paimon/manifest/ManifestFileMetaTest.java b/paimon-core/src/test/java/org/apache/paimon/manifest/ManifestFileMetaTest.java index a81e2de76077..fe345048c315 100644 --- a/paimon-core/src/test/java/org/apache/paimon/manifest/ManifestFileMetaTest.java +++ b/paimon-core/src/test/java/org/apache/paimon/manifest/ManifestFileMetaTest.java @@ -75,6 +75,7 @@ import java.util.concurrent.atomic.AtomicInteger; import java.util.stream.Collectors; import java.util.stream.IntStream; +import java.util.stream.LongStream; import static org.assertj.core.api.Assertions.assertThat; import static org.junit.jupiter.api.Assertions.assertThrows; @@ -1173,6 +1174,40 @@ private void beforeFirstRead() throws IOException { } } + private static class OpenTrackingFileIO extends LocalFileIO { + + private final AtomicInteger openInputStreams = new AtomicInteger(); + private final AtomicInteger maxOpenInputStreams = new AtomicInteger(); + + @Override + public SeekableInputStream newInputStream(Path path) throws IOException { + SeekableInputStream input = super.newInputStream(path); + int open = openInputStreams.incrementAndGet(); + maxOpenInputStreams.accumulateAndGet(open, Math::max); + return new SeekableInputStreamWrapper(input) { + + private boolean closed; + + @Override + public void close() throws IOException { + if (closed) { + return; + } + try { + super.close(); + } finally { + closed = true; + openInputStreams.decrementAndGet(); + } + } + }; + } + + private int maxOpenInputStreams() { + return maxOpenInputStreams.get(); + } + } + // ==================== Manifest Sort Tests ==================== /** @@ -1609,6 +1644,443 @@ public void testDataEvolutionManifestSortByPartitionAndRowId() { } } + @Test + public void testDataEvolutionManifestRunMergeUsesDefaultParallelism() throws Exception { + assumeTrue(Runtime.getRuntime().availableProcessors() > 1); + CountingReadFileIO fileIO = new CountingReadFileIO(); + manifestFile = createManifestFile(tempDir.toString(), fileIO); + ManifestFileMeta first = makeManifest(makeRowIdEntry(true, "row-0", 0, 0, 5)); + ManifestFileMeta second = makeManifest(makeRowIdEntry(true, "row-10", 0, 10, 5)); + + fileIO.blockManifestReads( + new HashSet<>(Arrays.asList(first.fileName(), second.fileName()))); + List merged; + try { + Options testOptions = new Options(); + testOptions.set("manifest-sort.enabled", "true"); + testOptions.set("data-evolution.enabled", "true"); + testOptions.set("manifest.full-compaction-threshold-size", "1B"); + merged = + ManifestFileMerger.merge( + Arrays.asList(first, second), + manifestFile, + getPartitionType(), + CoreOptions.fromMap(testOptions.toMap())); + } finally { + fileIO.stopBlockingManifestReads(); + } + + assertThat(fileIO.maxConcurrentManifestReads()).isGreaterThanOrEqualTo(2); + assertThat(readEntries(merged).stream().map(entry -> entry.file().fileName())) + .containsExactly("row-0", "row-10"); + } + + @Test + public void testDataEvolutionManifestRunMergeRespectsFileHandleLimit() { + OpenTrackingFileIO fileIO = new OpenTrackingFileIO(); + manifestFile = createManifestFile(tempDir.toString(), fileIO); + List input = + Arrays.asList( + makeManifest(makeRowIdEntry(true, "row-0", 0, 0, 5)), + makeManifest(makeRowIdEntry(true, "row-10", 0, 10, 5)), + makeManifest(makeRowIdEntry(true, "row-20", 0, 20, 5))); + + Options testOptions = new Options(); + testOptions.set("manifest-sort.enabled", "true"); + testOptions.set("data-evolution.enabled", "true"); + testOptions.set("manifest.full-compaction-threshold-size", "1B"); + testOptions.set("scan.manifest.parallelism", "1"); + testOptions.set("local-sort.max-num-file-handles", "2"); + List merged = + ManifestFileMerger.merge( + input, + manifestFile, + getPartitionType(), + CoreOptions.fromMap(testOptions.toMap())); + + assertThat(fileIO.maxOpenInputStreams()).isLessThanOrEqualTo(2); + assertThat(readEntries(merged).stream().map(entry -> entry.file().fileName())) + .containsExactly("row-0", "row-10", "row-20"); + } + + @Test + public void testDisablingRunMergeOptimizePreservesDataEvolutionRowIdSort() { + assertThat(CoreOptions.fromMap(Collections.emptyMap()).manifestMergeOptimizeEnabled()) + .isTrue(); + + List input = + Arrays.asList( + makeManifest( + makeRowIdEntry(true, "row-30", 0, 30, 5), + makeRowIdEntry(true, "row-10", 0, 10, 5)), + makeManifest( + makeRowIdEntry(true, "row-20", 0, 20, 5), + makeRowIdEntry(true, "row-0", 0, 0, 5))); + + Options testOptions = new Options(); + testOptions.set("manifest-sort.enabled", "true"); + testOptions.set("manifest.merge-optimize.enabled", "false"); + testOptions.set("data-evolution.enabled", "true"); + testOptions.set("manifest.full-compaction-threshold-size", "1B"); + CoreOptions coreOptions = CoreOptions.fromMap(testOptions.toMap()); + + assertThat(coreOptions.manifestMergeOptimizeEnabled()).isFalse(); + + List merged = + ManifestFileMerger.merge(input, manifestFile, getPartitionType(), coreOptions); + + assertEquivalentEntries(input, merged); + assertThat(readEntries(merged).stream().map(entry -> entry.file().fileName())) + .containsExactly("row-0", "row-10", "row-20", "row-30"); + } + + @Test + public void testDataEvolutionManifestRunMergeSecondaryKeys() { + List firstManifest = new ArrayList<>(); + firstManifest.add(makeRowIdEntry(true, "range-short", 0, 100, 5, 1)); + firstManifest.add(makeRowIdEntry(true, "sequence-newer", 0, 100, 10, 9)); + for (int i = 19; i >= 10; i--) { + firstManifest.add(makeRowIdEntry(true, String.format("tie-%02d", i), 0, 100, 10, 5)); + } + + List secondManifest = new ArrayList<>(); + for (int i = 9; i >= 0; i--) { + secondManifest.add(makeRowIdEntry(true, String.format("tie-%02d", i), 0, 100, 10, 5)); + } + + List input = new ArrayList<>(); + input.add(makeManifest(firstManifest.toArray(new ManifestEntry[0]))); + input.add(makeManifest(secondManifest.toArray(new ManifestEntry[0]))); + + Options testOptions = new Options(); + testOptions.set("manifest-sort.enabled", "true"); + testOptions.set("data-evolution.enabled", "true"); + testOptions.set("manifest.full-compaction-threshold-size", "1B"); + testOptions.set("scan.manifest.parallelism", "2"); + + List merged = + ManifestFileMerger.merge( + input, + manifestFile, + getPartitionType(), + CoreOptions.fromMap(testOptions.toMap())); + + List expected = new ArrayList<>(); + expected.add("range-short"); + expected.add("sequence-newer"); + for (int i = 0; i < 20; i++) { + expected.add(String.format("tie-%02d", i)); + } + assertThat(readEntries(merged).stream().map(e -> e.file().fileName())) + .containsExactlyElementsOf(expected); + } + + @Test + public void testDataEvolutionManifestRunMergeUsesExactDeleteIdentifier() { + List input = + Arrays.asList( + makeManifest( + makeRowIdEntry(true, "deleted", 0, 100, 5), + makeRowIdEntry(true, "same-row-id-survivor", 0, 100, 5)), + makeManifest(makeRowIdEntry(false, "deleted", 0, 100, 5))); + + Options testOptions = new Options(); + testOptions.set("manifest-sort.enabled", "true"); + testOptions.set("data-evolution.enabled", "true"); + testOptions.set("manifest.full-compaction-threshold-size", "1B"); + testOptions.set("scan.manifest.parallelism", "2"); + + List merged = + ManifestFileMerger.merge( + input, + manifestFile, + getPartitionType(), + CoreOptions.fromMap(testOptions.toMap())); + + assertThat(readEntries(merged).stream().map(entry -> entry.file().fileName())) + .containsExactly("same-row-id-survivor"); + } + + @Test + public void testDataEvolutionManifestRunMergeUsesRawDeleteIdentityFields() { + ManifestEntry deleted = + makeRowIdEntry( + true, + "same-file-name", + 0, + 100, + 5, + 0, + Arrays.asList("extra-a", "extra-b"), + new byte[] {1, 2}, + "external-a"); + ManifestEntry survivor = + makeRowIdEntry( + true, + "same-file-name", + 0, + 100, + 5, + 0, + Collections.singletonList("other-extra"), + new byte[] {3, 4}, + "external-b"); + ManifestEntry delete = + makeRowIdEntry( + false, + "same-file-name", + 0, + 100, + 5, + 0, + Arrays.asList("extra-a", "extra-b"), + new byte[] {1, 2}, + "external-a"); + List input = + Arrays.asList(makeManifest(deleted, survivor), makeManifest(delete)); + + Options testOptions = new Options(); + testOptions.set("manifest-sort.enabled", "true"); + testOptions.set("data-evolution.enabled", "true"); + testOptions.set("manifest.full-compaction-threshold-size", "1B"); + + List merged = + ManifestFileMerger.merge( + input, + manifestFile, + getPartitionType(), + CoreOptions.fromMap(testOptions.toMap())); + + assertThat(readEntries(merged)).singleElement().isEqualTo(survivor); + } + + @Test + public void testDataEvolutionManifestRunMergeManyPartitions() { + List firstManifest = new ArrayList<>(); + List secondManifest = new ArrayList<>(); + for (int partition = 39; partition >= 0; partition--) { + ManifestEntry entry = + makeRowIdEntry( + true, + String.format("partition-%02d", partition), + partition, + partition * 10L, + 5); + (partition >= 20 ? firstManifest : secondManifest).add(entry); + } + + List input = + Arrays.asList( + makeManifest(firstManifest.toArray(new ManifestEntry[0])), + makeManifest(secondManifest.toArray(new ManifestEntry[0]))); + Options testOptions = new Options(); + testOptions.set("manifest-sort.enabled", "true"); + testOptions.set("data-evolution.enabled", "true"); + testOptions.set("manifest.full-compaction-threshold-size", "1B"); + + List merged = + ManifestFileMerger.merge( + input, + manifestFile, + getPartitionType(), + CoreOptions.fromMap(testOptions.toMap())); + + assertThat(readEntries(merged).stream().map(e -> e.partition().getInt(0))) + .containsExactlyElementsOf( + IntStream.range(0, 40).boxed().collect(Collectors.toList())); + } + + @Test + public void testDataEvolutionManifestRunMergeFragmentedSmallManifests() { + List firstManifest = new ArrayList<>(); + List secondManifest = new ArrayList<>(); + for (long firstRowId = 199; firstRowId >= 0; firstRowId--) { + ManifestEntry entry = + makeRowIdEntry(true, String.format("row-%03d", firstRowId), 0, firstRowId, 1); + (firstRowId >= 100 ? firstManifest : secondManifest).add(entry); + } + + List input = + Arrays.asList( + makeManifest(firstManifest.toArray(new ManifestEntry[0])), + makeManifest(secondManifest.toArray(new ManifestEntry[0]))); + Options testOptions = new Options(); + testOptions.set("manifest-sort.enabled", "true"); + testOptions.set("data-evolution.enabled", "true"); + testOptions.set("manifest.full-compaction-threshold-size", "1B"); + + List merged = + ManifestFileMerger.merge( + input, + manifestFile, + getPartitionType(), + CoreOptions.fromMap(testOptions.toMap())); + + assertThat(readEntries(merged).stream().map(entry -> entry.file().nonNullFirstRowId())) + .containsExactlyElementsOf( + LongStream.range(0, 200).boxed().collect(Collectors.toList())); + } + + @Test + public void testDataEvolutionManifestRunMergeFallsBackForLargeFragmentedManifest() { + List firstManifest = new ArrayList<>(); + List secondManifest = new ArrayList<>(); + for (long firstRowId = 25_000; firstRowId >= 12_500; firstRowId--) { + firstManifest.add( + makeRowIdEntry(true, String.format("row-%05d", firstRowId), 0, firstRowId, 1)); + } + for (long firstRowId = 12_499; firstRowId >= 0; firstRowId--) { + secondManifest.add( + makeRowIdEntry(true, String.format("row-%05d", firstRowId), 0, firstRowId, 1)); + } + + List input = + Arrays.asList( + makeManifest(firstManifest.toArray(new ManifestEntry[0])), + makeManifest(secondManifest.toArray(new ManifestEntry[0]))); + Options testOptions = new Options(); + testOptions.set("manifest-sort.enabled", "true"); + testOptions.set("data-evolution.enabled", "true"); + testOptions.set("manifest.full-compaction-threshold-size", "1B"); + + List merged = + ManifestFileMerger.merge( + input, + manifestFile, + getPartitionType(), + CoreOptions.fromMap(testOptions.toMap())); + + assertThat(readEntries(merged).stream().map(entry -> entry.file().nonNullFirstRowId())) + .containsExactlyElementsOf( + LongStream.rangeClosed(0, 25_000).boxed().collect(Collectors.toList())); + } + + @Test + public void testDataEvolutionMinorRunMergeFallsBackForLargeFragmentedManifest() { + List fragmentedEntries = new ArrayList<>(); + for (long firstRowId = 25_000; firstRowId >= 0; firstRowId--) { + fragmentedEntries.add( + makeRowIdEntry(true, String.format("row-%05d", firstRowId), 0, firstRowId, 1)); + } + + List input = + Arrays.asList( + makeManifest(fragmentedEntries.toArray(new ManifestEntry[0])), + makeManifest( + makeRowIdEntry(false, "row-12500", 0, 12_500, 1), + makeRowIdEntry(true, "row-30000", 0, 30_000, 1))); + + List expected = + LongStream.rangeClosed(0, 25_000).boxed().collect(Collectors.toList()); + expected.remove(Long.valueOf(12_500L)); + expected.add(30_000L); + + List externalResult = readEntries(mergeMinorManifestEntries(input, false)); + List runMergeResult = readEntries(mergeMinorManifestEntries(input, true)); + assertThat(runMergeResult).containsExactlyElementsOf(externalResult); + assertThat(runMergeResult.stream().map(entry -> entry.file().nonNullFirstRowId())) + .containsExactlyElementsOf(expected); + } + + @Test + public void testDataEvolutionManifestRunMergeLimitsReadAmplification() { + CountingReadFileIO fileIO = new CountingReadFileIO(); + manifestFile = createManifestFile(tempDir.toString(), fileIO); + + int runCount = 20; + int entriesPerRun = 600; + List fragmentedEntries = new ArrayList<>(); + for (int run = runCount - 1; run >= 0; run--) { + long runStart = (long) run * entriesPerRun; + for (int entry = 0; entry < entriesPerRun; entry++) { + long firstRowId = runStart + entry; + fragmentedEntries.add( + makeRowIdEntry( + true, + String.format("fragmented-%05d", firstRowId), + 0, + firstRowId, + 1)); + } + } + + ManifestFileMeta fragmented = makeManifest(fragmentedEntries.toArray(new ManifestEntry[0])); + ManifestFileMeta overlap = + makeManifest(makeRowIdEntry(true, "overlap", 0, entriesPerRun / 2L, 1)); + fileIO.resetReadCounts(); + + Options testOptions = new Options(); + testOptions.set("manifest-sort.enabled", "true"); + testOptions.set("data-evolution.enabled", "true"); + testOptions.set("manifest.full-compaction-threshold-size", "1B"); + + List merged = + ManifestFileMerger.merge( + Arrays.asList(fragmented, overlap), + manifestFile, + getPartitionType(), + CoreOptions.fromMap(testOptions.toMap())); + + assertThat(fileIO.readCount(fragmented.fileName())).isEqualTo(2); + List expectedRowIds = + LongStream.range(0, (long) runCount * entriesPerRun) + .boxed() + .collect(Collectors.toList()); + expectedRowIds.add(entriesPerRun / 2L); + Collections.sort(expectedRowIds); + assertThat(readEntries(merged).stream().map(entry -> entry.file().nonNullFirstRowId())) + .containsExactlyElementsOf(expectedRowIds); + } + + @Test + public void testDataEvolutionManifestRunMergePreservesBlockStats() { + List spanningManifest = new ArrayList<>(); + List middleManifest = new ArrayList<>(); + for (long firstRowId = 0; firstRowId < 5_000; firstRowId++) { + spanningManifest.add( + makeRowIdEntry( + true, String.format("row-%05d", firstRowId), null, firstRowId, 1)); + } + for (long firstRowId = 10_000; firstRowId < 15_000; firstRowId++) { + spanningManifest.add( + makeRowIdEntry(true, String.format("row-%05d", firstRowId), 7, firstRowId, 1)); + } + for (long firstRowId = 5_000; firstRowId < 10_000; firstRowId++) { + middleManifest.add( + makeRowIdEntry( + true, String.format("row-%05d", firstRowId), null, firstRowId, 1)); + } + + List input = + Arrays.asList( + makeManifest(spanningManifest.toArray(new ManifestEntry[0])), + makeManifest(middleManifest.toArray(new ManifestEntry[0]))); + Options testOptions = new Options(); + testOptions.set("manifest-sort.enabled", "true"); + testOptions.set("data-evolution.enabled", "true"); + testOptions.set("manifest.full-compaction-threshold-size", "1B"); + + List merged = + ManifestFileMerger.merge( + input, + manifestFile, + getPartitionType(), + CoreOptions.fromMap(testOptions.toMap())); + + assertThat(merged).hasSize(1); + ManifestFileMeta output = merged.get(0); + assertThat(output.numAddedFiles()).isEqualTo(15_000); + assertThat(output.numDeletedFiles()).isZero(); + assertThat(output.minRowId()).isZero(); + assertThat(output.maxRowId()).isEqualTo(14_999); + assertThat(output.partitionStats().minValues().getInt(0)).isEqualTo(7); + assertThat(output.partitionStats().maxValues().getInt(0)).isEqualTo(7); + assertThat(output.partitionStats().nullCounts().getLong(0)).isEqualTo(10_000); + assertThat(readEntries(merged).stream().map(entry -> entry.file().nonNullFirstRowId())) + .containsExactlyElementsOf( + LongStream.range(0, 15_000).boxed().collect(Collectors.toList())); + } + @Test public void testDataEvolutionManifestSortUsesConfiguredPartitionFieldBeforeRowId() { RowType multiPartitionType = RowType.of(new IntType(), new IntType(), new IntType()); @@ -1740,6 +2212,99 @@ public void testDataEvolutionMinorManifestSortPreservesUnmatchedDeleteEntries() .containsExactly("ADD-new-row20", "ADD-survivor-row30", "DELETE-old-row10"); } + @Test + public void testDataEvolutionMinorRunMergeMatchesExternalSort() { + ManifestEntry deleted = + makeRowIdEntry( + true, + "same-file-name", + 0, + 100, + 5, + 0, + Arrays.asList("extra-a", "extra-b"), + new byte[] {1, 2}, + "external-a"); + ManifestEntry sameRowIdSurvivor = + makeRowIdEntry( + true, + "same-file-name", + 0, + 100, + 5, + 0, + Collections.singletonList("other-extra"), + new byte[] {3, 4}, + "external-b"); + ManifestEntry delete = + makeRowIdEntry( + false, + "same-file-name", + 0, + 100, + 5, + 0, + Arrays.asList("extra-a", "extra-b"), + new byte[] {1, 2}, + "external-a"); + ManifestEntry unmatchedDelete = makeRowIdEntry(false, "old-row-200", 0, 200, 5); + List input = + Arrays.asList( + makeManifest( + deleted, + sameRowIdSurvivor, + makeRowIdEntry(true, "survivor-row-300", 0, 300, 5)), + makeManifest(delete, unmatchedDelete, unmatchedDelete)); + + List externalResult = readEntries(mergeMinorManifestEntries(input, false)); + List runMergeResult = readEntries(mergeMinorManifestEntries(input, true)); + + assertThat(runMergeResult).containsExactlyElementsOf(externalResult); + assertThat( + runMergeResult.stream() + .map(entry -> entry.kind() + "-" + entry.file().fileName()) + .collect(Collectors.toList())) + .containsExactly( + "ADD-same-file-name", "ADD-survivor-row-300", "DELETE-old-row-200"); + assertThat(runMergeResult.get(0)).isEqualTo(sameRowIdSurvivor); + } + + @Test + public void testDataEvolutionMinorRunMergeCollectsDeletesDuringDiscovery() { + CountingReadFileIO fileIO = new CountingReadFileIO(); + manifestFile = createManifestFile(tempDir.toString(), fileIO); + ManifestFileMeta base = + makeManifest( + makeRowIdEntry(true, "deleted-row-10", 0, 10, 5), + makeRowIdEntry(true, "survivor-row-20", 0, 20, 5)); + ManifestFileMeta delta = + makeManifest( + makeRowIdEntry(true, "survivor-row-30", 0, 30, 5), + makeRowIdEntry(false, "deleted-row-10", 0, 10, 5)); + fileIO.resetReadCounts(); + + List merged = mergeMinorManifestEntries(Arrays.asList(base, delta), true); + + assertThat(fileIO.readCount(base.fileName())).isEqualTo(2); + assertThat(fileIO.readCount(delta.fileName())).isEqualTo(2); + assertThat( + readEntries(merged).stream() + .map(entry -> entry.file().fileName()) + .collect(Collectors.toList())) + .containsExactly("survivor-row-20", "survivor-row-30"); + } + + private List mergeMinorManifestEntries( + List input, boolean runMergeOptimizeEnabled) { + Options options = new Options(); + options.set("manifest-sort.enabled", "true"); + options.set("manifest.merge-optimize.enabled", Boolean.toString(runMergeOptimizeEnabled)); + options.set("data-evolution.enabled", "true"); + options.set("manifest.full-compaction-threshold-size", Long.MAX_VALUE + "B"); + return ManifestFileMerger.merge( + input, manifestFile, getPartitionType(), CoreOptions.fromMap(options.toMap())); + } + /** * Test manifest sort with a multi-field partition type. * @@ -2255,20 +2820,46 @@ private List readFileNames( /** Create a ManifestEntry with row ID metadata for data evolution manifest sort tests. */ private ManifestEntry makeRowIdEntry( - boolean isAdd, String fileName, int partition, long firstRowId, long rowCount) { + boolean isAdd, String fileName, Integer partition, long firstRowId, long rowCount) { return makeRowIdEntry(isAdd, fileName, partition, firstRowId, rowCount, 0); } private ManifestEntry makeRowIdEntry( boolean isAdd, String fileName, - int partition, + Integer partition, long firstRowId, long rowCount, long sequenceNumber) { + return makeRowIdEntry( + isAdd, + fileName, + partition, + firstRowId, + rowCount, + sequenceNumber, + Collections.emptyList(), + null, + null); + } + + private ManifestEntry makeRowIdEntry( + boolean isAdd, + String fileName, + Integer partition, + long firstRowId, + long rowCount, + long sequenceNumber, + List extraFiles, + byte[] embeddedIndex, + String externalPath) { BinaryRow binaryRow = new BinaryRow(1); BinaryRowWriter writer = new BinaryRowWriter(binaryRow); - writer.writeInt(0, partition); + if (partition == null) { + writer.setNullAt(0); + } else { + writer.writeInt(0, partition); + } writer.complete(); return ManifestEntry.create( @@ -2288,13 +2879,13 @@ private ManifestEntry makeRowIdEntry( sequenceNumber, 0, 0, - Collections.emptyList(), + extraFiles, Timestamp.fromEpochMillis(200000), 0L, - null, + embeddedIndex, FileSource.APPEND, null, - null, + externalPath, firstRowId, Collections.singletonList("f0"))); }