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 e4aab21645 [core] Build expired partitions from their values instead
of a joined string (#9277)
e4aab21645 is described below
commit e4aab2164529e1baf1e71cd4c5238225d83811a0
Author: ZIHAN DAI <[email protected]>
AuthorDate: Thu Aug 20 12:02:47 2026 +1000
[core] Build expired partitions from their values instead of a joined
string (#9277)
---
.../paimon/operation/NormalPartitionExpire.java | 11 ++--
.../paimon/operation/PartitionExpireTest.java | 60 ++++++++++++++++++++++
2 files changed, 66 insertions(+), 5 deletions(-)
diff --git
a/paimon-core/src/main/java/org/apache/paimon/operation/NormalPartitionExpire.java
b/paimon-core/src/main/java/org/apache/paimon/operation/NormalPartitionExpire.java
index 5a23f538a9..29aa8e0437 100644
---
a/paimon-core/src/main/java/org/apache/paimon/operation/NormalPartitionExpire.java
+++
b/paimon-core/src/main/java/org/apache/paimon/operation/NormalPartitionExpire.java
@@ -37,6 +37,7 @@ import java.time.Duration;
import java.time.LocalDateTime;
import java.util.ArrayList;
import java.util.Collection;
+import java.util.Comparator;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
@@ -216,12 +217,12 @@ public class NormalPartitionExpire implements
PartitionExpire {
private List<Map<String, String>> convertToPartitionString(
List<List<String>> expiredPartValues) {
+ // Sort on the joined form, but build the partition from the original
values: a value
+ // may itself contain DELIMITER, and splitting the joined string back
apart would then
+ // shift every following field by one.
return expiredPartValues.stream()
- .map(values -> String.join(DELIMITER, values))
- .sorted()
- // Use split(DELIMITER, -1) to preserve trailing empty strings
- .map(s -> s.split(DELIMITER, -1))
- .map(strategy::toPartitionString)
+ .sorted(Comparator.comparing(values -> String.join(DELIMITER,
values)))
+ .map(values -> strategy.toPartitionString(values.toArray()))
.limit(Math.min(expiredPartValues.size(), maxExpireNum))
.collect(Collectors.toList());
}
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 e6abe773eb..f9dcf0d6d4 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
@@ -85,6 +85,7 @@ import static org.apache.paimon.CoreOptions.createCommitUser;
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;
/** Test for {@link NormalPartitionExpire}. */
public class PartitionExpireTest {
@@ -446,6 +447,65 @@ public class PartitionExpireTest {
+ " job to failover restart continuously.
Expired partitions are: [20230101]");
}
+ @Test
+ public void testExpirePartitionValueContainingTheDelimiter() throws
Exception {
+ SchemaManager schemaManager = new SchemaManager(LocalFileIO.create(),
path);
+ schemaManager.createTable(
+ new Schema(
+ RowType.of(VarCharType.STRING_TYPE,
VarCharType.STRING_TYPE).getFields(),
+ Arrays.asList("f0", "f1"),
+ emptyList(),
+ Collections.emptyMap(),
+ ""));
+ newTable();
+ // f1 contains the delimiter that the expired partitions are joined on
+ write("20230101", "us,ca");
+ write("20230105", "51");
+
+ NormalPartitionExpire expire = newExpire();
+ expire.setLastCheck(date(1));
+ List<Map<String, String>> expired = expire.expire(date(6),
Long.MAX_VALUE);
+
+ assertThat(expired).hasSize(1);
+ assertThat(expired.get(0)).containsExactly(entry("f0", "20230101"),
entry("f1", "us,ca"));
+ assertThat(read()).containsExactlyInAnyOrder("20230105:51");
+ }
+
+ @Test
+ public void
testExpireKeepsALivePartitionWhoseNeighbourContainsTheDelimiter() throws
Exception {
+ SchemaManager schemaManager = new SchemaManager(LocalFileIO.create(),
path);
+ schemaManager.createTable(
+ new Schema(
+ RowType.of(VarCharType.STRING_TYPE,
VarCharType.STRING_TYPE).getFields(),
+ singletonList("f0"),
+ emptyList(),
+ Collections.emptyMap(),
+ ""));
+ newTable();
+ Map<String, String> options = new HashMap<>();
+ options.put(CoreOptions.PARTITION_EXPIRATION_STRATEGY.key(),
"update-time");
+ options.put(PARTITION_EXPIRATION_TIME.key(), "1 s");
+ options.put(PARTITION_EXPIRATION_CHECK_INTERVAL.key(), "1 d");
+ table = table.copy(options);
+
+ // "us,ca" is old enough to expire; "us" is written afterwards and
must survive. Under
+ // update-time any partition value is accepted, so the delimiter
reaches the round trip.
+ write("us,ca", "old");
+ Thread.sleep(2000);
+ write("us", "fresh");
+ // pin the check time now, so a later stall cannot move the cut-off
past "us"
+ LocalDateTime checkTime = LocalDateTime.now();
+
+ NormalPartitionExpire expire =
+ (NormalPartitionExpire) table.store().newPartitionExpire("",
table);
+ expire.setLastCheck(checkTime.minusDays(2));
+ List<Map<String, String>> expired = expire.expire(checkTime,
Long.MAX_VALUE);
+
+ assertThat(expired).hasSize(1);
+ assertThat(expired.get(0)).containsExactly(entry("f0", "us,ca"));
+ assertThat(read()).containsExactlyInAnyOrder("us:fresh");
+ }
+
private List<String> read() throws IOException {
List<String> ret = new ArrayList<>();
table.newRead()
