diff --git a/paimon-core/src/main/java/org/apache/paimon/manifest/ManifestEntrySortKey.java b/paimon-core/src/main/java/org/apache/paimon/manifest/ManifestEntrySortKey.java new file mode 100644 index 000000000000..6859e1113018 --- /dev/null +++ b/paimon-core/src/main/java/org/apache/paimon/manifest/ManifestEntrySortKey.java @@ -0,0 +1,143 @@ +/* + * 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.manifest; + +import org.apache.paimon.codegen.CodeGenUtils; +import org.apache.paimon.codegen.RecordComparator; +import org.apache.paimon.data.BinaryRow; +import org.apache.paimon.types.DataType; +import org.apache.paimon.utils.SerializationUtils; + +import java.io.Serializable; +import java.util.List; +import java.util.Objects; + +/** + * A lightweight, serializable sort key for {@link ManifestEntry}. Entries are ordered by {@code + * partition -> bucket -> level -> fileName}, which co-locates the ADD and DELETE of the same file + * (they share the same key) so they always land in the same Spark partition after {@code sortByKey} + * and can be cancelled during the manifest rewrite. The {@code partition -> bucket -> level} prefix + * also keeps {@link ManifestFileMeta} statistics (partitionStats / minBucket / maxBucket / minLevel + * / maxLevel) compact for scan pruning. + * + *

The partition is kept as serialized bytes and deserialized lazily on the first comparison; the + * generated {@link RecordComparator} is also created lazily per executor to avoid being serialized + * across the shuffle. This class is intentionally small so that it can be used as the key of a + * Spark {@code sortByKey} shuffle without moving the whole {@link ManifestEntry} payload. + * + *

Kryo compatibility: the partition is stored as a plain {@code byte[]} rather than a + * {@link BinaryRow} because this key travels through Spark's shuffle, where the serializer is + * {@code KryoSerializer} (Paimon's Spark test base randomly picks Kryo). {@code BinaryRow} only + * implements Java serialization (its {@code BinarySection.writeObject/readObject} callbacks), and + * Kryo does not invoke those callbacks while also skipping the {@code transient segments} field — + * so a {@code BinaryRow} key ends up with {@code null} segments after a Kryo shuffle and NPEs on + * the first comparison. {@code byte[]} and {@code String} are transparent to both Kryo and Java + * serialization, and the {@link BinaryRow} / {@link RecordComparator} are rebuilt lazily on the + * executor after the shuffle. + */ +public class ManifestEntrySortKey implements Serializable, Comparable { + + private static final long serialVersionUID = 1L; + + private final byte[] partitionBytes; + private final int bucket; + private final int level; + private final String fileName; + + private final List partitionFieldTypes; + + private transient BinaryRow partition; + private transient RecordComparator partitionComparator; + + public ManifestEntrySortKey( + BinaryRow partition, + int bucket, + int level, + String fileName, + List partitionFieldTypes) { + this.partitionBytes = SerializationUtils.serializeBinaryRow(partition); + this.bucket = bucket; + this.level = level; + this.fileName = fileName; + this.partitionFieldTypes = partitionFieldTypes; + } + + @Override + public int compareTo(ManifestEntrySortKey other) { + // 1. partition + int cmp = partitionComparator().compare(partition(), other.partition()); + if (cmp != 0) { + return cmp; + } + // 2. bucket + cmp = Integer.compare(bucket, other.bucket); + if (cmp != 0) { + return cmp; + } + // 3. level + cmp = Integer.compare(level, other.level); + if (cmp != 0) { + return cmp; + } + // 4. fileName — co-locates ADD and DELETE of the same file (same key) + return fileName.compareTo(other.fileName); + } + + private BinaryRow partition() { + if (partition == null) { + partition = SerializationUtils.deserializeBinaryRow(partitionBytes); + } + return partition; + } + + private RecordComparator partitionComparator() { + if (partitionComparator == null) { + partitionComparator = CodeGenUtils.newRecordComparator(partitionFieldTypes); + } + return partitionComparator; + } + + public int bucket() { + return bucket; + } + + public int level() { + return level; + } + + @Override + public boolean equals(Object o) { + if (this == o) { + return true; + } + if (o == null || getClass() != o.getClass()) { + return false; + } + ManifestEntrySortKey that = (ManifestEntrySortKey) o; + return bucket == that.bucket + && level == that.level + && Objects.deepEquals(partitionBytes, that.partitionBytes) + && Objects.equals(fileName, that.fileName); + } + + @Override + public int hashCode() { + return Objects.hash(bucket, level, fileName, java.util.Arrays.hashCode(partitionBytes)); + } +} diff --git a/paimon-core/src/main/java/org/apache/paimon/operation/FileStoreCommit.java b/paimon-core/src/main/java/org/apache/paimon/operation/FileStoreCommit.java index b039ffb9e9fc..7caa89e2da28 100644 --- a/paimon-core/src/main/java/org/apache/paimon/operation/FileStoreCommit.java +++ b/paimon-core/src/main/java/org/apache/paimon/operation/FileStoreCommit.java @@ -23,6 +23,7 @@ import org.apache.paimon.disk.IOManager; import org.apache.paimon.fs.FileIO; import org.apache.paimon.manifest.ManifestCommittable; +import org.apache.paimon.manifest.ManifestFileMeta; import org.apache.paimon.operation.metrics.CommitMetrics; import org.apache.paimon.stats.Statistics; import org.apache.paimon.table.sink.CommitMessage; @@ -83,6 +84,16 @@ FileStoreCommit rowIdCheckConflictForMaterializeDvCompaction( /** Compact the manifest entries only. */ void compactManifest(); + /** + * Replace the manifest entries with the given rewritten manifests. The {@code removedManifests} + * are the manifests the caller read and sorted (used for conflict detection); the {@code + * addedManifests} are the sorted rewrite result produced by the caller. The commit reuses the + * optimistic concurrency mode of {@link #compactManifest()}: on conflict, new delta manifests + * added by other commits are appended to the tail of {@code addedManifests}. + */ + void replaceManifest( + List removedManifests, List addedManifests); + /** Roll back to the target snapshot and materialize it as the latest snapshot. */ boolean rollbackToAsLatest(Snapshot targetSnapshot); 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 7f9a02f6022a..107faba81db6 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 @@ -1675,6 +1675,98 @@ public void compactManifest() { } } + @Override + public void replaceManifest( + List removedManifests, List addedManifests) { + int retryCount = 0; + long startMillis = System.currentTimeMillis(); + Set removedPathSet = + removedManifests.stream() + .map(ManifestFileMeta::fileName) + .collect(Collectors.toSet()); + while (true) { + Snapshot latestSnapshot = snapshotManager.latestSnapshot(); + if (latestSnapshot == null) { + throw new RuntimeException("Cannot replace manifests: the table has no snapshot."); + } + + List currentBase = manifestList.readDataManifests(latestSnapshot); + Set currentPathSet = + currentBase.stream() + .map(ManifestFileMeta::fileName) + .collect(Collectors.toSet()); + + if (!currentPathSet.containsAll(removedPathSet)) { + cleanUpRewrittenManifests(addedManifests); + throw new RuntimeException( + "Manifest conflict: the current snapshot does not contain all the " + + "manifests to replace. Another manifest rewrite may have " + + "happened concurrently; please retry."); + } + + List manifestsKept = + currentBase.stream() + .filter(m -> !removedPathSet.contains(m.fileName())) + .collect(Collectors.toList()); + + List manifestsToCommit = new ArrayList<>(manifestsKept); + manifestsToCommit.addAll(addedManifests); + + Pair baseManifestList = manifestList.write(manifestsToCommit); + Pair deltaManifestList = manifestList.write(emptyList()); + + Snapshot newSnapshot = + new Snapshot( + latestSnapshot.id() + 1, + latestSnapshot.schemaId(), + baseManifestList.getLeft(), + baseManifestList.getRight(), + deltaManifestList.getLeft(), + deltaManifestList.getRight(), + null, + null, + latestSnapshot.indexManifest(), + commitUser, + Long.MAX_VALUE, + CommitKind.COMPACT, + System.currentTimeMillis(), + latestSnapshot.totalRecordCount(), + 0L, + null, + latestSnapshot.watermark(), + latestSnapshot.statistics(), + latestSnapshot.properties(), + latestSnapshot.nextRowId(), + null); + + if (commitSnapshotImpl(latestSnapshot, newSnapshot, emptyList())) { + return; + } + + manifestList.delete(deltaManifestList.getLeft()); + manifestList.delete(baseManifestList.getLeft()); + + if (System.currentTimeMillis() - startMillis > options.commitTimeout() + || retryCount >= options.commitMaxRetries()) { + cleanUpRewrittenManifests(addedManifests); + throw new RuntimeException( + String.format( + "Commit failed after %s millis with %s retries, there maybe exist commit conflicts between multiple jobs.", + options.commitTimeout(), retryCount)); + } + + retryWaiter.retryWait(retryCount); + retryCount++; + } + } + + /** Delete the rewritten manifest files produced by this rewrite. */ + private void cleanUpRewrittenManifests(List addedManifests) { + for (ManifestFileMeta manifest : addedManifests) { + manifestFile.delete(manifest.fileName()); + } + } + private boolean compactManifestOnce() { Snapshot latestSnapshot = snapshotManager.latestSnapshot(); diff --git a/paimon-core/src/main/java/org/apache/paimon/table/format/FormatTableCommit.java b/paimon-core/src/main/java/org/apache/paimon/table/format/FormatTableCommit.java index 278e4c4f369f..2b1c483cc75b 100644 --- a/paimon-core/src/main/java/org/apache/paimon/table/format/FormatTableCommit.java +++ b/paimon-core/src/main/java/org/apache/paimon/table/format/FormatTableCommit.java @@ -27,6 +27,7 @@ import org.apache.paimon.fs.FileStatus; import org.apache.paimon.fs.Path; import org.apache.paimon.fs.TwoPhaseOutputStream; +import org.apache.paimon.manifest.ManifestFileMeta; import org.apache.paimon.metrics.MetricRegistry; import org.apache.paimon.options.CatalogOptions; import org.apache.paimon.options.Options; @@ -318,6 +319,12 @@ public void compactManifests() { throw new UnsupportedOperationException(); } + @Override + public void replaceManifests( + List removedManifests, List addedManifests) { + throw new UnsupportedOperationException(); + } + @Override public TableCommit withMetricRegistry(MetricRegistry registry) { throw new UnsupportedOperationException(); diff --git a/paimon-core/src/main/java/org/apache/paimon/table/sink/BatchTableCommit.java b/paimon-core/src/main/java/org/apache/paimon/table/sink/BatchTableCommit.java index 2cfc181eada2..09a4c88d306b 100644 --- a/paimon-core/src/main/java/org/apache/paimon/table/sink/BatchTableCommit.java +++ b/paimon-core/src/main/java/org/apache/paimon/table/sink/BatchTableCommit.java @@ -21,6 +21,7 @@ import org.apache.paimon.Snapshot; import org.apache.paimon.Snapshot.CommitKind; import org.apache.paimon.annotation.Public; +import org.apache.paimon.manifest.ManifestFileMeta; import org.apache.paimon.stats.Statistics; import java.util.List; @@ -73,6 +74,14 @@ public interface BatchTableCommit extends TableCommit { /** Compact the manifest entries. Generates a snapshot with {@link CommitKind#COMPACT}. */ void compactManifests(); + /** + * Replace the manifest entries with the given rewritten manifests. The {@code removedManifests} + * are the manifests the caller read and sorted; the {@code addedManifests} are the sorted + * rewrite result. Generates a snapshot with {@link CommitKind#COMPACT}. + */ + void replaceManifests( + List removedManifests, List addedManifests); + /** Set the logical operation type (e.g. WRITE, DELETE, MERGE) recorded in the snapshot. */ default BatchTableCommit withOperation(Snapshot.Operation operation) { return this; diff --git a/paimon-core/src/main/java/org/apache/paimon/table/sink/TableCommitImpl.java b/paimon-core/src/main/java/org/apache/paimon/table/sink/TableCommitImpl.java index 014b5e64daa1..a3a010c483b6 100644 --- a/paimon-core/src/main/java/org/apache/paimon/table/sink/TableCommitImpl.java +++ b/paimon-core/src/main/java/org/apache/paimon/table/sink/TableCommitImpl.java @@ -28,6 +28,7 @@ import org.apache.paimon.io.DataFileMeta; import org.apache.paimon.io.DataFilePathFactory; import org.apache.paimon.manifest.ManifestCommittable; +import org.apache.paimon.manifest.ManifestFileMeta; import org.apache.paimon.metrics.MetricRegistry; import org.apache.paimon.operation.FileStoreCommit; import org.apache.paimon.operation.PartitionExpire; @@ -235,6 +236,12 @@ public void compactManifests() { commit.compactManifest(); } + @Override + public void replaceManifests( + List removedManifests, List addedManifests) { + commit.replaceManifest(removedManifests, addedManifests); + } + public boolean rollbackToAsLatest(Tag targetTag) { checkCommitted(); boolean success = commit.rollbackToAsLatest(targetTag.trimToSnapshot()); diff --git a/paimon-core/src/test/java/org/apache/paimon/manifest/ManifestEntrySortKeyTest.java b/paimon-core/src/test/java/org/apache/paimon/manifest/ManifestEntrySortKeyTest.java new file mode 100644 index 000000000000..02030364c5e1 --- /dev/null +++ b/paimon-core/src/test/java/org/apache/paimon/manifest/ManifestEntrySortKeyTest.java @@ -0,0 +1,173 @@ +/* + * 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.manifest; + +import org.apache.paimon.data.BinaryRow; +import org.apache.paimon.data.BinaryRowWriter; +import org.apache.paimon.data.BinaryString; +import org.apache.paimon.types.DataType; +import org.apache.paimon.types.DataTypes; +import org.apache.paimon.types.RowType; + +import org.junit.jupiter.api.Test; + +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collections; +import java.util.List; + +import static org.assertj.core.api.Assertions.assertThat; + +/** Tests for {@link ManifestEntrySortKey}. */ +public class ManifestEntrySortKeyTest { + + // partition type: (dt INT, region STRING) + private final List partitionFieldTypes = + RowType.of(DataTypes.INT(), DataTypes.STRING()).getFieldTypes(); + + private ManifestEntrySortKey key(int dt, String region, int bucket, int level) { + return key(dt, region, bucket, level, "f1"); + } + + private ManifestEntrySortKey key(int dt, String region, int bucket, int level, String file) { + return new ManifestEntrySortKey( + partition(dt, region), bucket, level, file, partitionFieldTypes); + } + + private BinaryRow partition(int dt, String region) { + BinaryRow row = new BinaryRow(2); + BinaryRowWriter writer = new BinaryRowWriter(row); + writer.writeInt(0, dt); + writer.writeString(1, BinaryString.fromString(region)); + writer.complete(); + return row; + } + + @Test + public void testOrderByPartitionFirst() { + // same bucket & level, different partition -> ordered by partition + assertThat(key(1, "a", 0, 0)).isLessThan(key(2, "a", 0, 0)); + assertThat(key(2, "a", 0, 0)).isGreaterThan(key(1, "a", 0, 0)); + } + + @Test + public void testPartitionComparesAllFields() { + // dt equal, region differs -> ordered by region (second partition field) + assertThat(key(1, "a", 0, 0)).isLessThan(key(1, "b", 0, 0)); + assertThat(key(1, "b", 0, 0)).isGreaterThan(key(1, "a", 0, 0)); + // dt differs -> dt wins regardless of region + assertThat(key(2, "a", 0, 0)).isGreaterThan(key(1, "z", 0, 0)); + } + + @Test + public void testOrderByBucketWhenPartitionEquals() { + assertThat(key(1, "a", 0, 0)).isLessThan(key(1, "a", 1, 0)); + assertThat(key(1, "a", 1, 0)).isGreaterThan(key(1, "a", 0, 0)); + } + + @Test + public void testOrderByLevelWhenPartitionAndBucketEqual() { + assertThat(key(1, "a", 0, 0)).isLessThan(key(1, "a", 0, 1)); + assertThat(key(1, "a", 0, 1)).isGreaterThan(key(1, "a", 0, 0)); + } + + @Test + public void testOrderByFileNameWhenPartitionBucketLevelEqual() { + // same (p,b,l), different fileName -> ordered by fileName + assertThat(key(1, "a", 0, 0, "f1")).isLessThan(key(1, "a", 0, 0, "f2")); + assertThat(key(1, "a", 0, 0, "f2")).isGreaterThan(key(1, "a", 0, 0, "f1")); + } + + @Test + public void testAddAndDeleteShareSameKey() { + // ADD and DELETE of the same file share the same sort key, so they always land in the + // same Spark partition after sortByKey and can be cancelled. + ManifestEntrySortKey add = key(1, "a", 0, 0, "f1"); + ManifestEntrySortKey delete = key(1, "a", 0, 0, "f1"); + assertThat(add.compareTo(delete)).isEqualTo(0); + assertThat(add).isEqualTo(delete); + } + + @Test + public void testGlobalSort() { + // build keys in a deliberately shuffled order + List keys = + new ArrayList<>( + Arrays.asList( + key(2, "a", 1, 0, "f1"), + key(1, "b", 0, 1, "f1"), + key(1, "a", 0, 0, "f2"), + key(1, "a", 0, 0, "f1"), + key(2, "a", 0, 0, "f2"), + key(1, "a", 0, 1, "f1"), + key(2, "a", 0, 0, "f1"), + key(1, "a", 1, 0, "f1"))); + Collections.sort(keys); + + // expected: partition asc -> bucket asc -> level asc -> fileName asc + List expected = + Arrays.asList( + key(1, "a", 0, 0, "f1"), + key(1, "a", 0, 0, "f2"), + key(1, "a", 0, 1, "f1"), + key(1, "a", 1, 0, "f1"), + key(1, "b", 0, 1, "f1"), + key(2, "a", 0, 0, "f1"), + key(2, "a", 0, 0, "f2"), + key(2, "a", 1, 0, "f1")); + assertThat(keys).isEqualTo(expected); + } + + @Test + public void testEqualsAndHashCode() { + // keys with the same partition/bucket/level/fileName are equal even if built from different + // BinaryRow instances, because the partition is compared by serialized bytes + ManifestEntrySortKey a = key(1, "a", 3, 5, "f1"); + ManifestEntrySortKey b = key(1, "a", 3, 5, "f1"); + assertThat(a).isEqualTo(b); + assertThat(a.hashCode()).isEqualTo(b.hashCode()); + + // any field difference breaks equality + assertThat(a).isNotEqualTo(key(2, "a", 3, 5, "f1")); + assertThat(a).isNotEqualTo(key(1, "b", 3, 5, "f1")); + assertThat(a).isNotEqualTo(key(1, "a", 4, 5, "f1")); + assertThat(a).isNotEqualTo(key(1, "a", 3, 6, "f1")); + assertThat(a).isNotEqualTo(key(1, "a", 3, 5, "f2")); + } + + @Test + public void testSurvivesSerialization() throws Exception { + // the comparator and partition are transient; after Java serialization they must be + // rebuilt lazily and comparison must still work + ManifestEntrySortKey original = key(1, "a", 0, 0, "f1"); + + java.io.ByteArrayOutputStream baos = new java.io.ByteArrayOutputStream(); + try (java.io.ObjectOutputStream oos = new java.io.ObjectOutputStream(baos)) { + oos.writeObject(original); + } + try (java.io.ObjectInputStream ois = + new java.io.ObjectInputStream( + new java.io.ByteArrayInputStream(baos.toByteArray()))) { + ManifestEntrySortKey roundTripped = (ManifestEntrySortKey) ois.readObject(); + assertThat(roundTripped.compareTo(key(1, "a", 0, 0, "f1"))).isEqualTo(0); + assertThat(roundTripped).isLessThan(key(2, "a", 0, 0, "f1")); + assertThat(roundTripped).isLessThan(key(1, "a", 0, 0, "f2")); + } + } +} diff --git a/paimon-core/src/test/java/org/apache/paimon/operation/FileStoreCommitTest.java b/paimon-core/src/test/java/org/apache/paimon/operation/FileStoreCommitTest.java index f326be18dcab..abfc2c93db8d 100644 --- a/paimon-core/src/test/java/org/apache/paimon/operation/FileStoreCommitTest.java +++ b/paimon-core/src/test/java/org/apache/paimon/operation/FileStoreCommitTest.java @@ -50,6 +50,7 @@ import org.apache.paimon.operation.commit.ConflictDetection; import org.apache.paimon.operation.commit.ManifestEntryChanges; import org.apache.paimon.operation.commit.RetryCommitResult; +import org.apache.paimon.options.MemorySize; import org.apache.paimon.predicate.PredicateBuilder; import org.apache.paimon.schema.Schema; import org.apache.paimon.schema.SchemaManager; @@ -98,6 +99,7 @@ import java.util.Map; import java.util.Objects; import java.util.Optional; +import java.util.Set; import java.util.UUID; import java.util.concurrent.ThreadLocalRandom; import java.util.concurrent.atomic.AtomicReference; @@ -2369,4 +2371,102 @@ private void logData(Supplier> supplier, String name) { } LOG.debug("========== End of " + name + " =========="); } + + // --------------------------------------------------------------------------------------------- + // Tests for replaceManifest + // --------------------------------------------------------------------------------------------- + + @Test + public void testReplaceManifestsIdentity() throws Exception { + TestFileStore store = createStore(false, 2); + store.commitData(generateDataList(20), gen::getPartition, kv -> 0); + + Snapshot snapshot = store.snapshotManager().latestSnapshot(); + ManifestList manifestList = store.manifestListFactory().create(); + List manifests = manifestList.readDataManifests(snapshot); + assertThat(manifests).isNotEmpty(); + + Set fileNamesBefore = + manifests.stream().map(ManifestFileMeta::fileName).collect(Collectors.toSet()); + + try (FileStoreCommit commit = store.newCommit()) { + commit.replaceManifest(manifests, manifests); + } + + Snapshot newSnapshot = store.snapshotManager().latestSnapshot(); + assertThat(newSnapshot.id()).isEqualTo(snapshot.id() + 1); + assertThat(newSnapshot.commitKind()).isEqualTo(Snapshot.CommitKind.COMPACT); + List newManifests = manifestList.readDataManifests(newSnapshot); + Set fileNamesAfter = + newManifests.stream().map(ManifestFileMeta::fileName).collect(Collectors.toSet()); + assertThat(fileNamesAfter).isEqualTo(fileNamesBefore); + } + + @Test + public void testReplaceManifestsPreservesConcurrentDelta() throws Exception { + TestFileStore store = createStore(false, 2); + store.commitData(generateDataList(20), gen::getPartition, kv -> 0); + + Snapshot snapshotBefore = store.snapshotManager().latestSnapshot(); + ManifestList manifestList = store.manifestListFactory().create(); + List manifestsBefore = manifestList.readDataManifests(snapshotBefore); + + store.commitData(generateDataList(20), gen::getPartition, kv -> 0); + Snapshot snapshotAfter = store.snapshotManager().latestSnapshot(); + List manifestsAfter = manifestList.readDataManifests(snapshotAfter); + + Set beforeNames = + manifestsBefore.stream() + .map(ManifestFileMeta::fileName) + .collect(Collectors.toSet()); + List deltaManifests = + manifestsAfter.stream() + .filter(m -> !beforeNames.contains(m.fileName())) + .collect(Collectors.toList()); + assertThat(deltaManifests).isNotEmpty(); + + try (FileStoreCommit commit = store.newCommit()) { + commit.replaceManifest(manifestsBefore, manifestsBefore); + } + + Snapshot finalSnapshot = store.snapshotManager().latestSnapshot(); + List finalManifests = manifestList.readDataManifests(finalSnapshot); + Set finalNames = + finalManifests.stream().map(ManifestFileMeta::fileName).collect(Collectors.toSet()); + + for (String name : beforeNames) { + assertThat(finalNames).contains(name); + } + for (ManifestFileMeta delta : deltaManifests) { + assertThat(finalNames).contains(delta.fileName()); + } + } + + @Test + public void testReplaceManifestsThrowsOnConcurrentCompact() throws Exception { + // use a small manifest target size so compactManifest actually rewrites the manifests + // (changes file names), which makes the stale manifestsBefore absent from the current base + Map options = new HashMap<>(); + options.put(CoreOptions.MANIFEST_TARGET_FILE_SIZE.key(), "1KB"); + options.put( + CoreOptions.MANIFEST_FULL_COMPACTION_FILE_SIZE.key(), + MemorySize.ofBytes(1).toString()); + TestFileStore store = createStore(false, 2, CoreOptions.ChangelogProducer.NONE, options); + for (int i = 0; i < 3; i++) { + store.commitData(generateDataList(10), gen::getPartition, kv -> 0); + } + + Snapshot snapshotBefore = store.snapshotManager().latestSnapshot(); + ManifestList manifestList = store.manifestListFactory().create(); + List manifestsBefore = manifestList.readDataManifests(snapshotBefore); + + try (FileStoreCommit commit = store.newCommit()) { + commit.compactManifest(); + } + + try (FileStoreCommit commit = store.newCommit()) { + assertThatThrownBy(() -> commit.replaceManifest(manifestsBefore, manifestsBefore)) + .hasMessageContaining("Manifest conflict"); + } + } } diff --git a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/SparkProcedures.java b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/SparkProcedures.java index 60b5747e3db7..facdbec37753 100644 --- a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/SparkProcedures.java +++ b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/SparkProcedures.java @@ -58,6 +58,7 @@ import org.apache.paimon.spark.procedure.RescaleProcedure; import org.apache.paimon.spark.procedure.ResetConsumerProcedure; import org.apache.paimon.spark.procedure.RewriteFileIndexProcedure; +import org.apache.paimon.spark.procedure.RewriteManifestProcedure; import org.apache.paimon.spark.procedure.RollbackProcedure; import org.apache.paimon.spark.procedure.RollbackToTimestampProcedure; import org.apache.paimon.spark.procedure.RollbackToWatermarkProcedure; @@ -124,6 +125,7 @@ private static Map> initProcedureBuilders() { procedureBuilders.put("reset_consumer", ResetConsumerProcedure::builder); procedureBuilders.put("mark_partition_done", MarkPartitionDoneProcedure::builder); procedureBuilders.put("compact_manifest", CompactManifestProcedure::builder); + procedureBuilders.put("rewrite_manifest", RewriteManifestProcedure::builder); procedureBuilders.put("clear_consumers", ClearConsumersProcedure::builder); procedureBuilders.put("alter_view_dialect", AlterViewDialectProcedure::builder); procedureBuilders.put("create_function", CreateFunctionProcedure::builder); diff --git a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/RewriteManifestProcedure.java b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/RewriteManifestProcedure.java new file mode 100644 index 000000000000..bb2374f1b264 --- /dev/null +++ b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/RewriteManifestProcedure.java @@ -0,0 +1,479 @@ +/* + * 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.spark.procedure; + +import org.apache.paimon.Snapshot; +import org.apache.paimon.fs.Path; +import org.apache.paimon.manifest.FileKind; +import org.apache.paimon.manifest.ManifestEntry; +import org.apache.paimon.manifest.ManifestEntrySerializer; +import org.apache.paimon.manifest.ManifestEntrySortKey; +import org.apache.paimon.manifest.ManifestFile; +import org.apache.paimon.manifest.ManifestFile.ManifestEntryWriter; +import org.apache.paimon.manifest.ManifestFileMeta; +import org.apache.paimon.manifest.ManifestFileMetaSerializer; +import org.apache.paimon.manifest.ManifestList; +import org.apache.paimon.partition.PartitionPredicate; +import org.apache.paimon.predicate.Predicate; +import org.apache.paimon.predicate.PredicateBuilder; +import org.apache.paimon.spark.catalyst.analysis.expressions.ExpressionUtils; +import org.apache.paimon.stats.SimpleStats; +import org.apache.paimon.table.FileStoreTable; +import org.apache.paimon.table.Table; +import org.apache.paimon.table.sink.BatchTableCommit; +import org.apache.paimon.types.DataType; +import org.apache.paimon.types.RowType; +import org.apache.paimon.utils.FileStorePathFactory; +import org.apache.paimon.utils.ProcedureUtils; +import org.apache.paimon.utils.StringUtils; + +import org.apache.spark.api.java.JavaPairRDD; +import org.apache.spark.api.java.JavaSparkContext; +import org.apache.spark.api.java.function.FlatMapFunction; +import org.apache.spark.api.java.function.PairFlatMapFunction; +import org.apache.spark.sql.catalyst.InternalRow; +import org.apache.spark.sql.catalyst.expressions.Expression; +import org.apache.spark.sql.catalyst.plans.logical.LogicalPlan; +import org.apache.spark.sql.connector.catalog.Identifier; +import org.apache.spark.sql.connector.catalog.TableCatalog; +import org.apache.spark.sql.execution.datasources.v2.DataSourceV2Relation; +import org.apache.spark.sql.types.DataTypes; +import org.apache.spark.sql.types.Metadata; +import org.apache.spark.sql.types.StructField; +import org.apache.spark.sql.types.StructType; + +import java.io.IOException; +import java.util.ArrayList; +import java.util.Collections; +import java.util.HashMap; +import java.util.Iterator; +import java.util.List; +import java.util.stream.Collectors; + +import scala.Tuple2; + +import static org.apache.paimon.utils.Preconditions.checkArgument; +import static org.apache.spark.sql.types.DataTypes.StringType; + +/** + * Rewrite manifest procedure. It reads all manifest entries, sorts them globally by {@code + * partition -> bucket -> level -> fileName} (canceling ADD/DELETE pairs along the way) and writes + * them back as new manifest files so that {@link ManifestFileMeta} statistics become more compact + * for scan pruning. The sort runs distributed via a Spark {@code sortByKey} shuffle. + * + *

An optional {@code where} clause restricts the rewrite to manifests whose partition stats may + * match the predicate; the remaining manifests are left untouched. + * + * 


+ *  CALL sys.rewrite_manifest(table => 'tableId')
+ *  CALL sys.rewrite_manifest(table => 'tableId', where => 'dt = "2024-01-01"')
+ *
+ */ +public class RewriteManifestProcedure extends BaseProcedure { + + private static final ProcedureParameter[] PARAMETERS = + new ProcedureParameter[] { + ProcedureParameter.required("table", StringType), + ProcedureParameter.optional("where", StringType), + ProcedureParameter.optional("options", StringType) + }; + + private static final StructType OUTPUT_TYPE = + new StructType( + new StructField[] { + new StructField( + "rewritten_manifests_count", + DataTypes.IntegerType, + true, + Metadata.empty()), + new StructField( + "added_manifests_count", + DataTypes.IntegerType, + true, + Metadata.empty()) + }); + + protected RewriteManifestProcedure(TableCatalog tableCatalog) { + super(tableCatalog); + } + + @Override + public ProcedureParameter[] parameters() { + return PARAMETERS; + } + + @Override + public StructType outputType() { + return OUTPUT_TYPE; + } + + @Override + public InternalRow[] call(InternalRow args) { + Identifier tableIdent = toIdentifier(args.getString(0), PARAMETERS[0].name()); + String where = args.isNullAt(1) ? null : args.getString(1); + String options = args.isNullAt(2) ? null : args.getString(2); + + Table table = loadSparkTable(tableIdent).getTable(); + HashMap dynamicOptions = new HashMap<>(); + ProcedureUtils.putAllOptions(dynamicOptions, options); + FileStoreTable fileStoreTable = (FileStoreTable) table.copy(dynamicOptions); + + // 1. read the latest snapshot and its data manifests + Snapshot latestSnapshot = fileStoreTable.store().snapshotManager().latestSnapshot(); + if (latestSnapshot == null) { + return new InternalRow[] {newInternalRow(0, 0)}; + } + ManifestList manifestList = fileStoreTable.store().manifestListFactory().create(); + List currentManifests = manifestList.readDataManifests(latestSnapshot); + if (currentManifests.isEmpty()) { + return new InternalRow[] {newInternalRow(0, 0)}; + } + + // 2. filter manifests by the optional where clause (partition-stats pruning) + List manifestsToRewrite = currentManifests; + PartitionPredicate partitionPredicate = resolvePartitionPredicate(tableIdent, table, where); + if (partitionPredicate != null) { + manifestsToRewrite = filterManifests(manifestsToRewrite, partitionPredicate); + if (manifestsToRewrite.isEmpty()) { + return new InternalRow[] {newInternalRow(0, 0)}; + } + } + + // 3. globally sort manifest entries and write them back as new manifest files + List newManifests = rewriteManifests(fileStoreTable, manifestsToRewrite); + + // 4. commit the rewritten manifests (optimistic concurrency with retry) + try (BatchTableCommit commit = fileStoreTable.newBatchWriteBuilder().newCommit()) { + commit.replaceManifests(manifestsToRewrite, newManifests); + } catch (Exception e) { + throw new RuntimeException(e); + } + + // rewritten_manifests_count: number of manifests that were rewritten (input) + // added_manifests_count: number of new manifests produced by the rewrite (output) + int rewrittenCount = manifestsToRewrite.size(); + int addedCount = newManifests.size(); + return new InternalRow[] {newInternalRow(rewrittenCount, addedCount)}; + } + + /** + * Parse the {@code where} SQL string into a {@link PartitionPredicate}, validating that it only + * references partition columns. Returns {@code null} when {@code where} is blank. + */ + private PartitionPredicate resolvePartitionPredicate( + Identifier tableIdent, Table table, String where) { + if (StringUtils.isNullOrWhitespaceOnly(where)) { + return null; + } + FileStoreTable fileStoreTable = (FileStoreTable) table; + DataSourceV2Relation relation = createRelation(tableIdent); + Expression condition = ExpressionUtils.resolveFilter(spark(), relation, where); + checkArgument( + ExpressionUtils.isValidPredicate( + spark(), condition, fileStoreTable.partitionKeys().toArray(new String[0])), + "Only partition predicate is supported, your predicate is %s, but partition keys are %s", + condition, + fileStoreTable.partitionKeys()); + Predicate predicate = + ExpressionUtils.convertConditionToPaimonPredicate( + condition, + ((LogicalPlan) relation).output(), + table.rowType(), + false) + .getOrElse(null); + + // the predicate references fields by their index in the full row type; map them to the + // partition type index so PartitionPredicate can test against partitionStats / BinaryRow + // partitions (which only contain partition columns) + List partitionKeys = fileStoreTable.partitionKeys(); + int[] fieldIdxToPartitionIdx = + fileStoreTable.schema().fields().stream() + .mapToInt(f -> partitionKeys.indexOf(f.name())) + .toArray(); + Predicate partitionPredicate = + PredicateBuilder.transformFieldMapping(predicate, fieldIdxToPartitionIdx) + .orElse(null); + + RowType partitionType = fileStoreTable.store().partitionType(); + return PartitionPredicate.fromPredicate(partitionType, partitionPredicate); + } + + /** Keep only manifests whose partition stats may match the predicate. */ + private List filterManifests( + List manifests, PartitionPredicate predicate) { + return manifests.stream() + .filter( + m -> { + SimpleStats stats = m.partitionStats(); + return predicate.test( + m.numAddedFiles() + m.numDeletedFiles(), + stats.minValues(), + stats.maxValues(), + stats.nullCounts()); + }) + .collect(Collectors.toList()); + } + + private List rewriteManifests( + FileStoreTable table, List currentManifests) { + List partitionFieldTypes = table.store().partitionType().getFieldTypes(); + ManifestFileMetaSerializer metaSerializer = new ManifestFileMetaSerializer(); + + // serialize ManifestFileMeta to byte[] so they can travel through the RDD + List serializedMetas = new ArrayList<>(currentManifests.size()); + for (ManifestFileMeta meta : currentManifests) { + try { + serializedMetas.add(metaSerializer.serializeToBytes(meta)); + } catch (IOException e) { + throw new RuntimeException(e); + } + } + + int numPartitions = computeNumPartitions(table, currentManifests); + + JavaSparkContext jsc = new JavaSparkContext(spark().sparkContext()); + + // The table is captured by the closure (not broadcast) so that Spark serializes it with + // Java serialization (ClosureCleaner always uses JavaSerializer). This is required because + // the table's FileIO (e.g. HadoopFileIO) holds a SerializableConfiguration whose transient + // Configuration is restored only via Java writeObject/readObject — Kryo (used by broadcast + // when spark.serializer=KryoSerializer) would skip those callbacks and leave a null conf. + FileStoreTable tableRef = table; + + // Step 1: read all entries (ADD and DELETE) and pair them with their sort key + JavaPairRDD entryRDD = + jsc.parallelize(serializedMetas, Math.min(serializedMetas.size(), numPartitions)) + .flatMapToPair( + (PairFlatMapFunction) + manifestBytes -> { + ManifestFileMetaSerializer ser = + new ManifestFileMetaSerializer(); + ManifestFileMeta manifest = + ser.deserializeFromBytes(manifestBytes); + ManifestFile manifestFile = + tableRef.store().manifestFileFactory().create(); + ManifestEntrySerializer entrySer = + new ManifestEntrySerializer(); + + List> pairs = + new ArrayList<>(); + for (ManifestEntry entry : + manifestFile.read( + manifest.fileName(), + manifest.fileSize())) { + ManifestEntrySortKey key = + new ManifestEntrySortKey( + entry.partition(), + entry.bucket(), + entry.file().level(), + entry.fileName(), + partitionFieldTypes); + pairs.add( + new Tuple2<>( + key, + entrySer.serializeToBytes(entry))); + } + return pairs.iterator(); + }); + + // Step 2: global sort. The sort key is (partition, bucket, level, fileName), so + // the ADD and DELETE of the same file are adjacent. + JavaPairRDD sortedRDD = + entryRDD.sortByKey(true, numPartitions); + + // Step 3: within each partition, stream the sorted entries, cancel ADD/DELETE pairs per + // manifest entry, and write surviving entries to a single manifest file. Because the sort + // key + // is (partition, bucket, level, fileName), entries of the same file are consecutive; the + // per-key buffer holds at most one ADD and one DELETE, so memory is negligible. Each task + // produces at most one manifest, sized roughly to the manifest target file size. + List serializedResult = + sortedRDD + .mapPartitions( + (FlatMapFunction< + Iterator>, + byte[]>) + iter -> { + FileStoreTable t = tableRef; + FileStorePathFactory pathFactory = + t.store().pathFactory(); + Path manifestPath = pathFactory.newManifestFile(); + ManifestFile manifestFile = + t.store().manifestFileFactory().create(); + ManifestEntryWriter writer = + manifestFile.createManifestEntryWriter( + manifestPath); + ManifestEntrySerializer entrySer = + new ManifestEntrySerializer(); + + // per-current-key buffer: at most one ADD and one + // DELETE for the same Identifier + ManifestEntrySortKey currentKey = null; + ManifestEntry bufferedAdd = null; + ManifestEntry bufferedDelete = null; + try { + while (iter.hasNext()) { + Tuple2 pair = + iter.next(); + ManifestEntrySortKey key = pair._1; + ManifestEntry entry = + entrySer.deserializeFromBytes(pair._2); + + if (currentKey != null + && currentKey.compareTo(key) != 0) { + // key changed: flush the previous group + ManifestEntry survived = + mergeGroup( + bufferedAdd, + bufferedDelete); + if (survived != null) { + writer.write(survived); + } + bufferedAdd = null; + bufferedDelete = null; + } + + currentKey = key; + if (entry.kind() == FileKind.ADD) { + if (bufferedAdd != null) { + throw new IllegalStateException( + "Duplicate ADD entry for " + + entry.identifier()); + } + bufferedAdd = entry; + } else { + if (bufferedDelete != null) { + throw new IllegalStateException( + "Duplicate DELETE entry for " + + entry.identifier()); + } + bufferedDelete = entry; + } + } + + // flush the last group + if (currentKey != null) { + ManifestEntry survived = + mergeGroup(bufferedAdd, bufferedDelete); + if (survived != null) { + writer.write(survived); + } + } + } finally { + writer.close(); + } + + if (writer.recordCount() == 0) { + // nothing survived — delete the empty file and + // emit nothing + manifestFile.delete(writer.path().getName()); + return Collections.emptyList().iterator(); + } + ManifestFileMeta newMeta = writer.result(); + ManifestFileMetaSerializer ser = + new ManifestFileMetaSerializer(); + return Collections.singletonList( + ser.serializeToBytes(newMeta)) + .iterator(); + }) + .collect(); + + List newManifests = new ArrayList<>(serializedResult.size()); + for (byte[] bytes : serializedResult) { + try { + newManifests.add(metaSerializer.deserializeFromBytes(bytes)); + } catch (IOException e) { + throw new RuntimeException(e); + } + } + return newManifests; + } + + /** + * Merge the ADD and DELETE entries of the same file (same Identifier) in an order-independent + * way. Returns the surviving ADD if only an ADD is present, {@code null} if the ADD and DELETE + * cancel each other out, or the DELETE if only a DELETE is present (kept so it can match an ADD + * in a previous manifest). + */ + private static ManifestEntry mergeGroup(ManifestEntry addEntry, ManifestEntry deleteEntry) { + if (addEntry != null && deleteEntry != null) { + // ADD + DELETE cancel out + return null; + } + if (addEntry != null) { + return addEntry; + } + return deleteEntry; + } + + /** + * Estimate the number of output manifests (and thus the sort parallelism) from the input + * manifests. The surviving entry count is {@code added - deleted} (ADD/DELETE pairs cancel), + * the average entry size is {@code totalSize / (added + deleted)}, and the estimated output + * size is {@code avgEntrySize * survivingEntries}. The parallelism is the floor of that over + * the manifest target file size, so each task produces roughly one manifest of the target size. + */ + private int computeNumPartitions(FileStoreTable table, List manifests) { + long targetSizeBytes = table.coreOptions().manifestTargetSize().getBytes(); + long totalSizeBytes = 0L; + long addedEntries = 0L; + long deletedEntries = 0L; + for (ManifestFileMeta manifest : manifests) { + totalSizeBytes += manifest.fileSize(); + addedEntries += manifest.numAddedFiles(); + deletedEntries += manifest.numDeletedFiles(); + } + + if (totalSizeBytes <= 0) { + throw new IllegalStateException( + "Cannot compute parallelism: total manifest size is " + totalSizeBytes); + } + long totalEntries = addedEntries + deletedEntries; + if (totalEntries <= 0) { + throw new IllegalStateException( + "Cannot compute parallelism: total manifest entries is " + totalEntries); + } + long survivingEntries = addedEntries - deletedEntries; + if (survivingEntries < 0) { + throw new IllegalStateException( + "Cannot compute parallelism: surviving entries (added - deleted) is " + + survivingEntries); + } + double avgEntrySizeBytes = (double) totalSizeBytes / totalEntries; + long estimatedOutputSizeBytes = (long) (avgEntrySizeBytes * survivingEntries); + // floor division: prefer fewer, slightly-over-target manifests over more, under-target + // ones. For example, 35M estimated with 8M target -> 4 manifests (~8.75M each), not 5 + // (~7M each). At least one task is always produced. + return (int) Math.max(1, estimatedOutputSizeBytes / targetSizeBytes); + } + + @Override + public String description() { + return "This procedure rewrites and globally sorts manifest entries."; + } + + public static ProcedureBuilder builder() { + return new Builder() { + @Override + public RewriteManifestProcedure doBuild() { + return new RewriteManifestProcedure(tableCatalog()); + } + }; + } +} diff --git a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/procedure/RewriteManifestProcedureTest.scala b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/procedure/RewriteManifestProcedureTest.scala new file mode 100644 index 000000000000..43c066d1634c --- /dev/null +++ b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/procedure/RewriteManifestProcedureTest.scala @@ -0,0 +1,364 @@ +/* + * 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.spark.procedure + +import org.apache.paimon.spark.PaimonSparkTestBase + +import org.apache.spark.sql.streaming.StreamTest +import org.assertj.core.api.Assertions + +import scala.jdk.CollectionConverters._ + +/** Test rewrite manifest procedure. See [[RewriteManifestProcedure]]. */ +class RewriteManifestProcedureTest extends PaimonSparkTestBase with StreamTest { + + test("Paimon Procedure: rewrite manifest cleans delete entries and keeps data intact") { + // A small manifest target file size forces several manifest files, so that the global sort + // actually reshuffles entries across files. + spark.sql(s""" + |CREATE TABLE T (id INT, value STRING, dt STRING) + |TBLPROPERTIES ( + | 'bucket' = '2', + | 'bucket-key' = 'id', + | 'write-only' = 'true', + | 'file.format' = 'avro', + | 'manifest.target-file-size' = '1KB' + |) + |PARTITIONED BY (dt) + |""".stripMargin) + + // Insert partitions in non-sorted order across multiple commits so that manifest entries are + // interleaved and out of partition order. + spark.sql(s"INSERT INTO T VALUES (1, 'a', '2024-01-03')") + spark.sql(s"INSERT INTO T VALUES (2, 'b', '2024-01-01')") + spark.sql(s"INSERT INTO T VALUES (3, 'c', '2024-01-02')") + spark.sql(s"INSERT INTO T VALUES (4, 'd', '2024-01-01')") + spark.sql(s"INSERT INTO T VALUES (5, 'e', '2024-01-03')") + + // Partition-level overwrites produce ADD/DELETE pairs that rewrite_manifest should cancel, + // without dropping the other partitions. + spark.sql(s"INSERT OVERWRITE T PARTITION (dt = '2024-01-03') VALUES (1, 'a2')") + spark.sql(s"INSERT OVERWRITE T PARTITION (dt = '2024-01-01') VALUES (2, 'b2'), (4, 'd2')") + + Thread.sleep(10000) + + val expectedCount = spark.sql("SELECT count(*) FROM T").collectAsList().get(0).getLong(0) + val expectedIdSum = spark.sql("SELECT sum(id) FROM T").collectAsList().get(0).getLong(0) + + // before rewrite there should be some delete entries + val beforeDeleted = + spark + .sql("SELECT sum(num_deleted_files) FROM `T$manifests`") + .collectAsList() + .get(0) + .getLong(0) + Assertions.assertThat(beforeDeleted).isGreaterThan(0L) + + val beforeManifests = + spark.sql("SELECT count(*) FROM `T$manifests`").collectAsList().get(0).getLong(0) + + val result = + spark.sql("CALL sys.rewrite_manifest(table => 'T')").collectAsList().get(0) + + // rewritten_manifests_count == before (all manifests rewritten without where) + // added_manifests_count > 0 (new manifests produced) + val rewrittenCount = result.getInt(0) + val addedCount = result.getInt(1) + Assertions.assertThat(rewrittenCount).isEqualTo(beforeManifests.toInt) + Assertions.assertThat(addedCount).isGreaterThan(0) + + // after rewrite all delete entries must be cleaned + val afterDeleted = + spark + .sql("SELECT sum(num_deleted_files) FROM `T$manifests`") + .collectAsList() + .get(0) + .getLong(0) + Assertions.assertThat(afterDeleted).isEqualTo(0L) + + // data must be intact + Assertions + .assertThat(spark.sql("SELECT count(*) FROM T").collectAsList().get(0).getLong(0)) + .isEqualTo(expectedCount) + Assertions + .assertThat(spark.sql("SELECT sum(id) FROM T").collectAsList().get(0).getLong(0)) + .isEqualTo(expectedIdSum) + } + + test("Paimon Procedure: rewrite manifest orders manifest partitions globally") { + spark.sql(s""" + |CREATE TABLE T2 (id INT, dt STRING) + |TBLPROPERTIES ( + | 'bucket' = '-1', + | 'write-only' = 'true', + | 'manifest.target-file-size' = '1KB', + | 'file.format' = 'avro' + |) + |PARTITIONED BY (dt) + |""".stripMargin) + + // insert partitions in reverse order across commits + spark.sql(s"INSERT INTO T2 VALUES (1, '2024-01-03')") + spark.sql(s"INSERT INTO T2 VALUES (2, '2024-01-02')") + spark.sql(s"INSERT INTO T2 VALUES (3, '2024-01-01')") + spark.sql(s"INSERT INTO T2 VALUES (4, '2024-01-03')") + spark.sql(s"INSERT INTO T2 VALUES (5, '2024-01-01')") + + Thread.sleep(10000) + + val expectedCount = spark.sql("SELECT count(*) FROM T2").collectAsList().get(0).getLong(0) + + spark.sql("CALL sys.rewrite_manifest(table => 'T2')") + + // After global sort, manifest partition ranges must not overlap: sorting manifests by their + // min partition must yield a sequence where each manifest's min is >= the previous manifest's + // max. This is robust regardless of the row order Spark returns for the system table. + val ranges = spark + .sql("SELECT min_partition_stats, max_partition_stats FROM `T2$manifests`") + .collectAsList() + .asScala + .filter(r => !r.isNullAt(0)) + .map(r => (r.getString(0), r.getString(1))) + .sortBy(_._1) + + for (i <- 1 until ranges.length) { + Assertions.assertThat(ranges(i)._1.compareTo(ranges(i - 1)._2) >= 0).isTrue + } + + // data intact + Assertions + .assertThat(spark.sql("SELECT count(*) FROM T2").collectAsList().get(0).getLong(0)) + .isEqualTo(expectedCount) + } + + test("Paimon Procedure: rewrite manifest with where only rewrites matching manifests") { + spark.sql(s""" + |CREATE TABLE T3 (id INT, dt STRING) + |TBLPROPERTIES ( + | 'bucket' = '-1', + | 'write-only' = 'true', + | 'manifest.target-file-size' = '1KB' + |) + |PARTITIONED BY (dt) + |""".stripMargin) + + // insert into multiple partitions across commits + spark.sql(s"INSERT INTO T3 VALUES (1, '2024-01-01')") + spark.sql(s"INSERT INTO T3 VALUES (2, '2024-01-02')") + spark.sql(s"INSERT INTO T3 VALUES (3, '2024-01-03')") + spark.sql(s"INSERT INTO T3 VALUES (4, '2024-01-01')") + + Thread.sleep(10000) + + val expectedCount = spark.sql("SELECT count(*) FROM T3").collectAsList().get(0).getLong(0) + + // record manifest file names before rewrite + val allManifestsBefore = + spark.sql("SELECT file_name FROM `T3$manifests`").collectAsList().asScala.map(_.getString(0)) + + // rewrite only manifests that may match dt = '2024-01-01' + spark.sql("CALL sys.rewrite_manifest(table => 'T3', where => 'dt = \"2024-01-01\"')") + + val allManifestsAfter = + spark.sql("SELECT file_name FROM `T3$manifests`").collectAsList().asScala.map(_.getString(0)) + + // some manifests should have been rewritten (new file names appear) + val newManifests = allManifestsAfter.filter(!allManifestsBefore.contains(_)) + Assertions.assertThat(newManifests.nonEmpty).isTrue + + // but not all manifests are rewritten — at least one original manifest survives + val survivingOriginals = allManifestsBefore.filter(allManifestsAfter.contains(_)) + Assertions.assertThat(survivingOriginals.nonEmpty).isTrue + + // data intact + Assertions + .assertThat(spark.sql("SELECT count(*) FROM T3").collectAsList().get(0).getLong(0)) + .isEqualTo(expectedCount) + Assertions + .assertThat( + spark + .sql("SELECT count(*) FROM T3 WHERE dt = '2024-01-01'") + .collectAsList() + .get(0) + .getLong(0)) + .isEqualTo(2) + } + + test("Paimon Procedure: rewrite manifest with where matching nothing returns zero") { + spark.sql(s""" + |CREATE TABLE T4 (id INT, dt STRING) + |TBLPROPERTIES ( + | 'bucket' = '-1', + | 'write-only' = 'true', + | 'manifest.target-file-size' = '1KB', + | 'file.format' = 'avro' + |) + |PARTITIONED BY (dt) + |""".stripMargin) + + spark.sql(s"INSERT INTO T4 VALUES (1, '2024-01-01')") + spark.sql(s"INSERT INTO T4 VALUES (2, '2024-01-02')") + Thread.sleep(10000) + + val manifestsBefore = + spark.sql("SELECT file_name FROM `T4$manifests`").collectAsList().asScala.map(_.getString(0)) + + // where matches no partition — nothing should be rewritten + val result = + spark + .sql("CALL sys.rewrite_manifest(table => 'T4', where => 'dt = \"1999-01-01\"')") + .collectAsList() + .get(0) + Assertions.assertThat(result.getInt(0)).isEqualTo(0) + Assertions.assertThat(result.getInt(1)).isEqualTo(0) + + // all original manifests survive untouched + val manifestsAfter = + spark.sql("SELECT file_name FROM `T4$manifests`").collectAsList().asScala.map(_.getString(0)) + Assertions.assertThat(manifestsAfter).isEqualTo(manifestsBefore) + + // data intact + Assertions + .assertThat(spark.sql("SELECT count(*) FROM T4").collectAsList().get(0).getLong(0)) + .isEqualTo(2) + } + + test("Paimon Procedure: rewrite manifest with range where rewrites matching partitions") { + spark.sql(s""" + |CREATE TABLE T5 (id INT, dt STRING) + |TBLPROPERTIES ( + | 'bucket' = '-1', + | 'write-only' = 'true', + | 'manifest.target-file-size' = '1KB', + | 'file.format' = 'avro' + |) + |PARTITIONED BY (dt) + |""".stripMargin) + + spark.sql(s"INSERT INTO T5 VALUES (1, '2024-01-01')") + spark.sql(s"INSERT INTO T5 VALUES (2, '2024-01-02')") + spark.sql(s"INSERT INTO T5 VALUES (3, '2024-01-03')") + spark.sql(s"INSERT INTO T5 VALUES (4, '2024-01-04')") + Thread.sleep(10000) + + val expectedCount = spark.sql("SELECT count(*) FROM T5").collectAsList().get(0).getLong(0) + + // range where: only Jan 1-2 + spark.sql( + "CALL sys.rewrite_manifest(table => 'T5', where => 'dt >= \"2024-01-01\" AND dt <= \"2024-01-02\"')") + + // data intact across all partitions + Assertions + .assertThat(spark.sql("SELECT count(*) FROM T5").collectAsList().get(0).getLong(0)) + .isEqualTo(expectedCount) + Assertions + .assertThat( + spark + .sql("SELECT count(*) FROM T5 WHERE dt >= '2024-01-01' AND dt <= '2024-01-02'") + .collectAsList() + .get(0) + .getLong(0)) + .isEqualTo(2) + } + + test("Paimon Procedure: rewrite manifest on unpartitioned table") { + spark.sql(s""" + |CREATE TABLE T6 (id INT, value STRING) + |TBLPROPERTIES ( + | 'bucket' = '-1', + | 'write-only' = 'true', + | 'manifest.target-file-size' = '1KB', + | 'file.format' = 'avro' + |) + |""".stripMargin) + + spark.sql(s"INSERT INTO T6 VALUES (1, 'a')") + spark.sql(s"INSERT INTO T6 VALUES (2, 'b')") + Thread.sleep(10000) + + val expectedCount = spark.sql("SELECT count(*) FROM T6").collectAsList().get(0).getLong(0) + + spark.sql("CALL sys.rewrite_manifest(table => 'T6')") + + // data intact + Assertions + .assertThat(spark.sql("SELECT count(*) FROM T6").collectAsList().get(0).getLong(0)) + .isEqualTo(expectedCount) + } + + test("Paimon Procedure: rewrite manifest with non-partition where throws") { + spark.sql(s""" + |CREATE TABLE T7 (id INT, dt STRING) + |TBLPROPERTIES ( + | 'bucket' = '-1', + | 'write-only' = 'true', + | 'manifest.target-file-size' = '1KB', + | 'file.format' = 'avro' + |) + |PARTITIONED BY (dt) + |""".stripMargin) + + spark.sql(s"INSERT INTO T7 VALUES (1, '2024-01-01')") + Thread.sleep(10000) + + // where on a non-partition column (id) must fail + Assertions + .assertThatThrownBy( + () => spark.sql("CALL sys.rewrite_manifest(table => 'T7', where => 'id = 1')")) + .hasMessageContaining("Only partition predicate is supported") + } + + test("Paimon Procedure: rewritten manifest sizes are within the target bound") { + // Use a small target file size and enough data so that multiple manifests are produced. + // Each task writes a single non-rolling manifest, so every output manifest should be at most + // a modest multiple of the target size (allowing headroom for a single partition that + // slightly exceeds the range estimate). + val targetSize = 1024L + spark.sql(s""" + |CREATE TABLE T8 (id INT, value STRING, dt STRING) + |TBLPROPERTIES ( + | 'bucket' = '4', + | 'bucket-key' = 'id', + | 'write-only' = 'true', + | 'file.format' = 'avro', + | 'manifest.target-file-size' = '${targetSize}B' + |) + |PARTITIONED BY (dt) + |""".stripMargin) + + // Insert enough rows across enough partitions to produce multiple output manifests. + for (dt <- 0 until 20) { + val values = (0 until 20).map(i => s"(${dt * 20 + i}, '${"x" * 50}', '2024-01-${dt + 1}')") + spark.sql(s"INSERT INTO T8 VALUES ${values.mkString(", ")}") + } + Thread.sleep(10000) + + spark.sql("CALL sys.rewrite_manifest(table => 'T8')") + + val sizes = + spark.sql("SELECT file_size FROM `T8$manifests`").collectAsList().asScala.map(_.getLong(0)) + Assertions.assertThat(sizes.nonEmpty).isTrue + // Every rewritten manifest should be <= 3x target size (generous bound for the last manifest + // of a task and range-partitioner skew). The key invariant: no manifest should be wildly + // oversized (e.g. 10x), which would indicate the parallelism estimate is wrong. + for (size <- sizes) { + Assertions.assertThat(size).isLessThanOrEqualTo(targetSize * 3) + } + } +}