feat: add partition summary and write added/existing/deleted entries to manifest writer (#317)

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

@@ -42,6 +42,7 @@ set(ICEBERG_SOURCES
     name_mapping.cc
     partition_field.cc
     partition_spec.cc
+    partition_summary.cc
     row/arrow_array_wrapper.cc
     row/manifest_wrapper.cc
     row/struct_like.cc

src/iceberg/avro/avro_writer.cc

@@ -154,19 +154,19 @@ Status AvroWriter::Close() {
   return {};
 }
 
-std::optional<Metrics> AvroWriter::metrics() {
+Result<Metrics> AvroWriter::metrics() {
   if (impl_->Closed()) {
     // TODO(xiao.dong) implement metrics
     return {};
   }
-  return std::nullopt;
+  return Invalid("AvroWriter is not closed");
 }
 
-std::optional<int64_t> AvroWriter::length() {
+Result<int64_t> AvroWriter::length() {
   if (impl_->Closed()) {
     return impl_->length();
   }
-  return std::nullopt;
+  return Invalid("AvroWriter is not closed");
 }
 
 std::vector<int64_t> AvroWriter::split_offsets() { return {}; }

src/iceberg/avro/avro_writer.h

@@ -37,9 +37,9 @@ class ICEBERG_BUNDLE_EXPORT AvroWriter : public Writer {
 
   Status Write(ArrowArray* data) final;
 
-  std::optional<Metrics> metrics() final;
+  Result<Metrics> metrics() final;
 
-  std::optional<int64_t> length() final;
+  Result<int64_t> length() final;
 
   std::vector<int64_t> split_offsets() final;
 

src/iceberg/expression/literal.h

@@ -150,11 +150,11 @@ 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;
 
-  /// Check if this literal is NaN.
+  /// \brief Check if this literal is NaN.
   /// \return true if this literal is NaN, false otherwise
   bool IsNaN() const;
 

src/iceberg/file_writer.h

@@ -97,11 +97,11 @@ class ICEBERG_EXPORT Writer {
 
   /// \brief Get the file statistics.
   /// Only valid after the file is closed.
-  virtual std::optional<Metrics> metrics() = 0;
+  virtual Result<Metrics> metrics() = 0;
 
   /// \brief Get the file length.
   /// Only valid after the file is closed.
-  virtual std::optional<int64_t> length() = 0;
+  virtual Result<int64_t> length() = 0;
 
   /// \brief Returns a list of recommended split locations, if applicable, empty
   /// otherwise. When available, this information is used for planning scan tasks whose

src/iceberg/manifest_adapter.cc

@@ -19,6 +19,7 @@
 
 #include "iceberg/manifest_adapter.h"
 
+#include <memory>
 #include <utility>
 
 #include <nanoarrow/nanoarrow.h>
@@ -28,7 +29,6 @@
 #include "iceberg/manifest_list.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 +141,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_) {
@@ -386,6 +388,10 @@ Result<std::optional<int64_t>> ManifestEntryAdapter::GetContentSizeInBytes(
 }
 
 Status ManifestEntryAdapter::AppendInternal(const ManifestEntry& entry) {
+  if (entry.data_file == nullptr) [[unlikely]] {
+    return InvalidManifest("Missing required data_file field from manifest entry.");
+  }
+
   const auto& fields = manifest_schema_->fields();
   for (size_t i = 0; i < fields.size(); i++) {
     const auto& field = fields[i];

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,7 +60,8 @@ 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;
@@ -72,6 +72,12 @@ class ICEBERG_EXPORT ManifestEntryAdapter : public ManifestAdapter {
 
   ManifestContent content() const { return content_; }
 
+  std::optional<int64_t> snapshot_id() const { return snapshot_id_; }
+
+  const std::shared_ptr<PartitionSpec>& partition_spec() const { return partition_spec_; }
+
+  const std::shared_ptr<StructType>& partition_type() const { return partition_type_; }
+
  protected:
   Status AppendInternal(const ManifestEntry& entry);
   Status AppendDataFile(ArrowArray* array,
@@ -91,8 +97,10 @@ 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<StructType> partition_type_;
   std::shared_ptr<Schema> manifest_schema_;
   const ManifestContent content_;
 };

src/iceberg/manifest_entry.cc

@@ -22,7 +22,6 @@
 #include <memory>
 #include <vector>
 
-#include "iceberg/schema.h"
 #include "iceberg/schema_field.h"
 #include "iceberg/type.h"
 

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
@@ -277,6 +268,7 @@ struct ICEBERG_EXPORT DataFile {
 
   bool operator==(const DataFile& other) const = default;
 
+  /// \brief Get the schema of the data file with the given partition type.
   static std::shared_ptr<StructType> Type(std::shared_ptr<StructType> partition_type);
 };
 
@@ -315,6 +307,33 @@ 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;
+  }
+
+  ManifestEntry AsAdded() const {
+    ManifestEntry copy = *this;
+    copy.status = ManifestStatus::kAdded;
+    if (copy.data_file->first_row_id.has_value()) {
+      copy.data_file = std::make_unique<DataFile>(*copy.data_file);
+      copy.data_file->first_row_id = std::nullopt;
+    }
+    return copy;
+  }
+
+  ManifestEntry AsExisting() const {
+    ManifestEntry copy = *this;
+    copy.status = ManifestStatus::kExisting;
+    return copy;
+  }
+
+  ManifestEntry AsDeleted() const {
+    ManifestEntry copy = *this;
+    copy.status = ManifestStatus::kDeleted;
+    return copy;
+  }
+
   bool operator==(const ManifestEntry& other) const;
 
   static std::shared_ptr<StructType> TypeFromPartitionType(
@@ -323,6 +342,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 {