diff --git a/src/paimon/core/table/source/data_evolution_batch_scan.cpp b/src/paimon/core/table/source/data_evolution_batch_scan.cpp
index 2f86769a..91ee97b1 100644
--- a/src/paimon/core/table/source/data_evolution_batch_scan.cpp
+++ b/src/paimon/core/table/source/data_evolution_batch_scan.cpp
@@ -24,6 +24,7 @@
 
 #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"
@@ -44,25 +45,44 @@ 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 (core_options_.GetScanTagName() || core_options_.GetScanTimestampMillis()) {
+        return Status::NotImplemented("Global index scan does not support time travel");
+    }
+    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();
+            if (const std::optional<int64_t>& snapshot_id = core_options_.GetScanSnapshotId()) {
+                PAIMON_ASSIGN_OR_RAISE(Snapshot snapshot,
+                                       snapshot_manager->LoadSnapshot(snapshot_id.value()));
+                global_index_snapshot_id = snapshot.Id();
+            } else {
+                PAIMON_ASSIGN_OR_RAISE(std::optional<Snapshot> snapshot,
+                                       snapshot_manager->LatestSnapshot());
+                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.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());
     std::map<int64_t, float> id_to_score;
@@ -136,16 +156,16 @@ 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();
@@ -164,7 +184,8 @@ Result<std::shared_ptr<GlobalIndexResult>> DataEvolutionBatchScan::EvalGlobalInd
         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
diff --git a/src/paimon/core/table/source/data_evolution_batch_scan.h b/src/paimon/core/table/source/data_evolution_batch_scan.h
index 546ae101..5f0c5a91 100644
--- a/src/paimon/core/table/source/data_evolution_batch_scan.h
+++ b/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_;
diff --git a/test/inte/global_index_test.cpp b/test/inte/global_index_test.cpp
index e05dc4b2..3a8be983 100644
--- a/test/inte/global_index_test.cpp
+++ b/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,60 @@ 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));
+
+    auto empty_index_result = BitmapGlobalIndexResult::FromRanges({});
+    ASSERT_OK_AND_ASSIGN(
+        auto supplied_explicit_plan,
+        ScanGlobalIndexAndData(table_path, /*predicate=*/nullptr,
+                               {{Options::SCAN_SNAPSHOT_ID, "1"}}, empty_index_result));
+    ASSERT_TRUE(supplied_explicit_plan->Splits().empty());
+    ASSERT_EQ(supplied_explicit_plan->SnapshotId(), std::optional<int64_t>(1));
+
+    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().IsNotExist())
+        << nonexistent_snapshot_result.status().ToString();
+}
+
 TEST_P(GlobalIndexTest, TestDataEvolutionBatchScanWithOnlyOnePartitionHasIndex) {
     CreateTable(/*partition_keys=*/{"f1"});
     std::string table_path = PathUtil::JoinPath(dir_->Str(), "foo.db/bar");