This is an automated email from the ASF dual-hosted git repository.
lzljs3620320 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 69e8e230d [core] Sort the expired partitions and add max_expires param
(#4266)
69e8e230d is described below
commit 69e8e230d77f3e1b9fc9fe8cf58426f0c7a84ffa
Author: askwang <[email protected]>
AuthorDate: Tue Oct 8 17:28:42 2024 +0800
[core] Sort the expired partitions and add max_expires param (#4266)
---
.../apache/paimon/operation/PartitionExpire.java | 37 +++++++++--
.../paimon/partition/PartitionExpireStrategy.java | 11 +++-
.../flink/procedure/ExpirePartitionsProcedure.java | 56 ++++++++--------
.../ProcedurePositionalArgumentsITCase.java | 5 +-
.../flink/procedure/ExpirePartitionsProcedure.java | 10 ++-
.../procedure/ExpirePartitionsProcedureITCase.java | 55 ++++++++++++++++
.../spark/procedure/ExpirePartitionsProcedure.java | 8 ++-
.../procedure/ExpirePartitionsProcedureTest.scala | 77 ++++++++++++++++++++++
8 files changed, 223 insertions(+), 36 deletions(-)
diff --git
a/paimon-core/src/main/java/org/apache/paimon/operation/PartitionExpire.java
b/paimon-core/src/main/java/org/apache/paimon/operation/PartitionExpire.java
index 64074a4c1..62a9b7964 100644
--- a/paimon-core/src/main/java/org/apache/paimon/operation/PartitionExpire.java
+++ b/paimon-core/src/main/java/org/apache/paimon/operation/PartitionExpire.java
@@ -37,12 +37,15 @@ import java.util.Collection;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
+import java.util.stream.Collectors;
/** Expire partitions. */
public class PartitionExpire {
private static final Logger LOG =
LoggerFactory.getLogger(PartitionExpire.class);
+ private static final String DELIMITER = ",";
+
private final Duration expirationTime;
private final Duration checkInterval;
private final FileStoreScan scan;
@@ -51,6 +54,7 @@ public class PartitionExpire {
private LocalDateTime lastCheck;
private final PartitionExpireStrategy strategy;
private final boolean endInputCheckPartitionExpire;
+ private int maxExpires;
public PartitionExpire(
Duration expirationTime,
@@ -68,6 +72,7 @@ public class PartitionExpire {
this.metastoreClient = metastoreClient;
this.lastCheck = LocalDateTime.now();
this.endInputCheckPartitionExpire = endInputCheckPartitionExpire;
+ this.maxExpires = Integer.MAX_VALUE;
}
public PartitionExpire(
@@ -85,6 +90,11 @@ public class PartitionExpire {
return this;
}
+ public PartitionExpire withMaxExpires(int maxExpires) {
+ this.maxExpires = maxExpires;
+ return this;
+ }
+
public List<Map<String, String>> expire(long commitIdentifier) {
return expire(LocalDateTime.now(), commitIdentifier);
}
@@ -125,14 +135,18 @@ public class PartitionExpire {
private List<Map<String, String>> doExpire(
LocalDateTime expireDateTime, long commitIdentifier) {
- List<Map<String, String>> expired = new ArrayList<>();
- for (PartitionEntry partition : strategy.selectExpiredPartitions(scan,
expireDateTime)) {
+ List<PartitionEntry> partitionEntries =
+ strategy.selectExpiredPartitions(scan, expireDateTime);
+ List<List<String>> expiredPartValues = new
ArrayList<>(partitionEntries.size());
+ for (PartitionEntry partition : partitionEntries) {
Object[] array = strategy.convertPartition(partition.partition());
- Map<String, String> partString = strategy.toPartitionString(array);
- expired.add(partString);
- LOG.info("Expire Partition: {}", partString);
+ expiredPartValues.add(strategy.toPartitionValue(array));
}
- if (!expired.isEmpty()) {
+
+ List<Map<String, String>> expired = new ArrayList<>();
+ if (!expiredPartValues.isEmpty()) {
+ expired = convertToPartitionString(expiredPartValues);
+ LOG.info("Expire Partitions: {}", expired);
if (metastoreClient != null) {
deleteMetastorePartitions(expired);
}
@@ -153,4 +167,15 @@ public class PartitionExpire {
});
}
}
+
+ private List<Map<String, String>> convertToPartitionString(
+ List<List<String>> expiredPartValues) {
+ return expiredPartValues.stream()
+ .map(values -> String.join(DELIMITER, values))
+ .sorted()
+ .map(s -> s.split(DELIMITER))
+ .map(strategy::toPartitionString)
+ .limit(Math.min(expiredPartValues.size(), maxExpires))
+ .collect(Collectors.toList());
+ }
}
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 529d81510..2def5b43a 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
@@ -26,6 +26,7 @@ import org.apache.paimon.types.RowType;
import org.apache.paimon.utils.RowDataToObjectArrayConverter;
import java.time.LocalDateTime;
+import java.util.ArrayList;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
@@ -42,13 +43,21 @@ public abstract class PartitionExpireStrategy {
}
public Map<String, String> toPartitionString(Object[] array) {
- Map<String, String> map = new LinkedHashMap<>();
+ Map<String, String> map = new LinkedHashMap<>(partitionKeys.size());
for (int i = 0; i < partitionKeys.size(); i++) {
map.put(partitionKeys.get(i), array[i].toString());
}
return map;
}
+ public List<String> toPartitionValue(Object[] array) {
+ List<String> list = new ArrayList<>(partitionKeys.size());
+ for (int i = 0; i < partitionKeys.size(); i++) {
+ list.add(array[i].toString());
+ }
+ return list;
+ }
+
public Object[] convertPartition(BinaryRow partition) {
return toObjectArrayConverter.convert(partition);
}
diff --git
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/procedure/ExpirePartitionsProcedure.java
b/paimon-flink/paimon-flink-1.18/src/main/java/org/apache/paimon/flink/procedure/ExpirePartitionsProcedure.java
similarity index 72%
copy from
paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/procedure/ExpirePartitionsProcedure.java
copy to
paimon-flink/paimon-flink-1.18/src/main/java/org/apache/paimon/flink/procedure/ExpirePartitionsProcedure.java
index abbf4f486..c0e5a65c4 100644
---
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/procedure/ExpirePartitionsProcedure.java
+++
b/paimon-flink/paimon-flink-1.18/src/main/java/org/apache/paimon/flink/procedure/ExpirePartitionsProcedure.java
@@ -26,11 +26,7 @@ import org.apache.paimon.operation.PartitionExpire;
import org.apache.paimon.table.FileStoreTable;
import org.apache.paimon.utils.TimeUtils;
-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;
import java.time.Duration;
import java.util.HashMap;
@@ -42,29 +38,15 @@ import static
org.apache.paimon.partition.PartitionExpireStrategy.createPartitio
/** A procedure to expire partitions. */
public class ExpirePartitionsProcedure extends ProcedureBase {
+
+ public static final String IDENTIFIER = "expire_partitions";
+
@Override
public String identifier() {
- return "expire_partitions";
+ return IDENTIFIER;
}
- @ProcedureHint(
- argument = {
- @ArgumentHint(name = "table", type = @DataTypeHint("STRING")),
- @ArgumentHint(name = "expiration_time", type =
@DataTypeHint(value = "STRING")),
- @ArgumentHint(
- name = "timestamp_formatter",
- type = @DataTypeHint("STRING"),
- isOptional = true),
- @ArgumentHint(
- name = "timestamp_pattern",
- type = @DataTypeHint("STRING"),
- isOptional = true),
- @ArgumentHint(
- name = "expire_strategy",
- type = @DataTypeHint("STRING"),
- isOptional = true),
- })
- public @DataTypeHint("ROW< expired_partitions STRING>") Row[] call(
+ public String[] call(
ProcedureContext procedureContext,
String tableId,
String expirationTime,
@@ -72,6 +54,25 @@ public class ExpirePartitionsProcedure extends ProcedureBase
{
String timestampPattern,
String expireStrategy)
throws Catalog.TableNotExistException {
+ return call(
+ procedureContext,
+ tableId,
+ expirationTime,
+ timestampFormatter,
+ timestampPattern,
+ expireStrategy,
+ null);
+ }
+
+ public String[] call(
+ ProcedureContext procedureContext,
+ String tableId,
+ String expirationTime,
+ String timestampFormatter,
+ String timestampPattern,
+ String expireStrategy,
+ Integer maxExpires)
+ throws Catalog.TableNotExistException {
FileStoreTable fileStoreTable = (FileStoreTable) table(tableId);
FileStore fileStore = fileStoreTable.store();
Map<String, String> map = new HashMap<>();
@@ -93,15 +94,18 @@ public class ExpirePartitionsProcedure extends
ProcedureBase {
.metastoreClientFactory())
.map(MetastoreClient.Factory::create)
.orElse(null));
+ if (maxExpires != null) {
+ partitionExpire.withMaxExpires(maxExpires);
+ }
List<Map<String, String>> expired =
partitionExpire.expire(Long.MAX_VALUE);
return expired == null || expired.isEmpty()
- ? new Row[] {Row.of("No expired partitions.")}
+ ? new String[] {"No expired partitions."}
: expired.stream()
.map(
x -> {
String r = x.toString();
- return Row.of(r.substring(1, r.length() -
1));
+ return r.substring(1, r.length() - 1);
})
- .toArray(Row[]::new);
+ .toArray(String[]::new);
}
}
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 da2eaef0b..a48de667b 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
@@ -244,7 +244,10 @@ public class ProcedurePositionalArgumentsITCase extends
CatalogITCaseBase {
sql("INSERT INTO T VALUES ('1', '2024-06-01')");
sql("INSERT INTO T VALUES ('2', '9024-06-01')");
assertThat(read(table)).containsExactlyInAnyOrder("1:2024-06-01",
"2:9024-06-01");
- sql("CALL sys.expire_partitions('default.T', '1 d', 'yyyy-MM-dd',
'$dt', 'values-time')");
+ assertThat(
+ sql(
+ "CALL sys.expire_partitions('default.T', '1
d', 'yyyy-MM-dd', '$dt', 'values-time')"))
+ .containsExactly(Row.of("dt=2024-06-01"));
assertThat(read(table)).containsExactlyInAnyOrder("2:9024-06-01");
}
diff --git
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/procedure/ExpirePartitionsProcedure.java
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/procedure/ExpirePartitionsProcedure.java
index abbf4f486..ee6075a92 100644
---
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/procedure/ExpirePartitionsProcedure.java
+++
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/procedure/ExpirePartitionsProcedure.java
@@ -63,6 +63,10 @@ public class ExpirePartitionsProcedure extends ProcedureBase
{
name = "expire_strategy",
type = @DataTypeHint("STRING"),
isOptional = true),
+ @ArgumentHint(
+ name = "max_expires",
+ type = @DataTypeHint("INTEGER"),
+ isOptional = true)
})
public @DataTypeHint("ROW< expired_partitions STRING>") Row[] call(
ProcedureContext procedureContext,
@@ -70,7 +74,8 @@ public class ExpirePartitionsProcedure extends ProcedureBase {
String expirationTime,
String timestampFormatter,
String timestampPattern,
- String expireStrategy)
+ String expireStrategy,
+ Integer maxExpires)
throws Catalog.TableNotExistException {
FileStoreTable fileStoreTable = (FileStoreTable) table(tableId);
FileStore fileStore = fileStoreTable.store();
@@ -93,6 +98,9 @@ public class ExpirePartitionsProcedure extends ProcedureBase {
.metastoreClientFactory())
.map(MetastoreClient.Factory::create)
.orElse(null));
+ if (maxExpires != null) {
+ partitionExpire.withMaxExpires(maxExpires);
+ }
List<Map<String, String>> expired =
partitionExpire.expire(Long.MAX_VALUE);
return expired == null || expired.isEmpty()
? new Row[] {Row.of("No expired partitions.")}
diff --git
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/procedure/ExpirePartitionsProcedureITCase.java
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/procedure/ExpirePartitionsProcedureITCase.java
index 71a6dc466..bc2e84902 100644
---
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/procedure/ExpirePartitionsProcedureITCase.java
+++
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/procedure/ExpirePartitionsProcedureITCase.java
@@ -347,6 +347,61 @@ public class ExpirePartitionsProcedureITCase extends
CatalogITCaseBase {
.containsExactlyInAnyOrder("Never-expire:9999-09-09:99:99");
}
+ @Test
+ public void testSortAndLimitExpirePartition() throws Exception {
+ sql(
+ "CREATE TABLE T ("
+ + " k STRING,"
+ + " dt STRING,"
+ + " hm STRING,"
+ + " PRIMARY KEY (k, dt, hm) NOT ENFORCED"
+ + ") PARTITIONED BY (dt, hm) WITH ("
+ + " 'bucket' = '1'"
+ + ")");
+ FileStoreTable table = paimonTable("T");
+ // Test there are no expired partitions.
+ assertThat(
+ callExpirePartitions(
+ "CALL sys.expire_partitions("
+ + "`table` => 'default.T'"
+ + ", expiration_time => '1 d'"
+ + ", timestamp_formatter =>
'yyyy-MM-dd')"))
+ .containsExactlyInAnyOrder("No expired partitions.");
+
+ sql("INSERT INTO T VALUES ('3', '2024-06-02', '02:00')");
+ sql("INSERT INTO T VALUES ('2', '2024-06-02', '01:00')");
+ sql("INSERT INTO T VALUES ('4', '2024-06-03', '01:00')");
+ sql("INSERT INTO T VALUES ('1', '2024-06-01', '01:00')");
+ // This partition never expires.
+ sql("INSERT INTO T VALUES ('Never-expire', '9999-09-09', '99:99')");
+
+ Function<InternalRow, String> consumerReadResult =
+ (InternalRow row) ->
+ row.getString(0) + ":" + row.getString(1) + ":" +
row.getString(2);
+ assertThat(read(table, consumerReadResult))
+ .containsExactlyInAnyOrder(
+ "1:2024-06-01:01:00",
+ "2:2024-06-02:01:00",
+ "3:2024-06-02:02:00",
+ "4:2024-06-03:01:00",
+ "Never-expire:9999-09-09:99:99");
+
+ // Show a list of expired partitions.
+ assertThat(
+ callExpirePartitions(
+ "CALL sys.expire_partitions("
+ + "`table` => 'default.T'"
+ + ", expiration_time => '1 d'"
+ + ", timestamp_formatter =>
'yyyy-MM-dd', max_expires => 3)"))
+ .containsExactly(
+ "dt=2024-06-01, hm=01:00",
+ "dt=2024-06-02, hm=01:00",
+ "dt=2024-06-02, hm=02:00");
+
+ assertThat(read(table, consumerReadResult))
+ .containsExactlyInAnyOrder("4:2024-06-03:01:00",
"Never-expire:9999-09-09:99:99");
+ }
+
/** Return a list of expired partitions. */
public List<String> callExpirePartitions(String callSql) {
return sql(callSql).stream()
diff --git
a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/ExpirePartitionsProcedure.java
b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/ExpirePartitionsProcedure.java
index 81012abc9..7b388227e 100644
---
a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/ExpirePartitionsProcedure.java
+++
b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/ExpirePartitionsProcedure.java
@@ -40,6 +40,7 @@ import java.util.Map;
import java.util.Optional;
import static
org.apache.paimon.partition.PartitionExpireStrategy.createPartitionExpireStrategy;
+import static org.apache.spark.sql.types.DataTypes.IntegerType;
import static org.apache.spark.sql.types.DataTypes.StringType;
/** A procedure to expire partitions. */
@@ -51,7 +52,8 @@ public class ExpirePartitionsProcedure extends BaseProcedure {
ProcedureParameter.required("expiration_time", StringType),
ProcedureParameter.optional("timestamp_formatter", StringType),
ProcedureParameter.optional("timestamp_pattern", StringType),
- ProcedureParameter.optional("expire_strategy", StringType)
+ ProcedureParameter.optional("expire_strategy", StringType),
+ ProcedureParameter.optional("max_expires", IntegerType)
};
private static final StructType OUTPUT_TYPE =
@@ -81,6 +83,7 @@ public class ExpirePartitionsProcedure extends BaseProcedure {
String timestampFormatter = args.isNullAt(2) ? null :
args.getString(2);
String timestampPattern = args.isNullAt(3) ? null : args.getString(3);
String expireStrategy = args.isNullAt(4) ? null : args.getString(4);
+ Integer maxExpires = args.isNullAt(5) ? null : args.getInt(5);
return modifyPaimonTable(
tableIdent,
table -> {
@@ -105,6 +108,9 @@ public class ExpirePartitionsProcedure extends
BaseProcedure {
.metastoreClientFactory())
.map(MetastoreClient.Factory::create)
.orElse(null));
+ if (maxExpires != null) {
+ partitionExpire.withMaxExpires(maxExpires);
+ }
List<Map<String, String>> expired =
partitionExpire.expire(Long.MAX_VALUE);
return expired == null || expired.isEmpty()
? new InternalRow[] {
diff --git
a/paimon-spark/paimon-spark-common/src/test/scala/org/apache/paimon/spark/procedure/ExpirePartitionsProcedureTest.scala
b/paimon-spark/paimon-spark-common/src/test/scala/org/apache/paimon/spark/procedure/ExpirePartitionsProcedureTest.scala
index db4696047..4561e532f 100644
---
a/paimon-spark/paimon-spark-common/src/test/scala/org/apache/paimon/spark/procedure/ExpirePartitionsProcedureTest.scala
+++
b/paimon-spark/paimon-spark-common/src/test/scala/org/apache/paimon/spark/procedure/ExpirePartitionsProcedureTest.scala
@@ -474,4 +474,81 @@ class ExpirePartitionsProcedureTest extends
PaimonSparkTestBase with StreamTest
}
}
}
+
+ test("Paimon procedure : sorted the expired partitions with max_expires.") {
+ failAfter(streamingTimeout) {
+ withTempDir {
+ checkpointDir =>
+ spark.sql(s"""
+ |CREATE TABLE T (k STRING, pt STRING, hm STRING)
+ |TBLPROPERTIES ('primary-key'='k,pt,hm', 'bucket'='1')
+ | PARTITIONED BY (pt,hm)
+ |""".stripMargin)
+ val location = loadTable("T").location().toString
+
+ val inputData = MemoryStream[(String, String, String)]
+ val stream = inputData
+ .toDS()
+ .toDF("k", "pt", "hm")
+ .writeStream
+ .option("checkpointLocation", checkpointDir.getCanonicalPath)
+ .foreachBatch {
+ (batch: Dataset[Row], _: Long) =>
+ batch.write.format("paimon").mode("append").save(location)
+ }
+ .start()
+
+ val query = () => spark.sql("SELECT * FROM T")
+
+ try {
+ // Show results : There are no expired partitions.
+ checkAnswer(
+ spark.sql(
+ "CALL paimon.sys.expire_partitions(table => 'test.T',
expiration_time => '1 d'" +
+ ", timestamp_formatter => 'yyyy-MM-dd')"),
+ Row("No expired partitions.") :: Nil
+ )
+
+ inputData.addData(("a", "2024-06-02", "02:00"))
+ stream.processAllAvailable()
+ inputData.addData(("b", "2024-06-02", "01:00"))
+ stream.processAllAvailable()
+ inputData.addData(("d", "2024-06-03", "01:00"))
+ stream.processAllAvailable()
+ inputData.addData(("c", "2024-06-01", "01:00"))
+ stream.processAllAvailable()
+ // this snapshot never expires.
+ inputData.addData(("Never-expire", "9999-09-09", "99:99"))
+ stream.processAllAvailable()
+
+ checkAnswer(
+ query(),
+ Row("a", "2024-06-02", "02:00") :: Row("b", "2024-06-02",
"01:00") :: Row(
+ "d",
+ "2024-06-03",
+ "01:00") :: Row("c", "2024-06-01", "01:00") :: Row(
+ "Never-expire",
+ "9999-09-09",
+ "99:99") :: Nil
+ )
+
+ // sorted result of limited expired partitions.
+ checkAnswer(
+ spark.sql(
+ "CALL paimon.sys.expire_partitions(table => 'test.T'" +
+ ", expiration_time => '1 d'" +
+ ", timestamp_formatter => 'yyyy-MM-dd', max_expires => 3)"),
+ Row("pt=2024-06-01, hm=01:00") :: Row("pt=2024-06-02, hm=01:00")
:: Row(
+ "pt=2024-06-02, hm=02:00") :: Nil
+ )
+
+ checkAnswer(
+ query(),
+ Row("d", "2024-06-03", "01:00") :: Row("Never-expire",
"9999-09-09", "99:99") :: Nil)
+ } finally {
+ stream.stop()
+ }
+ }
+ }
+ }
}
