From 60b7e32b4ad12a1d705287d4273d3a3a4b18ce9c Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E4=BB=9F=E5=BC=8B?= Date: Sun, 20 Sep 2026 17:17:36 +0800 Subject: [PATCH 1/3] [core] Persist row ID reassignment plans and mark snapshots --- .../data-evolution-maintenance.mdx | 5 + .../DataEvolutionRowIdReassignPlan.java | 179 +++++++++++++++++ .../DataEvolutionRowIdReassigner.java | 26 ++- .../dataevolution/RowRangeMappingIndex.java | 26 +++ .../paimon/operation/FileDeletionBase.java | 11 + .../paimon/operation/FileStoreCommitImpl.java | 25 ++- .../paimon/operation/OrphanFilesClean.java | 6 + .../DataEvolutionRowIdReassignerTest.java | 188 ++++++++++++++++++ .../RowRangeMappingIndexTest.java | 24 +++ 9 files changed, 486 insertions(+), 4 deletions(-) create mode 100644 paimon-core/src/main/java/org/apache/paimon/append/dataevolution/DataEvolutionRowIdReassignPlan.java diff --git a/docs/docs/multimodal-table/data-evolution-maintenance.mdx b/docs/docs/multimodal-table/data-evolution-maintenance.mdx index 66870446aabf..44e910423163 100644 --- a/docs/docs/multimodal-table/data-evolution-maintenance.mdx +++ b/docs/docs/multimodal-table/data-evolution-maintenance.mdx @@ -187,6 +187,11 @@ 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 plan file in the table's `manifest/` directory. The plan contains the source +and target snapshot IDs and the row-ID mappings for each affected partition. It is +retained while its snapshot or a tag referencing that snapshot is retained. + ## 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/DataEvolutionRowIdReassignPlan.java b/paimon-core/src/main/java/org/apache/paimon/append/dataevolution/DataEvolutionRowIdReassignPlan.java new file mode 100644 index 000000000000..f738b28b8140 --- /dev/null +++ b/paimon-core/src/main/java/org/apache/paimon/append/dataevolution/DataEvolutionRowIdReassignPlan.java @@ -0,0 +1,179 @@ +/* + * 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.DataInputViewStreamWrapper; +import org.apache.paimon.io.DataOutputViewStreamWrapper; +import org.apache.paimon.utils.FileStorePathFactory; +import org.apache.paimon.utils.Range; + +import javax.annotation.Nullable; + +import java.io.BufferedInputStream; +import java.io.BufferedOutputStream; +import java.io.IOException; +import java.util.Collections; +import java.util.HashMap; +import java.util.LinkedHashMap; +import java.util.Map; +import java.util.Optional; +import java.util.UUID; +import java.util.zip.CRC32; +import java.util.zip.CheckedInputStream; +import java.util.zip.CheckedOutputStream; + +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; + +/** The absolute row-id mappings applied by one committed reassignment snapshot. */ +public final class DataEvolutionRowIdReassignPlan { + + /** A snapshot-local marker and reference to the plan in the manifest directory. */ + public static final String PLAN_FILE_PROPERTY = "row-id-reassign.plan"; + + private static final int VERSION = 1; + private static final String FILE_PREFIX = "row-id-reassign-plan-"; + + private final long sourceSnapshotId; + private final long snapshotId; + private final Map mappings; + + DataEvolutionRowIdReassignPlan( + long sourceSnapshotId, long snapshotId, Map mappings) { + checkArgument( + sourceSnapshotId >= Snapshot.FIRST_SNAPSHOT_ID + && snapshotId == Math.addExact(sourceSnapshotId, 1L), + "Invalid reassignment snapshot transition %s -> %s.", + sourceSnapshotId, + snapshotId); + checkArgument(!mappings.isEmpty(), "Reassignment mappings must not be empty."); + this.sourceSnapshotId = sourceSnapshotId; + this.snapshotId = snapshotId; + this.mappings = Collections.unmodifiableMap(new LinkedHashMap<>(mappings)); + } + + public long sourceSnapshotId() { + return sourceSnapshotId; + } + + public long snapshotId() { + return snapshotId; + } + + /** Returns a mapping only when the entire range maps to a contiguous range. */ + public Optional map(BinaryRow partition, Range range) { + RowRangeMappingIndex mapping = mappings.get(partition); + return mapping == null ? Optional.empty() : mapping.map(range); + } + + public boolean overlaps(BinaryRow partition, Range range) { + RowRangeMappingIndex mapping = mappings.get(partition); + return mapping != null && mapping.overlaps(range); + } + + @Nullable + public static String planFile(Snapshot snapshot) { + return snapshot.properties() == null ? null : snapshot.properties().get(PLAN_FILE_PROPERTY); + } + + /** Reassignment describes a transition and must not be inherited by another commit. */ + @Nullable + public static Map withoutPlan(@Nullable Map properties) { + if (properties == null || !properties.containsKey(PLAN_FILE_PROPERTY)) { + return properties; + } + Map result = new HashMap<>(properties); + result.remove(PLAN_FILE_PROPERTY); + return result.isEmpty() ? null : result; + } + + /** Streams the effective mappings without materializing another copy of the plan. */ + String write(FileIO fileIO, FileStorePathFactory pathFactory) throws IOException { + String fileName = FILE_PREFIX + UUID.randomUUID(); + Path path = pathFactory.toManifestFilePath(fileName); + try (DataOutputViewStreamWrapper out = + new DataOutputViewStreamWrapper( + new BufferedOutputStream(fileIO.newOutputStream(path, false)))) { + CRC32 checksum = new CRC32(); + DataOutputViewStreamWrapper payload = + new DataOutputViewStreamWrapper(new CheckedOutputStream(out, checksum)); + payload.writeInt(VERSION); + payload.writeLong(sourceSnapshotId); + payload.writeLong(snapshotId); + payload.writeInt(mappings.size()); + for (Map.Entry entry : mappings.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 DataEvolutionRowIdReassignPlan read( + FileIO fileIO, FileStorePathFactory pathFactory, Snapshot snapshot) throws IOException { + String fileName = planFile(snapshot); + checkArgument(fileName != null, "Snapshot %s has no reassignment plan.", snapshot.id()); + try (DataInputViewStreamWrapper in = + new DataInputViewStreamWrapper( + new BufferedInputStream( + fileIO.newInputStream(pathFactory.toManifestFilePath(fileName))))) { + CRC32 checksum = new CRC32(); + DataInputViewStreamWrapper payload = + new DataInputViewStreamWrapper(new CheckedInputStream(in, checksum)); + int version = payload.readInt(); + if (version != VERSION) { + throw new IOException("Unsupported row-id reassignment plan version: " + version); + } + long sourceSnapshotId = payload.readLong(); + long snapshotId = payload.readLong(); + if (snapshotId != snapshot.id()) { + throw new IOException("Row-id reassignment plan belongs to snapshot " + snapshotId); + } + int partitions = readCount(payload, "reassignment partitions"); + Map mappings = new LinkedHashMap<>(); + for (int i = 0; i < partitions; i++) { + BinaryRow partition = deserializeBinaryRow(payload); + if (mappings.put(partition, RowRangeMappingIndex.deserialize(payload)) != null) { + throw new IOException("Duplicate partition in row-id reassignment plan."); + } + } + long actualChecksum = checksum.getValue(); + if (in.readLong() != actualChecksum) { + throw new IOException("Row-id reassignment plan checksum mismatch."); + } + if (in.read() != -1) { + throw new IOException("Unexpected trailing bytes in row-id reassignment plan."); + } + return new DataEvolutionRowIdReassignPlan(sourceSnapshotId, snapshotId, mappings); + } catch (IllegalArgumentException | ArithmeticException e) { + throw new IOException("Invalid row-id reassignment plan " + fileName, e); + } + } +} 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 ec12b3fd2131..073068314706 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 @@ -53,6 +53,8 @@ import javax.annotation.Nullable; +import java.io.IOException; +import java.io.UncheckedIOException; import java.util.ArrayList; import java.util.Collections; import java.util.Comparator; @@ -427,6 +429,23 @@ private CommitAssignmentResult commitAssignment( Pair deltaManifestList = manifestList.write(Collections.emptyList()); RewrittenIndexManifest rewrittenIndexManifest = rewriteIndexManifest(assignment); + String planFile; + try { + planFile = + new DataEvolutionRowIdReassignPlan( + assignment.snapshot.id(), + assignment.snapshot.id() + 1, + assignment.rowIdMappings) + .write(table.fileIO(), table.store().pathFactory()); + } catch (IOException e) { + throw new UncheckedIOException("Failed to persist row-id reassignment plan.", e); + } + Map properties = + assignment.snapshot.properties() == null + ? new HashMap<>() + : new HashMap<>(assignment.snapshot.properties()); + properties.put(DataEvolutionRowIdReassignPlan.PLAN_FILE_PROPERTY, planFile); + boolean success; try (FileStoreCommitImpl commit = (FileStoreCommitImpl) table.store().newCommit(commitUser, table)) { @@ -438,7 +457,12 @@ private CommitAssignmentResult commitAssignment( baseManifestList, deltaManifestList, rewrittenIndexManifest.indexManifest, - assignment.nextRowId); + assignment.nextRowId, + properties); + } + if (!success) { + // Only clean a definitively rejected attempt. An exception may mean it committed. + table.fileIO().deleteQuietly(table.store().pathFactory().toManifestFilePath(planFile)); } 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 c4c0f07b8c34..6b02d5ea61fc 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.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 @@ boolean overlaps(Range oldRange) { 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/operation/FileDeletionBase.java b/paimon-core/src/main/java/org/apache/paimon/operation/FileDeletionBase.java index 1b8b245e9b39..f73c6bee792c 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 @@ -19,6 +19,7 @@ package org.apache.paimon.operation; import org.apache.paimon.Snapshot; +import org.apache.paimon.append.dataevolution.DataEvolutionRowIdReassignPlan; import org.apache.paimon.data.BinaryRow; import org.apache.paimon.fs.FileIO; import org.apache.paimon.fs.Path; @@ -381,6 +382,11 @@ protected List planManifestsCleaner( collectUnusedIndexManifests(snapshot, skippingSet, indexFiles, indexManifests); collectUnusedStatisticsManifests(snapshot, skippingSet, statistics); + String reassignPlan = DataEvolutionRowIdReassignPlan.planFile(snapshot); + if (reassignPlan != null && skippingSet.add(reassignPlan)) { + manifests.add(reassignPlan); + } + List tasks = new ArrayList<>(); for (String manifest : manifests) { tasks.add(() -> manifestFile.delete(manifest)); @@ -617,6 +623,11 @@ private Set manifestSkippingSet(Snapshot skippingSnapshot) { .forEach(skippingSet::add); } + String reassignPlan = DataEvolutionRowIdReassignPlan.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 93092e20d7f5..ea10fa16438d 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 @@ -22,6 +22,7 @@ import org.apache.paimon.Snapshot; import org.apache.paimon.Snapshot.CommitKind; import org.apache.paimon.annotation.VisibleForTesting; +import org.apache.paimon.append.dataevolution.DataEvolutionRowIdReassignPlan; import org.apache.paimon.catalog.SnapshotCommit; import org.apache.paimon.data.BinaryRow; import org.apache.paimon.data.InternalRow; @@ -1394,6 +1395,24 @@ public boolean replaceManifestList( Pair deltaManifestList, @Nullable String indexManifest, @Nullable Long nextRowId) { + return replaceManifestList( + latest, + totalRecordCount, + baseManifestList, + deltaManifestList, + indexManifest, + nextRowId, + DataEvolutionRowIdReassignPlan.withoutPlan(latest.properties())); + } + + public boolean replaceManifestList( + Snapshot latest, + long totalRecordCount, + Pair baseManifestList, + Pair deltaManifestList, + @Nullable String indexManifest, + @Nullable Long nextRowId, + @Nullable Map properties) { Snapshot newSnapshot = new Snapshot( latest.id() + 1, @@ -1416,7 +1435,7 @@ public boolean replaceManifestList( latest.watermark(), latest.statistics(), // if empty properties, just set to null - latest.properties(), + properties == null || properties.isEmpty() ? null : properties, nextRowId, null); @@ -1502,7 +1521,7 @@ public boolean rollbackToAsLatest(Snapshot targetSnapshot) { null, targetSnapshot.watermark(), targetSnapshot.statistics(), - targetSnapshot.properties(), + DataEvolutionRowIdReassignPlan.withoutPlan(targetSnapshot.properties()), nextRowId, null); @@ -1652,7 +1671,7 @@ private boolean compactManifestOnce() { null, latestSnapshot.watermark(), latestSnapshot.statistics(), - latestSnapshot.properties(), + DataEvolutionRowIdReassignPlan.withoutPlan(latestSnapshot.properties()), latestSnapshot.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 04b4ae63ff06..a6e22971eb04 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 @@ -19,6 +19,7 @@ package org.apache.paimon.operation; import org.apache.paimon.Snapshot; +import org.apache.paimon.append.dataevolution.DataEvolutionRowIdReassignPlan; import org.apache.paimon.blob.ManagedBlobReferenceFile; import org.apache.paimon.data.Timestamp; import org.apache.paimon.fs.FileIO; @@ -329,6 +330,11 @@ protected void collectWithoutDataFileWithManifestFlag( .forEach(name -> usedFileWithFlagConsumer.accept(Pair.of(name, false))); } + String reassignPlan = DataEvolutionRowIdReassignPlan.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 3cde9a1b7cd8..d9c656134167 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 @@ -49,6 +49,8 @@ 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.options.ExpireConfig; import org.apache.paimon.options.MemorySize; import org.apache.paimon.partition.PartitionPredicate; import org.apache.paimon.predicate.Predicate; @@ -67,6 +69,7 @@ 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,6 +77,8 @@ import org.junit.jupiter.api.Test; +import java.io.IOException; +import java.io.OutputStream; import java.lang.reflect.Field; import java.lang.reflect.Method; import java.util.ArrayList; @@ -643,6 +648,178 @@ public void testReassignRowIdsByPartition() throws Exception { 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 testReassignPlanIsNotInherited() throws Exception { + FileStoreTable table = createTableWithInterleavedPartitions(); + new DataEvolutionRowIdReassigner(table).reassign(); + Snapshot reassigned = table.snapshotManager().latestSnapshot(); + assertThat(DataEvolutionRowIdReassignPlan.planFile(reassigned)).isNotNull(); + + compactManifests(table); + Snapshot compacted = table.snapshotManager().latestSnapshot(); + assertThat(compacted.id()).isGreaterThan(reassigned.id()); + assertThat(DataEvolutionRowIdReassignPlan.planFile(compacted)).isNull(); + + try (FileStoreCommitImpl commit = + (FileStoreCommitImpl) table.store().newCommit("test-rollback-plan", table)) { + assertThat(commit.rollbackToAsLatest(reassigned)).isTrue(); + } + assertThat( + DataEvolutionRowIdReassignPlan.planFile( + table.snapshotManager().latestSnapshot())) + .isNull(); + assertThat(DataEvolutionRowIdReassignPlan.planFile(reassigned)).isNotNull(); + DataEvolutionRowIdReassignPlan.read( + table.fileIO(), table.store().pathFactory(), reassigned); + } + + @Test + public void testReassignPlanCleanup() throws Exception { + FileStoreTable table = createTableWithInterleavedPartitions(); + new DataEvolutionRowIdReassigner(table).reassign(); + Snapshot reassigned = table.snapshotManager().latestSnapshot(); + Path plan = + table.store() + .pathFactory() + .toManifestFilePath(DataEvolutionRowIdReassignPlan.planFile(reassigned)); + Path orphan = table.store().pathFactory().toManifestFilePath("row-id-reassign-plan-orphan"); + 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( + DataEvolutionRowIdReassignPlan.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( + DataEvolutionRowIdReassignPlan.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 testRejectInvalidReassignPlan() throws Exception { + FileStoreTable table = createTableWithInterleavedPartitions(); + new DataEvolutionRowIdReassigner(table).reassign(); + Snapshot snapshot = table.snapshotManager().latestSnapshot(); + Path path = + table.store() + .pathFactory() + .toManifestFilePath(DataEvolutionRowIdReassignPlan.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( + () -> + DataEvolutionRowIdReassignPlan.read( + table.fileIO(), table.store().pathFactory(), snapshot)) + .isInstanceOf(IOException.class) + .hasMessageContaining("checksum"); + + corrupted = bytes.clone(); + corrupted[3] = 99; + overwritePlan(table, path, corrupted); + assertThatThrownBy( + () -> + DataEvolutionRowIdReassignPlan.read( + table.fileIO(), table.store().pathFactory(), snapshot)) + .isInstanceOf(IOException.class) + .hasMessageContaining("version: 99"); + + // The target snapshot ID occupies bytes 12 through 19. + corrupted = bytes.clone(); + corrupted[19] ^= 1; + overwritePlan(table, path, corrupted); + assertThatThrownBy( + () -> + DataEvolutionRowIdReassignPlan.read( + table.fileIO(), table.store().pathFactory(), snapshot)) + .isInstanceOf(IOException.class) + .hasMessageContaining("belongs to snapshot"); + + overwritePlan(table, path, Arrays.copyOf(bytes, bytes.length - 1)); + assertThatThrownBy( + () -> + DataEvolutionRowIdReassignPlan.read( + table.fileIO(), table.store().pathFactory(), snapshot)) + .isInstanceOf(IOException.class); + } + + @Test + public void testRemovingReassignMarkerPreservesOtherProperties() { + Map properties = new HashMap<>(); + properties.put(DataEvolutionRowIdReassignPlan.PLAN_FILE_PROPERTY, "plan"); + properties.put("sequence.generation.max-sequence-number", "100"); + assertThat(DataEvolutionRowIdReassignPlan.withoutPlan(properties)) + .containsExactlyEntriesOf( + Collections.singletonMap("sequence.generation.max-sequence-number", "100")); + assertThat(properties).hasSize(2); + assertThat( + DataEvolutionRowIdReassignPlan.withoutPlan( + Collections.singletonMap( + DataEvolutionRowIdReassignPlan.PLAN_FILE_PROPERTY, "plan"))) + .isNull(); + } + + private void overwritePlan(FileStoreTable table, Path path, byte[] bytes) throws IOException { + try (OutputStream out = table.fileIO().newOutputStream(path, true)) { + out.write(bytes); + } + } + + private void assertPersistedPlan(FileStoreTable table) throws Exception { + Snapshot snapshot = Snapshot.fromJson(table.snapshotManager().latestSnapshot().toJson()); + DataEvolutionRowIdReassignPlan plan = + DataEvolutionRowIdReassignPlan.read( + table.fileIO(), table.store().pathFactory(), snapshot); + assertThat(plan.snapshotId()).isEqualTo(snapshot.id()); + assertThat(plan.sourceSnapshotId()).isEqualTo(snapshot.id() - 1); + Map previous = new HashMap<>(); + for (ManifestEntry entry : + table.store().newScan().withSnapshot(plan.sourceSnapshotId()).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(); + if (oldRange.equals(newRange)) { + assertThat(plan.map(entry.partition(), oldRange)).isEmpty(); + assertThat(plan.overlaps(entry.partition(), oldRange)).isFalse(); + } else { + assertThat(plan.map(entry.partition(), oldRange)).hasValue(newRange); + } + } } @Test @@ -679,6 +856,7 @@ public void testReassignNullPartitionAcrossManifestGroups() throws Exception { 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 +906,16 @@ public void testReassignRetriesWithLatestNextRowIdAfterConcurrentAppend() throws .containsEntry("pt=c/", Collections.singletonList(5L)); assertThat(valueStatsByFile(table)).containsAllEntriesOf(valueStatsBefore); assertThat(table.snapshotManager().latestSnapshot().nextRowId()).isEqualTo(11L); + assertPersistedPlan(table); + long planFiles = + Arrays.stream(table.fileIO().listStatus(table.store().pathFactory().manifestPath())) + .filter( + file -> + file.getPath() + .getName() + .startsWith("row-id-reassign-plan-")) + .count(); + assertThat(planFiles).isEqualTo(1L); } @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 9ed385bca0e9..61cd1cb69a51 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,28 @@ public void testRelativeMappingCanBeShiftedToAbsoluteRowIds() { 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); + RowRangeMappingIndex restored = + RowRangeMappingIndex.deserialize(new DataInputDeserializer(out.getCopyOfBuffer())); + + 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}; From 0fb305537ba9a104550bf4883b55372be4d4b5d9 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E4=BB=9F=E5=BC=8B?= Date: Sun, 20 Sep 2026 17:50:44 +0800 Subject: [PATCH 2/3] [core] Serialize the existing row ID assignment directly --- .../data-evolution-maintenance.mdx | 6 +- ...java => DataEvolutionRowIdAssignment.java} | 78 ++++++----- .../DataEvolutionRowIdReassigner.java | 130 ++++++++---------- .../paimon/operation/FileDeletionBase.java | 6 +- .../paimon/operation/FileStoreCommitImpl.java | 8 +- .../paimon/operation/OrphanFilesClean.java | 4 +- .../DataEvolutionRowIdReassignerTest.java | 84 ++++++----- 7 files changed, 149 insertions(+), 167 deletions(-) rename paimon-core/src/main/java/org/apache/paimon/append/dataevolution/{DataEvolutionRowIdReassignPlan.java => DataEvolutionRowIdAssignment.java} (73%) diff --git a/docs/docs/multimodal-table/data-evolution-maintenance.mdx b/docs/docs/multimodal-table/data-evolution-maintenance.mdx index 44e910423163..2ce37c933843 100644 --- a/docs/docs/multimodal-table/data-evolution-maintenance.mdx +++ b/docs/docs/multimodal-table/data-evolution-maintenance.mdx @@ -188,9 +188,9 @@ 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 plan file in the table's `manifest/` directory. The plan contains the source -and target snapshot IDs and the row-ID mappings for each affected partition. It is -retained while its snapshot or a tag referencing that snapshot is retained. +versioned plan file in the table's `manifest/` directory. The file stores the row-ID +mappings applied to each affected partition. It is retained while its snapshot or a +tag referencing that snapshot is retained. ## File Sizing diff --git a/paimon-core/src/main/java/org/apache/paimon/append/dataevolution/DataEvolutionRowIdReassignPlan.java b/paimon-core/src/main/java/org/apache/paimon/append/dataevolution/DataEvolutionRowIdAssignment.java similarity index 73% rename from paimon-core/src/main/java/org/apache/paimon/append/dataevolution/DataEvolutionRowIdReassignPlan.java rename to paimon-core/src/main/java/org/apache/paimon/append/dataevolution/DataEvolutionRowIdAssignment.java index f738b28b8140..9b9f6fe5d1c1 100644 --- a/paimon-core/src/main/java/org/apache/paimon/append/dataevolution/DataEvolutionRowIdReassignPlan.java +++ b/paimon-core/src/main/java/org/apache/paimon/append/dataevolution/DataEvolutionRowIdAssignment.java @@ -47,8 +47,8 @@ import static org.apache.paimon.utils.SerializationUtils.readCount; import static org.apache.paimon.utils.SerializationUtils.serializeBinaryRow; -/** The absolute row-id mappings applied by one committed reassignment snapshot. */ -public final class DataEvolutionRowIdReassignPlan { +/** The row-id assignment used to rewrite metadata and persisted for subsequent commits. */ +public final class DataEvolutionRowIdAssignment { /** A snapshot-local marker and reference to the plan in the manifest directory. */ public static final String PLAN_FILE_PROPERTY = "row-id-reassign.plan"; @@ -56,40 +56,49 @@ public final class DataEvolutionRowIdReassignPlan { private static final int VERSION = 1; private static final String FILE_PREFIX = "row-id-reassign-plan-"; - private final long sourceSnapshotId; - private final long snapshotId; - private final Map mappings; + private final Map rowIdMappings; + private final long firstAssignedRowId; + private final long nextRowId; - DataEvolutionRowIdReassignPlan( - long sourceSnapshotId, long snapshotId, Map mappings) { + DataEvolutionRowIdAssignment( + Map rowIdMappings, + long firstAssignedRowId, + long nextRowId) { + checkArgument(!rowIdMappings.isEmpty(), "Reassignment mappings must not be empty."); checkArgument( - sourceSnapshotId >= Snapshot.FIRST_SNAPSHOT_ID - && snapshotId == Math.addExact(sourceSnapshotId, 1L), - "Invalid reassignment snapshot transition %s -> %s.", - sourceSnapshotId, - snapshotId); - checkArgument(!mappings.isEmpty(), "Reassignment mappings must not be empty."); - this.sourceSnapshotId = sourceSnapshotId; - this.snapshotId = snapshotId; - this.mappings = Collections.unmodifiableMap(new LinkedHashMap<>(mappings)); + firstAssignedRowId >= 0 && nextRowId > firstAssignedRowId, + "Invalid assigned row-id range [%s, %s).", + firstAssignedRowId, + nextRowId); + this.rowIdMappings = Collections.unmodifiableMap(new LinkedHashMap<>(rowIdMappings)); + this.firstAssignedRowId = firstAssignedRowId; + this.nextRowId = nextRowId; } - public long sourceSnapshotId() { - return sourceSnapshotId; + Map rowIdMappings() { + return rowIdMappings; } - public long snapshotId() { - return snapshotId; + public long firstAssignedRowId() { + return firstAssignedRowId; + } + + public long nextRowId() { + return nextRowId; + } + + public long logicalRowCount() { + return nextRowId - firstAssignedRowId; } /** Returns a mapping only when the entire range maps to a contiguous range. */ public Optional map(BinaryRow partition, Range range) { - RowRangeMappingIndex mapping = mappings.get(partition); + RowRangeMappingIndex mapping = rowIdMappings.get(partition); return mapping == null ? Optional.empty() : mapping.map(range); } public boolean overlaps(BinaryRow partition, Range range) { - RowRangeMappingIndex mapping = mappings.get(partition); + RowRangeMappingIndex mapping = rowIdMappings.get(partition); return mapping != null && mapping.overlaps(range); } @@ -120,10 +129,10 @@ String write(FileIO fileIO, FileStorePathFactory pathFactory) throws IOException DataOutputViewStreamWrapper payload = new DataOutputViewStreamWrapper(new CheckedOutputStream(out, checksum)); payload.writeInt(VERSION); - payload.writeLong(sourceSnapshotId); - payload.writeLong(snapshotId); - payload.writeInt(mappings.size()); - for (Map.Entry entry : mappings.entrySet()) { + payload.writeLong(firstAssignedRowId); + payload.writeLong(nextRowId); + payload.writeInt(rowIdMappings.size()); + for (Map.Entry entry : rowIdMappings.entrySet()) { serializeBinaryRow(entry.getKey(), payload); entry.getValue().serialize(payload); } @@ -136,10 +145,8 @@ String write(FileIO fileIO, FileStorePathFactory pathFactory) throws IOException return fileName; } - public static DataEvolutionRowIdReassignPlan read( - FileIO fileIO, FileStorePathFactory pathFactory, Snapshot snapshot) throws IOException { - String fileName = planFile(snapshot); - checkArgument(fileName != null, "Snapshot %s has no reassignment plan.", snapshot.id()); + public static DataEvolutionRowIdAssignment read( + FileIO fileIO, FileStorePathFactory pathFactory, String fileName) throws IOException { try (DataInputViewStreamWrapper in = new DataInputViewStreamWrapper( new BufferedInputStream( @@ -151,11 +158,8 @@ public static DataEvolutionRowIdReassignPlan read( if (version != VERSION) { throw new IOException("Unsupported row-id reassignment plan version: " + version); } - long sourceSnapshotId = payload.readLong(); - long snapshotId = payload.readLong(); - if (snapshotId != snapshot.id()) { - throw new IOException("Row-id reassignment plan belongs to snapshot " + snapshotId); - } + long firstAssignedRowId = payload.readLong(); + long nextRowId = payload.readLong(); int partitions = readCount(payload, "reassignment partitions"); Map mappings = new LinkedHashMap<>(); for (int i = 0; i < partitions; i++) { @@ -171,8 +175,8 @@ public static DataEvolutionRowIdReassignPlan read( if (in.read() != -1) { throw new IOException("Unexpected trailing bytes in row-id reassignment plan."); } - return new DataEvolutionRowIdReassignPlan(sourceSnapshotId, snapshotId, mappings); - } catch (IllegalArgumentException | ArithmeticException e) { + return new DataEvolutionRowIdAssignment(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/append/dataevolution/DataEvolutionRowIdReassigner.java b/paimon-core/src/main/java/org/apache/paimon/append/dataevolution/DataEvolutionRowIdReassigner.java index 073068314706..f7c8870e5b40 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 @@ -159,25 +159,25 @@ public Result reassign(String commitUser) { CommittedAssignment committed = commitAssignmentWithRetry( optionalPlan.get(), latest, manifestFile, manifestList, commitUser); - Assignment assignment = committed.assignment; + DataEvolutionRowIdAssignment assignment = committed.assignment; CommitAssignmentResult commitResult = committed.commitResult; LOG.info( "Reassigned row IDs for table {} from {} to {}, partitions={}, files={}, rows={}.", table.name(), - assignment.firstAssignedRowId, - assignment.nextRowId, - assignment.rowIdMappings.size(), + assignment.firstAssignedRowId(), + assignment.nextRowId(), + assignment.rowIdMappings().size(), commitResult.fileCount, assignment.logicalRowCount()); return new Result( - assignment.snapshot.id(), - assignment.snapshot.id() + 1, + committed.previousSnapshotId, + committed.previousSnapshotId + 1, commitResult.fileCount, assignment.logicalRowCount(), commitResult.indexFileCount, - assignment.firstAssignedRowId, - assignment.nextRowId); + assignment.firstAssignedRowId(), + assignment.nextRowId()); } private Optional planAssignment(List manifestMetas) { @@ -381,11 +381,17 @@ private CommittedAssignment commitAssignmentWithRetry( latest = observedLatest; } - Assignment assignment = assignmentPlan.createAssignment(latest); + DataEvolutionRowIdAssignment assignment = assignmentPlan.createAssignment(latest); CommitAssignmentResult commitResult = - commitAssignment(assignment, manifestFile, manifestList, commitUser); + commitAssignment( + latest, + assignmentPlan.manifestMetasToRewrite, + assignment, + manifestFile, + manifestList, + commitUser); if (commitResult.success) { - return new CommittedAssignment(assignment, commitResult); + return new CommittedAssignment(latest.id(), assignment, commitResult); } if (System.currentTimeMillis() - startMillis > options.commitTimeout() @@ -415,36 +421,33 @@ private CommittedAssignment commitAssignmentWithRetry( } private CommitAssignmentResult commitAssignment( - Assignment assignment, + Snapshot snapshot, + List manifestMetasToRewrite, + DataEvolutionRowIdAssignment assignment, ManifestFile manifestFile, ManifestList manifestList, String commitUser) { RewrittenDataManifests rewrittenDataManifests = - writeManifestReplacements(assignment, manifestFile); + writeManifestReplacements(manifestMetasToRewrite, assignment, manifestFile); Pair baseManifestList = writeBaseManifestList( - manifestList.readDataManifests(assignment.snapshot), + manifestList.readDataManifests(snapshot), rewrittenDataManifests.manifestMetas, manifestList); Pair deltaManifestList = manifestList.write(Collections.emptyList()); - RewrittenIndexManifest rewrittenIndexManifest = rewriteIndexManifest(assignment); + RewrittenIndexManifest rewrittenIndexManifest = rewriteIndexManifest(snapshot, assignment); String planFile; try { - planFile = - new DataEvolutionRowIdReassignPlan( - assignment.snapshot.id(), - assignment.snapshot.id() + 1, - assignment.rowIdMappings) - .write(table.fileIO(), table.store().pathFactory()); + planFile = assignment.write(table.fileIO(), table.store().pathFactory()); } catch (IOException e) { throw new UncheckedIOException("Failed to persist row-id reassignment plan.", e); } Map properties = - assignment.snapshot.properties() == null + snapshot.properties() == null ? new HashMap<>() - : new HashMap<>(assignment.snapshot.properties()); - properties.put(DataEvolutionRowIdReassignPlan.PLAN_FILE_PROPERTY, planFile); + : new HashMap<>(snapshot.properties()); + properties.put(DataEvolutionRowIdAssignment.PLAN_FILE_PROPERTY, planFile); boolean success; try (FileStoreCommitImpl commit = @@ -452,12 +455,12 @@ private CommitAssignmentResult commitAssignment( beforeCommit.run(); success = commit.replaceManifestList( - assignment.snapshot, - assignment.snapshot.totalRecordCount(), + snapshot, + snapshot.totalRecordCount(), baseManifestList, deltaManifestList, rewrittenIndexManifest.indexManifest, - assignment.nextRowId, + assignment.nextRowId(), properties); } if (!success) { @@ -653,13 +656,13 @@ private Pair writeBaseManifestList( } private RewrittenDataManifests writeManifestReplacements( - Assignment assignment, ManifestFile manifestFile) { + List manifestMetasToRewrite, + DataEvolutionRowIdAssignment assignment, + ManifestFile manifestFile) { Integer parallelism = table.coreOptions().scanManifestParallelism(); - List rewritten = - new ArrayList<>(assignment.manifestMetasToRewrite.size()); - if (assignment.manifestMetasToRewrite.size() == 1 - || (parallelism != null && parallelism == 1)) { - for (ManifestFileMeta manifestMeta : assignment.manifestMetasToRewrite) { + List rewritten = new ArrayList<>(manifestMetasToRewrite.size()); + if (manifestMetasToRewrite.size() == 1 || (parallelism != null && parallelism == 1)) { + for (ManifestFileMeta manifestMeta : manifestMetasToRewrite) { rewritten.add(rewriteDataManifest(assignment, manifestFile, manifestMeta)); } } else { @@ -672,7 +675,7 @@ private RewrittenDataManifests writeManifestReplacements( manifestMeta)); try (CloseableBatchIterator results = sequentialBatchedExecuteCloseable( - rewriter, assignment.manifestMetasToRewrite, parallelism)) { + rewriter, manifestMetasToRewrite, parallelism)) { while (results.hasNext()) { rewritten.add(results.next()); } @@ -689,7 +692,9 @@ private RewrittenDataManifests writeManifestReplacements( } private RewrittenDataManifest rewriteDataManifest( - Assignment assignment, ManifestFile manifestFile, ManifestFileMeta manifestMeta) { + DataEvolutionRowIdAssignment assignment, + ManifestFile manifestFile, + ManifestFileMeta manifestMeta) { beforeManifestRewrite.accept(manifestMeta); ManifestEntrySerializer serializer = new ManifestEntrySerializer(); ManifestAvroWriter writer = manifestFile.createAvroWriter(); @@ -702,7 +707,7 @@ private RewrittenDataManifest rewriteDataManifest( while (entries.hasNext()) { ProjectedManifestEntry entry = entries.next(); ManifestEntry output = entry; - RowRangeMappingIndex mapping = assignment.rowIdMappings.get(entry.partition()); + RowRangeMappingIndex mapping = assignment.rowIdMappings().get(entry.partition()); if (mapping != null) { Optional reassignedRange = mapping.map(entry.file().nonNullRowIdRange()); if (reassignedRange.isPresent()) { @@ -750,14 +755,14 @@ private void validatePlanningEntry(ManifestEntry entry) { table.name()); } - private RewrittenIndexManifest rewriteIndexManifest(Assignment assignment) { - if (assignment.snapshot.indexManifest() == null) { + private RewrittenIndexManifest rewriteIndexManifest( + Snapshot snapshot, DataEvolutionRowIdAssignment assignment) { + if (snapshot.indexManifest() == null) { return new RewrittenIndexManifest(null, 0); } IndexManifestFile indexManifestFile = table.store().indexManifestFileFactory().create(); - List indexEntries = - indexManifestFile.read(assignment.snapshot.indexManifest()); + List indexEntries = indexManifestFile.read(snapshot.indexManifest()); if (indexEntries.isEmpty()) { return new RewrittenIndexManifest(null, 0); } @@ -768,12 +773,12 @@ private RewrittenIndexManifest rewriteIndexManifest(Assignment assignment) { checkState( entry.kind() == FileKind.ADD, "Index manifest '%s' contains non-current entry %s.", - assignment.snapshot.indexManifest(), + snapshot.indexManifest(), entry); IndexFileMeta indexFile = entry.indexFile(); GlobalIndexMeta globalIndex = indexFile.globalIndexMeta(); - RowRangeMappingIndex mappingIndex = assignment.rowIdMappings.get(entry.partition()); + RowRangeMappingIndex mappingIndex = assignment.rowIdMappings().get(entry.partition()); if (globalIndex == null || mappingIndex == null) { rewritten.add(entry); continue; @@ -936,7 +941,7 @@ private AssignmentPlan( this.relativeRowIdMappings = relativeRowIdMappings; } - private Assignment createAssignment(Snapshot snapshot) { + private DataEvolutionRowIdAssignment createAssignment(Snapshot snapshot) { Long firstAssignedRowId = snapshot.nextRowId(); checkState( firstAssignedRowId != null, @@ -948,41 +953,13 @@ private Assignment createAssignment(Snapshot snapshot) { absoluteRowIdMappings.put( mapping.getKey(), mapping.getValue().shiftNewStarts(firstAssignedRowId)); } - return new Assignment( - snapshot, - manifestMetasToRewrite, + return new DataEvolutionRowIdAssignment( absoluteRowIdMappings, firstAssignedRowId, Math.addExact(firstAssignedRowId, relativeRowIdMappings.totalOffset)); } } - private static class Assignment { - private final Snapshot snapshot; - private final List manifestMetasToRewrite; - private final Map rowIdMappings; - private final long firstAssignedRowId; - private final long nextRowId; - - private Assignment( - Snapshot snapshot, - List manifestMetasToRewrite, - Map rowIdMappings, - long firstAssignedRowId, - long nextRowId) { - this.snapshot = snapshot; - this.manifestMetasToRewrite = - Collections.unmodifiableList(new ArrayList<>(manifestMetasToRewrite)); - this.rowIdMappings = Collections.unmodifiableMap(new LinkedHashMap<>(rowIdMappings)); - this.firstAssignedRowId = firstAssignedRowId; - this.nextRowId = nextRowId; - } - - private long logicalRowCount() { - return nextRowId - firstAssignedRowId; - } - } - private static class RewrittenDataManifests { private final Map> manifestMetas; private final long fileCount; @@ -1010,10 +987,15 @@ private RewrittenDataManifest( } private static class CommittedAssignment { - private final Assignment assignment; + private final long previousSnapshotId; + private final DataEvolutionRowIdAssignment assignment; private final CommitAssignmentResult commitResult; - private CommittedAssignment(Assignment assignment, CommitAssignmentResult commitResult) { + private CommittedAssignment( + long previousSnapshotId, + DataEvolutionRowIdAssignment assignment, + CommitAssignmentResult commitResult) { + this.previousSnapshotId = previousSnapshotId; this.assignment = assignment; this.commitResult = commitResult; } 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 f73c6bee792c..6b54aeb92b03 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 @@ -19,7 +19,7 @@ package org.apache.paimon.operation; import org.apache.paimon.Snapshot; -import org.apache.paimon.append.dataevolution.DataEvolutionRowIdReassignPlan; +import org.apache.paimon.append.dataevolution.DataEvolutionRowIdAssignment; import org.apache.paimon.data.BinaryRow; import org.apache.paimon.fs.FileIO; import org.apache.paimon.fs.Path; @@ -382,7 +382,7 @@ protected List planManifestsCleaner( collectUnusedIndexManifests(snapshot, skippingSet, indexFiles, indexManifests); collectUnusedStatisticsManifests(snapshot, skippingSet, statistics); - String reassignPlan = DataEvolutionRowIdReassignPlan.planFile(snapshot); + String reassignPlan = DataEvolutionRowIdAssignment.planFile(snapshot); if (reassignPlan != null && skippingSet.add(reassignPlan)) { manifests.add(reassignPlan); } @@ -623,7 +623,7 @@ private Set manifestSkippingSet(Snapshot skippingSnapshot) { .forEach(skippingSet::add); } - String reassignPlan = DataEvolutionRowIdReassignPlan.planFile(skippingSnapshot); + String reassignPlan = DataEvolutionRowIdAssignment.planFile(skippingSnapshot); if (reassignPlan != null) { skippingSet.add(reassignPlan); } 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 ea10fa16438d..07fa455ce357 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 @@ -22,7 +22,7 @@ import org.apache.paimon.Snapshot; import org.apache.paimon.Snapshot.CommitKind; import org.apache.paimon.annotation.VisibleForTesting; -import org.apache.paimon.append.dataevolution.DataEvolutionRowIdReassignPlan; +import org.apache.paimon.append.dataevolution.DataEvolutionRowIdAssignment; import org.apache.paimon.catalog.SnapshotCommit; import org.apache.paimon.data.BinaryRow; import org.apache.paimon.data.InternalRow; @@ -1402,7 +1402,7 @@ public boolean replaceManifestList( deltaManifestList, indexManifest, nextRowId, - DataEvolutionRowIdReassignPlan.withoutPlan(latest.properties())); + DataEvolutionRowIdAssignment.withoutPlan(latest.properties())); } public boolean replaceManifestList( @@ -1521,7 +1521,7 @@ public boolean rollbackToAsLatest(Snapshot targetSnapshot) { null, targetSnapshot.watermark(), targetSnapshot.statistics(), - DataEvolutionRowIdReassignPlan.withoutPlan(targetSnapshot.properties()), + DataEvolutionRowIdAssignment.withoutPlan(targetSnapshot.properties()), nextRowId, null); @@ -1671,7 +1671,7 @@ private boolean compactManifestOnce() { null, latestSnapshot.watermark(), latestSnapshot.statistics(), - DataEvolutionRowIdReassignPlan.withoutPlan(latestSnapshot.properties()), + DataEvolutionRowIdAssignment.withoutPlan(latestSnapshot.properties()), latestSnapshot.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 a6e22971eb04..d0af01ad7698 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 @@ -19,7 +19,7 @@ package org.apache.paimon.operation; import org.apache.paimon.Snapshot; -import org.apache.paimon.append.dataevolution.DataEvolutionRowIdReassignPlan; +import org.apache.paimon.append.dataevolution.DataEvolutionRowIdAssignment; import org.apache.paimon.blob.ManagedBlobReferenceFile; import org.apache.paimon.data.Timestamp; import org.apache.paimon.fs.FileIO; @@ -330,7 +330,7 @@ protected void collectWithoutDataFileWithManifestFlag( .forEach(name -> usedFileWithFlagConsumer.accept(Pair.of(name, false))); } - String reassignPlan = DataEvolutionRowIdReassignPlan.planFile(snapshot); + String reassignPlan = DataEvolutionRowIdAssignment.planFile(snapshot); if (reassignPlan != null) { usedFileWithFlagConsumer.accept(Pair.of(reassignPlan, 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 d9c656134167..2253eb24f2d9 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 @@ -656,24 +656,24 @@ public void testReassignPlanIsNotInherited() throws Exception { FileStoreTable table = createTableWithInterleavedPartitions(); new DataEvolutionRowIdReassigner(table).reassign(); Snapshot reassigned = table.snapshotManager().latestSnapshot(); - assertThat(DataEvolutionRowIdReassignPlan.planFile(reassigned)).isNotNull(); + assertThat(DataEvolutionRowIdAssignment.planFile(reassigned)).isNotNull(); compactManifests(table); Snapshot compacted = table.snapshotManager().latestSnapshot(); assertThat(compacted.id()).isGreaterThan(reassigned.id()); - assertThat(DataEvolutionRowIdReassignPlan.planFile(compacted)).isNull(); + assertThat(DataEvolutionRowIdAssignment.planFile(compacted)).isNull(); try (FileStoreCommitImpl commit = (FileStoreCommitImpl) table.store().newCommit("test-rollback-plan", table)) { assertThat(commit.rollbackToAsLatest(reassigned)).isTrue(); } - assertThat( - DataEvolutionRowIdReassignPlan.planFile( - table.snapshotManager().latestSnapshot())) + assertThat(DataEvolutionRowIdAssignment.planFile(table.snapshotManager().latestSnapshot())) .isNull(); - assertThat(DataEvolutionRowIdReassignPlan.planFile(reassigned)).isNotNull(); - DataEvolutionRowIdReassignPlan.read( - table.fileIO(), table.store().pathFactory(), reassigned); + assertThat(DataEvolutionRowIdAssignment.planFile(reassigned)).isNotNull(); + DataEvolutionRowIdAssignment.read( + table.fileIO(), + table.store().pathFactory(), + DataEvolutionRowIdAssignment.planFile(reassigned)); } @Test @@ -684,7 +684,7 @@ public void testReassignPlanCleanup() throws Exception { Path plan = table.store() .pathFactory() - .toManifestFilePath(DataEvolutionRowIdReassignPlan.planFile(reassigned)); + .toManifestFilePath(DataEvolutionRowIdAssignment.planFile(reassigned)); Path orphan = table.store().pathFactory().toManifestFilePath("row-id-reassign-plan-orphan"); table.fileIO().newOutputStream(orphan, false).close(); @@ -694,9 +694,7 @@ public void testReassignPlanCleanup() throws Exception { table.createTag("reassign", reassigned.id()); writeOneRow(table, "c", 100); - assertThat( - DataEvolutionRowIdReassignPlan.planFile( - table.snapshotManager().latestSnapshot())) + assertThat(DataEvolutionRowIdAssignment.planFile(table.snapshotManager().latestSnapshot())) .isNull(); table.newExpireSnapshots() .config(ExpireConfig.builder().snapshotRetainMin(1).snapshotRetainMax(1).build()) @@ -718,7 +716,7 @@ public void testReassignPlanExpiresWithSnapshot() throws Exception { table.store() .pathFactory() .toManifestFilePath( - DataEvolutionRowIdReassignPlan.planFile( + DataEvolutionRowIdAssignment.planFile( table.snapshotManager().latestSnapshot())); writeOneRow(table, "c", 100); table.newExpireSnapshots() @@ -735,15 +733,17 @@ public void testRejectInvalidReassignPlan() throws Exception { Path path = table.store() .pathFactory() - .toManifestFilePath(DataEvolutionRowIdReassignPlan.planFile(snapshot)); + .toManifestFilePath(DataEvolutionRowIdAssignment.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( () -> - DataEvolutionRowIdReassignPlan.read( - table.fileIO(), table.store().pathFactory(), snapshot)) + DataEvolutionRowIdAssignment.read( + table.fileIO(), + table.store().pathFactory(), + DataEvolutionRowIdAssignment.planFile(snapshot))) .isInstanceOf(IOException.class) .hasMessageContaining("checksum"); @@ -752,43 +752,36 @@ public void testRejectInvalidReassignPlan() throws Exception { overwritePlan(table, path, corrupted); assertThatThrownBy( () -> - DataEvolutionRowIdReassignPlan.read( - table.fileIO(), table.store().pathFactory(), snapshot)) + DataEvolutionRowIdAssignment.read( + table.fileIO(), + table.store().pathFactory(), + DataEvolutionRowIdAssignment.planFile(snapshot))) .isInstanceOf(IOException.class) .hasMessageContaining("version: 99"); - // The target snapshot ID occupies bytes 12 through 19. - corrupted = bytes.clone(); - corrupted[19] ^= 1; - overwritePlan(table, path, corrupted); - assertThatThrownBy( - () -> - DataEvolutionRowIdReassignPlan.read( - table.fileIO(), table.store().pathFactory(), snapshot)) - .isInstanceOf(IOException.class) - .hasMessageContaining("belongs to snapshot"); - overwritePlan(table, path, Arrays.copyOf(bytes, bytes.length - 1)); assertThatThrownBy( () -> - DataEvolutionRowIdReassignPlan.read( - table.fileIO(), table.store().pathFactory(), snapshot)) + DataEvolutionRowIdAssignment.read( + table.fileIO(), + table.store().pathFactory(), + DataEvolutionRowIdAssignment.planFile(snapshot))) .isInstanceOf(IOException.class); } @Test public void testRemovingReassignMarkerPreservesOtherProperties() { Map properties = new HashMap<>(); - properties.put(DataEvolutionRowIdReassignPlan.PLAN_FILE_PROPERTY, "plan"); + properties.put(DataEvolutionRowIdAssignment.PLAN_FILE_PROPERTY, "plan"); properties.put("sequence.generation.max-sequence-number", "100"); - assertThat(DataEvolutionRowIdReassignPlan.withoutPlan(properties)) + assertThat(DataEvolutionRowIdAssignment.withoutPlan(properties)) .containsExactlyEntriesOf( Collections.singletonMap("sequence.generation.max-sequence-number", "100")); assertThat(properties).hasSize(2); assertThat( - DataEvolutionRowIdReassignPlan.withoutPlan( + DataEvolutionRowIdAssignment.withoutPlan( Collections.singletonMap( - DataEvolutionRowIdReassignPlan.PLAN_FILE_PROPERTY, "plan"))) + DataEvolutionRowIdAssignment.PLAN_FILE_PROPERTY, "plan"))) .isNull(); } @@ -800,24 +793,27 @@ private void overwritePlan(FileStoreTable table, Path path, byte[] bytes) throws private void assertPersistedPlan(FileStoreTable table) throws Exception { Snapshot snapshot = Snapshot.fromJson(table.snapshotManager().latestSnapshot().toJson()); - DataEvolutionRowIdReassignPlan plan = - DataEvolutionRowIdReassignPlan.read( - table.fileIO(), table.store().pathFactory(), snapshot); - assertThat(plan.snapshotId()).isEqualTo(snapshot.id()); - assertThat(plan.sourceSnapshotId()).isEqualTo(snapshot.id() - 1); + DataEvolutionRowIdAssignment assignment = + DataEvolutionRowIdAssignment.read( + table.fileIO(), + table.store().pathFactory(), + DataEvolutionRowIdAssignment.planFile(snapshot)); + assertThat(assignment.firstAssignedRowId()) + .isEqualTo(table.snapshotManager().snapshot(snapshot.id() - 1).nextRowId()); + assertThat(assignment.nextRowId()).isEqualTo(snapshot.nextRowId()); Map previous = new HashMap<>(); for (ManifestEntry entry : - table.store().newScan().withSnapshot(plan.sourceSnapshotId()).plan().files()) { + 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(); if (oldRange.equals(newRange)) { - assertThat(plan.map(entry.partition(), oldRange)).isEmpty(); - assertThat(plan.overlaps(entry.partition(), oldRange)).isFalse(); + assertThat(assignment.map(entry.partition(), oldRange)).isEmpty(); + assertThat(assignment.overlaps(entry.partition(), oldRange)).isFalse(); } else { - assertThat(plan.map(entry.partition(), oldRange)).hasValue(newRange); + assertThat(assignment.map(entry.partition(), oldRange)).hasValue(newRange); } } } From 1bf64c595d85c6ba18cbbbd0cbb0f3896272fe15 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E4=BB=9F=E5=BC=8B?= Date: Mon, 21 Sep 2026 11:32:50 +0800 Subject: [PATCH 3/3] [core] Isolate assignment persistence in SerializationAssignment --- .../DataEvolutionRowIdReassigner.java | 138 +++++++++--------- ...ment.java => SerializationAssignment.java} | 54 +++++-- .../paimon/operation/FileDeletionBase.java | 6 +- .../paimon/operation/FileStoreCommitImpl.java | 8 +- .../paimon/operation/OrphanFilesClean.java | 4 +- .../DataEvolutionRowIdReassignerTest.java | 46 +++--- 6 files changed, 144 insertions(+), 112 deletions(-) rename paimon-core/src/main/java/org/apache/paimon/append/dataevolution/{DataEvolutionRowIdAssignment.java => SerializationAssignment.java} (79%) 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 f7c8870e5b40..6c293c020e52 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 @@ -53,8 +53,6 @@ import javax.annotation.Nullable; -import java.io.IOException; -import java.io.UncheckedIOException; import java.util.ArrayList; import java.util.Collections; import java.util.Comparator; @@ -159,25 +157,25 @@ public Result reassign(String commitUser) { CommittedAssignment committed = commitAssignmentWithRetry( optionalPlan.get(), latest, manifestFile, manifestList, commitUser); - DataEvolutionRowIdAssignment assignment = committed.assignment; + Assignment assignment = committed.assignment; CommitAssignmentResult commitResult = committed.commitResult; LOG.info( "Reassigned row IDs for table {} from {} to {}, partitions={}, files={}, rows={}.", table.name(), - assignment.firstAssignedRowId(), - assignment.nextRowId(), - assignment.rowIdMappings().size(), + assignment.firstAssignedRowId, + assignment.nextRowId, + assignment.rowIdMappings.size(), commitResult.fileCount, assignment.logicalRowCount()); return new Result( - committed.previousSnapshotId, - committed.previousSnapshotId + 1, + assignment.snapshot.id(), + assignment.snapshot.id() + 1, commitResult.fileCount, assignment.logicalRowCount(), commitResult.indexFileCount, - assignment.firstAssignedRowId(), - assignment.nextRowId()); + assignment.firstAssignedRowId, + assignment.nextRowId); } private Optional planAssignment(List manifestMetas) { @@ -381,17 +379,11 @@ private CommittedAssignment commitAssignmentWithRetry( latest = observedLatest; } - DataEvolutionRowIdAssignment assignment = assignmentPlan.createAssignment(latest); + Assignment assignment = assignmentPlan.createAssignment(latest); CommitAssignmentResult commitResult = - commitAssignment( - latest, - assignmentPlan.manifestMetasToRewrite, - assignment, - manifestFile, - manifestList, - commitUser); + commitAssignment(assignment, manifestFile, manifestList, commitUser); if (commitResult.success) { - return new CommittedAssignment(latest.id(), assignment, commitResult); + return new CommittedAssignment(assignment, commitResult); } if (System.currentTimeMillis() - startMillis > options.commitTimeout() @@ -421,33 +413,27 @@ private CommittedAssignment commitAssignmentWithRetry( } private CommitAssignmentResult commitAssignment( - Snapshot snapshot, - List manifestMetasToRewrite, - DataEvolutionRowIdAssignment assignment, + Assignment assignment, ManifestFile manifestFile, ManifestList manifestList, String commitUser) { RewrittenDataManifests rewrittenDataManifests = - writeManifestReplacements(manifestMetasToRewrite, assignment, manifestFile); + writeManifestReplacements(assignment, manifestFile); Pair baseManifestList = writeBaseManifestList( - manifestList.readDataManifests(snapshot), + manifestList.readDataManifests(assignment.snapshot), rewrittenDataManifests.manifestMetas, manifestList); Pair deltaManifestList = manifestList.write(Collections.emptyList()); - RewrittenIndexManifest rewrittenIndexManifest = rewriteIndexManifest(snapshot, assignment); + RewrittenIndexManifest rewrittenIndexManifest = rewriteIndexManifest(assignment); - String planFile; - try { - planFile = assignment.write(table.fileIO(), table.store().pathFactory()); - } catch (IOException e) { - throw new UncheckedIOException("Failed to persist row-id reassignment plan.", e); - } Map properties = - snapshot.properties() == null - ? new HashMap<>() - : new HashMap<>(snapshot.properties()); - properties.put(DataEvolutionRowIdAssignment.PLAN_FILE_PROPERTY, planFile); + SerializationAssignment.writeProperties( + table, + assignment.snapshot, + assignment.rowIdMappings, + assignment.firstAssignedRowId, + assignment.nextRowId); boolean success; try (FileStoreCommitImpl commit = @@ -455,17 +441,16 @@ private CommitAssignmentResult commitAssignment( beforeCommit.run(); success = commit.replaceManifestList( - snapshot, - snapshot.totalRecordCount(), + assignment.snapshot, + assignment.snapshot.totalRecordCount(), baseManifestList, deltaManifestList, rewrittenIndexManifest.indexManifest, - assignment.nextRowId(), + assignment.nextRowId, properties); } if (!success) { - // Only clean a definitively rejected attempt. An exception may mean it committed. - table.fileIO().deleteQuietly(table.store().pathFactory().toManifestFilePath(planFile)); + SerializationAssignment.deletePlan(table, properties); } return new CommitAssignmentResult( success, rewrittenDataManifests.fileCount, rewrittenIndexManifest.indexFileCount); @@ -656,13 +641,13 @@ private Pair writeBaseManifestList( } private RewrittenDataManifests writeManifestReplacements( - List manifestMetasToRewrite, - DataEvolutionRowIdAssignment assignment, - ManifestFile manifestFile) { + Assignment assignment, ManifestFile manifestFile) { Integer parallelism = table.coreOptions().scanManifestParallelism(); - List rewritten = new ArrayList<>(manifestMetasToRewrite.size()); - if (manifestMetasToRewrite.size() == 1 || (parallelism != null && parallelism == 1)) { - for (ManifestFileMeta manifestMeta : manifestMetasToRewrite) { + List rewritten = + new ArrayList<>(assignment.manifestMetasToRewrite.size()); + if (assignment.manifestMetasToRewrite.size() == 1 + || (parallelism != null && parallelism == 1)) { + for (ManifestFileMeta manifestMeta : assignment.manifestMetasToRewrite) { rewritten.add(rewriteDataManifest(assignment, manifestFile, manifestMeta)); } } else { @@ -675,7 +660,7 @@ private RewrittenDataManifests writeManifestReplacements( manifestMeta)); try (CloseableBatchIterator results = sequentialBatchedExecuteCloseable( - rewriter, manifestMetasToRewrite, parallelism)) { + rewriter, assignment.manifestMetasToRewrite, parallelism)) { while (results.hasNext()) { rewritten.add(results.next()); } @@ -692,9 +677,7 @@ private RewrittenDataManifests writeManifestReplacements( } private RewrittenDataManifest rewriteDataManifest( - DataEvolutionRowIdAssignment assignment, - ManifestFile manifestFile, - ManifestFileMeta manifestMeta) { + Assignment assignment, ManifestFile manifestFile, ManifestFileMeta manifestMeta) { beforeManifestRewrite.accept(manifestMeta); ManifestEntrySerializer serializer = new ManifestEntrySerializer(); ManifestAvroWriter writer = manifestFile.createAvroWriter(); @@ -707,7 +690,7 @@ private RewrittenDataManifest rewriteDataManifest( while (entries.hasNext()) { ProjectedManifestEntry entry = entries.next(); ManifestEntry output = entry; - RowRangeMappingIndex mapping = assignment.rowIdMappings().get(entry.partition()); + RowRangeMappingIndex mapping = assignment.rowIdMappings.get(entry.partition()); if (mapping != null) { Optional reassignedRange = mapping.map(entry.file().nonNullRowIdRange()); if (reassignedRange.isPresent()) { @@ -755,14 +738,14 @@ private void validatePlanningEntry(ManifestEntry entry) { table.name()); } - private RewrittenIndexManifest rewriteIndexManifest( - Snapshot snapshot, DataEvolutionRowIdAssignment assignment) { - if (snapshot.indexManifest() == null) { + private RewrittenIndexManifest rewriteIndexManifest(Assignment assignment) { + if (assignment.snapshot.indexManifest() == null) { return new RewrittenIndexManifest(null, 0); } IndexManifestFile indexManifestFile = table.store().indexManifestFileFactory().create(); - List indexEntries = indexManifestFile.read(snapshot.indexManifest()); + List indexEntries = + indexManifestFile.read(assignment.snapshot.indexManifest()); if (indexEntries.isEmpty()) { return new RewrittenIndexManifest(null, 0); } @@ -773,12 +756,12 @@ private RewrittenIndexManifest rewriteIndexManifest( checkState( entry.kind() == FileKind.ADD, "Index manifest '%s' contains non-current entry %s.", - snapshot.indexManifest(), + assignment.snapshot.indexManifest(), entry); IndexFileMeta indexFile = entry.indexFile(); GlobalIndexMeta globalIndex = indexFile.globalIndexMeta(); - RowRangeMappingIndex mappingIndex = assignment.rowIdMappings().get(entry.partition()); + RowRangeMappingIndex mappingIndex = assignment.rowIdMappings.get(entry.partition()); if (globalIndex == null || mappingIndex == null) { rewritten.add(entry); continue; @@ -941,7 +924,7 @@ private AssignmentPlan( this.relativeRowIdMappings = relativeRowIdMappings; } - private DataEvolutionRowIdAssignment createAssignment(Snapshot snapshot) { + private Assignment createAssignment(Snapshot snapshot) { Long firstAssignedRowId = snapshot.nextRowId(); checkState( firstAssignedRowId != null, @@ -953,13 +936,41 @@ private DataEvolutionRowIdAssignment createAssignment(Snapshot snapshot) { absoluteRowIdMappings.put( mapping.getKey(), mapping.getValue().shiftNewStarts(firstAssignedRowId)); } - return new DataEvolutionRowIdAssignment( + return new Assignment( + snapshot, + manifestMetasToRewrite, absoluteRowIdMappings, firstAssignedRowId, Math.addExact(firstAssignedRowId, relativeRowIdMappings.totalOffset)); } } + private static class Assignment { + private final Snapshot snapshot; + private final List manifestMetasToRewrite; + private final Map rowIdMappings; + private final long firstAssignedRowId; + private final long nextRowId; + + private Assignment( + Snapshot snapshot, + List manifestMetasToRewrite, + Map rowIdMappings, + long firstAssignedRowId, + long nextRowId) { + this.snapshot = snapshot; + this.manifestMetasToRewrite = + Collections.unmodifiableList(new ArrayList<>(manifestMetasToRewrite)); + this.rowIdMappings = Collections.unmodifiableMap(new LinkedHashMap<>(rowIdMappings)); + this.firstAssignedRowId = firstAssignedRowId; + this.nextRowId = nextRowId; + } + + private long logicalRowCount() { + return nextRowId - firstAssignedRowId; + } + } + private static class RewrittenDataManifests { private final Map> manifestMetas; private final long fileCount; @@ -987,15 +998,10 @@ private RewrittenDataManifest( } private static class CommittedAssignment { - private final long previousSnapshotId; - private final DataEvolutionRowIdAssignment assignment; + private final Assignment assignment; private final CommitAssignmentResult commitResult; - private CommittedAssignment( - long previousSnapshotId, - DataEvolutionRowIdAssignment assignment, - CommitAssignmentResult commitResult) { - this.previousSnapshotId = previousSnapshotId; + private CommittedAssignment(Assignment assignment, CommitAssignmentResult commitResult) { this.assignment = assignment; this.commitResult = commitResult; } diff --git a/paimon-core/src/main/java/org/apache/paimon/append/dataevolution/DataEvolutionRowIdAssignment.java b/paimon-core/src/main/java/org/apache/paimon/append/dataevolution/SerializationAssignment.java similarity index 79% rename from paimon-core/src/main/java/org/apache/paimon/append/dataevolution/DataEvolutionRowIdAssignment.java rename to paimon-core/src/main/java/org/apache/paimon/append/dataevolution/SerializationAssignment.java index 9b9f6fe5d1c1..bdcc207b9112 100644 --- a/paimon-core/src/main/java/org/apache/paimon/append/dataevolution/DataEvolutionRowIdAssignment.java +++ b/paimon-core/src/main/java/org/apache/paimon/append/dataevolution/SerializationAssignment.java @@ -24,6 +24,7 @@ import org.apache.paimon.fs.Path; import org.apache.paimon.io.DataInputViewStreamWrapper; import org.apache.paimon.io.DataOutputViewStreamWrapper; +import org.apache.paimon.table.FileStoreTable; import org.apache.paimon.utils.FileStorePathFactory; import org.apache.paimon.utils.Range; @@ -32,6 +33,7 @@ import java.io.BufferedInputStream; import java.io.BufferedOutputStream; import java.io.IOException; +import java.io.UncheckedIOException; import java.util.Collections; import java.util.HashMap; import java.util.LinkedHashMap; @@ -47,8 +49,8 @@ import static org.apache.paimon.utils.SerializationUtils.readCount; import static org.apache.paimon.utils.SerializationUtils.serializeBinaryRow; -/** The row-id assignment used to rewrite metadata and persisted for subsequent commits. */ -public final class DataEvolutionRowIdAssignment { +/** 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"; @@ -60,7 +62,7 @@ public final class DataEvolutionRowIdAssignment { private final long firstAssignedRowId; private final long nextRowId; - DataEvolutionRowIdAssignment( + private SerializationAssignment( Map rowIdMappings, long firstAssignedRowId, long nextRowId) { @@ -75,10 +77,6 @@ public final class DataEvolutionRowIdAssignment { this.nextRowId = nextRowId; } - Map rowIdMappings() { - return rowIdMappings; - } - public long firstAssignedRowId() { return firstAssignedRowId; } @@ -87,10 +85,6 @@ public long nextRowId() { return nextRowId; } - public long logicalRowCount() { - return nextRowId - firstAssignedRowId; - } - /** Returns a mapping only when the entire range maps to a contiguous range. */ public Optional map(BinaryRow partition, Range range) { RowRangeMappingIndex mapping = rowIdMappings.get(partition); @@ -118,8 +112,40 @@ public static Map withoutPlan(@Nullable Map prop return result.isEmpty() ? null : result; } + /** Persists the assignment and adds its reference to this commit's snapshot properties. */ + static Map writeProperties( + FileStoreTable table, + Snapshot snapshot, + Map rowIdMappings, + long firstAssignedRowId, + long nextRowId) { + String planFile; + try { + planFile = + new SerializationAssignment(rowIdMappings, firstAssignedRowId, nextRowId) + .write(table.fileIO(), table.store().pathFactory()); + } catch (IOException e) { + throw new UncheckedIOException("Failed to persist row-id reassignment plan.", e); + } + Map properties = + snapshot.properties() == null + ? new HashMap<>() + : new HashMap<>(snapshot.properties()); + properties.put(PLAN_FILE_PROPERTY, planFile); + return properties; + } + + /** Only call after a definitively rejected commit, never when its outcome is uncertain. */ + static void deletePlan(FileStoreTable table, Map properties) { + table.fileIO() + .deleteQuietly( + table.store() + .pathFactory() + .toManifestFilePath(properties.get(PLAN_FILE_PROPERTY))); + } + /** Streams the effective mappings without materializing another copy of the plan. */ - String write(FileIO fileIO, FileStorePathFactory pathFactory) throws IOException { + private String write(FileIO fileIO, FileStorePathFactory pathFactory) throws IOException { String fileName = FILE_PREFIX + UUID.randomUUID(); Path path = pathFactory.toManifestFilePath(fileName); try (DataOutputViewStreamWrapper out = @@ -145,7 +171,7 @@ String write(FileIO fileIO, FileStorePathFactory pathFactory) throws IOException return fileName; } - public static DataEvolutionRowIdAssignment read( + public static SerializationAssignment read( FileIO fileIO, FileStorePathFactory pathFactory, String fileName) throws IOException { try (DataInputViewStreamWrapper in = new DataInputViewStreamWrapper( @@ -175,7 +201,7 @@ public static DataEvolutionRowIdAssignment read( if (in.read() != -1) { throw new IOException("Unexpected trailing bytes in row-id reassignment plan."); } - return new DataEvolutionRowIdAssignment(mappings, firstAssignedRowId, nextRowId); + return new SerializationAssignment(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 6b54aeb92b03..93d6501f4663 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 @@ -19,7 +19,7 @@ package org.apache.paimon.operation; import org.apache.paimon.Snapshot; -import org.apache.paimon.append.dataevolution.DataEvolutionRowIdAssignment; +import org.apache.paimon.append.dataevolution.SerializationAssignment; import org.apache.paimon.data.BinaryRow; import org.apache.paimon.fs.FileIO; import org.apache.paimon.fs.Path; @@ -382,7 +382,7 @@ protected List planManifestsCleaner( collectUnusedIndexManifests(snapshot, skippingSet, indexFiles, indexManifests); collectUnusedStatisticsManifests(snapshot, skippingSet, statistics); - String reassignPlan = DataEvolutionRowIdAssignment.planFile(snapshot); + String reassignPlan = SerializationAssignment.planFile(snapshot); if (reassignPlan != null && skippingSet.add(reassignPlan)) { manifests.add(reassignPlan); } @@ -623,7 +623,7 @@ private Set manifestSkippingSet(Snapshot skippingSnapshot) { .forEach(skippingSet::add); } - String reassignPlan = DataEvolutionRowIdAssignment.planFile(skippingSnapshot); + String reassignPlan = SerializationAssignment.planFile(skippingSnapshot); if (reassignPlan != null) { skippingSet.add(reassignPlan); } 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 07fa455ce357..b74c23d840be 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 @@ -22,7 +22,7 @@ import org.apache.paimon.Snapshot; import org.apache.paimon.Snapshot.CommitKind; import org.apache.paimon.annotation.VisibleForTesting; -import org.apache.paimon.append.dataevolution.DataEvolutionRowIdAssignment; +import org.apache.paimon.append.dataevolution.SerializationAssignment; import org.apache.paimon.catalog.SnapshotCommit; import org.apache.paimon.data.BinaryRow; import org.apache.paimon.data.InternalRow; @@ -1402,7 +1402,7 @@ public boolean replaceManifestList( deltaManifestList, indexManifest, nextRowId, - DataEvolutionRowIdAssignment.withoutPlan(latest.properties())); + SerializationAssignment.withoutPlan(latest.properties())); } public boolean replaceManifestList( @@ -1521,7 +1521,7 @@ public boolean rollbackToAsLatest(Snapshot targetSnapshot) { null, targetSnapshot.watermark(), targetSnapshot.statistics(), - DataEvolutionRowIdAssignment.withoutPlan(targetSnapshot.properties()), + SerializationAssignment.withoutPlan(targetSnapshot.properties()), nextRowId, null); @@ -1671,7 +1671,7 @@ private boolean compactManifestOnce() { null, latestSnapshot.watermark(), latestSnapshot.statistics(), - DataEvolutionRowIdAssignment.withoutPlan(latestSnapshot.properties()), + SerializationAssignment.withoutPlan(latestSnapshot.properties()), latestSnapshot.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 d0af01ad7698..6f6980e5b615 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 @@ -19,7 +19,7 @@ package org.apache.paimon.operation; import org.apache.paimon.Snapshot; -import org.apache.paimon.append.dataevolution.DataEvolutionRowIdAssignment; +import org.apache.paimon.append.dataevolution.SerializationAssignment; import org.apache.paimon.blob.ManagedBlobReferenceFile; import org.apache.paimon.data.Timestamp; import org.apache.paimon.fs.FileIO; @@ -330,7 +330,7 @@ protected void collectWithoutDataFileWithManifestFlag( .forEach(name -> usedFileWithFlagConsumer.accept(Pair.of(name, false))); } - String reassignPlan = DataEvolutionRowIdAssignment.planFile(snapshot); + String reassignPlan = SerializationAssignment.planFile(snapshot); if (reassignPlan != null) { usedFileWithFlagConsumer.accept(Pair.of(reassignPlan, 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 2253eb24f2d9..8c1ea945b9d8 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 @@ -656,24 +656,24 @@ public void testReassignPlanIsNotInherited() throws Exception { FileStoreTable table = createTableWithInterleavedPartitions(); new DataEvolutionRowIdReassigner(table).reassign(); Snapshot reassigned = table.snapshotManager().latestSnapshot(); - assertThat(DataEvolutionRowIdAssignment.planFile(reassigned)).isNotNull(); + assertThat(SerializationAssignment.planFile(reassigned)).isNotNull(); compactManifests(table); Snapshot compacted = table.snapshotManager().latestSnapshot(); assertThat(compacted.id()).isGreaterThan(reassigned.id()); - assertThat(DataEvolutionRowIdAssignment.planFile(compacted)).isNull(); + assertThat(SerializationAssignment.planFile(compacted)).isNull(); try (FileStoreCommitImpl commit = (FileStoreCommitImpl) table.store().newCommit("test-rollback-plan", table)) { assertThat(commit.rollbackToAsLatest(reassigned)).isTrue(); } - assertThat(DataEvolutionRowIdAssignment.planFile(table.snapshotManager().latestSnapshot())) + assertThat(SerializationAssignment.planFile(table.snapshotManager().latestSnapshot())) .isNull(); - assertThat(DataEvolutionRowIdAssignment.planFile(reassigned)).isNotNull(); - DataEvolutionRowIdAssignment.read( + assertThat(SerializationAssignment.planFile(reassigned)).isNotNull(); + SerializationAssignment.read( table.fileIO(), table.store().pathFactory(), - DataEvolutionRowIdAssignment.planFile(reassigned)); + SerializationAssignment.planFile(reassigned)); } @Test @@ -684,7 +684,7 @@ public void testReassignPlanCleanup() throws Exception { Path plan = table.store() .pathFactory() - .toManifestFilePath(DataEvolutionRowIdAssignment.planFile(reassigned)); + .toManifestFilePath(SerializationAssignment.planFile(reassigned)); Path orphan = table.store().pathFactory().toManifestFilePath("row-id-reassign-plan-orphan"); table.fileIO().newOutputStream(orphan, false).close(); @@ -694,7 +694,7 @@ public void testReassignPlanCleanup() throws Exception { table.createTag("reassign", reassigned.id()); writeOneRow(table, "c", 100); - assertThat(DataEvolutionRowIdAssignment.planFile(table.snapshotManager().latestSnapshot())) + assertThat(SerializationAssignment.planFile(table.snapshotManager().latestSnapshot())) .isNull(); table.newExpireSnapshots() .config(ExpireConfig.builder().snapshotRetainMin(1).snapshotRetainMax(1).build()) @@ -716,7 +716,7 @@ public void testReassignPlanExpiresWithSnapshot() throws Exception { table.store() .pathFactory() .toManifestFilePath( - DataEvolutionRowIdAssignment.planFile( + SerializationAssignment.planFile( table.snapshotManager().latestSnapshot())); writeOneRow(table, "c", 100); table.newExpireSnapshots() @@ -733,17 +733,17 @@ public void testRejectInvalidReassignPlan() throws Exception { Path path = table.store() .pathFactory() - .toManifestFilePath(DataEvolutionRowIdAssignment.planFile(snapshot)); + .toManifestFilePath(SerializationAssignment.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( () -> - DataEvolutionRowIdAssignment.read( + SerializationAssignment.read( table.fileIO(), table.store().pathFactory(), - DataEvolutionRowIdAssignment.planFile(snapshot))) + SerializationAssignment.planFile(snapshot))) .isInstanceOf(IOException.class) .hasMessageContaining("checksum"); @@ -752,36 +752,36 @@ public void testRejectInvalidReassignPlan() throws Exception { overwritePlan(table, path, corrupted); assertThatThrownBy( () -> - DataEvolutionRowIdAssignment.read( + SerializationAssignment.read( table.fileIO(), table.store().pathFactory(), - DataEvolutionRowIdAssignment.planFile(snapshot))) + SerializationAssignment.planFile(snapshot))) .isInstanceOf(IOException.class) .hasMessageContaining("version: 99"); overwritePlan(table, path, Arrays.copyOf(bytes, bytes.length - 1)); assertThatThrownBy( () -> - DataEvolutionRowIdAssignment.read( + SerializationAssignment.read( table.fileIO(), table.store().pathFactory(), - DataEvolutionRowIdAssignment.planFile(snapshot))) + SerializationAssignment.planFile(snapshot))) .isInstanceOf(IOException.class); } @Test public void testRemovingReassignMarkerPreservesOtherProperties() { Map properties = new HashMap<>(); - properties.put(DataEvolutionRowIdAssignment.PLAN_FILE_PROPERTY, "plan"); + properties.put(SerializationAssignment.PLAN_FILE_PROPERTY, "plan"); properties.put("sequence.generation.max-sequence-number", "100"); - assertThat(DataEvolutionRowIdAssignment.withoutPlan(properties)) + assertThat(SerializationAssignment.withoutPlan(properties)) .containsExactlyEntriesOf( Collections.singletonMap("sequence.generation.max-sequence-number", "100")); assertThat(properties).hasSize(2); assertThat( - DataEvolutionRowIdAssignment.withoutPlan( + SerializationAssignment.withoutPlan( Collections.singletonMap( - DataEvolutionRowIdAssignment.PLAN_FILE_PROPERTY, "plan"))) + SerializationAssignment.PLAN_FILE_PROPERTY, "plan"))) .isNull(); } @@ -793,11 +793,11 @@ private void overwritePlan(FileStoreTable table, Path path, byte[] bytes) throws private void assertPersistedPlan(FileStoreTable table) throws Exception { Snapshot snapshot = Snapshot.fromJson(table.snapshotManager().latestSnapshot().toJson()); - DataEvolutionRowIdAssignment assignment = - DataEvolutionRowIdAssignment.read( + SerializationAssignment assignment = + SerializationAssignment.read( table.fileIO(), table.store().pathFactory(), - DataEvolutionRowIdAssignment.planFile(snapshot)); + SerializationAssignment.planFile(snapshot)); assertThat(assignment.firstAssignedRowId()) .isEqualTo(table.snapshotManager().snapshot(snapshot.id() - 1).nextRowId()); assertThat(assignment.nextRowId()).isEqualTo(snapshot.nextRowId());