rebase and resolve conflicts

Author: wgtmac

src/iceberg/data/data_writer.cc

@@ -23,6 +23,7 @@
 
 #include "iceberg/file_writer.h"
 #include "iceberg/manifest/manifest_entry.h"
+#include "iceberg/partition_spec.h"
 #include "iceberg/util/macros.h"
 
 namespace iceberg {
@@ -43,18 +44,11 @@ class DataWriter::Impl {
     return std::unique_ptr<Impl>(new Impl(std::move(options), std::move(writer)));
   }
 
-  Status Write(ArrowArray* data) {
-    ICEBERG_DCHECK(writer_, "Writer not initialized");
-    return writer_->Write(data);
-  }
+  Status Write(ArrowArray* data) { return writer_->Write(data); }
 
-  Result<int64_t> Length() const {
-    ICEBERG_DCHECK(writer_, "Writer not initialized");
-    return writer_->length();
-  }
+  Result<int64_t> Length() const { return writer_->length(); }
 
   Status Close() {
-    ICEBERG_DCHECK(writer_, "Writer not initialized");
     if (closed_) {
       // Idempotent: no-op if already closed
       return {};
@@ -100,6 +94,8 @@ class DataWriter::Impl {
         .upper_bounds = std::move(upper_bounds_map),
         .split_offsets = std::move(split_offsets),
         .sort_order_id = options_.sort_order_id,
+        .partition_spec_id =
+            options_.spec ? std::make_optional(options_.spec->spec_id()) : std::nullopt,
     });
 
     FileWriter::WriteResult result;

src/iceberg/data/equality_delete_writer.cc

@@ -44,20 +44,12 @@ class EqualityDeleteWriter::Impl {
     return std::unique_ptr<Impl>(new Impl(std::move(options), std::move(writer)));
   }
 
-  Status Write(ArrowArray* data) {
-    ICEBERG_DCHECK(writer_, "Writer not initialized");
-    return writer_->Write(data);
-  }
+  Status Write(ArrowArray* data) { return writer_->Write(data); }
 
-  Result<int64_t> Length() const {
-    ICEBERG_DCHECK(writer_, "Writer not initialized");
-    return writer_->length();
-  }
+  Result<int64_t> Length() const { return writer_->length(); }
 
   Status Close() {
-    ICEBERG_DCHECK(writer_, "Writer not initialized");
     if (closed_) {
-      // Idempotent: no-op if already closed
       return {};
     }
     ICEBERG_RETURN_UNEXPECTED(writer_->Close());
@@ -84,6 +76,7 @@ class EqualityDeleteWriter::Impl {
       upper_bounds_map[col_id] = std::move(serialized);
     }
 
+    // TODO(anyone): add encryption key metadata for encrypted delete files
     auto data_file = std::make_shared<DataFile>(DataFile{
         .content = DataFile::Content::kEqualityDeletes,
         .file_path = options_.path,

src/iceberg/data/equality_delete_writer.h

@@ -50,7 +50,7 @@ struct ICEBERG_EXPORT EqualityDeleteWriterOptions {
   std::vector<int32_t> equality_field_ids;
   std::optional<int32_t> sort_order_id;
   std::unordered_map<std::string, std::string> properties;
-  // TODO: add key_metadata for encryption
+  // TODO(anyone): add key_metadata for encryption
 };
 
 /// \brief Writer for Iceberg equality delete files.

src/iceberg/data/position_delete_writer.cc

@@ -30,6 +30,7 @@
 #include "iceberg/file_writer.h"
 #include "iceberg/manifest/manifest_entry.h"
 #include "iceberg/metadata_columns.h"
+#include "iceberg/partition_spec.h"
 #include "iceberg/schema.h"
 #include "iceberg/schema_internal.h"
 #include "iceberg/util/macros.h"
@@ -154,6 +155,8 @@ class PositionDeleteWriter::Impl {
         .split_offsets = std::move(split_offsets),
         .sort_order_id = std::nullopt,
         .referenced_data_file = std::move(referenced_data_file),
+        .partition_spec_id =
+            options_.spec ? std::make_optional(options_.spec->spec_id()) : std::nullopt,
     });
 
     FileWriter::WriteResult result;

src/iceberg/type_fwd.h

@@ -142,20 +142,22 @@ class ManifestEvaluator;
 class ResidualEvaluator;
 class StrictMetricsEvaluator;
 
-/// \brief Scan.
+/// \brief Scan task.
 class ChangelogScanTask;
-class DataTableScan;
 class FileScanTask;
+class ScanTask;
+
+/// \brief Table scan
+class DataTableScan;
 template <typename ScanTaskType>
 class IncrementalScan;
 class IncrementalAppendScan;
 class IncrementalChangelogScan;
-class ScanTask;
 class TableScan;
+
+/// \brief Scan builder.
 template <typename ScanType>
 class TableScanBuilder;
-
-// Type aliases for incremental scan builders
 using DataTableScanBuilder = TableScanBuilder<DataTableScan>;
 using IncrementalAppendScanBuilder = TableScanBuilder<IncrementalAppendScan>;
 using IncrementalChangelogScanBuilder = TableScanBuilder<IncrementalChangelogScan>;