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 e28adfa817 [core][flink][spark] Support repairing the earliest
snapshot hint (#8883)
e28adfa817 is described below
commit e28adfa8178fa8197082dc5a2cb10f2e0a2061bc
Author: jerry <[email protected]>
AuthorDate: Thu Jul 30 19:21:16 2026 +0800
[core][flink][spark] Support repairing the earliest snapshot hint (#8883)
---
.../org/apache/paimon/utils/SnapshotManager.java | 50 +++++++++
.../apache/paimon/utils/SnapshotManagerTest.java | 122 +++++++++++++++++++++
.../ProcedurePositionalArgumentsITCase.java | 13 +++
.../procedure/RepairEarliestSnapshotProcedure.java | 54 +++++++++
.../services/org.apache.paimon.factories.Factory | 1 +
.../RepairEarliestSnapshotProcedureITCase.java | 53 +++++++++
.../org/apache/paimon/spark/SparkProcedures.java | 2 +
.../procedure/RepairEarliestSnapshotProcedure.java | 92 ++++++++++++++++
.../RepairEarliestSnapshotProcedureTest.scala | 42 +++++++
9 files changed, 429 insertions(+)
diff --git
a/paimon-core/src/main/java/org/apache/paimon/utils/SnapshotManager.java
b/paimon-core/src/main/java/org/apache/paimon/utils/SnapshotManager.java
index af21181214..984a98902e 100644
--- a/paimon-core/src/main/java/org/apache/paimon/utils/SnapshotManager.java
+++ b/paimon-core/src/main/java/org/apache/paimon/utils/SnapshotManager.java
@@ -272,6 +272,56 @@ public class SnapshotManager implements Serializable {
}
}
+ /**
+ * Repairs the earliest snapshot hint to the start of a continuous suffix
ending at the latest
+ * snapshot and returns the previous earliest snapshot id.
+ */
+ public long repairEarliestSnapshot(long snapshotId) {
+ long previous =
+ Preconditions.checkNotNull(
+ earliestSnapshotId(),
+ "Cannot repair earliest snapshot for an empty table.");
+ long latest =
+ Preconditions.checkNotNull(
+ latestSnapshotId(), "Cannot repair earliest snapshot
for an empty table.");
+ Preconditions.checkArgument(
+ snapshotId >= previous,
+ "Snapshot %s must not be earlier than current earliest
snapshot %s.",
+ snapshotId,
+ previous);
+ Preconditions.checkArgument(
+ snapshotId <= latest,
+ "Snapshot %s must not be later than latest snapshot %s.",
+ snapshotId,
+ latest);
+ Set<Long> snapshotIds;
+ try {
+ snapshotIds =
+ snapshotIdStream()
+ .filter(id -> id >= snapshotId && id <= latest)
+ .collect(Collectors.toSet());
+ } catch (IOException e) {
+ throw new UncheckedIOException(e);
+ }
+ for (long id = snapshotId; ; id++) {
+ Preconditions.checkArgument(
+ snapshotIds.contains(id), "Snapshot %s does not exist.",
id);
+ if (id == latest) {
+ break;
+ }
+ }
+ Preconditions.checkArgument(
+ snapshotId == previous || !snapshotExists(snapshotId - 1),
+ "Snapshot %s does not immediately follow a snapshot gap.",
+ snapshotId);
+ try {
+ commitEarliestHint(snapshotId);
+ } catch (IOException e) {
+ throw new UncheckedIOException(e);
+ }
+ return previous;
+ }
+
public @Nullable Long pickOrLatest(Predicate<Snapshot> predicate) {
Long latestId = latestSnapshotId();
Long earliestId = earliestSnapshotId();
diff --git
a/paimon-core/src/test/java/org/apache/paimon/utils/SnapshotManagerTest.java
b/paimon-core/src/test/java/org/apache/paimon/utils/SnapshotManagerTest.java
index 02b21ba906..e44af2cd90 100644
--- a/paimon-core/src/test/java/org/apache/paimon/utils/SnapshotManagerTest.java
+++ b/paimon-core/src/test/java/org/apache/paimon/utils/SnapshotManagerTest.java
@@ -30,6 +30,7 @@ import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.io.TempDir;
import org.junit.jupiter.params.ParameterizedTest;
import org.junit.jupiter.params.provider.ValueSource;
+import org.mockito.Mockito;
import javax.annotation.Nullable;
@@ -93,6 +94,127 @@ public class SnapshotManagerTest {
assertThat(snapshotManager.earliestSnapshot().id()).isEqualTo(isRaceCondition ?
1 : 0);
}
+ @Test
+ public void testRepairEarliestSnapshot() throws IOException {
+ FileIO fileIO = LocalFileIO.create();
+ SnapshotManager snapshotManager = newSnapshotManager(fileIO, new
Path(tempDir.toString()));
+ fileIO.tryToWriteAtomic(
+ snapshotManager.snapshotPath(1), createSnapshotWithMillis(1,
1000).toJson());
+ fileIO.tryToWriteAtomic(
+ snapshotManager.snapshotPath(5), createSnapshotWithMillis(5,
5000).toJson());
+ snapshotManager.commitEarliestHint(1);
+ snapshotManager.commitLatestHint(5);
+
+ assertThat(snapshotManager.repairEarliestSnapshot(5)).isEqualTo(1);
+ assertThat(snapshotManager.earliestSnapshotId()).isEqualTo(5);
+
+ assertThat(snapshotManager.repairEarliestSnapshot(5)).isEqualTo(5);
+ assertThat(snapshotManager.earliestSnapshotId()).isEqualTo(5);
+ }
+
+ @Test
+ public void testRepairEarliestSnapshotRejectsBackwardTarget() throws
IOException {
+ FileIO fileIO = LocalFileIO.create();
+ SnapshotManager snapshotManager = newSnapshotManager(fileIO, new
Path(tempDir.toString()));
+ fileIO.tryToWriteAtomic(
+ snapshotManager.snapshotPath(1), createSnapshotWithMillis(1,
1000).toJson());
+ fileIO.tryToWriteAtomic(
+ snapshotManager.snapshotPath(5), createSnapshotWithMillis(5,
5000).toJson());
+ snapshotManager.commitEarliestHint(5);
+ snapshotManager.commitLatestHint(5);
+
+ assertThatThrownBy(() -> snapshotManager.repairEarliestSnapshot(1))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("must not be earlier than current
earliest snapshot");
+ }
+
+ @Test
+ public void testRepairEarliestSnapshotRejectsMissingTarget() throws
IOException {
+ FileIO fileIO = LocalFileIO.create();
+ SnapshotManager snapshotManager = newSnapshotManager(fileIO, new
Path(tempDir.toString()));
+ fileIO.tryToWriteAtomic(
+ snapshotManager.snapshotPath(1), createSnapshotWithMillis(1,
1000).toJson());
+ fileIO.tryToWriteAtomic(
+ snapshotManager.snapshotPath(5), createSnapshotWithMillis(5,
5000).toJson());
+ snapshotManager.commitEarliestHint(1);
+ snapshotManager.commitLatestHint(5);
+
+ assertThatThrownBy(() -> snapshotManager.repairEarliestSnapshot(4))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("Snapshot 4 does not exist");
+ }
+
+ @Test
+ public void testRepairEarliestSnapshotRejectsNonContinuousSuffix() throws
IOException {
+ FileIO fileIO = LocalFileIO.create();
+ SnapshotManager snapshotManager = newSnapshotManager(fileIO, new
Path(tempDir.toString()));
+ for (long snapshotId : new long[] {1, 3, 5, 6}) {
+ fileIO.tryToWriteAtomic(
+ snapshotManager.snapshotPath(snapshotId),
+ createSnapshotWithMillis(snapshotId, snapshotId *
1000).toJson());
+ }
+ snapshotManager.commitEarliestHint(1);
+ snapshotManager.commitLatestHint(6);
+
+ assertThatThrownBy(() -> snapshotManager.repairEarliestSnapshot(3))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("Snapshot 4 does not exist");
+ }
+
+ @Test
+ public void testRepairEarliestSnapshotListsSnapshots() throws IOException {
+ FileIO fileIO = Mockito.spy(LocalFileIO.create());
+ SnapshotManager snapshotManager = newSnapshotManager(fileIO, new
Path(tempDir.toString()));
+ for (long snapshotId : new long[] {1, 3, 4, 5, 6}) {
+ fileIO.tryToWriteAtomic(
+ snapshotManager.snapshotPath(snapshotId),
+ createSnapshotWithMillis(snapshotId, snapshotId *
1000).toJson());
+ }
+ snapshotManager.commitEarliestHint(1);
+ snapshotManager.commitLatestHint(6);
+ Mockito.clearInvocations(fileIO);
+
+ assertThat(snapshotManager.repairEarliestSnapshot(3)).isEqualTo(1);
+ Mockito.verify(fileIO).listStatus(snapshotManager.snapshotDirectory());
+ Mockito.verify(fileIO,
Mockito.never()).exists(snapshotManager.snapshotPath(4));
+ }
+
+ @Test
+ public void testRepairEarliestSnapshotRejectsTargetAfterLatest() throws
IOException {
+ FileIO fileIO = LocalFileIO.create();
+ SnapshotManager snapshotManager = newSnapshotManager(fileIO, new
Path(tempDir.toString()));
+ fileIO.tryToWriteAtomic(
+ snapshotManager.snapshotPath(1), createSnapshotWithMillis(1,
1000).toJson());
+ fileIO.tryToWriteAtomic(
+ snapshotManager.snapshotPath(5), createSnapshotWithMillis(5,
5000).toJson());
+ fileIO.tryToWriteAtomic(
+ snapshotManager.snapshotPath(10), createSnapshotWithMillis(10,
10000).toJson());
+ snapshotManager.commitEarliestHint(1);
+ snapshotManager.commitLatestHint(5);
+
+ assertThatThrownBy(() -> snapshotManager.repairEarliestSnapshot(10))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("must not be later than latest
snapshot");
+ }
+
+ @Test
+ public void testRepairEarliestSnapshotRejectsContinuousTarget() throws
IOException {
+ FileIO fileIO = LocalFileIO.create();
+ SnapshotManager snapshotManager = newSnapshotManager(fileIO, new
Path(tempDir.toString()));
+ fileIO.tryToWriteAtomic(
+ snapshotManager.snapshotPath(1), createSnapshotWithMillis(1,
1000).toJson());
+ fileIO.tryToWriteAtomic(
+ snapshotManager.snapshotPath(4), createSnapshotWithMillis(4,
4000).toJson());
+ fileIO.tryToWriteAtomic(
+ snapshotManager.snapshotPath(5), createSnapshotWithMillis(5,
5000).toJson());
+ snapshotManager.commitEarliestHint(1);
+ snapshotManager.commitLatestHint(5);
+
+ assertThatThrownBy(() -> snapshotManager.repairEarliestSnapshot(5))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("does not immediately follow a snapshot
gap");
+ }
+
@Test
public void testEarliestSnapshotThrowsWhenRetryExhausted() throws
IOException {
FileIO localFileIO = LocalFileIO.create();
diff --git
a/paimon-flink/paimon-flink-1.18/src/test/java/org/apache/paimon/flink/procedure/ProcedurePositionalArgumentsITCase.java
b/paimon-flink/paimon-flink-1.18/src/test/java/org/apache/paimon/flink/procedure/ProcedurePositionalArgumentsITCase.java
index 5dac8cd200..b5350f495b 100644
---
a/paimon-flink/paimon-flink-1.18/src/test/java/org/apache/paimon/flink/procedure/ProcedurePositionalArgumentsITCase.java
+++
b/paimon-flink/paimon-flink-1.18/src/test/java/org/apache/paimon/flink/procedure/ProcedurePositionalArgumentsITCase.java
@@ -482,6 +482,19 @@ public class ProcedurePositionalArgumentsITCase extends
CatalogITCaseBase {
.doesNotThrowAnyException();
}
+ @Test
+ public void testRepairEarliestSnapshot() throws Exception {
+ sql("CREATE TABLE T (k INT)");
+ FileStoreTable table = paimonTable("T");
+
table.fileIO().tryToWriteAtomic(table.snapshotManager().snapshotPath(1), "");
+
table.fileIO().tryToWriteAtomic(table.snapshotManager().snapshotPath(5), "");
+ table.snapshotManager().commitEarliestHint(1);
+ table.snapshotManager().commitLatestHint(5);
+
+ assertThat(sql("CALL sys.repair_earliest_snapshot('default.T', 5)"))
+ .containsExactly(Row.of(1L, 5L));
+ }
+
@Test
public void testRewriteFileIndex() {
sql(
diff --git
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/procedure/RepairEarliestSnapshotProcedure.java
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/procedure/RepairEarliestSnapshotProcedure.java
new file mode 100644
index 0000000000..5ce16fa5b8
--- /dev/null
+++
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/procedure/RepairEarliestSnapshotProcedure.java
@@ -0,0 +1,54 @@
+/*
+ * 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.flink.procedure;
+
+import org.apache.paimon.catalog.Catalog;
+import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.utils.SnapshotManager;
+
+import org.apache.flink.table.annotation.ArgumentHint;
+import org.apache.flink.table.annotation.DataTypeHint;
+import org.apache.flink.table.annotation.ProcedureHint;
+import org.apache.flink.table.procedure.ProcedureContext;
+import org.apache.flink.types.Row;
+
+/** A procedure to repair the earliest snapshot. */
+public class RepairEarliestSnapshotProcedure extends ProcedureBase {
+
+ public static final String IDENTIFIER = "repair_earliest_snapshot";
+
+ @ProcedureHint(
+ argument = {
+ @ArgumentHint(name = "table", type = @DataTypeHint("STRING")),
+ @ArgumentHint(name = "snapshot_id", type =
@DataTypeHint("BIGINT"))
+ })
+ public @DataTypeHint(
+ "ROW<previous_earliest_snapshot_id BIGINT,
current_earliest_snapshot_id BIGINT>") Row[]
+ call(ProcedureContext procedureContext, String tableId, Long
snapshotId)
+ throws Catalog.TableNotExistException {
+ SnapshotManager snapshotManager = ((FileStoreTable)
table(tableId)).snapshotManager();
+ long previous = snapshotManager.repairEarliestSnapshot(snapshotId);
+ return new Row[] {Row.of(previous, snapshotId)};
+ }
+
+ @Override
+ public String identifier() {
+ return IDENTIFIER;
+ }
+}
diff --git
a/paimon-flink/paimon-flink-common/src/main/resources/META-INF/services/org.apache.paimon.factories.Factory
b/paimon-flink/paimon-flink-common/src/main/resources/META-INF/services/org.apache.paimon.factories.Factory
index 7cf7aefbf3..e0eee1005a 100644
---
a/paimon-flink/paimon-flink-common/src/main/resources/META-INF/services/org.apache.paimon.factories.Factory
+++
b/paimon-flink/paimon-flink-common/src/main/resources/META-INF/services/org.apache.paimon.factories.Factory
@@ -82,6 +82,7 @@ org.apache.paimon.flink.procedure.ExpireSnapshotsProcedure
org.apache.paimon.flink.procedure.ExpireChangelogsProcedure
org.apache.paimon.flink.procedure.ExpirePartitionsProcedure
org.apache.paimon.flink.procedure.PurgeFilesProcedure
+org.apache.paimon.flink.procedure.RepairEarliestSnapshotProcedure
org.apache.paimon.flink.procedure.privilege.InitFileBasedPrivilegeProcedure
org.apache.paimon.flink.procedure.privilege.CreatePrivilegedUserProcedure
org.apache.paimon.flink.procedure.privilege.DropPrivilegedUserProcedure
diff --git
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/procedure/RepairEarliestSnapshotProcedureITCase.java
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/procedure/RepairEarliestSnapshotProcedureITCase.java
new file mode 100644
index 0000000000..ded069c335
--- /dev/null
+++
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/procedure/RepairEarliestSnapshotProcedureITCase.java
@@ -0,0 +1,53 @@
+/*
+ * 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.flink.procedure;
+
+import org.apache.paimon.flink.CatalogITCaseBase;
+import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.utils.SnapshotManager;
+
+import org.apache.flink.types.Row;
+import org.junit.jupiter.api.Test;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** IT cases for {@link RepairEarliestSnapshotProcedure}. */
+public class RepairEarliestSnapshotProcedureITCase extends CatalogITCaseBase {
+
+ @Test
+ public void testRepairEarliestSnapshot() throws Exception {
+ sql("CREATE TABLE T (k INT)");
+ for (int i = 1; i <= 5; i++) {
+ sql("INSERT INTO T VALUES (" + i + ")");
+ }
+
+ FileStoreTable table = paimonTable("T");
+ SnapshotManager snapshotManager = table.snapshotManager();
+ snapshotManager.deleteSnapshot(2);
+ snapshotManager.deleteSnapshot(3);
+ snapshotManager.deleteSnapshot(4);
+
+ assertThat(
+ sql(
+ "CALL sys.repair_earliest_snapshot("
+ + "`table` => 'default.T', snapshot_id
=> 5)"))
+ .containsExactly(Row.of(1L, 5L));
+ assertThat(snapshotManager.earliestSnapshotId()).isEqualTo(5);
+ }
+}
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 6a2203b9cb..8fb1377323 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
@@ -50,6 +50,7 @@ import
org.apache.paimon.spark.procedure.RemoveOrphanFilesProcedure;
import org.apache.paimon.spark.procedure.RemoveUnexistingFilesProcedure;
import org.apache.paimon.spark.procedure.RenameBranchProcedure;
import org.apache.paimon.spark.procedure.RenameTagProcedure;
+import org.apache.paimon.spark.procedure.RepairEarliestSnapshotProcedure;
import org.apache.paimon.spark.procedure.RepairProcedure;
import org.apache.paimon.spark.procedure.ReplaceTagProcedure;
import org.apache.paimon.spark.procedure.RescaleProcedure;
@@ -113,6 +114,7 @@ public class SparkProcedures {
procedureBuilders.put("expire_snapshots",
ExpireSnapshotsProcedure::builder);
procedureBuilders.put("expire_partitions",
ExpirePartitionsProcedure::builder);
procedureBuilders.put("repair", RepairProcedure::builder);
+ procedureBuilders.put("repair_earliest_snapshot",
RepairEarliestSnapshotProcedure::builder);
procedureBuilders.put("fast_forward", FastForwardProcedure::builder);
procedureBuilders.put("merge_branch", MergeBranchProcedure::builder);
procedureBuilders.put("reset_consumer",
ResetConsumerProcedure::builder);
diff --git
a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/RepairEarliestSnapshotProcedure.java
b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/RepairEarliestSnapshotProcedure.java
new file mode 100644
index 0000000000..6ce3cb33dc
--- /dev/null
+++
b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/RepairEarliestSnapshotProcedure.java
@@ -0,0 +1,92 @@
+/*
+ * 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.table.FileStoreTable;
+import org.apache.paimon.utils.SnapshotManager;
+
+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.Metadata;
+import org.apache.spark.sql.types.StructField;
+import org.apache.spark.sql.types.StructType;
+
+import static org.apache.spark.sql.types.DataTypes.LongType;
+import static org.apache.spark.sql.types.DataTypes.StringType;
+
+/** A procedure to repair the earliest snapshot. */
+public class RepairEarliestSnapshotProcedure extends BaseProcedure {
+
+ private static final ProcedureParameter[] PARAMETERS =
+ new ProcedureParameter[] {
+ ProcedureParameter.required("table", StringType),
+ ProcedureParameter.required("snapshot_id", LongType)
+ };
+
+ private static final StructType OUTPUT_TYPE =
+ new StructType(
+ new StructField[] {
+ new StructField(
+ "previous_earliest_snapshot_id", LongType,
false, Metadata.empty()),
+ new StructField(
+ "current_earliest_snapshot_id", LongType,
false, Metadata.empty())
+ });
+
+ private RepairEarliestSnapshotProcedure(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());
+ long snapshotId = args.getLong(1);
+ return modifyPaimonTable(
+ tableIdent,
+ table -> {
+ SnapshotManager snapshotManager = ((FileStoreTable)
table).snapshotManager();
+ long previous =
snapshotManager.repairEarliestSnapshot(snapshotId);
+ return new InternalRow[] {newInternalRow(previous,
snapshotId)};
+ });
+ }
+
+ public static ProcedureBuilder builder() {
+ return new BaseProcedure.Builder<RepairEarliestSnapshotProcedure>() {
+ @Override
+ public RepairEarliestSnapshotProcedure doBuild() {
+ return new RepairEarliestSnapshotProcedure(tableCatalog());
+ }
+ };
+ }
+
+ @Override
+ public String description() {
+ return "RepairEarliestSnapshotProcedure";
+ }
+}
diff --git
a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/procedure/RepairEarliestSnapshotProcedureTest.scala
b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/procedure/RepairEarliestSnapshotProcedureTest.scala
new file mode 100644
index 0000000000..1895eb5f05
--- /dev/null
+++
b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/procedure/RepairEarliestSnapshotProcedureTest.scala
@@ -0,0 +1,42 @@
+/*
+ * 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.Row
+import org.assertj.core.api.Assertions.assertThat
+
+class RepairEarliestSnapshotProcedureTest extends PaimonSparkTestBase {
+
+ test("Paimon procedure: repair earliest snapshot") {
+ spark.sql("CREATE TABLE T (k INT) USING PAIMON")
+
+ val snapshotManager = loadTable("T").snapshotManager()
+ snapshotManager.fileIO.tryToWriteAtomic(snapshotManager.snapshotPath(1),
"")
+ snapshotManager.fileIO.tryToWriteAtomic(snapshotManager.snapshotPath(5),
"")
+ snapshotManager.commitEarliestHint(1)
+ snapshotManager.commitLatestHint(5)
+
+ checkAnswer(
+ spark.sql("CALL paimon.sys.repair_earliest_snapshot(table => 'test.T',
snapshot_id => 5)"),
+ Row(1L, 5L) :: Nil)
+ assertThat(snapshotManager.earliestSnapshotId).isEqualTo(5)
+ }
+}
