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

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

@@ -24,11 +24,26 @@
 
 #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/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();
+}
+
+}  // namespace
+
 DataEvolutionBatchScan::DataEvolutionBatchScan(
     const std::string& table_path, const std::shared_ptr<SnapshotReader>& snapshot_reader,
     std::unique_ptr<DataTableBatchScan>&& batch_scan,
@@ -44,27 +59,33 @@ 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();
+    PAIMON_ASSIGN_OR_RAISE(std::vector<Range> row_ranges, final_global_index_result->ToRanges());
+    if (row_ranges.empty()) {
+        if (!global_index_snapshot_id) {
+            return PlanImpl::EmptyPlan();
+        }
+        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.value()));
+    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 +157,55 @@ 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>();
+    }
+    if (UsesUnsupportedTimeTravel(core_options_)) {
+        return Status::NotImplemented("Global index scan does not support time travel");
     }
     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()) {
+    const StartupMode startup_mode = core_options_.GetStartupMode();
+    std::optional<Snapshot> snapshot;
+    if (startup_mode == StartupMode::FromSnapshot() ||
+        startup_mode == StartupMode::FromSnapshotFull()) {
+        const std::optional<int64_t>& snapshot_id = core_options_.GetScanSnapshotId();
+        if (!snapshot_id) {
+            if (startup_mode == StartupMode::FromSnapshotFull()) {
+                return Status::Invalid(
+                    "scan.snapshot-id must be set when startup mode is FROM_SNAPSHOT_FULL");
+            }
+            return Status::Invalid(
+                "scan.snapshot-id or scan.tag-name must be set when startup mode is "
+                "FROM_SNAPSHOT");
+        }
         PAIMON_ASSIGN_OR_RAISE(Snapshot loaded_snapshot,
                                snapshot_manager->LoadSnapshot(snapshot_id.value()));
         snapshot = std::move(loaded_snapshot);
+    } else if (startup_mode == StartupMode::FromTimestamp()) {
+        return Status::Invalid(
+            "scan.timestamp-millis or scan.timestamp must be set when startup mode is "
+            "FROM_TIMESTAMP");
     } else {
         PAIMON_ASSIGN_OR_RAISE(snapshot, snapshot_manager->LatestSnapshot());
     }
     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_;

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"
@@ -1504,6 +1507,99 @@ 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_TRUE(supplied_latest_plan->Splits().empty());
+    ASSERT_FALSE(supplied_latest_plan->SnapshotId());
+
+    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_FALSE(supplied_explicit_plan->SnapshotId());
+
+    ASSERT_NOK_WITH_MSG(
+        ScanGlobalIndexAndData(table_path, predicate, {{Options::SCAN_MODE, "from-snapshot"}}),
+        "scan.snapshot-id or scan.tag-name must be set when startup mode is FROM_SNAPSHOT");
+
+    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, options);
+        ASSERT_TRUE(result.status().IsNotImplemented()) << result.status().ToString();
+
+        ASSERT_OK_AND_ASSIGN(
+            std::shared_ptr<Plan> supplied_plan,
+            ScanGlobalIndexAndData(table_path, /*predicate=*/nullptr, options, empty_index_result));
+        ASSERT_TRUE(supplied_plan->Splits().empty());
+        ASSERT_FALSE(supplied_plan->SnapshotId());
+    }
+
+    auto unindexed_predicate = PredicateBuilder::Equal(/*field_index=*/3, /*field_name=*/"f3",
+                                                       FieldType::DOUBLE, Literal(99.9));
+    Result<std::shared_ptr<Plan>> unindexed_time_travel_result =
+        ScanGlobalIndexAndData(table_path, unindexed_predicate, time_travel_options.back());
+    ASSERT_TRUE(unindexed_time_travel_result.status().IsNotImplemented())
+        << unindexed_time_travel_result.status().ToString();
+
+    ASSERT_OK_AND_ASSIGN(
+        std::shared_ptr<Plan> no_index_plan,
+        ScanGlobalIndexAndData(table_path, /*predicate=*/nullptr, time_travel_options.back()));
+    ASSERT_EQ(no_index_plan->SnapshotId(), std::optional<int64_t>(2));
+
+    ASSERT_OK_AND_ASSIGN(
+        std::shared_ptr<Plan> supplied_nonexistent_snapshot_plan,
+        ScanGlobalIndexAndData(table_path, /*predicate=*/nullptr,
+                               {{Options::SCAN_SNAPSHOT_ID, "999"}}, empty_index_result));
+    ASSERT_TRUE(supplied_nonexistent_snapshot_plan->Splits().empty());
+    ASSERT_FALSE(supplied_nonexistent_snapshot_plan->SnapshotId());
+
+    ASSERT_NOK_WITH_MSG(
+        ScanGlobalIndexAndData(table_path, predicate, {{Options::SCAN_SNAPSHOT_ID, "999"}}),
+        "snapshot-999' not exists");
+}
+
 TEST_P(GlobalIndexTest, TestDataEvolutionBatchScanWithOnlyOnePartitionHasIndex) {
     CreateTable(/*partition_keys=*/{"f1"});
     std::string table_path = PathUtil::JoinPath(dir_->Str(), "foo.db/bar");