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/flink-table-store.git
The following commit(s) were added to refs/heads/master by this push:
new b3efd4f [FLINK-27085] Introduce snapshot.num-retained.min option
b3efd4f is described below
commit b3efd4f70e59b9d8069fecf13f506e72ed9a3e8d
Author: tsreaper <[email protected]>
AuthorDate: Wed Apr 6 18:05:53 2022 +0800
[FLINK-27085] Introduce snapshot.num-retained.min option
This closes #75
---
docs/content/docs/development/write-table.md | 9 ++++++-
.../flink/table/store/file/FileStoreImpl.java | 3 ++-
.../flink/table/store/file/FileStoreOptions.java | 29 ++++++++++++++++++----
.../store/file/operation/FileStoreExpireImpl.java | 17 +++++++------
.../flink/table/store/file/TestFileStore.java | 6 +++--
.../store/file/operation/FileStoreExpireTest.java | 27 ++++++++++++--------
6 files changed, 64 insertions(+), 27 deletions(-)
diff --git a/docs/content/docs/development/write-table.md
b/docs/content/docs/development/write-table.md
index 7b7c7ad..b866737 100644
--- a/docs/content/docs/development/write-table.md
+++ b/docs/content/docs/development/write-table.md
@@ -142,7 +142,14 @@ to eliminate expired snapshots:
<td>The maximum time of completed snapshots to retain.</td>
</tr>
<tr>
- <td><h5>snapshot.num-retained</h5></td>
+ <td><h5>snapshot.num-retained.min</h5></td>
+ <td>No</td>
+ <td style="word-wrap: break-word;">10</td>
+ <td>Integer</td>
+ <td>The minimum number of completed snapshots to retain.</td>
+ </tr>
+ <tr>
+ <td><h5>snapshot.num-retained.max</h5></td>
<td>No</td>
<td style="word-wrap: break-word;">Integer.MAX_VALUE</td>
<td>Integer</td>
diff --git
a/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/FileStoreImpl.java
b/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/FileStoreImpl.java
index 72e7c50..71397b2 100644
---
a/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/FileStoreImpl.java
+++
b/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/FileStoreImpl.java
@@ -135,7 +135,8 @@ public class FileStoreImpl implements FileStore {
@Override
public FileStoreExpireImpl newExpire() {
return new FileStoreExpireImpl(
- options.snapshotNumRetain(),
+ options.snapshotNumRetainMin(),
+ options.snapshotNumRetainMax(),
options.snapshotTimeRetain().toMillis(),
pathFactory(),
manifestFileFactory(),
diff --git
a/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/FileStoreOptions.java
b/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/FileStoreOptions.java
index 7f7bf8f..f9fbd39 100644
---
a/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/FileStoreOptions.java
+++
b/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/FileStoreOptions.java
@@ -87,8 +87,14 @@ public class FileStoreOptions implements Serializable {
"The default partition name in case the dynamic
partition"
+ " column value is null/empty string.");
- public static final ConfigOption<Integer> SNAPSHOT_NUM_RETAINED =
- ConfigOptions.key("snapshot.num-retained")
+ public static final ConfigOption<Integer> SNAPSHOT_NUM_RETAINED_MIN =
+ ConfigOptions.key("snapshot.num-retained.min")
+ .intType()
+ .defaultValue(10)
+ .withDescription("The minimum number of completed
snapshots to retain.");
+
+ public static final ConfigOption<Integer> SNAPSHOT_NUM_RETAINED_MAX =
+ ConfigOptions.key("snapshot.num-retained.max")
.intType()
.defaultValue(Integer.MAX_VALUE)
.withDescription("The maximum number of completed
snapshots to retain.");
@@ -116,7 +122,8 @@ public class FileStoreOptions implements Serializable {
allOptions.add(MANIFEST_TARGET_FILE_SIZE);
allOptions.add(MANIFEST_MERGE_MIN_COUNT);
allOptions.add(PARTITION_DEFAULT_NAME);
- allOptions.add(SNAPSHOT_NUM_RETAINED);
+ allOptions.add(SNAPSHOT_NUM_RETAINED_MIN);
+ allOptions.add(SNAPSHOT_NUM_RETAINED_MAX);
allOptions.add(SNAPSHOT_TIME_RETAINED);
return allOptions;
}
@@ -124,6 +131,14 @@ public class FileStoreOptions implements Serializable {
public FileStoreOptions(Configuration options) {
this.options = options;
// TODO validate all keys
+ Preconditions.checkArgument(
+ snapshotNumRetainMin() > 0,
+ SNAPSHOT_NUM_RETAINED_MIN.key() + " should be at least 1");
+ Preconditions.checkArgument(
+ snapshotNumRetainMin() <= snapshotNumRetainMax(),
+ SNAPSHOT_NUM_RETAINED_MIN.key()
+ + " should not be larger than "
+ + SNAPSHOT_NUM_RETAINED_MAX.key());
}
public int bucket() {
@@ -177,8 +192,12 @@ public class FileStoreOptions implements Serializable {
return new MergeTreeOptions(options);
}
- public int snapshotNumRetain() {
- return options.get(SNAPSHOT_NUM_RETAINED);
+ public int snapshotNumRetainMin() {
+ return options.get(SNAPSHOT_NUM_RETAINED_MIN);
+ }
+
+ public int snapshotNumRetainMax() {
+ return options.get(SNAPSHOT_NUM_RETAINED_MAX);
}
public Duration snapshotTimeRetain() {
diff --git
a/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/operation/FileStoreExpireImpl.java
b/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/operation/FileStoreExpireImpl.java
index 7d7d94c..feda420 100644
---
a/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/operation/FileStoreExpireImpl.java
+++
b/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/operation/FileStoreExpireImpl.java
@@ -52,8 +52,9 @@ public class FileStoreExpireImpl implements FileStoreExpire {
private static final Logger LOG =
LoggerFactory.getLogger(FileStoreExpireImpl.class);
+ private final int numRetainedMin;
// snapshots exceeding any constraint will be expired
- private final int numRetained;
+ private final int numRetainedMax;
private final long millisRetained;
private final FileStorePathFactory pathFactory;
@@ -63,12 +64,14 @@ public class FileStoreExpireImpl implements FileStoreExpire
{
private Lock lock;
public FileStoreExpireImpl(
- int numRetained,
+ int numRetainedMin,
+ int numRetainedMax,
long millisRetained,
FileStorePathFactory pathFactory,
ManifestFile.Factory manifestFileFactory,
ManifestList.Factory manifestListFactory) {
- this.numRetained = numRetained;
+ this.numRetainedMin = numRetainedMin;
+ this.numRetainedMax = numRetainedMax;
this.millisRetained = millisRetained;
this.pathFactory = pathFactory;
this.manifestFile = manifestFileFactory.create();
@@ -102,8 +105,8 @@ public class FileStoreExpireImpl implements FileStoreExpire
{
}
// find the earliest snapshot to retain
- for (long id = Math.max(latestSnapshotId - numRetained + 1, earliest);
- id <= latestSnapshotId;
+ for (long id = Math.max(latestSnapshotId - numRetainedMax + 1,
earliest);
+ id <= latestSnapshotId - numRetainedMin;
id++) {
Path snapshotPath = pathFactory.toSnapshotPath(id);
try {
@@ -121,8 +124,8 @@ public class FileStoreExpireImpl implements FileStoreExpire
{
}
}
- // no snapshot can be retained, expire all but last one
- expireUntil(earliest, latestSnapshotId);
+ // no snapshot can be retained, expire until there are only
numRetainedMin snapshots left
+ expireUntil(earliest, latestSnapshotId - numRetainedMin + 1);
}
private void expireUntil(long earliestId, long endExclusiveId) {
diff --git
a/flink-table-store-core/src/test/java/org/apache/flink/table/store/file/TestFileStore.java
b/flink-table-store-core/src/test/java/org/apache/flink/table/store/file/TestFileStore.java
index 15a85d0..1c5bbbf 100644
---
a/flink-table-store-core/src/test/java/org/apache/flink/table/store/file/TestFileStore.java
+++
b/flink-table-store-core/src/test/java/org/apache/flink/table/store/file/TestFileStore.java
@@ -122,9 +122,11 @@ public class TestFileStore extends FileStoreImpl {
this.valueSerializer = new RowDataSerializer(valueType);
}
- public FileStoreExpireImpl newExpire(int numRetained, long millisRetained)
{
+ public FileStoreExpireImpl newExpire(
+ int numRetainedMin, int numRetainedMax, long millisRetained) {
return new FileStoreExpireImpl(
- numRetained,
+ numRetainedMin,
+ numRetainedMax,
millisRetained,
pathFactory(),
manifestFileFactory(),
diff --git
a/flink-table-store-core/src/test/java/org/apache/flink/table/store/file/operation/FileStoreExpireTest.java
b/flink-table-store-core/src/test/java/org/apache/flink/table/store/file/operation/FileStoreExpireTest.java
index 2884a85..84da568 100644
---
a/flink-table-store-core/src/test/java/org/apache/flink/table/store/file/operation/FileStoreExpireTest.java
+++
b/flink-table-store-core/src/test/java/org/apache/flink/table/store/file/operation/FileStoreExpireTest.java
@@ -74,7 +74,7 @@ public class FileStoreExpireTest {
@Test
public void testNoSnapshot() {
- FileStoreExpire expire = store.newExpire(3, Long.MAX_VALUE);
+ FileStoreExpire expire = store.newExpire(1, 3, Long.MAX_VALUE);
expire.expire();
assertThat(pathFactory.latestSnapshotId()).isNull();
@@ -86,7 +86,7 @@ public class FileStoreExpireTest {
List<Integer> snapshotPositions = new ArrayList<>();
commit(2, allData, snapshotPositions);
int latestSnapshotId = pathFactory.latestSnapshotId().intValue();
- FileStoreExpire expire = store.newExpire(latestSnapshotId + 1,
Long.MAX_VALUE);
+ FileStoreExpire expire = store.newExpire(1, latestSnapshotId + 1,
Long.MAX_VALUE);
expire.expire();
FileSystem fs =
pathFactory.toSnapshotPath(latestSnapshotId).getFileSystem();
@@ -102,7 +102,7 @@ public class FileStoreExpireTest {
List<Integer> snapshotPositions = new ArrayList<>();
commit(5, allData, snapshotPositions);
int latestSnapshotId = pathFactory.latestSnapshotId().intValue();
- FileStoreExpire expire = store.newExpire(Integer.MAX_VALUE,
Long.MAX_VALUE);
+ FileStoreExpire expire = store.newExpire(1, Integer.MAX_VALUE,
Long.MAX_VALUE);
expire.expire();
FileSystem fs =
pathFactory.toSnapshotPath(latestSnapshotId).getFileSystem();
@@ -113,26 +113,31 @@ public class FileStoreExpireTest {
}
@Test
- public void testKeepAtLeastOneSnapshot() throws Exception {
+ public void testNumRetainedMin() throws Exception {
+ ThreadLocalRandom random = ThreadLocalRandom.current();
+
+ int numRetainedMin = random.nextInt(5) + 1;
List<KeyValue> allData = new ArrayList<>();
List<Integer> snapshotPositions = new ArrayList<>();
- commit(3, allData, snapshotPositions);
+ commit(numRetainedMin + random.nextInt(5), allData, snapshotPositions);
int latestSnapshotId = pathFactory.latestSnapshotId().intValue();
Thread.sleep(100);
- FileStoreExpire expire = store.newExpire(Integer.MAX_VALUE, 1);
+ FileStoreExpire expire = store.newExpire(numRetainedMin,
Integer.MAX_VALUE, 1);
expire.expire();
FileSystem fs =
pathFactory.toSnapshotPath(latestSnapshotId).getFileSystem();
- for (int i = 1; i < latestSnapshotId; i++) {
+ for (int i = 1; i <= latestSnapshotId - numRetainedMin; i++) {
assertThat(fs.exists(pathFactory.toSnapshotPath(i))).isFalse();
}
-
assertThat(fs.exists(pathFactory.toSnapshotPath(latestSnapshotId))).isTrue();
- assertSnapshot(latestSnapshotId, allData, snapshotPositions);
+ for (int i = latestSnapshotId - numRetainedMin + 1; i <=
latestSnapshotId; i++) {
+ assertThat(fs.exists(pathFactory.toSnapshotPath(i))).isTrue();
+ assertSnapshot(i, allData, snapshotPositions);
+ }
}
@Test
public void testExpireWithNumber() throws Exception {
- FileStoreExpire expire = store.newExpire(3, Long.MAX_VALUE);
+ FileStoreExpire expire = store.newExpire(1, 3, Long.MAX_VALUE);
List<KeyValue> allData = new ArrayList<>();
List<Integer> snapshotPositions = new ArrayList<>();
@@ -169,7 +174,7 @@ public class FileStoreExpireTest {
@Test
public void testExpireWithTime() throws Exception {
- FileStoreExpire expire = store.newExpire(Integer.MAX_VALUE, 1000);
+ FileStoreExpire expire = store.newExpire(1, Integer.MAX_VALUE, 1000);
List<KeyValue> allData = new ArrayList<>();
List<Integer> snapshotPositions = new ArrayList<>();
