basic support parquet writer

Author: HuaHuaY

.gitignore

@@ -23,3 +23,6 @@ cmake-build-release/
 
 # intellij files
 .idea
+
+# vscode files
+.vscode

src/iceberg/CMakeLists.txt

@@ -116,7 +116,8 @@ if(ICEBERG_BUILD_BUNDLE)
       parquet/parquet_data_util.cc
       parquet/parquet_reader.cc
       parquet/parquet_register.cc
-      parquet/parquet_schema_util.cc)
+      parquet/parquet_schema_util.cc
+      parquet/parquet_writer.cc)
 
   # Libraries to link with exported libiceberg_bundle.{so,a}.
   set(ICEBERG_BUNDLE_STATIC_BUILD_INTERFACE_LIBS)

src/iceberg/parquet/parquet_writer.cc

@@ -0,0 +1,149 @@
+/*
+ * 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/parquet/parquet_writer.h"
+
+#include <memory>
+
+#include <arrow/c/bridge.h>
+#include <arrow/record_batch.h>
+#include <arrow/util/key_value_metadata.h>
+#include <parquet/arrow/schema.h>
+#include <parquet/arrow/writer.h>
+#include <parquet/file_writer.h>
+#include <parquet/properties.h>
+
+#include "iceberg/arrow/arrow_error_transform_internal.h"
+#include "iceberg/arrow/arrow_fs_file_io_internal.h"
+#include "iceberg/schema_internal.h"
+#include "iceberg/util/checked_cast.h"
+#include "iceberg/util/macros.h"
+
+namespace iceberg::parquet {
+
+namespace {
+
+Result<std::shared_ptr<::arrow::io::OutputStream>> OpenOutputStream(
+    const WriterOptions& options) {
+  auto io = internal::checked_pointer_cast<arrow::ArrowFileSystemFileIO>(options.io);
+  ICEBERG_ARROW_ASSIGN_OR_RETURN(auto output, io->fs()->OpenOutputStream(options.path));
+  return output;
+}
+
+}  // namespace
+
+class ParquetWriter::Impl {
+ public:
+  Status Open(const WriterOptions& options) {
+    auto writer_properties =
+        ::parquet::WriterProperties::Builder().memory_pool(pool_)->build();
+    auto arrow_writer_properties = ::parquet::default_arrow_writer_properties();
+
+    ArrowSchema c_schema;
+    ICEBERG_RETURN_UNEXPECTED(ToArrowSchema(*options.schema, &c_schema));
+    ICEBERG_ARROW_ASSIGN_OR_RETURN(arrow_schema_, ::arrow::ImportSchema(&c_schema));
+
+    std::shared_ptr<::parquet::SchemaDescriptor> schema_descriptor;
+    ICEBERG_ARROW_RETURN_NOT_OK(
+        ::parquet::arrow::ToParquetSchema(arrow_schema_.get(), *writer_properties,
+                                          *arrow_writer_properties, &schema_descriptor));
+    auto schema_node = std::static_pointer_cast<::parquet::schema::GroupNode>(
+        schema_descriptor->schema_root());
+
+    std::shared_ptr<::arrow::KeyValueMetadata> metadata =
+        ::arrow::key_value_metadata(options.properties);
+
+    ICEBERG_ASSIGN_OR_RAISE(auto output_stream, OpenOutputStream(options));
+    auto file_writer = ::parquet::ParquetFileWriter::Open(
+        std::move(output_stream), schema_node, writer_properties, metadata);
+    ICEBERG_ARROW_RETURN_NOT_OK(::parquet::arrow::FileWriter::Make(
+        pool_, std::move(file_writer), arrow_schema_, arrow_writer_properties, &writer_));
+
+    return {};
+  }
+
+  Status Write(ArrowArray array) {
+    ICEBERG_ARROW_ASSIGN_OR_RETURN(auto batch,
+                                   ::arrow::ImportRecordBatch(&array, arrow_schema_));
+
+    ICEBERG_ARROW_RETURN_NOT_OK(writer_->WriteRecordBatch(*batch));
+
+    return {};
+  }
+
+  // Close the writer and release resources
+  Status Close() {
+    if (writer_ != nullptr) {
+      ICEBERG_ARROW_RETURN_NOT_OK(writer_->Close());
+      writer_.reset();
+    }
+    return {};
+  }
+
+  bool Closed() const { return writer_ == nullptr; }
+
+ private:
+  // TODO(gangwu): make memory pool configurable
+  ::arrow::MemoryPool* pool_ = ::arrow::default_memory_pool();
+  // Schema to write from the Parquet file.
+  std::shared_ptr<::arrow::Schema> arrow_schema_;
+  // Parquet file writer to write ArrowArray.
+  std::unique_ptr<::parquet::arrow::FileWriter> writer_;
+};
+
+ParquetWriter::~ParquetWriter() = default;
+
+Status ParquetWriter::Open(const WriterOptions& options) {
+  impl_ = std::make_unique<Impl>();
+  return impl_->Open(options);
+}
+
+Status ParquetWriter::Write(ArrowArray array) { return impl_->Write(array); }
+
+Status ParquetWriter::Close() { return impl_->Close(); }
+
+std::optional<Metrics> ParquetWriter::metrics() {
+  if (!impl_->Closed()) {
+    return std::nullopt;
+  }
+  return {};
+}
+
+std::optional<int64_t> ParquetWriter::length() {
+  if (!impl_->Closed()) {
+    return std::nullopt;
+  }
+  return {};
+}
+
+std::vector<int64_t> ParquetWriter::split_offsets() {
+  if (!impl_->Closed()) {
+    return {};
+  }
+  return {};
+}
+
+void ParquetWriter::Register() {
+  static WriterFactoryRegistry parquet_writer_register(
+      FileFormatType::kParquet, []() -> Result<std::unique_ptr<Writer>> {
+        return std::make_unique<ParquetWriter>();
+      });
+}
+
+}  // namespace iceberg::parquet

src/iceberg/parquet/parquet_writer.h

@@ -0,0 +1,53 @@
+/*
+ * 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::parquet {
+
+/// \brief A writer that writes ArrowArray to Parquet files.
+class ICEBERG_BUNDLE_EXPORT ParquetWriter : public Writer {
+ public:
+  ParquetWriter() = default;
+
+  ~ParquetWriter() override;
+
+  Status Open(const WriterOptions& options) final;
+
+  Status Close() final;
+
+  Status Write(ArrowArray array) final;
+
+  std::optional<Metrics> metrics() final;
+
+  std::optional<int64_t> length() final;
+
+  std::vector<int64_t> split_offsets() final;
+
+  static void Register();
+
+ private:
+  class Impl;
+  std::unique_ptr<Impl> impl_;
+};
+
+}  // namespace iceberg::parquet