diff --git a/docs/generated/core_configuration.html b/docs/generated/core_configuration.html index ab8443e4cbce..1d40881dae7f 100644 --- a/docs/generated/core_configuration.html +++ b/docs/generated/core_configuration.html @@ -1047,6 +1047,12 @@ Integer To avoid frequent manifest merges, this parameter specifies the minimum number of ManifestFileMeta to merge.
Note: when 'manifest-sort.enabled' is true, this minimum-count gate is only applied to the trailing sub-segment of a section that exceeds 'manifest-sort.max-rewrite-size'. Small under-budget sections are sorted and rewritten directly, so two small manifest files may be merged into one even when their count is below this threshold and full compaction is not triggered. + +
manifest.merge-optimize.enabled
+ true + Boolean + Whether to enable block-aware ordinary manifest merging. When disabled, ordinary manifest compaction uses the legacy full-entry merger. +
manifest.target-file-size
8 mb diff --git a/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java b/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java index b30e01544267..8e5e8c1300cb 100644 --- a/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java +++ b/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java @@ -589,6 +589,15 @@ public InlineElement getDescription() { + " skipped. Set to a larger value to allow more aggressive" + " sort rewriting. The cap only limits the sorted rewrite portion and full/minor cleanup may still happen beyond it."); + public static final ConfigOption MANIFEST_MERGE_OPTIMIZE_ENABLED = + key("manifest.merge-optimize.enabled") + .booleanType() + .defaultValue(true) + .withDescription( + "Whether to enable block-aware ordinary manifest merging. When" + + " disabled, ordinary manifest compaction uses the legacy" + + " full-entry merger."); + public static final ConfigOption PARTITION_DEFAULT_NAME = key("partition.default-name") .stringType() @@ -3066,6 +3075,10 @@ public long manifestSortMaxRewriteSize() { return options.get(MANIFEST_SORT_MAX_REWRITE_SIZE).getBytes(); } + public boolean manifestMergeOptimizeEnabled() { + return options.get(MANIFEST_MERGE_OPTIMIZE_ENABLED); + } + public String partitionDefaultName() { return options.get(PARTITION_DEFAULT_NAME); } diff --git a/paimon-core/src/main/java/org/apache/paimon/io/ProjectedDataFileMeta.java b/paimon-core/src/main/java/org/apache/paimon/io/ProjectedDataFileMeta.java index b707065eb19f..e525c4e2fc53 100644 --- a/paimon-core/src/main/java/org/apache/paimon/io/ProjectedDataFileMeta.java +++ b/paimon-core/src/main/java/org/apache/paimon/io/ProjectedDataFileMeta.java @@ -230,6 +230,16 @@ public boolean hasFirstRowId() { return !currentRow().isNullAt(requiredPosition(Fields.FIRST_ROW_ID)); } + @Override + public long nonNullFirstRowId() { + // Read the primitive value directly on manifest scan hot paths. Calling firstRowId() + // here would box every value as Long before immediately unboxing it again. + int position = requiredPosition(Fields.FIRST_ROW_ID); + InternalRow row = currentRow(); + checkState(!row.isNullAt(position), "First row id cannot be null."); + return row.getLong(position); + } + @Nullable @Override public Long firstRowId() { diff --git a/paimon-core/src/main/java/org/apache/paimon/manifest/CollectedDeletes.java b/paimon-core/src/main/java/org/apache/paimon/manifest/CollectedDeletes.java new file mode 100644 index 000000000000..49465c303d20 --- /dev/null +++ b/paimon-core/src/main/java/org/apache/paimon/manifest/CollectedDeletes.java @@ -0,0 +1,114 @@ +/* + * 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.manifest; + +import org.apache.paimon.data.BinaryRow; +import org.apache.paimon.io.ProjectedDataFileMeta; +import org.apache.paimon.manifest.FileEntry.ReusableIdentifier; + +import java.util.Collections; +import java.util.HashSet; +import java.util.Set; + +import static org.apache.paimon.utils.Preconditions.checkState; + +/** DELETE identifiers and optional RowID and partition indexes collected for manifest merging. */ +public final class CollectedDeletes { + + private final CompactFileIdentifierSet identifiers = new CompactFileIdentifierSet(); + private final DeletedRowIdSet rowIds = new DeletedRowIdSet(); + private Set partitions = new HashSet<>(); + private final boolean useRowIdFilter; + private boolean immutable; + + public CollectedDeletes(boolean useRowIdFilter) { + this.useRowIdFilter = useRowIdFilter; + } + + public void add( + ProjectedManifestEntry entry, boolean collectRowIds, boolean collectPartitions) { + checkState(!immutable, "Cannot modify an immutable DELETE collection."); + identifiers.add(entry); + if (collectPartitions) { + partitions.add(entry.partition().copy()); + } + if (collectRowIds) { + rowIds.add(entry.file().nonNullFirstRowId()); + } + } + + public void combine(CollectedDeletes other) { + checkState(!immutable, "Cannot modify an immutable DELETE collection."); + checkState( + useRowIdFilter == other.useRowIdFilter, + "Cannot combine DELETE collections with different RowID modes."); + identifiers.addAll(other.identifiers); + rowIds.addAll(other.rowIds); + partitions.addAll(other.partitions); + } + + public CollectedDeletes toImmutable() { + checkState(!immutable, "Cannot modify an immutable DELETE collection."); + if (useRowIdFilter) { + rowIds.prepareRangeIndex(); + } + partitions = Collections.unmodifiableSet(partitions); + immutable = true; + return this; + } + + public boolean isEmpty() { + return identifiers.isEmpty(); + } + + public Set partitions() { + return partitions; + } + + public boolean useRowIdFilter() { + return useRowIdFilter; + } + + public boolean isDeleted(ProjectedManifestEntry entry, ReusableIdentifier reusableIdentifier) { + if (useRowIdFilter) { + ProjectedDataFileMeta file = entry.file(); + checkState(file.hasFirstRowId(), "First row id should not be null."); + if (!rowIds.contains(file.nonNullFirstRowId())) { + return false; + } + } + return identifiers.contains(reusableIdentifier.replaceWithPartition(entry)); + } + + public boolean copyable( + ProjectedManifestEntry entry, + ReusableIdentifier reusableIdentifier, + boolean deferDeletedAddCheck) { + return entry.isAdd() && (deferDeletedAddCheck || !isDeleted(entry, reusableIdentifier)); + } + + public boolean intersectsRowIds(long minRowId, long maxRowId) { + return rowIds.intersects(minRowId, maxRowId); + } + + public void release() { + identifiers.release(); + rowIds.releaseRangeIndex(); + } +} diff --git a/paimon-core/src/main/java/org/apache/paimon/manifest/CompactFileIdentifierSet.java b/paimon-core/src/main/java/org/apache/paimon/manifest/CompactFileIdentifierSet.java index b11b21345889..eca97f75ddbc 100644 --- a/paimon-core/src/main/java/org/apache/paimon/manifest/CompactFileIdentifierSet.java +++ b/paimon-core/src/main/java/org/apache/paimon/manifest/CompactFileIdentifierSet.java @@ -74,6 +74,13 @@ public void add(int partitionId, ReusableIdentifier identifier) { add(partitionId, identifier.bytes(), identifier.length()); } + public void addAll(CompactFileIdentifierSet other) { + checkArgument(other != null, "Identifier set cannot be null."); + for (int entry = 0; entry < other.size; entry++) { + add(other.partitionIds[entry], other.arena, other.offsets[entry], other.lengths[entry]); + } + } + public boolean contains(ProjectedManifestEntry entry) { return contains(reusableIdentifier().replaceWithPartition(entry)); } @@ -104,9 +111,13 @@ public void release() { } void add(int partitionId, byte[] identifier, int length) { - checkIdentifier(identifier, length); - long hash = hash(partitionId, identifier, length); - if (contains(partitionId, identifier, length, hash)) { + add(partitionId, identifier, 0, length); + } + + private void add(int partitionId, byte[] identifier, int offset, int length) { + checkIdentifier(identifier, offset, length); + long hash = hash(partitionId, identifier, offset, length); + if (contains(partitionId, identifier, offset, length, hash)) { return; } if (size + 1 > (int) (buckets.length * LOAD_FACTOR)) { @@ -114,14 +125,14 @@ void add(int partitionId, byte[] identifier, int length) { } ensureEntryCapacity(size + 1); ensureArenaCapacity(length); - int offset = arenaSize; - System.arraycopy(identifier, 0, arena, offset, length); + int arenaOffset = arenaSize; + System.arraycopy(identifier, offset, arena, arenaOffset, length); arenaSize = Math.addExact(arenaSize, length); int bucket = bucket(hash); hashes[size] = hash; partitionIds[size] = partitionId; - offsets[size] = offset; + offsets[size] = arenaOffset; lengths[size] = length; next[size] = buckets[bucket]; buckets[bucket] = size; @@ -129,16 +140,18 @@ void add(int partitionId, byte[] identifier, int length) { } boolean contains(int partitionId, byte[] identifier, int length) { - checkIdentifier(identifier, length); - return contains(partitionId, identifier, length, hash(partitionId, identifier, length)); + checkIdentifier(identifier, 0, length); + return contains( + partitionId, identifier, 0, length, hash(partitionId, identifier, 0, length)); } - private boolean contains(int partitionId, byte[] identifier, int length, long hash) { + private boolean contains( + int partitionId, byte[] identifier, int offset, int length, long hash) { for (int entry = buckets[bucket(hash)]; entry >= 0; entry = next[entry]) { if (hashes[entry] == hash && partitionIds[entry] == partitionId && lengths[entry] == length - && bytesEqual(arena, offsets[entry], identifier, length)) { + && bytesEqual(arena, offsets[entry], identifier, offset, length)) { return true; } } @@ -194,20 +207,21 @@ private static int bucket(long hash, int bucketCount) { return ((int) (hash ^ (hash >>> 32))) & (bucketCount - 1); } - private static long hash(int partitionId, byte[] bytes, int length) { + private static long hash(int partitionId, byte[] bytes, int offset, int length) { long hash = 0xcbf29ce484222325L; hash ^= Integer.toUnsignedLong(partitionId); hash *= 0x100000001b3L; for (int i = 0; i < length; i++) { - hash ^= bytes[i] & 0xFFL; + hash ^= bytes[offset + i] & 0xFFL; hash *= 0x100000001b3L; } return hash; } - private static boolean bytesEqual(byte[] left, int leftOffset, byte[] right, int length) { + private static boolean bytesEqual( + byte[] left, int leftOffset, byte[] right, int rightOffset, int length) { for (int i = 0; i < length; i++) { - if (left[leftOffset + i] != right[i]) { + if (left[leftOffset + i] != right[rightOffset + i]) { return false; } } @@ -225,12 +239,13 @@ private ReusableIdentifier reusableIdentifier() { return reusableIdentifier; } - private static void checkIdentifier(byte[] identifier, int length) { + private static void checkIdentifier(byte[] identifier, int offset, int length) { checkArgument(identifier != null, "Identifier bytes cannot be null."); checkArgument( - length >= 0 && length <= identifier.length, - "Invalid identifier length %s.", - length); + offset >= 0 && length >= 0 && offset <= identifier.length - length, + "Invalid identifier range [%s, %s).", + offset, + offset + length); } private static int[] filledWithMinusOne(int length) { 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 new file mode 100644 index 000000000000..cd7203764d9d --- /dev/null +++ b/paimon-core/src/main/java/org/apache/paimon/manifest/DeletedRowIdSet.java @@ -0,0 +1,157 @@ +/* + * 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.manifest; + +import javax.annotation.Nullable; + +import java.util.Arrays; + +/** Primitive set used by RowID compaction to avoid rebuilding file identifiers. */ +final class DeletedRowIdSet { + + private static final long EMPTY = Long.MIN_VALUE; + + private long[] table = emptyTable(16); + private int size; + private boolean containsMinValue; + private @Nullable long[] sortedRowIds; + + void add(long value) { + if (value == EMPTY) { + if (!containsMinValue) { + containsMinValue = true; + size++; + sortedRowIds = null; + } + return; + } + if ((size + 1) * 2 > table.length) { + grow(); + } + int slot = slot(value, table.length); + while (table[slot] != EMPTY) { + if (table[slot] == value) { + return; + } + slot = (slot + 1) & (table.length - 1); + } + table[slot] = value; + size++; + sortedRowIds = null; + } + + void addAll(DeletedRowIdSet other) { + if (other.containsMinValue) { + add(EMPTY); + } + for (long value : other.table) { + if (value != EMPTY) { + add(value); + } + } + } + + boolean contains(long value) { + if (value == EMPTY) { + return containsMinValue; + } + int slot = slot(value, table.length); + while (table[slot] != EMPTY) { + if (table[slot] == value) { + return true; + } + slot = (slot + 1) & (table.length - 1); + } + return false; + } + + boolean intersects(long minInclusive, long maxInclusive) { + if (minInclusive > maxInclusive) { + return true; + } + long[] values = sortedRowIds(); + int position = Arrays.binarySearch(values, minInclusive); + if (position < 0) { + position = -position - 1; + } + return position < values.length && values[position] <= maxInclusive; + } + + void prepareRangeIndex() { + // Publish the immutable sorted snapshot before concurrent manifest planning starts. + sortedRowIds(); + } + + void releaseRangeIndex() { + sortedRowIds = null; + } + + private long[] sortedRowIds() { + if (sortedRowIds != null) { + return sortedRowIds; + } + long[] values = new long[size]; + int position = 0; + if (containsMinValue) { + values[position++] = EMPTY; + } + for (long value : table) { + if (value != EMPTY) { + values[position++] = value; + } + } + if (position != size) { + throw new IllegalStateException("Failed to snapshot deleted RowID set."); + } + Arrays.sort(values); + sortedRowIds = values; + return values; + } + + private void grow() { + long[] previous = table; + if (previous.length >= (1 << 30)) { + throw new IllegalStateException("Too many deleted RowIDs in one manifest group."); + } + table = emptyTable(previous.length << 1); + int previousSize = size; + size = containsMinValue ? 1 : 0; + for (long value : previous) { + if (value != EMPTY) { + add(value); + } + } + if (size != previousSize) { + throw new IllegalStateException("Failed to grow deleted RowID set."); + } + } + + private static int slot(long value, int length) { + value ^= value >>> 33; + value *= 0xff51afd7ed558ccdL; + value ^= value >>> 33; + return ((int) value) & (length - 1); + } + + private static long[] emptyTable(int length) { + long[] table = new long[length]; + Arrays.fill(table, EMPTY); + return table; + } +} diff --git a/paimon-core/src/main/java/org/apache/paimon/manifest/ManifestAvroReader.java b/paimon-core/src/main/java/org/apache/paimon/manifest/ManifestAvroReader.java index 3fbb0008714e..54d52434f824 100644 --- a/paimon-core/src/main/java/org/apache/paimon/manifest/ManifestAvroReader.java +++ b/paimon-core/src/main/java/org/apache/paimon/manifest/ManifestAvroReader.java @@ -74,6 +74,11 @@ public boolean hasNext() throws IOException { return blockReader.hasNextBlock(); } + /** Returns whether raw blocks can be copied into a writer for the current manifest schema. */ + public boolean rawBlockCopySupported() { + return rawBlockCopySupported; + } + /** Returns the next raw block without decompressing it. */ public RawBlock next() throws IOException { if (!hasNext()) { @@ -372,6 +377,12 @@ public boolean rawBlockCopySupported() { public AvroRawBlock encodedBlock() { return block; } + + /** Returns an independently owned block which remains valid after this reader advances. */ + public RawBlock stableCopy() { + return new RawBlock( + decoderContext, rawBlockCopySupported, block.stableCopy(), blockOrdinal); + } } /** Decoder state shared by the borrowed blocks produced by one reader. */ diff --git a/paimon-core/src/main/java/org/apache/paimon/manifest/ManifestAvroWriter.java b/paimon-core/src/main/java/org/apache/paimon/manifest/ManifestAvroWriter.java index 6244ed569c9e..16c4b68ab22c 100644 --- a/paimon-core/src/main/java/org/apache/paimon/manifest/ManifestAvroWriter.java +++ b/paimon-core/src/main/java/org/apache/paimon/manifest/ManifestAvroWriter.java @@ -19,6 +19,7 @@ package org.apache.paimon.manifest; import org.apache.paimon.data.BinaryRow; +import org.apache.paimon.data.InternalRow; import org.apache.paimon.format.SimpleColStats; import org.apache.paimon.format.SimpleStatsCollector; import org.apache.paimon.format.avro.AvroBlockWriter; @@ -29,6 +30,7 @@ import org.apache.paimon.fs.PositionOutputStream; import org.apache.paimon.io.RollingFileWriter; import org.apache.paimon.schema.SchemaManager; +import org.apache.paimon.stats.SimpleStats; import org.apache.paimon.stats.SimpleStatsConverter; import org.apache.paimon.types.RowType; import org.apache.paimon.utils.IOUtils; @@ -54,6 +56,8 @@ */ public final class ManifestAvroWriter implements AutoCloseable { + private static final int MAX_BUFFERED_ENCODED_PARTITIONS = 8_192; + private final FileIO fileIO; private final SchemaManager schemaManager; private final RowType partitionType; @@ -114,7 +118,19 @@ public void writeEncoded(ByteBuffer encodedRecord, EncodedEntry metadata) throws } } - public void writeEncodedBlock(AvroRawBlock block, EncodedBlock metadata) throws IOException { + /** Writes an already decoded manifest row while collecting its projected metadata. */ + public void writeRow(InternalRow row, EncodedEntry metadata) throws IOException { + try { + currentWriter().writeRow(row, metadata); + afterWrite(1, false); + } catch (IOException | RuntimeException | Error failure) { + abort(); + throw failure; + } + } + + public void writeEncodedBlock(AvroRawBlock block, EncodedBlockMeta metadata) + throws IOException { if (metadata.addedFiles < 0 || metadata.deletedFiles < 0) { throw new IllegalArgumentException( String.format( @@ -137,6 +153,38 @@ public void writeEncodedBlock(AvroRawBlock block, EncodedBlock metadata) throws } } + /** Copies all raw blocks from one manifest and reuses its aggregate metadata. */ + public void writeEncodedManifest(ManifestAvroReader reader, ManifestFileMeta metadata) + throws IOException { + if (!reader.rawBlockCopySupported()) { + throw new IllegalArgumentException( + "Manifest schema is incompatible with raw block copying."); + } + try { + FileWriter fileWriter = currentWriter(); + long copiedRecords = 0; + while (reader.hasNext()) { + ManifestAvroReader.RawBlock block = reader.next(); + fileWriter.ensureOpen(); + fileWriter.writer.addEncodedBlock(block.encodedBlock()); + copiedRecords = Math.addExact(copiedRecords, block.recordCount()); + } + long metadataRecords = + Math.addExact(metadata.numAddedFiles(), metadata.numDeletedFiles()); + if (copiedRecords != metadataRecords) { + throw new IllegalArgumentException( + String.format( + "Manifest record count mismatch: metadata %s, blocks %s.", + metadataRecords, copiedRecords)); + } + fileWriter.collectStats(metadata); + afterWrite(copiedRecords, true); + } catch (IOException | RuntimeException | Error failure) { + abort(); + throw failure; + } + } + private FileWriter currentWriter() { if (closed) { throw new IllegalStateException("Manifest writer has already closed."); @@ -213,6 +261,7 @@ public static final class EncodedEntry { private int bucket; private int level; private long schemaId; + private boolean hasRowId; private long firstRowId; private long rowCount; @@ -229,14 +278,33 @@ public EncodedEntry replace( this.bucket = bucket; this.level = level; this.schemaId = schemaId; + this.hasRowId = true; this.firstRowId = firstRowId; this.rowCount = rowCount; return this; } + + public EncodedEntry replace( + byte kind, + BinaryRow partition, + int bucket, + int level, + long schemaId, + long rowCount) { + this.kind = kind; + this.partition = partition; + this.bucket = bucket; + this.level = level; + this.schemaId = schemaId; + this.hasRowId = false; + this.firstRowId = 0; + this.rowCount = rowCount; + return this; + } } /** Aggregate statistics for an encoded Avro block copied without decompression. */ - public static final class EncodedBlock { + public static final class EncodedBlockMeta { private final long addedFiles; private final long deletedFiles; @@ -247,12 +315,9 @@ public static final class EncodedBlock { private final int maxLevel; private final long minRowId; private final long maxRowId; - private final @Nullable BinaryRow nullPartition; - private final long nullPartitionCount; - private final @Nullable BinaryRow minNonNullPartition; - private final @Nullable BinaryRow maxNonNullPartition; + private final SimpleStats partitionStats; - public EncodedBlock( + public EncodedBlockMeta( long addedFiles, long deletedFiles, long schemaId, @@ -262,10 +327,7 @@ public EncodedBlock( int maxLevel, long minRowId, long maxRowId, - @Nullable BinaryRow nullPartition, - long nullPartitionCount, - @Nullable BinaryRow minNonNullPartition, - @Nullable BinaryRow maxNonNullPartition) { + SimpleStats partitionStats) { this.addedFiles = addedFiles; this.deletedFiles = deletedFiles; this.schemaId = schemaId; @@ -275,10 +337,7 @@ public EncodedBlock( this.maxLevel = maxLevel; this.minRowId = minRowId; this.maxRowId = maxRowId; - this.nullPartition = nullPartition; - this.nullPartitionCount = nullPartitionCount; - this.minNonNullPartition = minNonNullPartition; - this.maxNonNullPartition = maxNonNullPartition; + this.partitionStats = partitionStats; } } @@ -289,6 +348,9 @@ private final class FileWriter { private final SimpleStatsConverter partitionStatsSerializer; private final Map encodedPartitionCounts = new IdentityHashMap<>(); private final long[] repeatedNullCounts = new long[partitionType.getFieldCount()]; + private final long[] copiedManifestNullCounts = new long[partitionType.getFieldCount()]; + private final long[] copiedManifestRepresentativeNullCounts = + new long[partitionType.getFieldCount()]; private @Nullable PositionOutputStream out; private @Nullable AvroBlockWriter writer; private @Nullable Long outputBytes; @@ -299,6 +361,8 @@ private final class FileWriter { private int maxBucket = Integer.MIN_VALUE; private int minLevel = Integer.MAX_VALUE; private int maxLevel = Integer.MIN_VALUE; + private boolean bucketStatsKnown = true; + private boolean levelStatsKnown = true; private @Nullable RowIdStats rowIdStats = new RowIdStats(); private boolean closed; @@ -348,20 +412,19 @@ private void writeEncoded(ByteBuffer encodedRecord, EncodedEntry metadata) addEncodedPartition(metadata.partition, 1); } - private void writeEncodedBlock(AvroRawBlock block, EncodedBlock metadata) + private void writeRow(InternalRow row, EncodedEntry metadata) throws IOException { + ensureOpen(); + writer.addElement(row); + collectStats(metadata); + addEncodedPartition(metadata.partition, 1); + } + + private void writeEncodedBlock(AvroRawBlock block, EncodedBlockMeta metadata) throws IOException { ensureOpen(); writer.addEncodedBlock(block); collectStats(metadata); - if (metadata.nullPartitionCount > 0) { - addEncodedPartition(metadata.nullPartition, metadata.nullPartitionCount); - } - if (metadata.minNonNullPartition != null) { - addEncodedPartition(metadata.minNonNullPartition, 1); - if (metadata.maxNonNullPartition != metadata.minNonNullPartition) { - addEncodedPartition(metadata.maxNonNullPartition, 1); - } - } + collectCopiedPartitionStats(metadata.partitionStats); } private void collectStats(ManifestEntry entry) { @@ -408,20 +471,77 @@ private void collectStats(EncodedEntry entry) { minLevel = Math.min(minLevel, entry.level); maxLevel = Math.max(maxLevel, entry.level); if (rowIdStats != null) { - rowIdStats.collect(entry.firstRowId, entry.rowCount); + if (!entry.hasRowId) { + rowIdStats = null; + } else { + rowIdStats.collect(entry.firstRowId, entry.rowCount); + } } } - private void collectStats(EncodedBlock block) { - numAddedFiles = Math.addExact(numAddedFiles, block.addedFiles); - numDeletedFiles = Math.addExact(numDeletedFiles, block.deletedFiles); - schemaId = Math.max(schemaId, block.schemaId); - minBucket = Math.min(minBucket, block.minBucket); - maxBucket = Math.max(maxBucket, block.maxBucket); - minLevel = Math.min(minLevel, block.minLevel); - maxLevel = Math.max(maxLevel, block.maxLevel); + private void collectStats(EncodedBlockMeta metadata) { + numAddedFiles = Math.addExact(numAddedFiles, metadata.addedFiles); + numDeletedFiles = Math.addExact(numDeletedFiles, metadata.deletedFiles); + schemaId = Math.max(schemaId, metadata.schemaId); + minBucket = Math.min(minBucket, metadata.minBucket); + maxBucket = Math.max(maxBucket, metadata.maxBucket); + minLevel = Math.min(minLevel, metadata.minLevel); + maxLevel = Math.max(maxLevel, metadata.maxLevel); if (rowIdStats != null) { - rowIdStats.collectRange(block.minRowId, block.maxRowId); + if (metadata.minRowId < 0 || metadata.maxRowId < 0) { + rowIdStats = null; + } else { + rowIdStats.collectRange(metadata.minRowId, metadata.maxRowId); + } + } + } + + private void collectStats(ManifestFileMeta manifest) { + numAddedFiles = Math.addExact(numAddedFiles, manifest.numAddedFiles()); + numDeletedFiles = Math.addExact(numDeletedFiles, manifest.numDeletedFiles()); + schemaId = Math.max(schemaId, manifest.schemaId()); + if (manifest.minBucket() == null || manifest.maxBucket() == null) { + bucketStatsKnown = false; + } else { + minBucket = Math.min(minBucket, manifest.minBucket()); + maxBucket = Math.max(maxBucket, manifest.maxBucket()); + } + if (manifest.minLevel() == null || manifest.maxLevel() == null) { + levelStatsKnown = false; + } else { + minLevel = Math.min(minLevel, manifest.minLevel()); + maxLevel = Math.max(maxLevel, manifest.maxLevel()); + } + if (rowIdStats != null) { + if (manifest.minRowId() == null || manifest.maxRowId() == null) { + rowIdStats = null; + } else { + rowIdStats.collectRange(manifest.minRowId(), manifest.maxRowId()); + } + } + + collectCopiedPartitionStats(manifest.partitionStats()); + } + + private void collectCopiedPartitionStats(SimpleStats partitionStats) { + collectCopiedPartitionRepresentative(partitionStats.minValues()); + if (!partitionStats.maxValues().equals(partitionStats.minValues())) { + collectCopiedPartitionRepresentative(partitionStats.maxValues()); + } + for (int field = 0; field < copiedManifestNullCounts.length; field++) { + copiedManifestNullCounts[field] = + Math.addExact( + copiedManifestNullCounts[field], + partitionStats.nullCounts().getLong(field)); + } + } + + private void collectCopiedPartitionRepresentative(BinaryRow partition) { + partitionStatsCollector.collect(partition); + for (int field = 0; field < partition.getFieldCount(); field++) { + if (partition.isNullAt(field)) { + copiedManifestRepresentativeNullCounts[field]++; + } } } @@ -432,9 +552,37 @@ private void addEncodedPartition(@Nullable BinaryRow partition, long count) { long[] value = encodedPartitionCounts.computeIfAbsent(partition, ignored -> new long[1]); value[0] = Math.addExact(value[0], count); + if (encodedPartitionCounts.size() >= MAX_BUFFERED_ENCODED_PARTITIONS) { + flushEncodedPartitions(); + } } private SimpleColStats[] partitionStats() { + flushEncodedPartitions(); + SimpleColStats[] stats = partitionStatsCollector.extract(); + for (int field = 0; field < stats.length; field++) { + // The collector sees only the min/max partition representatives of each copied + // block or manifest. Add the remaining nulls from its aggregate statistics. + long nullCountAdjustment = + Math.addExact( + repeatedNullCounts[field], + Math.subtractExact( + copiedManifestNullCounts[field], + copiedManifestRepresentativeNullCounts[field])); + if (nullCountAdjustment == 0) { + continue; + } + SimpleColStats current = stats[field]; + stats[field] = + new SimpleColStats( + current.min(), + current.max(), + Math.addExact(current.nullCount(), nullCountAdjustment)); + } + return stats; + } + + private void flushEncodedPartitions() { for (Map.Entry entry : encodedPartitionCounts.entrySet()) { BinaryRow partition = entry.getKey(); partitionStatsCollector.collect(partition); @@ -450,19 +598,6 @@ private SimpleColStats[] partitionStats() { } } encodedPartitionCounts.clear(); - SimpleColStats[] stats = partitionStatsCollector.extract(); - for (int field = 0; field < stats.length; field++) { - if (repeatedNullCounts[field] == 0) { - continue; - } - SimpleColStats current = stats[field]; - stats[field] = - new SimpleColStats( - current.min(), - current.max(), - Math.addExact(current.nullCount(), repeatedNullCounts[field])); - } - return stats; } private boolean reachTargetSize(boolean suggestedCheck, long targetSize) @@ -519,10 +654,10 @@ private ManifestFileMeta result() { numAddedFiles + numDeletedFiles > 0 ? schemaId : schemaManager.latest().get().id(), - minBucket, - maxBucket, - minLevel, - maxLevel, + bucketStatsKnown ? minBucket : null, + bucketStatsKnown ? maxBucket : null, + levelStatsKnown ? minLevel : null, + levelStatsKnown ? maxLevel : null, rowIdStats == null ? null : rowIdStats.minRowId, rowIdStats == null ? null : rowIdStats.maxRowId); } 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 b9fa876f61de..e91f84a2e65c 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 @@ -203,6 +203,15 @@ private static CloseableIterator createManifestIterator( } } + /** Opens a low-allocation reader over raw Avro manifest blocks. */ + public ManifestAvroReader scanAvroBlocks(String fileName, @Nullable Long fileSize) { + try { + return new ManifestAvroReader(fileIO.newInputStream(pathFactory.toPath(fileName))); + } catch (IOException e) { + throw new UncheckedIOException("Failed to read manifest file " + fileName, e); + } + } + @VisibleForTesting public long suggestedFileSize() { return suggestedFileSize; diff --git a/paimon-core/src/main/java/org/apache/paimon/manifest/PartitionDictionary.java b/paimon-core/src/main/java/org/apache/paimon/manifest/PartitionDictionary.java new file mode 100644 index 000000000000..b925457e68dd --- /dev/null +++ b/paimon-core/src/main/java/org/apache/paimon/manifest/PartitionDictionary.java @@ -0,0 +1,62 @@ +/* + * 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.manifest; + +import org.apache.paimon.data.BinaryRow; +import org.apache.paimon.utils.ByteArrayKey; +import org.apache.paimon.utils.ByteArrayLookupKey; +import org.apache.paimon.utils.SerializationUtils; + +import java.util.Arrays; +import java.util.HashMap; +import java.util.Map; + +/** Deduplicates serialized partitions and assigns compact integer identifiers. */ +public final class PartitionDictionary { + + private final Map ids = new HashMap<>(); + private final ByteArrayLookupKey lookup = new ByteArrayLookupKey(); + private BinaryRow[] partitions = new BinaryRow[16]; + private int partitionCount; + + public int id(byte[] bytes) { + lookup.reset(bytes); + try { + Integer existing = ids.get(lookup); + if (existing != null) { + return existing; + } + byte[] canonical = Arrays.copyOf(bytes, bytes.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 { + lookup.clear(); + } + } + + public BinaryRow partition(int id) { + return partitions[id]; + } +} diff --git a/paimon-core/src/main/java/org/apache/paimon/manifest/ProjectedManifestEntry.java b/paimon-core/src/main/java/org/apache/paimon/manifest/ProjectedManifestEntry.java index e7a279f0f0cd..591d86759e6e 100644 --- a/paimon-core/src/main/java/org/apache/paimon/manifest/ProjectedManifestEntry.java +++ b/paimon-core/src/main/java/org/apache/paimon/manifest/ProjectedManifestEntry.java @@ -49,6 +49,7 @@ public final class ProjectedManifestEntry implements ManifestEntry { private static final Projection FULL_PROJECTION = Projection.create(MANIFEST_ROW_TYPE); public static final Projection DELETE_ENTRY_PROJECTION = createDeleteEntryProjection(); public static final Projection ROW_RANGE_PROJECTION = createRowRangeProjection(); + public static final Projection ENTRY_LAYOUT_PROJECTION = createEntryLayoutProjection(); private final Projection projection; private final @Nullable ProjectedDataFileMeta file; @@ -132,6 +133,29 @@ private static Projection createRowRangeProjection() { DataFileMeta.FIRST_ROW_ID))))); } + private static Projection createEntryLayoutProjection() { + RowType manifestType = MANIFEST_ROW_TYPE; + return Projection.create( + new RowType( + false, + Arrays.asList( + manifestType.getField(ManifestEntry.KIND), + manifestType.getField(ManifestEntry.PARTITION), + manifestType.getField(ManifestEntry.BUCKET), + manifestType + .getField(ManifestEntry.FILE) + .newType( + DataFileMeta.SCHEMA.project( + DataFileMeta.FILE_NAME, + DataFileMeta.ROW_COUNT, + DataFileMeta.LEVEL, + DataFileMeta.SCHEMA_ID, + DataFileMeta.FIRST_ROW_ID, + DataFileMeta.EXTRA_FILES, + DataFileMeta.EMBEDDED_FILE_INDEX, + DataFileMeta.EXTERNAL_PATH))))); + } + /** Drops references to the current row before its reader batch is released. */ public void clear() { row = null; @@ -366,7 +390,7 @@ private static void validateProjection(RowType projectedType) { } } - RowType projectedType() { + public RowType projectedType() { return projectedType; } diff --git a/paimon-core/src/main/java/org/apache/paimon/operation/ManifestFileBlockMerger.java b/paimon-core/src/main/java/org/apache/paimon/operation/ManifestFileBlockMerger.java new file mode 100644 index 000000000000..baef36d807ce --- /dev/null +++ b/paimon-core/src/main/java/org/apache/paimon/operation/ManifestFileBlockMerger.java @@ -0,0 +1,916 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.paimon.operation; + +import org.apache.paimon.CoreOptions; +import org.apache.paimon.data.BinaryRow; +import org.apache.paimon.data.GenericRow; +import org.apache.paimon.format.SimpleColStats; +import org.apache.paimon.format.SimpleStatsCollector; +import org.apache.paimon.io.ProjectedDataFileMeta; +import org.apache.paimon.manifest.CollectedDeletes; +import org.apache.paimon.manifest.CompactFileIdentifierSet; +import org.apache.paimon.manifest.FileEntry.ReusableIdentifier; +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.ManifestFile; +import org.apache.paimon.manifest.ManifestFileMeta; +import org.apache.paimon.manifest.PartitionDictionary; +import org.apache.paimon.manifest.ProjectedManifestEntry; +import org.apache.paimon.partition.PartitionPredicate; +import org.apache.paimon.stats.SimpleStatsConverter; +import org.apache.paimon.types.RowType; +import org.apache.paimon.utils.CloseableIterator; +import org.apache.paimon.utils.Filter; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import javax.annotation.Nullable; + +import java.util.ArrayList; +import java.util.Collections; +import java.util.HashMap; +import java.util.Iterator; +import java.util.LinkedList; +import java.util.List; +import java.util.Map; +import java.util.Optional; +import java.util.function.Function; + +import static org.apache.paimon.manifest.ManifestFileMeta.allContainsRowId; +import static org.apache.paimon.utils.ManifestReadThreadPool.sequentialBatchedExecute; +import static org.apache.paimon.utils.Preconditions.checkArgument; +import static org.apache.paimon.utils.Preconditions.checkState; + +/** Block-aware manifest compaction which never delegates to the legacy full-entry merger. */ +final class ManifestFileBlockMerger { + + private static final Logger LOG = LoggerFactory.getLogger(ManifestFileBlockMerger.class); + + private ManifestFileBlockMerger() {} + + static List merge( + List input, + List newFilesForAbort, + ManifestFile manifestFile, + RowType partitionType, + CoreOptions options) + throws Exception { + long suggestedMetaSize = options.manifestTargetSize().getBytes(); + Integer manifestReadParallelism = options.scanManifestParallelism(); + Optional> fullCompacted = + tryFullCompaction( + input, + newFilesForAbort, + manifestFile, + suggestedMetaSize, + options.manifestFullCompactionThresholdSize().getBytes(), + partitionType, + manifestReadParallelism); + if (fullCompacted.isPresent()) { + return fullCompacted.get(); + } + return compactMinor( + input, + newFilesForAbort, + manifestFile, + partitionType, + suggestedMetaSize, + options.manifestMergeMinCount(), + manifestReadParallelism); + } + + static Optional> tryFullCompaction( + List inputs, + List newFilesForAbort, + ManifestFile manifestFile, + long suggestedMetaSize, + long sizeTrigger, + RowType partitionType, + @Nullable Integer manifestReadParallelism) + throws Exception { + checkArgument(sizeTrigger > 0, "Manifest full compaction size trigger cannot be zero."); + + Filter mustChange = + file -> file.numDeletedFiles() > 0 || file.fileSize() < suggestedMetaSize; + long totalManifestSize = 0; + long deltaDeleteFileNum = 0; + long totalDeltaFileSize = 0; + List deltaManifests = new ArrayList<>(); + for (ManifestFileMeta file : inputs) { + totalManifestSize += file.fileSize(); + if (mustChange.test(file)) { + totalDeltaFileSize += file.fileSize(); + deltaDeleteFileNum += file.numDeletedFiles(); + deltaManifests.add(file); + } + } + + if (totalDeltaFileSize < sizeTrigger) { + return Optional.empty(); + } + + LOG.info( + "Start Block-aware Manifest File Full Compaction: totalManifestSize: {}, deltaDeleteFileNum {}, totalDeltaFileSize {}", + totalManifestSize, + deltaDeleteFileNum, + totalDeltaFileSize); + + boolean useRowIdFilter = allContainsRowId(inputs); + final CollectedDeletes deletes = + collectDeletes( + deltaManifests, + manifestFile, + useRowIdFilter, + true, + manifestReadParallelism) + .toImmutable(); + try { + PartitionPredicate predicate; + if (deletes.isEmpty()) { + predicate = PartitionPredicate.ALWAYS_FALSE; + } else if (partitionType.getFieldCount() > 0) { + predicate = PartitionPredicate.fromMultiple(partitionType, deletes.partitions()); + } else { + predicate = PartitionPredicate.ALWAYS_TRUE; + } + + List result = new ArrayList<>(); + List toCompact = new LinkedList<>(inputs); + if (predicate != null) { + Iterator iterator = toCompact.iterator(); + while (iterator.hasNext()) { + ManifestFileMeta file = iterator.next(); + if (mustChange.test(file)) { + continue; + } + if (!predicate.test( + file.numAddedFiles() + file.numDeletedFiles(), + file.partitionStats().minValues(), + file.partitionStats().maxValues(), + file.partitionStats().nullCounts())) { + iterator.remove(); + result.add(file); + } + } + } + + if (toCompact.size() <= 1) { + return Optional.empty(); + } + + List rewritten = + rewriteManifests( + toCompact, + manifestFile, + partitionType, + deletes, + true, + mustChange, + result, + manifestReadParallelism); + result.addAll(rewritten); + newFilesForAbort.addAll(rewritten); + return Optional.of(result); + } finally { + deletes.release(); + } + } + + private static List compactMinor( + List input, + List newFilesForAbort, + ManifestFile manifestFile, + RowType partitionType, + long suggestedMetaSize, + int suggestedMinMetaCount, + @Nullable Integer manifestReadParallelism) + throws Exception { + List result = new ArrayList<>(); + List candidates = new ArrayList<>(); + long totalSize = 0; + for (ManifestFileMeta manifest : input) { + totalSize += manifest.fileSize(); + candidates.add(manifest); + if (totalSize >= suggestedMetaSize) { + compactMinorBatch( + candidates, + result, + newFilesForAbort, + manifestFile, + partitionType, + manifestReadParallelism); + candidates.clear(); + totalSize = 0; + } + } + + if (candidates.size() >= suggestedMinMetaCount) { + compactMinorBatch( + candidates, + result, + newFilesForAbort, + manifestFile, + partitionType, + manifestReadParallelism); + } else { + result.addAll(candidates); + } + return result; + } + + private static void compactMinorBatch( + List candidates, + List result, + List newFilesForAbort, + ManifestFile manifestFile, + RowType partitionType, + @Nullable Integer manifestReadParallelism) + throws Exception { + if (candidates.size() == 1) { + result.add(candidates.get(0)); + return; + } + + List compacted = + mergeMinorManifests( + candidates, manifestFile, partitionType, manifestReadParallelism); + result.addAll(compacted); + newFilesForAbort.addAll(compacted); + } + + private static CollectedDeletes collectDeletes( + List manifests, + ManifestFile manifestFile, + boolean collectRowIds, + boolean collectPartitions, + @Nullable Integer manifestReadParallelism) { + List manifestsWithDeletes = new ArrayList<>(); + for (ManifestFileMeta manifest : manifests) { + if (manifest.numDeletedFiles() > 0) { + manifestsWithDeletes.add(manifest); + } + } + + CollectedDeletes result = new CollectedDeletes(collectRowIds); + if ((manifestReadParallelism != null && manifestReadParallelism <= 1) + || manifestsWithDeletes.size() <= 1) { + for (ManifestFileMeta manifest : manifestsWithDeletes) { + CollectedDeletes deletes = + collectDeletedEntries( + manifest, manifestFile, collectRowIds, collectPartitions); + result.combine(deletes); + deletes.release(); + } + return result; + } + + Function> scan = + manifest -> + Collections.singletonList( + collectDeletedEntries( + manifest, manifestFile, collectRowIds, collectPartitions)); + for (CollectedDeletes deletes : + sequentialBatchedExecute(scan, manifestsWithDeletes, manifestReadParallelism)) { + result.combine(deletes); + deletes.release(); + } + return result; + } + + private static CollectedDeletes collectDeletedEntries( + ManifestFileMeta manifest, + ManifestFile manifestFile, + boolean collectRowIds, + boolean collectPartitions) { + CollectedDeletes deletes = new CollectedDeletes(collectRowIds); + try (CloseableIterator entries = + manifestFile.scan( + manifest.fileName(), ProjectedManifestEntry.DELETE_ENTRY_PROJECTION)) { + while (entries.hasNext()) { + ProjectedManifestEntry entry = entries.next(); + if (!entry.isDelete()) { + continue; + } + deletes.add(entry, collectRowIds, collectPartitions); + } + return deletes; + } catch (Exception e) { + deletes.release(); + throw new RuntimeException( + "Failed to collect DELETE entries from manifest " + manifest.fileName(), e); + } + } + + /** + * Compacts manifests in input order. RowID manifests can copy unaffected ADD-only Avro blocks + * verbatim; manifests without RowID use identifiers to filter decoded entries. + */ + private static List mergeMinorManifests( + List manifests, + ManifestFile manifestFile, + RowType partitionType, + @Nullable Integer manifestReadParallelism) + throws Exception { + boolean useRowIdFilter = allContainsRowId(manifests); + final CollectedDeletes deletes = + collectDeletes( + manifests, + manifestFile, + useRowIdFilter, + false, + manifestReadParallelism) + .toImmutable(); + try { + return rewriteManifests( + manifests, + manifestFile, + partitionType, + deletes, + false, + null, + null, + manifestReadParallelism); + } finally { + deletes.release(); + } + } + + private static List rewriteManifests( + List manifests, + ManifestFile manifestFile, + RowType partitionType, + CollectedDeletes deletes, + boolean fullCompaction, + @Nullable Filter mustChange, + @Nullable List unchangedManifests, + @Nullable Integer manifestReadParallelism) + throws Exception { + ManifestAvroWriter writer = manifestFile.createAvroWriter(); + CompactFileIdentifierSet matchedEntries = new CompactFileIdentifierSet(); + CompactFileIdentifierSet emittedDeletes = new CompactFileIdentifierSet(); + PartitionDictionary partitions = new PartitionDictionary(); + SimpleStatsConverter partitionStatsConverter = new SimpleStatsConverter(partitionType); + EncodedEntry metadata = new EncodedEntry(); + ReusableIdentifier reusableIdentifier = new ReusableIdentifier(); + boolean hasDeletes = !deletes.isEmpty(); + try { + // DELETE lookups are immutable, and every planning worker owns its lookup scratch and + // block statistics. Plan manifests in parallel while the single ordered writer keeps + // matched ADD and emitted DELETE state on this thread. + if (hasDeletes + && (manifestReadParallelism == null || manifestReadParallelism > 1) + && manifests.size() > 1) { + // Keep decompression and primitive entry inspection parallel. The batched executor + // bounds retained raw blocks to at most one manifest per planning thread, while the + // single writer still emits manifests in input order. + Function> planner = + manifest -> { + try { + return Collections.singletonList( + planManifestRewrite( + manifest, manifestFile, partitionType, deletes)); + } catch (Exception e) { + throw new RuntimeException( + "Failed to plan manifest rewrite for " + + manifest.fileName(), + e); + } + }; + for (ManifestRewritePlan plan : + sequentialBatchedExecute(planner, manifests, manifestReadParallelism)) { + if (fullCompaction + && mustChange != null + && !mustChange.test(plan.manifest) + && plan.unchanged()) { + checkState( + unchangedManifests != null, + "Full compaction requires an unchanged manifest result."); + unchangedManifests.add(plan.manifest); + continue; + } + for (PlannedBlock block : plan.blocks) { + if (block.compaction.canCopyEncodedBlock()) { + writer.writeEncodedBlock( + block.raw.encodedBlock(), block.compaction.metadata); + } else { + writeBlockEntries( + block.raw, + writer, + deletes, + reusableIdentifier, + fullCompaction, + matchedEntries, + emittedDeletes, + metadata, + plan.encodedRecordsCompatible); + } + } + } + } else { + // Otherwise keep the streaming path: add-only manifests already use raw copying, + // and a single worker or manifest provides no concurrency. Processing one manifest + // at a time also avoids retaining planned raw blocks unnecessarily. + for (ManifestFileMeta manifest : manifests) { + try (ManifestAvroReader reader = + manifestFile.scanAvroBlocks(manifest.fileName(), manifest.fileSize())) { + boolean encodedRecordsCompatible = reader.rawBlockCopySupported(); + if (fullCompaction && mustChange != null && !mustChange.test(manifest)) { + boolean rewritten = + rewriteOptionalManifest( + reader, + writer, + partitionType, + partitionStatsConverter, + partitions, + deletes, + reusableIdentifier, + matchedEntries, + emittedDeletes, + metadata, + encodedRecordsCompatible); + if (!rewritten) { + checkState( + unchangedManifests != null, + "Full compaction requires an unchanged manifest result."); + unchangedManifests.add(manifest); + } + continue; + } + if (!hasDeletes + && manifest.numDeletedFiles() == 0 + && encodedRecordsCompatible) { + writer.writeEncodedManifest(reader, manifest); + continue; + } + writeRemainingBlocks( + reader, + writer, + partitionType, + partitionStatsConverter, + partitions, + deletes, + reusableIdentifier, + fullCompaction, + matchedEntries, + emittedDeletes, + metadata, + encodedRecordsCompatible); + } + } + } + writer.close(); + return writer.result(); + } catch (Exception | Error failure) { + writer.abort(); + throw failure; + } finally { + reusableIdentifier.release(); + matchedEntries.release(); + emittedDeletes.release(); + } + } + + private static ManifestRewritePlan planManifestRewrite( + ManifestFileMeta manifest, + ManifestFile manifestFile, + RowType partitionType, + CollectedDeletes deletes) + throws Exception { + ReusableIdentifier reusableIdentifier = new ReusableIdentifier(); + PartitionDictionary partitions = new PartitionDictionary(); + SimpleStatsConverter partitionStatsConverter = new SimpleStatsConverter(partitionType); + try (ManifestAvroReader reader = + manifestFile.scanAvroBlocks(manifest.fileName(), manifest.fileSize())) { + boolean encodedRecordsCompatible = reader.rawBlockCopySupported(); + List blocks = new ArrayList<>(); + while (reader.hasNext()) { + RawBlock raw = reader.next(); + blocks.add( + new PlannedBlock( + raw.stableCopy(), + inspectBlock( + raw, + partitionType, + partitionStatsConverter, + partitions, + deletes, + reusableIdentifier, + encodedRecordsCompatible))); + } + return new ManifestRewritePlan(manifest, encodedRecordsCompatible, blocks); + } finally { + reusableIdentifier.release(); + } + } + + private static boolean rewriteOptionalManifest( + ManifestAvroReader reader, + ManifestAvroWriter writer, + RowType partitionType, + SimpleStatsConverter partitionStatsConverter, + PartitionDictionary partitions, + CollectedDeletes deletes, + ReusableIdentifier reusableIdentifier, + CompactFileIdentifierSet matchedEntries, + CompactFileIdentifierSet emittedDeletes, + EncodedEntry metadata, + boolean encodedRecordsCompatible) + throws Exception { + List pendingBlocks = new ArrayList<>(); + List pendingMetadata = new ArrayList<>(); + while (reader.hasNext()) { + RawBlock rawBlock = reader.next(); + CompactionBlock block = + inspectBlock( + rawBlock, + partitionType, + partitionStatsConverter, + partitions, + deletes, + reusableIdentifier, + encodedRecordsCompatible); + if (block.unchanged) { + pendingBlocks.add(rawBlock.stableCopy()); + pendingMetadata.add(block.metadata); + continue; + } + + for (int i = 0; i < pendingBlocks.size(); i++) { + RawBlock pending = pendingBlocks.get(i); + EncodedBlockMeta blockMetadata = pendingMetadata.get(i); + if (blockMetadata == null) { + writeBlockEntries( + pending, + writer, + deletes, + reusableIdentifier, + true, + matchedEntries, + emittedDeletes, + metadata, + encodedRecordsCompatible); + } else { + writer.writeEncodedBlock(pending.encodedBlock(), blockMetadata); + } + } + writeBlockEntries( + rawBlock, + writer, + deletes, + reusableIdentifier, + true, + matchedEntries, + emittedDeletes, + metadata, + encodedRecordsCompatible); + writeRemainingBlocks( + reader, + writer, + partitionType, + partitionStatsConverter, + partitions, + deletes, + reusableIdentifier, + true, + matchedEntries, + emittedDeletes, + metadata, + encodedRecordsCompatible); + return true; + } + return false; + } + + private static void writeRemainingBlocks( + ManifestAvroReader reader, + ManifestAvroWriter writer, + RowType partitionType, + SimpleStatsConverter partitionStatsConverter, + PartitionDictionary partitions, + CollectedDeletes deletes, + ReusableIdentifier reusableIdentifier, + boolean fullCompaction, + CompactFileIdentifierSet matchedEntries, + CompactFileIdentifierSet emittedDeletes, + EncodedEntry metadata, + boolean encodedRecordsCompatible) + throws Exception { + while (reader.hasNext()) { + RawBlock rawBlock = reader.next(); + if (encodedRecordsCompatible) { + CompactionBlock block = + inspectBlock( + rawBlock, + partitionType, + partitionStatsConverter, + partitions, + deletes, + reusableIdentifier, + true); + if (block.canCopyEncodedBlock()) { + writer.writeEncodedBlock(rawBlock.encodedBlock(), block.metadata); + continue; + } + } + + writeBlockEntries( + rawBlock, + writer, + deletes, + reusableIdentifier, + fullCompaction, + matchedEntries, + emittedDeletes, + metadata, + encodedRecordsCompatible); + } + } + + private static CompactionBlock inspectBlock( + RawBlock rawBlock, + RowType partitionType, + SimpleStatsConverter partitionStatsConverter, + PartitionDictionary partitions, + CollectedDeletes deletes, + ReusableIdentifier reusableIdentifier, + boolean encodedRecordsCompatible) + throws Exception { + boolean deferDeletedAddCheck = encodedRecordsCompatible && deletes.useRowIdFilter(); + CompactionBlock block = new CompactionBlock(encodedRecordsCompatible, partitionType); + RowIterator rows = + rawBlock.toRows(ProjectedManifestEntry.ENTRY_LAYOUT_PROJECTION.projectedType()); + ProjectedManifestEntry entry = ProjectedManifestEntry.ENTRY_LAYOUT_PROJECTION.createEntry(); + while (rows.hasNext()) { + entry.replace(rows.next()); + if (!block.collect( + entry, deletes, reusableIdentifier, partitions, deferDeletedAddCheck)) { + break; + } + } + block.finish(partitionStatsConverter, partitions); + block.finishFiltering(deletes, deferDeletedAddCheck); + return block; + } + + private static void writeBlockEntries( + RawBlock rawBlock, + ManifestAvroWriter writer, + CollectedDeletes deletes, + ReusableIdentifier reusableIdentifier, + boolean fullCompaction, + CompactFileIdentifierSet matchedEntries, + CompactFileIdentifierSet emittedDeletes, + EncodedEntry metadata, + boolean encodedRecordsCompatible) + throws Exception { + ProjectedManifestEntry.Projection projection = + (encodedRecordsCompatible + ? ProjectedManifestEntry.ENTRY_LAYOUT_PROJECTION + : ProjectedManifestEntry.fullProjection()); + RowIterator rows = rawBlock.toRows(projection.projectedType()); + ProjectedManifestEntry entry = projection.createEntry(); + while (rows.hasNext()) { + GenericRow sourceRow = rows.next(); + entry.replace(sourceRow); + if (fullCompaction) { + if (entry.isAdd() && !deletes.isDeleted(entry, reusableIdentifier)) { + writeCompactedEntry( + writer, rows, sourceRow, entry, metadata, encodedRecordsCompatible); + } + } else if (entry.isAdd()) { + if (deletes.isDeleted(entry, reusableIdentifier)) { + matchedEntries.add(reusableIdentifier.replaceWithPartition(entry)); + } else { + writeCompactedEntry( + writer, rows, sourceRow, entry, metadata, encodedRecordsCompatible); + } + } else { + ReusableIdentifier identifier = reusableIdentifier.replaceWithPartition(entry); + if (!matchedEntries.contains(identifier) && !emittedDeletes.contains(identifier)) { + emittedDeletes.add(identifier); + writeCompactedEntry( + writer, rows, sourceRow, entry, metadata, encodedRecordsCompatible); + } + } + } + } + + private static void writeCompactedEntry( + ManifestAvroWriter writer, + RowIterator rows, + GenericRow sourceRow, + ProjectedManifestEntry entry, + EncodedEntry metadata, + boolean encodedRecordsCompatible) + throws Exception { + ProjectedDataFileMeta file = entry.file(); + BinaryRow partition = entry.partition(); + if (file.hasFirstRowId()) { + metadata.replace( + entry.kind().toByteValue(), + partition, + entry.bucket(), + file.level(), + file.schemaId(), + file.nonNullFirstRowId(), + file.rowCount()); + } else { + metadata.replace( + entry.kind().toByteValue(), + partition, + entry.bucket(), + file.level(), + file.schemaId(), + file.rowCount()); + } + if (encodedRecordsCompatible) { + writer.writeEncoded(rows.encodedRecord(), metadata); + } else { + writer.writeRow(sourceRow, metadata); + } + } + + /** Aggregate metadata for one raw Avro block considered by ordinary manifest compaction. */ + private static final class CompactionBlock { + + private boolean unchanged; + private long addedFiles; + private long deletedFiles; + private long schemaId = Long.MIN_VALUE; + private int minBucket = Integer.MAX_VALUE; + private int maxBucket = Integer.MIN_VALUE; + private int minLevel = Integer.MAX_VALUE; + private int maxLevel = Integer.MIN_VALUE; + private long minRowId = Long.MAX_VALUE; + private long maxRowId = Long.MIN_VALUE; + private boolean hasRowIds = true; + private final RowType partitionType; + private final boolean collectMetadata; + private @Nullable Map partitionCounts; + private @Nullable EncodedBlockMeta metadata; + + private CompactionBlock(boolean collectMetadata, RowType partitionType) { + this.unchanged = true; + this.partitionType = partitionType; + this.collectMetadata = collectMetadata; + this.partitionCounts = collectMetadata ? new HashMap<>() : null; + } + + private boolean collect( + ProjectedManifestEntry entry, + CollectedDeletes deletes, + ReusableIdentifier reusableIdentifier, + PartitionDictionary partitions, + boolean deferDeletedAddCheck) { + if (!unchanged) { + return false; + } + if (!deletes.copyable(entry, reusableIdentifier, deferDeletedAddCheck)) { + unchanged = false; + partitionCounts = null; + return false; + } + if (!collectMetadata) { + return true; + } + + checkState(partitionCounts != null, "Partition counts have already been released."); + ProjectedDataFileMeta file = entry.file(); + if (entry.isAdd()) { + addedFiles++; + } else { + deletedFiles++; + } + schemaId = Math.max(schemaId, file.schemaId()); + int bucket = entry.bucket(); + minBucket = Math.min(minBucket, bucket); + maxBucket = Math.max(maxBucket, bucket); + int level = file.level(); + minLevel = Math.min(minLevel, level); + maxLevel = Math.max(maxLevel, level); + if (hasRowIds) { + if (file.hasFirstRowId()) { + long firstRowId = file.nonNullFirstRowId(); + minRowId = Math.min(minRowId, firstRowId); + maxRowId = Math.max(maxRowId, firstRowId + file.rowCount() - 1L); + } else { + hasRowIds = false; + } + } + partitionCounts.merge(partitions.id(entry.partitionBytes()), 1, Integer::sum); + return true; + } + + private void finish( + SimpleStatsConverter partitionStatsConverter, PartitionDictionary partitions) { + if (!unchanged || !collectMetadata) { + return; + } + + checkState(partitionCounts != null, "Partition counts have already been released."); + SimpleStatsCollector collector = new SimpleStatsCollector(partitionType); + long[] nullCounts = new long[partitionType.getFieldCount()]; + for (Map.Entry entry : partitionCounts.entrySet()) { + BinaryRow partition = partitions.partition(entry.getKey()); + collector.collect(partition); + for (int field = 0; field < nullCounts.length; field++) { + if (partition.isNullAt(field)) { + nullCounts[field] = Math.addExact(nullCounts[field], entry.getValue()); + } + } + } + SimpleColStats[] stats = collector.extract(); + for (int field = 0; field < stats.length; field++) { + stats[field] = + new SimpleColStats( + stats[field].min(), stats[field].max(), nullCounts[field]); + } + metadata = + new EncodedBlockMeta( + addedFiles, + deletedFiles, + schemaId, + minBucket, + maxBucket, + minLevel, + maxLevel, + hasRowIds ? minRowId : -1, + hasRowIds ? maxRowId : -1, + partitionStatsConverter.toBinaryAllMode(stats)); + partitionCounts = null; + } + + private void finishFiltering(CollectedDeletes deletes, boolean deferDeletedAddCheck) { + if (deferDeletedAddCheck && metadata != null) { + checkState(hasRowIds, "RowID filtering requires block RowID statistics."); + if (!deletes.intersectsRowIds(minRowId, maxRowId)) { + return; + } + metadata = null; + unchanged = false; + } + } + + private boolean canCopyEncodedBlock() { + return metadata != null; + } + } + + private static final class PlannedBlock { + + private final RawBlock raw; + private final CompactionBlock compaction; + + private PlannedBlock(RawBlock raw, CompactionBlock compaction) { + this.raw = raw; + this.compaction = compaction; + } + } + + private static final class ManifestRewritePlan { + + private final ManifestFileMeta manifest; + private final boolean encodedRecordsCompatible; + private final List blocks; + + private ManifestRewritePlan( + ManifestFileMeta manifest, + boolean encodedRecordsCompatible, + List blocks) { + this.manifest = manifest; + this.encodedRecordsCompatible = encodedRecordsCompatible; + this.blocks = blocks; + } + + private boolean unchanged() { + for (PlannedBlock block : blocks) { + if (!block.compaction.unchanged) { + return false; + } + } + return true; + } + } +} diff --git a/paimon-core/src/main/java/org/apache/paimon/operation/ManifestFileLegacyMerger.java b/paimon-core/src/main/java/org/apache/paimon/operation/ManifestFileLegacyMerger.java new file mode 100644 index 000000000000..12c633252634 --- /dev/null +++ b/paimon-core/src/main/java/org/apache/paimon/operation/ManifestFileLegacyMerger.java @@ -0,0 +1,290 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.paimon.operation; + +import org.apache.paimon.CoreOptions; +import org.apache.paimon.data.BinaryRow; +import org.apache.paimon.manifest.FileEntry; +import org.apache.paimon.manifest.ManifestAvroWriter; +import org.apache.paimon.manifest.ManifestEntry; +import org.apache.paimon.manifest.ManifestFile; +import org.apache.paimon.manifest.ManifestFileMeta; +import org.apache.paimon.partition.PartitionPredicate; +import org.apache.paimon.types.RowType; +import org.apache.paimon.utils.Filter; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import javax.annotation.Nullable; + +import java.util.ArrayList; +import java.util.HashSet; +import java.util.Iterator; +import java.util.LinkedHashMap; +import java.util.LinkedList; +import java.util.List; +import java.util.Map; +import java.util.Optional; +import java.util.Set; +import java.util.function.Function; + +import static java.util.Collections.singletonList; +import static org.apache.paimon.utils.ManifestReadThreadPool.sequentialBatchedExecute; +import static org.apache.paimon.utils.Preconditions.checkArgument; + +/** Legacy full-entry manifest merger used when optimized manifest merging is disabled. */ +final class ManifestFileLegacyMerger { + + private static final Logger LOG = LoggerFactory.getLogger(ManifestFileLegacyMerger.class); + + private ManifestFileLegacyMerger() {} + + static List merge( + List input, + List newFilesForAbort, + ManifestFile manifestFile, + RowType partitionType, + CoreOptions options) + throws Exception { + long suggestedMetaSize = options.manifestTargetSize().getBytes(); + Integer manifestReadParallelism = options.scanManifestParallelism(); + Optional> fullCompacted = + tryFullCompaction( + input, + newFilesForAbort, + manifestFile, + suggestedMetaSize, + options.manifestFullCompactionThresholdSize().getBytes(), + partitionType, + manifestReadParallelism); + if (fullCompacted.isPresent()) { + return fullCompacted.get(); + } + return compactMinor( + input, + newFilesForAbort, + manifestFile, + suggestedMetaSize, + options.manifestMergeMinCount(), + manifestReadParallelism); + } + + private static List compactMinor( + List input, + List newFilesForAbort, + ManifestFile manifestFile, + long suggestedMetaSize, + int suggestedMinMetaCount, + @Nullable Integer manifestReadParallelism) { + List result = new ArrayList<>(); + List candidates = new ArrayList<>(); + long totalSize = 0; + for (ManifestFileMeta manifest : input) { + totalSize += manifest.fileSize(); + candidates.add(manifest); + if (totalSize >= suggestedMetaSize) { + mergeCandidates( + candidates, + manifestFile, + result, + newFilesForAbort, + manifestReadParallelism); + candidates.clear(); + totalSize = 0; + } + } + + if (candidates.size() >= suggestedMinMetaCount) { + mergeCandidates( + candidates, manifestFile, result, newFilesForAbort, manifestReadParallelism); + } else { + result.addAll(candidates); + } + return result; + } + + private static void mergeCandidates( + List candidates, + ManifestFile manifestFile, + List result, + List newMetas, + @Nullable Integer manifestReadParallelism) { + if (candidates.size() == 1) { + result.add(candidates.get(0)); + return; + } + + Map map = new LinkedHashMap<>(); + FileEntry.mergeEntries(manifestFile, candidates, map, manifestReadParallelism); + if (!map.isEmpty()) { + List merged = manifestFile.write(new ArrayList<>(map.values())); + result.addAll(merged); + newMetas.addAll(merged); + } + } + + static Optional> tryFullCompaction( + List inputs, + List newFilesForAbort, + ManifestFile manifestFile, + long suggestedMetaSize, + long sizeTrigger, + RowType partitionType, + @Nullable Integer manifestReadParallelism) + throws Exception { + checkArgument(sizeTrigger > 0, "Manifest full compaction size trigger cannot be zero."); + + Filter mustChange = + file -> file.numDeletedFiles() > 0 || file.fileSize() < suggestedMetaSize; + long totalManifestSize = 0; + long deltaDeleteFileNum = 0; + long totalDeltaFileSize = 0; + for (ManifestFileMeta file : inputs) { + totalManifestSize += file.fileSize(); + if (mustChange.test(file)) { + totalDeltaFileSize += file.fileSize(); + deltaDeleteFileNum += file.numDeletedFiles(); + } + } + + if (totalDeltaFileSize < sizeTrigger) { + return Optional.empty(); + } + + LOG.info( + "Start Legacy Manifest File Full Compaction: totalManifestSize: {}, deltaDeleteFileNum {}, totalDeltaFileSize {}", + totalManifestSize, + deltaDeleteFileNum, + totalDeltaFileSize); + + Set deleteEntries = + FileEntry.readDeletedEntries(manifestFile, inputs, manifestReadParallelism); + + PartitionPredicate predicate; + if (deleteEntries.isEmpty()) { + predicate = PartitionPredicate.ALWAYS_FALSE; + } else if (partitionType.getFieldCount() > 0) { + predicate = + PartitionPredicate.fromMultiple( + partitionType, computeDeletePartitions(deleteEntries)); + } else { + predicate = PartitionPredicate.ALWAYS_TRUE; + } + + List result = new ArrayList<>(); + List toBeMerged = new LinkedList<>(inputs); + if (predicate != null) { + Iterator iterator = toBeMerged.iterator(); + while (iterator.hasNext()) { + ManifestFileMeta file = iterator.next(); + if (mustChange.test(file)) { + continue; + } + if (!predicate.test( + file.numAddedFiles() + file.numDeletedFiles(), + file.partitionStats().minValues(), + file.partitionStats().maxValues(), + file.partitionStats().nullCounts())) { + iterator.remove(); + result.add(file); + } + } + } + + if (toBeMerged.size() <= 1) { + return Optional.empty(); + } + + ManifestAvroWriter writer = manifestFile.createAvroWriter(); + Function> reader = + file -> + singletonList( + readForFullCompaction( + file, manifestFile, mustChange, deleteEntries)); + Exception exception = null; + try { + for (FullCompactionReadResult readResult : + sequentialBatchedExecute(reader, toBeMerged, manifestReadParallelism)) { + if (readResult.requireChange) { + writer.write(readResult.entries); + } else { + result.add(readResult.file); + } + } + } catch (Exception e) { + exception = e; + } finally { + if (exception != null) { + writer.abort(); + throw exception; + } + writer.close(); + } + + List merged = writer.result(); + result.addAll(merged); + newFilesForAbort.addAll(merged); + return Optional.of(result); + } + + private static FullCompactionReadResult readForFullCompaction( + ManifestFileMeta file, + ManifestFile manifestFile, + Filter mustChange, + Set deleteEntries) { + List entries = new ArrayList<>(); + boolean requireChange = mustChange.test(file); + for (ManifestEntry entry : + manifestFile.read( + file.fileName(), + file.fileSize(), + FileEntry.addFilter(), + Filter.alwaysTrue())) { + if (deleteEntries.contains(entry.identifier())) { + requireChange = true; + } else { + entries.add(entry); + } + } + return new FullCompactionReadResult(file, requireChange, entries); + } + + private static Set computeDeletePartitions(Set deleteEntries) { + Set partitions = new HashSet<>(); + for (FileEntry.Identifier identifier : deleteEntries) { + partitions.add(identifier.partition); + } + return partitions; + } + + private static final class FullCompactionReadResult { + + private final ManifestFileMeta file; + private final boolean requireChange; + private final List entries; + + private FullCompactionReadResult( + ManifestFileMeta file, boolean requireChange, List entries) { + this.file = file; + this.requireChange = requireChange; + this.entries = entries; + } + } +} diff --git a/paimon-core/src/main/java/org/apache/paimon/operation/ManifestFileMerger.java b/paimon-core/src/main/java/org/apache/paimon/operation/ManifestFileMerger.java index 7c5019f1e33c..b57666563a36 100644 --- a/paimon-core/src/main/java/org/apache/paimon/operation/ManifestFileMerger.java +++ b/paimon-core/src/main/java/org/apache/paimon/operation/ManifestFileMerger.java @@ -19,43 +19,23 @@ package org.apache.paimon.operation; import org.apache.paimon.CoreOptions; -import org.apache.paimon.data.BinaryRow; import org.apache.paimon.disk.IOManager; -import org.apache.paimon.manifest.FileEntry; -import org.apache.paimon.manifest.ManifestAvroWriter; import org.apache.paimon.manifest.ManifestEntry; import org.apache.paimon.manifest.ManifestFile; import org.apache.paimon.manifest.ManifestFileMeta; -import org.apache.paimon.partition.PartitionPredicate; import org.apache.paimon.types.RowType; -import org.apache.paimon.utils.Filter; - -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; import javax.annotation.Nullable; import java.util.ArrayList; -import java.util.HashSet; -import java.util.Iterator; -import java.util.LinkedHashMap; -import java.util.LinkedList; import java.util.List; -import java.util.Map; import java.util.Optional; -import java.util.Set; -import java.util.function.Function; -import static java.util.Collections.singletonList; import static org.apache.paimon.manifest.ManifestFileMeta.allContainsRowId; -import static org.apache.paimon.utils.ManifestReadThreadPool.sequentialBatchedExecute; -import static org.apache.paimon.utils.Preconditions.checkArgument; /** Manifest file merger with standard merge logic and optional sort rewrite. */ public class ManifestFileMerger { - private static final Logger LOG = LoggerFactory.getLogger(ManifestFileMerger.class); - /** * Merge several {@link ManifestFileMeta}s. {@link ManifestEntry}s representing first adding and * then deleting the same data file will cancel each other. @@ -76,12 +56,6 @@ public static List merge( RowType partitionType, CoreOptions options, @Nullable IOManager ioManager) { - // Extract configuration from options - long suggestedMetaSize = options.manifestTargetSize().getBytes(); - int suggestedMinMetaCount = options.manifestMergeMinCount(); - long manifestFullCompactionSize = options.manifestFullCompactionThresholdSize().getBytes(); - Integer manifestReadParallelism = options.scanManifestParallelism(); - // these are the newly created manifest files, clean them up if exception occurs List newFilesForAbort = new ArrayList<>(); @@ -94,27 +68,14 @@ public static List merge( || (options.dataEvolutionEnabled() && allContainsRowId(input)))) { return ManifestFileSorter.trySortCompaction( input, newFilesForAbort, manifestFile, partitionType, options, ioManager); - } else { - // Otherwise try full compaction first, then minor compaction if needed - Optional> fullCompacted = - tryFullCompaction( - input, - newFilesForAbort, - manifestFile, - suggestedMetaSize, - manifestFullCompactionSize, - partitionType, - manifestReadParallelism); - return fullCompacted.orElseGet( - () -> - tryMinorCompaction( - input, - newFilesForAbort, - manifestFile, - suggestedMetaSize, - suggestedMinMetaCount, - manifestReadParallelism)); } + + if (options.manifestMergeOptimizeEnabled()) { + return ManifestFileBlockMerger.merge( + input, newFilesForAbort, manifestFile, partitionType, options); + } + return ManifestFileLegacyMerger.merge( + input, newFilesForAbort, manifestFile, partitionType, options); } catch (Throwable e) { // exception occurs, clean up and rethrow for (ManifestFileMeta manifest : newFilesForAbort) { @@ -124,63 +85,6 @@ public static List merge( } } - private static List tryMinorCompaction( - List input, - List newFilesForAbort, - ManifestFile manifestFile, - long suggestedMetaSize, - int suggestedMinMetaCount, - @Nullable Integer manifestReadParallelism) { - List result = new ArrayList<>(); - List candidates = new ArrayList<>(); - long totalSize = 0; - // merge existing small manifest files - for (ManifestFileMeta manifest : input) { - totalSize += manifest.fileSize(); - candidates.add(manifest); - if (totalSize >= suggestedMetaSize) { - // reach suggested file size, perform merging and produce new file - mergeCandidates( - candidates, - manifestFile, - result, - newFilesForAbort, - manifestReadParallelism); - candidates.clear(); - totalSize = 0; - } - } - - // merge the last bit of manifests if there are too many - if (candidates.size() >= suggestedMinMetaCount) { - mergeCandidates( - candidates, manifestFile, result, newFilesForAbort, manifestReadParallelism); - } else { - result.addAll(candidates); - } - return result; - } - - private static void mergeCandidates( - List candidates, - ManifestFile manifestFile, - List result, - List newMetas, - @Nullable Integer manifestReadParallelism) { - if (candidates.size() == 1) { - result.add(candidates.get(0)); - return; - } - - Map map = new LinkedHashMap<>(); - FileEntry.mergeEntries(manifestFile, candidates, map, manifestReadParallelism); - if (!map.isEmpty()) { - List merged = manifestFile.write(new ArrayList<>(map.values())); - result.addAll(merged); - newMetas.addAll(merged); - } - } - public static Optional> tryFullCompaction( List inputs, List newFilesForAbort, @@ -190,154 +94,13 @@ public static Optional> tryFullCompaction( RowType partitionType, @Nullable Integer manifestReadParallelism) throws Exception { - checkArgument(sizeTrigger > 0, "Manifest full compaction size trigger cannot be zero."); - - // 1. should trigger full compaction - - Filter mustChange = - file -> file.numDeletedFiles() > 0 || file.fileSize() < suggestedMetaSize; - long totalManifestSize = 0; - long deltaDeleteFileNum = 0; - long totalDeltaFileSize = 0; - for (ManifestFileMeta file : inputs) { - totalManifestSize += file.fileSize(); - if (mustChange.test(file)) { - totalDeltaFileSize += file.fileSize(); - deltaDeleteFileNum += file.numDeletedFiles(); - } - } - - if (totalDeltaFileSize < sizeTrigger) { - return Optional.empty(); - } - - // 2. do full compaction - - LOG.info( - "Start Manifest File Full Compaction: totalManifestSize: {}, deltaDeleteFileNum {}, totalDeltaFileSize {}", - totalManifestSize, - deltaDeleteFileNum, - totalDeltaFileSize); - - // 2.1. read all delete entries - - Set deleteEntries = - FileEntry.readDeletedEntries(manifestFile, inputs, manifestReadParallelism); - - // 2.2. try to skip base files by partition filter - - PartitionPredicate predicate; - if (deleteEntries.isEmpty()) { - predicate = PartitionPredicate.ALWAYS_FALSE; - } else { - if (partitionType.getFieldCount() > 0) { - Set deletePartitions = computeDeletePartitions(deleteEntries); - predicate = PartitionPredicate.fromMultiple(partitionType, deletePartitions); - } else { - predicate = PartitionPredicate.ALWAYS_TRUE; - } - } - - List result = new ArrayList<>(); - List toBeMerged = new LinkedList<>(inputs); - - if (predicate != null) { - Iterator iterator = toBeMerged.iterator(); - while (iterator.hasNext()) { - ManifestFileMeta file = iterator.next(); - if (mustChange.test(file)) { - continue; - } - if (!predicate.test( - file.numAddedFiles() + file.numDeletedFiles(), - file.partitionStats().minValues(), - file.partitionStats().maxValues(), - file.partitionStats().nullCounts())) { - iterator.remove(); - result.add(file); - } - } - } - - // 2.2. merge - if (toBeMerged.size() <= 1) { - return Optional.empty(); - } - - ManifestAvroWriter writer = manifestFile.createAvroWriter(); - Function> reader = - file -> - singletonList( - readForFullCompaction( - file, manifestFile, mustChange, deleteEntries)); - Exception exception = null; - try { - for (FullCompactionReadResult readResult : - sequentialBatchedExecute(reader, toBeMerged, manifestReadParallelism)) { - if (readResult.requireChange) { - writer.write(readResult.entries); - } else { - result.add(readResult.file); - } - } - } catch (Exception e) { - exception = e; - } finally { - if (exception != null) { - writer.abort(); - throw exception; - } - writer.close(); - } - - List merged = writer.result(); - result.addAll(merged); - newFilesForAbort.addAll(merged); - return Optional.of(result); - } - - private static FullCompactionReadResult readForFullCompaction( - ManifestFileMeta file, - ManifestFile manifestFile, - Filter mustChange, - Set deleteEntries) { - List entries = new ArrayList<>(); - boolean requireChange = mustChange.test(file); - for (ManifestEntry entry : - manifestFile.read( - file.fileName(), - file.fileSize(), - FileEntry.addFilter(), - Filter.alwaysTrue())) { - if (deleteEntries.contains(entry.identifier())) { - requireChange = true; - } else { - entries.add(entry); - } - } - - return new FullCompactionReadResult(file, requireChange, entries); - } - - static Set computeDeletePartitions(Set deleteEntries) { - Set partitions = new HashSet<>(); - for (FileEntry.Identifier identifier : deleteEntries) { - partitions.add(identifier.partition); - } - return partitions; - } - - static class FullCompactionReadResult { - - final ManifestFileMeta file; - final boolean requireChange; - final List entries; - - FullCompactionReadResult( - ManifestFileMeta file, boolean requireChange, List entries) { - this.file = file; - this.requireChange = requireChange; - this.entries = entries; - } + return ManifestFileBlockMerger.tryFullCompaction( + inputs, + newFilesForAbort, + manifestFile, + suggestedMetaSize, + sizeTrigger, + partitionType, + manifestReadParallelism); } } diff --git a/paimon-core/src/main/java/org/apache/paimon/operation/commit/DataEvolutionConflictDetection.java b/paimon-core/src/main/java/org/apache/paimon/operation/commit/DataEvolutionConflictDetection.java index bfbe8a4f5778..3ab5c67c5bca 100644 --- a/paimon-core/src/main/java/org/apache/paimon/operation/commit/DataEvolutionConflictDetection.java +++ b/paimon-core/src/main/java/org/apache/paimon/operation/commit/DataEvolutionConflictDetection.java @@ -591,6 +591,12 @@ Optional checkRowIdExistence( List deltaEntries, @Nullable Long nextRowId, CommitKind commitKind) { + Optional exception = + checkDeletedFileRowIdExistence(baseEntries, deltaEntries); + if (exception.isPresent()) { + return exception; + } + List existingDataFiles = baseEntries.stream() .filter( @@ -605,6 +611,42 @@ Optional checkRowIdExistence( return checkNonCompactRowIdExistence(existingDataFiles, deltaEntries, nextRowId); } + private Optional checkDeletedFileRowIdExistence( + List baseEntries, List deltaEntries) { + // FileEntry.Identifier deliberately excludes RowID metadata because reassignment does not + // create a new physical data file. Reject a DELETE planned before reassignment by comparing + // it with the current ADD before identifier-based manifest merging cancels the two entries. + Map deletedFiles = new HashMap<>(); + for (SimpleFileEntry entry : deltaEntries) { + if (entry.kind() == FileKind.DELETE) { + deletedFiles.put(entry.identifier(), entry); + } + } + if (deletedFiles.isEmpty()) { + return Optional.empty(); + } + + for (SimpleFileEntry current : baseEntries) { + if (current.kind() != FileKind.ADD) { + continue; + } + SimpleFileEntry deleted = deletedFiles.get(current.identifier()); + if (deleted != null + && (!Objects.equals(current.firstRowId(), deleted.firstRowId()) + || current.rowCount() != deleted.rowCount())) { + return Optional.of( + new RowIdExistenceConflictException( + deleted.fileName(), + deleted.firstRowId(), + deleted.rowCount(), + current.firstRowId(), + current.rowCount(), + deleted.bucket())); + } + } + return Optional.empty(); + } + /** * Checks conflicts between compaction and concurrent Row ID reassignment, which may otherwise * cause reassigned Row IDs to fall back. For example, compaction produces a file with Row IDs diff --git a/paimon-core/src/main/java/org/apache/paimon/operation/commit/RowIdExistenceConflictException.java b/paimon-core/src/main/java/org/apache/paimon/operation/commit/RowIdExistenceConflictException.java index 4b14f928c2f5..2a1fce8de8dc 100644 --- a/paimon-core/src/main/java/org/apache/paimon/operation/commit/RowIdExistenceConflictException.java +++ b/paimon-core/src/main/java/org/apache/paimon/operation/commit/RowIdExistenceConflictException.java @@ -18,6 +18,8 @@ package org.apache.paimon.operation.commit; +import javax.annotation.Nullable; + /** Conflict caused by a staged file referencing a row-id range absent from the latest snapshot. */ public final class RowIdExistenceConflictException extends RuntimeException { @@ -31,4 +33,25 @@ public final class RowIdExistenceConflictException extends RuntimeException { + "concurrent compaction or removed by an overwrite.", fileName, firstRowId, rowCount, bucket)); } + + RowIdExistenceConflictException( + String fileName, + @Nullable Long deletedFirstRowId, + long deletedRowCount, + @Nullable Long currentFirstRowId, + long currentRowCount, + int bucket) { + super( + String.format( + "Row ID existence conflict: DELETE for file '%s' references " + + "firstRowId=%s, rowCount=%d in bucket %d, but the current file " + + "has firstRowId=%s, rowCount=%d. The file may have been " + + "reassigned by a concurrent commit.", + fileName, + deletedFirstRowId, + deletedRowCount, + bucket, + currentFirstRowId, + currentRowCount)); + } } diff --git a/paimon-core/src/test/java/org/apache/paimon/manifest/CompactFileIdentifierSetTest.java b/paimon-core/src/test/java/org/apache/paimon/manifest/CompactFileIdentifierSetTest.java index 95d071eabee3..7eeab708411d 100644 --- a/paimon-core/src/test/java/org/apache/paimon/manifest/CompactFileIdentifierSetTest.java +++ b/paimon-core/src/test/java/org/apache/paimon/manifest/CompactFileIdentifierSetTest.java @@ -79,6 +79,25 @@ void testGrowsAndReleases() { assertThat(identifiers.contains(1, new byte[] {1}, 1)).isTrue(); } + @Test + void testAddAll() { + CompactFileIdentifierSet first = new CompactFileIdentifierSet(); + first.add(1, new byte[] {1, 2}, 2); + first.add(2, new byte[] {3, 4, 5}, 3); + + CompactFileIdentifierSet second = new CompactFileIdentifierSet(); + second.add(2, new byte[] {3, 4, 5}, 3); + second.add(3, new byte[] {6, 7, 8, 9}, 4); + + first.addAll(second); + + assertThat(first.size()).isEqualTo(3); + assertThat(first.retainedIdentifierBytes()).isEqualTo(9); + assertThat(first.contains(1, new byte[] {1, 2}, 2)).isTrue(); + assertThat(first.contains(2, new byte[] {3, 4, 5}, 3)).isTrue(); + assertThat(first.contains(3, new byte[] {6, 7, 8, 9}, 4)).isTrue(); + } + @Test void testRejectsInvalidIdentifier() { CompactFileIdentifierSet identifiers = new CompactFileIdentifierSet(); 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 7dff697e8fc6..5383e5e8b4d8 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 @@ -48,8 +48,11 @@ import org.junit.jupiter.api.Test; import org.junit.jupiter.api.io.TempDir; import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.NullSource; import org.junit.jupiter.params.provider.ValueSource; +import javax.annotation.Nullable; + import java.io.IOException; import java.util.ArrayList; import java.util.Arrays; @@ -59,10 +62,12 @@ import java.util.Iterator; import java.util.LinkedList; import java.util.List; +import java.util.Map; import java.util.Optional; import java.util.Set; import java.util.TreeSet; import java.util.UUID; +import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.CountDownLatch; import java.util.concurrent.ThreadLocalRandom; import java.util.concurrent.TimeUnit; @@ -73,6 +78,7 @@ import static org.assertj.core.api.Assertions.assertThat; import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assumptions.assumeTrue; /** Tests for {@link ManifestFileMeta}. */ public class ManifestFileMetaTest extends ManifestFileMetaTestBase { @@ -236,6 +242,145 @@ public void testMergeWithoutDelta() { assertEquivalentEntries(input1, merged1); } + @ParameterizedTest + @ValueSource(booleans = {false, true}) + public void testAddOnlyCompactionCopiesRawBlocks(boolean fullCompaction) throws Exception { + List input = + Arrays.asList( + makeManifest(makeEntry(true, "a", 1), makeEntry(true, "b", 1)), + makeManifest(makeEntry(true, "c", 7)), + makeManifest(makeEntry(true, "d", 5), makeEntry(true, "e", 9))); + int inputBlocks = 0; + for (ManifestFileMeta meta : input) { + inputBlocks += rawBlockCount(manifestFile, meta); + } + + Options testOptions = new Options(); + testOptions.set("manifest.target-file-size", "1MB"); + testOptions.set("manifest.merge-min-count", "2"); + testOptions.set( + "manifest.full-compaction-threshold-size", + fullCompaction ? "1B" : Long.MAX_VALUE + "B"); + List merged = + ManifestFileMerger.merge( + input, + manifestFile, + getPartitionType(), + CoreOptions.fromMap(testOptions.toMap())); + + assertThat(merged).hasSize(1); + assertEquivalentEntries(input, merged); + ManifestFileMeta output = merged.get(0); + assertThat(rawBlockCount(manifestFile, output)).isEqualTo(inputBlocks); + assertThat(output.numAddedFiles()).isEqualTo(5); + assertThat(output.numDeletedFiles()).isZero(); + assertThat(output.partitionStats().minValues().getInt(0)).isEqualTo(1); + assertThat(output.partitionStats().maxValues().getInt(0)).isEqualTo(9); + assertThat(output.partitionStats().nullCounts().getLong(0)).isZero(); + assertThat(output.minRowId()).isNull(); + assertThat(output.maxRowId()).isNull(); + } + + @ParameterizedTest + @ValueSource(booleans = {false, true}) + public void testCompactionCopiesUnaffectedBlocksAroundDeletes(boolean fullCompaction) + throws Exception { + List input = + Arrays.asList( + makeManifest( + makeRowIdEntry(true, "deleted", 0, 0, 5), + makeRowIdEntry(true, "survivor-10", 0, 10, 5)), + makeManifest( + makeRowIdEntry(true, "survivor-100", 9, 100, 5), + makeRowIdEntry(true, "survivor-120", 1, 120, 5)), + makeManifest(makeRowIdEntry(false, "deleted", 0, 0, 5)), + makeManifest(makeRowIdEntry(true, "survivor-300", 0, 300, 5))); + + Options testOptions = new Options(); + testOptions.set("manifest.target-file-size", "1MB"); + testOptions.set("manifest.merge-min-count", "2"); + testOptions.set("scan.manifest.parallelism", "2"); + testOptions.set( + "manifest.full-compaction-threshold-size", + fullCompaction ? "1B" : Long.MAX_VALUE + "B"); + List merged = + ManifestFileMerger.merge( + input, + manifestFile, + getPartitionType(), + CoreOptions.fromMap(testOptions.toMap())); + + assertThat(merged).hasSize(1); + assertEquivalentEntries(input, merged); + ManifestFileMeta output = merged.get(0); + assertThat(rawBlockCount(manifestFile, output)).isEqualTo(3); + assertThat(output.numAddedFiles()).isEqualTo(4); + assertThat(output.numDeletedFiles()).isZero(); + assertThat(output.minRowId()).isEqualTo(10); + assertThat(output.maxRowId()).isEqualTo(304); + } + + @ParameterizedTest + @ValueSource(booleans = {false, true}) + public void testCompactionWithoutRowIdFiltersDeletes(boolean fullCompaction) { + List input = + Arrays.asList( + makeManifest(makeEntry(true, "deleted", 0), makeEntry(true, "survivor", 0)), + makeManifest(makeEntry(true, "other", 1)), + makeManifest(makeEntry(false, "deleted", 0))); + + Options testOptions = new Options(); + testOptions.set("manifest.target-file-size", "1MB"); + testOptions.set("manifest.merge-min-count", "2"); + testOptions.set("scan.manifest.parallelism", "2"); + testOptions.set( + "manifest.full-compaction-threshold-size", + fullCompaction ? "1B" : Long.MAX_VALUE + "B"); + List merged = + ManifestFileMerger.merge( + input, + manifestFile, + getPartitionType(), + CoreOptions.fromMap(testOptions.toMap())); + + assertThat(merged).hasSize(1); + assertEquivalentEntries(input, merged); + ManifestFileMeta output = merged.get(0); + assertThat(output.numAddedFiles()).isEqualTo(2); + assertThat(output.numDeletedFiles()).isZero(); + assertThat(output.minRowId()).isNull(); + assertThat(output.maxRowId()).isNull(); + } + + @Test + public void testDisablingManifestMergeOptimizeUsesLegacyMerger() throws Exception { + List input = + Arrays.asList( + makeManifest(makeEntry(true, "a", 0)), + makeManifest(makeEntry(true, "b", 1)), + makeManifest(makeEntry(true, "c", 2))); + int inputBlocks = 0; + for (ManifestFileMeta manifest : input) { + inputBlocks += rawBlockCount(manifestFile, manifest); + } + + Options testOptions = new Options(); + testOptions.set("manifest.merge-optimize.enabled", "false"); + testOptions.set("manifest.target-file-size", "1MB"); + testOptions.set("manifest.merge-min-count", "2"); + testOptions.set("manifest.full-compaction-threshold-size", Long.MAX_VALUE + "B"); + List merged = + ManifestFileMerger.merge( + input, + manifestFile, + getPartitionType(), + CoreOptions.fromMap(testOptions.toMap())); + + assertThat(merged).hasSize(1); + assertEquivalentEntries(input, merged); + assertThat(rawBlockCount(manifestFile, merged.get(0))).isLessThan(inputBlocks); + } + @Test public void testMergeWithoutBase() { List input = new ArrayList<>(); @@ -565,8 +710,8 @@ public void testMergeFullCompactionWithoutDeleteFile() { } @Test - public void testFullCompactionReadManifestsInParallel() throws Exception { - BlockingReadFileIO fileIO = new BlockingReadFileIO(); + public void testFullCompactionReadsSelectedManifestsOnce() throws Exception { + CountingReadFileIO fileIO = new CountingReadFileIO(); manifestFile = createManifestFile(tempDir.toString(), fileIO); List input = new ArrayList<>(); @@ -574,26 +719,157 @@ public void testFullCompactionReadManifestsInParallel() throws Exception { input.add(makeManifest(makeEntry(true, "parallel-" + i))); } - List newMetas = new ArrayList<>(); + fileIO.resetReadCounts(); + Optional> fullCompacted = + ManifestFileMerger.tryFullCompaction( + input, + new ArrayList<>(), + manifestFile, + Long.MAX_VALUE, + 1, + getPartitionType(), + 2); + + assertThat(fullCompacted).isPresent(); + for (ManifestFileMeta manifest : input) { + assertThat(fileIO.readCount(manifest.fileName())).isEqualTo(1); + } + assertEquivalentEntries(input, fullCompacted.get()); + } + + @Test + public void testFullCompactionCollectsDeletesWithDefaultParallelism() throws Exception { + assumeTrue(Runtime.getRuntime().availableProcessors() > 1); + CountingReadFileIO fileIO = new CountingReadFileIO(); + manifestFile = createManifestFile(tempDir.toString(), fileIO); + ManifestFileMeta base = + makeManifest( + makeRowIdEntry(true, "deleted-a", 0, 0, 5), + makeRowIdEntry(true, "deleted-b", 0, 10, 5), + makeRowIdEntry(true, "survivor", 0, 20, 5)); + ManifestFileMeta firstDelete = makeManifest(makeRowIdEntry(false, "deleted-a", 0, 0, 5)); + ManifestFileMeta secondDelete = makeManifest(makeRowIdEntry(false, "deleted-b", 0, 10, 5)); + + fileIO.blockManifestReads( + new HashSet<>(Arrays.asList(firstDelete.fileName(), secondDelete.fileName()))); Optional> fullCompacted; - fileIO.blockManifestReads(); try { fullCompacted = ManifestFileMerger.tryFullCompaction( - input, - newMetas, + Arrays.asList(base, firstDelete, secondDelete), + new ArrayList<>(), manifestFile, Long.MAX_VALUE, 1, getPartitionType(), - 2); + null); } finally { fileIO.stopBlockingManifestReads(); } assertThat(fileIO.maxConcurrentManifestReads()).isGreaterThanOrEqualTo(2); assertThat(fullCompacted).isPresent(); - assertEquivalentEntries(input, fullCompacted.get()); + assertThat(readEntries(fullCompacted.get()).stream().map(entry -> entry.file().fileName())) + .containsExactly("survivor"); + } + + @Test + public void testFullCompactionPlansRowIdManifestsWithDefaultParallelism() throws Exception { + assumeTrue(Runtime.getRuntime().availableProcessors() > 1); + CountingReadFileIO fileIO = new CountingReadFileIO(); + manifestFile = createManifestFile(tempDir.toString(), fileIO); + ManifestFileMeta first = + makeManifest( + makeRowIdEntry(true, "deleted", 0, 0, 5), + makeRowIdEntry(true, "survivor-a", 0, 10, 5)); + ManifestFileMeta second = makeManifest(makeRowIdEntry(true, "survivor-b", 0, 20, 5)); + ManifestFileMeta delta = makeManifest(makeRowIdEntry(false, "deleted", 0, 0, 5)); + + fileIO.blockManifestReads( + new HashSet<>(Arrays.asList(first.fileName(), second.fileName()))); + Optional> fullCompacted; + try { + fullCompacted = + ManifestFileMerger.tryFullCompaction( + Arrays.asList(first, second, delta), + new ArrayList<>(), + manifestFile, + Long.MAX_VALUE, + 1, + getPartitionType(), + null); + } finally { + fileIO.stopBlockingManifestReads(); + } + + assertThat(fileIO.maxConcurrentManifestReads()).isGreaterThanOrEqualTo(2); + assertThat(fullCompacted).isPresent(); + assertThat(readEntries(fullCompacted.get()).stream().map(entry -> entry.file().fileName())) + .containsExactlyInAnyOrder("survivor-a", "survivor-b"); + } + + @ParameterizedTest + @NullSource + @ValueSource(ints = 2) + public void testFullCompactionPlansIdentifierManifestsInParallel(@Nullable Integer parallelism) + throws Exception { + if (parallelism == null) { + assumeTrue(Runtime.getRuntime().availableProcessors() > 1); + } + CountingReadFileIO fileIO = new CountingReadFileIO(); + manifestFile = createManifestFile(tempDir.toString(), fileIO); + ManifestFileMeta first = + makeManifest(makeEntry(true, "deleted", 0), makeEntry(true, "survivor-a", 0)); + ManifestFileMeta second = makeManifest(makeEntry(true, "survivor-b", 0)); + ManifestFileMeta delta = makeManifest(makeEntry(false, "deleted", 0)); + + fileIO.blockManifestReads( + new HashSet<>(Arrays.asList(first.fileName(), second.fileName()))); + Optional> fullCompacted; + try { + fullCompacted = + ManifestFileMerger.tryFullCompaction( + Arrays.asList(first, second, delta), + new ArrayList<>(), + manifestFile, + 1, + 1, + getPartitionType(), + parallelism); + } finally { + fileIO.stopBlockingManifestReads(); + } + + assertThat(fileIO.maxConcurrentManifestReads()).isGreaterThanOrEqualTo(2); + assertThat(fullCompacted).isPresent(); + assertThat(readEntries(fullCompacted.get()).stream().map(entry -> entry.file().fileName())) + .containsExactlyInAnyOrder("survivor-a", "survivor-b"); + } + + @Test + public void testFullCompactionReadsDeleteCandidateOnce() throws Exception { + CountingReadFileIO fileIO = new CountingReadFileIO(); + manifestFile = createManifestFile(tempDir.toString(), fileIO); + ManifestFileMeta base = + makeManifest(makeEntry(true, "deleted", 0), makeEntry(true, "survivor", 0)); + ManifestFileMeta delta = makeManifest(makeEntry(false, "deleted", 0)); + fileIO.resetReadCounts(); + + Optional> fullCompacted = + ManifestFileMerger.tryFullCompaction( + Arrays.asList(base, delta), + new ArrayList<>(), + manifestFile, + 1, + 1, + getPartitionType(), + null); + + assertThat(fullCompacted).isPresent(); + assertThat(fileIO.readCount(base.fileName())).isEqualTo(1); + assertThat(fileIO.readCount(delta.fileName())).isEqualTo(2); + assertThat(readEntries(fullCompacted.get()).stream().map(entry -> entry.file().fileName())) + .containsExactly("survivor"); } @RepeatedTest(10) @@ -795,15 +1071,39 @@ ManifestFile getManifestFile() { return manifestFile; } - private static class BlockingReadFileIO extends LocalFileIO { + private static class CountingReadFileIO extends LocalFileIO { - private final AtomicBoolean blockManifestReads = new AtomicBoolean(false); - private final AtomicInteger activeManifestReads = new AtomicInteger(0); - private final AtomicInteger maxConcurrentManifestReads = new AtomicInteger(0); + private final Map readCounts = new ConcurrentHashMap<>(); + private final AtomicInteger activeManifestReads = new AtomicInteger(); + private final AtomicInteger maxConcurrentManifestReads = new AtomicInteger(); private final CountDownLatch readersReady = new CountDownLatch(2); private final CountDownLatch releaseReaders = new CountDownLatch(1); + private final AtomicBoolean blockManifestReads = new AtomicBoolean(); + private volatile Set blockedManifests = Collections.emptySet(); - private void blockManifestReads() { + @Override + public SeekableInputStream newInputStream(Path path) throws IOException { + readCounts + .computeIfAbsent(path.getName(), ignored -> new AtomicInteger()) + .incrementAndGet(); + SeekableInputStream input = super.newInputStream(path); + if (!blockManifestReads.get() || !blockedManifests.contains(path.getName())) { + return input; + } + return new BlockingSeekableInputStream(input); + } + + private int readCount(String fileName) { + AtomicInteger count = readCounts.get(fileName); + return count == null ? 0 : count.get(); + } + + private void resetReadCounts() { + readCounts.clear(); + } + + private void blockManifestReads(Set manifests) { + blockedManifests = new HashSet<>(manifests); blockManifestReads.set(true); } @@ -816,15 +1116,6 @@ private int maxConcurrentManifestReads() { return maxConcurrentManifestReads.get(); } - @Override - public SeekableInputStream newInputStream(Path path) throws IOException { - SeekableInputStream inputStream = super.newInputStream(path); - if (!blockManifestReads.get() || !path.toString().contains("/manifest/")) { - return inputStream; - } - return new BlockingSeekableInputStream(inputStream); - } - private class BlockingSeekableInputStream extends SeekableInputStreamWrapper { private boolean entered; @@ -870,7 +1161,6 @@ private void beforeFirstRead() throws IOException { if (readersReady.getCount() == 0) { releaseReaders.countDown(); } - try { if (!releaseReaders.await(3, TimeUnit.SECONDS)) { throw new IOException("Manifest reads were not parallelized."); @@ -1960,4 +2250,16 @@ private List readEntries(List manifestMetas) { } return entries; } + + private int rawBlockCount(ManifestFile manifestFile, ManifestFileMeta meta) throws Exception { + int blocks = 0; + try (ManifestAvroReader reader = + manifestFile.scanAvroBlocks(meta.fileName(), meta.fileSize())) { + while (reader.hasNext()) { + reader.next(); + blocks++; + } + } + return blocks; + } } diff --git a/paimon-core/src/test/java/org/apache/paimon/manifest/ManifestFileTest.java b/paimon-core/src/test/java/org/apache/paimon/manifest/ManifestFileTest.java index 3bd052ab7ad3..d3ba2527de25 100644 --- a/paimon-core/src/test/java/org/apache/paimon/manifest/ManifestFileTest.java +++ b/paimon-core/src/test/java/org/apache/paimon/manifest/ManifestFileTest.java @@ -124,6 +124,29 @@ void testWriteEncodedDeleteOnlyBlockCountsDeletes() throws Exception { assertEncodedBlockCounts(FileKind.DELETE, FileKind.DELETE); } + @Test + void testWriteEncodedBlockWithoutRowIds() throws Exception { + List entries = Arrays.asList(gen.next(), gen.next()); + ManifestFile manifestFile = createManifestFile(tempDir.toString(), Long.MAX_VALUE); + ManifestFileMeta sourceMeta = writeSingleManifest(manifestFile, entries); + assertThat(sourceMeta.minRowId()).isNull(); + assertThat(sourceMeta.maxRowId()).isNull(); + + ManifestAvroWriter writer = manifestFile.createAvroWriter(); + try (ManifestAvroReader reader = openManifestReader(sourceMeta)) { + assertThat(reader.hasNext()).isTrue(); + ManifestAvroReader.RawBlock block = reader.next(); + writer.writeEncodedBlock(block.encodedBlock(), encodedBlockMeta(sourceMeta)); + assertThat(reader.hasNext()).isFalse(); + } + writer.close(); + + ManifestFileMeta result = writer.result().get(0); + assertThat(result.minRowId()).isNull(); + assertThat(result.maxRowId()).isNull(); + assertThat(manifestFile.read(result.fileName())).containsExactlyElementsOf(entries); + } + @Test void testWriteEncodedRecords() throws Exception { ManifestEntry source = gen.next(); @@ -184,6 +207,84 @@ void testWriteEncodedRecords() throws Exception { assertThat(manifestFile.read(result.fileName())).containsExactlyElementsOf(entries); } + @Test + void testWriteEncodedRecordsFlushesPartitionStatsBuffer() throws Exception { + ManifestEntry generated = gen.next(); + ManifestEntry source = + ManifestEntry.create( + generated.kind(), + generated.partition(), + generated.bucket(), + generated.totalBuckets(), + generated.file().newFirstRowId(0L)); + ManifestFile manifestFile = createManifestFile(tempDir.toString(), Long.MAX_VALUE); + ManifestFileMeta sourceMeta = + writeSingleManifest(manifestFile, Collections.singletonList(source)); + ManifestAvroWriter writer = manifestFile.createAvroWriter(); + ManifestAvroWriter.EncodedEntry metadata = new ManifestAvroWriter.EncodedEntry(); + int recordCount = 8_200; + + try (ManifestAvroReader reader = openManifestReader(sourceMeta)) { + ManifestAvroReader.RowIterator rows = + reader.next().toRows(ManifestEntry.MANIFEST_ROW_TYPE); + rows.next(); + ByteBuffer encodedRecord = rows.encodedRecord(); + for (int i = 0; i < recordCount; i++) { + writer.writeEncoded( + encodedRecord.duplicate(), + metadata.replace( + source.kind().toByteValue(), + source.partition().copy(), + source.bucket(), + source.level(), + source.file().schemaId(), + source.file().firstRowId(), + source.file().rowCount())); + } + } + writer.close(); + + ManifestFileMeta result = writer.result().get(0); + assertThat(result.numAddedFiles()).isEqualTo(recordCount); + assertThat(result.numDeletedFiles()).isZero(); + assertThat(result.partitionStats()).isEqualTo(sourceMeta.partitionStats()); + } + + @Test + void testWriteEncodedManifestPreservesUnknownAggregateStats() throws Exception { + List entries = Arrays.asList(gen.next(), gen.next()); + ManifestFile manifestFile = createManifestFile(tempDir.toString(), Long.MAX_VALUE); + ManifestFileMeta source = writeSingleManifest(manifestFile, entries); + ManifestFileMeta unknownStats = + new ManifestFileMeta( + source.fileName(), + source.fileSize(), + source.numAddedFiles(), + source.numDeletedFiles(), + source.partitionStats(), + source.schemaId(), + null, + null, + null, + null, + source.minRowId(), + source.maxRowId()); + + ManifestAvroWriter writer = manifestFile.createAvroWriter(); + try (ManifestAvroReader reader = openManifestReader(source)) { + writer.writeEncodedManifest(reader, unknownStats); + } + writer.close(); + + ManifestFileMeta result = writer.result().get(0); + assertThat(result.minBucket()).isNull(); + assertThat(result.maxBucket()).isNull(); + assertThat(result.minLevel()).isNull(); + assertThat(result.maxLevel()).isNull(); + assertThat(result.partitionStats()).isEqualTo(source.partitionStats()); + assertThat(manifestFile.read(result.fileName())).containsExactlyElementsOf(entries); + } + @Test void testReadMissingManifestFile() { ManifestFile manifestFile = createManifestFile(tempDir.toString()); @@ -889,7 +990,7 @@ private void assertEncodedBlockCounts(FileKind... kinds) throws Exception { assertThat(reader.hasNext()).isTrue(); ManifestAvroReader.RawBlock block = reader.next(); assertThat(block.recordCount()).isEqualTo(entries.size()); - writer.writeEncodedBlock(block.encodedBlock(), encodedBlock(sourceMeta, entries)); + writer.writeEncodedBlock(block.encodedBlock(), encodedBlockMeta(sourceMeta)); assertThat(reader.hasNext()).isFalse(); } writer.close(); @@ -903,10 +1004,8 @@ private void assertEncodedBlockCounts(FileKind... kinds) throws Exception { assertThat(manifestFile.read(result.fileName())).containsExactlyElementsOf(entries); } - private ManifestAvroWriter.EncodedBlock encodedBlock( - ManifestFileMeta meta, List entries) { - boolean nullPartition = entries.get(0).partition().isNullAt(0); - return new ManifestAvroWriter.EncodedBlock( + private ManifestAvroWriter.EncodedBlockMeta encodedBlockMeta(ManifestFileMeta meta) { + return new ManifestAvroWriter.EncodedBlockMeta( meta.numAddedFiles(), meta.numDeletedFiles(), meta.schemaId(), @@ -914,12 +1013,9 @@ private ManifestAvroWriter.EncodedBlock encodedBlock( meta.maxBucket(), meta.minLevel(), meta.maxLevel(), - meta.minRowId(), - meta.maxRowId(), - nullPartition ? entries.get(0).partition() : null, - nullPartition ? entries.size() : 0, - nullPartition ? null : entries.get(0).partition(), - nullPartition ? null : entries.get(0).partition()); + meta.minRowId() == null ? -1 : meta.minRowId(), + meta.maxRowId() == null ? -1 : meta.maxRowId(), + meta.partitionStats()); } private ManifestAvroReader openManifestReader(ManifestFileMeta manifest) throws IOException { diff --git a/paimon-core/src/test/java/org/apache/paimon/operation/commit/ConflictDetectionTest.java b/paimon-core/src/test/java/org/apache/paimon/operation/commit/ConflictDetectionTest.java index 21e9f71ab989..de0a04bb3d84 100644 --- a/paimon-core/src/test/java/org/apache/paimon/operation/commit/ConflictDetectionTest.java +++ b/paimon-core/src/test/java/org/apache/paimon/operation/commit/ConflictDetectionTest.java @@ -1094,6 +1094,74 @@ void testCheckRowIdExistenceSkipsDeleteEntries() { .isEmpty(); } + @Test + void testCheckRowIdExistenceRejectsStaleDeleteAfterReassign() { + DataEvolutionConflictDetection detection = createConflictDetection(); + + List baseEntries = + Collections.singletonList(createFileEntryWithRowId("f1", ADD, 100L, 10L)); + List deltaEntries = + Collections.singletonList(createFileEntryWithRowId("f1", DELETE, 0L, 10L)); + + Optional result = + detection.checkConflicts( + snapshot(1), + baseEntries, + deltaEntries, + Collections.emptyList(), + null, + Snapshot.CommitKind.APPEND); + + assertThat(result).isPresent(); + assertThat(result.get()) + .isInstanceOf(RowIdExistenceConflictException.class) + .hasMessageContaining("DELETE for file 'f1'") + .hasMessageContaining("firstRowId=0, rowCount=10") + .hasMessageContaining("firstRowId=100, rowCount=10"); + } + + @Test + void testCheckRowIdExistenceRejectsStaleDeleteWithDifferentRowCount() { + DataEvolutionConflictDetection detection = createConflictDetection(); + + List baseEntries = + Collections.singletonList(createFileEntryWithRowId("f1", ADD, 0L, 20L)); + List deltaEntries = + Collections.singletonList(createFileEntryWithRowId("f1", DELETE, 0L, 10L)); + + Optional result = + detection.checkConflicts( + snapshot(1), + baseEntries, + deltaEntries, + Collections.emptyList(), + null, + Snapshot.CommitKind.APPEND); + + assertThat(result).isPresent(); + assertThat(result.get()).isInstanceOf(RowIdExistenceConflictException.class); + } + + @Test + void testCheckRowIdExistenceAllowsDeleteWithCurrentAssignment() { + DataEvolutionConflictDetection detection = createConflictDetection(); + + List baseEntries = + Collections.singletonList(createFileEntryWithRowId("f1", ADD, 100L, 10L)); + List deltaEntries = + Collections.singletonList(createFileEntryWithRowId("f1", DELETE, 100L, 10L)); + + assertThat( + detection.checkConflicts( + snapshot(1), + baseEntries, + deltaEntries, + Collections.emptyList(), + null, + Snapshot.CommitKind.APPEND)) + .isEmpty(); + } + @Test void testCheckRowIdExistenceSkipsWhenNextRowIdNull() { DataEvolutionConflictDetection detection = createConflictDetection(); diff --git a/paimon-format/src/main/java/org/apache/avro/file/RawBlock.java b/paimon-format/src/main/java/org/apache/avro/file/RawBlock.java index 3386074b9df4..8600279380d0 100644 --- a/paimon-format/src/main/java/org/apache/avro/file/RawBlock.java +++ b/paimon-format/src/main/java/org/apache/avro/file/RawBlock.java @@ -57,6 +57,22 @@ public long recordCount() { return block.getNumEntries(); } + public Schema schema() { + return schema; + } + + /** Returns an independently owned copy which remains valid after the reader advances. */ + public RawBlock stableCopy() { + ByteBuffer source = block.getAsByteBuffer().duplicate(); + ByteBuffer copy = ByteBuffer.allocate(source.remaining()); + copy.put(source); + copy.flip(); + DataFileStream.DataBlock copiedBlock = + new DataFileStream.DataBlock(copy, block.getNumEntries()); + copiedBlock.setFlushOnWrite(block.isFlushOnWrite()); + return new RawBlock(copiedBlock, codec, schema); + } + public ByteBuffer decompress(ByteBuffer reuse) throws IOException { if (!decompressed) { if (codec instanceof ZstandardCodec) { @@ -100,20 +116,16 @@ public ByteBuffer decompress(ByteBuffer reuse) throws IOException { target.limit(size); decompressedBuffer = target.duplicate(); } else { - block.decompressUsing(codec); - decompressedBuffer = block.getAsByteBuffer().duplicate(); + decompressedBuffer = codec.decompress(block.getAsByteBuffer().duplicate()); } decompressed = true; } return decompressedBuffer.duplicate(); } - /** Returns a single-block stream for appending this compressed block to an Avro writer. */ - public DataFileStream asStream() throws IOException { - if (decompressed) { - throw new IllegalStateException("A decompressed Avro block cannot be copied raw."); - } - return new SingleBlockStream(schema, codec, block); + /** Returns a single-block stream using a binary-compatible target schema. */ + public DataFileStream asStream(Schema targetSchema) throws IOException { + return new SingleBlockStream(targetSchema, codec, block); } private static final class SingleBlockStream extends DataFileStream { diff --git a/paimon-format/src/main/java/org/apache/paimon/format/avro/AvroBlockReader.java b/paimon-format/src/main/java/org/apache/paimon/format/avro/AvroBlockReader.java index c79b0e198573..c4359afcda43 100644 --- a/paimon-format/src/main/java/org/apache/paimon/format/avro/AvroBlockReader.java +++ b/paimon-format/src/main/java/org/apache/paimon/format/avro/AvroBlockReader.java @@ -22,6 +22,7 @@ import org.apache.paimon.utils.IOUtils; import org.apache.avro.AvroRuntimeException; +import org.apache.avro.Schema; import org.apache.avro.file.RawBlock; import org.apache.avro.file.RawBlockReader; @@ -60,8 +61,51 @@ public AvroRecordDecoder createRecordDecoder() { /** Returns whether blocks use the default Avro schema for the given row type. */ public boolean supportsRawBlockCopy(RowType rowType) { - return AvroSchemaConverter.convertToSchema(rowType, Collections.emptyMap()) - .equals(reader.getSchema()); + return hasSameBinaryLayout( + AvroSchemaConverter.convertToSchema(rowType, Collections.emptyMap()), + reader.getSchema()); + } + + static boolean hasSameBinaryLayout(Schema expected, Schema actual) { + if (expected.getType() != actual.getType()) { + return false; + } + switch (expected.getType()) { + case RECORD: + if (expected.getFields().size() != actual.getFields().size()) { + return false; + } + for (int i = 0; i < expected.getFields().size(); i++) { + if (!expected.getFields().get(i).name().equals(actual.getFields().get(i).name()) + || !hasSameBinaryLayout( + expected.getFields().get(i).schema(), + actual.getFields().get(i).schema())) { + return false; + } + } + return true; + case ARRAY: + return hasSameBinaryLayout(expected.getElementType(), actual.getElementType()); + case MAP: + return hasSameBinaryLayout(expected.getValueType(), actual.getValueType()); + case UNION: + if (expected.getTypes().size() != actual.getTypes().size()) { + return false; + } + for (int i = 0; i < expected.getTypes().size(); i++) { + if (!hasSameBinaryLayout( + expected.getTypes().get(i), actual.getTypes().get(i))) { + return false; + } + } + return true; + case FIXED: + return expected.getFixedSize() == actual.getFixedSize(); + case ENUM: + return expected.getEnumSymbols().equals(actual.getEnumSymbols()); + default: + return true; + } } /** Returns whether another block is available. */ diff --git a/paimon-format/src/main/java/org/apache/paimon/format/avro/AvroBlockWriter.java b/paimon-format/src/main/java/org/apache/paimon/format/avro/AvroBlockWriter.java index 377a7862d099..5cf4a803e03a 100644 --- a/paimon-format/src/main/java/org/apache/paimon/format/avro/AvroBlockWriter.java +++ b/paimon-format/src/main/java/org/apache/paimon/format/avro/AvroBlockWriter.java @@ -22,6 +22,7 @@ import org.apache.paimon.format.FormatWriter; import org.apache.paimon.fs.PositionOutputStream; +import org.apache.avro.Schema; import org.apache.avro.file.DataFileWriter; import java.io.IOException; @@ -32,10 +33,13 @@ public final class AvroBlockWriter implements FormatWriter { private final DataFileWriter writer; private final PositionOutputStream out; + private final Schema schema; - public AvroBlockWriter(DataFileWriter writer, PositionOutputStream out) { + public AvroBlockWriter( + DataFileWriter writer, PositionOutputStream out, Schema schema) { this.writer = writer; this.out = out; + this.schema = schema; } @Override @@ -48,7 +52,11 @@ public void addEncoded(ByteBuffer record) throws IOException { } public void addEncodedBlock(AvroRawBlock block) throws IOException { - writer.appendAllFrom(block.asStream(), false); + if (!AvroBlockReader.hasSameBinaryLayout(schema, block.rawBlock().schema())) { + throw new IllegalArgumentException( + "Avro block schema is not binary-compatible with the writer schema."); + } + writer.appendAllFrom(block.rawBlock().asStream(schema), false); } @Override diff --git a/paimon-format/src/main/java/org/apache/paimon/format/avro/AvroFileFormat.java b/paimon-format/src/main/java/org/apache/paimon/format/avro/AvroFileFormat.java index 73b08eec474f..69512d5ee7cf 100644 --- a/paimon-format/src/main/java/org/apache/paimon/format/avro/AvroFileFormat.java +++ b/paimon-format/src/main/java/org/apache/paimon/format/avro/AvroFileFormat.java @@ -95,7 +95,7 @@ public AvroBlockWriter createBlockWriter( writer.setCodec(createCodecFactory(compression)); writer.setFlushOnEveryBlock(false); writer.create(schema, new CloseShieldOutputStream(out)); - return new AvroBlockWriter(writer, out); + return new AvroBlockWriter(writer, out, schema); } @Override diff --git a/paimon-format/src/main/java/org/apache/paimon/format/avro/AvroRawBlock.java b/paimon-format/src/main/java/org/apache/paimon/format/avro/AvroRawBlock.java index c7d21d4cc8d9..152d5dc30087 100644 --- a/paimon-format/src/main/java/org/apache/paimon/format/avro/AvroRawBlock.java +++ b/paimon-format/src/main/java/org/apache/paimon/format/avro/AvroRawBlock.java @@ -18,7 +18,6 @@ package org.apache.paimon.format.avro; -import org.apache.avro.file.DataFileStream; import org.apache.avro.file.RawBlock; import javax.annotation.Nullable; @@ -48,6 +47,11 @@ public long recordCount() { return block.recordCount(); } + /** Returns an independently owned copy which is not reused by the reader. */ + public AvroRawBlock stableCopy() { + return new AvroRawBlock(block.stableCopy()); + } + /** * Lazily decompresses this block, reusing the supplied heap buffer when possible. * @@ -57,8 +61,4 @@ public long recordCount() { public ByteBuffer decompress(@Nullable ByteBuffer reuse) throws IOException { return block.decompress(reuse); } - - DataFileStream asStream() throws IOException { - return block.asStream(); - } } diff --git a/paimon-format/src/test/java/org/apache/paimon/format/avro/AvroFileFormatTest.java b/paimon-format/src/test/java/org/apache/paimon/format/avro/AvroFileFormatTest.java index 8c05c5b4bf05..ab6031531b35 100644 --- a/paimon-format/src/test/java/org/apache/paimon/format/avro/AvroFileFormatTest.java +++ b/paimon-format/src/test/java/org/apache/paimon/format/avro/AvroFileFormatTest.java @@ -197,6 +197,47 @@ void testReadBorrowedRawBlocks() throws IOException { assertThat(records).isEqualTo(numRecords); } + @Test + void testRawBlockCompatibilityUsesBinaryLayoutAndFieldIdentity() { + Schema expected = + SchemaBuilder.record("Expected") + .fields() + .requiredInt("id") + .name("nested") + .type( + SchemaBuilder.record("ExpectedNested") + .fields() + .requiredLong("value") + .endRecord()) + .noDefault() + .endRecord(); + Schema renamedRecords = + SchemaBuilder.record("Actual") + .fields() + .requiredInt("id") + .name("nested") + .type( + SchemaBuilder.record("ActualNested") + .fields() + .requiredLong("value") + .endRecord()) + .noDefault() + .endRecord(); + Schema renamedField = + SchemaBuilder.record("Actual") + .fields() + .requiredInt("other_id") + .name("nested") + .type(renamedRecords.getField("nested").schema()) + .noDefault() + .endRecord(); + Schema missingField = SchemaBuilder.record("Actual").fields().requiredInt("id").endRecord(); + + assertThat(AvroBlockReader.hasSameBinaryLayout(expected, renamedRecords)).isTrue(); + assertThat(AvroBlockReader.hasSameBinaryLayout(expected, renamedField)).isFalse(); + assertThat(AvroBlockReader.hasSameBinaryLayout(expected, missingField)).isFalse(); + } + @Test void testRowReaderProjectsIntoReusedRow() throws IOException { Schema writerSchema =