feat: add manifest list reader (#143)

- Add manifest list reader
- Integrate with avro reader
- Add simple ut

---------

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

Author: dongxiao1198

src/iceberg/CMakeLists.txt

@@ -44,6 +44,9 @@ set(ICEBERG_SOURCES
     transform.cc
     transform_function.cc
     type.cc
+    manifest_reader.cc
+    manifest_reader_internal.cc
+    arrow_c_data_guard_internal.cc
     util/murmurhash3_internal.cc
     util/timepoint.cc
     util/gzip_internal.cc)

src/iceberg/arrow_c_data.h

@@ -73,4 +73,45 @@ struct ArrowArray {
 
 #endif  // ARROW_C_DATA_INTERFACE
 
+#ifndef ARROW_C_STREAM_INTERFACE
+#  define ARROW_C_STREAM_INTERFACE
+
+struct ArrowArrayStream {
+  // Callback to get the stream type
+  // (will be the same for all arrays in the stream).
+  //
+  // Return value: 0 if successful, an `errno`-compatible error code otherwise.
+  //
+  // If successful, the ArrowSchema must be released independently from the stream.
+  int (*get_schema)(struct ArrowArrayStream*, struct ArrowSchema* out);
+
+  // Callback to get the next array
+  // (if no error and the array is released, the stream has ended)
+  //
+  // Return value: 0 if successful, an `errno`-compatible error code otherwise.
+  //
+  // If successful, the ArrowArray must be released independently from the stream.
+  int (*get_next)(struct ArrowArrayStream*, struct ArrowArray* out);
+
+  // Callback to get optional detailed error information.
+  // This must only be called if the last stream operation failed
+  // with a non-0 return code.
+  //
+  // Return value: pointer to a null-terminated character array describing
+  // the last error, or NULL if no description is available.
+  //
+  // The returned pointer is only valid until the next operation on this stream
+  // (including release).
+  const char* (*get_last_error)(struct ArrowArrayStream*);
+
+  // Release callback: release the stream's own resources.
+  // Note that arrays returned by `get_next` must be individually released.
+  void (*release)(struct ArrowArrayStream*);
+
+  // Opaque producer-specific data
+  void* private_data;
+};
+
+#endif  // ARROW_C_STREAM_INTERFACE
+
 }  // extern "C"

src/iceberg/arrow_c_data_guard_internal.cc

@@ -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.
+ */
+
+#include "iceberg/arrow_c_data_guard_internal.h"
+
+namespace iceberg::internal {
+
+ArrowArrayGuard::~ArrowArrayGuard() {
+  if (array_ != nullptr) {
+    ArrowArrayRelease(array_);
+  }
+}
+
+ArrowSchemaGuard::~ArrowSchemaGuard() {
+  if (schema_ != nullptr) {
+    ArrowSchemaRelease(schema_);
+  }
+}
+
+ArrowArrayViewGuard::~ArrowArrayViewGuard() {
+  if (view_ != nullptr) {
+    ArrowArrayViewReset(view_);
+  }
+}
+
+ArrowArrayBufferGuard::~ArrowArrayBufferGuard() {
+  if (buffer_ != nullptr) {
+    ArrowBufferReset(buffer_);
+  }
+}
+
+}  // namespace iceberg::internal

src/iceberg/arrow_c_data_guard_internal.h

@@ -0,0 +1,64 @@
+/*
+ * 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 <nanoarrow/nanoarrow.h>
+
+#include "iceberg/arrow_c_data.h"
+
+namespace iceberg::internal {
+
+class ArrowArrayGuard {
+ public:
+  explicit ArrowArrayGuard(ArrowArray* array) : array_(array) {}
+  ~ArrowArrayGuard();
+
+ private:
+  ArrowArray* array_;
+};
+
+class ArrowSchemaGuard {
+ public:
+  explicit ArrowSchemaGuard(ArrowSchema* schema) : schema_(schema) {}
+  ~ArrowSchemaGuard();
+
+ private:
+  ArrowSchema* schema_;
+};
+
+class ArrowArrayViewGuard {
+ public:
+  explicit ArrowArrayViewGuard(ArrowArrayView* view) : view_(view) {}
+  ~ArrowArrayViewGuard();
+
+ private:
+  ArrowArrayView* view_;
+};
+
+class ArrowArrayBufferGuard {
+ public:
+  explicit ArrowArrayBufferGuard(ArrowBuffer* buffer) : buffer_(buffer) {}
+  ~ArrowArrayBufferGuard();
+
+ private:
+  ArrowBuffer* buffer_;
+};
+
+}  // namespace iceberg::internal

src/iceberg/manifest_list.cc

@@ -19,9 +19,7 @@
 
 #include "iceberg/manifest_list.h"
 
-#include <vector>
-
-#include "iceberg/type.h"
+#include "iceberg/schema.h"
 
 namespace iceberg {
 

src/iceberg/manifest_reader.cc

@@ -0,0 +1,58 @@
+/*
+ * 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_reader.h"
+
+#include "iceberg/manifest_entry.h"
+#include "iceberg/manifest_list.h"
+#include "iceberg/manifest_reader_internal.h"
+#include "iceberg/schema.h"
+#include "iceberg/util/macros.h"
+
+namespace iceberg {
+
+Result<std::unique_ptr<ManifestReader>> ManifestReader::MakeReader(
+    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 reader, ReaderFactoryRegistry::Open(FileFormatType::kAvro,
+                                               {.path = std::string(manifest_location),
+                                                .io = std::move(file_io),
+                                                .projection = schema}));
+  return std::make_unique<ManifestReaderImpl>(std::move(reader), std::move(schema));
+}
+
+Result<std::unique_ptr<ManifestListReader>> ManifestListReader::MakeReader(
+    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 reader, ReaderFactoryRegistry::Open(
+                                           FileFormatType::kAvro,
+                                           {.path = std::string(manifest_list_location),
+                                            .io = std::move(file_io),
+                                            .projection = schema}));
+  return std::make_unique<ManifestListReaderImpl>(std::move(reader), std::move(schema));
+}
+
+}  // namespace iceberg

src/iceberg/manifest_reader.h

@@ -23,7 +23,7 @@
 /// Data reader interface for manifest files.
 
 #include <memory>
-#include <span>
+#include <vector>
 
 #include "iceberg/file_reader.h"
 #include "iceberg/iceberg_export.h"
@@ -35,35 +35,29 @@ namespace iceberg {
 class ICEBERG_EXPORT ManifestReader {
  public:
   virtual ~ManifestReader() = default;
-  virtual Result<std::span<std::unique_ptr<ManifestEntry>>> Entries() const = 0;
-
- private:
-  std::unique_ptr<Reader> reader_;
+  virtual Result<std::vector<ManifestEntry>> Entries() const = 0;
+
+  /// \brief Creates a reader for a manifest file.
+  /// \param manifest_location Path to the manifest file.
+  /// \param file_io File IO implementation to use.
+  /// \return A Result containing the reader or an error.
+  static Result<std::unique_ptr<ManifestReader>> MakeReader(
+      std::string_view manifest_location, std::shared_ptr<FileIO> file_io,
+      std::shared_ptr<Schema> partition_schema);
 };
 
 /// \brief Read manifest files from a manifest list file.
 class ICEBERG_EXPORT ManifestListReader {
  public:
   virtual ~ManifestListReader() = default;
-  virtual Result<std::span<std::unique_ptr<ManifestFile>>> Files() const = 0;
-
- private:
-  std::unique_ptr<Reader> reader_;
+  virtual Result<std::vector<ManifestFile>> Files() const = 0;
+
+  /// \brief Creates a reader for the manifest list.
+  /// \param manifest_list_location Path to the manifest list file.
+  /// \param file_io File IO implementation to use.
+  /// \return A Result containing the reader or an error.
+  static Result<std::unique_ptr<ManifestListReader>> MakeReader(
+      std::string_view manifest_list_location, std::shared_ptr<FileIO> file_io);
 };
 
-/// \brief Creates a reader for the manifest list.
-/// \param file_path Path to the manifest list file.
-/// \return A Result containing the reader or an error.
-Result<std::unique_ptr<ManifestListReader>> CreateManifestListReader(
-    std::string_view file_path) {
-  return NotImplemented("CreateManifestListReader is not implemented yet.");
-}
-
-/// \brief Creates a reader for a manifest file.
-/// \param file_path Path to the manifest file.
-/// \return A Result containing the reader or an error.
-Result<std::unique_ptr<ManifestReader>> CreateManifestReader(std::string_view file_path) {
-  return NotImplemented("CreateManifestReader is not implemented yet.");
-}
-
 }  // namespace iceberg