chore(log): add logging for scanner and compact manager

src/paimon/core/append/bucketed_append_compact_manager.cpp

@@ -41,8 +41,7 @@ BucketedAppendCompactManager::BucketedAppendCompactManager(
           [](const std::shared_ptr<DataFileMeta>& lhs, const std::shared_ptr<DataFileMeta>& rhs) {
               return lhs->min_sequence_number > rhs->min_sequence_number;
           }),
-      cancellation_controller_(cancellation_controller),
-      logger_(Logger::GetLogger("BucketedAppendCompactManager")) {
+      cancellation_controller_(cancellation_controller) {
     assert(cancellation_controller_ != nullptr);
     for (const auto& file : restored) {
         to_compact_.push(file);

src/paimon/core/append/bucketed_append_compact_manager.h

@@ -61,7 +61,13 @@ class BucketedAppendCompactManager : public CompactFutureManager {
                 return false;
             }
             if (!ignore_overlap && IsOverlap(lhs, rhs)) {
-                // TODO(yonghao.fyh): add log
+                static auto logger = Logger::GetLogger("BucketedAppendCompactManager");
+                PAIMON_LOG_WARN(logger,
+                                "Overlap detected between files %s and %s, "
+                                "sequence range [%ld, %ld] and [%ld, %ld].",
+                                lhs->file_name.c_str(), rhs->file_name.c_str(),
+                                lhs->min_sequence_number, lhs->max_sequence_number,
+                                rhs->min_sequence_number, rhs->max_sequence_number);
             }
             return lhs->min_sequence_number < rhs->min_sequence_number;
         };
@@ -200,7 +206,6 @@ class BucketedAppendCompactManager : public CompactFutureManager {
     std::optional<std::vector<std::shared_ptr<DataFileMeta>>> compacting_;
     DataFileMetaPriorityQueue to_compact_;
     std::shared_ptr<CancellationController> cancellation_controller_;
-    std::unique_ptr<Logger> logger_;
 };
 
 }  // namespace paimon

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

@@ -19,21 +19,29 @@
 #include <memory>
 
 #include "paimon/core/table/source/snapshot/follow_up_scanner.h"
+#include "paimon/logging.h"
 
 namespace paimon {
 class DeltaFollowUpScanner : public FollowUpScanner {
  public:
+    DeltaFollowUpScanner() : logger_(Logger::GetLogger("DeltaFollowUpScanner")) {}
+
     bool NeedScanSnapshot(const Snapshot& snapshot) const override {
         if (snapshot.GetCommitKind() == Snapshot::CommitKind::Append()) {
             return true;
         }
-        // TODO(liancheng.lsz): log
+        PAIMON_LOG_DEBUG(
+            logger_, "Ignore snapshot #%ld with commit kind %s in delta follow-up scanner.",
+            snapshot.Id(), Snapshot::CommitKind::ToString(snapshot.GetCommitKind()).c_str());
         return false;
     }
     Result<std::shared_ptr<Plan>> Scan(
         const Snapshot& snapshot,
         const std::shared_ptr<SnapshotReader>& snapshot_reader) const override {
         return snapshot_reader->WithMode(ScanMode::DELTA)->WithSnapshot(snapshot)->Read();
     }
+
+ private:
+    std::unique_ptr<Logger> logger_;
 };
 }  // namespace paimon

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

@@ -19,6 +19,7 @@
 #include <memory>
 
 #include "paimon/core/table/source/snapshot/starting_scanner.h"
+#include "paimon/logging.h"
 
 namespace paimon {
 /// `StartingScanner` for the `StartupMode::FromSnapshot()` or `StartupMode::FromSnapshotFull()`
@@ -27,7 +28,8 @@ class StaticFromSnapshotStartingScanner : public StartingScanner {
  public:
     StaticFromSnapshotStartingScanner(const std::shared_ptr<SnapshotManager>& snapshot_manager,
                                       int64_t snapshot_id)
-        : StartingScanner(snapshot_manager) {
+        : StartingScanner(snapshot_manager),
+          logger_(Logger::GetLogger("StaticFromSnapshotStartingScanner")) {
         starting_snapshot_id_ = snapshot_id;
     }
 
@@ -38,7 +40,8 @@ class StaticFromSnapshotStartingScanner : public StartingScanner {
         PAIMON_ASSIGN_OR_RAISE(std::optional<int64_t> latest,
                                snapshot_manager_->LatestSnapshotId());
         if (earliest == std::nullopt || latest == std::nullopt) {
-            // TODO(liancheng.lsz): Log
+            PAIMON_LOG_INFO(
+                logger_, "There is currently no snapshot. Waiting for snapshot generation.%s", "");
             return std::make_shared<StartingScanner::NoSnapshot>();
         }
         if (starting_snapshot_id_.value() < earliest.value() ||
@@ -55,5 +58,8 @@ class StaticFromSnapshotStartingScanner : public StartingScanner {
             snapshot_reader->WithMode(ScanMode::ALL)->WithSnapshot(snapshot)->Read());
         return std::make_shared<StartingScanner::CurrentSnapshot>(plan);
     }
+
+ private:
+    std::unique_ptr<Logger> logger_;
 };
 }  // namespace paimon