This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/paimon.git
The following commit(s) were added to refs/heads/master by this push:
new d15d250cf6 [core][spark] Support remove_unexisting_manifests procedure
(#10083)
d15d250cf6 is described below
commit d15d250cf63e601b448318392158f1531b30193e
Author: Wenchao Wu <[email protected]>
AuthorDate: Thu Sep 24 14:44:43 2026 +0800
[core][spark] Support remove_unexisting_manifests procedure (#10083)
---
docs/docs/spark/procedures.md | 1 +
docs/docs/spark/procedures/maintenance.md | 16 ++
.../operation/RemoveUnexistingManifests.java | 78 +++++---
.../operation/RemoveUnexistingManifestsTest.java | 213 +++++++++++++++++++++
.../action/RemoveUnexistingManifestsAction.java | 74 +------
.../org/apache/paimon/spark/SparkProcedures.java | 3 +
.../RemoveUnexistingManifestsProcedure.java | 102 ++++++++++
.../RemoveUnexistingManifestsProcedureTest.scala | 159 +++++++++++++++
8 files changed, 547 insertions(+), 99 deletions(-)
diff --git a/docs/docs/spark/procedures.md b/docs/docs/spark/procedures.md
index 045f4bce92..2048c49b07 100644
--- a/docs/docs/spark/procedures.md
+++ b/docs/docs/spark/procedures.md
@@ -69,6 +69,7 @@ Choose a group, then use the page contents to jump to a
procedure:
[`remove_orphan_files`](./procedures/maintenance#remove_orphan_files),
[`remove_orphan_blobs`](./procedures/maintenance#remove_orphan_blobs),
[`remove_unexisting_files`](./procedures/maintenance#remove_unexisting_files),
+[`remove_unexisting_manifests`](./procedures/maintenance#remove_unexisting_manifests),
[`purge_files`](./procedures/maintenance#purge_files),
[`repair`](./procedures/maintenance#repair),
[`repair_earliest_snapshot`](./procedures/maintenance#repair_earliest_snapshot)
diff --git a/docs/docs/spark/procedures/maintenance.md
b/docs/docs/spark/procedures/maintenance.md
index 1b2a77a507..366dd13e53 100644
--- a/docs/docs/spark/procedures/maintenance.md
+++ b/docs/docs/spark/procedures/maintenance.md
@@ -317,6 +317,22 @@ CALL sys.remove_unexisting_files(table => 'mydb.myt');
CALL sys.remove_unexisting_files(table => 'mydb.myt', dry_run => true);
```
+## remove_unexisting_manifests
+
+Remove missing manifest files from the latest snapshot's manifest list and
commit a replacement snapshot.
+
+This procedure may cause data loss when used outside of the documented repair
cases.
+
+**Arguments**
+
+- `table` (`STRING`, required): the target table identifier. To repair a
branch, backtick-quote the table name so `$` stays inside the identifier.
+
+```sql
+CALL sys.remove_unexisting_manifests(table => 'mydb.myt');
+
+CALL sys.remove_unexisting_manifests(table => 'mydb.`myt$branch_rt`');
+```
+
## purge_files
Clear table with purge files.
diff --git
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/RemoveUnexistingManifestsAction.java
b/paimon-core/src/main/java/org/apache/paimon/operation/RemoveUnexistingManifests.java
similarity index 61%
copy from
paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/RemoveUnexistingManifestsAction.java
copy to
paimon-core/src/main/java/org/apache/paimon/operation/RemoveUnexistingManifests.java
index e0fe8ac870..d2ed729581 100644
---
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/RemoveUnexistingManifestsAction.java
+++
b/paimon-core/src/main/java/org/apache/paimon/operation/RemoveUnexistingManifests.java
@@ -16,17 +16,16 @@
* limitations under the License.
*/
-package org.apache.paimon.flink.action;
+package org.apache.paimon.operation;
import org.apache.paimon.Snapshot;
-import org.apache.paimon.catalog.Identifier;
import org.apache.paimon.fs.FileIO;
import org.apache.paimon.fs.Path;
+import org.apache.paimon.manifest.FileEntry;
+import org.apache.paimon.manifest.FileKind;
import org.apache.paimon.manifest.ManifestEntry;
import org.apache.paimon.manifest.ManifestFileMeta;
import org.apache.paimon.manifest.ManifestList;
-import org.apache.paimon.operation.FileStoreCommitImpl;
-import org.apache.paimon.operation.ManifestsReader;
import org.apache.paimon.table.FileStoreTable;
import org.apache.paimon.table.source.ScanMode;
import org.apache.paimon.utils.FileStorePathFactory;
@@ -37,36 +36,37 @@ import org.slf4j.LoggerFactory;
import java.util.ArrayList;
import java.util.Collections;
+import java.util.HashSet;
import java.util.List;
-import java.util.Map;
+import java.util.Set;
import java.util.UUID;
-import static org.apache.paimon.manifest.ManifestEntry.recordCount;
-
-/** Action to remove the un-existing manifest file. */
-public class RemoveUnexistingManifestsAction extends ActionBase implements
LocalAction {
+/**
+ * Remove unexisting manifest files from the latest snapshot's manifest list.
+ *
+ * <p>Note that callers are on their own risk using this, which may cause data
loss when used
+ * outside the documented repair cases.
+ */
+public class RemoveUnexistingManifests {
- private static final Logger LOG =
- LoggerFactory.getLogger(RemoveUnexistingManifestsAction.class);
+ private static final Logger LOG =
LoggerFactory.getLogger(RemoveUnexistingManifests.class);
- private final String databaseName;
- private final String tableName;
+ private final FileStoreTable table;
- public RemoveUnexistingManifestsAction(
- String databaseName, String tableName, Map<String, String>
catalogConfig) {
- super(catalogConfig);
- this.databaseName = databaseName;
- this.tableName = tableName;
+ public RemoveUnexistingManifests(FileStoreTable table) {
+ this.table = table;
}
- @Override
- public void executeLocally() throws Exception {
- Identifier identifier = new Identifier(databaseName, tableName);
- FileStoreTable table = (FileStoreTable) catalog.getTable(identifier);
+ /**
+ * Drop missing manifest files from the latest snapshot and commit a
replacement snapshot.
+ *
+ * @return {@code true} if a repair snapshot was committed
+ */
+ public boolean execute() {
FileIO fileIO = table.fileIO();
Snapshot latest = table.snapshotManager().latestSnapshot();
if (latest == null) {
- return;
+ return false;
}
ManifestsReader manifestsReader =
table.store().newScan().manifestsReader();
@@ -82,22 +82,22 @@ public class RemoveUnexistingManifestsAction extends
ActionBase implements Local
Path path = pathFactory.toManifestFilePath(meta.fileName());
if (!fileIO.exists(path)) {
brokenManifestFile = true;
- LOG.warn("Drop manifest file: " + meta.fileName());
+ LOG.warn("Drop manifest file: {}", meta.fileName());
} else {
baseManifestEntries.addAll(table.store().newScan().readManifest(meta));
existingManifestFiles.add(meta);
}
} catch (Exception e) {
- throw new RuntimeException("Exception happens", e);
+ throw new RuntimeException("Failed to read manifest file " +
meta.fileName(), e);
}
}
if (!brokenManifestFile) {
- return;
+ return false;
}
ManifestList manifestList =
table.store().manifestListFactory().create();
- long totalRecordCount = recordCount(baseManifestEntries);
+ long totalRecordCount = visibleRecordCount(baseManifestEntries);
Pair<String, Long> baseManifestList =
manifestList.write(existingManifestFiles);
Pair<String, Long> deltaManifestList =
manifestList.write(Collections.emptyList());
@@ -112,5 +112,29 @@ public class RemoveUnexistingManifestsAction extends
ActionBase implements Local
"Failed, snapshot conflict, maybe multiple jobs is
running to commit snapshots.");
}
}
+ return true;
+ }
+
+ /**
+ * Rows a scan can still read from the remaining manifests.
+ *
+ * <p>A scan keeps an ADD entry only when no remaining manifest deletes
that file. Summing every
+ * entry's row count counts DELETE entries as extra rows. Subtracting
DELETE row counts
+ * under-counts when the matching ADD was in a manifest that has been
dropped.
+ */
+ private static long visibleRecordCount(List<ManifestEntry> entries) {
+ Set<FileEntry.Identifier> deleted = new HashSet<>();
+ for (ManifestEntry entry : entries) {
+ if (entry.kind() == FileKind.DELETE) {
+ deleted.add(entry.identifier());
+ }
+ }
+ long total = 0L;
+ for (ManifestEntry entry : entries) {
+ if (entry.kind() == FileKind.ADD &&
!deleted.contains(entry.identifier())) {
+ total += entry.file().rowCount();
+ }
+ }
+ return total;
}
}
diff --git
a/paimon-core/src/test/java/org/apache/paimon/operation/RemoveUnexistingManifestsTest.java
b/paimon-core/src/test/java/org/apache/paimon/operation/RemoveUnexistingManifestsTest.java
new file mode 100644
index 0000000000..abd8fec4cd
--- /dev/null
+++
b/paimon-core/src/test/java/org/apache/paimon/operation/RemoveUnexistingManifestsTest.java
@@ -0,0 +1,213 @@
+/*
+ * 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.operation;
+
+import org.apache.paimon.Snapshot;
+import org.apache.paimon.catalog.Catalog;
+import org.apache.paimon.catalog.CatalogContext;
+import org.apache.paimon.catalog.CatalogFactory;
+import org.apache.paimon.catalog.Identifier;
+import org.apache.paimon.data.BinaryRow;
+import org.apache.paimon.data.BinaryString;
+import org.apache.paimon.data.GenericRow;
+import org.apache.paimon.fs.Path;
+import org.apache.paimon.manifest.FileKind;
+import org.apache.paimon.manifest.ManifestEntry;
+import org.apache.paimon.manifest.ManifestFileMeta;
+import org.apache.paimon.reader.RecordReader;
+import org.apache.paimon.schema.Schema;
+import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.table.sink.BatchTableCommit;
+import org.apache.paimon.table.sink.BatchTableWrite;
+import org.apache.paimon.table.sink.BatchWriteBuilder;
+import org.apache.paimon.table.source.ReadBuilder;
+import org.apache.paimon.table.source.ScanMode;
+import org.apache.paimon.types.DataTypes;
+import org.apache.paimon.utils.TraceableFileIO;
+
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.function.Predicate;
+
+import static org.apache.paimon.manifest.ManifestEntry.recordCount;
+import static org.apache.paimon.manifest.ManifestEntry.recordCountAdd;
+import static org.apache.paimon.manifest.ManifestEntry.recordCountDelete;
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** Tests for {@link RemoveUnexistingManifests}. */
+public class RemoveUnexistingManifestsTest {
+
+ @TempDir java.nio.file.Path tempDir;
+
+ private Catalog catalog;
+ private FileStoreTable table;
+
+ @BeforeEach
+ public void beforeEach() throws Exception {
+ Path warehouse = new Path(TraceableFileIO.SCHEME + "://" +
tempDir.toString());
+ catalog =
CatalogFactory.createCatalog(CatalogContext.create(warehouse));
+ Identifier identifier = new Identifier("default", "T");
+ catalog.createDatabase(identifier.getDatabaseName(), true);
+ Schema schema =
+ Schema.newBuilder()
+ .column("k", DataTypes.INT())
+ .column("v", DataTypes.STRING())
+ .primaryKey("k")
+ .option("bucket", "1")
+ .option("manifest.target-file-size", "1 B")
+ .build();
+ catalog.createTable(identifier, schema, true);
+ table = (FileStoreTable) catalog.getTable(identifier);
+ }
+
+ @AfterEach
+ public void afterEach() throws IOException {
+ Predicate<Path> pathPredicate = path ->
path.toString().contains(tempDir.toString());
+ assertThat(TraceableFileIO.openInputStreams(pathPredicate)).isEmpty();
+ assertThat(TraceableFileIO.openOutputStreams(pathPredicate)).isEmpty();
+ }
+
+ @Test
+ public void testNoSnapshotIsNoOp() {
+ assertThat(new RemoveUnexistingManifests(table).execute()).isFalse();
+ assertThat(table.snapshotManager().latestSnapshot()).isNull();
+ }
+
+ @Test
+ public void testExistingManifestsAreNoOp() throws Exception {
+ commit(GenericRow.of(1, BinaryString.fromString("a")));
+ long snapshotId = table.snapshotManager().latestSnapshot().id();
+
+ assertThat(new RemoveUnexistingManifests(table).execute()).isFalse();
+
assertThat(table.snapshotManager().latestSnapshot().id()).isEqualTo(snapshotId);
+ assertThat(readCount()).isEqualTo(1);
+ }
+
+ @Test
+ public void
testTotalRecordCountMatchesVisibleRowsAfterDroppingDeletedAdd() throws
Exception {
+ commit(GenericRow.of(1, BinaryString.fromString("a")));
+ commit(GenericRow.of(2, BinaryString.fromString("b")));
+ compact();
+
+ List<ManifestFileMeta> manifests = manifests();
+ List<List<ManifestEntry>> entries = new ArrayList<>();
+ for (ManifestFileMeta meta : manifests) {
+ entries.add(table.store().newScan().readManifest(meta));
+ }
+ int dropIndex = indexOfAddDeletedElsewhere(entries);
+ assertThat(dropIndex).isGreaterThanOrEqualTo(0);
+
+ List<ManifestEntry> remaining = new ArrayList<>();
+ for (int i = 0; i < entries.size(); i++) {
+ if (i != dropIndex) {
+ remaining.addAll(entries.get(i));
+ }
+ }
+ long summedRowCount = recordCount(remaining);
+ long addedMinusDeleted = recordCountAdd(remaining) -
recordCountDelete(remaining);
+
+ Path path =
+
table.store().pathFactory().toManifestFilePath(manifests.get(dropIndex).fileName());
+ assertThat(table.fileIO().delete(path, false)).isTrue();
+
+ long snapshotId = table.snapshotManager().latestSnapshot().id();
+ assertThat(new RemoveUnexistingManifests(table).execute()).isTrue();
+
+ table = (FileStoreTable) catalog.getTable(new Identifier("default",
"T"));
+ Snapshot latest = table.snapshotManager().latestSnapshot();
+ long visibleRows = readCount();
+ assertThat(latest.id()).isEqualTo(snapshotId + 1);
+ assertThat(latest.totalRecordCount()).isEqualTo(visibleRows);
+ assertThat(summedRowCount).isNotEqualTo(visibleRows);
+ assertThat(addedMinusDeleted).isNotEqualTo(visibleRows);
+
assertThat(manifestNames()).doesNotContain(manifests.get(dropIndex).fileName());
+ }
+
+ private int indexOfAddDeletedElsewhere(List<List<ManifestEntry>> entries) {
+ for (int i = 0; i < entries.size(); i++) {
+ for (ManifestEntry entry : entries.get(i)) {
+ if (entry.kind() != FileKind.ADD) {
+ continue;
+ }
+ for (int j = 0; j < entries.size(); j++) {
+ if (i == j) {
+ continue;
+ }
+ for (ManifestEntry other : entries.get(j)) {
+ if (other.kind() == FileKind.DELETE
+ &&
other.identifier().equals(entry.identifier())) {
+ return i;
+ }
+ }
+ }
+ }
+ }
+ return -1;
+ }
+
+ private List<ManifestFileMeta> manifests() {
+ return table.store()
+ .newScan()
+ .manifestsReader()
+ .read(table.snapshotManager().latestSnapshot(), ScanMode.ALL)
+ .allManifests;
+ }
+
+ private List<String> manifestNames() {
+ List<String> names = new ArrayList<>();
+ for (ManifestFileMeta meta : manifests()) {
+ names.add(meta.fileName());
+ }
+ return names;
+ }
+
+ private long readCount() throws Exception {
+ ReadBuilder readBuilder = table.newReadBuilder();
+ try (RecordReader<?> reader =
+
readBuilder.newRead().createReader(readBuilder.newScan().plan())) {
+ long[] count = new long[1];
+ reader.forEachRemaining(row -> count[0]++);
+ return count[0];
+ }
+ }
+
+ private void commit(GenericRow row) throws Exception {
+ BatchWriteBuilder builder = table.newBatchWriteBuilder();
+ try (BatchTableWrite write = builder.newWrite();
+ BatchTableCommit commit = builder.newCommit()) {
+ write.write(row);
+ commit.commit(write.prepareCommit());
+ }
+ }
+
+ private void compact() throws Exception {
+ BatchWriteBuilder builder = table.newBatchWriteBuilder();
+ try (BatchTableWrite write = builder.newWrite();
+ BatchTableCommit commit = builder.newCommit()) {
+ write.compact(BinaryRow.EMPTY_ROW, 0, true);
+ commit.commit(write.prepareCommit());
+ }
+ }
+}
diff --git
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/RemoveUnexistingManifestsAction.java
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/RemoveUnexistingManifestsAction.java
index e0fe8ac870..fa075ffcab 100644
---
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/RemoveUnexistingManifestsAction.java
+++
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/RemoveUnexistingManifestsAction.java
@@ -18,37 +18,15 @@
package org.apache.paimon.flink.action;
-import org.apache.paimon.Snapshot;
import org.apache.paimon.catalog.Identifier;
-import org.apache.paimon.fs.FileIO;
-import org.apache.paimon.fs.Path;
-import org.apache.paimon.manifest.ManifestEntry;
-import org.apache.paimon.manifest.ManifestFileMeta;
-import org.apache.paimon.manifest.ManifestList;
-import org.apache.paimon.operation.FileStoreCommitImpl;
-import org.apache.paimon.operation.ManifestsReader;
+import org.apache.paimon.operation.RemoveUnexistingManifests;
import org.apache.paimon.table.FileStoreTable;
-import org.apache.paimon.table.source.ScanMode;
-import org.apache.paimon.utils.FileStorePathFactory;
-import org.apache.paimon.utils.Pair;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import java.util.ArrayList;
-import java.util.Collections;
-import java.util.List;
import java.util.Map;
-import java.util.UUID;
-
-import static org.apache.paimon.manifest.ManifestEntry.recordCount;
/** Action to remove the un-existing manifest file. */
public class RemoveUnexistingManifestsAction extends ActionBase implements
LocalAction {
- private static final Logger LOG =
- LoggerFactory.getLogger(RemoveUnexistingManifestsAction.class);
-
private final String databaseName;
private final String tableName;
@@ -63,54 +41,6 @@ public class RemoveUnexistingManifestsAction extends
ActionBase implements Local
public void executeLocally() throws Exception {
Identifier identifier = new Identifier(databaseName, tableName);
FileStoreTable table = (FileStoreTable) catalog.getTable(identifier);
- FileIO fileIO = table.fileIO();
- Snapshot latest = table.snapshotManager().latestSnapshot();
- if (latest == null) {
- return;
- }
-
- ManifestsReader manifestsReader =
table.store().newScan().manifestsReader();
- ManifestsReader.Result manifestsResult = manifestsReader.read(latest,
ScanMode.ALL);
- List<ManifestFileMeta> manifests = manifestsResult.allManifests;
- List<ManifestFileMeta> existingManifestFiles = new ArrayList<>();
- List<ManifestEntry> baseManifestEntries = new ArrayList<>();
-
- FileStorePathFactory pathFactory = table.store().pathFactory();
- boolean brokenManifestFile = false;
- for (ManifestFileMeta meta : manifests) {
- try {
- Path path = pathFactory.toManifestFilePath(meta.fileName());
- if (!fileIO.exists(path)) {
- brokenManifestFile = true;
- LOG.warn("Drop manifest file: " + meta.fileName());
- } else {
-
baseManifestEntries.addAll(table.store().newScan().readManifest(meta));
- existingManifestFiles.add(meta);
- }
- } catch (Exception e) {
- throw new RuntimeException("Exception happens", e);
- }
- }
-
- if (!brokenManifestFile) {
- return;
- }
-
- ManifestList manifestList =
table.store().manifestListFactory().create();
- long totalRecordCount = recordCount(baseManifestEntries);
- Pair<String, Long> baseManifestList =
manifestList.write(existingManifestFiles);
- Pair<String, Long> deltaManifestList =
manifestList.write(Collections.emptyList());
-
- try (FileStoreCommitImpl fileStoreCommit =
- (FileStoreCommitImpl)
- table.store().newCommit("Repair-table-" +
UUID.randomUUID(), table)) {
- boolean result =
- fileStoreCommit.replaceManifestList(
- latest, totalRecordCount, baseManifestList,
deltaManifestList);
- if (!result) {
- throw new RuntimeException(
- "Failed, snapshot conflict, maybe multiple jobs is
running to commit snapshots.");
- }
- }
+ new RemoveUnexistingManifests(table).execute();
}
}
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 672099d293..f534773959 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
@@ -56,6 +56,7 @@ import
org.apache.paimon.spark.procedure.ReassignRowIdProcedure;
import org.apache.paimon.spark.procedure.RemoveOrphanBlobsProcedure;
import org.apache.paimon.spark.procedure.RemoveOrphanFilesProcedure;
import org.apache.paimon.spark.procedure.RemoveUnexistingFilesProcedure;
+import org.apache.paimon.spark.procedure.RemoveUnexistingManifestsProcedure;
import org.apache.paimon.spark.procedure.RenameBranchProcedure;
import org.apache.paimon.spark.procedure.RenameTagProcedure;
import org.apache.paimon.spark.procedure.RepairEarliestSnapshotProcedure;
@@ -123,6 +124,8 @@ public class SparkProcedures {
procedureBuilders.put("remove_orphan_files",
RemoveOrphanFilesProcedure::builder);
procedureBuilders.put("remove_orphan_blobs",
RemoveOrphanBlobsProcedure::builder);
procedureBuilders.put("remove_unexisting_files",
RemoveUnexistingFilesProcedure::builder);
+ procedureBuilders.put(
+ "remove_unexisting_manifests",
RemoveUnexistingManifestsProcedure::builder);
procedureBuilders.put("expire_snapshots",
ExpireSnapshotsProcedure::builder);
procedureBuilders.put("expire_partitions",
ExpirePartitionsProcedure::builder);
procedureBuilders.put("repair", RepairProcedure::builder);
diff --git
a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/RemoveUnexistingManifestsProcedure.java
b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/RemoveUnexistingManifestsProcedure.java
new file mode 100644
index 0000000000..92ecd99c3d
--- /dev/null
+++
b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/RemoveUnexistingManifestsProcedure.java
@@ -0,0 +1,102 @@
+/*
+ * 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.operation.RemoveUnexistingManifests;
+import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.utils.Preconditions;
+
+import org.apache.spark.sql.catalyst.InternalRow;
+import org.apache.spark.sql.connector.catalog.Identifier;
+import org.apache.spark.sql.connector.catalog.TableCatalog;
+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 org.apache.spark.unsafe.types.UTF8String;
+
+/**
+ * Procedure to remove unexisting manifest files from the manifest list. See
{@link
+ * RemoveUnexistingManifests} for detailed use cases.
+ *
+ * <pre><code>
+ * -- remove unexisting manifest files in table `mydb.myt`
+ * CALL sys.remove_unexisting_manifests(table => 'mydb.myt')
+ *
+ * -- remove unexisting manifest files in a branch
+ * CALL sys.remove_unexisting_manifests(table => 'mydb.`myt$branch_rt`')
+ * </code></pre>
+ *
+ * <p>Note that the user is on their own risk using this procedure, which may
cause data loss when
+ * used outside of the documented repair cases.
+ */
+public class RemoveUnexistingManifestsProcedure extends BaseProcedure {
+
+ private static final ProcedureParameter[] PARAMETERS =
+ new ProcedureParameter[] {ProcedureParameter.required("table",
DataTypes.StringType)};
+
+ private static final StructType OUTPUT_TYPE =
+ new StructType(
+ new StructField[] {
+ new StructField("result", DataTypes.StringType, false,
Metadata.empty())
+ });
+
+ private RemoveUnexistingManifestsProcedure(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());
+ return modifyPaimonTable(
+ tableIdent,
+ table -> {
+ Preconditions.checkArgument(
+ table instanceof FileStoreTable,
+ "%s is not a file store table",
+ tableIdent);
+ new RemoveUnexistingManifests((FileStoreTable)
table).execute();
+ return new InternalRow[]
{newInternalRow(UTF8String.fromString("Success"))};
+ });
+ }
+
+ public static ProcedureBuilder builder() {
+ return new BaseProcedure.Builder<RemoveUnexistingManifestsProcedure>()
{
+ @Override
+ public RemoveUnexistingManifestsProcedure doBuild() {
+ return new RemoveUnexistingManifestsProcedure(tableCatalog());
+ }
+ };
+ }
+
+ @Override
+ public String description() {
+ return "RemoveUnexistingManifestsProcedure";
+ }
+}
diff --git
a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/procedure/RemoveUnexistingManifestsProcedureTest.scala
b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/procedure/RemoveUnexistingManifestsProcedureTest.scala
new file mode 100644
index 0000000000..055dbbad85
--- /dev/null
+++
b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/procedure/RemoveUnexistingManifestsProcedureTest.scala
@@ -0,0 +1,159 @@
+/*
+ * 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.manifest.ManifestFileMeta
+import org.apache.paimon.spark.PaimonSparkTestBase
+import org.apache.paimon.table.FileStoreTable
+import org.apache.paimon.table.source.ScanMode
+
+import org.apache.spark.sql.Row
+
+import scala.collection.JavaConverters._
+
+class RemoveUnexistingManifestsProcedureTest extends PaimonSparkTestBase {
+
+ private val tableProperties =
+ """
+ |TBLPROPERTIES (
+ | 'primary-key' = 'k',
+ | 'bucket' = '1',
+ | 'write-only' = 'true',
+ | 'manifest.target-file-size' = '10 B')
+ |""".stripMargin
+
+ test("Paimon Procedure: remove unexisting manifests") {
+ spark.sql(s"""
+ |CREATE TABLE T (k BIGINT, v STRING)
+ |$tableProperties
+ |""".stripMargin)
+
+ spark.sql("INSERT INTO T VALUES (1, 'Hi')")
+ spark.sql("INSERT INTO T VALUES (2, 'Hello')")
+ spark.sql("INSERT INTO T VALUES (3, 'Paimon')")
+
+ val originals =
+ Set(Row(1L, "Hi"), Row(2L, "Hello"), Row(3L, "Paimon"))
+ checkAnswer(spark.sql("SELECT * FROM T ORDER BY k"), originals.toList)
+
+ val (table, manifests) = loadManifests("T")
+ assert(manifests.size() >= 2)
+ val dropped = manifests.get(1)
+ val droppedRows = rowCount(table, dropped)
+ assert(droppedRows > 0 && droppedRows < originals.size)
+
+ val beforeId = table.snapshotManager().latestSnapshot().id()
+
assert(table.fileIO.delete(table.store.pathFactory.toManifestFilePath(dropped.fileName),
false))
+
+ checkAnswer(
+ spark.sql("CALL sys.remove_unexisting_manifests(table => 'test.T')"),
+ Row("Success") :: Nil)
+
+ val remaining = spark.sql("SELECT * FROM T ORDER BY k").collect()
+ assert(remaining.length == originals.size - droppedRows)
+ assert(remaining.forall(originals.contains))
+
+ val repaired = loadTable("T")
+ val latest = repaired.snapshotManager().latestSnapshot()
+ assert(latest.id() == beforeId + 1)
+ assert(latest.totalRecordCount() == remaining.length.toLong)
+ assert(!manifestNames(repaired).contains(dropped.fileName))
+ }
+
+ test("Paimon Procedure: remove unexisting manifests is a no-op when files
exist") {
+ spark.sql("CREATE TABLE T (k INT, v STRING)")
+ spark.sql("INSERT INTO T VALUES (1, 'a')")
+ val beforeId = loadTable("T").snapshotManager().latestSnapshot().id()
+
+ checkAnswer(
+ spark.sql("CALL sys.remove_unexisting_manifests(table => 'test.T')"),
+ Row("Success") :: Nil)
+ checkAnswer(spark.sql("SELECT * FROM T"), Row(1, "a") :: Nil)
+ assert(loadTable("T").snapshotManager().latestSnapshot().id() == beforeId)
+ }
+
+ test("Paimon Procedure: remove unexisting manifests on a branch") {
+ spark.sql(s"""
+ |CREATE TABLE T (k BIGINT, v STRING)
+ |$tableProperties
+ |""".stripMargin)
+ spark.sql("INSERT INTO T VALUES (1, 'Hi')")
+ spark.sql("INSERT INTO T VALUES (2, 'Hello')")
+ spark.sql("INSERT INTO T VALUES (3, 'Paimon')")
+ spark.sql("CALL sys.create_tag(table => 'test.T', tag => 'base')")
+ spark.sql("CALL sys.create_branch(table => 'test.T', branch => 'rt', tag
=> 'base')")
+ spark.sql("INSERT INTO `T$branch_rt` VALUES (4, 'Hi 4')")
+ spark.sql("INSERT INTO `T$branch_rt` VALUES (5, 'Hello 5')")
+ spark.sql("INSERT INTO `T$branch_rt` VALUES (6, 'Paimon 6')")
+
+ val branchRowsBefore = spark.sql("SELECT * FROM `T$branch_rt`").collect()
+ val mainNames = manifestNames(loadTable("T")).toSet
+ val (branch, branchManifests) = loadManifests("T$branch_rt")
+ val dropped = branchManifests.asScala.find(meta =>
!mainNames.contains(meta.fileName()))
+ assert(dropped.isDefined)
+ val droppedMeta = dropped.get
+ val droppedRows = rowCount(branch, droppedMeta)
+ assert(droppedRows > 0)
+
+ assert(
+ branch.fileIO
+
.delete(branch.store.pathFactory.toManifestFilePath(droppedMeta.fileName),
false))
+
+ checkAnswer(
+ spark.sql("CALL sys.remove_unexisting_manifests(table =>
'test.`T$branch_rt`')"),
+ Row("Success") :: Nil)
+
+ val branchRows = spark.sql("SELECT * FROM `T$branch_rt` ORDER BY
k").collect()
+ assert(branchRows.length == branchRowsBefore.length - droppedRows)
+ val repaired = loadTable("T$branch_rt")
+ assert(
+ repaired.snapshotManager().latestSnapshot().totalRecordCount() ==
branchRows.length.toLong)
+ assert(!manifestNames(repaired).contains(droppedMeta.fileName))
+
+ checkAnswer(
+ spark.sql("SELECT * FROM T ORDER BY k"),
+ Row(1L, "Hi") :: Row(2L, "Hello") :: Row(3L, "Paimon") :: Nil)
+ }
+
+ private def loadManifests(
+ tableName: String): (FileStoreTable, java.util.List[ManifestFileMeta]) =
{
+ val table = loadTable(tableName)
+ val manifests = table.store
+ .newScan()
+ .manifestsReader()
+ .read(table.snapshotManager.latestSnapshot, ScanMode.ALL)
+ .allManifests
+ (table, manifests)
+ }
+
+ private def manifestNames(table: FileStoreTable): Seq[String] = {
+ table.store
+ .newScan()
+ .manifestsReader()
+ .read(table.snapshotManager.latestSnapshot, ScanMode.ALL)
+ .allManifests
+ .asScala
+ .map(_.fileName())
+ .toSeq
+ }
+
+ private def rowCount(table: FileStoreTable, manifest: ManifestFileMeta):
Long = {
+
table.store.newScan().readManifest(manifest).asScala.map(_.file().rowCount()).sum
+ }
+}
