feat: support manifest&list writer part 1

1 add avro writer and factory(without write func since converter pull/166 not finished yet)
2 add manifest and manifest list writer internal implement definition

Author: xiao.dong

src/iceberg/CMakeLists.txt

@@ -23,6 +23,7 @@ set(ICEBERG_SOURCES
     expression/expression.cc
     expression/literal.cc
     file_reader.cc
+    file_writer.cc
     json_internal.cc
     manifest_entry.cc
     manifest_list.cc
@@ -46,6 +47,8 @@ set(ICEBERG_SOURCES
     type.cc
     manifest_reader.cc
     manifest_reader_internal.cc
+    manifest_writer.cc
+    manifest_writer_internal.cc
     arrow_c_data_guard_internal.cc
     util/murmurhash3_internal.cc
     util/timepoint.cc
@@ -107,6 +110,7 @@ if(ICEBERG_BUILD_BUNDLE)
       arrow/arrow_fs_file_io.cc
       avro/avro_data_util.cc
       avro/avro_reader.cc
+      avro/avro_writer.cc
       avro/avro_schema_util.cc
       avro/avro_register.cc
       avro/avro_stream_internal.cc

src/iceberg/avro/avro_writer.cc

@@ -0,0 +1,134 @@
+/*
+ * 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/avro/avro_writer.h"
+
+#include <memory>
+
+#include <arrow/array/builder_base.h>
+#include <arrow/c/bridge.h>
+#include <arrow/record_batch.h>
+#include <arrow/result.h>
+#include <avro/DataFile.hh>
+#include <avro/GenericDatum.hh>
+#include <avro/NodeImpl.hh>
+
+#include "iceberg/arrow/arrow_fs_file_io_internal.h"
+#include "iceberg/avro/avro_schema_util_internal.h"
+#include "iceberg/avro/avro_stream_internal.h"
+#include "iceberg/schema.h"
+#include "iceberg/schema_internal.h"
+#include "iceberg/util/checked_cast.h"
+#include "iceberg/util/macros.h"
+
+namespace iceberg::avro {
+
+namespace {
+
+Result<std::unique_ptr<AvroOutputStream>> CreateOutputStream(const WriterOptions& options,
+                                                             int64_t buffer_size) {
+  auto io = internal::checked_pointer_cast<arrow::ArrowFileSystemFileIO>(options.io);
+  auto result = io->fs()->OpenOutputStream(options.path);
+  if (!result.ok()) {
+    return IOError("Failed to open file {} for {}", options.path,
+                   result.status().message());
+  }
+  return std::make_unique<AvroOutputStream>(result.MoveValueUnsafe(), buffer_size);
+}
+
+}  // namespace
+
+// A stateful context to keep track of the writing progress.
+struct WriteContext {};
+
+class AvroWriter::Impl {
+ public:
+  Status Open(const WriterOptions& options) {
+    write_arrow_schema_ = options.schema;
+    ICEBERG_ASSIGN_OR_RAISE(write_schema_, FromArrowSchema(options.schema, std::nullopt));
+
+    auto root = std::make_shared<::avro::NodeRecord>();
+    ToAvroNodeVisitor visitor;
+    for (const auto& field : write_schema_->fields()) {
+      ::avro::NodePtr node;
+      ICEBERG_RETURN_UNEXPECTED(visitor.Visit(field, &node));
+      root->addLeaf(node);
+    }
+    avro_schema_ = std::make_shared<::avro::ValidSchema>(root);
+
+    // Open the output stream and adapt to the avro interface.
+    constexpr int64_t kDefaultBufferSize = 1024 * 1024;
+    ICEBERG_ASSIGN_OR_RAISE(auto output_stream,
+                            CreateOutputStream(options, kDefaultBufferSize));
+
+    writer_ = std::make_unique<::avro::DataFileWriter<::avro::GenericDatum>>(
+        std::move(output_stream), *avro_schema_);
+    return {};
+  }
+
+  Status Write(ArrowArray /*data*/) {
+    if (!context_) {
+      ICEBERG_RETURN_UNEXPECTED(InitWriteContext());
+    }
+    // TODO(xiao.dong) convert data and write to avro
+    return {};
+  }
+
+  Status Close() {
+    if (writer_ != nullptr) {
+      writer_->close();
+      writer_.reset();
+    }
+    context_.reset();
+    return {};
+  }
+
+ private:
+  Status InitWriteContext() { return {}; }
+
+ private:
+  ArrowSchema write_arrow_schema_;
+  // The schema to write.
+  std::shared_ptr<::iceberg::Schema> write_schema_;
+  // The avro schema to write.
+  std::shared_ptr<::avro::ValidSchema> avro_schema_;
+  // The avro writer to write the data into a datum.
+  std::unique_ptr<::avro::DataFileWriter<::avro::GenericDatum>> writer_;
+  // The context to keep track of the writing progress.
+  std::unique_ptr<WriteContext> context_;
+};
+
+AvroWriter::~AvroWriter() = default;
+
+Status AvroWriter::Write(ArrowArray data) { return impl_->Write(data); }
+
+Status AvroWriter::Open(const WriterOptions& options) {
+  impl_ = std::make_unique<Impl>();
+  return impl_->Open(options);
+}
+
+Status AvroWriter::Close() { return impl_->Close(); }
+
+void AvroWriter::Register() {
+  static WriterFactoryRegistry avro_writer_register(
+      FileFormatType::kAvro,
+      []() -> Result<std::unique_ptr<Writer>> { return std::make_unique<AvroWriter>(); });
+}
+
+}  // namespace iceberg::avro

src/iceberg/avro/avro_writer.h

@@ -0,0 +1,48 @@
+/*
+ * 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
+
+#include "iceberg/file_writer.h"
+#include "iceberg/iceberg_bundle_export.h"
+
+namespace iceberg::avro {
+
+/// \brief A writer ArrowArray to Avro files.
+class ICEBERG_BUNDLE_EXPORT AvroWriter : public Writer {
+ public:
+  AvroWriter() = default;
+
+  ~AvroWriter() override;
+
+  Status Open(const WriterOptions& options) final;
+
+  Status Close() final;
+
+  Status Write(ArrowArray data) final;
+
+  /// \brief Register this Avro writer implementation.
+  static void Register();
+
+ private:
+  class Impl;
+  std::unique_ptr<Impl> impl_;
+};
+
+}  // namespace iceberg::avro

src/iceberg/file_writer.cc

@@ -0,0 +1,62 @@
+/*
+ * 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/file_writer.h"
+
+#include <unordered_map>
+
+#include "iceberg/result.h"
+#include "iceberg/util/formatter.h"
+#include "iceberg/util/macros.h"
+
+namespace iceberg {
+
+namespace {
+
+WriterFactory GetNotImplementedFactory(FileFormatType format_type) {
+  return [format_type]() -> Result<std::unique_ptr<Writer>> {
+    return NotImplemented("Missing writer factory for file format: {}", format_type);
+  };
+}
+
+}  // namespace
+
+WriterFactory& WriterFactoryRegistry::GetFactory(FileFormatType format_type) {
+  static std::unordered_map<FileFormatType, WriterFactory> factories = {
+      {FileFormatType::kAvro, GetNotImplementedFactory(FileFormatType::kAvro)},
+      {FileFormatType::kParquet, GetNotImplementedFactory(FileFormatType::kParquet)},
+      {FileFormatType::kOrc, GetNotImplementedFactory(FileFormatType::kOrc)},
+      {FileFormatType::kPuffin, GetNotImplementedFactory(FileFormatType::kPuffin)},
+  };
+  return factories.at(format_type);
+}
+
+WriterFactoryRegistry::WriterFactoryRegistry(FileFormatType format_type,
+                                             WriterFactory factory) {
+  GetFactory(format_type) = std::move(factory);
+}
+
+Result<std::unique_ptr<Writer>> WriterFactoryRegistry::Open(
+    FileFormatType format_type, const WriterOptions& options) {
+  ICEBERG_ASSIGN_OR_RAISE(auto writer, GetFactory(format_type)());
+  ICEBERG_RETURN_UNEXPECTED(writer->Open(options));
+  return writer;
+}
+
+}  // namespace iceberg

src/iceberg/manifest_writer.cc

@@ -0,0 +1,56 @@
+/*
+ * 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/manifest_writer_internal.h"
+#include "iceberg/schema.h"
+#include "iceberg/util/macros.h"
+
+namespace iceberg {
+
+Result<std::unique_ptr<ManifestWriter>> ManifestWriter::MakeWriter(
+    std::string_view manifest_location, std::shared_ptr<FileIO> file_io,
+    std::shared_ptr<Schema> partition_schema) {
+  auto manifest_entry_schema = ManifestEntry::TypeFromPartitionType(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, WriterFactoryRegistry::Open(FileFormatType::kAvro,
+                                               {.path = std::string(manifest_location),
+                                                .io = std::move(file_io)}));
+  return std::make_unique<ManifestWriterImpl>(std::move(writer), std::move(schema));
+}
+
+Result<std::unique_ptr<ManifestListWriter>> ManifestListWriter::MakeWriter(
+    std::string_view manifest_list_location, std::shared_ptr<FileIO> file_io) {
+  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, WriterFactoryRegistry::Open(
+                                           FileFormatType::kAvro,
+                                           {.path = std::string(manifest_list_location),
+                                            .io = std::move(file_io)}));
+  return std::make_unique<ManifestListWriterImpl>(std::move(writer), std::move(schema));
+}
+
+}  // namespace iceberg

src/iceberg/manifest_writer_internal.cc

@@ -0,0 +1,41 @@
+/*
+ * 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 "manifest_writer_internal.h"
+
+#include "iceberg/manifest_entry.h"
+#include "iceberg/manifest_list.h"
+#include "iceberg/schema.h"
+#include "iceberg/type.h"
+
+namespace iceberg {
+
+Status ManifestWriterImpl::WriteManifestEntries(
+    const std::vector<ManifestEntry>& /*entries*/) const {
+  // TODO(xiao.dong) convert entries to arrow data
+  return {};
+}
+
+Status ManifestListWriterImpl::WriteManifestFiles(
+    const std::vector<ManifestFile>& /*files*/) const {
+  // TODO(xiao.dong) convert manifest files to arrow data
+  return {};
+}
+
+}  // namespace iceberg