fix: review comments

Author: zhjwpku

src/iceberg/CMakeLists.txt

@@ -49,6 +49,7 @@ set(ICEBERG_SOURCES
     manifest/manifest_group.cc
     manifest/manifest_list.cc
     manifest/manifest_reader.cc
+    manifest/manifest_util_internal.cc
     manifest/manifest_writer.cc
     manifest/rolling_manifest_writer.cc
     manifest/v1_metadata.cc

src/iceberg/constants.h

@@ -32,6 +32,7 @@ namespace iceberg {
 
 constexpr std::string_view kParquetFieldIdKey = "PARQUET:field_id";
 constexpr int64_t kInvalidSnapshotId = -1;
+constexpr int64_t kInvalidSequenceNumber = -1;
 /// \brief Stand-in for the current sequence number that will be assigned when the commit
 /// is successful. This is replaced when writing a manifest list by the ManifestFile
 /// adapter.

src/iceberg/manifest/manifest_util_internal.cc

@@ -0,0 +1,75 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+#include "iceberg/manifest/manifest_util_internal.h"
+
+#include <memory>
+#include <optional>
+
+#include "iceberg/inheritable_metadata.h"
+#include "iceberg/manifest/manifest_entry.h"
+#include "iceberg/manifest/manifest_reader.h"
+#include "iceberg/manifest/manifest_writer.h"
+#include "iceberg/result.h"
+#include "iceberg/schema.h"
+#include "iceberg/snapshot.h"
+#include "iceberg/util/macros.h"
+
+namespace iceberg {
+
+Result<ManifestFile> CopyAppendManifest(
+    const ManifestFile& manifest, std::shared_ptr<FileIO> file_io,
+    std::shared_ptr<Schema> schema, std::shared_ptr<PartitionSpec> spec,
+    int64_t snapshot_id, const std::string& output_path, int8_t format_version,
+    SnapshotSummaryBuilder* summary_builder) {
+  ICEBERG_ASSIGN_OR_RAISE(auto reader,
+                          ManifestReader::Make(manifest, file_io, schema, spec));
+  ICEBERG_ASSIGN_OR_RAISE(auto entries, reader->Entries());
+
+  // use metadata that will add the current snapshot's ID for the rewrite
+  ICEBERG_ASSIGN_OR_RAISE(auto inheritable_metadata,
+                          InheritableMetadataFactory::ForCopy(snapshot_id));
+
+  // do not produce row IDs for the copy
+  ICEBERG_ASSIGN_OR_RAISE(
+      auto writer,
+      ManifestWriter::MakeWriter(format_version, snapshot_id, output_path, file_io, spec,
+                                 schema, ManifestContent::kData));
+
+  // Write all entries as added entries with the new snapshot ID
+  for (auto& entry : entries) {
+    ICEBERG_PRECHECK(entry.status == ManifestStatus::kAdded,
+                     "Manifest to copy must only contain added entries");
+
+    ICEBERG_RETURN_UNEXPECTED(inheritable_metadata->Apply(entry));
+
+    if (summary_builder != nullptr && entry.data_file != nullptr) {
+      summary_builder->AddedFile(*spec, *entry.data_file);
+    }
+
+    ICEBERG_RETURN_UNEXPECTED(writer->WriteAddedEntry(entry));
+  }
+
+  ICEBERG_RETURN_UNEXPECTED(writer->Close());
+  ICEBERG_ASSIGN_OR_RAISE(auto new_manifest, writer->ToManifestFile());
+
+  return new_manifest;
+}
+
+}  // namespace iceberg

src/iceberg/manifest/manifest_util_internal.h

@@ -0,0 +1,57 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+#pragma once
+
+/// \file iceberg/manifest/manifest_util_internal.h
+/// Internal utility functions for manifest operations.
+
+#include <cstdint>
+#include <memory>
+#include <optional>
+#include <string>
+
+#include "iceberg/iceberg_export.h"
+#include "iceberg/result.h"
+#include "iceberg/type_fwd.h"
+
+namespace iceberg {
+
+/// \brief Copy an append manifest with a new snapshot ID.
+///
+/// This function copies a manifest file that contains only ADDED entries,
+/// rewriting it with a new snapshot ID. This is similar to Java's
+/// ManifestFiles.copyAppendManifest.
+///
+/// \param manifest The manifest file to copy
+/// \param file_io File IO implementation to use
+/// \param schema Table schema
+/// \param spec Partition spec for the manifest
+/// \param snapshot_id The new snapshot ID to assign to entries
+/// \param output_path Path where the new manifest will be written
+/// \param format_version Table format version
+/// \param summary_builder Optional summary builder to update with file metrics
+/// \return The copied manifest file, or an error
+ICEBERG_EXPORT Result<ManifestFile> CopyAppendManifest(
+    const ManifestFile& manifest, std::shared_ptr<FileIO> file_io,
+    std::shared_ptr<Schema> schema, std::shared_ptr<PartitionSpec> spec,
+    int64_t snapshot_id, const std::string& output_path, int8_t format_version,
+    SnapshotSummaryBuilder* summary_builder = nullptr);
+
+}  // namespace iceberg

src/iceberg/meson.build

@@ -67,6 +67,7 @@ iceberg_sources = files(
     'manifest/manifest_group.cc',
     'manifest/manifest_list.cc',
     'manifest/manifest_reader.cc',
+    'manifest/manifest_util_internal.cc',
     'manifest/manifest_writer.cc',
     'manifest/rolling_manifest_writer.cc',
     'manifest/v1_metadata.cc',

src/iceberg/table_metadata.h

@@ -75,7 +75,6 @@ struct ICEBERG_EXPORT TableMetadata {
   static constexpr int8_t kMinFormatVersionRowLineage = 3;
   static constexpr int8_t kMinFormatVersionDefaultValues = 3;
   static constexpr int64_t kInitialSequenceNumber = 0;
-  static constexpr int64_t kInvalidSequenceNumber = -1;
   static constexpr int64_t kInitialRowId = 0;
 
   static inline const std::unordered_map<TypeId, int8_t> kMinFormatVersions = {};

src/iceberg/test/manifest_writer_versions_test.cc

@@ -27,6 +27,7 @@
 
 #include "iceberg/arrow/arrow_file_io.h"
 #include "iceberg/avro/avro_register.h"
+#include "iceberg/constants.h"
 #include "iceberg/file_format.h"
 #include "iceberg/manifest/manifest_entry.h"
 #include "iceberg/manifest/manifest_list.h"
@@ -411,12 +412,11 @@ class ManifestWriterVersionsTest : public ::testing::Test {
 
 TEST_F(ManifestWriterVersionsTest, TestV1Write) {
   auto manifest = WriteManifest(/*format_version=*/1, {data_file_});
-  CheckManifest(manifest, TableMetadata::kInvalidSequenceNumber,
-                TableMetadata::kInvalidSequenceNumber);
+  CheckManifest(manifest, kInvalidSequenceNumber, kInvalidSequenceNumber);
   auto entries = ReadManifest(manifest);
   ASSERT_EQ(entries.size(), 1);
-  CheckEntry(entries[0], TableMetadata::kInvalidSequenceNumber,
-             TableMetadata::kInvalidSequenceNumber, DataFile::Content::kData);
+  CheckEntry(entries[0], kInvalidSequenceNumber, kInvalidSequenceNumber,
+             DataFile::Content::kData);
 }
 
 TEST_F(ManifestWriterVersionsTest, TestV1WriteDelete) {
@@ -449,13 +449,12 @@ TEST_F(ManifestWriterVersionsTest, TestV1WriteWithInheritance) {
 
 TEST_F(ManifestWriterVersionsTest, TestV2Write) {
   auto manifest = WriteManifest(/*format_version=*/2, {data_file_});
-  CheckManifest(manifest, TableMetadata::kInvalidSequenceNumber,
-                TableMetadata::kInvalidSequenceNumber);
+  CheckManifest(manifest, kInvalidSequenceNumber, kInvalidSequenceNumber);
   auto entries = ReadManifest(manifest);
   ASSERT_EQ(entries.size(), 1);
   ASSERT_EQ(manifest.content, ManifestContent::kData);
-  CheckEntry(entries[0], TableMetadata::kInvalidSequenceNumber,
-             TableMetadata::kInvalidSequenceNumber, DataFile::Content::kData);
+  CheckEntry(entries[0], kInvalidSequenceNumber, kInvalidSequenceNumber,
+             DataFile::Content::kData);
 }
 
 TEST_F(ManifestWriterVersionsTest, TestV2WriteWithInheritance) {
@@ -470,8 +469,7 @@ TEST_F(ManifestWriterVersionsTest, TestV2WriteWithInheritance) {
 
 TEST_F(ManifestWriterVersionsTest, TestV2PlusWriteDeleteV2) {
   auto manifest = WriteDeleteManifest(/*format_version=*/2, delete_file_);
-  CheckManifest(manifest, TableMetadata::kInvalidSequenceNumber,
-                TableMetadata::kInvalidSequenceNumber);
+  CheckManifest(manifest, kInvalidSequenceNumber, kInvalidSequenceNumber);
   auto entries = ReadManifest(manifest);
   ASSERT_EQ(entries.size(), 1);
   ASSERT_EQ(manifest.content, ManifestContent::kDeletes);
@@ -507,7 +505,7 @@ TEST_F(ManifestWriterVersionsTest, TestV2ManifestRewriteWithInheritance) {
 
   // rewrite the manifest file using a v2 manifest
   auto rewritten_manifest = RewriteManifest(manifests[0], 2);
-  CheckRewrittenManifest(rewritten_manifest, TableMetadata::kInvalidSequenceNumber,
+  CheckRewrittenManifest(rewritten_manifest, kInvalidSequenceNumber,
                          TableMetadata::kInitialSequenceNumber);
 
   // add the v2 manifest to a v2 manifest list, with a sequence number
@@ -525,14 +523,12 @@ TEST_F(ManifestWriterVersionsTest, TestV2ManifestRewriteWithInheritance) {
 
 TEST_F(ManifestWriterVersionsTest, TestV3Write) {
   auto manifest = WriteManifest(/*format_version=*/3, {data_file_});
-  CheckManifest(manifest, TableMetadata::kInvalidSequenceNumber,
-                TableMetadata::kInvalidSequenceNumber);
+  CheckManifest(manifest, kInvalidSequenceNumber, kInvalidSequenceNumber);
   auto entries = ReadManifest(manifest);
   ASSERT_EQ(entries.size(), 1);
   ASSERT_EQ(manifest.content, ManifestContent::kData);
-  CheckEntry(entries[0], TableMetadata::kInvalidSequenceNumber,
-             TableMetadata::kInvalidSequenceNumber, DataFile::Content::kData,
-             ManifestStatus::kAdded, kFirstRowId);
+  CheckEntry(entries[0], kInvalidSequenceNumber, kInvalidSequenceNumber,
+             DataFile::Content::kData, ManifestStatus::kAdded, kFirstRowId);
 }
 
 TEST_F(ManifestWriterVersionsTest, TestV3WriteWithInheritance) {
@@ -598,7 +594,7 @@ TEST_F(ManifestWriterVersionsTest, TestV3ManifestRewriteWithInheritance) {
 
   // rewrite the manifest file using a v3 manifest
   auto rewritten_manifest = RewriteManifest(manifests[0], 3);
-  CheckRewrittenManifest(rewritten_manifest, TableMetadata::kInvalidSequenceNumber,
+  CheckRewrittenManifest(rewritten_manifest, kInvalidSequenceNumber,
                          TableMetadata::kInitialSequenceNumber);
 
   // add the v3 manifest to a v3 manifest list, with a sequence number

src/iceberg/type_fwd.h

@@ -120,6 +120,7 @@ struct SnapshotLogEntry;
 struct SnapshotRef;
 struct StatisticsFile;
 struct TableMetadata;
+class SnapshotSummaryBuilder;
 
 /// \brief Expression.
 class BoundPredicate;
@@ -187,7 +188,6 @@ class TableUpdateContext;
 class Transaction;
 
 /// \brief Update family.
-class AppendFiles;
 class ExpireSnapshots;
 class FastAppend;
 class PendingUpdate;