feat: adopt partition summary & add AddEntry variants

including AddEntry, AddExistingEntry, AddDeleteEntry

Author: zhjwpku

src/iceberg/CMakeLists.txt

@@ -39,6 +39,7 @@ set(ICEBERG_SOURCES
     name_mapping.cc
     partition_field.cc
     partition_spec.cc
+    partition_summary_internal.cc
     row/arrow_array_wrapper.cc
     row/manifest_wrapper.cc
     schema.cc

src/iceberg/expression/literal.cc

@@ -504,6 +504,17 @@ bool Literal::IsAboveMax() const { return std::holds_alternative<AboveMax>(value
 
 bool Literal::IsNull() const { return std::holds_alternative<std::monostate>(value_); }
 
+bool Literal::IsNan() const {
+  if (type_->type_id() == TypeId::kFloat) {
+    auto val = std::get<float>(value_);
+    return std::isnan(val);
+  } else if (type_->type_id() == TypeId::kDouble) {
+    auto val = std::get<double>(value_);
+    return std::isnan(val);
+  }
+  return false;
+}
+
 // LiteralCaster implementation
 
 Result<Literal> LiteralCaster::CastTo(const Literal& literal,

src/iceberg/expression/literal.h

@@ -150,10 +150,14 @@ class ICEBERG_EXPORT Literal : public util::Formattable {
   /// \return true if this literal represents a BelowMin value, false otherwise
   bool IsBelowMin() const;
 
-  /// Check if this literal is null.
+  /// \brief Check if this literal is null.
   /// \return true if this literal is null, false otherwise
   bool IsNull() const;
 
+  /// \brief Check if this literal is NaN (Not a Number).
+  /// \return true if this literal is NaN, false otherwise
+  bool IsNan() const;
+
   std::string ToString() const override;
 
  private:

src/iceberg/manifest_adapter.cc

@@ -26,6 +26,7 @@
 #include "iceberg/arrow/nanoarrow_status_internal.h"
 #include "iceberg/manifest_entry.h"
 #include "iceberg/manifest_list.h"
+#include "iceberg/partition_summary_internal.h"
 #include "iceberg/result.h"
 #include "iceberg/schema.h"
 #include "iceberg/schema_internal.h"
@@ -141,10 +142,12 @@ Result<ArrowArray*> ManifestAdapter::FinishAppending() {
   return &array_;
 }
 
-ManifestEntryAdapter::ManifestEntryAdapter(std::shared_ptr<PartitionSpec> partition_spec,
+ManifestEntryAdapter::ManifestEntryAdapter(std::optional<int64_t> snapshot_id_,
+                                           std::shared_ptr<PartitionSpec> partition_spec,
                                            std::shared_ptr<Schema> current_schema,
                                            ManifestContent content)
-    : partition_spec_(std::move(partition_spec)),
+    : snapshot_id_(snapshot_id_),
+      partition_spec_(std::move(partition_spec)),
       current_schema_(std::move(current_schema)),
       content_(content) {
   if (!partition_spec_) {
@@ -161,6 +164,82 @@ ManifestEntryAdapter::~ManifestEntryAdapter() {
   }
 }
 
+Status ManifestEntryAdapter::AddEntry(ManifestEntry& entry) {
+  entry.status = ManifestStatus::kAdded;
+  entry.snapshot_id = snapshot_id_;
+  if (entry.sequence_number.has_value() &&
+      entry.sequence_number.value() < TableMetadata::kInitialSequenceNumber) {
+    entry.sequence_number = std::nullopt;
+  }
+  entry.file_sequence_number = std::nullopt;
+  return AddEntryInternal(entry);
+}
+
+Status ManifestEntryAdapter::AddDeleteEntry(ManifestEntry& entry) {
+  entry.status = ManifestStatus::kDeleted;
+  entry.snapshot_id = snapshot_id_;
+  return AddEntryInternal(entry);
+}
+
+Status ManifestEntryAdapter::AddExistingEntry(ManifestEntry& entry) {
+  entry.status = ManifestStatus::kExisting;
+  return AddEntryInternal(entry);
+}
+
+std::unique_ptr<ManifestFile> ManifestEntryAdapter::ToManifestFile() const {
+  auto manifest_file = std::make_unique<ManifestFile>();
+  manifest_file->partition_spec_id = partition_spec_->spec_id();
+  manifest_file->content = content_;
+  // sequence_number and min_sequence_number with UNASSIGNED_SEQUENCE_NUMBER will be
+  // replace with real sequence number in `ManifestListWriter`.
+  manifest_file->sequence_number = TableMetadata::kInvalidSequenceNumber;
+  manifest_file->min_sequence_number =
+      min_sequence_number_.value_or(TableMetadata::kInvalidSequenceNumber);
+  manifest_file->existing_files_count = existing_files_count_;
+  manifest_file->added_snapshot_id = snapshot_id_.value_or(Snapshot::kInvalidSnapshotId);
+  manifest_file->added_files_count = add_files_count_;
+  manifest_file->existing_files_count = existing_files_count_;
+  manifest_file->deleted_files_count = delete_files_count_;
+  manifest_file->added_rows_count = add_rows_count_;
+  manifest_file->existing_rows_count = existing_rows_count_;
+  manifest_file->deleted_rows_count = delete_rows_count_;
+  manifest_file->partitions = std::move(partition_summary_->Summaries());
+  return manifest_file;
+}
+
+Status ManifestEntryAdapter::AddEntryInternal(const ManifestEntry& entry) {
+  if (entry.data_file == nullptr) [[unlikely]] {
+    return InvalidManifest("Missing required data_file field from manifest entry.");
+  }
+  switch (entry.status) {
+    case ManifestStatus::kAdded:
+      add_files_count_++;
+      add_rows_count_ += entry.data_file->record_count;
+      break;
+    case ManifestStatus::kExisting:
+      existing_files_count_++;
+      existing_rows_count_ += entry.data_file->record_count;
+      break;
+    case ManifestStatus::kDeleted:
+      delete_files_count_++;
+      delete_rows_count_ += entry.data_file->record_count;
+      break;
+    default:
+      std::unreachable();
+  }
+
+  partition_summary_->Update(entry.data_file->partition);
+
+  if (entry.IsAlive() && entry.sequence_number.has_value()) {
+    if (!min_sequence_number_.has_value() ||
+        entry.sequence_number.value() < min_sequence_number_.value()) {
+      min_sequence_number_ = entry.sequence_number.value();
+    }
+  }
+
+  return AppendInternal(entry);
+}
+
 Status ManifestEntryAdapter::AppendPartitionValues(
     ArrowArray* array, const std::shared_ptr<StructType>& partition_type,
     const std::vector<Literal>& partition_values) {

src/iceberg/manifest_adapter.h

@@ -25,7 +25,6 @@
 #include <memory>
 #include <optional>
 #include <unordered_map>
-#include <unordered_set>
 #include <vector>
 
 #include "iceberg/arrow_c_data.h"
@@ -61,18 +60,44 @@ class ICEBERG_EXPORT ManifestAdapter {
 /// Implemented by different versions with version-specific schemas.
 class ICEBERG_EXPORT ManifestEntryAdapter : public ManifestAdapter {
  public:
-  ManifestEntryAdapter(std::shared_ptr<PartitionSpec> partition_spec,
+  ManifestEntryAdapter(std::optional<int64_t> snapshot_id_,
+                       std::shared_ptr<PartitionSpec> partition_spec,
                        std::shared_ptr<Schema> current_schema, ManifestContent content);
 
   ~ManifestEntryAdapter() override;
 
-  virtual Status Append(const ManifestEntry& entry) = 0;
+  /// \brief Add a new entry to the manifest.
+  ///
+  /// This method will update following status of the entry:
+  /// - Update the entry status to `Added`
+  /// - Set the snapshot id to the current snapshot id
+  /// - Set the sequence number to nullopt if it is invalid(smaller than 0)
+  /// - Set the file sequence number to nullopt
+  virtual Status AddEntry(ManifestEntry& entry);
+
+  /// \brief Add a delete entry to the manifest.
+  ///
+  /// This method will update following status of the entry:
+  /// - Update the entry status to `Deleted`
+  /// - Set the snapshot id to the current snapshot id
+  virtual Status AddDeleteEntry(ManifestEntry& entry);
+
+  /// \brief Add an existing entry to the manifest.
+  ///
+  /// This method will update following status of the entry:
+  /// - Update the entry status to `Existing`
+  virtual Status AddExistingEntry(ManifestEntry& entry);
 
   const std::shared_ptr<Schema>& schema() const { return manifest_schema_; }
 
   ManifestContent content() const { return content_; }
 
+  /// \brief Create a ManifestFile object without setting file metadata, such as
+  /// location, file size, key metadata, etc.
+  std::unique_ptr<ManifestFile> ToManifestFile() const;
+
  protected:
+  Status AddEntryInternal(const ManifestEntry& entry);
   Status AppendInternal(const ManifestEntry& entry);
   Status AppendDataFile(ArrowArray* array,
                         const std::shared_ptr<StructType>& data_file_type,
@@ -91,10 +116,19 @@ class ICEBERG_EXPORT ManifestEntryAdapter : public ManifestAdapter {
       const DataFile& file) const;
 
  protected:
+  std::optional<int64_t> snapshot_id_;
   std::shared_ptr<PartitionSpec> partition_spec_;
   std::shared_ptr<Schema> current_schema_;
   std::shared_ptr<Schema> manifest_schema_;
   const ManifestContent content_;
+  int32_t add_files_count_{0};
+  int32_t existing_files_count_{0};
+  int32_t delete_files_count_{0};
+  int64_t add_rows_count_{0L};
+  int64_t existing_rows_count_{0L};
+  int64_t delete_rows_count_{0L};
+  std::optional<int64_t> min_sequence_number_{std::nullopt};
+  std::unique_ptr<PartitionSummary> partition_summary_;
 };
 
 /// \brief Adapter for appending a list of `ManifestFile`s to an `ArrowArray`.

src/iceberg/manifest_entry.h

@@ -57,15 +57,6 @@ ICEBERG_EXPORT constexpr Result<ManifestStatus> ManifestStatusFromInt(
   }
 }
 
-enum class ManifestContent {
-  kData = 0,
-  kDeletes = 1,
-};
-
-ICEBERG_EXPORT constexpr std::string_view ToString(ManifestContent content) noexcept;
-ICEBERG_EXPORT constexpr Result<ManifestContent> ManifestContentFromString(
-    std::string_view str) noexcept;
-
 /// \brief DataFile carries data file path, partition tuple, metrics, ...
 struct ICEBERG_EXPORT DataFile {
   /// \brief Content of a data file
@@ -315,6 +306,17 @@ struct ICEBERG_EXPORT ManifestEntry {
   inline static const SchemaField kFileSequenceNumber =
       SchemaField::MakeOptional(4, "file_sequence_number", iceberg::int64());
 
+  /// \brief Check if this manifest entry is deleted.
+  constexpr bool IsAlive() const {
+    return status == ManifestStatus::kAdded || status == ManifestStatus::kExisting;
+  }
+
+  /// \brief Create a copy of this manifest entry.
+  ManifestEntry Copy() const {
+    ManifestEntry copy = *this;
+    return copy;
+  }
+
   bool operator==(const ManifestEntry& other) const;
 
   static std::shared_ptr<StructType> TypeFromPartitionType(

src/iceberg/manifest_list.h

@@ -25,7 +25,6 @@
 #include <optional>
 #include <string>
 #include <string_view>
-#include <utility>
 
 #include "iceberg/iceberg_export.h"
 #include "iceberg/partition_spec.h"
@@ -72,17 +71,17 @@ struct ICEBERG_EXPORT PartitionFieldSummary {
   static const StructType& Type();
 };
 
+/// \brief The type of files tracked by the manifest, either data or delete files; 0 for
+/// all v1 manifests
+enum class ManifestContent {
+  /// The manifest content is data.
+  kData = 0,
+  /// The manifest content is deletes.
+  kDeletes = 1,
+};
+
 /// \brief Entry in a manifest list.
 struct ICEBERG_EXPORT ManifestFile {
-  /// \brief The type of files tracked by the manifest, either data or delete files; 0 for
-  /// all v1 manifests
-  enum class Content {
-    /// The manifest content is data.
-    kData = 0,
-    /// The manifest content is deletes.
-    kDeletes = 1,
-  };
-
   /// Field id: 500
   /// Location of the manifest file
   std::string manifest_path;
@@ -96,7 +95,7 @@ struct ICEBERG_EXPORT ManifestFile {
   /// Field id: 517
   /// The type of files tracked by the manifest, either data or delete files; 0 for all v1
   /// manifests
-  Content content = Content::kData;
+  ManifestContent content = ManifestContent::kData;
   /// Field id: 515
   /// The sequence number when the manifest was added to the table; use 0 when reading v1
   /// manifest lists
@@ -218,21 +217,20 @@ struct ICEBERG_EXPORT ManifestList {
 };
 
 /// \brief Get the relative manifest content type name
-ICEBERG_EXPORT constexpr std::string_view ToString(ManifestFile::Content type) noexcept {
+ICEBERG_EXPORT constexpr std::string_view ToString(ManifestContent type) noexcept {
   switch (type) {
-    case ManifestFile::Content::kData:
+    case ManifestContent::kData:
       return "data";
-    case ManifestFile::Content::kDeletes:
+    case ManifestContent::kDeletes:
       return "deletes";
   }
-  std::unreachable();
 }
 
 /// \brief Get the relative manifest content type from name
-ICEBERG_EXPORT constexpr Result<ManifestFile::Content> ManifestFileContentFromString(
+ICEBERG_EXPORT constexpr Result<ManifestContent> ManifestContentFromString(
     std::string_view str) noexcept {
-  if (str == "data") return ManifestFile::Content::kData;
-  if (str == "deletes") return ManifestFile::Content::kDeletes;
+  if (str == "data") return ManifestContent::kData;
+  if (str == "deletes") return ManifestContent::kDeletes;
   return InvalidArgument("Invalid manifest content type: {}", str);
 }
 

src/iceberg/manifest_reader_internal.cc

@@ -237,7 +237,7 @@ Result<std::vector<ManifestFile>> ParseManifestList(ArrowSchema* schema,
         break;
       case ManifestFileField::kContent:
         PARSE_PRIMITIVE_FIELD(manifest_files[row_idx].content, view_of_column,
-                              ManifestFile::Content);
+                              ManifestContent);
         break;
       case ManifestFileField::kSequenceNumber:
         PARSE_PRIMITIVE_FIELD(manifest_files[row_idx].sequence_number, view_of_column,