add some tests

Author: HuaHuaY

src/iceberg/constants.h

@@ -0,0 +1,28 @@
+/*
+ * 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 <string_view>
+
+namespace iceberg {
+
+constexpr std::string_view kParquetFieldIdKey = "PARQUET:field_id";
+
+}  // namespace iceberg

src/iceberg/parquet/parquet_reader.cc

@@ -125,9 +125,9 @@ class ParquetReader::Impl {
     arrow_reader_properties.set_arrow_extensions_enabled(true);
 
     // Open the Parquet file reader
-    ICEBERG_ASSIGN_OR_RAISE(auto input_stream, OpenInputStream(options));
+    ICEBERG_ASSIGN_OR_RAISE(input_stream_, OpenInputStream(options));
     auto file_reader =
-        ::parquet::ParquetFileReader::Open(std::move(input_stream), reader_properties);
+        ::parquet::ParquetFileReader::Open(input_stream_, reader_properties);
     ICEBERG_ARROW_RETURN_NOT_OK(::parquet::arrow::FileReader::Make(
         pool_, std::move(file_reader), arrow_reader_properties, &reader_));
 
@@ -169,6 +169,7 @@ class ParquetReader::Impl {
     }
 
     reader_.reset();
+    ICEBERG_ARROW_RETURN_NOT_OK(input_stream_->Close());
     return {};
   }
 
@@ -240,6 +241,8 @@ class ParquetReader::Impl {
   std::unique_ptr<::parquet::arrow::FileReader> reader_;
   // The context to keep track of the reading progress.
   std::unique_ptr<ReadContext> context_;
+  // The input stream to read Parquet file.
+  std::shared_ptr<::arrow::io::RandomAccessFile> input_stream_;
 };
 
 ParquetReader::~ParquetReader() = default;

src/iceberg/parquet/parquet_register.cc

@@ -21,8 +21,6 @@
 
 namespace iceberg::parquet {
 
-void RegisterWriter() {}
-
 void RegisterAll() {
   RegisterReader();
   RegisterWriter();

src/iceberg/parquet/parquet_schema_util.cc

@@ -26,6 +26,7 @@
 #include <parquet/arrow/schema.h>
 #include <parquet/schema.h>
 
+#include "iceberg/constants.h"
 #include "iceberg/metadata_columns.h"
 #include "iceberg/parquet/parquet_schema_util_internal.h"
 #include "iceberg/result.h"
@@ -38,8 +39,6 @@ namespace iceberg::parquet {
 
 namespace {
 
-constexpr std::string_view kParquetFieldIdKey = "PARQUET:field_id";
-
 std::optional<int32_t> FieldIdFromMetadata(
     const std::shared_ptr<const ::arrow::KeyValueMetadata>& metadata) {
   if (!metadata) {

src/iceberg/parquet/parquet_writer.cc

@@ -69,9 +69,9 @@ class ParquetWriter::Impl {
     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_ASSIGN_OR_RAISE(output_stream_, OpenOutputStream(options));
+    auto file_writer = ::parquet::ParquetFileWriter::Open(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_));
 
@@ -89,10 +89,13 @@ class ParquetWriter::Impl {
 
   // Close the writer and release resources
   Status Close() {
-    if (writer_ != nullptr) {
-      ICEBERG_ARROW_RETURN_NOT_OK(writer_->Close());
-      writer_.reset();
+    if (writer_ == nullptr) {
+      return {};  // Already closed
     }
+
+    ICEBERG_ARROW_RETURN_NOT_OK(writer_->Close());
+    writer_.reset();
+    ICEBERG_ARROW_RETURN_NOT_OK(output_stream_->Close());
     return {};
   }
 
@@ -105,6 +108,8 @@ class ParquetWriter::Impl {
   std::shared_ptr<::arrow::Schema> arrow_schema_;
   // Parquet file writer to write ArrowArray.
   std::unique_ptr<::parquet::arrow::FileWriter> writer_;
+  // The output stream to write Parquet file.
+  std::shared_ptr<::arrow::io::OutputStream> output_stream_;
 };
 
 ParquetWriter::~ParquetWriter() = default;
@@ -139,7 +144,7 @@ std::vector<int64_t> ParquetWriter::split_offsets() {
   return {};
 }
 
-void ParquetWriter::Register() {
+void RegisterWriter() {
   static WriterFactoryRegistry parquet_writer_register(
       FileFormatType::kParquet, []() -> Result<std::unique_ptr<Writer>> {
         return std::make_unique<ParquetWriter>();

src/iceberg/parquet/parquet_writer.h

@@ -43,8 +43,6 @@ class ICEBERG_BUNDLE_EXPORT ParquetWriter : public Writer {
 
   std::vector<int64_t> split_offsets() final;
 
-  static void Register();
-
  private:
   class Impl;
   std::unique_ptr<Impl> impl_;

src/iceberg/schema_internal.h

@@ -24,6 +24,7 @@
 
 #include <nanoarrow/nanoarrow.h>
 
+#include "iceberg/constants.h"
 #include "iceberg/iceberg_export.h"
 #include "iceberg/result.h"
 #include "iceberg/type_fwd.h"
@@ -33,7 +34,7 @@ namespace iceberg {
 // Apache Arrow C++ uses "PARQUET:field_id" to store field IDs for Parquet.
 // Here we follow a similar convention for Iceberg but we might also add
 // "PARQUET:field_id" in the future once we implement a Parquet writer.
-constexpr std::string_view kFieldIdKey = "ICEBERG:field_id";
+constexpr std::string_view kFieldIdKey = kParquetFieldIdKey;
 
 /// \brief Convert an Iceberg schema to an Arrow schema.
 ///

src/iceberg/type.cc

@@ -319,4 +319,16 @@ std::shared_ptr<FixedType> fixed(int32_t length) {
   return std::make_shared<FixedType>(length);
 }
 
+std::shared_ptr<MapType> map(SchemaField key, SchemaField value) {
+  return std::make_shared<MapType>(key, value);
+}
+
+std::shared_ptr<ListType> list(SchemaField element) {
+  return std::make_shared<ListType>(std::move(element));
+}
+
+std::shared_ptr<StructType> struct_(std::vector<SchemaField> fields) {
+  return std::make_shared<StructType>(std::move(fields));
+}
+
 }  // namespace iceberg

src/iceberg/type.h

@@ -487,6 +487,22 @@ ICEBERG_EXPORT std::shared_ptr<DecimalType> decimal(int32_t precision, int32_t s
 /// \return A shared pointer to the FixedType instance.
 ICEBERG_EXPORT std::shared_ptr<FixedType> fixed(int32_t length);
 
+/// \brief Create a StructType with the given fields.
+/// \param fields The fields of the struct.
+/// \return A shared pointer to the StructType instance.
+ICEBERG_EXPORT std::shared_ptr<StructType> struct_(std::vector<SchemaField> fields);
+
+/// \brief Create a ListType with the given element field.
+/// \param element The element field of the list.
+/// \return A shared pointer to the ListType instance.
+ICEBERG_EXPORT std::shared_ptr<ListType> list(SchemaField element);
+
+/// \brief Create a MapType with the given key and value fields.
+/// \param key The key field of the map.
+/// \param value The value field of the map.
+/// \return A shared pointer to the MapType instance.
+ICEBERG_EXPORT std::shared_ptr<MapType> map(SchemaField key, SchemaField value);
+
 /// @}
 
 }  // namespace iceberg