[core] throw exception if table is recreated when it still being read

paimon-core/src/main/java/org/apache/paimon/utils/NextSnapshotFetcher.java

@@ -45,8 +45,23 @@ 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) {
+                throw new OutOfRangeException(
+                        String.format(
+                                "The earliest snapshot is null now, but the next expected snapshot id is %d. "
+                                        + "Most possible cause might be the table had been recreated.",
+                                nextSnapshotId));
+            }
+            if (latestSnapshotId != null && nextSnapshotId > latestSnapshotId + 1) {
+                throw new OutOfRangeException(
+                        String.format(
+                                "The next expected snapshot with id %d is greater than latest snapshot with id %d plus one. "
+                                        + "Most possible cause might be the table had been recreated.",
+                                nextSnapshotId, latestSnapshotId));
+            }
             LOG.debug(
                     "Next snapshot id {} does not exist, wait for the snapshot generation.",
                     nextSnapshotId);

paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/PrimaryKeyFileStoreTableITCase.java

@@ -23,6 +23,7 @@
 import org.apache.paimon.flink.util.AbstractTestBase;
 import org.apache.paimon.fs.Path;
 import org.apache.paimon.fs.local.LocalFileIO;
+import org.apache.paimon.table.source.OutOfRangeException;
 import org.apache.paimon.utils.FailingFileIO;
 
 import org.apache.flink.api.common.JobStatus;
@@ -37,6 +38,8 @@
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.Timeout;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
 
 import java.io.IOException;
 import java.nio.file.Files;
@@ -52,6 +55,7 @@
 import java.util.stream.Stream;
 
 import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatCode;
 
 /** Tests for changelog table with primary keys. */
 public class PrimaryKeyFileStoreTableITCase extends AbstractTestBase {
@@ -335,6 +339,82 @@ 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);
+
+        // 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"
+                        + ")");
+
+        String exceptionMsg =
+                "The earliest snapshot is null now, but the next expected snapshot id is 3. "
+                        + "Most possible cause might be the table had been recreated.";
+        // if reload data, it will generate a new snapshot for recreated table
+        if (isReloadData) {
+            bEnv.executeSql("INSERT INTO t VALUES " + String.join(", ", values)).await();
+            exceptionMsg =
+                    "The next expected snapshot with id 3 is greater than latest snapshot with id 1 plus one. "
+                            + "Most possible cause might be the table had been recreated.";
+        }
+        assertThatCode(
+                        () -> {
+                            while (true) {
+                                if (it.hasNext()) {
+                                    it.next();
+                                }
+                            }
+                        })
+                .hasRootCauseInstanceOf(OutOfRangeException.class)
+                .hasRootCauseMessage(exceptionMsg);
+    }
+
     @Test
     @Timeout(120)
     public void testChangelogCompactInBatchWrite() throws Exception {