From c9f317aba32b6332847f795c773910ef310e917a Mon Sep 17 00:00:00 2001 From: big face cat <731030576@qq.com> Date: Sat, 5 Sep 2026 10:14:21 +0800 Subject: [PATCH 1/4] [core] expire snapshot branch snapshots from chain delta commits --- .../paimon/table/PrimaryKeyFileStoreTable.java | 18 ++++++++++++++++-- 1 file changed, 16 insertions(+), 2 deletions(-) diff --git a/paimon-core/src/main/java/org/apache/paimon/table/PrimaryKeyFileStoreTable.java b/paimon-core/src/main/java/org/apache/paimon/table/PrimaryKeyFileStoreTable.java index f521f2e9b17f..c0fa777f4d82 100644 --- a/paimon-core/src/main/java/org/apache/paimon/table/PrimaryKeyFileStoreTable.java +++ b/paimon-core/src/main/java/org/apache/paimon/table/PrimaryKeyFileStoreTable.java @@ -41,6 +41,7 @@ import org.apache.paimon.table.source.PrimaryKeyBatchScan; import org.apache.paimon.table.source.SplitGenerator; import org.apache.paimon.types.RowType; +import org.apache.paimon.utils.ChainTableUtils; import org.apache.paimon.utils.RowKindFilter; import javax.annotation.Nullable; @@ -213,8 +214,21 @@ public LocalTableQuery newLocalTableQuery() { protected Runnable newExpireRunnable() { if (coreOptions().bucket() == BucketMode.POSTPONE_BUCKET) { return null; - } else { - return super.newExpireRunnable(); } + + Runnable expire = super.newExpireRunnable(); + CoreOptions options = coreOptions(); + if (expire == null || !ChainTableUtils.isScanFallbackDeltaBranch(options)) { + return expire; + } + + ExpireSnapshots snapshotBranchExpire = + switchToBranch(options.scanFallbackSnapshotBranch()) + .newExpireSnapshots() + .config(options.expireConfig()); + return () -> { + expire.run(); + snapshotBranchExpire.expire(); + }; } } From ddf703af1f1ae5d11a07b6cc6143c2f5a1fb819e Mon Sep 17 00:00:00 2001 From: big face cat <731030576@qq.com> Date: Sat, 5 Sep 2026 10:14:36 +0800 Subject: [PATCH 2/4] [core] add chain table snapshot expiration regression test --- .../table/ChainTableSnapshotExpireTest.java | 153 ++++++++++++++++++ 1 file changed, 153 insertions(+) create mode 100644 paimon-core/src/test/java/org/apache/paimon/table/ChainTableSnapshotExpireTest.java diff --git a/paimon-core/src/test/java/org/apache/paimon/table/ChainTableSnapshotExpireTest.java b/paimon-core/src/test/java/org/apache/paimon/table/ChainTableSnapshotExpireTest.java new file mode 100644 index 000000000000..9a559c4c260c --- /dev/null +++ b/paimon-core/src/test/java/org/apache/paimon/table/ChainTableSnapshotExpireTest.java @@ -0,0 +1,153 @@ +/* + * 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.table; + +import org.apache.paimon.CoreOptions; +import org.apache.paimon.data.BinaryString; +import org.apache.paimon.data.GenericRow; +import org.apache.paimon.fs.Path; +import org.apache.paimon.fs.local.LocalFileIO; +import org.apache.paimon.options.Options; +import org.apache.paimon.schema.FileSystemSchemaManager; +import org.apache.paimon.schema.Schema; +import org.apache.paimon.schema.SchemaChange; +import org.apache.paimon.schema.SchemaManager; +import org.apache.paimon.schema.TableSchema; +import org.apache.paimon.table.sink.CommitMessage; +import org.apache.paimon.table.sink.StreamTableWrite; +import org.apache.paimon.table.sink.TableCommitImpl; +import org.apache.paimon.types.DataTypes; +import org.apache.paimon.types.RowType; + +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.io.TempDir; + +import java.util.Arrays; +import java.util.Collections; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.UUID; + +import static org.assertj.core.api.Assertions.assertThat; + +/** Tests snapshot expiration maintenance for chain table branches. */ +public class ChainTableSnapshotExpireTest { + + @TempDir java.nio.file.Path tempDir; + + @Test + public void testDeltaCommitExpiresSnapshotBranchSnapshots() throws Exception { + Path tablePath = new Path(tempDir.toUri().toString(), "chain_snapshot_expire"); + createChainTable(tablePath); + + FileStoreTable mainTable = loadTable(tablePath); + FileStoreTable snapshotTable = mainTable.switchToBranch("snapshot"); + FileStoreTable deltaTable = mainTable.switchToBranch("delta"); + String commitUser = UUID.randomUUID().toString(); + + write(snapshotTable, commitUser, "20250101", "v1"); + write(snapshotTable, commitUser, "20250201", "v2"); + write(snapshotTable, commitUser, "20250301", "v3"); + + snapshotTable = loadTable(tablePath).switchToBranch("snapshot"); + assertThat(snapshotTable.snapshotManager().snapshotCount()).isEqualTo(3); + + Map expireOptions = new HashMap<>(); + expireOptions.put(CoreOptions.WRITE_ONLY.key(), "false"); + expireOptions.put(CoreOptions.SNAPSHOT_NUM_RETAINED_MIN.key(), "1"); + expireOptions.put(CoreOptions.SNAPSHOT_NUM_RETAINED_MAX.key(), "1"); + expireOptions.put(CoreOptions.SNAPSHOT_TIME_RETAINED.key(), "0 ms"); + expireOptions.put(CoreOptions.SNAPSHOT_EXPIRE_EXECUTION_MODE.key(), "sync"); + deltaTable = deltaTable.copy(expireOptions); + + write(deltaTable, commitUser, "20250401", "v4"); + + snapshotTable = loadTable(tablePath).switchToBranch("snapshot"); + assertThat(snapshotTable.snapshotManager().snapshotCount()).isEqualTo(1); + } + + private void createChainTable(Path tablePath) throws Exception { + LocalFileIO fileIO = LocalFileIO.create(); + SchemaManager schemaManager = new FileSystemSchemaManager(fileIO, tablePath); + + Map options = new HashMap<>(); + options.put(CoreOptions.BUCKET.key(), "1"); + options.put(CoreOptions.MERGE_ENGINE.key(), "deduplicate"); + options.put(CoreOptions.SEQUENCE_FIELD.key(), "v"); + + Schema schema = + new Schema( + RowType.of( + new org.apache.paimon.types.DataType[] { + DataTypes.STRING(), DataTypes.STRING(), DataTypes.STRING() + }, + new String[] {"dt", "pk", "v"}) + .getFields(), + Collections.singletonList("dt"), + Arrays.asList("pk", "dt"), + options, + ""); + schemaManager.createTable(schema); + + FileStoreTable mainTable = loadTable(tablePath); + mainTable.createBranch("snapshot"); + mainTable.createBranch("delta"); + + List chainOptions = + Arrays.asList( + SchemaChange.setOption(CoreOptions.CHAIN_TABLE_ENABLED.key(), "true"), + SchemaChange.setOption( + CoreOptions.SCAN_FALLBACK_SNAPSHOT_BRANCH.key(), "snapshot"), + SchemaChange.setOption(CoreOptions.SCAN_FALLBACK_DELTA_BRANCH.key(), "delta"), + SchemaChange.setOption( + CoreOptions.PARTITION_TIMESTAMP_PATTERN.key(), "$dt"), + SchemaChange.setOption( + CoreOptions.PARTITION_TIMESTAMP_FORMATTER.key(), "yyyyMMdd")); + schemaManager.commitChanges(chainOptions); + new FileSystemSchemaManager(fileIO, tablePath, "snapshot").commitChanges(chainOptions); + new FileSystemSchemaManager(fileIO, tablePath, "delta").commitChanges(chainOptions); + } + + private FileStoreTable loadTable(Path tablePath) { + LocalFileIO fileIO = LocalFileIO.create(); + Options options = new Options(); + options.set(CoreOptions.PATH, tablePath.toString()); + String branchName = CoreOptions.branch(options.toMap()); + TableSchema tableSchema = + new FileSystemSchemaManager(fileIO, tablePath, branchName).latest().get(); + return FileStoreTableFactory.create( + fileIO, tablePath, tableSchema, CatalogEnvironment.empty()); + } + + private void write(FileStoreTable table, String commitUser, String dt, String value) + throws Exception { + StreamTableWrite write = table.newWrite(commitUser); + write.write( + GenericRow.of( + BinaryString.fromString(dt), + BinaryString.fromString(value), + BinaryString.fromString(value))); + try (TableCommitImpl commit = table.newCommit(commitUser)) { + List commitMessages = write.prepareCommit(true, Long.MAX_VALUE); + commit.commit(Long.MAX_VALUE, commitMessages); + } + write.close(); + } +} From 37bff717f766422adbcfd80a04f20f7e2ed4f068 Mon Sep 17 00:00:00 2001 From: big face cat <731030576@qq.com> Date: Sat, 5 Sep 2026 10:24:15 +0800 Subject: [PATCH 3/4] [core] strengthen chain table snapshot expiration regression test --- .../table/ChainTableSnapshotExpireTest.java | 47 +++++++++++++++++-- 1 file changed, 42 insertions(+), 5 deletions(-) diff --git a/paimon-core/src/test/java/org/apache/paimon/table/ChainTableSnapshotExpireTest.java b/paimon-core/src/test/java/org/apache/paimon/table/ChainTableSnapshotExpireTest.java index 9a559c4c260c..3f80c6781dc9 100644 --- a/paimon-core/src/test/java/org/apache/paimon/table/ChainTableSnapshotExpireTest.java +++ b/paimon-core/src/test/java/org/apache/paimon/table/ChainTableSnapshotExpireTest.java @@ -23,6 +23,7 @@ import org.apache.paimon.data.GenericRow; import org.apache.paimon.fs.Path; import org.apache.paimon.fs.local.LocalFileIO; +import org.apache.paimon.manifest.PartitionEntry; import org.apache.paimon.options.Options; import org.apache.paimon.schema.FileSystemSchemaManager; import org.apache.paimon.schema.Schema; @@ -38,22 +39,29 @@ import org.junit.jupiter.api.Test; import org.junit.jupiter.api.io.TempDir; +import java.time.LocalDate; +import java.time.format.DateTimeFormatter; import java.util.Arrays; import java.util.Collections; import java.util.HashMap; import java.util.List; import java.util.Map; import java.util.UUID; +import java.util.stream.Collectors; import static org.assertj.core.api.Assertions.assertThat; /** Tests snapshot expiration maintenance for chain table branches. */ public class ChainTableSnapshotExpireTest { + private static final DateTimeFormatter PARTITION_FORMATTER = + DateTimeFormatter.ofPattern("yyyyMMdd"); + @TempDir java.nio.file.Path tempDir; @Test - public void testDeltaCommitExpiresSnapshotBranchSnapshots() throws Exception { + public void testDeltaCommitExpiresSnapshotBranchSnapshotsAfterPartitionExpiration() + throws Exception { Path tablePath = new Path(tempDir.toUri().toString(), "chain_snapshot_expire"); createChainTable(tablePath); @@ -62,24 +70,41 @@ public void testDeltaCommitExpiresSnapshotBranchSnapshots() throws Exception { FileStoreTable deltaTable = mainTable.switchToBranch("delta"); String commitUser = UUID.randomUUID().toString(); - write(snapshotTable, commitUser, "20250101", "v1"); - write(snapshotTable, commitUser, "20250201", "v2"); - write(snapshotTable, commitUser, "20250301", "v3"); + String day90 = partitionDate(-90); + String day65 = partitionDate(-65); + String day40 = partitionDate(-40); + String day10 = partitionDate(-10); + + // Build three snapshot anchors while partition expiration is disabled. + write(snapshotTable, commitUser, day90, "v1"); + write(snapshotTable, commitUser, day65, "v2"); + write(snapshotTable, commitUser, day40, "v3"); snapshotTable = loadTable(tablePath).switchToBranch("snapshot"); + assertThat(listPartitions(snapshotTable)).containsExactly(day90, day65, day40); assertThat(snapshotTable.snapshotManager().snapshotCount()).isEqualTo(3); + long latestSnapshotBeforeExpiration = snapshotTable.snapshotManager().latestSnapshotId(); Map expireOptions = new HashMap<>(); expireOptions.put(CoreOptions.WRITE_ONLY.key(), "false"); + expireOptions.put(CoreOptions.PARTITION_EXPIRATION_TIME.key(), "30 d"); + expireOptions.put(CoreOptions.END_INPUT_CHECK_PARTITION_EXPIRE.key(), "true"); expireOptions.put(CoreOptions.SNAPSHOT_NUM_RETAINED_MIN.key(), "1"); expireOptions.put(CoreOptions.SNAPSHOT_NUM_RETAINED_MAX.key(), "1"); expireOptions.put(CoreOptions.SNAPSHOT_TIME_RETAINED.key(), "0 ms"); expireOptions.put(CoreOptions.SNAPSHOT_EXPIRE_EXECUTION_MODE.key(), "sync"); deltaTable = deltaTable.copy(expireOptions); - write(deltaTable, commitUser, "20250401", "v4"); + // A bounded Delta commit deterministically triggers ChainTablePartitionExpire. The two + // oldest snapshot anchors are expired and the latest expired-time anchor (day40) is kept. + // Dropping those Snapshot-branch partitions creates a new Snapshot-branch metadata + // snapshot, which must then be expired according to the same snapshot retention policy. + write(deltaTable, commitUser, day10, "v4"); snapshotTable = loadTable(tablePath).switchToBranch("snapshot"); + assertThat(listPartitions(snapshotTable)).containsExactly(day40); + assertThat(snapshotTable.snapshotManager().latestSnapshotId()) + .isGreaterThan(latestSnapshotBeforeExpiration); assertThat(snapshotTable.snapshotManager().snapshotCount()).isEqualTo(1); } @@ -150,4 +175,16 @@ private void write(FileStoreTable table, String commitUser, String dt, String va } write.close(); } + + private List listPartitions(FileStoreTable table) { + return table.newSnapshotReader().partitionEntries().stream() + .map(PartitionEntry::partition) + .map(partition -> partition.getString(0).toString()) + .sorted() + .collect(Collectors.toList()); + } + + private String partitionDate(int daysFromToday) { + return LocalDate.now().plusDays(daysFromToday).format(PARTITION_FORMATTER); + } } From 825941009a64ee93f36d5d2f437e14ed8bc641b7 Mon Sep 17 00:00:00 2001 From: big face cat <731030576@qq.com> Date: Sat, 5 Sep 2026 10:33:58 +0800 Subject: [PATCH 4/4] [core] Fix spotless formatting in chain table snapshot expiration test --- .../apache/paimon/table/ChainTableSnapshotExpireTest.java | 7 +++++-- 1 file changed, 5 insertions(+), 2 deletions(-) diff --git a/paimon-core/src/test/java/org/apache/paimon/table/ChainTableSnapshotExpireTest.java b/paimon-core/src/test/java/org/apache/paimon/table/ChainTableSnapshotExpireTest.java index 3f80c6781dc9..43e7a77bd51b 100644 --- a/paimon-core/src/test/java/org/apache/paimon/table/ChainTableSnapshotExpireTest.java +++ b/paimon-core/src/test/java/org/apache/paimon/table/ChainTableSnapshotExpireTest.java @@ -121,7 +121,9 @@ private void createChainTable(Path tablePath) throws Exception { new Schema( RowType.of( new org.apache.paimon.types.DataType[] { - DataTypes.STRING(), DataTypes.STRING(), DataTypes.STRING() + DataTypes.STRING(), + DataTypes.STRING(), + DataTypes.STRING() }, new String[] {"dt", "pk", "v"}) .getFields(), @@ -140,7 +142,8 @@ private void createChainTable(Path tablePath) throws Exception { SchemaChange.setOption(CoreOptions.CHAIN_TABLE_ENABLED.key(), "true"), SchemaChange.setOption( CoreOptions.SCAN_FALLBACK_SNAPSHOT_BRANCH.key(), "snapshot"), - SchemaChange.setOption(CoreOptions.SCAN_FALLBACK_DELTA_BRANCH.key(), "delta"), + SchemaChange.setOption( + CoreOptions.SCAN_FALLBACK_DELTA_BRANCH.key(), "delta"), SchemaChange.setOption( CoreOptions.PARTITION_TIMESTAMP_PATTERN.key(), "$dt"), SchemaChange.setOption(