fix(scan): preserve snapshot id for empty data evolution index plans

src/paimon/core/table/source/data_evolution_batch_scan.cpp

@@ -24,11 +24,57 @@
 
 #include "paimon/core/global_index/global_index_scan_impl.h"
 #include "paimon/core/global_index/indexed_split_impl.h"
+#include "paimon/core/snapshot.h"
 #include "paimon/core/table/source/data_split_impl.h"
+#include "paimon/core/table/source/snapshot/static_from_snapshot_starting_scanner.h"
 #include "paimon/core/utils/snapshot_manager.h"
 #include "paimon/global_index/bitmap_global_index_result.h"
 
 namespace paimon {
+namespace {
+
+bool UsesUnsupportedTimeTravel(const CoreOptions& core_options) {
+    const StartupMode startup_mode = core_options.GetStartupMode();
+    if (startup_mode == StartupMode::FromTimestamp()) {
+        return core_options.GetScanTimestampMillis().has_value();
+    }
+    return startup_mode == StartupMode::FromSnapshot() &&
+           !core_options.GetScanSnapshotId().has_value() &&
+           core_options.GetScanTagName().has_value();
+}
+
+Result<std::optional<Snapshot>> ResolveGlobalIndexScanSnapshot(
+    const CoreOptions& core_options, const std::shared_ptr<SnapshotManager>& snapshot_manager) {
+    const StartupMode startup_mode = core_options.GetStartupMode();
+    if (startup_mode == StartupMode::FromSnapshot() ||
+        startup_mode == StartupMode::FromSnapshotFull()) {
+        if (const std::optional<int64_t>& snapshot_id = core_options.GetScanSnapshotId()) {
+            return StaticFromSnapshotStartingScanner::ResolveSnapshot(snapshot_manager,
+                                                                      snapshot_id.value());
+        }
+        if (startup_mode == StartupMode::FromSnapshotFull()) {
+            return Status::Invalid(
+                "scan.snapshot-id must be set when startup mode is FROM_SNAPSHOT_FULL");
+        }
+        if (!core_options.GetScanTagName()) {
+            return Status::Invalid(
+                "scan.snapshot-id or scan.tag-name must be set when startup mode is "
+                "FROM_SNAPSHOT");
+        }
+    } else if (startup_mode == StartupMode::FromTimestamp() &&
+               !core_options.GetScanTimestampMillis()) {
+        return Status::Invalid(
+            "scan.timestamp-millis or scan.timestamp must be set when startup mode is "
+            "FROM_TIMESTAMP");
+    }
+
+    // Tag and timestamp scans are rejected only when the predicate uses a Global Index. Use the
+    // latest snapshot here to check whether an applicable index exists.
+    return snapshot_manager->LatestSnapshot();
+}
+
+}  // namespace
+
 DataEvolutionBatchScan::DataEvolutionBatchScan(
     const std::string& table_path, const std::shared_ptr<SnapshotReader>& snapshot_reader,
     std::unique_ptr<DataTableBatchScan>&& batch_scan,
@@ -44,27 +90,43 @@ DataEvolutionBatchScan::DataEvolutionBatchScan(
       executor_(executor) {}
 
 Result<std::shared_ptr<Plan>> DataEvolutionBatchScan::CreatePlan() {
-    std::optional<std::vector<Range>> row_ranges;
+    std::optional<int64_t> global_index_snapshot_id;
     std::shared_ptr<GlobalIndexResult> final_global_index_result = global_index_result_;
     if (!final_global_index_result) {
-        PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<GlobalIndexResult> index_result, EvalGlobalIndex());
-        if (index_result) {
-            final_global_index_result = index_result;
-            PAIMON_ASSIGN_OR_RAISE(row_ranges, index_result->ToRanges());
+        PAIMON_ASSIGN_OR_RAISE(std::optional<EvaluatedGlobalIndex> evaluated_index,
+                               EvalGlobalIndex());
+        if (evaluated_index) {
+            final_global_index_result = evaluated_index->result;
+            global_index_snapshot_id = evaluated_index->snapshot_id;
         }
-    } else {
-        PAIMON_ASSIGN_OR_RAISE(row_ranges, final_global_index_result->ToRanges());
     }
-    if (!row_ranges) {
+    if (!final_global_index_result) {
         return batch_scan_->CreatePlan();
     }
-    if (row_ranges.value().empty()) {
-        return PlanImpl::EmptyPlan();
+    if (UsesUnsupportedTimeTravel(core_options_)) {
+        return Status::NotImplemented("Global index scan does not support time travel");
     }
-    PAIMON_ASSIGN_OR_RAISE(RowRangeIndex row_range_index,
-                           RowRangeIndex::Create(row_ranges.value()));
+    PAIMON_ASSIGN_OR_RAISE(std::vector<Range> row_ranges, final_global_index_result->ToRanges());
+    if (row_ranges.empty()) {
+        if (!global_index_snapshot_id) {
+            const std::shared_ptr<SnapshotManager>& snapshot_manager =
+                snapshot_reader_->GetSnapshotManager();
+            PAIMON_ASSIGN_OR_RAISE(std::optional<Snapshot> snapshot,
+                                   ResolveGlobalIndexScanSnapshot(core_options_, snapshot_manager));
+            if (!snapshot) {
+                return PlanImpl::EmptyPlan();
+            }
+            global_index_snapshot_id = snapshot->Id();
+        }
+        return std::make_shared<PlanImpl>(global_index_snapshot_id,
+                                          std::vector<std::shared_ptr<Split>>());
+    }
+    PAIMON_ASSIGN_OR_RAISE(RowRangeIndex row_range_index, RowRangeIndex::Create(row_ranges));
     batch_scan_->WithRowRangeIndex(row_range_index);
     PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<Plan> data_plan, batch_scan_->CreatePlan());
+    if (global_index_snapshot_id && data_plan->SnapshotId() != global_index_snapshot_id) {
+        return Status::Invalid("Global index and data scan resolved different snapshots");
+    }
     std::map<int64_t, float> id_to_score;
     if (auto scored_result =
             std::dynamic_pointer_cast<ScoredGlobalIndexResult>(final_global_index_result)) {
@@ -136,35 +198,30 @@ Result<std::shared_ptr<Plan>> DataEvolutionBatchScan::WrapToIndexedSplits(
     return std::make_shared<PlanImpl>(data_plan->SnapshotId(), indexed_splits);
 }
 
-Result<std::shared_ptr<GlobalIndexResult>> DataEvolutionBatchScan::EvalGlobalIndex() const {
+Result<std::optional<DataEvolutionBatchScan::EvaluatedGlobalIndex>>
+DataEvolutionBatchScan::EvalGlobalIndex() const {
     auto predicate = batch_scan_->GetNonPartitionPredicate();
     if (!predicate) {
-        return std::shared_ptr<GlobalIndexResult>(nullptr);
+        return std::optional<EvaluatedGlobalIndex>();
     }
     if (!core_options_.GlobalIndexEnabled()) {
-        return std::shared_ptr<GlobalIndexResult>(nullptr);
+        return std::optional<EvaluatedGlobalIndex>();
     }
     auto partition_filter = batch_scan_->GetPartitionPredicate();
-    // TODO(lisizhuo.lsz): support time travel
-    std::optional<Snapshot> snapshot;
     const std::shared_ptr<SnapshotManager>& snapshot_manager =
         snapshot_reader_->GetSnapshotManager();
-    if (const std::optional<int64_t>& snapshot_id = core_options_.GetScanSnapshotId()) {
-        PAIMON_ASSIGN_OR_RAISE(Snapshot loaded_snapshot,
-                               snapshot_manager->LoadSnapshot(snapshot_id.value()));
-        snapshot = std::move(loaded_snapshot);
-    } else {
-        PAIMON_ASSIGN_OR_RAISE(snapshot, snapshot_manager->LatestSnapshot());
-    }
+    PAIMON_ASSIGN_OR_RAISE(std::optional<Snapshot> snapshot,
+                           ResolveGlobalIndexScanSnapshot(core_options_, snapshot_manager));
     if (!snapshot) {
-        return Status::Invalid("not found latest snapshot");
+        return std::optional<EvaluatedGlobalIndex>();
     }
 
     PAIMON_ASSIGN_OR_RAISE(
         std::unique_ptr<GlobalIndexScanImpl> index_scan,
         GlobalIndexScanImpl::Create(table_path_, table_schema_, snapshot.value(), partition_filter,
                                     core_options_, executor_, pool_));
-    return index_scan->Scan(predicate);
+    PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<GlobalIndexResult> result, index_scan->Scan(predicate));
+    return std::optional<EvaluatedGlobalIndex>(EvaluatedGlobalIndex{result, snapshot->Id()});
 }
 
 }  // namespace paimon

src/paimon/core/table/source/data_evolution_batch_scan.h

@@ -21,6 +21,7 @@
 #include <cstdint>
 #include <map>
 #include <memory>
+#include <optional>
 #include <utility>
 #include <vector>
 
@@ -51,7 +52,12 @@ class DataEvolutionBatchScan : public AbstractTableScan {
         const std::map<int64_t, float>& id_to_score);
 
  private:
-    Result<std::shared_ptr<GlobalIndexResult>> EvalGlobalIndex() const;
+    struct EvaluatedGlobalIndex {
+        std::shared_ptr<GlobalIndexResult> result;
+        int64_t snapshot_id;
+    };
+
+    Result<std::optional<EvaluatedGlobalIndex>> EvalGlobalIndex() const;
 
  private:
     std::shared_ptr<MemoryPool> pool_;

src/paimon/core/table/source/snapshot/static_from_snapshot_starting_scanner.h

@@ -20,6 +20,7 @@
 #pragma once
 
 #include <memory>
+#include <utility>
 
 #include "paimon/core/table/source/snapshot/starting_scanner.h"
 #include "paimon/logging.h"
@@ -36,29 +37,36 @@ class StaticFromSnapshotStartingScanner : public StartingScanner {
         starting_snapshot_id_ = snapshot_id;
     }
 
-    Result<std::shared_ptr<ScanResult>> Scan(
-        const std::shared_ptr<SnapshotReader>& snapshot_reader) override {
+    static Result<std::optional<Snapshot>> ResolveSnapshot(
+        const std::shared_ptr<SnapshotManager>& snapshot_manager, int64_t snapshot_id) {
         PAIMON_ASSIGN_OR_RAISE(std::optional<int64_t> earliest,
-                               snapshot_manager_->EarliestSnapshotId());
-        PAIMON_ASSIGN_OR_RAISE(std::optional<int64_t> latest,
-                               snapshot_manager_->LatestSnapshotId());
-        if (earliest == std::nullopt || latest == std::nullopt) {
-            PAIMON_LOG_INFO(
-                logger_, "There is currently no snapshot. Waiting for snapshot generation.%s", "");
-            return std::make_shared<StartingScanner::NoSnapshot>();
+                               snapshot_manager->EarliestSnapshotId());
+        PAIMON_ASSIGN_OR_RAISE(std::optional<int64_t> latest, snapshot_manager->LatestSnapshotId());
+        if (!earliest || !latest) {
+            return std::optional<Snapshot>();
         }
-        if (starting_snapshot_id_.value() < earliest.value() ||
-            starting_snapshot_id_.value() > latest.value()) {
+        if (snapshot_id < earliest.value() || snapshot_id > latest.value()) {
             return Status::Invalid(
                 fmt::format("The specified scan snapshotId {} is out of "
                             "available snapshotId range [{}, {}].",
-                            starting_snapshot_id_.value(), earliest.value(), latest.value()));
+                            snapshot_id, earliest.value(), latest.value()));
+        }
+        PAIMON_ASSIGN_OR_RAISE(Snapshot snapshot, snapshot_manager->LoadSnapshot(snapshot_id));
+        return std::optional<Snapshot>(std::move(snapshot));
+    }
+
+    Result<std::shared_ptr<ScanResult>> Scan(
+        const std::shared_ptr<SnapshotReader>& snapshot_reader) override {
+        PAIMON_ASSIGN_OR_RAISE(std::optional<Snapshot> snapshot,
+                               ResolveSnapshot(snapshot_manager_, starting_snapshot_id_.value()));
+        if (!snapshot) {
+            PAIMON_LOG_INFO(
+                logger_, "There is currently no snapshot. Waiting for snapshot generation.%s", "");
+            return std::make_shared<StartingScanner::NoSnapshot>();
         }
-        PAIMON_ASSIGN_OR_RAISE(Snapshot snapshot,
-                               snapshot_manager_->LoadSnapshot(starting_snapshot_id_.value()));
         PAIMON_ASSIGN_OR_RAISE(
             std::shared_ptr<Plan> plan,
-            snapshot_reader->WithMode(ScanMode::ALL)->WithSnapshot(snapshot)->Read());
+            snapshot_reader->WithMode(ScanMode::ALL)->WithSnapshot(snapshot.value())->Read());
         return std::make_shared<StartingScanner::CurrentSnapshot>(plan);
     }
 

test/inte/global_index_test.cpp

@@ -15,6 +15,9 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
+
+#include <limits>
+
 #include "arrow/type.h"
 #include "gtest/gtest.h"
 #include "paimon/common/factories/io_hook.h"
@@ -1509,6 +1512,83 @@ TEST_P(GlobalIndexTest, TestDataEvolutionBatchScan) {
     }
 }
 
+TEST_P(GlobalIndexTest, TestDataEvolutionGlobalIndexSnapshotSelection) {
+    CreateTable();
+    std::string table_path = PathUtil::JoinPath(dir_->Str(), "foo.db/bar");
+    auto schema = arrow::schema(fields_);
+    std::vector<std::string> write_cols = schema->field_names();
+    auto src_array = arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(fields_), R"([
+["Alice", 10, 1, 11.1],
+["Bob", 20, 0, 12.1]
+    ])")
+                         .ValueOrDie();
+
+    ASSERT_OK_AND_ASSIGN(auto commit_msgs, WriteArray(table_path, write_cols, src_array));
+    ASSERT_OK(Commit(table_path, commit_msgs));
+    ASSERT_OK(WriteIndex(table_path, /*partition_filters=*/{}, "f0", "bitmap", /*options=*/{},
+                         Range(0, 1)));
+
+    auto predicate =
+        PredicateBuilder::Equal(/*field_index=*/0, /*field_name=*/"f0", FieldType::STRING,
+                                Literal(FieldType::STRING, "missing", 7));
+
+    ASSERT_OK_AND_ASSIGN(auto latest_plan, ScanGlobalIndexAndData(table_path, predicate));
+    ASSERT_TRUE(latest_plan->Splits().empty());
+    ASSERT_EQ(latest_plan->SnapshotId(), std::optional<int64_t>(2));
+
+    const std::map<std::string, std::string> explicit_latest_options = {
+        {Options::SCAN_MODE, "latest"},
+        {Options::SCAN_SNAPSHOT_ID, "999"},
+        {Options::SCAN_TAG_NAME, "ignored"},
+        {Options::SCAN_TIMESTAMP_MILLIS, "0"}};
+    ASSERT_OK_AND_ASSIGN(auto explicit_latest_plan,
+                         ScanGlobalIndexAndData(table_path, predicate, explicit_latest_options));
+    ASSERT_TRUE(explicit_latest_plan->Splits().empty());
+    ASSERT_EQ(explicit_latest_plan->SnapshotId(), std::optional<int64_t>(2));
+
+    auto empty_index_result = BitmapGlobalIndexResult::FromRanges({});
+    ASSERT_OK_AND_ASSIGN(auto supplied_latest_plan,
+                         ScanGlobalIndexAndData(table_path, /*predicate=*/nullptr,
+                                                explicit_latest_options, empty_index_result));
+    ASSERT_EQ(supplied_latest_plan->SnapshotId(), std::optional<int64_t>(2));
+
+    ASSERT_OK_AND_ASSIGN(auto supplied_explicit_plan,
+                         ScanGlobalIndexAndData(table_path, /*predicate=*/nullptr,
+                                                {{Options::SCAN_SNAPSHOT_ID, "1"},
+                                                 {Options::SCAN_TAG_NAME, "ignored"},
+                                                 {Options::SCAN_TIMESTAMP_MILLIS, "0"}},
+                                                empty_index_result));
+    ASSERT_TRUE(supplied_explicit_plan->Splits().empty());
+    ASSERT_EQ(supplied_explicit_plan->SnapshotId(), std::optional<int64_t>(1));
+
+    Result<std::shared_ptr<Plan>> missing_selector_result =
+        ScanGlobalIndexAndData(table_path, /*predicate=*/nullptr,
+                               {{Options::SCAN_MODE, "from-snapshot"}}, empty_index_result);
+    ASSERT_TRUE(missing_selector_result.status().IsInvalid())
+        << missing_selector_result.status().ToString();
+
+    std::vector<std::map<std::string, std::string>> time_travel_options = {
+        {{Options::SCAN_TAG_NAME, "tag"}},
+        {{Options::SCAN_TIMESTAMP_MILLIS, std::to_string(std::numeric_limits<int64_t>::max())}}};
+    for (const auto& options : time_travel_options) {
+        Result<std::shared_ptr<Plan>> result =
+            ScanGlobalIndexAndData(table_path, /*predicate=*/nullptr, options, empty_index_result);
+        ASSERT_TRUE(result.status().IsNotImplemented()) << result.status().ToString();
+    }
+
+    auto unindexed_predicate = PredicateBuilder::Equal(/*field_index=*/3, /*field_name=*/"f3",
+                                                       FieldType::DOUBLE, Literal(99.9));
+    ASSERT_OK_AND_ASSIGN(auto fallback_plan, ScanGlobalIndexAndData(table_path, unindexed_predicate,
+                                                                    time_travel_options.back()));
+    ASSERT_EQ(fallback_plan->SnapshotId(), std::optional<int64_t>(2));
+
+    Result<std::shared_ptr<Plan>> nonexistent_snapshot_result =
+        ScanGlobalIndexAndData(table_path, /*predicate=*/nullptr,
+                               {{Options::SCAN_SNAPSHOT_ID, "999"}}, empty_index_result);
+    ASSERT_TRUE(nonexistent_snapshot_result.status().IsInvalid())
+        << nonexistent_snapshot_result.status().ToString();
+}
+
 TEST_P(GlobalIndexTest, TestDataEvolutionBatchScanWithOnlyOnePartitionHasIndex) {
     CreateTable(/*partition_keys=*/{"f1"});
     std::string table_path = PathUtil::JoinPath(dir_->Str(), "foo.db/bar");