fix: define IncrementalScan's PlanFiles as pure virtual and no need instantiate in header

Author: WZhuo

src/iceberg/manifest/manifest_list.h

@@ -230,7 +230,9 @@ struct ICEBERG_EXPORT ManifestFile {
       kFirstRowIdFieldId, "first_row_id", int64(),
       "Starting row ID to assign to new rows in ADDED data files");
 
-  bool operator==(const ManifestFile& other) const = default;
+  bool operator==(const ManifestFile& other) const {
+    return manifest_path == other.manifest_path;
+  }
 
   static const std::shared_ptr<StructType>& Type();
 };
@@ -275,7 +277,7 @@ ICEBERG_EXPORT inline constexpr Result<ManifestContent> ManifestContentFromStrin
 
 namespace std {
 template <>
-struct std::hash<iceberg::ManifestFile> {
+struct hash<iceberg::ManifestFile> {
   size_t operator()(const iceberg::ManifestFile& manifest_file) const {
     return std::hash<std::string>{}(manifest_file.manifest_path);
   }

src/iceberg/table_scan.cc

@@ -639,6 +639,30 @@ Result<std::vector<std::shared_ptr<FileScanTask>>> DataTableScan::PlanFiles() co
   return manifest_group->PlanFiles();
 }
 
+// Friend function template for IncrementalScan that implements the shared PlanFiles
+// logic. It resolves the from/to snapshot range from the scan context and delegates
+// to the two-arg virtual PlanFiles() override in the concrete subclass.
+// Defined as a friend to access the protected two-arg PlanFiles().
+template <typename ScanTaskType>
+Result<std::vector<std::shared_ptr<ScanTaskType>>> ResolvePlanFiles(
+    const IncrementalScan<ScanTaskType>& scan) {
+  if (IsScanCurrentLineage(scan.context())) {
+    if (scan.metadata()->current_snapshot_id == kInvalidSnapshotId) {
+      return std::vector<std::shared_ptr<ScanTaskType>>{};
+    }
+  }
+
+  ICEBERG_ASSIGN_OR_RAISE(
+      int64_t to_snapshot_id_inclusive,
+      internal::ToSnapshotIdInclusive(scan.context(), *scan.metadata()));
+  ICEBERG_ASSIGN_OR_RAISE(
+      std::optional<int64_t> from_snapshot_id_exclusive,
+      internal::FromSnapshotIdExclusive(scan.context(), *scan.metadata(),
+                                        to_snapshot_id_inclusive));
+
+  return scan.PlanFiles(from_snapshot_id_exclusive, to_snapshot_id_inclusive);
+}
+
 // IncrementalAppendScan implementation
 
 Result<std::unique_ptr<IncrementalAppendScan>> IncrementalAppendScan::Make(
@@ -651,6 +675,11 @@ Result<std::unique_ptr<IncrementalAppendScan>> IncrementalAppendScan::Make(
       std::move(metadata), std::move(schema), std::move(io), std::move(context)));
 }
 
+Result<std::vector<std::shared_ptr<FileScanTask>>> IncrementalAppendScan::PlanFiles()
+    const {
+  return ResolvePlanFiles<FileScanTask>(*this);
+}
+
 Result<std::vector<std::shared_ptr<FileScanTask>>> IncrementalAppendScan::PlanFiles(
     std::optional<int64_t> from_snapshot_id_exclusive,
     int64_t to_snapshot_id_inclusive) const {
@@ -725,35 +754,15 @@ Result<std::unique_ptr<IncrementalChangelogScan>> IncrementalChangelogScan::Make
   return NotImplemented("IncrementalChangelogScan is not implemented");
 }
 
+Result<std::vector<std::shared_ptr<ChangelogScanTask>>>
+IncrementalChangelogScan::PlanFiles() const {
+  return ResolvePlanFiles<ChangelogScanTask>(*this);
+}
+
 Result<std::vector<std::shared_ptr<ChangelogScanTask>>>
 IncrementalChangelogScan::PlanFiles(std::optional<int64_t> from_snapshot_id_exclusive,
                                     int64_t to_snapshot_id_inclusive) const {
   return NotImplemented("IncrementalChangelogScan::PlanFiles is not implemented");
 }
 
-// Explicit template implementations for IncrementalScan
-// This moves the template implementation from header to source file
-template <typename ScanTaskType>
-Result<std::vector<std::shared_ptr<ScanTaskType>>>
-IncrementalScan<ScanTaskType>::PlanFiles() const {
-  if (IsScanCurrentLineage(context_)) {
-    ICEBERG_ASSIGN_OR_RAISE(auto current_snapshot, metadata_->Snapshot());
-    if (current_snapshot == nullptr) {
-      return std::vector<std::shared_ptr<ScanTaskType>>{};
-    }
-  }
-
-  ICEBERG_ASSIGN_OR_RAISE(int64_t to_snapshot_id_inclusive,
-                          internal::ToSnapshotIdInclusive(context_, *metadata_));
-  ICEBERG_ASSIGN_OR_RAISE(
-      std::optional<int64_t> from_snapshot_id_exclusive,
-      internal::FromSnapshotIdExclusive(context_, *metadata_, to_snapshot_id_inclusive));
-
-  return PlanFiles(from_snapshot_id_exclusive, to_snapshot_id_inclusive);
-}
-
-// Explicitly instantiate the templates
-template class IncrementalScan<FileScanTask>;
-template class IncrementalScan<ChangelogScanTask>;
-
 }  // namespace iceberg

src/iceberg/table_scan.h

@@ -356,24 +356,24 @@ class ICEBERG_EXPORT DataTableScan : public TableScan {
 /// \brief A base template class for incremental scans that read changes between
 /// snapshots, and return scan tasks of the specified type.
 template <typename ScanTaskType>
-class ICEBERG_EXPORT IncrementalScan : public TableScan {
+class IncrementalScan : public TableScan {
  public:
   ~IncrementalScan() override = default;
 
-  /// \brief Plans the scan tasks by resolving manifests and data files.
-  /// \return A Result containing scan tasks or an error.
-  Result<std::vector<std::shared_ptr<ScanTaskType>>> PlanFiles() const;
+  virtual Result<std::vector<std::shared_ptr<ScanTaskType>>> PlanFiles() const = 0;
 
  protected:
   virtual Result<std::vector<std::shared_ptr<ScanTaskType>>> PlanFiles(
       std::optional<int64_t> from_snapshot_id_exclusive,
       int64_t to_snapshot_id_inclusive) const = 0;
 
   using TableScan::TableScan;
-};
 
-extern template class IncrementalScan<FileScanTask>;
-extern template class IncrementalScan<ChangelogScanTask>;
+  // Allow the free function ResolvePlanFiles to access protected members.
+  template <typename T>
+  friend Result<std::vector<std::shared_ptr<T>>> ResolvePlanFiles(
+      const IncrementalScan<T>& scan);
+};
 
 /// \brief A scan that reads data files added between snapshots (incremental appends).
 class ICEBERG_EXPORT IncrementalAppendScan : public IncrementalScan<FileScanTask> {
@@ -385,8 +385,7 @@ class ICEBERG_EXPORT IncrementalAppendScan : public IncrementalScan<FileScanTask
 
   ~IncrementalAppendScan() override = default;
 
-  // Bring the public PlanFiles() from base class into scope
-  using IncrementalScan<FileScanTask>::PlanFiles;
+  Result<std::vector<std::shared_ptr<FileScanTask>>> PlanFiles() const override;
 
  protected:
   Result<std::vector<std::shared_ptr<FileScanTask>>> PlanFiles(
@@ -407,8 +406,7 @@ class ICEBERG_EXPORT IncrementalChangelogScan
 
   ~IncrementalChangelogScan() override = default;
 
-  // Bring the public PlanFiles() from base class into scope
-  using IncrementalScan<ChangelogScanTask>::PlanFiles;
+  Result<std::vector<std::shared_ptr<ChangelogScanTask>>> PlanFiles() const override;
 
  protected:
   Result<std::vector<std::shared_ptr<ChangelogScanTask>>> PlanFiles(