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"
@@ -161,6 +162,39 @@ ManifestEntryAdapter::~ManifestEntryAdapter() {
   }
 }
 
+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_data_sequence_number_.has_value() ||
+        entry.sequence_number.value() < min_data_sequence_number_.value()) {
+      min_data_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"
@@ -66,13 +65,16 @@ class ICEBERG_EXPORT ManifestEntryAdapter : public ManifestAdapter {
 
   ~ManifestEntryAdapter() override;
 
-  virtual Status Append(const ManifestEntry& entry) = 0;
+  virtual Status AddEntry(ManifestEntry& entry) = 0;
+  virtual Status AddDeleteEntry(ManifestEntry& entry) = 0;
+  virtual Status AddExistingEntry(ManifestEntry& entry) = 0;
 
   const std::shared_ptr<Schema>& schema() const { return manifest_schema_; }
 
   ManifestContent content() const { return content_; }
 
  protected:
+  Status AddEntryInternal(const ManifestEntry& entry);
   Status AppendInternal(const ManifestEntry& entry);
   Status AppendDataFile(ArrowArray* array,
                         const std::shared_ptr<StructType>& data_file_type,
@@ -95,6 +97,14 @@ class ICEBERG_EXPORT ManifestEntryAdapter : public ManifestAdapter {
   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_data_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,

src/iceberg/manifest_writer.cc

@@ -29,18 +29,39 @@
 
 namespace iceberg {
 
-Status ManifestWriter::Add(const ManifestEntry& entry) {
+Status ManifestWriter::AddEntry(const ManifestEntry& entry) {
   if (adapter_->size() >= kBatchSize) {
     ICEBERG_ASSIGN_OR_RAISE(auto array, adapter_->FinishAppending());
     ICEBERG_RETURN_UNEXPECTED(writer_->Write(array));
     ICEBERG_RETURN_UNEXPECTED(adapter_->StartAppending());
   }
-  return adapter_->Append(entry);
+  auto copy = entry.Copy();
+  return adapter_->AddEntry(copy);
+}
+
+Status ManifestWriter::AddDelete(const ManifestEntry& entry) {
+  if (adapter_->size() >= kBatchSize) {
+    ICEBERG_ASSIGN_OR_RAISE(auto array, adapter_->FinishAppending());
+    ICEBERG_RETURN_UNEXPECTED(writer_->Write(array));
+    ICEBERG_RETURN_UNEXPECTED(adapter_->StartAppending());
+  }
+  auto copy = entry.Copy();
+  return adapter_->AddDeleteEntry(copy);
+}
+
+Status ManifestWriter::AddExisting(const ManifestEntry& entry) {
+  if (adapter_->size() >= kBatchSize) {
+    ICEBERG_ASSIGN_OR_RAISE(auto array, adapter_->FinishAppending());
+    ICEBERG_RETURN_UNEXPECTED(writer_->Write(array));
+    ICEBERG_RETURN_UNEXPECTED(adapter_->StartAppending());
+  }
+  auto copy = entry.Copy();
+  return adapter_->AddExistingEntry(copy);
 }
 
 Status ManifestWriter::AddAll(const std::vector<ManifestEntry>& entries) {
   for (const auto& entry : entries) {
-    ICEBERG_RETURN_UNEXPECTED(Add(entry));
+    ICEBERG_RETURN_UNEXPECTED(AddEntry(entry));
   }
   return {};
 }
@@ -53,6 +74,8 @@ Status ManifestWriter::Close() {
   return writer_->Close();
 }
 
+std::optional<Metrics> ManifestWriter::Metrics() const { return writer_->metrics(); }
+
 ManifestContent ManifestWriter::content() const { return adapter_->content(); }
 
 Result<std::unique_ptr<Writer>> OpenFileWriter(

src/iceberg/manifest_writer.h

@@ -28,6 +28,7 @@
 #include "iceberg/file_writer.h"
 #include "iceberg/iceberg_export.h"
 #include "iceberg/manifest_adapter.h"
+#include "iceberg/metrics.h"
 #include "iceberg/type_fwd.h"
 
 namespace iceberg {
@@ -41,10 +42,20 @@ class ICEBERG_EXPORT ManifestWriter {
 
   ~ManifestWriter() = default;
 
-  /// \brief Write manifest entry to file.
+  /// \brief Add a new Manifest entry.
   /// \param entry Manifest entry to write.
   /// \return Status::OK() if entry was written successfully
-  Status Add(const ManifestEntry& entry);
+  Status AddEntry(const ManifestEntry& entry);
+
+  /// \brief Add an existing Manifest entry.
+  /// \param entry Manifest entry to write.
+  /// \return Status::OK() if entry was written successfully
+  Status AddExisting(const ManifestEntry& entry);
+
+  /// \brief Add a delete Manifest entry.
+  /// \param entry Manifest entry to write.
+  /// \return Status::OK() if entry was written successfully
+  Status AddDelete(const ManifestEntry& entry);
 
   /// \brief Write manifest entries to file.
   /// \param entries Manifest entries to write.
@@ -54,6 +65,9 @@ class ICEBERG_EXPORT ManifestWriter {
   /// \brief Close writer and flush to storage.
   Status Close();
 
+  /// \brief Get the metrics of written manifest file.
+  std::optional<Metrics> Metrics() const;
+
   /// \brief Get the content of the manifest.
   ManifestContent content() const;