feat: adopt partition summary & add AddEntry variants

including AddEntry, AddExistingEntry, AddDeleteEntry

Author: zhjwpku

.gitignore

@@ -16,6 +16,7 @@
 # under the License.
 
 build/
+cmake-build/
 cmake-build-debug/
 cmake-build-release/
 .DS_Store

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

@@ -19,16 +19,17 @@
 
 #include "iceberg/manifest_adapter.h"
 
+#include <memory>
 #include <utility>
 
 #include <nanoarrow/nanoarrow.h>
 
 #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"
 #include "iceberg/util/checked_cast.h"
 #include "iceberg/util/macros.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,110 @@ ManifestEntryAdapter::~ManifestEntryAdapter() {
   }
 }
 
+Status ManifestEntryAdapter::AddEntry(ManifestEntry& entry) {
+  ICEBERG_RETURN_UNEXPECTED(CheckDataFile(*entry.data_file));
+  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) {
+  ICEBERG_RETURN_UNEXPECTED(CheckDataFile(*entry.data_file));
+  entry.status = ManifestStatus::kDeleted;
+  entry.snapshot_id = snapshot_id_;
+  return AddEntryInternal(entry);
+}
+
+Status ManifestEntryAdapter::AddExistingEntry(ManifestEntry& entry) {
+  ICEBERG_RETURN_UNEXPECTED(CheckDataFile(*entry.data_file));
+  entry.status = ManifestStatus::kExisting;
+  return AddEntryInternal(entry);
+}
+
+ManifestFile ManifestEntryAdapter::ToManifestFile() const {
+  ManifestFile manifest_file;
+  manifest_file.partition_spec_id = partition_spec_->spec_id();
+  manifest_file.content = content_;
+  // sequence_number and min_sequence_number with kInvalidSequenceNumber 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::CheckDataFile(const DataFile& file) const {
+  switch (content_) {
+    case ManifestContent::kData:
+      if (file.content != DataFile::Content::kData) {
+        return InvalidArgument(
+            "Manifest content type: data, data file content should be: data, but got: {}",
+            ToString(file.content));
+      }
+      break;
+    case ManifestContent::kDeletes:
+      if (file.content != DataFile::Content::kPositionDeletes &&
+          file.content != DataFile::Content::kEqualityDeletes) {
+        return InvalidArgument(
+            "Manifest content type: deletes, data file content should be: "
+            "position_deletes or equality_deletes, but got: {}",
+            ToString(file.content));
+      }
+      break;
+    default:
+      std::unreachable();
+  }
+  return {};
+}
+
+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();
+  }
+
+  ICEBERG_RETURN_UNEXPECTED(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,45 @@ 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.
+  ManifestFile ToManifestFile() const;
+
  protected:
+  Status CheckDataFile(const DataFile& file) const;
+  Status AddEntryInternal(const ManifestEntry& entry);
   Status AppendInternal(const ManifestEntry& entry);
   Status AppendDataFile(ArrowArray* array,
                         const std::shared_ptr<StructType>& data_file_type,
@@ -91,10 +117,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`.
@@ -104,7 +139,7 @@ class ICEBERG_EXPORT ManifestFileAdapter : public ManifestAdapter {
   ManifestFileAdapter() = default;
   ~ManifestFileAdapter() override;
 
-  virtual Status Append(const ManifestFile& file) = 0;
+  virtual Status Append(ManifestFile& file) = 0;
 
   const std::shared_ptr<Schema>& schema() const { return manifest_list_schema_; }
 

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(
@@ -323,6 +325,19 @@ struct ICEBERG_EXPORT ManifestEntry {
       std::shared_ptr<StructType> datafile_type);
 };
 
+/// \brief Get the relative datafile content type name
+ICEBERG_EXPORT constexpr std::string_view ToString(DataFile::Content type) noexcept {
+  switch (type) {
+    case DataFile::Content::kData:
+      return "data";
+    case DataFile::Content::kPositionDeletes:
+      return "position_deletes";
+    case DataFile::Content::kEqualityDeletes:
+      return "equality_deletes";
+  }
+  std::unreachable();
+}
+
 /// \brief Get the relative data file content type from int
 ICEBERG_EXPORT constexpr Result<DataFile::Content> DataFileContentFromInt(
     int content) noexcept {