This is an automated email from the ASF dual-hosted git repository.
JingsongLi 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 bbcac771ff [core] Optimize manifest sorting with binary entries (#8890)
bbcac771ff is described below
commit bbcac771fff860bedabe717b2ae7ba921440324c
Author: Jingsong Lee <[email protected]>
AuthorDate: Wed Jul 29 00:19:19 2026 +0800
[core] Optimize manifest sorting with binary entries (#8890)
---
.../benchmark/ManifestFileSorterBenchmark.java | 345 +++++++++++++++++++++
.../DataEvolutionRowIdAssignmentPlanner.java | 30 +-
.../org/apache/paimon/io/BinaryDataFileMeta.java | 11 +-
.../org/apache/paimon/io/SingleFileWriter.java | 24 +-
.../paimon/manifest/BinaryManifestEntry.java | 128 +++++++-
.../paimon/manifest}/DeletedIdentifierSet.java | 59 +++-
.../org/apache/paimon/manifest/ManifestEntry.java | 3 +
.../org/apache/paimon/manifest/ManifestFile.java | 9 +-
.../operation/ManifestEntryExternalSort.java | 238 +++++++-------
.../paimon/operation/ManifestFileSorter.java | 153 +++++++--
.../BinaryManifestEntryReusableIdentifierTest.java | 23 +-
.../paimon/manifest/BinaryManifestEntryTest.java | 61 +++-
.../paimon/manifest}/DeletedIdentifierSetTest.java | 6 +-
.../paimon/manifest/ManifestFileMetaTest.java | 4 +-
.../apache/paimon/manifest/ManifestFileTest.java | 3 +-
15 files changed, 898 insertions(+), 199 deletions(-)
diff --git
a/paimon-benchmark/paimon-micro-benchmarks/src/test/java/org/apache/paimon/benchmark/ManifestFileSorterBenchmark.java
b/paimon-benchmark/paimon-micro-benchmarks/src/test/java/org/apache/paimon/benchmark/ManifestFileSorterBenchmark.java
new file mode 100644
index 0000000000..937906490c
--- /dev/null
+++
b/paimon-benchmark/paimon-micro-benchmarks/src/test/java/org/apache/paimon/benchmark/ManifestFileSorterBenchmark.java
@@ -0,0 +1,345 @@
+/*
+ * 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.benchmark;
+
+import org.apache.paimon.CoreOptions;
+import org.apache.paimon.data.BinaryRow;
+import org.apache.paimon.data.BinaryRowWriter;
+import org.apache.paimon.format.FileFormat;
+import org.apache.paimon.fs.FileIO;
+import org.apache.paimon.fs.Path;
+import org.apache.paimon.fs.local.LocalFileIO;
+import org.apache.paimon.io.DataFileMeta;
+import org.apache.paimon.manifest.FileKind;
+import org.apache.paimon.manifest.FileSource;
+import org.apache.paimon.manifest.ManifestEntry;
+import org.apache.paimon.manifest.ManifestFile;
+import org.apache.paimon.manifest.ManifestFileMeta;
+import org.apache.paimon.operation.ManifestFileMerger;
+import org.apache.paimon.options.Options;
+import org.apache.paimon.schema.SchemaManager;
+import org.apache.paimon.stats.SimpleStats;
+import org.apache.paimon.types.IntType;
+import org.apache.paimon.types.RowType;
+import org.apache.paimon.utils.FileStorePathFactory;
+
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+import java.lang.management.ManagementFactory;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
+/**
+ * End-to-end allocation and throughput benchmark for full and minor manifest
sort compaction.
+ *
+ * <p>Run with:
+ *
+ * <pre>
+ * mvn -pl paimon-benchmark/paimon-micro-benchmarks -am -Pfast-build \
+ * -DskipTests package
+ * mvn -pl paimon-benchmark/paimon-micro-benchmarks -am -Pfast-build \
+ * -DfailIfNoTests=false -Dtest=ManifestFileSorterBenchmark test
+ * </pre>
+ *
+ * <p>The data size can be changed with {@code
manifest.sort.benchmark.manifests}, {@code
+ * manifest.sort.benchmark.entries-per-manifest}, {@code
manifest.sort.benchmark.iterations}, and
+ * {@code manifest.sort.benchmark.parallelism}. The external sort buffer can
be changed with {@code
+ * manifest.sort.benchmark.sort-buffer-size}.
+ */
+public class ManifestFileSorterBenchmark {
+
+ private static final RowType PARTITION_TYPE = RowType.of(new IntType());
+ private static final long TARGET_MANIFEST_SIZE = 1024L * 1024L;
+
+ @TempDir java.nio.file.Path tempDir;
+
+ @Test
+ public void benchmarkFullCompaction() throws Exception {
+ int manifestCount =
integerProperty("manifest.sort.benchmark.manifests", 24);
+ int entriesPerManifest =
+
integerProperty("manifest.sort.benchmark.entries-per-manifest", 2_000);
+ int iterations = integerProperty("manifest.sort.benchmark.iterations",
4);
+ int totalEntries = Math.multiplyExact(manifestCount,
entriesPerManifest);
+
+ ManifestFile manifestFile = createManifestFile();
+ List<ManifestFileMeta> input = createInput(manifestFile,
manifestCount, entriesPerManifest);
+ runBenchmark("full", manifestFile, input, totalEntries, totalEntries,
iterations, "1B");
+ }
+
+ @Test
+ public void benchmarkMinorCompactionWithDeletes() throws Exception {
+ int manifestCount =
integerProperty("manifest.sort.benchmark.manifests", 24);
+ int entriesPerManifest =
+
integerProperty("manifest.sort.benchmark.entries-per-manifest", 2_000);
+ int iterations = integerProperty("manifest.sort.benchmark.iterations",
4);
+ int addCount = Math.multiplyExact(manifestCount, entriesPerManifest);
+
+ ManifestFile manifestFile = createManifestFile();
+ List<ManifestFileMeta> input = createInput(manifestFile,
manifestCount, entriesPerManifest);
+ int matchedDeletes = 0;
+ int unmatchedDeletes = 0;
+ int partitionCount = partitionCount(entriesPerManifest);
+ List<ManifestEntry> deletes = new ArrayList<>(entriesPerManifest);
+ for (int manifest = 0; manifest < manifestCount; manifest++) {
+ for (int entry = 0; entry < entriesPerManifest; entry += 4) {
+ int partition = partition(manifest, entry, partitionCount);
+ deletes.add(entry(FileKind.DELETE, fileName(manifest, entry),
partition));
+ matchedDeletes++;
+ if ((entry & 7) == 0) {
+ deletes.add(
+ entry(
+ FileKind.DELETE,
+ "unmatched-" + fileName(manifest, entry),
+ partition));
+ unmatchedDeletes++;
+ }
+ if (deletes.size() >= entriesPerManifest) {
+ input.add(manifestFile.write(deletes).get(0));
+ deletes.clear();
+ }
+ }
+ }
+ if (!deletes.isEmpty()) {
+ input.add(manifestFile.write(deletes).get(0));
+ }
+
+ runBenchmark(
+ "minor-delete",
+ manifestFile,
+ input,
+ addCount + matchedDeletes + unmatchedDeletes,
+ addCount - matchedDeletes + unmatchedDeletes,
+ iterations,
+ Long.MAX_VALUE + "B");
+ }
+
+ private void runBenchmark(
+ String name,
+ ManifestFile manifestFile,
+ List<ManifestFileMeta> input,
+ int processedEntries,
+ int expectedOutputEntries,
+ int iterations,
+ String fullCompactionThreshold) {
+ Set<String> inputNames = new HashSet<>();
+ for (ManifestFileMeta meta : input) {
+ inputNames.add(meta.fileName());
+ }
+
+ Options options = new Options();
+ options.set("manifest-sort.enabled", "true");
+ options.set("manifest.target-file-size", TARGET_MANIFEST_SIZE + "B");
+ options.set("manifest.full-compaction-threshold-size",
fullCompactionThreshold);
+ options.set("manifest-sort.max-rewrite-size", Long.MAX_VALUE + "B");
+ options.set(
+ "scan.manifest.parallelism",
+
Integer.toString(integerProperty("manifest.sort.benchmark.parallelism", 1)));
+ options.set(
+ "sort-spill-buffer-size",
+ System.getProperty("manifest.sort.benchmark.sort-buffer-size",
"64MB"));
+ CoreOptions coreOptions = CoreOptions.fromMap(options.toMap());
+
+ long bestNanos = Long.MAX_VALUE;
+ long totalNanos = 0;
+ long bestAllocatedBytes = Long.MAX_VALUE;
+ for (int iteration = -1; iteration < iterations; iteration++) {
+ System.gc();
+ Map<Long, Long> allocatedBefore = allocatedBytesByThread();
+ long start = System.nanoTime();
+ List<ManifestFileMeta> output =
+ ManifestFileMerger.merge(input, manifestFile,
PARTITION_TYPE, coreOptions);
+ long elapsed = System.nanoTime() - start;
+ long allocated = allocatedBytesSince(allocatedBefore);
+
+ long outputEntries = 0;
+ for (ManifestFileMeta meta : output) {
+ outputEntries += meta.numAddedFiles() + meta.numDeletedFiles();
+ if (!inputNames.contains(meta.fileName())) {
+ manifestFile.delete(meta.fileName());
+ }
+ }
+ if (outputEntries != expectedOutputEntries) {
+ throw new AssertionError(
+ "Expected "
+ + expectedOutputEntries
+ + " output entries, but got "
+ + outputEntries);
+ }
+
+ if (iteration >= 0) {
+ bestNanos = Math.min(bestNanos, elapsed);
+ totalNanos += elapsed;
+ bestAllocatedBytes = Math.min(bestAllocatedBytes, allocated);
+ System.out.printf(
+ "ManifestFileSorter %s iteration %d: %.1f ms, %.1f MiB
allocated%n",
+ name, iteration + 1, elapsed / 1_000_000.0, allocated
/ 1024.0 / 1024.0);
+ }
+ }
+
+ System.out.printf(
+ "ManifestFileSorter %s result: entries=%d, manifests=%d, "
+ + "best/avg=%.1f/%.1f ms, "
+ + "best allocation=%.1f MiB, best rate=%.1f K
entries/s%n",
+ name,
+ processedEntries,
+ input.size(),
+ bestNanos / 1_000_000.0,
+ totalNanos / iterations / 1_000_000.0,
+ bestAllocatedBytes / 1024.0 / 1024.0,
+ processedEntries / (bestNanos / 1_000_000_000.0) / 1_000.0);
+ }
+
+ private List<ManifestFileMeta> createInput(
+ ManifestFile manifestFile, int manifestCount, int
entriesPerManifest) {
+ List<ManifestFileMeta> manifests = new ArrayList<>(manifestCount);
+ int partitionCount = partitionCount(entriesPerManifest);
+ for (int manifest = 0; manifest < manifestCount; manifest++) {
+ List<ManifestEntry> entries = new ArrayList<>(entriesPerManifest);
+ for (int entry = 0; entry < entriesPerManifest; entry++) {
+ int partition = partition(manifest, entry, partitionCount);
+ entries.add(entry(FileKind.ADD, fileName(manifest, entry),
partition));
+ }
+ manifests.add(manifestFile.write(entries).get(0));
+ }
+ return manifests;
+ }
+
+ private static int partitionCount(int entriesPerManifest) {
+ return Math.max(128, entriesPerManifest / 2);
+ }
+
+ private static int partition(int manifest, int entry, int partitionCount) {
+ return Math.floorMod(entry * 104_729 + manifest * 32_749,
partitionCount);
+ }
+
+ private static String fileName(int manifest, int entry) {
+ return String.format(
+
"data-m%03d-e%06d-padding-0123456789abcdef0123456789abcdef.parquet",
+ manifest, entry);
+ }
+
+ private static ManifestEntry entry(FileKind kind, String fileName, int
partition) {
+ BinaryRow partitionRow = new BinaryRow(1);
+ BinaryRowWriter writer = new BinaryRowWriter(partitionRow);
+ writer.writeInt(0, partition);
+ writer.complete();
+ return ManifestEntry.create(
+ kind,
+ partitionRow,
+ 0,
+ 1,
+ DataFileMeta.create(
+ fileName,
+ 128L * 1024L,
+ 1_000L,
+ partitionRow,
+ partitionRow,
+ SimpleStats.EMPTY_STATS,
+ SimpleStats.EMPTY_STATS,
+ 0,
+ 0,
+ 1,
+ 0,
+ Collections.emptyList(),
+ 0L,
+ null,
+ FileSource.APPEND,
+ null,
+ null,
+ null,
+ null));
+ }
+
+ private ManifestFile createManifestFile() {
+ Path tablePath = new Path(tempDir.toString());
+ FileIO fileIO = LocalFileIO.create();
+ FileStorePathFactory pathFactory =
+ new FileStorePathFactory(
+ tablePath,
+ PARTITION_TYPE,
+ "default",
+ CoreOptions.FILE_FORMAT.defaultValue(),
+ CoreOptions.DATA_FILE_PREFIX.defaultValue(),
+ CoreOptions.CHANGELOG_FILE_PREFIX.defaultValue(),
+
CoreOptions.PARTITION_GENERATE_LEGACY_NAME.defaultValue(),
+
CoreOptions.FILE_SUFFIX_INCLUDE_COMPRESSION.defaultValue(),
+ CoreOptions.FILE_COMPRESSION.defaultValue(),
+ null,
+ null,
+ CoreOptions.ExternalPathStrategy.NONE,
+ null,
+ false,
+ null);
+ return new ManifestFile.Factory(
+ fileIO,
+ new SchemaManager(fileIO, tablePath),
+ PARTITION_TYPE,
+ FileFormat.fromIdentifier("avro", new Options()),
+ "zstd",
+ pathFactory,
+ TARGET_MANIFEST_SIZE,
+ null)
+ .create();
+ }
+
+ private static int integerProperty(String name, int defaultValue) {
+ return Integer.parseInt(System.getProperty(name,
Integer.toString(defaultValue)));
+ }
+
+ private static Map<Long, Long> allocatedBytesByThread() {
+ java.lang.management.ThreadMXBean bean =
ManagementFactory.getThreadMXBean();
+ if (!(bean instanceof com.sun.management.ThreadMXBean)) {
+ return Collections.emptyMap();
+ }
+ com.sun.management.ThreadMXBean allocationBean =
(com.sun.management.ThreadMXBean) bean;
+ if (!allocationBean.isThreadAllocatedMemorySupported()) {
+ return Collections.emptyMap();
+ }
+ if (!allocationBean.isThreadAllocatedMemoryEnabled()) {
+ allocationBean.setThreadAllocatedMemoryEnabled(true);
+ }
+ long[] threadIds = bean.getAllThreadIds();
+ long[] allocated = allocationBean.getThreadAllocatedBytes(threadIds);
+ Map<Long, Long> result = new HashMap<>();
+ for (int i = 0; i < threadIds.length; i++) {
+ if (allocated[i] >= 0) {
+ result.put(threadIds[i], allocated[i]);
+ }
+ }
+ return result;
+ }
+
+ private static long allocatedBytesSince(Map<Long, Long> allocatedBefore) {
+ long total = 0;
+ for (Map.Entry<Long, Long> entry :
allocatedBytesByThread().entrySet()) {
+ Long before = allocatedBefore.get(entry.getKey());
+ long delta = entry.getValue() - (before == null ? 0 : before);
+ if (delta > 0) {
+ total += delta;
+ }
+ }
+ return total;
+ }
+}
diff --git
a/paimon-core/src/main/java/org/apache/paimon/append/dataevolution/DataEvolutionRowIdAssignmentPlanner.java
b/paimon-core/src/main/java/org/apache/paimon/append/dataevolution/DataEvolutionRowIdAssignmentPlanner.java
index 09c2a11362..d40e0fd52c 100644
---
a/paimon-core/src/main/java/org/apache/paimon/append/dataevolution/DataEvolutionRowIdAssignmentPlanner.java
+++
b/paimon-core/src/main/java/org/apache/paimon/append/dataevolution/DataEvolutionRowIdAssignmentPlanner.java
@@ -27,6 +27,7 @@ import org.apache.paimon.io.DataFileMeta;
import org.apache.paimon.manifest.BinaryManifestEntry;
import org.apache.paimon.manifest.BinaryManifestEntry.Projection;
import org.apache.paimon.manifest.BinaryManifestEntry.ReusableIdentifier;
+import org.apache.paimon.manifest.DeletedIdentifierSet;
import org.apache.paimon.manifest.ManifestEntry;
import org.apache.paimon.manifest.ManifestFile;
import org.apache.paimon.manifest.ManifestFileMeta;
@@ -38,10 +39,8 @@ import org.apache.paimon.types.RowType;
import org.apache.paimon.utils.ByteArrayKey;
import org.apache.paimon.utils.ByteArrayLookupKey;
import org.apache.paimon.utils.CloseableIterator;
-import org.apache.paimon.utils.DeletedIdentifierSet;
import org.apache.paimon.utils.PrimitiveRowRanges;
import org.apache.paimon.utils.SerializationUtils;
-import org.apache.paimon.utils.VersionedObjectSerializer;
import javax.annotation.Nullable;
@@ -72,14 +71,6 @@ final class DataEvolutionRowIdAssignmentPlanner {
BinaryString.fromString(SpecialFields.ROW_ID.name());
private static final BinaryString BLOB_FILE_SUFFIX =
BinaryString.fromString(".blob");
private static final BinaryString VECTOR_FILE_MARKER =
BinaryString.fromString(".vector.");
- private static final Projection DELETE_PROJECTION =
- manifestProjection(
- true,
- DataFileMeta.FILE_NAME,
- DataFileMeta.LEVEL,
- DataFileMeta.EXTRA_FILES,
- DataFileMeta.EMBEDDED_FILE_INDEX,
- DataFileMeta.EXTERNAL_PATH);
private static final Projection ADD_IDENTIFIER_PROJECTION =
manifestProjection(
true,
@@ -132,15 +123,14 @@ final class DataEvolutionRowIdAssignmentPlanner {
private static Projection manifestProjection(
boolean includeBucket, String... projectedFileFields) {
- RowType manifestType =
VersionedObjectSerializer.versionType(ManifestEntry.SCHEMA);
List<DataField> fields = new ArrayList<>();
- fields.add(manifestType.getField(ManifestEntry.KIND));
- fields.add(manifestType.getField(ManifestEntry.PARTITION));
+
fields.add(ManifestEntry.MANIFEST_ROW_TYPE.getField(ManifestEntry.KIND));
+
fields.add(ManifestEntry.MANIFEST_ROW_TYPE.getField(ManifestEntry.PARTITION));
if (includeBucket) {
- fields.add(manifestType.getField(ManifestEntry.BUCKET));
+
fields.add(ManifestEntry.MANIFEST_ROW_TYPE.getField(ManifestEntry.BUCKET));
}
fields.add(
- manifestType
+ ManifestEntry.MANIFEST_ROW_TYPE
.getField(ManifestEntry.FILE)
.newType(DataFileMeta.SCHEMA.project(projectedFileFields)));
return Projection.create(new RowType(false, fields));
@@ -191,7 +181,9 @@ final class DataEvolutionRowIdAssignmentPlanner {
}
try (CloseableIterator<BinaryManifestEntry> entries =
manifestFile.scan(
- manifestMeta.fileName(), manifestMeta.fileSize(),
DELETE_PROJECTION)) {
+ manifestMeta.fileName(),
+ manifestMeta.fileSize(),
+ BinaryManifestEntry.DELETE_ENTRY_PROJECTION)) {
while (entries.hasNext()) {
BinaryManifestEntry entry = entries.next();
if (!entry.isDelete()) {
@@ -202,8 +194,7 @@ final class DataEvolutionRowIdAssignmentPlanner {
continue;
}
identifier.replace(entry);
- group.deletedIdentifiers.add(
- partition.id, identifier.bytes(),
identifier.length());
+ group.deletedIdentifiers.add(partition.id, identifier);
}
} catch (Exception e) {
throw scanException(manifestMeta, e);
@@ -240,8 +231,7 @@ final class DataEvolutionRowIdAssignmentPlanner {
}
if (!group.deletedIdentifiers.isEmpty()) {
identifier.replace(entry);
- if (group.deletedIdentifiers.contains(
- partition.id, identifier.bytes(),
identifier.length())) {
+ if (group.deletedIdentifiers.contains(partition.id,
identifier)) {
continue;
}
}
diff --git
a/paimon-core/src/main/java/org/apache/paimon/io/BinaryDataFileMeta.java
b/paimon-core/src/main/java/org/apache/paimon/io/BinaryDataFileMeta.java
index 3a36bf8594..be56deb207 100644
--- a/paimon-core/src/main/java/org/apache/paimon/io/BinaryDataFileMeta.java
+++ b/paimon-core/src/main/java/org/apache/paimon/io/BinaryDataFileMeta.java
@@ -60,11 +60,12 @@ public final class BinaryDataFileMeta implements
DataFileMeta {
/** Replaces the backing row and returns this reusable view. */
public BinaryDataFileMeta replace(InternalRow row) {
checkArgument(row != null, "Data file row cannot be null.");
- checkArgument(
- row.getFieldCount() == projection.fieldCount,
- "Data file row field count %s does not match projected field
count %s.",
- row.getFieldCount(),
- projection.fieldCount);
+ if (row.getFieldCount() != projection.fieldCount) {
+ throw new IllegalArgumentException(
+ String.format(
+ "Data file row field count %s does not match
projected field count %s.",
+ row.getFieldCount(), projection.fieldCount));
+ }
this.row = row;
return this;
}
diff --git
a/paimon-core/src/main/java/org/apache/paimon/io/SingleFileWriter.java
b/paimon-core/src/main/java/org/apache/paimon/io/SingleFileWriter.java
index 473acecea9..1355ea3518 100644
--- a/paimon-core/src/main/java/org/apache/paimon/io/SingleFileWriter.java
+++ b/paimon-core/src/main/java/org/apache/paimon/io/SingleFileWriter.java
@@ -143,8 +143,7 @@ public abstract class SingleFileWriter<T, R> implements
FileWriter<T, R> {
try {
InternalRow rowData = converter.apply(record);
- writer.addElement(rowData);
- recordCount++;
+ writeRowInternal(rowData);
return rowData;
} catch (Throwable e) {
LOG.warn("Exception occurs when writing file {}. Cleaning up.",
path, e);
@@ -153,6 +152,27 @@ public abstract class SingleFileWriter<T, R> implements
FileWriter<T, R> {
}
}
+ /** Writes an already converted row without invoking this writer's record
converter. */
+ protected InternalRow writeRow(InternalRow rowData) throws IOException {
+ if (closed) {
+ throw new RuntimeException("Writer has already closed!");
+ }
+
+ try {
+ writeRowInternal(rowData);
+ return rowData;
+ } catch (Throwable e) {
+ LOG.warn("Exception occurs when writing file {}. Cleaning up.",
path, e);
+ abort();
+ throw e;
+ }
+ }
+
+ private void writeRowInternal(InternalRow rowData) throws IOException {
+ writer.addElement(rowData);
+ recordCount++;
+ }
+
@Override
public long recordCount() {
return recordCount;
diff --git
a/paimon-core/src/main/java/org/apache/paimon/manifest/BinaryManifestEntry.java
b/paimon-core/src/main/java/org/apache/paimon/manifest/BinaryManifestEntry.java
index 86d3bca3cf..73e361a904 100644
---
a/paimon-core/src/main/java/org/apache/paimon/manifest/BinaryManifestEntry.java
+++
b/paimon-core/src/main/java/org/apache/paimon/manifest/BinaryManifestEntry.java
@@ -22,10 +22,11 @@ import org.apache.paimon.data.BinaryRow;
import org.apache.paimon.data.BinaryString;
import org.apache.paimon.data.InternalRow;
import org.apache.paimon.io.BinaryDataFileMeta;
+import org.apache.paimon.io.DataFileMeta;
+import org.apache.paimon.memory.MemorySegment;
import org.apache.paimon.memory.MemorySegmentUtils;
import org.apache.paimon.types.DataField;
import org.apache.paimon.types.RowType;
-import org.apache.paimon.utils.VersionedObjectSerializer;
import javax.annotation.Nullable;
@@ -34,7 +35,6 @@ import java.util.List;
import static org.apache.paimon.utils.Preconditions.checkArgument;
import static org.apache.paimon.utils.Preconditions.checkState;
-import static org.apache.paimon.utils.SerializationUtils.deserializeBinaryRow;
/**
* Reusable binary view of a projected manifest entry.
@@ -46,11 +46,13 @@ import static
org.apache.paimon.utils.SerializationUtils.deserializeBinaryRow;
*/
public final class BinaryManifestEntry implements ManifestEntry {
- private static final RowType MANIFEST_TYPE =
- VersionedObjectSerializer.versionType(ManifestEntry.SCHEMA);
+ private static final Projection FULL_PROJECTION =
+ Projection.create(ManifestEntry.MANIFEST_ROW_TYPE);
+ public static final Projection DELETE_ENTRY_PROJECTION =
createDeleteEntryProjection();
private final Projection projection;
private final @Nullable BinaryDataFileMeta file;
+ private final ReusablePartition partitionView = new ReusablePartition();
private @Nullable InternalRow row;
private BinaryManifestEntry(Projection projection) {
@@ -64,6 +66,12 @@ public final class BinaryManifestEntry implements
ManifestEntry {
/** Replaces the backing row and returns this reusable view. */
public BinaryManifestEntry replace(InternalRow row) {
checkArgument(row != null, "Manifest row cannot be null.");
+ if (row.getFieldCount() != projection.projectedType.getFieldCount()) {
+ throw new IllegalArgumentException(
+ String.format(
+ "Manifest row field count %s does not match
projected field count %s.",
+ row.getFieldCount(),
projection.projectedType.getFieldCount()));
+ }
if (projection.filePosition >= 0) {
InternalRow fileRow =
row.getRow(projection.filePosition,
projection.projectedFileFieldCount);
@@ -71,12 +79,48 @@ public final class BinaryManifestEntry implements
ManifestEntry {
file.replace(fileRow);
}
this.row = row;
+ this.partitionView.reset();
return this;
}
+ /** Returns the backing row when this entry uses the complete versioned
manifest schema. */
+ public InternalRow fullRow() {
+ checkState(
+ projection.fullProjection,
+ "The selected binary manifest projection is not the complete
manifest schema.");
+ checkState(row != null, "Binary manifest entry is not backed by a
row.");
+ return row;
+ }
+
+ /** Returns the reusable projection for the complete versioned manifest
schema. */
+ public static Projection fullProjection() {
+ return FULL_PROJECTION;
+ }
+
+ private static Projection createDeleteEntryProjection() {
+ RowType manifestType = ManifestEntry.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.LEVEL,
+
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;
+ partitionView.reset();
if (file != null) {
file.clear();
}
@@ -99,17 +143,54 @@ public final class BinaryManifestEntry implements
ManifestEntry {
}
public byte[] partitionBytes() {
- byte[] partition =
- row.getBinary(
- requiredOuterPosition(
- projection.partitionPosition,
ManifestEntry.PARTITION));
- checkState(partition != null, "Serialized manifest partition cannot be
null.");
- return partition;
+ return partitionView.getBytes(
+ row, requiredOuterPosition(projection.partitionPosition,
ManifestEntry.PARTITION));
}
@Override
public BinaryRow partition() {
- return deserializeBinaryRow(partitionBytes());
+ return partitionView.getRow(
+ row, requiredOuterPosition(projection.partitionPosition,
ManifestEntry.PARTITION));
+ }
+
+ private static final class ReusablePartition {
+
+ private final MemorySegment[] segments = new MemorySegment[1];
+ private @Nullable byte[] bytes;
+ private @Nullable BinaryRow row;
+
+ private byte[] getBytes(InternalRow entryRow, int position) {
+ if (bytes == null) {
+ bytes = entryRow.getBinary(position);
+ checkState(bytes != null, "Serialized manifest partition
cannot be null.");
+ }
+ return bytes;
+ }
+
+ private BinaryRow getRow(InternalRow entryRow, int position) {
+ if (segments[0] == null) {
+ byte[] bytes = getBytes(entryRow, position);
+ checkState(
+ bytes.length >= Integer.BYTES,
+ "Serialized manifest partition is too short.");
+ int arity =
+ ((bytes[0] & 0xff) << 24)
+ | ((bytes[1] & 0xff) << 16)
+ | ((bytes[2] & 0xff) << 8)
+ | (bytes[3] & 0xff);
+ if (row == null || row.getFieldCount() != arity) {
+ row = new BinaryRow(arity);
+ }
+ segments[0] = MemorySegment.wrap(bytes);
+ row.pointTo(segments, Integer.BYTES, bytes.length -
Integer.BYTES);
+ }
+ return row;
+ }
+
+ private void reset() {
+ bytes = null;
+ segments[0] = null;
+ }
}
@Override
@@ -241,6 +322,18 @@ public final class BinaryManifestEntry implements
ManifestEntry {
public ReusableIdentifier replace(BinaryManifestEntry entry) {
checkArgument(entry != null, "Binary manifest entry cannot be
null.");
length = 0;
+ return appendEntryFields(entry);
+ }
+
+ /** Replaces this encoding with the entry's partition and identity
fields. */
+ public ReusableIdentifier replaceWithPartition(BinaryManifestEntry
entry) {
+ checkArgument(entry != null, "Binary manifest entry cannot be
null.");
+ length = 0;
+ putBytes(entry.partitionBytes());
+ return appendEntryFields(entry);
+ }
+
+ private ReusableIdentifier appendEntryFields(BinaryManifestEntry
entry) {
putInt(entry.bucket());
BinaryDataFileMeta file = entry.file();
putInt(file.level());
@@ -331,6 +424,7 @@ public final class BinaryManifestEntry implements
ManifestEntry {
private final int filePosition;
private final int projectedFileFieldCount;
private final @Nullable BinaryDataFileMeta.Projection fileProjection;
+ private final boolean fullProjection;
private Projection(
RowType projectedType,
@@ -340,7 +434,8 @@ public final class BinaryManifestEntry implements
ManifestEntry {
int totalBucketsPosition,
int filePosition,
int projectedFileFieldCount,
- @Nullable BinaryDataFileMeta.Projection fileProjection) {
+ @Nullable BinaryDataFileMeta.Projection fileProjection,
+ boolean fullProjection) {
this.projectedType = projectedType;
this.kindPosition = kindPosition;
this.partitionPosition = partitionPosition;
@@ -349,6 +444,7 @@ public final class BinaryManifestEntry implements
ManifestEntry {
this.filePosition = filePosition;
this.projectedFileFieldCount = projectedFileFieldCount;
this.fileProjection = fileProjection;
+ this.fullProjection = fullProjection;
}
public static Projection create(RowType projectedType) {
@@ -373,17 +469,19 @@ public final class BinaryManifestEntry implements
ManifestEntry {
projectedType.getFieldIndex(ManifestEntry.TOTAL_BUCKETS),
filePosition,
projectedFileFieldCount,
- fileProjection);
+ fileProjection,
+ projectedType.equals(ManifestEntry.MANIFEST_ROW_TYPE));
}
private static void validateProjection(RowType projectedType) {
for (DataField projectedField : projectedType.getFields()) {
checkArgument(
- MANIFEST_TYPE.containsField(projectedField.id()),
+
ManifestEntry.MANIFEST_ROW_TYPE.containsField(projectedField.id()),
"Unknown projected manifest field '%s' (id %s).",
projectedField.name(),
projectedField.id());
- DataField manifestField =
MANIFEST_TYPE.getField(projectedField.id());
+ DataField manifestField =
+
ManifestEntry.MANIFEST_ROW_TYPE.getField(projectedField.id());
checkArgument(
projectedField.isPrunedFrom(manifestField),
"Projected manifest field '%s' does not match %s.",
diff --git
a/paimon-common/src/main/java/org/apache/paimon/utils/DeletedIdentifierSet.java
b/paimon-core/src/main/java/org/apache/paimon/manifest/DeletedIdentifierSet.java
similarity index 76%
rename from
paimon-common/src/main/java/org/apache/paimon/utils/DeletedIdentifierSet.java
rename to
paimon-core/src/main/java/org/apache/paimon/manifest/DeletedIdentifierSet.java
index 6fd4290418..8b97cb533c 100644
---
a/paimon-common/src/main/java/org/apache/paimon/utils/DeletedIdentifierSet.java
+++
b/paimon-core/src/main/java/org/apache/paimon/manifest/DeletedIdentifierSet.java
@@ -16,18 +16,28 @@
* limitations under the License.
*/
-package org.apache.paimon.utils;
+package org.apache.paimon.manifest;
+
+import org.apache.paimon.manifest.BinaryManifestEntry.ReusableIdentifier;
+
+import javax.annotation.Nullable;
import java.util.Arrays;
import static org.apache.paimon.utils.Preconditions.checkArgument;
import static org.apache.paimon.utils.Preconditions.checkState;
-/** Compact, collision-safe set backed by primitive arrays and one identifier
byte arena. */
+/**
+ * Compact, collision-safe set backed by primitive arrays and one identifier
byte arena.
+ *
+ * <p>The reusable identifier is only used as lookup scratch. Added identifier
bytes are copied into
+ * the arena.
+ */
public final class DeletedIdentifierSet {
private static final float LOAD_FACTOR = 0.75f;
+ private @Nullable ReusableIdentifier reusableIdentifier;
private int[] buckets = filledWithMinusOne(16);
private long[] hashes = new long[16];
private int[] partitionIds = new int[16];
@@ -50,7 +60,37 @@ public final class DeletedIdentifierSet {
return arenaSize;
}
+ public void add(BinaryManifestEntry entry) {
+ add(reusableIdentifier().replaceWithPartition(entry));
+ }
+
+ public void add(ReusableIdentifier identifier) {
+ add(0, identifier);
+ }
+
+ public void add(int partitionId, ReusableIdentifier identifier) {
+ checkIdentifier(identifier);
+ add(partitionId, identifier.bytes(), identifier.length());
+ }
+
+ public boolean contains(BinaryManifestEntry entry) {
+ return contains(reusableIdentifier().replaceWithPartition(entry));
+ }
+
+ public boolean contains(ReusableIdentifier identifier) {
+ return contains(0, identifier);
+ }
+
+ public boolean contains(int partitionId, ReusableIdentifier identifier) {
+ checkIdentifier(identifier);
+ return contains(partitionId, identifier.bytes(), identifier.length());
+ }
+
public void release() {
+ if (reusableIdentifier != null) {
+ reusableIdentifier.release();
+ reusableIdentifier = null;
+ }
buckets = filledWithMinusOne(16);
hashes = new long[0];
partitionIds = new int[0];
@@ -62,7 +102,7 @@ public final class DeletedIdentifierSet {
size = 0;
}
- public void add(int partitionId, byte[] identifier, int length) {
+ void add(int partitionId, byte[] identifier, int length) {
checkIdentifier(identifier, length);
long hash = hash(partitionId, identifier, length);
if (contains(partitionId, identifier, length, hash)) {
@@ -87,7 +127,7 @@ public final class DeletedIdentifierSet {
size++;
}
- public boolean contains(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));
}
@@ -174,6 +214,17 @@ public final class DeletedIdentifierSet {
return true;
}
+ private static void checkIdentifier(ReusableIdentifier identifier) {
+ checkArgument(identifier != null, "Identifier cannot be null.");
+ }
+
+ private ReusableIdentifier reusableIdentifier() {
+ if (reusableIdentifier == null) {
+ reusableIdentifier = new ReusableIdentifier();
+ }
+ return reusableIdentifier;
+ }
+
private static void checkIdentifier(byte[] identifier, int length) {
checkArgument(identifier != null, "Identifier bytes cannot be null.");
checkArgument(
diff --git
a/paimon-core/src/main/java/org/apache/paimon/manifest/ManifestEntry.java
b/paimon-core/src/main/java/org/apache/paimon/manifest/ManifestEntry.java
index 822f257321..788a86794b 100644
--- a/paimon-core/src/main/java/org/apache/paimon/manifest/ManifestEntry.java
+++ b/paimon-core/src/main/java/org/apache/paimon/manifest/ManifestEntry.java
@@ -25,6 +25,7 @@ import org.apache.paimon.types.DataField;
import org.apache.paimon.types.IntType;
import org.apache.paimon.types.RowType;
import org.apache.paimon.types.TinyIntType;
+import org.apache.paimon.utils.VersionedObjectSerializer;
import javax.annotation.Nullable;
@@ -57,6 +58,8 @@ public interface ManifestEntry extends FileEntry {
new DataField(3, TOTAL_BUCKETS, new
IntType(false)),
new DataField(4, FILE, DataFileMeta.SCHEMA)));
+ RowType MANIFEST_ROW_TYPE = VersionedObjectSerializer.versionType(SCHEMA);
+
static ManifestEntry create(
FileKind kind, BinaryRow partition, int bucket, int totalBuckets,
DataFileMeta file) {
return new PojoManifestEntry(kind, partition, bucket, totalBuckets,
file);
diff --git
a/paimon-core/src/main/java/org/apache/paimon/manifest/ManifestFile.java
b/paimon-core/src/main/java/org/apache/paimon/manifest/ManifestFile.java
index abeaa36a64..c5ee9d9c0c 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
@@ -42,7 +42,6 @@ import org.apache.paimon.utils.Filter;
import org.apache.paimon.utils.ObjectsFile;
import org.apache.paimon.utils.PathFactory;
import org.apache.paimon.utils.SegmentsCache;
-import org.apache.paimon.utils.VersionedObjectSerializer;
import javax.annotation.Nullable;
@@ -271,7 +270,11 @@ public class ManifestFile extends
ObjectsFile<ManifestEntry> {
@Override
public void write(ManifestEntry entry) throws IOException {
- super.write(entry);
+ if (entry instanceof BinaryManifestEntry) {
+ writeRow(((BinaryManifestEntry) entry).fullRow());
+ } else {
+ super.write(entry);
+ }
switch (entry.kind()) {
case ADD:
@@ -367,7 +370,7 @@ public class ManifestFile extends
ObjectsFile<ManifestEntry> {
}
public ManifestFile create() {
- RowType entryType =
VersionedObjectSerializer.versionType(ManifestEntry.SCHEMA);
+ RowType entryType = ManifestEntry.MANIFEST_ROW_TYPE;
return new ManifestFile(
fileIO,
schemaManager,
diff --git
a/paimon-core/src/main/java/org/apache/paimon/operation/ManifestEntryExternalSort.java
b/paimon-core/src/main/java/org/apache/paimon/operation/ManifestEntryExternalSort.java
index 5f2c342cc4..94cc2caf75 100644
---
a/paimon-core/src/main/java/org/apache/paimon/operation/ManifestEntryExternalSort.java
+++
b/paimon-core/src/main/java/org/apache/paimon/operation/ManifestEntryExternalSort.java
@@ -19,33 +19,29 @@
package org.apache.paimon.operation;
import org.apache.paimon.CoreOptions;
-import org.apache.paimon.codegen.CodeGenUtils;
-import org.apache.paimon.codegen.RecordComparator;
import org.apache.paimon.compression.CompressOptions;
import org.apache.paimon.data.BinaryRow;
+import org.apache.paimon.data.GenericRow;
+import org.apache.paimon.data.serializer.InternalRowSerializer;
import org.apache.paimon.disk.IOManager;
import org.apache.paimon.io.RollingFileWriter;
-import org.apache.paimon.manifest.FileEntry;
-import org.apache.paimon.manifest.FileKind;
+import org.apache.paimon.manifest.BinaryManifestEntry;
+import org.apache.paimon.manifest.BinaryManifestEntry.ReusableIdentifier;
+import org.apache.paimon.manifest.DeletedIdentifierSet;
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.options.MemorySize;
import org.apache.paimon.sort.BinaryExternalSortBuffer;
-import org.apache.paimon.utils.Filter;
+import org.apache.paimon.utils.CloseableIterator;
import org.apache.paimon.utils.MutableObjectIterator;
import org.apache.paimon.utils.Pair;
import javax.annotation.Nullable;
import java.util.ArrayList;
-import java.util.Collection;
import java.util.Collections;
-import java.util.HashMap;
import java.util.List;
-import java.util.Map;
-import java.util.Set;
import java.util.function.Function;
import static
org.apache.paimon.utils.ManifestReadThreadPool.sequentialBatchedExecute;
@@ -62,29 +58,23 @@ public class ManifestEntryExternalSort {
@Nullable Integer manifestReadParallelism)
throws Exception {
try (EntrySorter sorter = new EntrySorter(sortKey, config)) {
- Map<FileEntry.Identifier, ManifestEntry> deleteEntries = new
HashMap<>();
- Function<ManifestFileMeta, List<ManifestEntry>> reader =
- meta -> manifestFile.read(meta.fileName(),
meta.fileSize());
- for (ManifestEntry entry :
- sequentialBatchedExecute(reader, section,
manifestReadParallelism)) {
- if (entry.kind() == FileKind.DELETE) {
- deleteEntries.put(entry.identifier(), entry);
- } else {
- sorter.write(entry);
- }
- }
+ DeletedIdentifierSet deleteEntries = new DeletedIdentifierSet();
+ try {
+ scanEntries(
+ section,
+ manifestFile,
+ manifestReadParallelism,
+ entry -> {
+ if (entry.isDelete()) {
+ deleteEntries.add(entry);
+ }
+ sorter.write(entry);
+ });
- List<ManifestFileMeta> addFiles =
- sorter.writeSurvivingAddsToManifest(manifestFile,
deleteEntries);
- // Register ADD files for abort cleanup right after they are
written, before the
- // DELETE files below. Otherwise, if sortAndWriteDeleteEntries
throws, the already
- // written ADD manifest files would not be in newFilesForAbort and
would leak as
- // orphan files on commit abort.
- newFilesForAbort.addAll(addFiles);
- List<ManifestFileMeta> deleteFiles =
- sortAndWriteDeleteEntries(deleteEntries.values(), sortKey,
manifestFile);
- newFilesForAbort.addAll(deleteFiles);
- return Pair.of(addFiles, deleteFiles);
+ return sorter.writeMinorToManifest(manifestFile,
deleteEntries, newFilesForAbort);
+ } finally {
+ deleteEntries.release();
+ }
}
}
@@ -94,35 +84,81 @@ public class ManifestEntryExternalSort {
ExternalSortConfig config,
ManifestFile manifestFile,
List<ManifestFileMeta> newFilesForAbort,
- Set<FileEntry.Identifier> deleteEntries,
+ DeletedIdentifierSet deleteEntries,
@Nullable Integer manifestReadParallelism)
throws Exception {
try (EntrySorter sorter = new EntrySorter(sortKey, config)) {
- Function<ManifestFileMeta, List<ManifestEntry>> reader =
- meta -> {
- List<ManifestEntry> batch = new ArrayList<>();
- for (ManifestEntry entry :
- manifestFile.read(
- meta.fileName(),
- meta.fileSize(),
- FileEntry.addFilter(),
- Filter.alwaysTrue())) {
- if (!deleteEntries.contains(entry.identifier())) {
- batch.add(entry);
- }
+ scanEntries(
+ section,
+ manifestFile,
+ manifestReadParallelism,
+ entry -> {
+ if (entry.isAdd()
+ && (deleteEntries.isEmpty() ||
!deleteEntries.contains(entry))) {
+ sorter.write(entry);
}
- return batch;
- };
- for (ManifestEntry entry :
- sequentialBatchedExecute(reader, section,
manifestReadParallelism)) {
- sorter.write(entry);
- }
+ });
List<ManifestFileMeta> files =
sorter.writeToManifest(manifestFile);
newFilesForAbort.addAll(files);
return files;
}
}
+ private static void scanEntries(
+ List<ManifestFileMeta> section,
+ ManifestFile manifestFile,
+ @Nullable Integer manifestReadParallelism,
+ BinaryEntryConsumer consumer)
+ throws Exception {
+ if (section.size() <= 1
+ || (manifestReadParallelism != null && manifestReadParallelism
<= 1)) {
+ for (ManifestFileMeta meta : section) {
+ try (CloseableIterator<BinaryManifestEntry> entries =
+ manifestFile.scan(
+ meta.fileName(),
+ meta.fileSize(),
+ BinaryManifestEntry.fullProjection())) {
+ while (entries.hasNext()) {
+ consumer.accept(entries.next());
+ }
+ }
+ }
+ return;
+ }
+
+ Function<ManifestFileMeta, List<BinaryRow>> reader =
+ meta -> readBinaryRows(manifestFile, meta);
+ BinaryManifestEntry entry =
BinaryManifestEntry.fullProjection().createEntry();
+ for (BinaryRow row : sequentialBatchedExecute(reader, section,
manifestReadParallelism)) {
+ consumer.accept(entry.replace(row));
+ }
+ entry.clear();
+ }
+
+ private static List<BinaryRow> readBinaryRows(
+ ManifestFile manifestFile, ManifestFileMeta meta) {
+ long entryCount = meta.numAddedFiles() + meta.numDeletedFiles();
+ List<BinaryRow> rows = new ArrayList<>((int) Math.min(entryCount, 1 <<
20));
+ InternalRowSerializer serializer =
+ new InternalRowSerializer(ManifestEntry.MANIFEST_ROW_TYPE);
+ try (CloseableIterator<BinaryManifestEntry> entries =
+ manifestFile.scan(
+ meta.fileName(), meta.fileSize(),
BinaryManifestEntry.fullProjection())) {
+ while (entries.hasNext()) {
+
rows.add(serializer.toBinaryRow(entries.next().fullRow()).copy());
+ }
+ } catch (Exception e) {
+ throw new RuntimeException(
+ String.format("Failed to scan manifest file '%s'.",
meta.fileName()), e);
+ }
+ return rows;
+ }
+
+ @FunctionalInterface
+ private interface BinaryEntryConsumer {
+ void accept(BinaryManifestEntry entry) throws Exception;
+ }
+
/** Config used by manifest entry external sort. */
static class ExternalSortConfig {
final long bufferSize;
@@ -158,17 +194,17 @@ public class ManifestEntryExternalSort {
}
}
- /** Spillable sorter that stores sort keys plus serialized manifest
entries in BinaryRow. */
+ /** Spillable sorter that stores sort keys plus complete binary manifest
rows. */
private static class EntrySorter implements AutoCloseable {
private final ManifestFileSorter.ManifestSortKey sortKey;
- private final ManifestEntrySerializer entrySerializer;
+ private final GenericRow externalSortRow;
private final IOManager ioManager;
private final boolean ownedIOManager;
private final BinaryExternalSortBuffer sortBuffer;
private EntrySorter(ManifestFileSorter.ManifestSortKey sortKey,
ExternalSortConfig config) {
this.sortKey = sortKey;
- this.entrySerializer = new ManifestEntrySerializer();
+ this.externalSortRow = new
GenericRow(sortKey.externalSortRowType().getFieldCount());
this.ioManager =
config.ioManager == null
?
IOManager.create(System.getProperty("java.io.tmpdir"))
@@ -186,9 +222,9 @@ public class ManifestEntryExternalSort {
config.maxDiskSize);
}
- private void write(ManifestEntry entry) throws Exception {
- sortBuffer.write(
- sortKey.toExternalSortRow(entry,
entrySerializer.serializeToBytes(entry)));
+ private void write(BinaryManifestEntry entry) throws Exception {
+ sortKey.replaceExternalSortRow(externalSortRow, entry,
entry.fullRow());
+ sortBuffer.write(externalSortRow);
}
private boolean isEmpty() {
@@ -206,10 +242,12 @@ public class ManifestEntryExternalSort {
try {
MutableObjectIterator<BinaryRow> iterator =
sortBuffer.sortedIterator();
BinaryRow reuse = new
BinaryRow(sortKey.externalSortRowType().getFieldCount());
+ BinaryManifestEntry entry =
BinaryManifestEntry.fullProjection().createEntry();
BinaryRow row;
while ((row = iterator.next(reuse)) != null) {
-
writer.write(entrySerializer.deserializeFromBytes(sortKey.entryBytes(row)));
+
writer.write(entry.replace(sortKey.binaryManifestRow(row)));
}
+ entry.clear();
} catch (Exception e) {
exception = e;
} finally {
@@ -222,38 +260,61 @@ public class ManifestEntryExternalSort {
return writer.result();
}
- private List<ManifestFileMeta> writeSurvivingAddsToManifest(
- ManifestFile manifestFile, Map<FileEntry.Identifier,
ManifestEntry> deleteEntries)
+ private Pair<List<ManifestFileMeta>, List<ManifestFileMeta>>
writeMinorToManifest(
+ ManifestFile manifestFile,
+ DeletedIdentifierSet deleteEntries,
+ List<ManifestFileMeta> newFilesForAbort)
throws Exception {
if (isEmpty()) {
- return Collections.emptyList();
+ return Pair.of(Collections.emptyList(),
Collections.emptyList());
}
- RollingFileWriter<ManifestEntry, ManifestFileMeta> writer =
+ RollingFileWriter<ManifestEntry, ManifestFileMeta> addWriter =
manifestFile.createRollingWriter();
+ RollingFileWriter<ManifestEntry, ManifestFileMeta> deleteWriter =
+ manifestFile.createRollingWriter();
+ DeletedIdentifierSet matchedEntries = new DeletedIdentifierSet();
+ DeletedIdentifierSet emittedDeletes = new DeletedIdentifierSet();
+ ReusableIdentifier identifier = new ReusableIdentifier();
Exception exception = null;
try {
MutableObjectIterator<BinaryRow> iterator =
sortBuffer.sortedIterator();
BinaryRow reuse = new
BinaryRow(sortKey.externalSortRowType().getFieldCount());
+ BinaryManifestEntry entry =
BinaryManifestEntry.fullProjection().createEntry();
BinaryRow row;
while ((row = iterator.next(reuse)) != null) {
- ManifestEntry entry =
-
entrySerializer.deserializeFromBytes(sortKey.entryBytes(row));
- if (deleteEntries.remove(entry.identifier()) != null) {
- continue;
+ entry.replace(sortKey.binaryManifestRow(row));
+ identifier.replaceWithPartition(entry);
+ if (entry.isAdd()) {
+ if (deleteEntries.contains(identifier)) {
+ matchedEntries.add(identifier);
+ } else {
+ addWriter.write(entry);
+ }
+ } else if (!matchedEntries.contains(identifier)
+ && !emittedDeletes.contains(identifier)) {
+ emittedDeletes.add(identifier);
+ deleteWriter.write(entry);
}
- writer.write(entry);
}
+ entry.clear();
+ addWriter.close();
+ newFilesForAbort.addAll(addWriter.result());
+ deleteWriter.close();
+ newFilesForAbort.addAll(deleteWriter.result());
} catch (Exception e) {
exception = e;
} finally {
+ identifier.release();
+ matchedEntries.release();
+ emittedDeletes.release();
if (exception != null) {
- writer.abort();
+ addWriter.abort();
+ deleteWriter.abort();
throw exception;
}
- writer.close();
}
- return writer.result();
+ return Pair.of(addWriter.result(), deleteWriter.result());
}
@Override
@@ -264,41 +325,4 @@ public class ManifestEntryExternalSort {
}
}
}
-
- private static List<ManifestFileMeta> sortAndWriteDeleteEntries(
- Collection<ManifestEntry> entries,
- ManifestFileSorter.ManifestSortKey sortKey,
- ManifestFile manifestFile)
- throws Exception {
- if (entries.isEmpty()) {
- return Collections.emptyList();
- }
-
- List<ManifestEntry> sorted = new ArrayList<>(entries);
- RecordComparator comparator =
- CodeGenUtils.newRecordComparator(
- sortKey.externalSortRowType().getFieldTypes(),
- sortKey.externalSortKeyFields());
- sorted.sort(
- (a, b) ->
- comparator.compare(
- sortKey.toExternalSortRow(a, new byte[0]),
- sortKey.toExternalSortRow(b, new byte[0])));
-
- RollingFileWriter<ManifestEntry, ManifestFileMeta> writer =
- manifestFile.createRollingWriter();
- Exception exception = null;
- try {
- writer.write(sorted);
- } catch (Exception e) {
- exception = e;
- } finally {
- if (exception != null) {
- writer.abort();
- throw exception;
- }
- writer.close();
- }
- return writer.result();
- }
}
diff --git
a/paimon-core/src/main/java/org/apache/paimon/operation/ManifestFileSorter.java
b/paimon-core/src/main/java/org/apache/paimon/operation/ManifestFileSorter.java
index 9bf96538e9..40a76a5914 100644
---
a/paimon-core/src/main/java/org/apache/paimon/operation/ManifestFileSorter.java
+++
b/paimon-core/src/main/java/org/apache/paimon/operation/ManifestFileSorter.java
@@ -26,7 +26,8 @@ import org.apache.paimon.data.BinaryString;
import org.apache.paimon.data.GenericRow;
import org.apache.paimon.data.InternalRow;
import org.apache.paimon.disk.IOManager;
-import org.apache.paimon.manifest.FileEntry;
+import org.apache.paimon.manifest.BinaryManifestEntry;
+import org.apache.paimon.manifest.DeletedIdentifierSet;
import org.apache.paimon.manifest.ManifestEntry;
import org.apache.paimon.manifest.ManifestFile;
import org.apache.paimon.manifest.ManifestFileMeta;
@@ -34,6 +35,7 @@ import org.apache.paimon.partition.PartitionPredicate;
import org.apache.paimon.types.DataType;
import org.apache.paimon.types.DataTypes;
import org.apache.paimon.types.RowType;
+import org.apache.paimon.utils.CloseableIterator;
import org.apache.paimon.utils.Pair;
import org.slf4j.Logger;
@@ -54,6 +56,9 @@ import java.util.Map;
import java.util.Optional;
import java.util.PriorityQueue;
import java.util.Set;
+import java.util.function.Function;
+
+import static
org.apache.paimon.utils.ManifestReadThreadPool.sequentialBatchedExecute;
/**
* Manifest file sorter that sorts and rewrites manifest files by a configured
partition field, or
@@ -62,13 +67,12 @@ import java.util.Set;
public class ManifestFileSorter {
private static final Logger LOG =
LoggerFactory.getLogger(ManifestFileSorter.class);
-
/** Context object that carries shared state across compaction methods. */
static class CompactionContext {
final boolean fullCompaction;
final ManifestSortKey sortKey;
final ManifestEntryExternalSort.ExternalSortConfig externalSortConfig;
- final Set<FileEntry.Identifier> deleteEntries;
+ final DeletedIdentifierSet deleteEntries;
/**
* Manifest files that need unsorted compaction.
*
@@ -86,7 +90,7 @@ public class ManifestFileSorter {
boolean fullCompaction,
ManifestSortKey sortKey,
ManifestEntryExternalSort.ExternalSortConfig
externalSortConfig,
- Set<FileEntry.Identifier> deleteEntries,
+ DeletedIdentifierSet deleteEntries,
Map<ManifestFileMeta, Boolean> compactWithoutSort,
List<ManifestAdjacentSortedRun> levelRuns,
List<ManifestAdjacentSortedRun> pickedRuns) {
@@ -108,7 +112,7 @@ public class ManifestFileSorter {
/** Result of classifying manifest files. */
private static class ClassifyResult {
final List<ManifestFileMeta> lsmFiles;
- final Set<FileEntry.Identifier> deleteEntries;
+ final DeletedIdentifierSet deleteEntries;
/**
* Manifest files that need unsorted compaction.
*
@@ -121,7 +125,7 @@ public class ManifestFileSorter {
ClassifyResult(
List<ManifestFileMeta> lsmFiles,
- Set<FileEntry.Identifier> deleteEntries,
+ DeletedIdentifierSet deleteEntries,
Map<ManifestFileMeta, Boolean> compactWithoutSort) {
this.lsmFiles = lsmFiles;
this.deleteEntries = deleteEntries;
@@ -129,6 +133,17 @@ public class ManifestFileSorter {
}
}
+ /** Binary identifiers and partition values collected from DELETE entries.
*/
+ private static class DeletedEntryInfo {
+ final DeletedIdentifierSet identifiers;
+ final Set<BinaryRow> partitions;
+
+ private DeletedEntryInfo(DeletedIdentifierSet identifiers,
Set<BinaryRow> partitions) {
+ this.identifiers = identifiers;
+ this.partitions = partitions;
+ }
+ }
+
/**
* Try to sort-rewrite the merged manifest list by a configured partition
field. If the sort
* field cannot be resolved, the input is returned as-is.
@@ -496,19 +511,20 @@ public class ManifestFileSorter {
// Initialize classification containers and read delete entries
Map<ManifestFileMeta, Boolean> compactWithoutSort = new
LinkedHashMap<>();
List<ManifestFileMeta> lsmFiles = new LinkedList<>(input);
- Set<FileEntry.Identifier> classifiedDeleteEntries =
Collections.emptySet();
+ DeletedIdentifierSet classifiedDeleteEntries = new
DeletedIdentifierSet();
+ Set<BinaryRow> deletePartitions = Collections.emptySet();
PartitionPredicate predicate = null;
if (fullCompaction) {
- classifiedDeleteEntries =
- FileEntry.readDeletedEntries(manifestFile, input,
manifestReadParallelism);
+ DeletedEntryInfo deletedEntries =
+ readDeletedEntries(manifestFile, input,
manifestReadParallelism);
+ classifiedDeleteEntries = deletedEntries.identifiers;
+ deletePartitions = deletedEntries.partitions;
// Build partition predicate from delete entries for overlap
detection.
if (classifiedDeleteEntries.isEmpty()) {
predicate = PartitionPredicate.ALWAYS_FALSE;
} else {
if (partitionType.getFieldCount() > 0) {
- Set<BinaryRow> deletePartitions =
-
ManifestFileMerger.computeDeletePartitions(classifiedDeleteEntries);
predicate = PartitionPredicate.fromMultiple(partitionType,
deletePartitions);
} else {
predicate = PartitionPredicate.ALWAYS_TRUE;
@@ -537,6 +553,71 @@ public class ManifestFileSorter {
return new ClassifyResult(lsmFiles, classifiedDeleteEntries,
compactWithoutSort);
}
+ private static DeletedEntryInfo readDeletedEntries(
+ ManifestFile manifestFile,
+ List<ManifestFileMeta> manifestFiles,
+ @Nullable Integer manifestReadParallelism) {
+ DeletedIdentifierSet identifiers = new DeletedIdentifierSet();
+ Set<BinaryRow> partitions = new HashSet<>();
+ List<ManifestFileMeta> filesWithDeletes = new ArrayList<>();
+ for (ManifestFileMeta meta : manifestFiles) {
+ if (meta.numDeletedFiles() > 0) {
+ filesWithDeletes.add(meta);
+ }
+ }
+
+ if (filesWithDeletes.size() <= 1
+ || (manifestReadParallelism != null && manifestReadParallelism
<= 1)) {
+ for (ManifestFileMeta meta : filesWithDeletes) {
+ collectDeletedEntries(meta, manifestFile, identifiers,
partitions, false);
+ }
+ } else {
+ Function<ManifestFileMeta, List<Boolean>> reader =
+ meta -> {
+ collectDeletedEntries(meta, manifestFile, identifiers,
partitions, true);
+ return Collections.singletonList(Boolean.TRUE);
+ };
+ for (Boolean ignored :
+ sequentialBatchedExecute(reader, filesWithDeletes,
manifestReadParallelism)) {
+ // Iteration waits for each bounded batch of parallel reads.
+ }
+ }
+ return new DeletedEntryInfo(identifiers, partitions);
+ }
+
+ private static void collectDeletedEntries(
+ ManifestFileMeta meta,
+ ManifestFile manifestFile,
+ DeletedIdentifierSet identifiers,
+ Set<BinaryRow> partitions,
+ boolean synchronize) {
+ try (CloseableIterator<BinaryManifestEntry> entries =
+ manifestFile.scan(
+ meta.fileName(),
+ meta.fileSize(),
+ BinaryManifestEntry.DELETE_ENTRY_PROJECTION)) {
+ while (entries.hasNext()) {
+ BinaryManifestEntry entry = entries.next();
+ if (!entry.isDelete()) {
+ continue;
+ }
+ BinaryRow partition = entry.partition().copy();
+ if (synchronize) {
+ synchronized (identifiers) {
+ identifiers.add(entry);
+ partitions.add(partition);
+ }
+ } else {
+ identifiers.add(entry);
+ partitions.add(partition);
+ }
+ }
+ } catch (Exception e) {
+ throw new RuntimeException(
+ String.format("Failed to scan manifest file '%s'.",
meta.fileName()), e);
+ }
+ }
+
/**
* Build level-sorted runs from a list of manifest files. Sorts files by
min partition value,
* greedy-scans to build non-overlapping SortedRuns, then assigns levels
by totalSize (Top-4
@@ -1115,9 +1196,10 @@ public class ManifestFileSorter {
int[] externalSortKeyFields();
- InternalRow toExternalSortRow(ManifestEntry entry, byte[] entryBytes);
+ void replaceExternalSortRow(
+ GenericRow row, ManifestEntry entry, InternalRow
binaryManifestRow);
- byte[] entryBytes(BinaryRow row);
+ InternalRow binaryManifestRow(BinaryRow row);
}
private static class PartitionSortKey implements ManifestSortKey {
@@ -1139,7 +1221,7 @@ public class ManifestFileSorter {
sortFieldType,
DataTypes.TINYINT(),
DataTypes.STRING(),
- DataTypes.BYTES());
+ ManifestEntry.MANIFEST_ROW_TYPE);
this.externalSortKeyFields = createSequentialFields(sortFieldNum);
}
@@ -1173,18 +1255,21 @@ public class ManifestFileSorter {
}
@Override
- public InternalRow toExternalSortRow(ManifestEntry entry, byte[]
entryBytes) {
- GenericRow row = new
GenericRow(externalSortRowType.getFieldCount());
+ public void replaceExternalSortRow(
+ GenericRow row, ManifestEntry entry, InternalRow
binaryManifestRow) {
row.setField(0, sortFieldGetter.getFieldOrNull(entry.partition()));
row.setField(1, entry.kind().toByteValue());
- row.setField(2, BinaryString.fromString(entry.file().fileName()));
- row.setField(3, entryBytes);
- return row;
+ row.setField(
+ 2,
+ entry instanceof BinaryManifestEntry
+ ? ((BinaryManifestEntry)
entry).file().fileNameBinary()
+ :
BinaryString.fromString(entry.file().fileName()));
+ row.setField(3, binaryManifestRow);
}
@Override
- public byte[] entryBytes(BinaryRow row) {
- return row.getBinary(sortFieldNum);
+ public InternalRow binaryManifestRow(BinaryRow row) {
+ return row.getRow(sortFieldNum,
ManifestEntry.MANIFEST_ROW_TYPE.getFieldCount());
}
}
@@ -1208,12 +1293,15 @@ public class ManifestFileSorter {
for (int partitionSortField : partitionSortFields) {
fieldTypes.add(partitionType.getTypeAt(partitionSortField));
}
+ // ADD must precede DELETE for the same partition. Minor
compaction streams the sorted
+ // rows once and uses this ordering to eliminate a matching pair
without retaining all
+ // ADD identifiers.
+ fieldTypes.add(DataTypes.TINYINT());
fieldTypes.add(DataTypes.BIGINT());
fieldTypes.add(DataTypes.BIGINT());
fieldTypes.add(DataTypes.BIGINT());
- fieldTypes.add(DataTypes.TINYINT());
fieldTypes.add(DataTypes.STRING());
- fieldTypes.add(DataTypes.BYTES());
+ fieldTypes.add(ManifestEntry.MANIFEST_ROW_TYPE);
this.externalSortRowType = DataTypes.ROW(fieldTypes.toArray(new
DataType[0]));
this.sortFieldNum = externalSortRowType.getFieldCount() - 1;
this.externalSortKeyFields = createSequentialFields(sortFieldNum);
@@ -1262,24 +1350,27 @@ public class ManifestFileSorter {
}
@Override
- public InternalRow toExternalSortRow(ManifestEntry entry, byte[]
entryBytes) {
- GenericRow row = new
GenericRow(externalSortRowType.getFieldCount());
+ public void replaceExternalSortRow(
+ GenericRow row, ManifestEntry entry, InternalRow
binaryManifestRow) {
int pos = 0;
for (InternalRow.FieldGetter partitionFieldGetter :
partitionFieldGetters) {
row.setField(pos++,
partitionFieldGetter.getFieldOrNull(entry.partition()));
}
+ row.setField(pos++, entry.kind().toByteValue());
row.setField(pos++, entry.file().nonNullFirstRowId());
row.setField(pos++, rowIdRangeEnd(entry));
row.setField(pos++, Long.MAX_VALUE -
entry.file().maxSequenceNumber());
- row.setField(pos++, entry.kind().toByteValue());
- row.setField(pos++,
BinaryString.fromString(entry.file().fileName()));
- row.setField(pos, entryBytes);
- return row;
+ row.setField(
+ pos++,
+ entry instanceof BinaryManifestEntry
+ ? ((BinaryManifestEntry)
entry).file().fileNameBinary()
+ :
BinaryString.fromString(entry.file().fileName()));
+ row.setField(pos, binaryManifestRow);
}
@Override
- public byte[] entryBytes(BinaryRow row) {
- return row.getBinary(sortFieldNum);
+ public InternalRow binaryManifestRow(BinaryRow row) {
+ return row.getRow(sortFieldNum,
ManifestEntry.MANIFEST_ROW_TYPE.getFieldCount());
}
private int comparePartitionMin(ManifestFileMeta a, ManifestFileMeta
b) {
diff --git
a/paimon-core/src/test/java/org/apache/paimon/manifest/BinaryManifestEntryReusableIdentifierTest.java
b/paimon-core/src/test/java/org/apache/paimon/manifest/BinaryManifestEntryReusableIdentifierTest.java
index 4e0f5ddc12..294418cfa3 100644
---
a/paimon-core/src/test/java/org/apache/paimon/manifest/BinaryManifestEntryReusableIdentifierTest.java
+++
b/paimon-core/src/test/java/org/apache/paimon/manifest/BinaryManifestEntryReusableIdentifierTest.java
@@ -18,6 +18,7 @@
package org.apache.paimon.manifest;
+import org.apache.paimon.data.BinaryRow;
import org.apache.paimon.data.BinaryString;
import org.apache.paimon.data.GenericArray;
import org.apache.paimon.data.GenericRow;
@@ -25,7 +26,6 @@ import org.apache.paimon.io.DataFileMeta;
import org.apache.paimon.manifest.BinaryManifestEntry.ReusableIdentifier;
import org.apache.paimon.types.DataField;
import org.apache.paimon.types.RowType;
-import org.apache.paimon.utils.VersionedObjectSerializer;
import org.junit.jupiter.api.Test;
@@ -34,6 +34,7 @@ import javax.annotation.Nullable;
import java.util.Arrays;
import java.util.List;
+import static org.apache.paimon.utils.SerializationUtils.serializeBinaryRow;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
@@ -78,6 +79,22 @@ class BinaryManifestEntryReusableIdentifierTest {
.isInstanceOf(IllegalArgumentException.class);
}
+ @Test
+ void testDeletedIdentifierSetReusesIdentifierForEntryLookup() {
+ DeletedIdentifierSet identifiers = new DeletedIdentifierSet();
+ BinaryManifestEntry first = entry(1, 0, "first", new String[0], null,
null);
+ BinaryManifestEntry second = entry(2, 0, "second", new String[0],
null, null);
+
+ identifiers.add(first);
+ assertThat(identifiers.contains(first)).isTrue();
+ assertThat(identifiers.contains(second)).isFalse();
+ assertThat(identifiers.contains(first)).isTrue();
+
+ identifiers.add(second);
+ assertThat(identifiers.contains(first)).isTrue();
+ assertThat(identifiers.contains(second)).isTrue();
+ }
+
private static BinaryManifestEntry entry(
int bucket,
int level,
@@ -85,7 +102,7 @@ class BinaryManifestEntryReusableIdentifierTest {
String[] extraFiles,
@Nullable byte[] embeddedIndex,
@Nullable String externalPath) {
- RowType manifestType =
VersionedObjectSerializer.versionType(ManifestEntry.SCHEMA);
+ RowType manifestType = ManifestEntry.MANIFEST_ROW_TYPE;
RowType fileType =
DataFileMeta.SCHEMA.project(
DataFileMeta.LEVEL,
@@ -95,6 +112,7 @@ class BinaryManifestEntryReusableIdentifierTest {
DataFileMeta.EXTERNAL_PATH);
List<DataField> fields =
Arrays.asList(
+ manifestType.getField(ManifestEntry.PARTITION),
manifestType.getField(ManifestEntry.BUCKET),
manifestType.getField(ManifestEntry.FILE).newType(fileType));
Object[] extraFileValues = new Object[extraFiles.length];
@@ -105,6 +123,7 @@ class BinaryManifestEntryReusableIdentifierTest {
.createEntry()
.replace(
GenericRow.of(
+ serializeBinaryRow(BinaryRow.EMPTY_ROW),
bucket,
GenericRow.of(
level,
diff --git
a/paimon-core/src/test/java/org/apache/paimon/manifest/BinaryManifestEntryTest.java
b/paimon-core/src/test/java/org/apache/paimon/manifest/BinaryManifestEntryTest.java
index 1dfb38b6ea..07f06be10b 100644
---
a/paimon-core/src/test/java/org/apache/paimon/manifest/BinaryManifestEntryTest.java
+++
b/paimon-core/src/test/java/org/apache/paimon/manifest/BinaryManifestEntryTest.java
@@ -19,6 +19,7 @@
package org.apache.paimon.manifest;
import org.apache.paimon.data.BinaryRow;
+import org.apache.paimon.data.BinaryRowWriter;
import org.apache.paimon.data.BinaryString;
import org.apache.paimon.data.GenericArray;
import org.apache.paimon.data.GenericRow;
@@ -26,11 +27,11 @@ import org.apache.paimon.io.BinaryDataFileMeta;
import org.apache.paimon.io.DataFileMeta;
import org.apache.paimon.types.DataField;
import org.apache.paimon.types.RowType;
-import org.apache.paimon.utils.VersionedObjectSerializer;
import org.junit.jupiter.api.Test;
import java.util.ArrayList;
+import java.util.Arrays;
import java.util.List;
import static org.apache.paimon.utils.SerializationUtils.serializeBinaryRow;
@@ -138,7 +139,7 @@ public class BinaryManifestEntryTest {
BinaryRow partition = BinaryRow.EMPTY_ROW;
BinaryRow minKey = BinaryRow.EMPTY_ROW;
BinaryRow maxKey = BinaryRow.EMPTY_ROW.copy();
- RowType manifestType =
VersionedObjectSerializer.versionType(ManifestEntry.SCHEMA);
+ RowType manifestType = ManifestEntry.MANIFEST_ROW_TYPE;
RowType projectedFileType =
DataFileMeta.SCHEMA.project(
DataFileMeta.MAX_KEY, DataFileMeta.FILE_NAME,
DataFileMeta.MIN_KEY);
@@ -203,9 +204,38 @@ public class BinaryManifestEntryTest {
.hasMessageContaining("not backed by a row");
}
+ @Test
+ void testReusesPartitionAndPartitionedIdentifierViews() {
+ BinaryManifestEntry entry =
+ projection(
+ true,
+ DataFileMeta.FILE_NAME,
+ DataFileMeta.LEVEL,
+ DataFileMeta.EXTRA_FILES,
+ DataFileMeta.EMBEDDED_FILE_INDEX,
+ DataFileMeta.EXTERNAL_PATH)
+ .createEntry();
+ BinaryManifestEntry.ReusableIdentifier identifier =
+ new BinaryManifestEntry.ReusableIdentifier();
+
+ entry.replace(identityRow(partition(1)));
+ BinaryRow partitionView = entry.partition();
+ assertThat(partitionView.getInt(0)).isEqualTo(1);
+ assertThat(entry.partition()).isSameAs(partitionView);
+ identifier.replaceWithPartition(entry);
+ byte[] firstIdentifier = Arrays.copyOf(identifier.bytes(),
identifier.length());
+
+ entry.replace(identityRow(partition(2)));
+ assertThat(entry.partition()).isSameAs(partitionView);
+ assertThat(partitionView.getInt(0)).isEqualTo(2);
+ identifier.replaceWithPartition(entry);
+ assertThat(Arrays.copyOf(identifier.bytes(), identifier.length()))
+ .isNotEqualTo(firstIdentifier);
+ }
+
@Test
void testProjectionWithoutFile() {
- RowType manifestType =
VersionedObjectSerializer.versionType(ManifestEntry.SCHEMA);
+ RowType manifestType = ManifestEntry.MANIFEST_ROW_TYPE;
RowType projectedType =
new RowType(
false,
@@ -222,7 +252,7 @@ public class BinaryManifestEntryTest {
@Test
void testDoesNotValidateFileKindOnReplace() {
- RowType manifestType =
VersionedObjectSerializer.versionType(ManifestEntry.SCHEMA);
+ RowType manifestType = ManifestEntry.MANIFEST_ROW_TYPE;
RowType projectedType =
new RowType(
false,
@@ -239,7 +269,7 @@ public class BinaryManifestEntryTest {
private static BinaryManifestEntry.Projection projection(
boolean includeBucket, String... projectedFileFields) {
- RowType manifestType =
VersionedObjectSerializer.versionType(ManifestEntry.SCHEMA);
+ RowType manifestType = ManifestEntry.MANIFEST_ROW_TYPE;
List<DataField> fields = new ArrayList<>();
fields.add(manifestType.getField(ManifestEntry.KIND));
fields.add(manifestType.getField(ManifestEntry.PARTITION));
@@ -253,6 +283,27 @@ public class BinaryManifestEntryTest {
return BinaryManifestEntry.Projection.create(new RowType(false,
fields));
}
+ private static GenericRow identityRow(BinaryRow partition) {
+ return GenericRow.of(
+ FileKind.ADD.toByteValue(),
+ serializeBinaryRow(partition),
+ 3,
+ GenericRow.of(
+ BinaryString.fromString("data.parquet"),
+ 2,
+ new GenericArray(new Object[0]),
+ null,
+ null));
+ }
+
+ private static BinaryRow partition(int value) {
+ BinaryRow partition = new BinaryRow(1);
+ BinaryRowWriter writer = new BinaryRowWriter(partition);
+ writer.writeInt(0, value);
+ writer.complete();
+ return partition;
+ }
+
private static void assertUnsupported(ThrowingSupplier call, String field)
{
assertThatThrownBy(call::get)
.isInstanceOf(UnsupportedOperationException.class)
diff --git
a/paimon-common/src/test/java/org/apache/paimon/utils/DeletedIdentifierSetTest.java
b/paimon-core/src/test/java/org/apache/paimon/manifest/DeletedIdentifierSetTest.java
similarity index 93%
rename from
paimon-common/src/test/java/org/apache/paimon/utils/DeletedIdentifierSetTest.java
rename to
paimon-core/src/test/java/org/apache/paimon/manifest/DeletedIdentifierSetTest.java
index dd213580ca..0de63608e8 100644
---
a/paimon-common/src/test/java/org/apache/paimon/utils/DeletedIdentifierSetTest.java
+++
b/paimon-core/src/test/java/org/apache/paimon/manifest/DeletedIdentifierSetTest.java
@@ -16,7 +16,7 @@
* limitations under the License.
*/
-package org.apache.paimon.utils;
+package org.apache.paimon.manifest;
import org.junit.jupiter.api.Test;
@@ -83,11 +83,13 @@ class DeletedIdentifierSetTest {
void testRejectsInvalidIdentifier() {
DeletedIdentifierSet identifiers = new DeletedIdentifierSet();
- assertThatThrownBy(() -> identifiers.add(0, null, 0))
+ assertThatThrownBy(() -> identifiers.add(0, (byte[]) null, 0))
.isInstanceOf(IllegalArgumentException.class);
assertThatThrownBy(() -> identifiers.add(0, new byte[1], -1))
.isInstanceOf(IllegalArgumentException.class);
assertThatThrownBy(() -> identifiers.contains(0, new byte[1], 2))
.isInstanceOf(IllegalArgumentException.class);
+ assertThatThrownBy(() -> identifiers.add(0,
(BinaryManifestEntry.ReusableIdentifier) null))
+ .isInstanceOf(IllegalArgumentException.class);
}
}
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 4c719dfb16..7dff697e8f 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
@@ -1362,7 +1362,9 @@ public class ManifestFileMetaTest extends
ManifestFileMetaTestBase {
makeRowIdEntry(true, "survivor-row30", 0, 30, 5, 1)));
input.add(
makeManifest(
- makeRowIdEntry(false, "base-row0", 0, 0, 5, 1),
+ // A newer sequence makes this DELETE sort before its
matching ADD unless
+ // the minor-compaction key explicitly orders ADD
first.
+ makeRowIdEntry(false, "base-row0", 0, 0, 5, 2),
makeRowIdEntry(false, "old-row10", 0, 10, 5, 1),
makeRowIdEntry(true, "new-row20", 0, 20, 5, 2)));
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 c17495b649..e99cf205c1 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
@@ -34,7 +34,6 @@ import org.apache.paimon.types.RowType;
import org.apache.paimon.utils.CloseableIterator;
import org.apache.paimon.utils.FailingFileIO;
import org.apache.paimon.utils.FileStorePathFactory;
-import org.apache.paimon.utils.VersionedObjectSerializer;
import org.junit.jupiter.api.RepeatedTest;
import org.junit.jupiter.api.Test;
@@ -287,7 +286,7 @@ public class ManifestFileTest {
}
private BinaryManifestEntry.Projection projection(String...
projectedFileFields) {
- RowType manifestType =
VersionedObjectSerializer.versionType(ManifestEntry.SCHEMA);
+ RowType manifestType = ManifestEntry.MANIFEST_ROW_TYPE;
List<DataField> fields =
Arrays.asList(
manifestType.getField(ManifestEntry.KIND),
