You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by lz...@apache.org on 2023/02/03 15:12:09 UTC
[flink-table-store] branch release-0.3 updated: [FLINK-30882] Introduce partition.expiration-time to automatically delete expired partitions
This is an automated email from the ASF dual-hosted git repository.
lzljs3620320 pushed a commit to branch release-0.3
in repository https://gitbox.apache.org/repos/asf/flink-table-store.git
The following commit(s) were added to refs/heads/release-0.3 by this push:
new 5283f98e [FLINK-30882] Introduce partition.expiration-time to automatically delete expired partitions
5283f98e is described below
commit 5283f98ed3b2faf85699c4e661f0ebe258062b7f
Author: Jingsong Lee <ji...@gmail.com>
AuthorDate: Fri Feb 3 23:12:04 2023 +0800
[FLINK-30882] Introduce partition.expiration-time to automatically delete expired partitions
This closes #497
---
docs/content/docs/how-to/creating-tables.md | 4 +-
docs/content/docs/maintenance/configurations.md | 2 +-
docs/content/docs/maintenance/manage-partition.md | 81 +++++++++++
.../shortcodes/generated/core_configuration.html | 24 ++++
.../org/apache/flink/table/store/CoreOptions.java | 78 +++++++++++
.../flink/table/store/file/AbstractFileStore.java | 19 +++
.../apache/flink/table/store/file/FileStore.java | 6 +
.../store/file/operation/PartitionExpire.java | 125 +++++++++++++++++
.../file/partition/PartitionTimeExtractor.java | 125 +++++++++++++++++
.../table/store/table/AbstractFileStoreTable.java | 8 +-
.../flink/table/store/table/sink/TableCommit.java | 16 ++-
.../store/file/operation/PartitionExpireTest.java | 148 +++++++++++++++++++++
.../file/partition/PartitionTimeExtractorTest.java | 90 +++++++++++++
13 files changed, 721 insertions(+), 5 deletions(-)
diff --git a/docs/content/docs/how-to/creating-tables.md b/docs/content/docs/how-to/creating-tables.md
index eba5d91d..c7afbcff 100644
--- a/docs/content/docs/how-to/creating-tables.md
+++ b/docs/content/docs/how-to/creating-tables.md
@@ -107,9 +107,11 @@ CREATE TABLE MyTable (
{{< /tabs >}}
{{< hint info >}}
-
Partition keys must be a subset of primary keys if primary keys are defined.
+{{< /hint >}}
+{{< hint info >}}
+By configuring [partition.expiration-time]({{< ref "docs/maintenance/manage-partition" >}}), expired partitions can be automatically deleted.
{{< /hint >}}
### Create Table Like
diff --git a/docs/content/docs/maintenance/configurations.md b/docs/content/docs/maintenance/configurations.md
index f70c2d5b..34ea60b0 100644
--- a/docs/content/docs/maintenance/configurations.md
+++ b/docs/content/docs/maintenance/configurations.md
@@ -1,6 +1,6 @@
---
title: "Configurations"
-weight: 4
+weight: 10
type: docs
aliases:
- /development/configurations.html
diff --git a/docs/content/docs/maintenance/manage-partition.md b/docs/content/docs/maintenance/manage-partition.md
new file mode 100644
index 00000000..54d68a31
--- /dev/null
+++ b/docs/content/docs/maintenance/manage-partition.md
@@ -0,0 +1,81 @@
+---
+title: "Manage Partition"
+weight: 4
+type: docs
+aliases:
+- /maintenance/manage-partition.html
+---
+<!--
+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.
+-->
+
+## Expiring Partitions
+
+You can set `partition.expiration-time` when creating a partitioned table. Table Store will periodically check
+the status of partitions and delete expired partitions according to time.
+
+How to determine whether a partition has expired: compare the time extracted from the partition with the current
+time to see if survival time has exceeded the `partition.expiration-time`.
+
+An example:
+```sql
+CREATE TABLE T (...) PARTITIONED BY (dt) WITH (
+ 'partition.expiration-time' = '7 d',
+ 'partition.expiration-check-interval' = '1 d',
+ 'partition.timestamp-formatter' = 'yyyyMMdd'
+);
+```
+
+More options:
+
+<table class="table table-bordered">
+ <thead>
+ <tr>
+ <th class="text-left" style="width: 20%">Option</th>
+ <th class="text-left" style="width: 5%">Default</th>
+ <th class="text-left" style="width: 10%">Type</th>
+ <th class="text-left" style="width: 60%">Description</th>
+ </tr>
+ </thead>
+ <tbody>
+ <tr>
+ <td><h5>partition.expiration-check-interval</h5></td>
+ <td style="word-wrap: break-word;">1 h</td>
+ <td>Duration</td>
+ <td>The check interval of partition expiration.</td>
+ </tr>
+ <tr>
+ <td><h5>partition.expiration-time</h5></td>
+ <td style="word-wrap: break-word;">(none)</td>
+ <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.timestamp-formatter</h5></td>
+ <td style="word-wrap: break-word;">(none)</td>
+ <td>String</td>
+ <td>The formatter to format timestamp from string. It can be used with 'partition.timestamp-pattern' to create a formatter using the specified value.<ul><li>Default formatter is 'yyyy-MM-dd HH:mm:ss' and 'yyyy-MM-dd'.</li><li>Supports multiple partition fields like '$year-$month-$day $hour:00:00'.</li><li>The timestamp-formatter is compatible with Java's DateTimeFormatter.</li></ul></td>
+ </tr>
+ <tr>
+ <td><h5>partition.timestamp-pattern</h5></td>
+ <td style="word-wrap: break-word;">(none)</td>
+ <td>String</td>
+ <td>You can specify a pattern to get a timestamp from partitions. The formatter pattern is defined by 'partition.timestamp-formatter'.<ul><li>By default, read from the first field.</li><li>If the timestamp in the partition is a single field called 'dt', you can use '$dt'.</li><li>If it is spread across multiple fields for year, month, day, and hour, you can use '$year-$month-$day $hour:00:00'.</li><li>If the timestamp is in fields dt and hour, you can use '$dt $hour:00:00'.</ [...]
+ </tr>
+ </tbody>
+</table>
diff --git a/docs/layouts/shortcodes/generated/core_configuration.html b/docs/layouts/shortcodes/generated/core_configuration.html
index feae87cf..b123fa65 100644
--- a/docs/layouts/shortcodes/generated/core_configuration.html
+++ b/docs/layouts/shortcodes/generated/core_configuration.html
@@ -188,6 +188,30 @@
<td>String</td>
<td>The default partition name in case the dynamic partition column value is null/empty string.</td>
</tr>
+ <tr>
+ <td><h5>partition.expiration-check-interval</h5></td>
+ <td style="word-wrap: break-word;">1 h</td>
+ <td>Duration</td>
+ <td>The check interval of partition expiration.</td>
+ </tr>
+ <tr>
+ <td><h5>partition.expiration-time</h5></td>
+ <td style="word-wrap: break-word;">(none)</td>
+ <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.timestamp-formatter</h5></td>
+ <td style="word-wrap: break-word;">(none)</td>
+ <td>String</td>
+ <td>The formatter to format timestamp from string. It can be used with 'partition.timestamp-pattern' to create a formatter using the specified value.<ul><li>Default formatter is 'yyyy-MM-dd HH:mm:ss' and 'yyyy-MM-dd'.</li><li>Supports multiple partition fields like '$year-$month-$day $hour:00:00'.</li><li>The timestamp-formatter is compatible with Java's DateTimeFormatter.</li></ul></td>
+ </tr>
+ <tr>
+ <td><h5>partition.timestamp-pattern</h5></td>
+ <td style="word-wrap: break-word;">(none)</td>
+ <td>String</td>
+ <td>You can specify a pattern to get a timestamp from partitions. The formatter pattern is defined by 'partition.timestamp-formatter'.<ul><li>By default, read from the first field.</li><li>If the timestamp in the partition is a single field called 'dt', you can use '$dt'.</li><li>If it is spread across multiple fields for year, month, day, and hour, you can use '$year-$month-$day $hour:00:00'.</li><li>If the timestamp is in fields dt and hour, you can use '$dt $hour:00:00'.</ [...]
+ </tr>
<tr>
<td><h5>scan.mode</h5></td>
<td style="word-wrap: break-word;">default</td>
diff --git a/flink-table-store-core/src/main/java/org/apache/flink/table/store/CoreOptions.java b/flink-table-store-core/src/main/java/org/apache/flink/table/store/CoreOptions.java
index 21fc29e9..c5c3b1e6 100644
--- a/flink-table-store-core/src/main/java/org/apache/flink/table/store/CoreOptions.java
+++ b/flink-table-store-core/src/main/java/org/apache/flink/table/store/CoreOptions.java
@@ -401,6 +401,61 @@ public class CoreOptions implements Serializable {
.withDescription(
"Whether to create underlying storage when reading and writing the table.");
+ public static final ConfigOption<Duration> PARTITION_EXPIRATION_TIME =
+ key("partition.expiration-time")
+ .durationType()
+ .noDefaultValue()
+ .withDescription(
+ "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.");
+
+ public static final ConfigOption<Duration> PARTITION_EXPIRATION_CHECK_INTERVAL =
+ key("partition.expiration-check-interval")
+ .durationType()
+ .defaultValue(Duration.ofHours(1))
+ .withDescription("The check interval of partition expiration.");
+
+ public static final ConfigOption<String> PARTITION_TIMESTAMP_FORMATTER =
+ key("partition.timestamp-formatter")
+ .stringType()
+ .noDefaultValue()
+ .withDescription(
+ Description.builder()
+ .text(
+ "The formatter to format timestamp from string. It can be used"
+ + " with 'partition.timestamp-pattern' to create a formatter"
+ + " using the specified value.")
+ .list(
+ text(
+ "Default formatter is 'yyyy-MM-dd HH:mm:ss' and 'yyyy-MM-dd'."),
+ text(
+ "Supports multiple partition fields like '$year-$month-$day $hour:00:00'."),
+ text(
+ "The timestamp-formatter is compatible with Java's DateTimeFormatter."))
+ .build());
+
+ public static final ConfigOption<String> PARTITION_TIMESTAMP_PATTERN =
+ key("partition.timestamp-pattern")
+ .stringType()
+ .noDefaultValue()
+ .withDescription(
+ Description.builder()
+ .text(
+ "You can specify a pattern to get a timestamp from partitions. "
+ + "The formatter pattern is defined by 'partition.timestamp-formatter'.")
+ .list(
+ text("By default, read from the first field."),
+ text(
+ "If the timestamp in the partition is a single field called 'dt', you can use '$dt'."),
+ text(
+ "If it is spread across multiple fields for year, month, day, and hour,"
+ + " you can use '$year-$month-$day $hour:00:00'."),
+ text(
+ "If the timestamp is in fields dt and hour, you can use '$dt "
+ + "$hour:00:00'."))
+ .build());
+
private final Configuration options;
public CoreOptions(Map<String, String> options) {
@@ -596,6 +651,22 @@ public class CoreOptions implements Serializable {
return options.get(WRITE_ONLY);
}
+ public Duration partitionExpireTime() {
+ return options.get(PARTITION_EXPIRATION_TIME);
+ }
+
+ public Duration partitionExpireCheckInterval() {
+ return options.get(PARTITION_EXPIRATION_CHECK_INTERVAL);
+ }
+
+ public String partitionTimestampFormatter() {
+ return options.get(PARTITION_TIMESTAMP_FORMATTER);
+ }
+
+ public String partitionTimestampPattern() {
+ return options.get(PARTITION_TIMESTAMP_PATTERN);
+ }
+
/** Specifies the merge engine for table with primary key. */
public enum MergeEngine implements DescribedEnum {
DEDUPLICATE("deduplicate", "De-duplicate and keep the last row."),
@@ -847,6 +918,13 @@ public class CoreOptions implements Serializable {
"Cannot define any primary key in an append-only table. Set 'write-mode'='change-log' if "
+ "still want to keep the primary key definition.");
}
+
+ if (schema.options().containsKey(CoreOptions.PARTITION_EXPIRATION_TIME.key())) {
+ if (schema.partitionKeys().isEmpty()) {
+ throw new IllegalArgumentException(
+ "Can not set 'partition.expiration-time' for non-partitioned table.");
+ }
+ }
}
private static void checkOptionExistInMode(
diff --git a/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/AbstractFileStore.java b/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/AbstractFileStore.java
index 91e37620..0abb637c 100644
--- a/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/AbstractFileStore.java
+++ b/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/AbstractFileStore.java
@@ -25,11 +25,13 @@ import org.apache.flink.table.store.file.manifest.ManifestFile;
import org.apache.flink.table.store.file.manifest.ManifestList;
import org.apache.flink.table.store.file.operation.FileStoreCommitImpl;
import org.apache.flink.table.store.file.operation.FileStoreExpireImpl;
+import org.apache.flink.table.store.file.operation.PartitionExpire;
import org.apache.flink.table.store.file.schema.SchemaManager;
import org.apache.flink.table.store.file.utils.FileStorePathFactory;
import org.apache.flink.table.store.file.utils.SnapshotManager;
import org.apache.flink.table.types.logical.RowType;
+import java.time.Duration;
import java.util.Comparator;
/**
@@ -123,5 +125,22 @@ public abstract class AbstractFileStore<T> implements FileStore<T> {
manifestListFactory());
}
+ @Override
+ public PartitionExpire newPartitionExpire(String commitUser) {
+ Duration partitionExpireTime = options.partitionExpireTime();
+ if (partitionExpireTime == null || partitionType().getFieldCount() == 0) {
+ return null;
+ }
+
+ return new PartitionExpire(
+ partitionType(),
+ partitionExpireTime,
+ options.partitionExpireCheckInterval(),
+ options.partitionTimestampPattern(),
+ options.partitionTimestampFormatter(),
+ newScan(),
+ newCommit(commitUser));
+ }
+
public abstract Comparator<RowData> newKeyComparator();
}
diff --git a/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/FileStore.java b/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/FileStore.java
index 157d22d2..abb12d7b 100644
--- a/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/FileStore.java
+++ b/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/FileStore.java
@@ -24,9 +24,12 @@ import org.apache.flink.table.store.file.operation.FileStoreExpire;
import org.apache.flink.table.store.file.operation.FileStoreRead;
import org.apache.flink.table.store.file.operation.FileStoreScan;
import org.apache.flink.table.store.file.operation.FileStoreWrite;
+import org.apache.flink.table.store.file.operation.PartitionExpire;
import org.apache.flink.table.store.file.utils.SnapshotManager;
import org.apache.flink.table.types.logical.RowType;
+import javax.annotation.Nullable;
+
import java.io.Serializable;
/**
@@ -51,4 +54,7 @@ public interface FileStore<T> extends Serializable {
FileStoreCommit newCommit(String commitUser);
FileStoreExpire newExpire();
+
+ @Nullable
+ PartitionExpire newPartitionExpire(String commitUser);
}
diff --git a/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/operation/PartitionExpire.java b/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/operation/PartitionExpire.java
new file mode 100644
index 00000000..b1732a6b
--- /dev/null
+++ b/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/operation/PartitionExpire.java
@@ -0,0 +1,125 @@
+/*
+ * 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.flink.table.store.file.operation;
+
+import org.apache.flink.annotation.VisibleForTesting;
+import org.apache.flink.table.data.binary.BinaryRowData;
+import org.apache.flink.table.store.file.manifest.ManifestCommittable;
+import org.apache.flink.table.store.file.manifest.ManifestEntry;
+import org.apache.flink.table.store.file.partition.PartitionTimeExtractor;
+import org.apache.flink.table.store.utils.RowDataToObjectArrayConverter;
+import org.apache.flink.table.types.logical.RowType;
+
+import java.time.Duration;
+import java.time.LocalDateTime;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+
+/** Expire partitions. */
+public class PartitionExpire {
+
+ private final List<String> partitionKeys;
+ private final RowDataToObjectArrayConverter toObjectArrayConverter;
+ private final Duration expirationTime;
+ private final Duration checkInterval;
+ private final PartitionTimeExtractor timeExtractor;
+ private final FileStoreScan scan;
+ private final FileStoreCommit commit;
+
+ private LocalDateTime lastCheck;
+
+ public PartitionExpire(
+ RowType partitionType,
+ Duration expirationTime,
+ Duration checkInterval,
+ String timePattern,
+ String timeFormatter,
+ FileStoreScan scan,
+ FileStoreCommit commit) {
+ this.partitionKeys = partitionType.getFieldNames();
+ this.toObjectArrayConverter = new RowDataToObjectArrayConverter(partitionType);
+ this.expirationTime = expirationTime;
+ this.checkInterval = checkInterval;
+ this.timeExtractor = new PartitionTimeExtractor(timePattern, timeFormatter);
+ this.scan = scan;
+ this.commit = commit;
+ this.lastCheck = LocalDateTime.now();
+ }
+
+ public PartitionExpire withLock(Lock lock) {
+ this.commit.withLock(lock);
+ return this;
+ }
+
+ public void expire() {
+ expire(LocalDateTime.now());
+ }
+
+ @VisibleForTesting
+ void setLastCheck(LocalDateTime time) {
+ lastCheck = time;
+ }
+
+ @VisibleForTesting
+ void expire(LocalDateTime now) {
+ if (now.isAfter(lastCheck.plus(checkInterval))) {
+ doExpire(now.minus(expirationTime));
+ lastCheck = now;
+ }
+ }
+
+ private void doExpire(LocalDateTime expireDateTime) {
+ List<BinaryRowData> partitions = readPartitions();
+ List<Map<String, String>> expired = new ArrayList<>();
+ for (BinaryRowData partition : partitions) {
+ Object[] array = toObjectArrayConverter.convert(partition);
+ LocalDateTime partTime = timeExtractor.extract(partitionKeys, Arrays.asList(array));
+ if (expireDateTime.isAfter(partTime)) {
+ expired.add(toPartitionString(array));
+ }
+ }
+
+ if (expired.size() > 0) {
+ // identifier is MAX_VALUE to avoid conflict.
+ ManifestCommittable committable = new ManifestCommittable(Long.MAX_VALUE);
+ commit.overwrite(expired, committable, Collections.emptyMap());
+ }
+ }
+
+ private Map<String, String> toPartitionString(Object[] array) {
+ Map<String, String> map = new LinkedHashMap<>();
+ for (int i = 0; i < partitionKeys.size(); i++) {
+ map.put(partitionKeys.get(i), array[i].toString());
+ }
+ return map;
+ }
+
+ private List<BinaryRowData> readPartitions() {
+ // TODO optimize this to read partition only
+ return scan.plan().files().stream()
+ .map(ManifestEntry::partition)
+ .distinct()
+ .collect(Collectors.toList());
+ }
+}
diff --git a/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/partition/PartitionTimeExtractor.java b/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/partition/PartitionTimeExtractor.java
new file mode 100644
index 00000000..15a9bb18
--- /dev/null
+++ b/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/partition/PartitionTimeExtractor.java
@@ -0,0 +1,125 @@
+/*
+ * 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.flink.table.store.file.partition;
+
+import javax.annotation.Nullable;
+
+import java.time.LocalDate;
+import java.time.LocalDateTime;
+import java.time.LocalTime;
+import java.time.format.DateTimeFormatter;
+import java.time.format.DateTimeFormatterBuilder;
+import java.time.format.DateTimeParseException;
+import java.time.format.ResolverStyle;
+import java.time.format.SignStyle;
+import java.time.temporal.ChronoField;
+import java.util.List;
+import java.util.Locale;
+import java.util.Objects;
+
+import static java.time.temporal.ChronoField.DAY_OF_MONTH;
+import static java.time.temporal.ChronoField.HOUR_OF_DAY;
+import static java.time.temporal.ChronoField.MINUTE_OF_HOUR;
+import static java.time.temporal.ChronoField.MONTH_OF_YEAR;
+import static java.time.temporal.ChronoField.SECOND_OF_MINUTE;
+import static java.time.temporal.ChronoField.YEAR;
+
+/** Time extractor to extract time from partition values. */
+public class PartitionTimeExtractor {
+
+ private static final DateTimeFormatter TIMESTAMP_FORMATTER =
+ new DateTimeFormatterBuilder()
+ .appendValue(YEAR, 1, 10, SignStyle.NORMAL)
+ .appendLiteral('-')
+ .appendValue(MONTH_OF_YEAR, 1, 2, SignStyle.NORMAL)
+ .appendLiteral('-')
+ .appendValue(DAY_OF_MONTH, 1, 2, SignStyle.NORMAL)
+ .optionalStart()
+ .appendLiteral(" ")
+ .appendValue(HOUR_OF_DAY, 1, 2, SignStyle.NORMAL)
+ .appendLiteral(':')
+ .appendValue(MINUTE_OF_HOUR, 1, 2, SignStyle.NORMAL)
+ .appendLiteral(':')
+ .appendValue(SECOND_OF_MINUTE, 1, 2, SignStyle.NORMAL)
+ .optionalStart()
+ .appendFraction(ChronoField.NANO_OF_SECOND, 1, 9, true)
+ .optionalEnd()
+ .optionalEnd()
+ .toFormatter()
+ .withResolverStyle(ResolverStyle.LENIENT);
+
+ private static final DateTimeFormatter DATE_FORMATTER =
+ new DateTimeFormatterBuilder()
+ .appendValue(YEAR, 1, 10, SignStyle.NORMAL)
+ .appendLiteral('-')
+ .appendValue(MONTH_OF_YEAR, 1, 2, SignStyle.NORMAL)
+ .appendLiteral('-')
+ .appendValue(DAY_OF_MONTH, 1, 2, SignStyle.NORMAL)
+ .toFormatter()
+ .withResolverStyle(ResolverStyle.LENIENT);
+
+ @Nullable private final String pattern;
+ @Nullable private final String formatter;
+
+ public PartitionTimeExtractor(@Nullable String pattern, @Nullable String formatter) {
+ this.pattern = pattern;
+ this.formatter = formatter;
+ }
+
+ public LocalDateTime extract(List<String> partitionKeys, List<Object> partitionValues) {
+ String timestampString;
+ if (pattern == null) {
+ timestampString = partitionValues.get(0).toString();
+ } else {
+ timestampString = pattern;
+ for (int i = 0; i < partitionKeys.size(); i++) {
+ timestampString =
+ timestampString.replaceAll(
+ "\\$" + partitionKeys.get(i), partitionValues.get(i).toString());
+ }
+ }
+ return toLocalDateTime(timestampString, this.formatter);
+ }
+
+ private static LocalDateTime toLocalDateTime(
+ String timestampString, @Nullable String formatterPattern) {
+
+ if (formatterPattern == null) {
+ return PartitionTimeExtractor.toLocalDateTimeDefault(timestampString);
+ }
+ DateTimeFormatter dateTimeFormatter =
+ DateTimeFormatter.ofPattern(Objects.requireNonNull(formatterPattern), Locale.ROOT);
+ try {
+ return LocalDateTime.parse(timestampString, Objects.requireNonNull(dateTimeFormatter));
+ } catch (DateTimeParseException e) {
+ return LocalDateTime.of(
+ LocalDate.parse(timestampString, Objects.requireNonNull(dateTimeFormatter)),
+ LocalTime.MIDNIGHT);
+ }
+ }
+
+ public static LocalDateTime toLocalDateTimeDefault(String timestampString) {
+ try {
+ return LocalDateTime.parse(timestampString, TIMESTAMP_FORMATTER);
+ } catch (DateTimeParseException e) {
+ return LocalDateTime.of(
+ LocalDate.parse(timestampString, DATE_FORMATTER), LocalTime.MIDNIGHT);
+ }
+ }
+}
diff --git a/flink-table-store-core/src/main/java/org/apache/flink/table/store/table/AbstractFileStoreTable.java b/flink-table-store-core/src/main/java/org/apache/flink/table/store/table/AbstractFileStoreTable.java
index 83022584..5c2a7715 100644
--- a/flink-table-store-core/src/main/java/org/apache/flink/table/store/table/AbstractFileStoreTable.java
+++ b/flink-table-store-core/src/main/java/org/apache/flink/table/store/table/AbstractFileStoreTable.java
@@ -18,6 +18,7 @@
package org.apache.flink.table.store.table;
+import org.apache.flink.annotation.VisibleForTesting;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.core.fs.Path;
import org.apache.flink.table.store.CoreOptions;
@@ -45,7 +46,8 @@ public abstract class AbstractFileStoreTable implements FileStoreTable {
this.tableSchema = tableSchema;
}
- protected abstract FileStore<?> store();
+ @VisibleForTesting
+ public abstract FileStore<?> store();
protected abstract FileStoreTable copy(TableSchema newTableSchema);
@@ -107,6 +109,8 @@ public abstract class AbstractFileStoreTable implements FileStoreTable {
@Override
public TableCommit newCommit(String commitUser) {
return new TableCommit(
- store().newCommit(commitUser), options().writeOnly() ? null : store().newExpire());
+ store().newCommit(commitUser),
+ options().writeOnly() ? null : store().newExpire(),
+ options().writeOnly() ? null : store().newPartitionExpire(commitUser));
}
}
diff --git a/flink-table-store-core/src/main/java/org/apache/flink/table/store/table/sink/TableCommit.java b/flink-table-store-core/src/main/java/org/apache/flink/table/store/table/sink/TableCommit.java
index 56e5905a..972b0c32 100644
--- a/flink-table-store-core/src/main/java/org/apache/flink/table/store/table/sink/TableCommit.java
+++ b/flink-table-store-core/src/main/java/org/apache/flink/table/store/table/sink/TableCommit.java
@@ -22,6 +22,7 @@ import org.apache.flink.table.store.file.manifest.ManifestCommittable;
import org.apache.flink.table.store.file.operation.FileStoreCommit;
import org.apache.flink.table.store.file.operation.FileStoreExpire;
import org.apache.flink.table.store.file.operation.Lock;
+import org.apache.flink.table.store.file.operation.PartitionExpire;
import javax.annotation.Nullable;
@@ -38,13 +39,18 @@ public class TableCommit implements AutoCloseable {
private final FileStoreCommit commit;
@Nullable private final FileStoreExpire expire;
+ @Nullable private final PartitionExpire partitionExpire;
@Nullable private List<Map<String, String>> overwritePartitions = null;
@Nullable private Lock lock;
- public TableCommit(FileStoreCommit commit, @Nullable FileStoreExpire expire) {
+ public TableCommit(
+ FileStoreCommit commit,
+ @Nullable FileStoreExpire expire,
+ @Nullable PartitionExpire partitionExpire) {
this.commit = commit;
this.expire = expire;
+ this.partitionExpire = partitionExpire;
}
public TableCommit withOverwritePartition(@Nullable Map<String, String> overwritePartition) {
@@ -67,6 +73,10 @@ public class TableCommit implements AutoCloseable {
expire.withLock(lock);
}
+ if (partitionExpire != null) {
+ partitionExpire.withLock(lock);
+ }
+
this.lock = lock;
return this;
}
@@ -113,6 +123,10 @@ public class TableCommit implements AutoCloseable {
if (expire != null) {
expire.expire();
}
+
+ if (partitionExpire != null) {
+ partitionExpire.expire();
+ }
}
@Override
diff --git a/flink-table-store-core/src/test/java/org/apache/flink/table/store/file/operation/PartitionExpireTest.java b/flink-table-store-core/src/test/java/org/apache/flink/table/store/file/operation/PartitionExpireTest.java
new file mode 100644
index 00000000..171c479c
--- /dev/null
+++ b/flink-table-store-core/src/test/java/org/apache/flink/table/store/file/operation/PartitionExpireTest.java
@@ -0,0 +1,148 @@
+/*
+ * 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.flink.table.store.file.operation;
+
+import org.apache.flink.core.fs.Path;
+import org.apache.flink.table.data.GenericRowData;
+import org.apache.flink.table.data.StringData;
+import org.apache.flink.table.store.CoreOptions;
+import org.apache.flink.table.store.file.schema.SchemaManager;
+import org.apache.flink.table.store.file.schema.UpdateSchema;
+import org.apache.flink.table.store.file.utils.RecordReaderUtils;
+import org.apache.flink.table.store.table.AbstractFileStoreTable;
+import org.apache.flink.table.store.table.FileStoreTable;
+import org.apache.flink.table.store.table.FileStoreTableFactory;
+import org.apache.flink.table.store.table.sink.TableWrite;
+import org.apache.flink.table.types.logical.RowType;
+import org.apache.flink.table.types.logical.VarCharType;
+
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+import java.io.IOException;
+import java.time.LocalDate;
+import java.time.LocalDateTime;
+import java.time.LocalTime;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import static org.apache.flink.table.store.CoreOptions.PARTITION_EXPIRATION_TIME;
+import static org.apache.flink.table.store.CoreOptions.WRITE_ONLY;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/** Test for {@link PartitionExpire}. */
+public class PartitionExpireTest {
+
+ @TempDir java.nio.file.Path tempDir;
+
+ private Path path;
+ private FileStoreTable table;
+
+ @BeforeEach
+ public void beforeEach() {
+ path = new Path(tempDir.toUri());
+ }
+
+ @Test
+ public void testNonPartitionedTable() {
+ SchemaManager schemaManager = new SchemaManager(path);
+ assertThatThrownBy(
+ () ->
+ schemaManager.commitNewVersion(
+ new UpdateSchema(
+ RowType.of(VarCharType.STRING_TYPE),
+ Collections.emptyList(),
+ Collections.emptyList(),
+ Collections.singletonMap(
+ PARTITION_EXPIRATION_TIME.key(), "1 d"),
+ "")))
+ .hasMessageContaining(
+ "Can not set 'partition.expiration-time' for non-partitioned table");
+ }
+
+ @Test
+ public void test() throws Exception {
+ SchemaManager schemaManager = new SchemaManager(path);
+ schemaManager.commitNewVersion(
+ new UpdateSchema(
+ RowType.of(VarCharType.STRING_TYPE, VarCharType.STRING_TYPE),
+ Collections.singletonList("f0"),
+ Collections.emptyList(),
+ Collections.emptyMap(),
+ ""));
+ table = FileStoreTableFactory.create(path);
+
+ write("20230101", "11");
+ write("20230101", "12");
+ write("20230103", "31");
+ write("20230103", "32");
+ write("20230105", "51");
+
+ PartitionExpire expire = newExpire();
+ expire.setLastCheck(date(1));
+
+ expire.expire(date(3));
+ assertThat(read())
+ .containsExactlyInAnyOrder(
+ "20230101:11", "20230101:12", "20230103:31", "20230103:32", "20230105:51");
+
+ expire.expire(date(5));
+ assertThat(read()).containsExactlyInAnyOrder("20230103:31", "20230103:32", "20230105:51");
+
+ // PARTITION_EXPIRATION_INTERVAL not trigger
+ expire.expire(date(6));
+ assertThat(read()).containsExactlyInAnyOrder("20230103:31", "20230103:32", "20230105:51");
+
+ expire.expire(date(8));
+ assertThat(read()).isEmpty();
+ }
+
+ private List<String> read() throws IOException {
+ List<String> ret = new ArrayList<>();
+ RecordReaderUtils.forEachRemaining(
+ table.newRead().createReader(table.newScan().plan().splits()),
+ row -> ret.add(row.getString(0) + ":" + row.getString(1)));
+ return ret;
+ }
+
+ private LocalDateTime date(int day) {
+ return LocalDateTime.of(LocalDate.of(2023, 1, day), LocalTime.MIN);
+ }
+
+ private void write(String f0, String f1) throws Exception {
+ TableWrite write =
+ table.copy(Collections.singletonMap(WRITE_ONLY.key(), "true")).newWrite("");
+ write.write(GenericRowData.of(StringData.fromString(f0), StringData.fromString(f1)));
+ table.newCommit("").commit(0, write.prepareCommit(true, 0));
+ write.close();
+ }
+
+ private PartitionExpire newExpire() {
+ Map<String, String> options = new HashMap<>();
+ options.put(PARTITION_EXPIRATION_TIME.key(), "2 d");
+ options.put(CoreOptions.PARTITION_EXPIRATION_CHECK_INTERVAL.key(), "1 d");
+ options.put(CoreOptions.PARTITION_TIMESTAMP_FORMATTER.key(), "yyyyMMdd");
+ return ((AbstractFileStoreTable) table.copy(options)).store().newPartitionExpire("");
+ }
+}
diff --git a/flink-table-store-core/src/test/java/org/apache/flink/table/store/file/partition/PartitionTimeExtractorTest.java b/flink-table-store-core/src/test/java/org/apache/flink/table/store/file/partition/PartitionTimeExtractorTest.java
new file mode 100644
index 00000000..816f1922
--- /dev/null
+++ b/flink-table-store-core/src/test/java/org/apache/flink/table/store/file/partition/PartitionTimeExtractorTest.java
@@ -0,0 +1,90 @@
+/*
+ * 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.flink.table.store.file.partition;
+
+import org.junit.jupiter.api.Test;
+
+import java.time.LocalDateTime;
+import java.util.Arrays;
+import java.util.Collections;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** Test for {@link PartitionTimeExtractor}. */
+public class PartitionTimeExtractorTest {
+
+ @Test
+ public void testDefault() {
+ PartitionTimeExtractor extractor = new PartitionTimeExtractor(null, null);
+ assertThat(
+ extractor.extract(
+ Collections.emptyList(),
+ Collections.singletonList("2023-01-01 20:08:08")))
+ .isEqualTo(LocalDateTime.parse("2023-01-01T20:08:08"));
+
+ assertThat(
+ extractor.extract(
+ Collections.emptyList(),
+ Collections.singletonList("2023-1-1 20:08:08")))
+ .isEqualTo(LocalDateTime.parse("2023-01-01T20:08:08"));
+
+ assertThat(
+ extractor.extract(
+ Collections.emptyList(), Collections.singletonList("2023-01-01")))
+ .isEqualTo(LocalDateTime.parse("2023-01-01T00:00:00"));
+
+ assertThat(
+ extractor.extract(
+ Collections.emptyList(), Collections.singletonList("2023-1-1")))
+ .isEqualTo(LocalDateTime.parse("2023-01-01T00:00:00"));
+ }
+
+ @Test
+ public void testPattern() {
+ PartitionTimeExtractor extractor =
+ new PartitionTimeExtractor("$year-$month-$day 00:00:00", null);
+ assertThat(
+ extractor.extract(
+ Arrays.asList("year", "month", "day"),
+ Arrays.asList("2023", "01", "01")))
+ .isEqualTo(LocalDateTime.parse("2023-01-01T00:00:00"));
+
+ extractor = new PartitionTimeExtractor("$year-$month-$day $hour:00:00", null);
+ assertThat(
+ extractor.extract(
+ Arrays.asList("year", "month", "day", "hour"),
+ Arrays.asList("2023", "01", "01", "01")))
+ .isEqualTo(LocalDateTime.parse("2023-01-01T01:00:00"));
+
+ extractor = new PartitionTimeExtractor("$dt", null);
+ assertThat(
+ extractor.extract(
+ Arrays.asList("other", "dt"), Arrays.asList("dummy", "2023-01-01")))
+ .isEqualTo(LocalDateTime.parse("2023-01-01T00:00:00"));
+ }
+
+ @Test
+ public void testFormatter() {
+ PartitionTimeExtractor extractor = new PartitionTimeExtractor(null, "yyyyMMdd");
+ assertThat(
+ extractor.extract(
+ Collections.emptyList(), Collections.singletonList("20230101")))
+ .isEqualTo(LocalDateTime.parse("2023-01-01T00:00:00"));
+ }
+}