(paimon-cpp) branch main updated: fix(scan): handle missing latest snapshot in range errors (#398)

[lxy]

This is an automated email from the ASF dual-hosted git repository.

lxy-9602 pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/paimon-cpp.git


The following commit(s) were added to refs/heads/main by this push:
     new d7c5f607 fix(scan): handle missing latest snapshot in range errors 
(#398)
d7c5f607 is described below

commit d7c5f607900ffd63b2e067800dc4b9097c24d888
Author: QuakeWang <45645138+quakew...@users.noreply.github.com>
AuthorDate: Tue Sep 29 22:13:50 2026 +0800

    fix(scan): handle missing latest snapshot in range errors (#398)
---
 .../core/table/source/data_table_stream_scan.cpp   |  3 +-
 src/paimon/core/table/source/table_scan_test.cpp   | 52 ++++++++++++++++++++++
 2 files changed, 54 insertions(+), 1 deletion(-)

diff --git a/src/paimon/core/table/source/data_table_stream_scan.cpp 
b/src/paimon/core/table/source/data_table_stream_scan.cpp
index bae9800d..439c302f 100644
--- a/src/paimon/core/table/source/data_table_stream_scan.cpp
+++ b/src/paimon/core/table/source/data_table_stream_scan.cpp
@@ -122,7 +122,8 @@ Result<std::optional<Snapshot>> 
DataTableStreamScan::GetNextSnapshot(
             return Status::Invalid(fmt::format(
                 "The next expected snapshot is too big! Most possible cause 
might be the table had "
                 "been recreated. The next snapshot id is {}, while the latest 
snapshot id is {}",
-                next_snapshot_id, latest.value()));
+                next_snapshot_id,
+                latest == std::nullopt ? "null" : fmt::format("{}", 
latest.value())));
         }
         return std::optional<Snapshot>();
     }
diff --git a/src/paimon/core/table/source/table_scan_test.cpp 
b/src/paimon/core/table/source/table_scan_test.cpp
index dbc5b992..435e2254 100644
--- a/src/paimon/core/table/source/table_scan_test.cpp
+++ b/src/paimon/core/table/source/table_scan_test.cpp
@@ -26,6 +26,7 @@
 
 #include "gtest/gtest.h"
 #include "paimon/defs.h"
+#include "paimon/fs/local/local_file_system.h"
 #include "paimon/metrics.h"
 #include "paimon/scan_context.h"
 #include "paimon/status.h"
@@ -76,6 +77,57 @@ TEST(TableScanTest, TestNonExistTable) {
     ASSERT_NOK_WITH_MSG(TableScan::Create(std::move(context)), "not found 
latest schema");
 }
 
+TEST(TableScanTest, TestStreamingSnapshotsRemoved) {
+    auto dir = UniqueTestDirectory::Create("local");
+    ASSERT_TRUE(dir);
+    ASSERT_TRUE(TestUtil::CopyDirectory(GetDataDir() + 
"/orc/append_09.db/append_09", dir->Str()));
+    ScanContextBuilder builder(dir->Str());
+    builder.AddOption(Options::FILE_FORMAT, "orc")
+        .AddOption(Options::SCAN_MODE, "latest")
+        .WithStreamingMode(true);
+    ASSERT_OK_AND_ASSIGN(std::unique_ptr<ScanContext> context, 
builder.Finish());
+    ASSERT_OK_AND_ASSIGN(std::unique_ptr<TableScan> table_scan,
+                         TableScan::Create(std::move(context)));
+    ASSERT_OK_AND_ASSIGN(std::shared_ptr<Plan> initial_plan, 
table_scan->CreatePlan());
+    ASSERT_TRUE(initial_plan->Splits().empty());
+    ASSERT_OK_AND_ASSIGN(std::shared_ptr<Plan> waiting_plan, 
table_scan->CreatePlan());
+    ASSERT_TRUE(waiting_plan->Splits().empty());
+
+    LocalFileSystem fs;
+    ASSERT_OK(fs.Delete(dir->Str() + "/snapshot", /*recursive=*/true));
+    ASSERT_NOK_WITH_MSG(table_scan->CreatePlan(),
+                        "The next snapshot id is 6, while the latest snapshot 
id is null");
+}
+
+TEST(TableScanTest, TestStreamingNextSnapshotBoundary) {
+    std::string path = GetDataDir() + "/orc/append_09.db/append_09";
+    for (const std::string next_snapshot_id : {"6", "7"}) {
+        SCOPED_TRACE(next_snapshot_id);
+        ScanContextBuilder builder(path);
+        builder.AddOption(Options::FILE_FORMAT, "orc")
+            .AddOption(Options::SCAN_MODE, "from-snapshot")
+            .AddOption(Options::SCAN_SNAPSHOT_ID, next_snapshot_id)
+            .WithStreamingMode(true);
+        ASSERT_OK_AND_ASSIGN(std::unique_ptr<ScanContext> context, 
builder.Finish());
+        ASSERT_OK_AND_ASSIGN(std::unique_ptr<TableScan> table_scan,
+                             TableScan::Create(std::move(context)));
+        ASSERT_OK_AND_ASSIGN(std::shared_ptr<Plan> initial_plan, 
table_scan->CreatePlan());
+        ASSERT_TRUE(initial_plan->Splits().empty());
+
+        auto result = table_scan->CreatePlan();
+        if (next_snapshot_id == "6") {
+            ASSERT_OK(result);
+            ASSERT_TRUE(result.value()->Splits().empty());
+            ASSERT_FALSE(result.value()->SnapshotId());
+        } else {
+            ASSERT_FALSE(result.ok());
+            ASSERT_TRUE(result.status().IsInvalid());
+            ASSERT_NOK_WITH_MSG(result,
+                                "The next snapshot id is 7, while the latest 
snapshot id is 5");
+        }
+    }
+}
+
 TEST(TableScanTest, TestPkSchemaEvolutionScan) {
     std::string path =
         paimon::test::GetDataDir() + 
"/orc/pk_table_with_alter_table.db/pk_table_with_alter_table/";