This is an automated email from the ASF dual-hosted git repository.
yuzelin pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/paimon.git
The following commit(s) were added to refs/heads/master by this push:
new 55fe85f641 [core] Fix shared decompression buffer in stable manifest
blocks (#9430)
55fe85f641 is described below
commit 55fe85f641b82ba8cebf6d555884641eb9546078
Author: yuzelin <[email protected]>
AuthorDate: Fri Aug 28 00:21:12 2026 +0800
[core] Fix shared decompression buffer in stable manifest blocks (#9430)
---
.../apache/paimon/manifest/ManifestAvroReader.java | 5 +-
.../paimon/operation/ManifestFileMergerTest.java | 142 +++++++++++++++++++++
.../paimon/format/avro/AvroRecordDecoder.java | 7 +
3 files changed, 153 insertions(+), 1 deletion(-)
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 54d52434f8..a61a65f900 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
@@ -381,7 +381,10 @@ public final class ManifestAvroReader implements
AutoCloseable {
/** Returns an independently owned block which remains valid after
this reader advances. */
public RawBlock stableCopy() {
return new RawBlock(
- decoderContext, rawBlockCopySupported, block.stableCopy(),
blockOrdinal);
+ new DecoderContext(decoderContext.decoder.copy()),
+ rawBlockCopySupported,
+ block.stableCopy(),
+ blockOrdinal);
}
}
diff --git
a/paimon-core/src/test/java/org/apache/paimon/operation/ManifestFileMergerTest.java
b/paimon-core/src/test/java/org/apache/paimon/operation/ManifestFileMergerTest.java
index 7df1318902..ff6e738f4d 100644
---
a/paimon-core/src/test/java/org/apache/paimon/operation/ManifestFileMergerTest.java
+++
b/paimon-core/src/test/java/org/apache/paimon/operation/ManifestFileMergerTest.java
@@ -19,20 +19,32 @@
package org.apache.paimon.operation;
import org.apache.paimon.CoreOptions;
+import org.apache.paimon.format.FormatWriter;
+import org.apache.paimon.fs.FileIO;
+import org.apache.paimon.fs.Path;
+import org.apache.paimon.fs.PositionOutputStream;
+import org.apache.paimon.fs.local.LocalFileIO;
+import org.apache.paimon.io.DataFileMeta;
import org.apache.paimon.manifest.FileKind;
+import org.apache.paimon.manifest.ManifestAvroReader;
import org.apache.paimon.manifest.ManifestEntry;
+import org.apache.paimon.manifest.ManifestEntrySerializer;
import org.apache.paimon.manifest.ManifestFile;
import org.apache.paimon.manifest.ManifestFileMeta;
import org.apache.paimon.manifest.ManifestFileMetaTestBase;
import org.apache.paimon.options.Options;
+import org.apache.paimon.types.DataField;
import org.apache.paimon.types.RowType;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.io.TempDir;
+import java.util.ArrayList;
import java.util.Arrays;
+import java.util.Collections;
import java.util.List;
+import java.util.Optional;
import java.util.stream.Collectors;
import static org.assertj.core.api.Assertions.assertThat;
@@ -90,6 +102,87 @@ public class ManifestFileMergerTest extends
ManifestFileMetaTestBase {
.containsExactly("replacement");
}
+ @Test
+ public void testFullCompactionWithStableLegacyManifestBlocks() throws
Exception {
+ List<ManifestEntry> baseEntries = new ArrayList<>();
+
+ // Pending blocks contain many short records and therefore have a
larger record count.
+ for (int i = 0; i < 2_000; i++) {
+ baseEntries.add(makeEntry(true, "short-" + i, null));
+ }
+
+ // Current blocks contain fewer long records while retaining a similar
decompressed size.
+ String longName = String.join("", Collections.nCopies(2_048, "x"));
+ for (int i = 0; i < 200; i++) {
+ baseEntries.add(makeEntry(true, "long-" + i + "-" + longName,
null));
+ }
+
+ ManifestFileMeta base = makeLegacyManifest(baseEntries);
+
+ List<Long> blockRecordCounts = new ArrayList<>();
+ try (ManifestAvroReader reader =
+ manifestFile.scanAvroBlocks(base.fileName(), base.fileSize()))
{
+ assertThat(reader.rawBlockCopySupported()).isFalse();
+
+ while (reader.hasNext()) {
+ blockRecordCounts.add(reader.next().recordCount());
+ }
+ }
+
+ assertThat(blockRecordCounts.size()).isGreaterThan(2);
+
+ // Find the first block whose record count decreases because it
contains long file names.
+ int changedBlock = -1;
+ long changedBlockStart = 0;
+ long blockStart = blockRecordCounts.get(0);
+
+ for (int i = 1; i < blockRecordCounts.size() - 1; i++) {
+ if (blockRecordCounts.get(i) < blockRecordCounts.get(i - 1)) {
+ changedBlock = i;
+ changedBlockStart = blockStart;
+ break;
+ }
+
+ blockStart += blockRecordCounts.get(i);
+ }
+
+ assertThat(changedBlock).isGreaterThan(0);
+
+ // Delete an entry from the middle of the selected block so that it
must be rewritten.
+ int deletedEntryIndex =
+ Math.toIntExact(changedBlockStart +
blockRecordCounts.get(changedBlock) / 2);
+
+ String deletedFileName = baseEntries.get(deletedEntryIndex).fileName();
+ ManifestFileMeta delta = makeManifest(makeEntry(false,
deletedFileName, null));
+
+ List<ManifestFileMeta> newFilesForAbort = new ArrayList<>();
+ Optional<List<ManifestFileMeta>> compacted =
+ ManifestFileBlockMerger.tryFullCompaction(
+ Arrays.asList(base, delta),
+ newFilesForAbort,
+ manifestFile,
+ 1,
+ 1,
+ NO_PARTITION_TYPE,
+ 1);
+
+ assertThat(compacted).isPresent();
+
+ List<ManifestEntry> actual =
+ compacted.get().stream()
+ .flatMap(
+ meta ->
+ manifestFile.read(meta.fileName(),
meta.fileSize())
+ .stream())
+ .collect(Collectors.toList());
+
+ assertThat(actual.size()).isEqualTo(baseEntries.size() - 1);
+ assertThat(actual).allMatch(entry -> entry.kind() == FileKind.ADD);
+ assertThat(actual)
+ .extracting(entry -> entry.file().fileName())
+ .doesNotContain(deletedFileName);
+ }
+
@Override
public ManifestFile getManifestFile() {
return manifestFile;
@@ -99,4 +192,53 @@ public class ManifestFileMergerTest extends
ManifestFileMetaTestBase {
public RowType getPartitionType() {
return NO_PARTITION_TYPE;
}
+
+ private ManifestFileMeta makeLegacyManifest(List<ManifestEntry> entries)
throws Exception {
+ ManifestFileMeta current = makeManifest(entries.toArray(new
ManifestEntry[0]));
+
+ RowType legacyFileType =
+ DataFileMeta.SCHEMA.project(
+ new int[] {
+ 0, 1, 2, 3, 4, 5, 6, 7, 8,
+ 9, 10, 11, 12, 13, 14, 15, 16, 17
+ });
+
+ List<DataField> legacyManifestFields =
+ ManifestEntry.MANIFEST_ROW_TYPE.getFields().stream()
+ .map(
+ field ->
+ ManifestEntry.FILE.equals(field.name())
+ ? field.newType(legacyFileType)
+ : field)
+ .collect(Collectors.toList());
+
+ RowType legacyManifestType = new RowType(false, legacyManifestFields);
+
+ Path path = new Path(new Path(tempDir.toUri()), "manifest/" +
current.fileName());
+
+ FileIO fileIO = LocalFileIO.create();
+ ManifestEntrySerializer serializer = new ManifestEntrySerializer();
+
+ try (PositionOutputStream out = fileIO.newOutputStream(path, true);
+ FormatWriter writer =
+
avro.createWriterFactory(legacyManifestType).create(out, "zstd")) {
+ for (ManifestEntry entry : entries) {
+ writer.addElement(serializer.toRow(entry));
+ }
+ }
+
+ return new ManifestFileMeta(
+ current.fileName(),
+ fileIO.getFileStatus(path).getLen(),
+ current.numAddedFiles(),
+ current.numDeletedFiles(),
+ current.partitionStats(),
+ current.schemaId(),
+ current.minBucket(),
+ current.maxBucket(),
+ current.minLevel(),
+ current.maxLevel(),
+ current.minRowId(),
+ current.maxRowId());
+ }
}
diff --git
a/paimon-format/src/main/java/org/apache/paimon/format/avro/AvroRecordDecoder.java
b/paimon-format/src/main/java/org/apache/paimon/format/avro/AvroRecordDecoder.java
index a4dafc1296..48788ad651 100644
---
a/paimon-format/src/main/java/org/apache/paimon/format/avro/AvroRecordDecoder.java
+++
b/paimon-format/src/main/java/org/apache/paimon/format/avro/AvroRecordDecoder.java
@@ -35,6 +35,7 @@ import java.nio.ByteBuffer;
*/
public final class AvroRecordDecoder {
+ private final Schema writerSchema;
private final Schema recordSchema;
private final int recordBranch;
@@ -44,6 +45,7 @@ public final class AvroRecordDecoder {
private int blockLength;
AvroRecordDecoder(Schema writerSchema) {
+ this.writerSchema = writerSchema;
if (writerSchema.getType() == Schema.Type.UNION) {
int branchIndex = -1;
Schema record = null;
@@ -71,6 +73,11 @@ public final class AvroRecordDecoder {
}
}
+ /** Creates an independent decoder with the same writer schema and no
current block. */
+ public AvroRecordDecoder copy() {
+ return new AvroRecordDecoder(writerSchema);
+ }
+
/** Returns the number of fields in the writer record. */
public int fieldCount() {
return recordSchema.getFields().size();
