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 ef0b12e16a [spark] Support expire_changelogs procedure (#10062)
ef0b12e16a is described below
commit ef0b12e16ae71b8fed55a2c3c9ae10b5b436eb4b
Author: Wenchao Wu <[email protected]>
AuthorDate: Sat Oct 3 17:20:20 2026 +0800
[spark] Support expire_changelogs procedure (#10062)
---
docs/docs/spark/procedures.md | 1 +
docs/docs/spark/procedures/maintenance.md | 30 +++
.../org/apache/paimon/utils/ProcedureUtils.java | 29 +++
.../apache/paimon/table/ExpireChangelogImpl.java | 63 ++++-
.../apache/paimon/table/ChangelogExpireTest.java | 86 ++++++-
.../org/apache/paimon/spark/SparkProcedures.java | 2 +
.../spark/procedure/ExpireChangelogsProcedure.java | 157 ++++++++++++
.../procedure/ExpireChangelogsProcedureTest.scala | 265 +++++++++++++++++++++
8 files changed, 623 insertions(+), 10 deletions(-)
diff --git a/docs/docs/spark/procedures.md b/docs/docs/spark/procedures.md
index 947de28348..441033b332 100644
--- a/docs/docs/spark/procedures.md
+++ b/docs/docs/spark/procedures.md
@@ -65,6 +65,7 @@ Choose a group, then use the page contents to jump to a
procedure:
[`materialize_deletion_vectors`](./procedures/maintenance#materialize_deletion_vectors),
[`rescale`](./procedures/maintenance#rescale),
[`expire_snapshots`](./procedures/maintenance#expire_snapshots),
+[`expire_changelogs`](./procedures/maintenance#expire_changelogs),
[`expire_partitions`](./procedures/maintenance#expire_partitions),
[`remove_orphan_files`](./procedures/maintenance#remove_orphan_files),
[`remove_orphan_blobs`](./procedures/maintenance#remove_orphan_blobs),
diff --git a/docs/docs/spark/procedures/maintenance.md
b/docs/docs/spark/procedures/maintenance.md
index 366dd13e53..2153d68247 100644
--- a/docs/docs/spark/procedures/maintenance.md
+++ b/docs/docs/spark/procedures/maintenance.md
@@ -199,6 +199,36 @@ Expire snapshots.
CALL sys.expire_snapshots(table => 'default.T', retain_max => 10, options =>
'snapshot.expire.limit=1');
```
+## expire_changelogs
+
+Expire separated changelog files. Unspecified arguments fall back to table
properties
+(`changelog.num-retained.max` / `changelog.num-retained.min` /
`changelog.time-retained`, or the
+corresponding `snapshot.*` options when changelog options are not set, and
`snapshot.expire.limit`
+for `max_deletes`). `retain_min` and `retain_max` are applied against the
latest snapshot id, not
+the long-lived changelog count. The procedure returns
`deleted_changelogs_count`, the number of
+changelogs successfully deleted. A missing changelog, one skipped after a
file-skipper failure, or
+one whose changelog file is still present after a failed deletion, is not
included.
+
+**Arguments**
+
+- `table` (`STRING`, required): the target table identifier.
+- `retain_max` (`INT`, optional): the maximum number of completed changelogs
to retain.
+- `retain_min` (`INT`, optional): the minimum number of completed changelogs
to retain.
+- `older_than` (`STRING`, optional): timestamp before which changelogs will be
removed.
+- `max_deletes` (`INT`, optional): the maximum number of changelogs that can
be deleted at once.
+- `delete_all` (`BOOLEAN`, optional): when true, delete all separated
changelogs. Cannot be combined with `retain_max`, `retain_min`, `older_than`,
`max_deletes` or `options`.
+- `options` (`STRING`, optional): the additional dynamic options of the table.
These override stored table properties and are overridden by explicit procedure
arguments.
+
+```sql
+CALL sys.expire_changelogs(table => 'default.T', retain_max => 8);
+
+CALL sys.expire_changelogs(table => 'default.T', older_than => '2024-01-01
12:00:00', max_deletes => 2);
+
+CALL sys.expire_changelogs(table => 'default.T', options =>
'changelog.num-retained.max=8, changelog.num-retained.min=4');
+
+CALL sys.expire_changelogs(table => 'default.T', delete_all => true);
+```
+
## expire_partitions
Expire partitions.
diff --git
a/paimon-common/src/main/java/org/apache/paimon/utils/ProcedureUtils.java
b/paimon-common/src/main/java/org/apache/paimon/utils/ProcedureUtils.java
index d13cb3c6df..9e4f966beb 100644
--- a/paimon-common/src/main/java/org/apache/paimon/utils/ProcedureUtils.java
+++ b/paimon-common/src/main/java/org/apache/paimon/utils/ProcedureUtils.java
@@ -95,4 +95,33 @@ public class ProcedureUtils {
}
return builder;
}
+
+ public static ExpireConfig.Builder fillInChangelogOptions(
+ CoreOptions tableOptions,
+ Integer retainMax,
+ Integer retainMin,
+ String olderThanStr,
+ Integer maxDeletes) {
+
+ ExpireConfig.Builder builder = ExpireConfig.builder();
+ builder.snapshotRetainMax(tableOptions.snapshotNumRetainMax())
+ .snapshotRetainMin(tableOptions.snapshotNumRetainMin())
+ .snapshotTimeRetain(tableOptions.snapshotTimeRetain())
+ .snapshotMaxDeletes(tableOptions.snapshotExpireLimit())
+ .changelogRetainMax(
+
Optional.ofNullable(retainMax).orElse(tableOptions.changelogNumRetainMax()))
+ .changelogRetainMin(
+
Optional.ofNullable(retainMin).orElse(tableOptions.changelogNumRetainMin()))
+ .changelogMaxDeletes(
+
Optional.ofNullable(maxDeletes).orElse(tableOptions.snapshotExpireLimit()))
+ .changelogTimeRetain(tableOptions.changelogTimeRetain());
+ if (!StringUtils.isNullOrWhitespaceOnly(olderThanStr)) {
+ long olderThanMills =
+ DateTimeUtils.parseTimestampData(olderThanStr, 3,
TimeZone.getDefault())
+ .getMillisecond();
+ builder.changelogTimeRetain(
+ Duration.ofMillis(System.currentTimeMillis() -
olderThanMills));
+ }
+ return builder;
+ }
}
diff --git
a/paimon-core/src/main/java/org/apache/paimon/table/ExpireChangelogImpl.java
b/paimon-core/src/main/java/org/apache/paimon/table/ExpireChangelogImpl.java
index b5d0af3954..fbb044c1a0 100644
--- a/paimon-core/src/main/java/org/apache/paimon/table/ExpireChangelogImpl.java
+++ b/paimon-core/src/main/java/org/apache/paimon/table/ExpireChangelogImpl.java
@@ -21,6 +21,8 @@ package org.apache.paimon.table;
import org.apache.paimon.Changelog;
import org.apache.paimon.Snapshot;
import org.apache.paimon.consumer.ConsumerManager;
+import org.apache.paimon.fs.FileIO;
+import org.apache.paimon.fs.Path;
import org.apache.paimon.manifest.ExpireFileEntry;
import org.apache.paimon.operation.ChangelogDeletion;
import org.apache.paimon.options.ExpireConfig;
@@ -149,6 +151,12 @@ public class ExpireChangelogImpl implements
ExpireSnapshots {
return expireUntil(earliestChangelogId, maxExclusive);
}
+ /**
+ * Expire changelogs in {@code [earliestId, endExclusiveId)}.
+ *
+ * @return number of changelogs successfully deleted. A missing changelog,
one skipped after a
+ * file-skipper failure, or one whose file remains after a failed
deletion, is not counted.
+ */
public int expireUntil(long earliestId, long endExclusiveId) {
if (LOG.isDebugEnabled()) {
LOG.debug("Changelog expire range is [{}, {})", earliestId,
endExclusiveId);
@@ -169,6 +177,7 @@ public class ExpireChangelogImpl implements ExpireSnapshots
{
}
skippingSnapshots.add(snapshotManager.earliestSnapshot());
Set<String> manifestSkippSet =
changelogDeletion.manifestSkippingSet(skippingSnapshots);
+ int deleted = 0;
for (long id = earliestId; id < endExclusiveId; id++) {
if (LOG.isDebugEnabled()) {
LOG.debug("Ready to delete changelog files from changelog
#{}", id);
@@ -193,34 +202,53 @@ public class ExpireChangelogImpl implements
ExpireSnapshots {
changelogDeletion.cleanDeletedDataFiles(changelog, skipper);
changelogDeletion.cleanUnusedManifests(changelog,
manifestSkippSet);
-
changelogManager.fileIO().deleteQuietly(changelogManager.longLivedChangelogPath(id));
+ if (deleteChangelogFile(
+ changelogManager.fileIO(),
changelogManager.longLivedChangelogPath(id))) {
+ deleted++;
+ }
}
changelogDeletion.cleanEmptyDirectories();
writeEarliestHintFile(endExclusiveId);
- return (int) (endExclusiveId - earliestId);
+ return deleted;
}
- /** expire all separated changelogs, only used by
ExpireChangelogsProcedure. */
+ /**
+ * Expire all separated changelogs.
+ *
+ * <p>The return type stays {@code void} so callers compiled against
{@code ()V}, including the
+ * Flink procedure, remain compatible. Use {@link
#expireAllDeletedCount()} when the deleted
+ * count is needed.
+ */
public void expireAll() {
+ expireAllDeletedCount();
+ }
+
+ /**
+ * Expire all separated changelogs.
+ *
+ * @return number of changelogs successfully deleted. A missing changelog,
one skipped after a
+ * file-skipper failure, or one whose file remains after a failed
deletion, is not counted.
+ */
+ public int expireAllDeletedCount() {
Long latestSnapshotId = snapshotManager.latestSnapshotId();
if (latestSnapshotId == null) {
// no snapshot, nothing to expire
- return;
+ return 0;
}
Long earliestSnapshotId = snapshotManager.earliestSnapshotId();
if (earliestSnapshotId == null) {
- return;
+ return 0;
}
Long latestChangelogId = changelogManager.latestLongLivedChangelogId();
if (latestChangelogId == null) {
- return;
+ return 0;
}
Long earliestChangelogId =
changelogManager.earliestLongLivedChangelogId();
if (earliestChangelogId == null) {
- return;
+ return 0;
}
LOG.info(
@@ -240,6 +268,7 @@ public class ExpireChangelogImpl implements ExpireSnapshots
{
skippingSnapshots.add(snapshotManager.snapshot(earliestSnapshotId));
Set<String> manifestSkippSet =
changelogDeletion.manifestSkippingSet(skippingSnapshots);
+ int deleted = 0;
for (long id = earliestChangelogId; id <= latestChangelogId; id++) {
LOG.info("Ready to delete changelog files from changelog #{}", id);
@@ -264,7 +293,10 @@ public class ExpireChangelogImpl implements
ExpireSnapshots {
changelogDeletion.cleanDeletedDataFiles(changelog, skipper);
changelogDeletion.cleanUnusedManifests(changelog,
manifestSkippSet);
-
changelogManager.fileIO().deleteQuietly(changelogManager.longLivedChangelogPath(id));
+ if (deleteChangelogFile(
+ changelogManager.fileIO(),
changelogManager.longLivedChangelogPath(id))) {
+ deleted++;
+ }
}
// try delete changelog hint file
@@ -276,6 +308,21 @@ public class ExpireChangelogImpl implements
ExpireSnapshots {
}
changelogDeletion.cleanEmptyDirectories();
+ return deleted;
+ }
+
+ /**
+ * Delete a changelog file and report whether it is actually gone. {@link
FileIO#deleteQuietly}
+ * swallows I/O failures, so a file that is still present is not a
successful removal.
+ */
+ static boolean deleteChangelogFile(FileIO fileIO, Path path) {
+ fileIO.deleteQuietly(path);
+ try {
+ return !fileIO.exists(path);
+ } catch (IOException e) {
+ LOG.warn("Failed to check whether changelog file {} still
exists.", path, e);
+ return false;
+ }
}
private void writeEarliestHintFile(long earliest) {
diff --git
a/paimon-core/src/test/java/org/apache/paimon/table/ChangelogExpireTest.java
b/paimon-core/src/test/java/org/apache/paimon/table/ChangelogExpireTest.java
index 281dd0f1a6..fa08f69f7c 100644
--- a/paimon-core/src/test/java/org/apache/paimon/table/ChangelogExpireTest.java
+++ b/paimon-core/src/test/java/org/apache/paimon/table/ChangelogExpireTest.java
@@ -26,6 +26,7 @@ import org.apache.paimon.catalog.CatalogFactory;
import org.apache.paimon.catalog.Identifier;
import org.apache.paimon.fs.FileIO;
import org.apache.paimon.fs.Path;
+import org.apache.paimon.fs.local.LocalFileIO;
import org.apache.paimon.manifest.ManifestFileMeta;
import org.apache.paimon.manifest.ManifestList;
import org.apache.paimon.options.ExpireConfig;
@@ -184,13 +185,94 @@ public class ChangelogExpireTest extends
IndexFileExpireTableTest {
assertThat(fileIO.exists(changelogManager.longLivedChangelogPath(middleId))).isTrue();
fileIO.deleteQuietly(changelogManager.longLivedChangelogPath(middleId));
+ int expectedDeleted = 0;
+ for (long id = earliestChangelogId; id < latestChangelogId; id++) {
+ if (changelogManager.longLivedChangelogExists(id)) {
+ expectedDeleted++;
+ }
+ }
+ assertThat(expectedDeleted).isLessThan((int) (latestChangelogId -
earliestChangelogId));
+
ExpireChangelogImpl expire =
(ExpireChangelogImpl)
table.newExpireChangelog().config(expireConfig);
- // should not throw even though a middle changelog is missing
- assertThatCode(expire::expire).doesNotThrowAnyException();
+ // should not throw even though a middle changelog is missing, and the
count excludes it
+ assertThat(expire.expire()).isEqualTo(expectedDeleted);
// earliest should be advanced past the deleted range
assertThat(changelogManager.earliestLongLivedChangelogId()).isEqualTo(latestChangelogId);
+
assertThat(changelogManager.longLivedChangelogExists(latestChangelogId)).isTrue();
+ }
+
+ @Test
+ public void testDeleteChangelogFileRequiresActualRemoval() throws
Exception {
+ LocalFileIO fileIO = new LocalFileIO();
+ java.nio.file.Path tempDir =
java.nio.file.Files.createTempDirectory("changelog-delete");
+ Path dir = new Path(tempDir.toString());
+ try {
+ Path removed = new Path(dir, "changelog-1");
+ fileIO.writeFile(removed, "{}", false);
+ assertThat(ExpireChangelogImpl.deleteChangelogFile(fileIO,
removed)).isTrue();
+ assertThat(fileIO.exists(removed)).isFalse();
+
+ Path blocked = new Path(dir, "changelog-2");
+ assertThat(fileIO.mkdirs(blocked)).isTrue();
+ fileIO.writeFile(new Path(blocked, "keep"), "x", false);
+ assertThat(ExpireChangelogImpl.deleteChangelogFile(fileIO,
blocked)).isFalse();
+ assertThat(fileIO.exists(blocked)).isTrue();
+ } finally {
+ fileIO.delete(dir, true);
+ }
+ }
+
+ @Test
+ public void testExpireAllDeletedCountSkipsMissingChangelog() throws
Exception {
+ StreamWriteBuilder writeBuilder = table.newStreamWriteBuilder();
+ StreamTableWrite write = writeBuilder.newWrite();
+ StreamTableCommit commit = writeBuilder.newCommit();
+ for (int i = 1; i <= 10; i++) {
+ write(write, createRow(1, 0, i, i * 10));
+ commit.commit(i, write.prepareCommit(true, i));
+ }
+ write.close();
+ commit.close();
+
+ SnapshotManager snapshotManager = table.snapshotManager();
+ long latestSnapshotId = snapshotManager.latestSnapshotId();
+
+ // changelogRetainMax > snapshotRetainMax to ensure
changelogDecoupled=true
+ ExpireConfig expireConfig =
+ ExpireConfig.builder()
+ .changelogRetainMax((int) latestSnapshotId)
+ .changelogRetainMin(1)
+ .changelogTimeRetain(Duration.ofMillis(0))
+ .snapshotRetainMax(1)
+ .snapshotRetainMin(1)
+ .build();
+ ExpireSnapshotsImpl expireSnapshots =
+ (ExpireSnapshotsImpl)
table.newExpireSnapshots().config(expireConfig);
+ expireSnapshots.expire();
+
+ ChangelogManager changelogManager = table.changelogManager();
+ FileIO fileIO = table.fileIO();
+ long latestChangelogId = changelogManager.latestLongLivedChangelogId();
+ long earliestChangelogId =
changelogManager.earliestLongLivedChangelogId();
+
+ long middleId = (earliestChangelogId + latestChangelogId) / 2;
+
assertThat(fileIO.exists(changelogManager.longLivedChangelogPath(middleId))).isTrue();
+
fileIO.deleteQuietly(changelogManager.longLivedChangelogPath(middleId));
+
+ int existing = 0;
+ for (long id = earliestChangelogId; id <= latestChangelogId; id++) {
+ if (changelogManager.longLivedChangelogExists(id)) {
+ existing++;
+ }
+ }
+ assertThat(existing).isEqualTo((int) (latestChangelogId -
earliestChangelogId));
+
+ ExpireChangelogImpl expire = (ExpireChangelogImpl)
table.newExpireChangelog();
+ assertThat(expire.expireAllDeletedCount()).isEqualTo(existing);
+ assertThat(changelogManager.latestLongLivedChangelogId()).isNull();
+ assertThat(changelogManager.earliestLongLivedChangelogId()).isNull();
}
}
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 18bb328375..aebc19389e 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
@@ -37,6 +37,7 @@ import org.apache.paimon.spark.procedure.DeleteTagProcedure;
import org.apache.paimon.spark.procedure.DropFunctionProcedure;
import org.apache.paimon.spark.procedure.DropGlobalIndexProcedure;
import org.apache.paimon.spark.procedure.DropPolicyProcedure;
+import org.apache.paimon.spark.procedure.ExpireChangelogsProcedure;
import org.apache.paimon.spark.procedure.ExpirePartitionsProcedure;
import org.apache.paimon.spark.procedure.ExpireSnapshotsProcedure;
import org.apache.paimon.spark.procedure.ExpireTagsProcedure;
@@ -129,6 +130,7 @@ public class SparkProcedures {
procedureBuilders.put(
"remove_unexisting_manifests",
RemoveUnexistingManifestsProcedure::builder);
procedureBuilders.put("expire_snapshots",
ExpireSnapshotsProcedure::builder);
+ procedureBuilders.put("expire_changelogs",
ExpireChangelogsProcedure::builder);
procedureBuilders.put("expire_partitions",
ExpirePartitionsProcedure::builder);
procedureBuilders.put("repair", RepairProcedure::builder);
procedureBuilders.put("repair_earliest_snapshot",
RepairEarliestSnapshotProcedure::builder);
diff --git
a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/ExpireChangelogsProcedure.java
b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/ExpireChangelogsProcedure.java
new file mode 100644
index 0000000000..0f696e234e
--- /dev/null
+++
b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/ExpireChangelogsProcedure.java
@@ -0,0 +1,157 @@
+/*
+ * 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.CoreOptions;
+import org.apache.paimon.options.ExpireConfig;
+import org.apache.paimon.table.ExpireChangelogImpl;
+import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.utils.ProcedureUtils;
+import org.apache.paimon.utils.StringUtils;
+
+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 java.util.HashMap;
+
+import static org.apache.paimon.utils.Preconditions.checkArgument;
+import static org.apache.spark.sql.types.DataTypes.BooleanType;
+import static org.apache.spark.sql.types.DataTypes.IntegerType;
+import static org.apache.spark.sql.types.DataTypes.StringType;
+
+/** A procedure to expire changelogs. */
+public class ExpireChangelogsProcedure extends BaseProcedure {
+
+ private static final ProcedureParameter[] PARAMETERS =
+ new ProcedureParameter[] {
+ ProcedureParameter.required("table", StringType),
+ ProcedureParameter.optional("retain_max", IntegerType),
+ ProcedureParameter.optional("retain_min", IntegerType),
+ ProcedureParameter.optional("older_than", StringType),
+ ProcedureParameter.optional("max_deletes", IntegerType),
+ ProcedureParameter.optional("delete_all", BooleanType),
+ ProcedureParameter.optional("options", StringType)
+ };
+
+ private static final StructType OUTPUT_TYPE =
+ new StructType(
+ new StructField[] {
+ new StructField(
+ "deleted_changelogs_count", IntegerType,
false, Metadata.empty())
+ });
+
+ private ExpireChangelogsProcedure(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());
+ Integer retainMax = args.isNullAt(1) ? null : args.getInt(1);
+ Integer retainMin = args.isNullAt(2) ? null : args.getInt(2);
+ String olderThanStr = args.isNullAt(3) ? null : args.getString(3);
+ Integer maxDeletes = args.isNullAt(4) ? null : args.getInt(4);
+ boolean deleteAll = !args.isNullAt(5) && args.getBoolean(5);
+ String options = args.isNullAt(6) ? null : args.getString(6);
+
+ return modifyPaimonTable(
+ tableIdent,
+ table -> {
+ checkArgument(
+ table instanceof FileStoreTable,
+ "Only FileStoreTable supports expire_changelogs
procedure. The table type is '%s'.",
+ table.getClass().getName());
+
+ if (deleteAll) {
+ checkArgument(
+ retainMax == null
+ && retainMin == null
+ && olderThanStr == null
+ && maxDeletes == null
+ &&
StringUtils.isNullOrWhitespaceOnly(options),
+ "delete_all cannot be used with retain_max,
retain_min, older_than, max_deletes or options.");
+ ExpireChangelogImpl expireChangelogs =
+ (ExpireChangelogImpl)
table.newExpireChangelog();
+ int deleted = expireChangelogs.expireAllDeletedCount();
+ return new InternalRow[] {newInternalRow(deleted)};
+ }
+
+ HashMap<String, String> dynamicOptions = new HashMap<>();
+ ProcedureUtils.putAllOptions(dynamicOptions, options);
+ // Explicit arguments override options. Merge them before
copy so schema
+ // validation sees the final config, not an intermediate
one.
+ overrideWithExplicitArguments(dynamicOptions, retainMax,
retainMin, maxDeletes);
+ table = table.copy(dynamicOptions);
+ ExpireChangelogImpl expireChangelogs =
+ (ExpireChangelogImpl) table.newExpireChangelog();
+ CoreOptions tableOptions = ((FileStoreTable)
table).store().options();
+ ExpireConfig.Builder builder =
+ ProcedureUtils.fillInChangelogOptions(
+ tableOptions, retainMax, retainMin,
olderThanStr, maxDeletes);
+ int deleted =
expireChangelogs.config(builder.build()).expire();
+ return new InternalRow[] {newInternalRow(deleted)};
+ });
+ }
+
+ private static void overrideWithExplicitArguments(
+ HashMap<String, String> dynamicOptions,
+ Integer retainMax,
+ Integer retainMin,
+ Integer maxDeletes) {
+ if (retainMax != null) {
+ dynamicOptions.put(
+ CoreOptions.CHANGELOG_NUM_RETAINED_MAX.key(),
String.valueOf(retainMax));
+ }
+ if (retainMin != null) {
+ dynamicOptions.put(
+ CoreOptions.CHANGELOG_NUM_RETAINED_MIN.key(),
String.valueOf(retainMin));
+ }
+ if (maxDeletes != null) {
+ dynamicOptions.put(CoreOptions.SNAPSHOT_EXPIRE_LIMIT.key(),
String.valueOf(maxDeletes));
+ }
+ }
+
+ public static ProcedureBuilder builder() {
+ return new BaseProcedure.Builder<ExpireChangelogsProcedure>() {
+ @Override
+ public ExpireChangelogsProcedure doBuild() {
+ return new ExpireChangelogsProcedure(tableCatalog());
+ }
+ };
+ }
+
+ @Override
+ public String description() {
+ return "ExpireChangelogsProcedure";
+ }
+}
diff --git
a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/procedure/ExpireChangelogsProcedureTest.scala
b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/procedure/ExpireChangelogsProcedureTest.scala
new file mode 100644
index 0000000000..bc0b12f638
--- /dev/null
+++
b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/procedure/ExpireChangelogsProcedureTest.scala
@@ -0,0 +1,265 @@
+/*
+ * 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.paimon.utils.{ChangelogManager, SnapshotManager}
+
+import org.apache.spark.sql.Row
+import org.assertj.core.api.Assertions.assertThat
+
+import java.sql.Timestamp
+
+class ExpireChangelogsProcedureTest extends PaimonSparkTestBase {
+
+ test("Paimon Procedure: expire changelogs") {
+ createInputChangelogTable()
+
+ val table = loadTable("T")
+ val snapshotManager = table.snapshotManager
+ val changelogManager = table.changelogManager
+
+ for (i <- 0 until 10) {
+ spark.sql(s"INSERT INTO T VALUES ('$i', $i)")
+ }
+ checkSnapshots(snapshotManager, 7, 10)
+ checkChangelogs(changelogManager, 1, 6)
+
+ checkAnswer(
+ spark.sql("CALL sys.expire_changelogs(table => 'test.T', retain_max =>
8)"),
+ Row(2) :: Nil)
+ checkChangelogs(changelogManager, 3, 6)
+
+ val ts = new Timestamp(snapshotManager.latestSnapshot().timeMillis)
+ checkAnswer(
+ spark.sql(
+ s"CALL sys.expire_changelogs(table => 'test.T', older_than =>
'${ts.toString}', max_deletes => 1)"),
+ Row(1) :: Nil)
+ checkChangelogs(changelogManager, 4, 6)
+
+ checkAnswer(
+ spark.sql(
+ s"CALL sys.expire_changelogs(table => 'test.T', older_than =>
'${ts.toString}', retain_min => 6)"),
+ Row(1) :: Nil)
+ checkChangelogs(changelogManager, 5, 6)
+
+ checkAnswer(
+ spark.sql(s"CALL sys.expire_changelogs(table => 'test.T', older_than =>
'${ts.toString}')"),
+ Row(1) :: Nil)
+ checkChangelogs(changelogManager, 6, 6)
+
+ for (i <- 10 until 12) {
+ spark.sql(s"INSERT INTO T VALUES ('$i', $i)")
+ }
+ checkSnapshots(snapshotManager, 9, 12)
+ checkChangelogs(changelogManager, 6, 8)
+
+ checkAnswer(
+ spark.sql("CALL sys.expire_changelogs(table => 'test.T', retain_max =>
4)"),
+ Row(2) :: Nil)
+ checkChangelogs(changelogManager, 8, 8)
+
+ checkAnswer(
+ spark.sql("CALL sys.expire_changelogs(table => 'test.T', delete_all =>
true)"),
+ Row(1) :: Nil)
+ checkAllDeleted(changelogManager)
+ }
+
+ test("Paimon Procedure: expire changelogs load table property first") {
+ createInputChangelogTable()
+
+ val table = loadTable("T")
+ val snapshotManager = table.snapshotManager
+ val changelogManager = table.changelogManager
+
+ for (i <- 0 until 10) {
+ spark.sql(s"INSERT INTO T VALUES ('$i', $i)")
+ }
+ checkSnapshots(snapshotManager, 7, 10)
+ checkChangelogs(changelogManager, 1, 6)
+
+ // Commits already honor changelog.num-retained.max=10, so empty CALL
would delete 0.
+ // Tighten table properties and verify the procedure reads them.
+ spark.sql("""
+ |ALTER TABLE T SET TBLPROPERTIES (
+ | 'changelog.num-retained.max' = '8',
+ | 'changelog.num-retained.min' = '4'
+ |)
+ |""".stripMargin)
+
+ checkAnswer(spark.sql("CALL sys.expire_changelogs(table => 'test.T')"),
Row(2) :: Nil)
+ checkChangelogs(changelogManager, 3, 6)
+ }
+
+ test("Paimon Procedure: expire changelogs add options parameter") {
+ createInputChangelogTable()
+
+ val table = loadTable("T")
+ val snapshotManager = table.snapshotManager
+ val changelogManager = table.changelogManager
+
+ for (i <- 0 until 10) {
+ spark.sql(s"INSERT INTO T VALUES ('$i', $i)")
+ }
+ checkSnapshots(snapshotManager, 7, 10)
+ checkChangelogs(changelogManager, 1, 6)
+
+ checkAnswer(spark.sql("CALL sys.expire_changelogs(table => 'test.T')"),
Row(0) :: Nil)
+ checkChangelogs(changelogManager, 1, 6)
+
+ checkAnswer(
+ spark.sql(
+ "CALL sys.expire_changelogs(table => 'test.T', options =>
'changelog.num-retained.max=8, changelog.num-retained.min=4')"),
+ Row(2) :: Nil)
+ checkChangelogs(changelogManager, 3, 6)
+ }
+
+ test("Paimon Procedure: expire changelogs explicit args override options") {
+ createInputChangelogTable()
+
+ val table = loadTable("T")
+ val snapshotManager = table.snapshotManager
+ val changelogManager = table.changelogManager
+
+ for (i <- 0 until 10) {
+ spark.sql(s"INSERT INTO T VALUES ('$i', $i)")
+ }
+ checkSnapshots(snapshotManager, 7, 10)
+ checkChangelogs(changelogManager, 1, 6)
+
+ // options alone would be max=2 < table min=4 and fail validation.
+ // retain_max overrides it, so the validated config is min=4, max=8.
+ checkAnswer(
+ spark.sql(
+ "CALL sys.expire_changelogs(table => 'test.T', retain_max => 8,
options => 'changelog.num-retained.max=2')"),
+ Row(2) :: Nil)
+ checkChangelogs(changelogManager, 3, 6)
+ }
+
+ test("Paimon Procedure: expire changelogs without separated changelogs") {
+ spark.sql("CREATE TABLE T (word STRING, cnt INT)")
+ spark.sql("INSERT INTO T VALUES ('a', 1)")
+
+ checkAnswer(spark.sql("CALL sys.expire_changelogs(table => 'test.T')"),
Row(0) :: Nil)
+ checkAnswer(
+ spark.sql("CALL sys.expire_changelogs(table => 'test.T', delete_all =>
true)"),
+ Row(0) :: Nil)
+ }
+
+ test("Paimon Procedure: expire changelogs retainMax retainMin value check") {
+ createInputChangelogTable()
+
+ for (i <- 0 until 10) {
+ spark.sql(s"INSERT INTO T VALUES ('$i', $i)")
+ }
+
+ assertThrows[IllegalArgumentException] {
+ spark.sql("CALL sys.expire_changelogs(table => 'test.T', retain_max =>
2, retain_min => 3)")
+ }
+ }
+
+ test("Paimon Procedure: expire changelogs skips missing changelog") {
+ createInputChangelogTable()
+
+ val table = loadTable("T")
+ val changelogManager = table.changelogManager
+
+ for (i <- 0 until 10) {
+ spark.sql(s"INSERT INTO T VALUES ('$i', $i)")
+ }
+ checkChangelogs(changelogManager, 1, 6)
+
+ // retain_max 8 deletes changelogs 1 and 2; a missing id 2 must not be
counted
+ val missingId = 2L
+ assertThat(changelogManager.longLivedChangelogExists(missingId)).isTrue()
+
changelogManager.fileIO.deleteQuietly(changelogManager.longLivedChangelogPath(missingId))
+
+ checkAnswer(
+ spark.sql("CALL sys.expire_changelogs(table => 'test.T', retain_max =>
8)"),
+ Row(1) :: Nil)
+ checkChangelogs(changelogManager, 3, 6)
+ }
+
+ test("Paimon Procedure: expire changelogs delete_all skips missing
changelog") {
+ createInputChangelogTable()
+
+ val table = loadTable("T")
+ val changelogManager = table.changelogManager
+
+ for (i <- 0 until 10) {
+ spark.sql(s"INSERT INTO T VALUES ('$i', $i)")
+ }
+ checkChangelogs(changelogManager, 1, 6)
+
+ val missingId = 3L
+ assertThat(changelogManager.longLivedChangelogExists(missingId)).isTrue()
+
changelogManager.fileIO.deleteQuietly(changelogManager.longLivedChangelogPath(missingId))
+
+ checkAnswer(
+ spark.sql("CALL sys.expire_changelogs(table => 'test.T', delete_all =>
true)"),
+ Row(5) :: Nil)
+ checkAllDeleted(changelogManager)
+ }
+
+ test("Paimon Procedure: expire changelogs delete_all cannot mix with other
arguments") {
+ createInputChangelogTable()
+ spark.sql("INSERT INTO T VALUES ('0', 0)")
+
+ assertThrows[IllegalArgumentException] {
+ spark.sql(
+ "CALL sys.expire_changelogs(table => 'test.T', delete_all => true,
retain_max => 8)")
+ }
+ assertThrows[IllegalArgumentException] {
+ spark.sql(
+ "CALL sys.expire_changelogs(table => 'test.T', delete_all => true,
options => 'changelog.num-retained.max=8')")
+ }
+ }
+
+ private def createInputChangelogTable(): Unit = {
+ spark.sql("""
+ |CREATE TABLE T (word STRING, cnt INT)
+ |TBLPROPERTIES (
+ | 'primary-key' = 'word',
+ | 'bucket' = '1',
+ | 'num-sorted-run.compaction-trigger' = '9999',
+ | 'changelog-producer' = 'input',
+ | 'snapshot.num-retained.min' = '4',
+ | 'snapshot.num-retained.max' = '4',
+ | 'changelog.num-retained.min' = '4',
+ | 'changelog.num-retained.max' = '10'
+ |)
+ |""".stripMargin)
+ }
+
+ private def checkSnapshots(sm: SnapshotManager, earliest: Int, latest: Int):
Unit = {
+ assertThat(sm.snapshotCount).isEqualTo(latest - earliest + 1)
+ assertThat(sm.earliestSnapshotId).isEqualTo(earliest)
+ assertThat(sm.latestSnapshotId).isEqualTo(latest)
+ }
+
+ private def checkChangelogs(cm: ChangelogManager, earliest: Int, latest:
Int): Unit = {
+ assertThat(cm.earliestLongLivedChangelogId).isEqualTo(earliest)
+ assertThat(cm.latestLongLivedChangelogId).isEqualTo(latest)
+ }
+
+ private def checkAllDeleted(cm: ChangelogManager): Unit = {
+ assertThat(cm.latestLongLivedChangelogId).isNull()
+ assertThat(cm.earliestLongLivedChangelogId).isNull()
+ }
+}
