perf(read): reduce snapshot and option parsing overhead

src/paimon/common/utils/options_utils.h

@@ -50,43 +50,37 @@ class OptionsUtils {
     template <typename T>
     static Result<T> GetValueFromMap(const std::map<std::string, std::string>& key_value_map,
                                      const std::string& key, const T& default_value) {
-        auto value = GetValueFromMap<T>(key_value_map, key);
-        if (value.ok()) {
-            return value.value();
-        } else if (value.status().IsNotExist()) {
-            return default_value;
-        }
-        return value.status();
+        PAIMON_ASSIGN_OR_RAISE(std::optional<T> value,
+                               GetOptionalValueFromMap<T>(key_value_map, key));
+        return value.value_or(default_value);
     }
 
     template <typename T>
     static Result<T> GetValueFromMap(const std::map<std::string, std::string>& key_value_map,
                                      const std::string& key) {
-        static_assert(is_supported_type<T>::value, "T must be trivially copyable or string");
-        auto iter = key_value_map.find(key);
-        if (iter == key_value_map.end()) {
+        PAIMON_ASSIGN_OR_RAISE(std::optional<T> value,
+                               GetOptionalValueFromMap<T>(key_value_map, key));
+        if (!value) {
             return Status::NotExist(fmt::format("key {} does not exist in map", key));
         }
-        const auto& value_str = iter->second;
-        std::optional<T> value = StringUtils::StringToValue<T>(value_str);
-        if (value == std::nullopt) {
-            return Status::Invalid(fmt::format("convert key {}, value {} to {} failed", key,
-                                               value_str, GetTypeName<T>()));
-        }
         return value.value();
     }
 
     template <typename T>
     static Result<std::optional<T>> GetOptionalValueFromMap(
         const std::map<std::string, std::string>& key_value_map, const std::string& key) {
-        Result<T> value = GetValueFromMap<T>(key_value_map, key);
-        if (value.ok()) {
-            return std::optional<T>(value.value());
-        }
-        if (value.status().IsNotExist()) {
+        static_assert(is_supported_type<T>::value, "T must be trivially copyable or string");
+        auto iter = key_value_map.find(key);
+        if (iter == key_value_map.end()) {
             return std::optional<T>();
         }
-        return value.status();
+        const auto& value_str = iter->second;
+        std::optional<T> value = StringUtils::StringToValue<T>(value_str);
+        if (value == std::nullopt) {
+            return Status::Invalid(fmt::format("convert key {}, value {} to {} failed", key,
+                                               value_str, GetTypeName<T>()));
+        }
+        return value;
     }
 
     static Result<std::string> GetNonEmptyValueFromMap(

src/paimon/common/utils/options_utils_test.cpp

@@ -61,6 +61,16 @@ TEST(OptionsUtilsTest, TestGetValueFromMap) {
     ASSERT_OK_AND_ASSIGN(auto empty,
                          OptionsUtils::GetValueFromMap<int32_t>(key_value_map, "", 999));
     ASSERT_EQ(999, empty);
+
+    ASSERT_OK_AND_ASSIGN(auto present,
+                         OptionsUtils::GetValueFromMap<int32_t>(key_value_map, "key_int", 233));
+    ASSERT_EQ(10, present);
+    ASSERT_TRUE(
+        OptionsUtils::GetValueFromMap<int32_t>(key_value_map, "missing").status().IsNotExist());
+    key_value_map["empty_string"] = "";
+    ASSERT_OK_AND_ASSIGN(std::string empty_string, OptionsUtils::GetValueFromMap<std::string>(
+                                                       key_value_map, "empty_string", "default"));
+    ASSERT_TRUE(empty_string.empty());
 }
 
 TEST(OptionsUtilsTest, TestGetOptionalValueFromMap) {

src/paimon/core/operation/file_store_scan.h

@@ -110,6 +110,10 @@ class FileStoreScan {
         return this;
     }
 
+    const std::optional<Snapshot>& GetSpecifiedSnapshot() const {
+        return specified_snapshot_;
+    }
+
     FileStoreScan* WithLevelFilter(const std::function<bool(int32_t)>& level_filter) {
         level_filter_ = level_filter;
         return this;

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

@@ -86,7 +86,11 @@ Result<std::shared_ptr<Plan>> PrimaryKeyIndexBatchScan::CreatePlan() {
     int64_t snapshot_id = data_plan->SnapshotId().value();
     const std::shared_ptr<SnapshotManager>& snapshot_manager =
         snapshot_reader_->GetSnapshotManager();
-    Result<Snapshot> snapshot_result = snapshot_manager->LoadSnapshot(snapshot_id);
+    // The data scan already loaded this snapshot. Reuse it for index planning.
+    const std::optional<Snapshot>& planned_snapshot = snapshot_reader_->GetSpecifiedSnapshot();
+    Result<Snapshot> snapshot_result = planned_snapshot && planned_snapshot->Id() == snapshot_id
+                                           ? Result<Snapshot>(planned_snapshot.value())
+                                           : snapshot_manager->LoadSnapshot(snapshot_id);
     if (!snapshot_result.ok()) {
         static auto logger = Logger::GetLogger("PrimaryKeyIndexBatchScan");
         PAIMON_LOG_WARN(logger,

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

@@ -92,6 +92,10 @@ class SnapshotReader {
         return scan_->GetSnapshotManager();
     }
 
+    const std::optional<Snapshot>& GetSpecifiedSnapshot() const {
+        return scan_->GetSpecifiedSnapshot();
+    }
+
     const std::unique_ptr<IndexFileHandler>& GetIndexFileHandler() const {
         return index_file_handler_;
     }

test/inte/primary_key_sorted_index_inte_test.cpp

@@ -75,6 +75,11 @@ class TrackingLocalFileSystem : public LocalFileSystem {
         return index_paths;
     }
 
+    int64_t OpenCount(const std::string& path) const {
+        std::scoped_lock lock(mutex_);
+        return std::count(opened_paths_.begin(), opened_paths_.end(), path);
+    }
+
  private:
     mutable std::mutex mutex_;
     mutable std::vector<std::string> opened_paths_;
@@ -292,11 +297,15 @@ TEST_P(PrimaryKeySortedIndexInteTest, HistoricalSnapshotsMixedFilesAndDisabledIn
     std::shared_ptr<Predicate> predicate = ScoreEqual(/*partitioned=*/false, 0);
     const std::vector<int64_t> snapshot_ids = {2, 3, 5};
     for (int64_t snapshot_id : snapshot_ids) {
+        auto tracking_file_system = std::make_shared<TrackingLocalFileSystem>();
         ASSERT_OK_AND_ASSIGN(
             std::shared_ptr<Plan> plan,
             Scan(/*partitioned=*/false, predicate, snapshot_id, /*index_enabled=*/true,
-                 /*partition_filters=*/{}, /*branch=*/"", /*file_system=*/nullptr));
+                 /*partition_filters=*/{}, /*branch=*/"", tracking_file_system));
         ASSERT_EQ(snapshot_id, plan->SnapshotId());
+        ASSERT_EQ(1, tracking_file_system->OpenCount(
+                         PathUtil::JoinPath(TablePath(/*partitioned=*/false),
+                                            fmt::format("snapshot/snapshot-{}", snapshot_id))));
         const auto [indexed_count, data_count] = CountSplitKinds(plan);
         ASSERT_GT(indexed_count, 0);
         if (snapshot_id == 3) {