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 ba7e1991da [core] Support retaining the latest expired partition
(#10766)
ba7e1991da is described below
commit ba7e1991da8845c0a25c70345c1608216226b667
Author: tzphh <[email protected]>
AuthorDate: Tue Oct 27 14:11:42 2026 +0800
[core] Support retaining the latest expired partition (#10766)
---
docs/docs/maintenance/manage-partitions.md | 23 +++
docs/generated/core_configuration.html | 6 +
.../main/java/org/apache/paimon/CoreOptions.java | 15 ++
.../paimon/partition/PartitionExpireStrategy.java | 20 ++
.../PartitionUpdateTimeExpireStrategy.java | 19 +-
.../PartitionValuesTimeExpireStrategy.java | 23 ++-
.../paimon/operation/PartitionExpireTest.java | 207 +++++++++++++++++++++
.../flink/action/ExpirePartitionsAction.java | 3 +
.../flink/action/ExpirePartitionsActionITCase.java | 58 ++++++
9 files changed, 369 insertions(+), 5 deletions(-)
diff --git a/docs/docs/maintenance/manage-partitions.md
b/docs/docs/maintenance/manage-partitions.md
index ab2bfc7922..9fe1172d18 100644
--- a/docs/docs/maintenance/manage-partitions.md
+++ b/docs/docs/maintenance/manage-partitions.md
@@ -100,6 +100,23 @@ insert into t values('pk', 'par-1');
```
+### Retaining the Latest Partition
+
+Set `'partition.expiration-retain-latest' = 'true'` (default: `false`) to
retain the latest
+partition when all partitions have expired. It remains available until a newer
partition
+is committed, after which the previous partition can expire normally.
+
+```sql
+CREATE TABLE t (...) PARTITIONED BY (dt) WITH (
+ 'partition.expiration-time' = '7 d',
+ 'partition.timestamp-formatter' = 'yyyyMMdd',
+ 'partition.expiration-retain-latest' = 'true'
+);
+```
+
+This option applies to normal tables. The latest partition is determined by
the parsed
+partition timestamp for `values-time`, or the last file creation time for
`update-time`.
+
More options:
<table class="table table-bordered">
@@ -135,6 +152,12 @@ More options:
<td>Duration</td>
<td>The expiration interval of a partition. A partition will be
expired if it's lifetime is over this value. Partition time is extracted from
the partition value.</td>
</tr>
+ <tr>
+ <td><h5>partition.expiration-retain-latest</h5></td>
+ <td style="word-wrap: break-word;">false</td>
+ <td>Boolean</td>
+ <td>Retain the latest partition when all partitions are expired.
Applies to one partition across the whole table with the built-in expiration
strategies on normal tables.</td>
+ </tr>
<tr>
<td><h5>partition.timestamp-formatter</h5></td>
<td style="word-wrap: break-word;">(none)</td>
diff --git a/docs/generated/core_configuration.html
b/docs/generated/core_configuration.html
index ef54534e46..0fc8c42fd2 100644
--- a/docs/generated/core_configuration.html
+++ b/docs/generated/core_configuration.html
@@ -1308,6 +1308,12 @@ For an internal format table in a REST catalog, it also
makes the catalog own th
<td>Integer</td>
<td>The default deleted num of partition expiration.</td>
</tr>
+ <tr>
+ <td><h5>partition.expiration-retain-latest</h5></td>
+ <td style="word-wrap: break-word;">false</td>
+ <td>Boolean</td>
+ <td>Whether to retain the latest partition when all partitions are
expired. The latest partition is determined by the parsed partition time for
values-time or the last file creation time for update-time. Retains one
partition for the whole table. Supported by the built-in expiration strategies
on normal tables. Explicit partition deletion and truncation are not
affected.</td>
+ </tr>
<tr>
<td><h5>partition.expiration-strategy</h5></td>
<td style="word-wrap: break-word;">"values-time"</td>
diff --git a/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java
b/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java
index 87fc141839..c3202188dc 100644
--- a/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java
+++ b/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java
@@ -1477,6 +1477,17 @@ public class CoreOptions implements Serializable {
.defaultValue(100)
.withDescription("The default deleted num of partition
expiration.");
+ public static final ConfigOption<Boolean>
PARTITION_EXPIRATION_RETAIN_LATEST =
+ key("partition.expiration-retain-latest")
+ .booleanType()
+ .defaultValue(false)
+ .withDescription(
+ "Whether to retain the latest partition when all
partitions are expired. "
+ + "The latest partition is determined by
the parsed partition time for values-time "
+ + "or the last file creation time for
update-time. Retains one partition for the whole table. "
+ + "Supported by the built-in expiration
strategies on normal tables. "
+ + "Explicit partition deletion and
truncation are not affected.");
+
public static final ConfigOption<Integer> PARTITION_EXPIRATION_BATCH_SIZE =
key("partition.expiration-batch-size")
.intType()
@@ -4331,6 +4342,10 @@ public class CoreOptions implements Serializable {
return options.get(PARTITION_EXPIRATION_MAX_NUM);
}
+ public boolean partitionExpireRetainLatest() {
+ return options.get(PARTITION_EXPIRATION_RETAIN_LATEST);
+ }
+
public int partitionExpireBatchSize() {
return options.getOptional(PARTITION_EXPIRATION_BATCH_SIZE)
.orElse(options.get(PARTITION_EXPIRATION_MAX_NUM));
diff --git
a/paimon-core/src/main/java/org/apache/paimon/partition/PartitionExpireStrategy.java
b/paimon-core/src/main/java/org/apache/paimon/partition/PartitionExpireStrategy.java
index 26ca00f2f7..fc00691c9a 100644
---
a/paimon-core/src/main/java/org/apache/paimon/partition/PartitionExpireStrategy.java
+++
b/paimon-core/src/main/java/org/apache/paimon/partition/PartitionExpireStrategy.java
@@ -28,12 +28,16 @@ import org.apache.paimon.types.DataTypeRoot;
import org.apache.paimon.types.RowType;
import org.apache.paimon.utils.RowDataToObjectArrayConverter;
+import
org.apache.paimon.shade.guava30.com.google.common.primitives.UnsignedBytes;
+
import javax.annotation.Nullable;
import java.io.Serializable;
import java.time.LocalDate;
import java.time.LocalDateTime;
import java.util.ArrayList;
+import java.util.Collections;
+import java.util.Comparator;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
@@ -89,6 +93,22 @@ public abstract class PartitionExpireStrategy implements
Serializable {
public abstract List<PartitionEntry> selectExpiredPartitions(
FileStoreScan scan, LocalDateTime expirationTime);
+ protected static void retainLatestPartition(
+ List<PartitionEntry> partitions,
+ List<PartitionEntry> expired,
+ Comparator<PartitionEntry> timeComparator) {
+ if (expired.isEmpty() || expired.size() != partitions.size()) {
+ return;
+ }
+
+ // Use the complete partition key to break ties independently of scan
order.
+ Comparator<PartitionEntry> comparator =
+ timeComparator.thenComparing(
+ entry -> entry.partition().toBytes(),
+ UnsignedBytes.lexicographicalComparator());
+ expired.remove(Collections.max(expired, comparator));
+ }
+
public static PartitionExpireStrategy createPartitionExpireStrategy(
CoreOptions options,
RowType partitionType,
diff --git
a/paimon-core/src/main/java/org/apache/paimon/partition/PartitionUpdateTimeExpireStrategy.java
b/paimon-core/src/main/java/org/apache/paimon/partition/PartitionUpdateTimeExpireStrategy.java
index 628f4c3733..6c41d90055 100644
---
a/paimon-core/src/main/java/org/apache/paimon/partition/PartitionUpdateTimeExpireStrategy.java
+++
b/paimon-core/src/main/java/org/apache/paimon/partition/PartitionUpdateTimeExpireStrategy.java
@@ -25,6 +25,7 @@ import org.apache.paimon.types.RowType;
import java.time.LocalDateTime;
import java.time.ZoneId;
+import java.util.Comparator;
import java.util.List;
import java.util.stream.Collectors;
@@ -36,17 +37,29 @@ public class PartitionUpdateTimeExpireStrategy extends
PartitionExpireStrategy {
private static final long serialVersionUID = 1L;
+ private final boolean retainLatest;
+
public PartitionUpdateTimeExpireStrategy(CoreOptions options, RowType
partitionType) {
super(partitionType, options.partitionDefaultName());
+ this.retainLatest = options.partitionExpireRetainLatest();
}
@Override
public List<PartitionEntry> selectExpiredPartitions(
FileStoreScan scan, LocalDateTime expirationTime) {
+ List<PartitionEntry> partitions = scan.readPartitionEntries();
long expirationMilli =
expirationTime.atZone(ZoneId.systemDefault()).toInstant().toEpochMilli();
- return scan.readPartitionEntries().stream()
- .filter(partitionEntry -> expirationMilli >
partitionEntry.lastFileCreationTime())
- .collect(Collectors.toList());
+ List<PartitionEntry> expired =
+ partitions.stream()
+ .filter(entry -> expirationMilli >
entry.lastFileCreationTime())
+ .collect(Collectors.toList());
+ if (retainLatest) {
+ retainLatestPartition(
+ partitions,
+ expired,
+
Comparator.comparingLong(PartitionEntry::lastFileCreationTime));
+ }
+ return expired;
}
}
diff --git
a/paimon-core/src/main/java/org/apache/paimon/partition/PartitionValuesTimeExpireStrategy.java
b/paimon-core/src/main/java/org/apache/paimon/partition/PartitionValuesTimeExpireStrategy.java
index 6238817da4..0291bcc6c9 100644
---
a/paimon-core/src/main/java/org/apache/paimon/partition/PartitionValuesTimeExpireStrategy.java
+++
b/paimon-core/src/main/java/org/apache/paimon/partition/PartitionValuesTimeExpireStrategy.java
@@ -32,6 +32,7 @@ import org.slf4j.LoggerFactory;
import java.time.LocalDateTime;
import java.time.format.DateTimeParseException;
import java.util.Arrays;
+import java.util.Comparator;
import java.util.List;
import java.util.Objects;
import java.util.stream.Collectors;
@@ -51,25 +52,43 @@ public class PartitionValuesTimeExpireStrategy extends
PartitionExpireStrategy {
LoggerFactory.getLogger(PartitionValuesTimeExpireStrategy.class);
private final PartitionTimeExtractor timeExtractor;
+ private final boolean retainLatest;
public PartitionValuesTimeExpireStrategy(CoreOptions options, RowType
partitionType) {
super(partitionType, options.partitionDefaultName());
String timePattern = options.partitionTimestampPattern();
String timeFormatter = options.partitionTimestampFormatter();
this.timeExtractor = new PartitionTimeExtractor(timePattern,
timeFormatter);
+ this.retainLatest = options.partitionExpireRetainLatest();
}
@Override
public List<PartitionEntry> selectExpiredPartitions(
FileStoreScan scan, LocalDateTime expirationTime) {
- return scan.withPartitionFilter(new
PartitionValuesTimePredicate(expirationTime))
- .readPartitionEntries();
+ PartitionValuesTimePredicate predicate = new
PartitionValuesTimePredicate(expirationTime);
+ if (!retainLatest) {
+ return scan.withPartitionFilter(predicate).readPartitionEntries();
+ }
+
+ List<PartitionEntry> partitions = scan.readPartitionEntries();
+ List<PartitionEntry> expired =
+ partitions.stream()
+ .filter(entry -> predicate.test(entry.partition()))
+ .collect(Collectors.toList());
+ retainLatestPartition(
+ partitions, expired,
Comparator.comparing(this::extractPartitionTime));
+ return expired;
}
public boolean isExpired(LocalDateTime expireDateTime, BinaryRow
partition) {
return new
PartitionValuesTimePredicate(expireDateTime).test(partition);
}
+ private LocalDateTime extractPartitionTime(PartitionEntry entry) {
+ return timeExtractor.extract(
+ partitionKeys,
Arrays.asList(convertPartition(entry.partition())));
+ }
+
/** The expired partition predicate uses the date-format value of the
partition. */
private class PartitionValuesTimePredicate implements PartitionPredicate {
diff --git
a/paimon-core/src/test/java/org/apache/paimon/operation/PartitionExpireTest.java
b/paimon-core/src/test/java/org/apache/paimon/operation/PartitionExpireTest.java
index 57dabaea4c..e0a2372287 100644
---
a/paimon-core/src/test/java/org/apache/paimon/operation/PartitionExpireTest.java
+++
b/paimon-core/src/test/java/org/apache/paimon/operation/PartitionExpireTest.java
@@ -30,8 +30,11 @@ import org.apache.paimon.fs.local.LocalFileIO;
import org.apache.paimon.io.CompactIncrement;
import org.apache.paimon.io.DataFileMeta;
import org.apache.paimon.io.DataIncrement;
+import org.apache.paimon.manifest.PartitionEntry;
import org.apache.paimon.options.Options;
+import org.apache.paimon.partition.PartitionExpireStrategy;
import org.apache.paimon.partition.PartitionStatistics;
+import org.apache.paimon.partition.PartitionUpdateTimeExpireStrategy;
import org.apache.paimon.partition.PartitionValuesTimeExpireStrategy;
import org.apache.paimon.partition.actions.AddDonePartitionAction;
import org.apache.paimon.schema.FileSystemSchemaManager;
@@ -61,12 +64,14 @@ import org.junit.jupiter.api.BeforeEach;
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.CsvSource;
import org.junit.jupiter.params.provider.ValueSource;
import java.io.IOException;
import java.time.LocalDate;
import java.time.LocalDateTime;
import java.time.LocalTime;
+import java.time.ZoneId;
import java.time.format.DateTimeFormatter;
import java.time.temporal.ChronoUnit;
import java.util.ArrayList;
@@ -85,6 +90,7 @@ import static java.util.Collections.emptyList;
import static java.util.Collections.singletonList;
import static org.apache.paimon.CoreOptions.METASTORE_PARTITIONED_TABLE;
import static
org.apache.paimon.CoreOptions.PARTITION_EXPIRATION_CHECK_INTERVAL;
+import static org.apache.paimon.CoreOptions.PARTITION_EXPIRATION_RETAIN_LATEST;
import static org.apache.paimon.CoreOptions.PARTITION_EXPIRATION_TIME;
import static org.apache.paimon.CoreOptions.PARTITION_TIMESTAMP_FORMATTER;
import static org.apache.paimon.CoreOptions.PATH;
@@ -94,6 +100,8 @@ import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatCode;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
import static org.assertj.core.api.Assertions.entry;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
/** Test for {@link NormalPartitionExpire}. */
public class PartitionExpireTest {
@@ -694,6 +702,205 @@ public class PartitionExpireTest {
assertThat(read()).containsExactlyInAnyOrder("us:fresh");
}
+ @ParameterizedTest
+ @CsvSource({"false,false", "true,false", "false,true", "true,true"})
+ public void testRetainLatestPartitionLifecycle(boolean primaryKey, boolean
metastore)
+ throws Exception {
+ newRetainLatestTable(primaryKey);
+ table =
+ table.copy(
+ Collections.singletonMap(
+ METASTORE_PARTITIONED_TABLE.key(),
String.valueOf(metastore)));
+ NormalPartitionExpire expire =
+ (NormalPartitionExpire)
table.store().newPartitionExpire("expire", table);
+ assertThat(expire.expire(date(10), Long.MAX_VALUE)).isEmpty();
+
+ write("20230101", "old");
+ assertThat(expire.expire(date(10), Long.MAX_VALUE)).isEmpty();
+ write("20230103", "latest");
+ if (metastore) {
+ AddDonePartitionAction doneAction =
+ new
AddDonePartitionAction(table.catalogEnvironment().partitionModification());
+ doneAction.markDone("f0=20230101");
+ doneAction.markDone("f0=20230103");
+ }
+ assertThat(expire.expire(date(10), Long.MAX_VALUE))
+ .containsExactly(Collections.singletonMap("f0", "20230101"));
+ assertThat(read()).containsExactly("20230103:latest");
+ long snapshotId = table.snapshotManager().latestSnapshotId();
+ assertThat(expire.expire(date(20), Long.MAX_VALUE)).isEmpty();
+
assertThat(table.snapshotManager().latestSnapshotId()).isEqualTo(snapshotId);
+ if (metastore) {
+ assertThat(createdPartitions)
+ .containsExactlyInAnyOrder(
+ Collections.singletonMap("f0", "20230103"),
+ Collections.singletonMap("f0", "20230103.done"));
+ assertThat(deletedPartitions)
+ .containsExactlyInAnyOrder(
+ Collections.singletonMap("f0", "20230101"),
+ Collections.singletonMap("f0", "20230101.done"));
+ }
+
+ // Prepared files are not a replacement until their commit becomes
visible.
+ try (StreamTableWrite write = table.newWrite("replacement");
+ TableCommitImpl commit = table.newCommit("replacement")) {
+ write.write(
+ GenericRow.of(
+ BinaryString.fromString("20230121"),
BinaryString.fromString("new")));
+ List<CommitMessage> messages = write.prepareCommit(true, 0);
+ assertThat(expire.expire(date(22), Long.MAX_VALUE)).isEmpty();
+ assertThat(read()).containsExactly("20230103:latest");
+ commit.commit(0, messages);
+ }
+ assertThat(expire.expire(date(22), Long.MAX_VALUE))
+ .containsExactly(Collections.singletonMap("f0", "20230103"));
+ assertThat(expire.expire(date(30), Long.MAX_VALUE)).isEmpty();
+ assertThat(read()).containsExactly("20230121:new");
+ if (metastore) {
+ assertThat(createdPartitions)
+ .containsExactly(Collections.singletonMap("f0",
"20230121"));
+ }
+ }
+
+ @ParameterizedTest
+ @CsvSource({"1,1", "100,1", "100,0"})
+ public void testRetainLatestWithLimitsAndBatches(int limit, int batchSize)
throws Exception {
+ newRetainLatestTable(false);
+ Map<String, String> options = new HashMap<>();
+ options.put(CoreOptions.PARTITION_EXPIRATION_MAX_NUM.key(),
String.valueOf(limit));
+ options.put(CoreOptions.PARTITION_EXPIRATION_BATCH_SIZE.key(),
String.valueOf(batchSize));
+ options.put(PARTITION_TIMESTAMP_FORMATTER.key(), "ddMMyyyy");
+ table = table.copy(options);
+ write("12012023", "old");
+ write("15012023", "old");
+ // Lexical order differs from timestamp order.
+ write("02022023", "latest");
+ NormalPartitionExpire expire =
+ (NormalPartitionExpire)
table.store().newPartitionExpire("expire", table);
+ for (int i = 0; i < 4; i++) {
+ assertThat(expire.expire(LocalDateTime.of(2023, 3, 1, 0, 0),
Long.MAX_VALUE))
+ .hasSizeLessThanOrEqualTo(limit);
+ }
+ assertThat(read()).containsExactly("02022023:latest");
+ }
+
+ @Test
+ public void
testRetainLatestDoesNotKeepExtraPartitionWhenUnparseablePartitionRemains()
+ throws Exception {
+ newRetainLatestTable(false);
+ write("20230101", "old");
+ write("invalid", "retained");
+ NormalPartitionExpire expire =
+ (NormalPartitionExpire)
table.store().newPartitionExpire("expire", table);
+ assertThat(expire.expire(date(10), Long.MAX_VALUE)).hasSize(1);
+ assertThat(read()).containsExactly("invalid:retained");
+ }
+
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void testWriterTriggeredExpirationRetainsLatest(boolean primaryKey)
throws Exception {
+ newRetainLatestTable(primaryKey);
+ table = table.copy(Collections.singletonMap(WRITE_ONLY.key(),
"false"));
+ write("20230101", "old");
+ assertThat(read()).containsExactly("20230101:old");
+ write("20230103", "latest");
+ assertThat(read()).containsExactly("20230103:latest");
+ }
+
+ @Test
+ public void testRetainLatestDoesNotPreventExplicitDeletion() throws
Exception {
+ newRetainLatestTable(false);
+ write("20230101", "old");
+ try (FileStoreCommit commit = table.store().newCommit("drop", table)) {
+ commit.dropPartitions(singletonList(Collections.singletonMap("f0",
"20230101")), 0);
+ assertThat(read()).isEmpty();
+ write("20230103", "latest");
+ commit.truncateTable(1);
+ assertThat(read()).isEmpty();
+ }
+ }
+
+ @ParameterizedTest
+ @ValueSource(strings = {"values-time", "update-time"})
+ public void testRetainLatestOrderingAndStableTies(String
expirationStrategy) {
+ RowType type = RowType.of(DataTypes.DATE(), DataTypes.STRING());
+ Options options = new Options();
+ options.set(PARTITION_EXPIRATION_RETAIN_LATEST, true);
+ boolean updateTime = "update-time".equals(expirationStrategy);
+ PartitionExpireStrategy strategy =
+ updateTime
+ ? new PartitionUpdateTimeExpireStrategy(new
CoreOptions(options), type)
+ : new PartitionValuesTimeExpireStrategy(new
CoreOptions(options), type);
+ InternalRowSerializer serializer = new InternalRowSerializer(type);
+ // For update-time, the latest partition may have an older partition
timestamp.
+ PartitionEntry old = partitionEntry(serializer, updateTime ? 5 : 1,
"z", 1);
+ PartitionEntry recent = partitionEntry(serializer, 3, "a", 3);
+ PartitionEntry tie = partitionEntry(serializer, 3, "b", 3);
+ FileStoreScan scan = mock(FileStoreScan.class);
+ when(scan.readPartitionEntries()).thenReturn(Arrays.asList(old,
recent, tie));
+ List<PartitionEntry> expired = strategy.selectExpiredPartitions(scan,
date(10));
+ assertThat(expired).hasSize(2).contains(old);
+ when(scan.readPartitionEntries()).thenReturn(Arrays.asList(tie,
recent, old));
+ assertThat(strategy.selectExpiredPartitions(scan, date(10)))
+ .containsExactlyInAnyOrderElementsOf(expired);
+ when(scan.readPartitionEntries()).thenReturn(singletonList(recent));
+ assertThat(strategy.selectExpiredPartitions(scan, date(10))).isEmpty();
+
+ // Any live partition means there is no need to protect an expired one.
+ PartitionEntry live =
+ partitionEntry(
+ serializer,
+ 20,
+ "live",
+
date(20).atZone(ZoneId.systemDefault()).toInstant().toEpochMilli());
+ when(scan.readPartitionEntries()).thenReturn(Arrays.asList(old,
recent, live));
+ assertThat(strategy.selectExpiredPartitions(scan,
date(10))).containsExactly(old, recent);
+ if (!updateTime) {
+ PartitionEntry nullTime = partitionEntry(serializer, null, "null",
1);
+ when(scan.readPartitionEntries()).thenReturn(Arrays.asList(old,
recent, nullTime));
+ assertThat(strategy.selectExpiredPartitions(scan, date(10)))
+ .containsExactly(old, recent);
+ }
+ when(scan.readPartitionEntries()).thenReturn(emptyList());
+ assertThat(strategy.selectExpiredPartitions(scan, date(10))).isEmpty();
+ }
+
+ private PartitionEntry partitionEntry(
+ InternalRowSerializer serializer, Integer day, String group, long
fileCreationTime) {
+ BinaryRow partition =
+ serializer
+ .toBinaryRow(
+ GenericRow.of(
+ day == null
+ ? null
+ : (int)
date(day).toLocalDate().toEpochDay(),
+ BinaryString.fromString(group)))
+ .copy();
+ return new PartitionEntry(partition, 1, 1, 1, fileCreationTime, 1);
+ }
+
+ private void newRetainLatestTable(boolean primaryKey) throws Exception {
+ Map<String, String> options = new HashMap<>();
+ options.put(PARTITION_EXPIRATION_RETAIN_LATEST.key(), "true");
+ options.put(PARTITION_EXPIRATION_TIME.key(), "2 d");
+ options.put(PARTITION_EXPIRATION_CHECK_INTERVAL.key(), "0 s");
+ options.put(PARTITION_TIMESTAMP_FORMATTER.key(), "yyyyMMdd");
+ options.put(WRITE_ONLY.key(), "true");
+ if (primaryKey) {
+ options.put(CoreOptions.BUCKET.key(), "1");
+ }
+ new FileSystemSchemaManager(LocalFileIO.create(), path)
+ .createTable(
+ new Schema(
+ RowType.of(VarCharType.STRING_TYPE,
VarCharType.STRING_TYPE)
+ .getFields(),
+ singletonList("f0"),
+ primaryKey ? Arrays.asList("f0", "f1") :
emptyList(),
+ options,
+ ""));
+ newTable();
+ }
+
private List<String> read() throws IOException {
List<String> ret = new ArrayList<>();
table.newRead()
diff --git
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/ExpirePartitionsAction.java
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/ExpirePartitionsAction.java
index 7efb65d2d6..3dfd15e45f 100644
---
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/ExpirePartitionsAction.java
+++
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/ExpirePartitionsAction.java
@@ -59,6 +59,9 @@ public class ExpirePartitionsAction extends TableActionBase
implements LocalActi
map.put(CoreOptions.PARTITION_EXPIRATION_STRATEGY.key(),
expireStrategy);
map.put(CoreOptions.PARTITION_TIMESTAMP_FORMATTER.key(),
timestampFormatter);
map.put(CoreOptions.PARTITION_TIMESTAMP_PATTERN.key(),
timestampPattern);
+ map.put(
+ CoreOptions.PARTITION_EXPIRATION_RETAIN_LATEST.key(),
+
table.options().get(CoreOptions.PARTITION_EXPIRATION_RETAIN_LATEST.key()));
}
@Override
diff --git
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/action/ExpirePartitionsActionITCase.java
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/action/ExpirePartitionsActionITCase.java
index 130d23d466..1a528c8d1f 100644
---
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/action/ExpirePartitionsActionITCase.java
+++
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/action/ExpirePartitionsActionITCase.java
@@ -31,12 +31,15 @@ import org.apache.paimon.utils.SnapshotManager;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.CsvSource;
import org.junit.jupiter.params.provider.ValueSource;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
+import java.util.HashMap;
import java.util.List;
+import java.util.Map;
import static org.apache.paimon.flink.util.ReadWriteTableTestUtil.init;
import static org.assertj.core.api.Assertions.assertThat;
@@ -190,6 +193,61 @@ public class ExpirePartitionsActionITCase extends
ActionITCaseBase {
assertThat(actual).isEqualTo(expected);
}
+ @ParameterizedTest
+ @CsvSource({"true,,true", ",true,true", "true,false,false", ",,false"})
+ public void testRetainLatestPartition(
+ String tableOption, String actionOption, boolean retainLatest)
throws Exception {
+ Map<String, String> options = new HashMap<>();
+ if (tableOption != null) {
+ options.put("partition.expiration-retain-latest", tableOption);
+ }
+ // Explicit action arguments must still override the table's strategy
and formatter.
+ options.put("partition.expiration-strategy", "update-time");
+ options.put("partition.timestamp-formatter", "yyyy-MM-dd");
+ FileStoreTable table =
+ createFileStoreTable(
+ ROW_TYPE,
+ Collections.singletonList("dt"),
+ Arrays.asList("k", "dt"),
+ Collections.singletonList("k"),
+ options);
+ StreamWriteBuilder writeBuilder =
table.newStreamWriteBuilder().withCommitUser(commitUser);
+ write = writeBuilder.newWrite();
+ commit = writeBuilder.newCommit();
+ writeData(
+ rowData(
+ BinaryString.fromString("1"),
+ BinaryString.fromString("20010101"),
+ BinaryString.fromString("01:00")));
+ assertThat(getData(tableName)).hasSize(1);
+
+ List<String> args =
+ new ArrayList<>(
+ Arrays.asList(
+ "expire_partitions",
+ "--warehouse",
+ warehouse,
+ "--database",
+ database,
+ "--table",
+ tableName,
+ "--expiration_time",
+ "1 d",
+ "--timestamp_formatter",
+ "yyyyMMdd",
+ "--expire_strategy",
+ "values-time"));
+ if (actionOption != null) {
+ args.add("--table_conf");
+ args.add("partition.expiration-retain-latest=" + actionOption);
+ }
+ ExpirePartitionsAction action =
createAction(ExpirePartitionsAction.class, args);
+ for (int i = 0; i < 2; i++) {
+ action.executeLocally();
+ assertThat(getData(tableName)).hasSize(retainLatest ? 1 : 0);
+ }
+ }
+
private FileStoreTable prepareTable() throws Exception {
init(warehouse);
