yuzelin commented on code in PR #497:
URL: https://github.com/apache/flink-table-store/pull/497#discussion_r1095528290
##########
flink-table-store-core/src/main/java/org/apache/flink/table/store/CoreOptions.java:
##########
@@ -401,6 +401,56 @@ 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(
+ "Judge whether the partition is expired according
to the time 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', "
+ + "creates a formatter
using the specified value. "
Review Comment:
`The formatter to format timestamp from string. It can be used with
'partition.timestamp-pattern' to create a formatter using the specified value. `
##########
flink-table-store-core/src/main/java/org/apache/flink/table/store/CoreOptions.java:
##########
@@ -401,6 +401,56 @@ 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(
+ "Judge whether the partition is expired according
to the time extracted from the partition value.");
+
Review Comment:
I think we can use description like:
The expiration interval of a partition. A partition will be expired if
it‘s lifetime is over this value.
##########
flink-table-store-core/src/main/java/org/apache/flink/table/store/CoreOptions.java:
##########
@@ -401,6 +401,56 @@ 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(
+ "Judge whether the partition is expired according
to the time 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', "
+ + "creates a formatter
using the specified value. "
+ + "Supports multiple
partition fields like '$year-$month-$day $hour:00:00'.")
+ .list(
+ 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, a format of
'yyyy-MM-dd hh:mm:ss' is read from the first field."),
Review Comment:
HH:mm:ss for 24-hour clock?
##########
docs/content/docs/how-to/creating-tables.md:
##########
@@ -106,6 +106,8 @@ CREATE TABLE MyTable (
{{< /tabs >}}
+Configure [partition.expiration-time]({{< ref
"docs/maintenance/manage-partition" >}}), expired partitions can be
automatically deleted.
Review Comment:
`By configuring`.
BTW, is it better to place this statement into a `hint info`?
##########
flink-table-store-core/src/test/java/org/apache/flink/table/store/file/operation/PartitionExpireTest.java:
##########
@@ -0,0 +1,149 @@
+/*
+ * 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 {
+ path = new Path(tempDir.toUri());
Review Comment:
Has been set in `beforeEach`.
##########
flink-table-store-core/src/main/java/org/apache/flink/table/store/CoreOptions.java:
##########
@@ -401,6 +401,56 @@ 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(
+ "Judge whether the partition is expired according
to the time 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', "
+ + "creates a formatter
using the specified value. "
+ + "Supports multiple
partition fields like '$year-$month-$day $hour:00:00'.")
+ .list(
+ text(
+ "The timestamp-formatter
is compatible with "
+ + "Java's
DateTimeFormatter."))
Review Comment:
Could describe the default formatter more clearly.
##########
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 objectArrayConverter;
Review Comment:
`toObjectArrayConverter`?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
