fix: align merging snapshot updates and manifest managers with Java

src/iceberg/deletes/dv_util.cc

@@ -17,6 +17,7 @@
  * under the License.
  */
 
+#include <atomic>
 #include <cstddef>
 #include <cstdint>
 #include <format>
@@ -39,17 +40,41 @@
 #include "iceberg/result.h"
 #include "iceberg/util/content_file_util.h"
 #include "iceberg/util/macros.h"
+#include "iceberg/util/task_group.h"
 #include "iceberg/version.h"
 
 namespace iceberg {
 
 Result<std::vector<std::shared_ptr<DataFile>>> DVUtil::MergeAndWriteDVs(
     std::span<const DeletionVectorMergeGroup> groups, std::string_view output_path,
-    const std::shared_ptr<FileIO>& io) {
+    const std::shared_ptr<FileIO>& io, OptionalExecutor executor) {
   if (groups.empty()) {
     return std::vector<std::shared_ptr<DataFile>>{};
   }
 
+  // Read all duplicate DVs in parallel, retaining their original order.
+  std::vector<std::shared_ptr<DataFile>> duplicate_dvs;
+  for (const auto& group : groups) {
+    duplicate_dvs.insert(duplicate_dvs.end(), group.delete_files.begin(),
+                         group.delete_files.end());
+  }
+  std::vector<PositionDeleteIndex> duplicated_dv_positions(duplicate_dvs.size());
+  std::atomic<bool> failed{false};
+  auto tasks = TaskGroup().SetExecutor(executor);
+  for (size_t index = 0; index < duplicate_dvs.size(); ++index) {
+    tasks.Submit([&, index]() -> Status {
+      if (failed.load()) return {};
+      auto positions = ReadDV(duplicate_dvs[index], io);
+      if (!positions) {
+        failed.store(true);
+        return ::iceberg::unexpected(std::move(positions.error()));
+      }
+      duplicated_dv_positions[index] = std::move(positions.value());
+      return {};
+    });
+  }
+  ICEBERG_RETURN_UNEXPECTED(std::move(tasks).Run());
+
   ICEBERG_ASSIGN_OR_RAISE(
       auto writer,
       DVWriter::Make(DVWriterOptions{
@@ -58,11 +83,11 @@ Result<std::vector<std::shared_ptr<DataFile>>> DVUtil::MergeAndWriteDVs(
           .load_previous_deletes = [](std::string_view)
               -> Result<std::optional<PositionDeleteIndex>> { return std::nullopt; }}));
 
+  size_t index = 0;
   for (const auto& group : groups) {
     PositionDeleteIndex merged;
-    for (const auto& delete_file : group.delete_files) {
-      ICEBERG_ASSIGN_OR_RAISE(auto index, DVUtil::ReadDV(delete_file, io));
-      merged.Merge(index);
+    for (size_t pos = 0; pos < group.delete_files.size(); ++pos) {
+      merged.Merge(duplicated_dv_positions[index++]);
     }
     ICEBERG_RETURN_UNEXPECTED(
         writer->Delete(group.referenced_data_file, merged, group.spec, group.partition));

src/iceberg/deletes/dv_util_internal.h

@@ -36,6 +36,7 @@
 #include "iceberg/result.h"
 #include "iceberg/row/partition_values.h"
 #include "iceberg/type_fwd.h"
+#include "iceberg/util/executor.h"
 
 namespace iceberg {
 
@@ -50,7 +51,7 @@ class ICEBERG_EXPORT DVUtil {
  public:
   static Result<std::vector<std::shared_ptr<DataFile>>> MergeAndWriteDVs(
       std::span<const DeletionVectorMergeGroup> groups, std::string_view output_path,
-      const std::shared_ptr<FileIO>& io);
+      const std::shared_ptr<FileIO>& io, OptionalExecutor executor = std::nullopt);
 
   static Result<PositionDeleteIndex> ReadDV(const std::shared_ptr<DataFile>& delete_file,
                                             const std::shared_ptr<FileIO>& io);

src/iceberg/inheritable_metadata.cc

@@ -65,8 +65,7 @@ Status BaseInheritableMetadata::Apply(ManifestEntry& entry) {
   if (entry.data_file) {
     entry.data_file->partition_spec_id = spec_id_;
     entry.data_file->data_sequence_number = entry.sequence_number;
-    // TODO(gangwu): Also propagate file_sequence_number and manifest_location
-    // when C++ adds consumers that need Java-style inherited file metadata.
+    entry.data_file->manifest_location = manifest_location_;
   } else {
     return InvalidManifest("Manifest entry has no data file");
   }

src/iceberg/manifest/manifest_entry.cc

@@ -20,13 +20,42 @@
 #include "iceberg/manifest/manifest_entry.h"
 
 #include <memory>
+#include <tuple>
 #include <vector>
 
 #include "iceberg/schema_field.h"
 #include "iceberg/type.h"
 
 namespace iceberg {
 
+bool DataFile::operator==(const DataFile& other) const {
+  return std::tie(content, file_path, file_format, partition, record_count,
+                  file_size_in_bytes, column_sizes, value_counts, null_value_counts,
+                  nan_value_counts, lower_bounds, upper_bounds, key_metadata,
+                  split_offsets, equality_ids, sort_order_id, first_row_id,
+                  referenced_data_file, content_offset, content_size_in_bytes,
+                  partition_spec_id, data_sequence_number) ==
+         std::tie(other.content, other.file_path, other.file_format, other.partition,
+                  other.record_count, other.file_size_in_bytes, other.column_sizes,
+                  other.value_counts, other.null_value_counts, other.nan_value_counts,
+                  other.lower_bounds, other.upper_bounds, other.key_metadata,
+                  other.split_offsets, other.equality_ids, other.sort_order_id,
+                  other.first_row_id, other.referenced_data_file, other.content_offset,
+                  other.content_size_in_bytes, other.partition_spec_id,
+                  other.data_sequence_number);
+}
+
+std::shared_ptr<DataFile> DataFile::CopyWithoutStats() const {
+  auto copy = std::make_shared<DataFile>(*this);
+  copy->column_sizes.clear();
+  copy->value_counts.clear();
+  copy->null_value_counts.clear();
+  copy->nan_value_counts.clear();
+  copy->lower_bounds.clear();
+  copy->upper_bounds.clear();
+  return copy;
+}
+
 bool ManifestEntry::operator==(const ManifestEntry& other) const {
   return status == other.status && snapshot_id == other.snapshot_id &&
          sequence_number == other.sequence_number &&

src/iceberg/manifest/manifest_entry.h

@@ -187,6 +187,12 @@ struct ICEBERG_EXPORT DataFile {
   /// the DataFile struct.
   std::optional<int64_t> data_sequence_number;
 
+  /// \brief Location of the manifest from which this file was read (not serialized).
+  std::optional<std::string> manifest_location;
+
+  /// \brief Copy this file without column-level metrics, retaining file metadata.
+  std::shared_ptr<DataFile> CopyWithoutStats() const;
+
   static constexpr int32_t kContentFieldId = 134;
   inline static const SchemaField kContent = SchemaField::MakeOptional(
       kContentFieldId, "content", int32(),
@@ -301,7 +307,8 @@ struct ICEBERG_EXPORT DataFile {
       SchemaField::MakeOptional(kContentSizeFieldId, "content_size_in_bytes", int64(),
                                 "The length of referenced content stored in the file");
 
-  bool operator==(const DataFile& other) const = default;
+  /// Value equality excludes transient manifest provenance.
+  bool operator==(const DataFile& other) const;
 
   /// \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);