This is an automated email from the ASF dual-hosted git repository.
leaves12138 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 ef01d08e7b [core] Persist row ID reassignment plans and mark snapshots
(#10008)
ef01d08e7b is described below
commit ef01d08e7b5794cb9536a2a9518e10c392815e61
Author: YeJunHao <[email protected]>
AuthorDate: Wed Sep 23 11:23:30 2026 +0800
[core] Persist row ID reassignment plans and mark snapshots (#10008)
---
.../data-evolution-maintenance.mdx | 10 +
.../DataEvolutionRowIdReassigner.java | 12 +-
.../append/dataevolution/RowRangeMappingIndex.java | 26 ++
.../dataevolution/SerializationAssignment.java | 205 ++++++++++
.../apache/paimon/operation/FileDeletionBase.java | 12 +
.../paimon/operation/FileStoreCommitImpl.java | 20 +-
.../apache/paimon/operation/OrphanFilesClean.java | 6 +
.../DataEvolutionRowIdReassignerTest.java | 424 +++++++++++++++++++++
.../dataevolution/RowRangeMappingIndexTest.java | 25 ++
9 files changed, 738 insertions(+), 2 deletions(-)
diff --git a/docs/docs/multimodal-table/data-evolution-maintenance.mdx
b/docs/docs/multimodal-table/data-evolution-maintenance.mdx
index 66870446aa..7666899d42 100644
--- a/docs/docs/multimodal-table/data-evolution-maintenance.mdx
+++ b/docs/docs/multimodal-table/data-evolution-maintenance.mdx
@@ -187,6 +187,16 @@ are not permanent application identifiers. The
`reassign_row_id` procedure is
documented in the [Spark](../spark/procedures) and
[Flink](../flink/procedures) procedure references.
+Each reassignment snapshot records a `row-id-reassign.plan` property
referencing a
+versioned `snapshot-<snapshotId>-<uuid>.reassign-plan` file in the table's
`manifest/`
+directory, together with a `reassign-snapshot-id` property identifying the
snapshot
+that committed it. The plan is recognized only when that ID matches the current
+snapshot ID; inherited properties on later snapshots are ignored. The file also
+stores the reassignment snapshot ID and the row-ID mappings applied to each
affected
+partition. It is retained while its
+owning snapshot or a tag referencing that snapshot is retained. Snapshot commit
+failures leave unused plans for orphan-file cleanup.
+
## File Sizing
`target-file-size` controls normal-file sizing. `blob.target-file-size` and
diff --git
a/paimon-core/src/main/java/org/apache/paimon/append/dataevolution/DataEvolutionRowIdReassigner.java
b/paimon-core/src/main/java/org/apache/paimon/append/dataevolution/DataEvolutionRowIdReassigner.java
index ec12b3fd21..e357decb6b 100644
---
a/paimon-core/src/main/java/org/apache/paimon/append/dataevolution/DataEvolutionRowIdReassigner.java
+++
b/paimon-core/src/main/java/org/apache/paimon/append/dataevolution/DataEvolutionRowIdReassigner.java
@@ -67,6 +67,7 @@ import java.util.function.Consumer;
import java.util.function.Function;
import static java.util.Collections.singletonList;
+import static
org.apache.paimon.append.dataevolution.SerializationAssignment.writeProperties;
import static
org.apache.paimon.utils.ManifestReadThreadPool.sequentialBatchedExecuteCloseable;
import static org.apache.paimon.utils.Preconditions.checkArgument;
import static org.apache.paimon.utils.Preconditions.checkState;
@@ -427,6 +428,14 @@ public class DataEvolutionRowIdReassigner {
Pair<String, Long> deltaManifestList =
manifestList.write(Collections.emptyList());
RewrittenIndexManifest rewrittenIndexManifest =
rewriteIndexManifest(assignment);
+ Map<String, String> properties =
+ writeProperties(
+ table,
+ assignment.snapshot,
+ assignment.rowIdMappings,
+ assignment.firstAssignedRowId,
+ assignment.nextRowId);
+
boolean success;
try (FileStoreCommitImpl commit =
(FileStoreCommitImpl) table.store().newCommit(commitUser,
table)) {
@@ -438,7 +447,8 @@ public class DataEvolutionRowIdReassigner {
baseManifestList,
deltaManifestList,
rewrittenIndexManifest.indexManifest,
- assignment.nextRowId);
+ assignment.nextRowId,
+ properties);
}
return new CommitAssignmentResult(
success, rewrittenDataManifests.fileCount,
rewrittenIndexManifest.indexFileCount);
diff --git
a/paimon-core/src/main/java/org/apache/paimon/append/dataevolution/RowRangeMappingIndex.java
b/paimon-core/src/main/java/org/apache/paimon/append/dataevolution/RowRangeMappingIndex.java
index c4c0f07b8c..6b02d5ea61 100644
---
a/paimon-core/src/main/java/org/apache/paimon/append/dataevolution/RowRangeMappingIndex.java
+++
b/paimon-core/src/main/java/org/apache/paimon/append/dataevolution/RowRangeMappingIndex.java
@@ -18,8 +18,11 @@
package org.apache.paimon.append.dataevolution;
+import org.apache.paimon.io.DataInputView;
+import org.apache.paimon.io.DataOutputView;
import org.apache.paimon.utils.Range;
+import java.io.IOException;
import java.util.ArrayList;
import java.util.Collections;
import java.util.Comparator;
@@ -27,6 +30,7 @@ import java.util.List;
import java.util.Optional;
import static org.apache.paimon.utils.Preconditions.checkArgument;
+import static org.apache.paimon.utils.SerializationUtils.readCount;
/** Index for row-range mappings. */
final class RowRangeMappingIndex {
@@ -166,6 +170,28 @@ final class RowRangeMappingIndex {
return index < oldStarts.length && oldStarts[index] <= oldRange.to;
}
+ void serialize(DataOutputView out) throws IOException {
+ out.writeInt(oldStarts.length);
+ for (int i = 0; i < oldStarts.length; i++) {
+ out.writeLong(oldStarts[i]);
+ out.writeLong(oldEnds[i]);
+ out.writeLong(Math.addExact(newStarts[i], newStartOffset));
+ }
+ }
+
+ static RowRangeMappingIndex deserialize(DataInputView in) throws
IOException {
+ int size = readCount(in, "row-id mappings");
+ long[] oldStarts = new long[size];
+ long[] oldEnds = new long[size];
+ long[] newStarts = new long[size];
+ for (int i = 0; i < size; i++) {
+ oldStarts[i] = in.readLong();
+ oldEnds[i] = in.readLong();
+ newStarts[i] = in.readLong();
+ }
+ return createFromOwnedArrays(oldStarts, oldEnds, newStarts);
+ }
+
private static int lowerBound(long[] sorted, long target) {
int left = 0;
int right = sorted.length;
diff --git
a/paimon-core/src/main/java/org/apache/paimon/append/dataevolution/SerializationAssignment.java
b/paimon-core/src/main/java/org/apache/paimon/append/dataevolution/SerializationAssignment.java
new file mode 100644
index 0000000000..667129e472
--- /dev/null
+++
b/paimon-core/src/main/java/org/apache/paimon/append/dataevolution/SerializationAssignment.java
@@ -0,0 +1,205 @@
+/*
+ * 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.append.dataevolution;
+
+import org.apache.paimon.Snapshot;
+import org.apache.paimon.data.BinaryRow;
+import org.apache.paimon.fs.FileIO;
+import org.apache.paimon.fs.Path;
+import org.apache.paimon.io.DataInputDeserializer;
+import org.apache.paimon.io.DataOutputViewStreamWrapper;
+import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.utils.FileStorePathFactory;
+
+import javax.annotation.Nullable;
+
+import java.io.BufferedOutputStream;
+import java.io.IOException;
+import java.io.OutputStream;
+import java.io.UncheckedIOException;
+import java.nio.ByteBuffer;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.LinkedHashMap;
+import java.util.Map;
+import java.util.UUID;
+import java.util.zip.CRC32;
+import java.util.zip.CheckedOutputStream;
+
+import static org.apache.paimon.utils.IOUtils.readFully;
+import static org.apache.paimon.utils.Preconditions.checkArgument;
+import static org.apache.paimon.utils.SerializationUtils.deserializeBinaryRow;
+import static org.apache.paimon.utils.SerializationUtils.readCount;
+import static org.apache.paimon.utils.SerializationUtils.serializeBinaryRow;
+
+/** Persisted row-id mappings and allocation bounds copied from a reassignment
attempt. */
+public final class SerializationAssignment {
+
+ /** A snapshot-local marker and reference to the plan in the manifest
directory. */
+ public static final String PLAN_FILE_PROPERTY = "row-id-reassign.plan";
+
+ public static final String REASSIGN_SNAPSHOT_ID = "reassign-snapshot-id";
+
+ private static final int VERSION = 1;
+ private static final String FILE_PREFIX = "snapshot-";
+
+ private final long snapshotId;
+ private final Map<BinaryRow, RowRangeMappingIndex> rowIdMappings;
+ private final long firstAssignedRowId;
+ private final long nextRowId;
+
+ private SerializationAssignment(
+ long snapshotId,
+ Map<BinaryRow, RowRangeMappingIndex> rowIdMappings,
+ long firstAssignedRowId,
+ long nextRowId) {
+ checkArgument(
+ snapshotId >= Snapshot.FIRST_SNAPSHOT_ID,
+ "Invalid reassignment snapshot ID: %s.",
+ snapshotId);
+ checkArgument(!rowIdMappings.isEmpty(), "Reassignment mappings must
not be empty.");
+ checkArgument(
+ firstAssignedRowId >= 0 && nextRowId > firstAssignedRowId,
+ "Invalid assigned row-id range [%s, %s).",
+ firstAssignedRowId,
+ nextRowId);
+ this.snapshotId = snapshotId;
+ this.rowIdMappings = Collections.unmodifiableMap(new
LinkedHashMap<>(rowIdMappings));
+ this.firstAssignedRowId = firstAssignedRowId;
+ this.nextRowId = nextRowId;
+ }
+
+ /** The reassignment snapshot that references this plan file. */
+ public long snapshotId() {
+ return snapshotId;
+ }
+
+ public long firstAssignedRowId() {
+ return firstAssignedRowId;
+ }
+
+ public long nextRowId() {
+ return nextRowId;
+ }
+
+ /** Returns a plan only for the snapshot that committed it, ignoring
inherited properties. */
+ @Nullable
+ public static String planFile(Snapshot snapshot) {
+ Map<String, String> properties = snapshot.properties();
+ if (properties == null
+ ||
!Long.toString(snapshot.id()).equals(properties.get(REASSIGN_SNAPSHOT_ID))) {
+ return null;
+ }
+ return properties.get(PLAN_FILE_PROPERTY);
+ }
+
+ /** Persists the assignment and adds its reference to this commit's
snapshot properties. */
+ static Map<String, String> writeProperties(
+ FileStoreTable table,
+ Snapshot snapshot,
+ Map<BinaryRow, RowRangeMappingIndex> rowIdMappings,
+ long firstAssignedRowId,
+ long nextRowId) {
+ long snapshotId = snapshot.id() + 1;
+ String planFile;
+ try {
+ planFile =
+ new SerializationAssignment(
+ snapshotId, rowIdMappings,
firstAssignedRowId, nextRowId)
+ .write(table.fileIO(),
table.store().pathFactory());
+ } catch (IOException e) {
+ throw new UncheckedIOException("Failed to persist row-id
reassignment plan.", e);
+ }
+ Map<String, String> properties =
+ snapshot.properties() == null
+ ? new HashMap<>()
+ : new HashMap<>(snapshot.properties());
+ properties.put(PLAN_FILE_PROPERTY, planFile);
+ properties.put(REASSIGN_SNAPSHOT_ID, Long.toString(snapshotId));
+ return properties;
+ }
+
+ /** Streams the effective mappings without materializing another copy of
the plan. */
+ private String write(FileIO fileIO, FileStorePathFactory pathFactory)
throws IOException {
+ String fileName = FILE_PREFIX + snapshotId + "-" + UUID.randomUUID() +
".reassign-plan";
+ Path path = pathFactory.toManifestFilePath(fileName);
+ // A failed create may mean another attempt owns this path. Do not
delete its plan.
+ OutputStream fileOut = fileIO.newOutputStream(path, false);
+ try (DataOutputViewStreamWrapper out =
+ new DataOutputViewStreamWrapper(new
BufferedOutputStream(fileOut))) {
+ CRC32 checksum = new CRC32();
+ DataOutputViewStreamWrapper payload =
+ new DataOutputViewStreamWrapper(new
CheckedOutputStream(out, checksum));
+ payload.writeInt(VERSION);
+ payload.writeLong(snapshotId);
+ payload.writeLong(firstAssignedRowId);
+ payload.writeLong(nextRowId);
+ payload.writeInt(rowIdMappings.size());
+ for (Map.Entry<BinaryRow, RowRangeMappingIndex> entry :
rowIdMappings.entrySet()) {
+ serializeBinaryRow(entry.getKey(), payload);
+ entry.getValue().serialize(payload);
+ }
+ payload.flush();
+ out.writeLong(checksum.getValue());
+ } catch (IOException | RuntimeException e) {
+ fileIO.deleteQuietly(path);
+ throw e;
+ }
+ return fileName;
+ }
+
+ public static SerializationAssignment readPlan(
+ FileIO fileIO, FileStorePathFactory pathFactory, String fileName)
throws IOException {
+ Path path = pathFactory.toManifestFilePath(fileName);
+ byte[] bytes = readFully(fileIO.newInputStream(path), true);
+ if (bytes.length < Long.BYTES) {
+ throw new IOException("Truncated row-id reassignment plan.");
+ }
+ int payloadSize = bytes.length - Long.BYTES;
+ CRC32 checksum = new CRC32();
+ checksum.update(bytes, 0, payloadSize);
+ if (ByteBuffer.wrap(bytes).getLong(payloadSize) !=
checksum.getValue()) {
+ throw new IOException("Row-id reassignment plan checksum
mismatch.");
+ }
+ try {
+ DataInputDeserializer in = new DataInputDeserializer(bytes, 0,
payloadSize);
+ int version = in.readInt();
+ if (version != VERSION) {
+ throw new IOException("Unsupported row-id reassignment plan
version: " + version);
+ }
+ long snapshotId = in.readLong();
+ long firstAssignedRowId = in.readLong();
+ long nextRowId = in.readLong();
+ int partitions = readCount(in, "reassignment partitions");
+ Map<BinaryRow, RowRangeMappingIndex> mappings = new
LinkedHashMap<>();
+ for (int i = 0; i < partitions; i++) {
+ BinaryRow partition = deserializeBinaryRow(in);
+ if (mappings.put(partition,
RowRangeMappingIndex.deserialize(in)) != null) {
+ throw new IOException("Duplicate partition in row-id
reassignment plan.");
+ }
+ }
+ if (in.available() != 0) {
+ throw new IOException("Unexpected trailing bytes in row-id
reassignment plan.");
+ }
+ return new SerializationAssignment(snapshotId, mappings,
firstAssignedRowId, nextRowId);
+ } catch (IllegalArgumentException e) {
+ throw new IOException("Invalid row-id reassignment plan " +
fileName, e);
+ }
+ }
+}
diff --git
a/paimon-core/src/main/java/org/apache/paimon/operation/FileDeletionBase.java
b/paimon-core/src/main/java/org/apache/paimon/operation/FileDeletionBase.java
index 1b8b245e9b..4f74342c0a 100644
---
a/paimon-core/src/main/java/org/apache/paimon/operation/FileDeletionBase.java
+++
b/paimon-core/src/main/java/org/apache/paimon/operation/FileDeletionBase.java
@@ -69,6 +69,8 @@ import java.util.concurrent.Executor;
import java.util.function.Consumer;
import java.util.function.Predicate;
+import static
org.apache.paimon.append.dataevolution.SerializationAssignment.planFile;
+
/**
* Base class for file deletion including methods for clean data files,
manifest files and empty
* data directories.
@@ -381,6 +383,11 @@ public abstract class FileDeletionBase<T extends Snapshot>
{
collectUnusedIndexManifests(snapshot, skippingSet, indexFiles,
indexManifests);
collectUnusedStatisticsManifests(snapshot, skippingSet, statistics);
+ String reassignPlan = planFile(snapshot);
+ if (reassignPlan != null && skippingSet.add(reassignPlan)) {
+ manifests.add(reassignPlan);
+ }
+
List<Runnable> tasks = new ArrayList<>();
for (String manifest : manifests) {
tasks.add(() -> manifestFile.delete(manifest));
@@ -617,6 +624,11 @@ public abstract class FileDeletionBase<T extends Snapshot>
{
.forEach(skippingSet::add);
}
+ String reassignPlan = planFile(skippingSnapshot);
+ if (reassignPlan != null) {
+ skippingSet.add(reassignPlan);
+ }
+
// statistics
if (skippingSnapshot.statistics() != null) {
skippingSet.add(skippingSnapshot.statistics());
diff --git
a/paimon-core/src/main/java/org/apache/paimon/operation/FileStoreCommitImpl.java
b/paimon-core/src/main/java/org/apache/paimon/operation/FileStoreCommitImpl.java
index 926e972cc7..e518c8f0bc 100644
---
a/paimon-core/src/main/java/org/apache/paimon/operation/FileStoreCommitImpl.java
+++
b/paimon-core/src/main/java/org/apache/paimon/operation/FileStoreCommitImpl.java
@@ -1433,6 +1433,24 @@ public class FileStoreCommitImpl implements
FileStoreCommit {
Pair<String, Long> deltaManifestList,
@Nullable String indexManifest,
@Nullable Long nextRowId) {
+ return replaceManifestList(
+ latest,
+ totalRecordCount,
+ baseManifestList,
+ deltaManifestList,
+ indexManifest,
+ nextRowId,
+ latest.properties());
+ }
+
+ public boolean replaceManifestList(
+ Snapshot latest,
+ long totalRecordCount,
+ Pair<String, Long> baseManifestList,
+ Pair<String, Long> deltaManifestList,
+ @Nullable String indexManifest,
+ @Nullable Long nextRowId,
+ @Nullable Map<String, String> properties) {
Snapshot newSnapshot =
new Snapshot(
latest.id() + 1,
@@ -1455,7 +1473,7 @@ public class FileStoreCommitImpl implements
FileStoreCommit {
latest.watermark(),
latest.statistics(),
// if empty properties, just set to null
- latest.properties(),
+ properties == null || properties.isEmpty() ? null :
properties,
nextRowId,
null);
diff --git
a/paimon-core/src/main/java/org/apache/paimon/operation/OrphanFilesClean.java
b/paimon-core/src/main/java/org/apache/paimon/operation/OrphanFilesClean.java
index 04b4ae63ff..31d80bf519 100644
---
a/paimon-core/src/main/java/org/apache/paimon/operation/OrphanFilesClean.java
+++
b/paimon-core/src/main/java/org/apache/paimon/operation/OrphanFilesClean.java
@@ -61,6 +61,7 @@ import java.util.function.Predicate;
import java.util.stream.Collectors;
import static java.util.Collections.emptyList;
+import static
org.apache.paimon.append.dataevolution.SerializationAssignment.planFile;
import static org.apache.paimon.catalog.Identifier.DEFAULT_MAIN_BRANCH;
import static org.apache.paimon.utils.ChangelogManager.CHANGELOG_PREFIX;
import static org.apache.paimon.utils.FileStorePathFactory.BUCKET_PATH_PREFIX;
@@ -329,6 +330,11 @@ public abstract class OrphanFilesClean implements
Serializable {
.forEach(name ->
usedFileWithFlagConsumer.accept(Pair.of(name, false)));
}
+ String reassignPlan = planFile(snapshot);
+ if (reassignPlan != null) {
+ usedFileWithFlagConsumer.accept(Pair.of(reassignPlan, false));
+ }
+
// statistic file
if (snapshot.statistics() != null) {
usedFileWithFlagConsumer.accept(Pair.of(snapshot.statistics(),
false));
diff --git
a/paimon-core/src/test/java/org/apache/paimon/append/dataevolution/DataEvolutionRowIdReassignerTest.java
b/paimon-core/src/test/java/org/apache/paimon/append/dataevolution/DataEvolutionRowIdReassignerTest.java
index 3cde9a1b7c..178577375f 100644
---
a/paimon-core/src/test/java/org/apache/paimon/append/dataevolution/DataEvolutionRowIdReassignerTest.java
+++
b/paimon-core/src/test/java/org/apache/paimon/append/dataevolution/DataEvolutionRowIdReassignerTest.java
@@ -21,12 +21,15 @@ package org.apache.paimon.append.dataevolution;
import org.apache.paimon.CoreOptions;
import org.apache.paimon.Snapshot;
import org.apache.paimon.catalog.Identifier;
+import org.apache.paimon.catalog.RenamingSnapshotCommit;
+import org.apache.paimon.catalog.SnapshotCommit;
import org.apache.paimon.data.BinaryRow;
import org.apache.paimon.data.BinaryString;
import org.apache.paimon.data.GenericRow;
import org.apache.paimon.data.InternalRow;
import org.apache.paimon.data.serializer.InternalRowSerializer;
import org.apache.paimon.fs.Path;
+import org.apache.paimon.fs.local.LocalFileIO;
import org.apache.paimon.globalindex.GlobalIndexBuilderUtils;
import org.apache.paimon.globalindex.ScanResult;
import org.apache.paimon.globalindex.btree.BTreeIndexOptions;
@@ -49,14 +52,20 @@ import org.apache.paimon.manifest.ManifestFile;
import org.apache.paimon.manifest.ManifestFileMeta;
import org.apache.paimon.manifest.ManifestList;
import org.apache.paimon.operation.FileStoreCommitImpl;
+import org.apache.paimon.operation.LocalOrphanFilesClean;
+import org.apache.paimon.operation.Lock;
+import org.apache.paimon.options.ExpireConfig;
import org.apache.paimon.options.MemorySize;
import org.apache.paimon.partition.PartitionPredicate;
+import org.apache.paimon.partition.PartitionStatistics;
import org.apache.paimon.predicate.Predicate;
import org.apache.paimon.predicate.PredicateBuilder;
import org.apache.paimon.schema.Schema;
import org.apache.paimon.stats.SimpleStats;
import org.apache.paimon.stats.Statistics;
+import org.apache.paimon.table.CatalogEnvironment;
import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.table.FileStoreTableFactory;
import org.apache.paimon.table.TableTestBase;
import org.apache.paimon.table.sink.BatchTableCommit;
import org.apache.paimon.table.sink.BatchTableWrite;
@@ -67,6 +76,7 @@ import org.apache.paimon.table.source.DataSplit;
import org.apache.paimon.table.source.ReadBuilder;
import org.apache.paimon.types.DataTypes;
import org.apache.paimon.types.RowType;
+import org.apache.paimon.utils.IOUtils;
import org.apache.paimon.utils.Pair;
import org.apache.paimon.utils.Range;
import org.apache.paimon.utils.SegmentsCache;
@@ -74,8 +84,13 @@ import org.apache.paimon.utils.SnapshotManager;
import org.junit.jupiter.api.Test;
+import javax.annotation.Nullable;
+
+import java.io.IOException;
+import java.io.OutputStream;
import java.lang.reflect.Field;
import java.lang.reflect.Method;
+import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
@@ -90,13 +105,19 @@ import java.util.Random;
import java.util.Set;
import java.util.UUID;
import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
+import java.util.zip.CRC32;
+import static
org.apache.paimon.append.dataevolution.SerializationAssignment.PLAN_FILE_PROPERTY;
+import static
org.apache.paimon.append.dataevolution.SerializationAssignment.REASSIGN_SNAPSHOT_ID;
+import static
org.apache.paimon.append.dataevolution.SerializationAssignment.planFile;
+import static
org.apache.paimon.append.dataevolution.SerializationAssignment.readPlan;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
@@ -643,6 +664,403 @@ public class DataEvolutionRowIdReassignerTest extends
TableTestBase {
assertThat(rowIdsByPartition).containsEntry("pt=a/", Arrays.asList(5L,
6L, 7L));
assertThat(rowIdsByPartition).containsEntry("pt=b/", Arrays.asList(8L,
9L));
assertThat(table.snapshotManager().latestSnapshot().nextRowId()).isEqualTo(10L);
+ assertPersistedPlan(table);
+ }
+
+ @Test
+ public void testInheritedPropertiesDoNotMarkReassignment() throws
Exception {
+ FileStoreTable table = createTableWithInterleavedPartitions();
+ new DataEvolutionRowIdReassigner(table).reassign();
+ Snapshot reassigned = table.snapshotManager().latestSnapshot();
+ assertThat(planFile(reassigned)).isNotNull();
+
+ compactManifests(table);
+ Snapshot compacted = table.snapshotManager().latestSnapshot();
+ assertThat(compacted.id()).isGreaterThan(reassigned.id());
+ assertThat(compacted.properties()).isEqualTo(reassigned.properties());
+ assertThat(planFile(compacted)).isNull();
+
+ try (FileStoreCommitImpl commit =
+ (FileStoreCommitImpl)
table.store().newCommit("test-rollback-plan", table)) {
+ assertThat(commit.rollbackToAsLatest(reassigned)).isTrue();
+ }
+ Snapshot rolledBack = table.snapshotManager().latestSnapshot();
+ assertThat(rolledBack.properties()).isEqualTo(reassigned.properties());
+ assertThat(planFile(rolledBack)).isNull();
+ assertThat(planFile(reassigned)).isNotNull();
+ assertThat(
+ readPlan(table.fileIO(), table.store().pathFactory(),
planFile(reassigned))
+ .snapshotId())
+ .isEqualTo(reassigned.id());
+
+ Path plan =
table.store().pathFactory().toManifestFilePath(planFile(reassigned));
+ table.newExpireSnapshots()
+
.config(ExpireConfig.builder().snapshotRetainMin(1).snapshotRetainMax(1).build())
+ .expire();
+
assertThat(table.snapshotManager().snapshotExists(reassigned.id())).isFalse();
+ assertThat(table.fileIO().exists(plan)).isFalse();
+
assertThat(planFile(table.snapshotManager().latestSnapshot())).isNull();
+ }
+
+ @Test
+ public void testReassignPlanCleanup() throws Exception {
+ FileStoreTable table = createTableWithInterleavedPartitions();
+ new DataEvolutionRowIdReassigner(table).reassign();
+ Snapshot reassigned = table.snapshotManager().latestSnapshot();
+ Path plan =
table.store().pathFactory().toManifestFilePath(planFile(reassigned));
+ Path orphan =
+ table.store()
+ .pathFactory()
+ .toManifestFilePath(
+
"snapshot-99999-00000000-0000-0000-0000-000000000000.reassign-plan");
+ table.fileIO().newOutputStream(orphan, false).close();
+
+ new LocalOrphanFilesClean(table, System.currentTimeMillis() +
2000).clean();
+ assertThat(table.fileIO().exists(plan)).isTrue();
+ assertThat(table.fileIO().exists(orphan)).isFalse();
+
+ table.createTag("reassign", reassigned.id());
+ writeOneRow(table, "c", 100);
+
assertThat(planFile(table.snapshotManager().latestSnapshot())).isNull();
+ table.newExpireSnapshots()
+
.config(ExpireConfig.builder().snapshotRetainMin(1).snapshotRetainMax(1).build())
+ .expire();
+
assertThat(table.snapshotManager().snapshotExists(reassigned.id())).isFalse();
+ assertThat(table.fileIO().exists(plan)).isTrue();
+ new LocalOrphanFilesClean(table, System.currentTimeMillis() +
2000).clean();
+ assertThat(table.fileIO().exists(plan)).isTrue();
+
+ table.deleteTag("reassign");
+ assertThat(table.fileIO().exists(plan)).isFalse();
+ }
+
+ @Test
+ public void testReassignPlanExpiresWithSnapshot() throws Exception {
+ FileStoreTable table = createTableWithInterleavedPartitions();
+ new DataEvolutionRowIdReassigner(table).reassign();
+ Path plan =
+ table.store()
+ .pathFactory()
+
.toManifestFilePath(planFile(table.snapshotManager().latestSnapshot()));
+ writeOneRow(table, "c", 100);
+ table.newExpireSnapshots()
+
.config(ExpireConfig.builder().snapshotRetainMin(1).snapshotRetainMax(1).build())
+ .expire();
+ assertThat(table.fileIO().exists(plan)).isFalse();
+ }
+
+ @Test
+ public void testReassignPlansAreIndependentAcrossBranches() throws
Exception {
+ FileStoreTable table = createTableWithInterleavedPartitions();
+ table.createTag("base", table.snapshotManager().latestSnapshot().id());
+ table.createBranch("other", "base");
+ FileStoreTable branch = table.switchToBranch("other");
+ assertThat(branch.store().pathFactory().manifestPath())
+ .isEqualTo(table.store().pathFactory().manifestPath());
+
+ new DataEvolutionRowIdReassigner(table).reassign("main-reassign");
+ new DataEvolutionRowIdReassigner(branch).reassign("branch-reassign");
+ Snapshot mainSnapshot = table.snapshotManager().latestSnapshot();
+ Snapshot branchSnapshot = branch.snapshotManager().latestSnapshot();
+ assertThat(mainSnapshot.id()).isEqualTo(branchSnapshot.id());
+
assertThat(planFile(mainSnapshot)).isNotEqualTo(planFile(branchSnapshot));
+ assertPersistedPlan(table);
+ assertPersistedPlan(branch);
+ }
+
+ @Test
+ public void testReassignAfterRollbackUsesNewPlanFile() throws Exception {
+ FileStoreTable table = createTableWithInterleavedPartitions();
+ long base = table.snapshotManager().latestSnapshot().id();
+ new DataEvolutionRowIdReassigner(table).reassign("before-rollback");
+ String oldPlan = planFile(table.snapshotManager().latestSnapshot());
+ table.rollbackTo(base);
+
assertThat(table.snapshotManager().latestSnapshot().id()).isEqualTo(base);
+
assertThat(table.fileIO().exists(table.store().pathFactory().toManifestFilePath(oldPlan)))
+ .isTrue();
+
+ new DataEvolutionRowIdReassigner(table).reassign("after-rollback");
+ Snapshot latest = table.snapshotManager().latestSnapshot();
+ assertThat(latest.id()).isEqualTo(base + 1);
+ assertThat(planFile(latest)).isNotEqualTo(oldPlan);
+ assertPersistedPlan(table);
+ }
+
+ @Test
+ public void testLosingReassignDoesNotDeleteCommittedPlan() throws
Exception {
+ FileStoreTable original = createTableWithInterleavedPartitions();
+ long targetId = original.snapshotManager().latestSnapshot().id() + 1;
+ String planPrefix = "snapshot-" + targetId + "-";
+ CountDownLatch checkedAbsent = new CountDownLatch(2);
+ Set<String> planNames = Collections.synchronizedSet(new HashSet<>());
+ LocalFileIO racingIO =
+ new LocalFileIO() {
+ @Override
+ public boolean exists(Path path) throws IOException {
+ boolean result = super.exists(path);
+ if (!result
+ && path.getName().startsWith(planPrefix)
+ && path.getName().endsWith(".reassign-plan")) {
+ // Both creates observe an absent path before
either opens its stream.
+ planNames.add(path.getName());
+ checkedAbsent.countDown();
+ awaitReassignLatch(checkedAbsent);
+ }
+ return result;
+ }
+ };
+ FileStoreTable table =
+ FileStoreTableFactory.create(racingIO, original.location(),
original.schema());
+ CountDownLatch plansWritten = new CountDownLatch(2);
+ CountDownLatch releaseLoser = new CountDownLatch(1);
+ ExecutorService executor = Executors.newFixedThreadPool(2);
+ try {
+ Future<DataEvolutionRowIdReassigner.Result> winner =
+ executor.submit(
+ () ->
+ new DataEvolutionRowIdReassigner(
+ table,
+ null,
+ () -> {
+
plansWritten.countDown();
+
awaitReassignLatch(plansWritten);
+ })
+ .reassign("winner"));
+ Future<DataEvolutionRowIdReassigner.Result> loser =
+ executor.submit(
+ () ->
+ new DataEvolutionRowIdReassigner(
+ table,
+ null,
+ () -> {
+
plansWritten.countDown();
+
awaitReassignLatch(releaseLoser);
+ })
+ .reassign("loser"));
+ assertThat(winner.get(30, TimeUnit.SECONDS).reassigned).isTrue();
+ Snapshot committed = table.snapshotManager().latestSnapshot();
+ assertThat(committed.commitUser()).isEqualTo("winner");
+ releaseLoser.countDown();
+ assertThatThrownBy(() -> loser.get(30, TimeUnit.SECONDS))
+ .isInstanceOf(ExecutionException.class)
+ .hasStackTraceContaining("OVERWRITE snapshot");
+ assertThat(planNames).hasSize(2);
+ for (String planName : planNames) {
+ assertThat(
+ table.fileIO()
+ .exists(
+ table.store()
+ .pathFactory()
+
.toManifestFilePath(planName)))
+ .isTrue();
+ }
+ new LocalOrphanFilesClean(table, System.currentTimeMillis() +
2000).clean();
+ for (String planName : planNames) {
+ assertThat(
+ table.fileIO()
+ .exists(
+ table.store()
+ .pathFactory()
+
.toManifestFilePath(planName)))
+ .isEqualTo(planName.equals(planFile(committed)));
+ }
+ assertPersistedPlan(table);
+ } finally {
+ releaseLoser.countDown();
+ executor.shutdownNow();
+ }
+ }
+
+ @Test
+ public void testCommittedPlanSurvivesFalseCommitResult() throws Exception {
+ FileStoreTable original = createTableWithInterleavedPartitions();
+ AtomicBoolean committedButReturnedFalse = new AtomicBoolean();
+ CatalogEnvironment environment =
+ new CatalogEnvironment(null, null, null, null, null, null,
false, false) {
+ @Override
+ public SnapshotCommit snapshotCommit(SnapshotManager
manager) {
+ SnapshotCommit delegate = new
RenamingSnapshotCommit(manager, Lock.empty());
+ return new SnapshotCommit() {
+ @Override
+ public boolean commit(
+ @Nullable String baseUuid,
+ Snapshot snapshot,
+ String branch,
+ List<PartitionStatistics> statistics)
+ throws Exception {
+ boolean committed =
+ delegate.commit(baseUuid, snapshot,
branch, statistics);
+ if (committed) {
+ committedButReturnedFalse.set(true);
+ return false;
+ }
+ return committed;
+ }
+
+ @Override
+ public void close() throws Exception {
+ delegate.close();
+ }
+ };
+ }
+ };
+ FileStoreTable table =
+ FileStoreTableFactory.create(
+ original.fileIO(), original.location(),
original.schema(), environment);
+ // Reuse the false-success failure model covered by
FileStoreCommitTest.
+ assertThatThrownBy(() -> new
DataEvolutionRowIdReassigner(table).reassign("false-result"))
+ .hasMessageContaining("OVERWRITE snapshot");
+ assertThat(committedButReturnedFalse).isTrue();
+ Snapshot committed = table.snapshotManager().latestSnapshot();
+ assertThat(committed.commitUser()).isEqualTo("false-result");
+ new LocalOrphanFilesClean(table, System.currentTimeMillis() +
2000).clean();
+ assertPersistedPlan(table);
+ }
+
+ private static void awaitReassignLatch(CountDownLatch latch) {
+ try {
+ assertThat(latch.await(20, TimeUnit.SECONDS))
+ .as("Concurrent reassignment barrier")
+ .isTrue();
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new RuntimeException(e);
+ }
+ }
+
+ @Test
+ public void testValidatePlanChecksumBeforeDeserialization() throws
Exception {
+ FileStoreTable table = createTableWithInterleavedPartitions();
+ new DataEvolutionRowIdReassigner(table).reassign();
+ Snapshot snapshot = table.snapshotManager().latestSnapshot();
+ Path path =
table.store().pathFactory().toManifestFilePath(planFile(snapshot));
+ byte[] bytes = IOUtils.readFully(table.fileIO().newInputStream(path),
true);
+ // Version and the three long fields precede the partition count.
+ int partitionCountOffset = Integer.BYTES + 3 * Long.BYTES;
+ int partitionSizeOffset = partitionCountOffset + Integer.BYTES;
+ int partitionSize = ByteBuffer.wrap(bytes).getInt(partitionSizeOffset);
+ int arityOffset = partitionSizeOffset + Integer.BYTES;
+ int mappingCountOffset = arityOffset + partitionSize;
+ int[] offsets = {
+ partitionCountOffset, partitionSizeOffset, arityOffset,
mappingCountOffset
+ };
+ for (int offset : offsets) {
+ for (int length : new int[] {-1, Integer.MAX_VALUE}) {
+ byte[] corrupted = bytes.clone();
+ ByteBuffer.wrap(corrupted).putInt(offset, length);
+ overwritePlan(table, path, corrupted);
+ assertThatThrownBy(
+ () ->
+ readPlan(
+ table.fileIO(),
+ table.store().pathFactory(),
+ planFile(snapshot)))
+ .as("Invalid length %s at offset %s", length, offset)
+ .isInstanceOf(IOException.class)
+ .hasMessageContaining("checksum");
+ }
+ }
+ }
+
+ @Test
+ public void testRejectInvalidReassignPlan() throws Exception {
+ FileStoreTable table = createTableWithInterleavedPartitions();
+ new DataEvolutionRowIdReassigner(table).reassign();
+ Snapshot snapshot = table.snapshotManager().latestSnapshot();
+ Path path =
table.store().pathFactory().toManifestFilePath(planFile(snapshot));
+ byte[] bytes = IOUtils.readFully(table.fileIO().newInputStream(path),
true);
+ byte[] corrupted = bytes.clone();
+ corrupted[corrupted.length - 1] ^= 1;
+ overwritePlan(table, path, corrupted);
+ assertThatThrownBy(
+ () ->
+ readPlan(
+ table.fileIO(),
+ table.store().pathFactory(),
+ planFile(snapshot)))
+ .isInstanceOf(IOException.class)
+ .hasMessageContaining("checksum");
+
+ corrupted = bytes.clone();
+ corrupted[3] = 99;
+ CRC32 checksum = new CRC32();
+ checksum.update(corrupted, 0, corrupted.length - Long.BYTES);
+ ByteBuffer.wrap(corrupted).putLong(corrupted.length - Long.BYTES,
checksum.getValue());
+ overwritePlan(table, path, corrupted);
+ assertThatThrownBy(
+ () ->
+ readPlan(
+ table.fileIO(),
+ table.store().pathFactory(),
+ planFile(snapshot)))
+ .isInstanceOf(IOException.class)
+ .hasMessageContaining("version: 99");
+
+ for (int length : new int[] {0, Long.BYTES - 1, bytes.length - 1}) {
+ overwritePlan(table, path, Arrays.copyOf(bytes, length));
+ assertThatThrownBy(
+ () ->
+ readPlan(
+ table.fileIO(),
+ table.store().pathFactory(),
+ planFile(snapshot)))
+ .isInstanceOf(IOException.class);
+ }
+ }
+
+ private long planFileCount(FileStoreTable table) throws IOException {
+ return
Arrays.stream(table.fileIO().listStatus(table.store().pathFactory().manifestPath()))
+ .filter(
+ file ->
+
file.getPath().getName().startsWith("snapshot-")
+ &&
file.getPath().getName().endsWith(".reassign-plan"))
+ .count();
+ }
+
+ private void overwritePlan(FileStoreTable table, Path path, byte[] bytes)
throws IOException {
+ try (OutputStream out = table.fileIO().newOutputStream(path, true)) {
+ out.write(bytes);
+ }
+ }
+
+ @SuppressWarnings("unchecked")
+ private void assertPersistedPlan(FileStoreTable table) throws Exception {
+ Snapshot snapshot =
Snapshot.fromJson(table.snapshotManager().latestSnapshot().toJson());
+ assertThat(snapshot.properties())
+ .containsEntry(REASSIGN_SNAPSHOT_ID,
Long.toString(snapshot.id()))
+ .containsKey(PLAN_FILE_PROPERTY);
+ String prefix = "snapshot-" + snapshot.id() + "-";
+ String fileName = planFile(snapshot);
+ assertThat(fileName).startsWith(prefix).endsWith(".reassign-plan");
+ UUID.fromString(
+ fileName.substring(prefix.length(), fileName.length() -
".reassign-plan".length()));
+ SerializationAssignment assignment =
+ readPlan(table.fileIO(), table.store().pathFactory(),
planFile(snapshot));
+ assertThat(assignment.snapshotId()).isEqualTo(snapshot.id());
+ assertThat(assignment.firstAssignedRowId())
+ .isEqualTo(table.snapshotManager().snapshot(snapshot.id() -
1).nextRowId());
+ assertThat(assignment.nextRowId()).isEqualTo(snapshot.nextRowId());
+ Map<BinaryRow, RowRangeMappingIndex> mappings =
+ (Map<BinaryRow, RowRangeMappingIndex>) fieldValue(assignment,
"rowIdMappings");
+ Map<String, ManifestEntry> previous = new HashMap<>();
+ for (ManifestEntry entry :
+ table.store().newScan().withSnapshot(snapshot.id() -
1).plan().files()) {
+ previous.put(entry.file().fileName(), entry);
+ }
+ for (ManifestEntry entry : currentEntries(table)) {
+ Range oldRange =
previous.get(entry.file().fileName()).file().nonNullRowIdRange();
+ Range newRange = entry.file().nonNullRowIdRange();
+ RowRangeMappingIndex mapping = mappings.get(entry.partition());
+ if (oldRange.equals(newRange)) {
+ if (mapping != null) {
+ assertThat(mapping.map(oldRange)).isEmpty();
+ assertThat(mapping.overlaps(oldRange)).isFalse();
+ }
+ } else {
+ assertThat(mapping).isNotNull();
+ assertThat(mapping.map(oldRange)).hasValue(newRange);
+ }
+ }
}
@Test
@@ -679,6 +1097,7 @@ public class DataEvolutionRowIdReassignerTest extends
TableTestBase {
new DataEvolutionRowIdReassigner(table)
.reassign("test-reassign-null-partition-row-id");
+ assertPersistedPlan(table);
assertThat(result.firstAssignedRowId).isEqualTo(4L);
assertThat(result.nextRowId).isEqualTo(6L);
assertThat(result.fileCount).isEqualTo(2L);
@@ -728,6 +1147,11 @@ public class DataEvolutionRowIdReassignerTest extends
TableTestBase {
.containsEntry("pt=c/", Collections.singletonList(5L));
assertThat(valueStatsByFile(table)).containsAllEntriesOf(valueStatsBefore);
assertThat(table.snapshotManager().latestSnapshot().nextRowId()).isEqualTo(11L);
+ assertPersistedPlan(table);
+ assertThat(planFileCount(table)).isEqualTo(2L);
+ new LocalOrphanFilesClean(table, System.currentTimeMillis() +
2000).clean();
+ assertThat(planFileCount(table)).isEqualTo(1L);
+ assertPersistedPlan(table);
}
@Test
diff --git
a/paimon-core/src/test/java/org/apache/paimon/append/dataevolution/RowRangeMappingIndexTest.java
b/paimon-core/src/test/java/org/apache/paimon/append/dataevolution/RowRangeMappingIndexTest.java
index 9ed385bca0..894be9d52d 100644
---
a/paimon-core/src/test/java/org/apache/paimon/append/dataevolution/RowRangeMappingIndexTest.java
+++
b/paimon-core/src/test/java/org/apache/paimon/append/dataevolution/RowRangeMappingIndexTest.java
@@ -18,6 +18,8 @@
package org.apache.paimon.append.dataevolution;
+import org.apache.paimon.io.DataInputDeserializer;
+import org.apache.paimon.io.DataOutputSerializer;
import org.apache.paimon.utils.Range;
import org.junit.jupiter.api.Test;
@@ -86,6 +88,29 @@ public class RowRangeMappingIndexTest {
assertThat(absolute.map(new Range(20, 24))).hasValue(new Range(105,
109));
}
+ @Test
+ public void testSerializeEffectiveMappingAfterMultipleShifts() throws
Exception {
+ RowRangeMappingIndex original =
+ RowRangeMappingIndex.create(
+ Arrays.asList(
+ RowRangeMappingIndex.mapping(10, 14,
0),
+ RowRangeMappingIndex.mapping(15, 19,
5),
+ RowRangeMappingIndex.mapping(30, 39,
10)))
+ .shiftNewStarts(100)
+ .shiftNewStarts(20);
+ DataOutputSerializer out = new DataOutputSerializer(128);
+ original.serialize(out);
+ byte[] serialized = out.getCopyOfBuffer();
+ RowRangeMappingIndex restored =
+ RowRangeMappingIndex.deserialize(new
DataInputDeserializer(serialized));
+
+ assertThat(restored.map(new Range(12, 17))).hasValue(new Range(122,
127));
+ assertThat(restored.map(new Range(30, 39))).hasValue(new Range(130,
139));
+ assertThat(restored.map(new Range(19, 30))).isEmpty();
+ assertThat(restored.overlaps(new Range(20, 29))).isFalse();
+ assertThat(restored.shiftNewStarts(5).map(new Range(12,
17))).hasValue(new Range(127, 132));
+ }
+
@Test
public void testPrimitiveArrayMappingsCanBeShifted() {
long[] oldStarts = {20, 30};
