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 2022/04/21 02:15:25 UTC
[flink-table-store] branch master updated: [FLINK-27316] Prevent users from changing bucket number
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 c9feaf5 [FLINK-27316] Prevent users from changing bucket number
c9feaf5 is described below
commit c9feaf5b66d0c87fc228fc1e2b2ead1a7669eb54
Author: Jane Chan <55...@users.noreply.github.com>
AuthorDate: Thu Apr 21 10:15:20 2022 +0800
[FLINK-27316] Prevent users from changing bucket number
This closes #96
---
.../store/connector/ReadWriteTableITCase.java | 60 ++++++++++++++++++++++
.../flink/table/store/file/FileStoreImpl.java | 6 ++-
.../store/file/operation/FileStoreScanImpl.java | 13 ++++-
3 files changed, 77 insertions(+), 2 deletions(-)
diff --git a/flink-table-store-connector/src/test/java/org/apache/flink/table/store/connector/ReadWriteTableITCase.java b/flink-table-store-connector/src/test/java/org/apache/flink/table/store/connector/ReadWriteTableITCase.java
index 5d32712..8ed0076 100644
--- a/flink-table-store-connector/src/test/java/org/apache/flink/table/store/connector/ReadWriteTableITCase.java
+++ b/flink-table-store-connector/src/test/java/org/apache/flink/table/store/connector/ReadWriteTableITCase.java
@@ -65,6 +65,7 @@ import static org.apache.flink.table.store.connector.TableStoreFactoryOptions.SC
import static org.apache.flink.table.store.connector.TableStoreFactoryOptions.SINK_PARALLELISM;
import static org.apache.flink.table.store.connector.TableStoreTestBase.createResolvedTable;
import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
/** IT cases for managed table dml. */
public class ReadWriteTableITCase extends ReadWriteTableTestBase {
@@ -1408,6 +1409,65 @@ public class ReadWriteTableITCase extends ReadWriteTableTestBase {
changelogRow("+I", "Euro", 119L, "2022-01-02")));
}
+ @Test
+ public void testChangeBucketNumber() throws Exception {
+ rootPath = TEMPORARY_FOLDER.newFolder().getPath();
+ tEnv = StreamTableEnvironment.create(buildBatchEnv(), EnvironmentSettings.inBatchMode());
+ tEnv.executeSql(
+ String.format(
+ "CREATE TABLE IF NOT EXISTS rates (\n"
+ + "currency STRING,\n"
+ + " rate BIGINT\n"
+ + ") WITH (\n"
+ + " 'bucket' = '2',\n"
+ + " 'path' = '%s'\n"
+ + ")",
+ rootPath));
+ tEnv.executeSql("INSERT INTO rates VALUES('US Dollar', 102)").await();
+
+ // increase bucket num from 2 to 3
+ tEnv.executeSql("ALTER TABLE rates SET ('bucket' = '3')");
+ assertThatThrownBy(
+ () -> tEnv.executeSql("INSERT INTO rates VALUES('US Dollar', 102)").await())
+ .hasRootCauseInstanceOf(IllegalStateException.class)
+ .hasRootCauseMessage(
+ "Bucket number has been changed. Manifest might be corrupted.");
+ assertThatThrownBy(() -> tEnv.executeSql("SELECT * FROM rates").await())
+ .hasRootCauseInstanceOf(IllegalStateException.class)
+ .hasRootCauseMessage(
+ "Bucket number has been changed. Manifest might be corrupted.");
+
+ // decrease bucket num from 3 to 1
+ tEnv.executeSql("ALTER TABLE rates RESET ('bucket')");
+ assertThatThrownBy(() -> tEnv.executeSql("SELECT * FROM rates").await())
+ .hasRootCauseInstanceOf(IllegalStateException.class)
+ .hasRootCauseMessage(
+ "Bucket number has been changed. Manifest might be corrupted.");
+ }
+
+ @Test
+ public void testSuccessiveWriteAndRead() throws Exception {
+ String managedTable =
+ collectAndCheckBatchReadWrite(false, false, null, Collections.emptyList(), rates());
+ // write rates() twice
+ tEnv.executeSql(
+ String.format(
+ "INSERT INTO `%s`"
+ + " VALUES ('US Dollar', 102),\n"
+ + "('Euro', 114),\n"
+ + "('Yen', 1),\n"
+ + "('Euro', 114),\n"
+ + "('Euro', 119)",
+ managedTable))
+ .await();
+ collectAndCheck(
+ tEnv,
+ managedTable,
+ Collections.emptyMap(),
+ "currency = 'Yen'",
+ Collections.nCopies(2, changelogRow("+I", "Yen", 1L)));
+ }
+
// ------------------------ Tools ----------------------------------
private String collectAndCheckBatchReadWrite(
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 71397b2..46b6cf9 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
@@ -146,7 +146,11 @@ public class FileStoreImpl implements FileStore {
@Override
public FileStoreScanImpl newScan() {
return new FileStoreScanImpl(
- partitionType, pathFactory(), manifestFileFactory(), manifestListFactory());
+ partitionType,
+ pathFactory(),
+ manifestFileFactory(),
+ manifestListFactory(),
+ options.bucket());
}
@Override
diff --git a/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/operation/FileStoreScanImpl.java b/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/operation/FileStoreScanImpl.java
index 8a5e307..3987f02 100644
--- a/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/operation/FileStoreScanImpl.java
+++ b/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/operation/FileStoreScanImpl.java
@@ -57,6 +57,7 @@ public class FileStoreScanImpl implements FileStoreScan {
private final FileStorePathFactory pathFactory;
private final ManifestFile.Factory manifestFileFactory;
private final ManifestList manifestList;
+ private final int numOfBuckets;
private Predicate partitionFilter;
private Predicate keyFilter;
@@ -71,11 +72,13 @@ public class FileStoreScanImpl implements FileStoreScan {
RowType partitionType,
FileStorePathFactory pathFactory,
ManifestFile.Factory manifestFileFactory,
- ManifestList.Factory manifestListFactory) {
+ ManifestList.Factory manifestListFactory,
+ int numOfBuckets) {
this.partitionConverter = new RowDataToObjectArrayConverter(partitionType);
this.pathFactory = pathFactory;
this.manifestFileFactory = manifestFileFactory;
this.manifestList = manifestListFactory.create();
+ this.numOfBuckets = numOfBuckets;
}
@Override
@@ -215,6 +218,9 @@ public class FileStoreScanImpl implements FileStoreScan {
Map<ManifestEntry.Identifier, ManifestEntry> map = new HashMap<>();
for (ManifestEntry entry : entries) {
ManifestEntry.Identifier identifier = entry.identifier();
+ Preconditions.checkState(
+ entry.totalBuckets() == numOfBuckets,
+ "Bucket number has been changed. Manifest might be corrupted.");
switch (entry.kind()) {
case ADD:
Preconditions.checkState(
@@ -261,6 +267,11 @@ public class FileStoreScanImpl implements FileStoreScan {
}
private boolean filterManifestEntry(ManifestEntry entry) {
+ if (specifiedBucket != null) {
+ Preconditions.checkState(
+ specifiedBucket < entry.totalBuckets(),
+ "Bucket number has been changed. Manifest might be corrupted.");
+ }
return (partitionFilter == null
|| partitionFilter.test(partitionConverter.convert(entry.partition())))
&& (keyFilter == null