feat: add manifest&manifest list writer (#176)

1 add v1v2v3 writer definition
2 add v1v2v3 metadata wrapper

---------

Co-authored-by: xiao.dong <[email protected]>

Author: dongxiao1198

src/iceberg/CMakeLists.txt

@@ -47,6 +47,7 @@ set(ICEBERG_SOURCES
     type.cc
     manifest_reader.cc
     manifest_reader_internal.cc
+    manifest_writer.cc
     arrow_c_data_guard_internal.cc
     util/murmurhash3_internal.cc
     util/timepoint.cc

src/iceberg/manifest_adapter.h

@@ -0,0 +1,66 @@
+/*
+ * 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/metadata_adapter.h
+/// Base class of adapter for v1v2v3v4 metadata.
+
+#include "iceberg/arrow_c_data.h"
+#include "iceberg/result.h"
+#include "iceberg/type_fwd.h"
+
+namespace iceberg {
+
+// \brief Base class to append manifest metadata to Arrow array.
+class ICEBERG_EXPORT ManifestAdapter {
+ public:
+  ManifestAdapter() = default;
+  virtual ~ManifestAdapter() = default;
+
+  virtual Status StartAppending() = 0;
+  virtual Result<ArrowArray> FinishAppending() = 0;
+  int64_t size() const { return size_; }
+
+ protected:
+  ArrowArray array_;
+  int64_t size_ = 0;
+};
+
+// \brief Implemented by different versions with different schemas to
+// append a list of `ManifestEntry`s to an `ArrowArray`.
+class ICEBERG_EXPORT ManifestEntryAdapter : public ManifestAdapter {
+ public:
+  ManifestEntryAdapter() = default;
+  ~ManifestEntryAdapter() override = default;
+
+  virtual Status Append(const ManifestEntry& entry) = 0;
+};
+
+// \brief Implemented by different versions with different schemas to
+// append a list of `ManifestFile`s to an `ArrowArray`.
+class ICEBERG_EXPORT ManifestFileAdapter : public ManifestAdapter {
+ public:
+  ManifestFileAdapter() = default;
+  ~ManifestFileAdapter() override = default;
+
+  virtual Status Append(const ManifestFile& file) = 0;
+};
+
+}  // namespace iceberg

src/iceberg/manifest_writer.cc

@@ -0,0 +1,182 @@
+/*
+ * 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_writer.h"
+
+#include "iceberg/manifest_entry.h"
+#include "iceberg/manifest_list.h"
+#include "iceberg/schema.h"
+#include "iceberg/util/macros.h"
+#include "iceberg/v1_metadata.h"
+#include "iceberg/v2_metadata.h"
+#include "iceberg/v3_metadata.h"
+
+namespace iceberg {
+
+Status ManifestWriter::Add(const ManifestEntry& entry) {
+  if (adapter_->size() >= kBatchSize) {
+    ICEBERG_ASSIGN_OR_RAISE(auto array, adapter_->FinishAppending());
+    ICEBERG_RETURN_UNEXPECTED(writer_->Write(array));
+    ICEBERG_RETURN_UNEXPECTED(adapter_->StartAppending());
+  }
+  return adapter_->Append(entry);
+}
+
+Status ManifestWriter::AddAll(const std::vector<ManifestEntry>& entries) {
+  for (const auto& entry : entries) {
+    ICEBERG_RETURN_UNEXPECTED(Add(entry));
+  }
+  return {};
+}
+
+Status ManifestWriter::Close() {
+  if (adapter_->size() > 0) {
+    ICEBERG_ASSIGN_OR_RAISE(auto array, adapter_->FinishAppending());
+    ICEBERG_RETURN_UNEXPECTED(writer_->Write(array));
+  }
+  return {};
+}
+
+Result<std::unique_ptr<Writer>> OpenFileWriter(std::string_view location,
+                                               std::shared_ptr<Schema> schema,
+                                               std::shared_ptr<FileIO> file_io) {
+  ICEBERG_ASSIGN_OR_RAISE(
+      auto writer,
+      WriterFactoryRegistry::Open(FileFormatType::kAvro, {.path = std::string(location),
+                                                          .schema = std::move(schema),
+                                                          .io = std::move(file_io)}));
+  return writer;
+}
+
+Result<std::unique_ptr<ManifestWriter>> ManifestWriter::MakeV1Writer(
+    std::optional<int64_t> snapshot_id, std::string_view manifest_location,
+    std::shared_ptr<FileIO> file_io, std::shared_ptr<Schema> partition_schema) {
+  // TODO(xiao.dong) parse v1 schema
+  auto manifest_entry_schema =
+      ManifestEntry::TypeFromPartitionType(std::move(partition_schema));
+  auto fields_span = manifest_entry_schema->fields();
+  std::vector<SchemaField> fields(fields_span.begin(), fields_span.end());
+  auto schema = std::make_shared<Schema>(fields);
+  ICEBERG_ASSIGN_OR_RAISE(auto writer,
+                          OpenFileWriter(manifest_location, schema, std::move(file_io)));
+  auto adapter = std::make_unique<ManifestEntryAdapterV1>(snapshot_id, std::move(schema));
+  return std::make_unique<ManifestWriter>(std::move(writer), std::move(adapter));
+}
+
+Result<std::unique_ptr<ManifestWriter>> ManifestWriter::MakeV2Writer(
+    std::optional<int64_t> snapshot_id, std::string_view manifest_location,
+    std::shared_ptr<FileIO> file_io, std::shared_ptr<Schema> partition_schema) {
+  // TODO(xiao.dong) parse v2 schema
+  auto manifest_entry_schema =
+      ManifestEntry::TypeFromPartitionType(std::move(partition_schema));
+  auto fields_span = manifest_entry_schema->fields();
+  std::vector<SchemaField> fields(fields_span.begin(), fields_span.end());
+  auto schema = std::make_shared<Schema>(fields);
+  ICEBERG_ASSIGN_OR_RAISE(auto writer,
+                          OpenFileWriter(manifest_location, schema, std::move(file_io)));
+  auto adapter = std::make_unique<ManifestEntryAdapterV2>(snapshot_id, std::move(schema));
+  return std::make_unique<ManifestWriter>(std::move(writer), std::move(adapter));
+}
+
+Result<std::unique_ptr<ManifestWriter>> ManifestWriter::MakeV3Writer(
+    std::optional<int64_t> snapshot_id, std::optional<int64_t> first_row_id,
+    std::string_view manifest_location, std::shared_ptr<FileIO> file_io,
+    std::shared_ptr<Schema> partition_schema) {
+  // TODO(xiao.dong) parse v3 schema
+  auto manifest_entry_schema =
+      ManifestEntry::TypeFromPartitionType(std::move(partition_schema));
+  auto fields_span = manifest_entry_schema->fields();
+  std::vector<SchemaField> fields(fields_span.begin(), fields_span.end());
+  auto schema = std::make_shared<Schema>(fields);
+  ICEBERG_ASSIGN_OR_RAISE(auto writer,
+                          OpenFileWriter(manifest_location, schema, std::move(file_io)));
+  auto adapter = std::make_unique<ManifestEntryAdapterV3>(snapshot_id, first_row_id,
+                                                          std::move(schema));
+  return std::make_unique<ManifestWriter>(std::move(writer), std::move(adapter));
+}
+
+Status ManifestListWriter::Add(const ManifestFile& file) {
+  if (adapter_->size() >= kBatchSize) {
+    ICEBERG_ASSIGN_OR_RAISE(auto array, adapter_->FinishAppending());
+    ICEBERG_RETURN_UNEXPECTED(writer_->Write(array));
+    ICEBERG_RETURN_UNEXPECTED(adapter_->StartAppending());
+  }
+  return adapter_->Append(file);
+}
+
+Status ManifestListWriter::AddAll(const std::vector<ManifestFile>& files) {
+  for (const auto& file : files) {
+    ICEBERG_RETURN_UNEXPECTED(Add(file));
+  }
+  return {};
+}
+
+Status ManifestListWriter::Close() {
+  if (adapter_->size() > 0) {
+    ICEBERG_ASSIGN_OR_RAISE(auto array, adapter_->FinishAppending());
+    ICEBERG_RETURN_UNEXPECTED(writer_->Write(array));
+  }
+  return {};
+}
+
+Result<std::unique_ptr<ManifestListWriter>> ManifestListWriter::MakeV1Writer(
+    int64_t snapshot_id, std::optional<int64_t> parent_snapshot_id,
+    std::string_view manifest_list_location, std::shared_ptr<FileIO> file_io) {
+  // TODO(xiao.dong) parse v1 schema
+  std::vector<SchemaField> fields(ManifestFile::Type().fields().begin(),
+                                  ManifestFile::Type().fields().end());
+  auto schema = std::make_shared<Schema>(fields);
+  ICEBERG_ASSIGN_OR_RAISE(
+      auto writer, OpenFileWriter(manifest_list_location, schema, std::move(file_io)));
+  auto adapter = std::make_unique<ManifestFileAdapterV1>(snapshot_id, parent_snapshot_id,
+                                                         std::move(schema));
+  return std::make_unique<ManifestListWriter>(std::move(writer), std::move(adapter));
+}
+
+Result<std::unique_ptr<ManifestListWriter>> ManifestListWriter::MakeV2Writer(
+    int64_t snapshot_id, std::optional<int64_t> parent_snapshot_id,
+    int64_t sequence_number, std::string_view manifest_list_location,
+    std::shared_ptr<FileIO> file_io) {
+  // TODO(xiao.dong) parse v2 schema
+  std::vector<SchemaField> fields(ManifestFile::Type().fields().begin(),
+                                  ManifestFile::Type().fields().end());
+  auto schema = std::make_shared<Schema>(fields);
+  ICEBERG_ASSIGN_OR_RAISE(
+      auto writer, OpenFileWriter(manifest_list_location, schema, std::move(file_io)));
+  auto adapter = std::make_unique<ManifestFileAdapterV2>(
+      snapshot_id, parent_snapshot_id, sequence_number, std::move(schema));
+  return std::make_unique<ManifestListWriter>(std::move(writer), std::move(adapter));
+}
+
+Result<std::unique_ptr<ManifestListWriter>> ManifestListWriter::MakeV3Writer(
+    int64_t snapshot_id, std::optional<int64_t> parent_snapshot_id,
+    int64_t sequence_number, std::optional<int64_t> first_row_id,
+    std::string_view manifest_list_location, std::shared_ptr<FileIO> file_io) {
+  // TODO(xiao.dong) parse v3 schema
+  std::vector<SchemaField> fields(ManifestFile::Type().fields().begin(),
+                                  ManifestFile::Type().fields().end());
+  auto schema = std::make_shared<Schema>(fields);
+  ICEBERG_ASSIGN_OR_RAISE(
+      auto writer, OpenFileWriter(manifest_list_location, schema, std::move(file_io)));
+  auto adapter = std::make_unique<ManifestFileAdapterV3>(
+      snapshot_id, parent_snapshot_id, sequence_number, first_row_id, std::move(schema));
+  return std::make_unique<ManifestListWriter>(std::move(writer), std::move(adapter));
+}
+
+}  // namespace iceberg