You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@paimon.apache.org by "LsomeYeah (via GitHub)" <gi...@apache.org> on 2024/11/04 09:40:59 UTC
[PR] [core] throw exception if table is recreated when it still being read [paimon]
LsomeYeah opened a new pull request, #4445:
URL: https://github.com/apache/paimon/pull/4445
<!-- Please specify the module before the PR name: [core] ... or [flink] ... -->
### Purpose
<!-- Linking this pull request to the issue -->
Linked issue: close #xxx
<!-- What is the purpose of the change -->
A table which is read in streaming read, will not throw exception even if the table is recreated currently. For example, the expected snapshot id is 10001, but the latest snapshot of the recreated table might be null or much less than 10001.
This is because when try to get the next snapshot, if the next snapshot is not exists in filesystem and its id is greater than earliest snapshot id, it will be treat as the next snapshot is still in generating progress.
This feature adds a check in `NextSnapshotFetcher#getNextSnapshot`. If next expected snapshot id is greater than the latest snapshot id plus one, it will throw an exception.
### Tests
<!-- List UT and IT cases to verify this change -->
`PrimaryKeyFileStoreTableITCase#testRecreateTableWithException`
### API and Format
<!-- Does this change affect API or storage format -->
### Documentation
<!-- Does this change introduce a new feature -->
--
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: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [core] throw exception if table is recreated when it still being read [paimon]
Posted by "tsreaper (via GitHub)" <gi...@apache.org>.
tsreaper commented on code in PR #4445:
URL: https://github.com/apache/paimon/pull/4445#discussion_r1830328316
##########
paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/PrimaryKeyFileStoreTableITCase.java:
##########
@@ -335,6 +338,88 @@ public void testBatchJobWithConflictAndRestart() throws Exception {
}
}
+ @Timeout(120)
+ @ParameterizedTest()
+ @ValueSource(booleans = {false, true})
+ public void testRecreateTableWithException(boolean isReloadData) throws Exception {
+ TableEnvironment bEnv = tableEnvironmentBuilder().batchMode().build();
+ bEnv.executeSql(createCatalogSql("testCatalog", path + "/warehouse"));
+ bEnv.executeSql("USE CATALOG testCatalog");
+ bEnv.executeSql(
+ "CREATE TABLE t ( pt INT, k INT, v INT, PRIMARY KEY (pt, k) NOT ENFORCED ) "
+ + "PARTITIONED BY (pt) "
+ + "WITH ("
+ + " 'bucket' = '2'\n"
+ + " ,'continuous.discovery-interval' = '1s'\n"
+ + ")");
+
+ TableEnvironment sEnv =
+ tableEnvironmentBuilder()
+ .streamingMode()
+ .parallelism(4)
+ .checkpointIntervalMs(1000)
+ .build();
+ sEnv.executeSql(createCatalogSql("testCatalog", path + "/warehouse"));
+ sEnv.executeSql("USE CATALOG testCatalog");
+ CloseableIterator<Row> it = sEnv.executeSql("SELECT * FROM t").collect();
+
+ // first write
+ List<String> values = new ArrayList<>();
+ for (int i = 0; i < 10; i++) {
+ values.add(String.format("(0, %d, %d)", i, i));
+ values.add(String.format("(1, %d, %d)", i, i));
+ }
+ bEnv.executeSql("INSERT INTO t VALUES " + String.join(", ", values)).await();
+ List<Row> expected = new ArrayList<>();
+ for (int i = 0; i < 10; i++) {
+ expected.add(Row.ofKind(RowKind.INSERT, 0, i, i));
+ expected.add(Row.ofKind(RowKind.INSERT, 1, i, i));
+ }
+ assertStreamingResult(it, expected);
+
+ // second write
+ values.clear();
+ for (int i = 0; i < 10; i++) {
+ values.add(String.format("(0, %d, %d)", i, i + 1));
+ values.add(String.format("(1, %d, %d)", i, i + 1));
+ }
+ bEnv.executeSql("INSERT INTO t VALUES " + String.join(", ", values)).await();
+
+ // start a read job
+ for (int i = 0; i < 10; i++) {
+ expected.add(Row.ofKind(RowKind.UPDATE_BEFORE, 0, i, i));
+ expected.add(Row.ofKind(RowKind.UPDATE_BEFORE, 1, i, i));
+ expected.add(Row.ofKind(RowKind.UPDATE_AFTER, 0, i, i + 1));
+ expected.add(Row.ofKind(RowKind.UPDATE_AFTER, 1, i, i + 1));
+ }
+ assertStreamingResult(it, expected.subList(20, 60));
+
+ // delete table and recreate a same table
+ bEnv.executeSql("DROP TABLE t");
+ bEnv.executeSql(
+ "CREATE TABLE t ( pt INT, k INT, v INT, PRIMARY KEY (pt, k) NOT ENFORCED ) "
+ + "PARTITIONED BY (pt) "
+ + "WITH ("
+ + " 'bucket' = '2'\n"
+ + ")");
+
+ // if reload data, it will generate a new snapshot for recreated table
+ if (isReloadData) {
+ bEnv.executeSql("INSERT INTO t VALUES " + String.join(", ", values)).await();
+ }
+ assertThatCode(
+ () -> {
+ while (true) {
+ if (it.hasNext()) {
+ it.next();
+ }
+ }
+ })
Review Comment:
```suggestion
assertThatCode(() -> it.next())
```
--
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: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [core] throw exception if table is recreated when it still being read [paimon]
Posted by "tsreaper (via GitHub)" <gi...@apache.org>.
tsreaper commented on code in PR #4445:
URL: https://github.com/apache/paimon/pull/4445#discussion_r1828666439
##########
paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/PrimaryKeyFileStoreTableITCase.java:
##########
@@ -335,6 +338,77 @@ public void testBatchJobWithConflictAndRestart() throws Exception {
}
}
+ @Timeout(120)
+ @ParameterizedTest()
+ @ValueSource(booleans = {false, true})
+ public void testRecreateTableWithException(boolean isReloadData) throws Exception {
+ TableEnvironment bEnv = tableEnvironmentBuilder().batchMode().build();
+ bEnv.executeSql(createCatalogSql("testCatalog", path + "/warehouse"));
+ bEnv.executeSql("USE CATALOG testCatalog");
+ bEnv.executeSql(
+ "CREATE TABLE t ( pt INT, k INT, v INT, PRIMARY KEY (pt, k) NOT ENFORCED ) "
+ + "PARTITIONED BY (pt) "
+ + "WITH ("
+ + " 'bucket' = '2'\n"
+ + ")");
Review Comment:
Set `continuous.discovery-interval` to a smaller value (for example `1s` or `500ms`) so this test can run faster. No need to wait for 10 seconds.
##########
paimon-core/src/main/java/org/apache/paimon/utils/NextSnapshotFetcher.java:
##########
@@ -45,8 +45,19 @@ public Snapshot getNextSnapshot(long nextSnapshotId) {
}
Long earliestSnapshotId = snapshotManager.earliestSnapshotId();
+ Long latestSnapshotId = snapshotManager.latestSnapshotId();
// No snapshot now
if (earliestSnapshotId == null || earliestSnapshotId <= nextSnapshotId) {
+ if ((earliestSnapshotId == null && nextSnapshotId > 1)
+ || (latestSnapshotId != null && nextSnapshotId > latestSnapshotId + 1)) {
+ throw new OutOfRangeException(
+ String.format(
+ "The next expected snapshot is too big! Most possible cause might be the table had been recreated."
+ + "The next snapshot id is %d, while the latest snapshot id is %s",
+ nextSnapshotId,
+ latestSnapshotId == null ? "null" : latestSnapshotId));
Review Comment:
```suggestion
latestSnapshotId));
```
##########
paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/PrimaryKeyFileStoreTableITCase.java:
##########
@@ -335,6 +338,77 @@ public void testBatchJobWithConflictAndRestart() throws Exception {
}
}
+ @Timeout(120)
+ @ParameterizedTest()
+ @ValueSource(booleans = {false, true})
+ public void testRecreateTableWithException(boolean isReloadData) throws Exception {
+ TableEnvironment bEnv = tableEnvironmentBuilder().batchMode().build();
+ bEnv.executeSql(createCatalogSql("testCatalog", path + "/warehouse"));
+ bEnv.executeSql("USE CATALOG testCatalog");
+ bEnv.executeSql(
+ "CREATE TABLE t ( pt INT, k INT, v INT, PRIMARY KEY (pt, k) NOT ENFORCED ) "
+ + "PARTITIONED BY (pt) "
+ + "WITH ("
+ + " 'bucket' = '2'\n"
+ + ")");
+
+ TableEnvironment sEnv =
+ tableEnvironmentBuilder()
+ .streamingMode()
+ .parallelism(4)
+ .checkpointIntervalMs(1000)
+ .build();
+ sEnv.executeSql(createCatalogSql("testCatalog", path + "/warehouse"));
+ sEnv.executeSql("USE CATALOG testCatalog");
+
+ // first write
+ List<String> values = new ArrayList<>();
+ for (int i = 0; i < 10; i++) {
+ values.add(String.format("(0, %d, %d)", i, i));
+ values.add(String.format("(1, %d, %d)", i, i));
+ }
+ bEnv.executeSql("INSERT INTO t VALUES " + String.join(", ", values)).await();
+
+ // second write
+ values.clear();
+ for (int i = 0; i < 10; i++) {
+ values.add(String.format("(0, %d, %d)", i, i + 1));
+ values.add(String.format("(1, %d, %d)", i, i + 1));
+ }
+ bEnv.executeSql("INSERT INTO t VALUES " + String.join(", ", values)).await();
+
+ // start a read job
+ CloseableIterator<Row> it = sEnv.executeSql("SELECT * FROM t").collect();
+
+ // wait the read job to read the current table
+ Thread.sleep(10000);
Review Comment:
Assert that all snapshots are read by checking the result of `SELECT`. Don't sleep for this long!
--
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: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [core] throw exception if table is recreated when it still being read [paimon]
Posted by "tsreaper (via GitHub)" <gi...@apache.org>.
tsreaper merged PR #4445:
URL: https://github.com/apache/paimon/pull/4445
--
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: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org