feat: support scan metrics of FileStoreScan to align with ScanMetrics

Author: SteNicholas

src/paimon/core/operation/file_store_scan.cpp

@@ -129,7 +129,7 @@ Result<std::shared_ptr<FileStoreScan::RawPlan>> FileStoreScan::CreatePlan() cons
         }
     }
     const int64_t all_data_files = std::accumulate(
-        all_manifest_file_metas.begin(), all_manifest_file_metas.end(), 0,
+        all_manifest_file_metas.begin(), all_manifest_file_metas.end(), int64_t{0},
         [](const int64_t sum, const ManifestFileMeta& manifest_file_meta) {
             return sum + manifest_file_meta.NumAddedFiles() - manifest_file_meta.NumDeletedFiles();
         });
@@ -138,7 +138,7 @@ Result<std::shared_ptr<FileStoreScan::RawPlan>> FileStoreScan::CreatePlan() cons
                              std::chrono::high_resolution_clock::now() - started)
                              .count());
     metrics_->SetCounter(ScanMetrics::LAST_SCANNED_SNAPSHOT_ID,
-                         snapshot.has_value() ? snapshot.value().Id() : 0);
+                         snapshot.has_value() ? snapshot.value().Id() : int64_t{0});
     metrics_->SetCounter(ScanMetrics::LAST_SCANNED_MANIFESTS, filtered_manifest_file_metas.size());
     metrics_->SetCounter(ScanMetrics::LAST_SCAN_SKIPPED_TABLE_FILES,
                          all_data_files - manifest_entries.size());
@@ -151,14 +151,15 @@ Status FileStoreScan::ReadManifests(std::optional<Snapshot>* snapshot_ptr,
                                     std::vector<ManifestFileMeta>* all_manifests_ptr,
                                     std::vector<ManifestFileMeta>* filter_manifests_ptr) const {
     auto& snapshot = *snapshot_ptr;
-    auto& filtered_manifests = *filter_manifests_ptr;
     auto& all_manifests = *all_manifests_ptr;
+    auto& filtered_manifests = *filter_manifests_ptr;
     if (specified_snapshot_ != std::nullopt) {
         snapshot = specified_snapshot_;
     } else {
         PAIMON_ASSIGN_OR_RAISE(snapshot, snapshot_manager_->LatestSnapshot());
     }
     if (snapshot == std::nullopt) {
+        all_manifests = std::vector<ManifestFileMeta>();
         filtered_manifests = std::vector<ManifestFileMeta>();
         return Status::OK();
     }

src/paimon/core/operation/key_value_file_store_scan_test.cpp

@@ -144,12 +144,27 @@ TEST_F(KeyValueFileStoreScanTest, TestMaxSequenceNumber) {
                              CreateFileStoreScan(table_path, scan_filter,
                                                  /*table_schema_id=*/0, /*snapshot_id=*/2));
 
+        const auto started = std::chrono::high_resolution_clock::now();
         ASSERT_OK_AND_ASSIGN(std::shared_ptr<FileStoreScan::RawPlan> raw_plan, scan->CreatePlan());
         std::shared_ptr<Metrics> metrics = scan->GetScanMetrics();
         ASSERT_TRUE(metrics);
+        ASSERT_OK_AND_ASSIGN(uint64_t last_scan_duration,
+                             metrics->GetCounter(ScanMetrics::LAST_SCAN_DURATION));
+        ASSERT_LE(last_scan_duration, std::chrono::duration_cast<std::chrono::milliseconds>(
+                                          std::chrono::high_resolution_clock::now() - started)
+                                          .count());
         ASSERT_OK_AND_ASSIGN(uint64_t last_scanned_snapshot_id,
                              metrics->GetCounter(ScanMetrics::LAST_SCANNED_SNAPSHOT_ID));
-        ASSERT_EQ(2u, last_scanned_snapshot_id);
+        ASSERT_EQ(last_scanned_snapshot_id, 2u);
+        ASSERT_OK_AND_ASSIGN(uint64_t last_scanned_manifests,
+                             metrics->GetCounter(ScanMetrics::LAST_SCANNED_MANIFESTS));
+        ASSERT_EQ(last_scanned_manifests, 2u);
+        ASSERT_OK_AND_ASSIGN(uint64_t last_scan_skipped_table_files,
+                             metrics->GetCounter(ScanMetrics::LAST_SCAN_SKIPPED_TABLE_FILES));
+        ASSERT_EQ(last_scan_skipped_table_files, 0u);
+        ASSERT_OK_AND_ASSIGN(uint64_t last_scan_resulted_table_files,
+                             metrics->GetCounter(ScanMetrics::LAST_SCAN_RESULTED_TABLE_FILES));
+        ASSERT_EQ(last_scan_resulted_table_files, 2u);
         int64_t max_sequence_num = GetMaxSequenceNumberOfRawPlan(raw_plan);
         ASSERT_EQ(max_sequence_num, 1);
         // test multiple scan