diff --git a/docs/docs/multimodal-table/data-evolution-maintenance.mdx b/docs/docs/multimodal-table/data-evolution-maintenance.mdx index 66870446aabf..2ce37c933843 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 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 `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 ec12b3fd2131..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 @@ -427,6 +427,14 @@ private CommitAssignmentResult commitAssignment( Pair deltaManifestList = manifestList.write(Collections.emptyList()); RewrittenIndexManifest rewrittenIndexManifest = rewriteIndexManifest(assignment); + Map properties = + SerializationAssignment.writeProperties( + table, + assignment.snapshot, + assignment.rowIdMappings, + assignment.firstAssignedRowId, + assignment.nextRowId); + boolean success; try (FileStoreCommitImpl commit = (FileStoreCommitImpl) table.store().newCommit(commitUser, table)) { @@ -438,7 +446,11 @@ private CommitAssignmentResult commitAssignment( baseManifestList, deltaManifestList, rewrittenIndexManifest.indexManifest, - assignment.nextRowId); + assignment.nextRowId, + properties); + } + if (!success) { + SerializationAssignment.deletePlan(table, 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 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/append/dataevolution/SerializationAssignment.java b/paimon-core/src/main/java/org/apache/paimon/append/dataevolution/SerializationAssignment.java new file mode 100644 index 000000000000..bdcc207b9112 --- /dev/null +++ b/paimon-core/src/main/java/org/apache/paimon/append/dataevolution/SerializationAssignment.java @@ -0,0 +1,209 @@ +/* + * 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.table.FileStoreTable; +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.io.UncheckedIOException; +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; + +/** 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"; + + private static final int VERSION = 1; + private static final String FILE_PREFIX = "row-id-reassign-plan-"; + + private final Map rowIdMappings; + private final long firstAssignedRowId; + private final long nextRowId; + + private SerializationAssignment( + Map rowIdMappings, + long firstAssignedRowId, + long nextRowId) { + checkArgument(!rowIdMappings.isEmpty(), "Reassignment mappings must not be empty."); + checkArgument( + 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 firstAssignedRowId() { + return firstAssignedRowId; + } + + public long nextRowId() { + return nextRowId; + } + + /** 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); + return mapping == null ? Optional.empty() : mapping.map(range); + } + + public boolean overlaps(BinaryRow partition, Range range) { + RowRangeMappingIndex mapping = rowIdMappings.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; + } + + /** 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. */ + private 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(firstAssignedRowId); + payload.writeLong(nextRowId); + payload.writeInt(rowIdMappings.size()); + for (Map.Entry 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 read( + FileIO fileIO, FileStorePathFactory pathFactory, String fileName) throws IOException { + 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 firstAssignedRowId = payload.readLong(); + long nextRowId = payload.readLong(); + 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 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 1b8b245e9b39..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,6 +19,7 @@ package org.apache.paimon.operation; import org.apache.paimon.Snapshot; +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; @@ -381,6 +382,11 @@ protected List planManifestsCleaner( collectUnusedIndexManifests(snapshot, skippingSet, indexFiles, indexManifests); collectUnusedStatisticsManifests(snapshot, skippingSet, statistics); + String reassignPlan = SerializationAssignment.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 = SerializationAssignment.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..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,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.SerializationAssignment; 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, + SerializationAssignment.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(), + SerializationAssignment.withoutPlan(targetSnapshot.properties()), nextRowId, null); @@ -1652,7 +1671,7 @@ private boolean compactManifestOnce() { null, latestSnapshot.watermark(), latestSnapshot.statistics(), - 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 04b4ae63ff06..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,6 +19,7 @@ package org.apache.paimon.operation; import org.apache.paimon.Snapshot; +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; @@ -329,6 +330,11 @@ protected void collectWithoutDataFileWithManifestFlag( .forEach(name -> usedFileWithFlagConsumer.accept(Pair.of(name, false))); } + String reassignPlan = SerializationAssignment.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..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 @@ -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,174 @@ 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(SerializationAssignment.planFile(reassigned)).isNotNull(); + + compactManifests(table); + Snapshot compacted = table.snapshotManager().latestSnapshot(); + assertThat(compacted.id()).isGreaterThan(reassigned.id()); + assertThat(SerializationAssignment.planFile(compacted)).isNull(); + + try (FileStoreCommitImpl commit = + (FileStoreCommitImpl) table.store().newCommit("test-rollback-plan", table)) { + assertThat(commit.rollbackToAsLatest(reassigned)).isTrue(); + } + assertThat(SerializationAssignment.planFile(table.snapshotManager().latestSnapshot())) + .isNull(); + assertThat(SerializationAssignment.planFile(reassigned)).isNotNull(); + SerializationAssignment.read( + table.fileIO(), + table.store().pathFactory(), + SerializationAssignment.planFile(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(SerializationAssignment.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(SerializationAssignment.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( + SerializationAssignment.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(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( + () -> + SerializationAssignment.read( + table.fileIO(), + table.store().pathFactory(), + SerializationAssignment.planFile(snapshot))) + .isInstanceOf(IOException.class) + .hasMessageContaining("checksum"); + + corrupted = bytes.clone(); + corrupted[3] = 99; + overwritePlan(table, path, corrupted); + assertThatThrownBy( + () -> + SerializationAssignment.read( + table.fileIO(), + table.store().pathFactory(), + SerializationAssignment.planFile(snapshot))) + .isInstanceOf(IOException.class) + .hasMessageContaining("version: 99"); + + overwritePlan(table, path, Arrays.copyOf(bytes, bytes.length - 1)); + assertThatThrownBy( + () -> + SerializationAssignment.read( + table.fileIO(), + table.store().pathFactory(), + SerializationAssignment.planFile(snapshot))) + .isInstanceOf(IOException.class); + } + + @Test + public void testRemovingReassignMarkerPreservesOtherProperties() { + Map properties = new HashMap<>(); + properties.put(SerializationAssignment.PLAN_FILE_PROPERTY, "plan"); + properties.put("sequence.generation.max-sequence-number", "100"); + assertThat(SerializationAssignment.withoutPlan(properties)) + .containsExactlyEntriesOf( + Collections.singletonMap("sequence.generation.max-sequence-number", "100")); + assertThat(properties).hasSize(2); + assertThat( + SerializationAssignment.withoutPlan( + Collections.singletonMap( + SerializationAssignment.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()); + SerializationAssignment assignment = + SerializationAssignment.read( + table.fileIO(), + table.store().pathFactory(), + SerializationAssignment.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(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(assignment.map(entry.partition(), oldRange)).isEmpty(); + assertThat(assignment.overlaps(entry.partition(), oldRange)).isFalse(); + } else { + assertThat(assignment.map(entry.partition(), oldRange)).hasValue(newRange); + } + } } @Test @@ -679,6 +852,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 +902,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};