You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iceberg.apache.org by fo...@apache.org on 2025/06/04 18:25:39 UTC
(iceberg-cpp) branch main updated: feat: implement Apache Avro file reader (#113)
This is an automated email from the ASF dual-hosted git repository.
fokko pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/iceberg-cpp.git
The following commit(s) were added to refs/heads/main by this push:
new ed49d1e feat: implement Apache Avro file reader (#113)
ed49d1e is described below
commit ed49d1e8f0dc31d99d4fff437a6c1583a0dfced5
Author: Gang Wu <us...@gmail.com>
AuthorDate: Thu Jun 5 02:25:34 2025 +0800
feat: implement Apache Avro file reader (#113)
- Refactor a little bit of Reader and ReaderFactory interfaces.
- Implement the skeleton of Avro reader to read data into ArrowArray.
- `AppendDatumToBuilder` is not implemented yet.
---
src/iceberg/CMakeLists.txt | 2 +
src/iceberg/arrow/arrow_fs_file_io.h | 3 +
.../avro/{demo_avro.h => avro_data_util.cc} | 29 +--
.../{demo_avro.h => avro_data_util_internal.h} | 26 +--
src/iceberg/avro/avro_reader.cc | 222 +++++++++++++++++++++
src/iceberg/avro/{demo_avro.h => avro_reader.h} | 30 +--
src/iceberg/avro/demo_avro.cc | 11 +-
src/iceberg/avro/demo_avro.h | 2 +
src/iceberg/expected.h | 19 +-
src/iceberg/file_reader.cc | 9 +-
src/iceberg/file_reader.h | 36 +++-
src/iceberg/result.h | 2 +
src/iceberg/schema_internal.h | 3 +-
test/avro_test.cc | 2 +-
14 files changed, 319 insertions(+), 77 deletions(-)
diff --git a/src/iceberg/CMakeLists.txt b/src/iceberg/CMakeLists.txt
index 09b419e..4c660d7 100644
--- a/src/iceberg/CMakeLists.txt
+++ b/src/iceberg/CMakeLists.txt
@@ -95,6 +95,8 @@ if(ICEBERG_BUILD_BUNDLE)
arrow/demo_arrow.cc
arrow/arrow_fs_file_io.cc
avro/demo_avro.cc
+ avro/avro_data_util.cc
+ avro/avro_reader.cc
avro/avro_schema_util.cc
avro/avro_stream_internal.cc)
diff --git a/src/iceberg/arrow/arrow_fs_file_io.h b/src/iceberg/arrow/arrow_fs_file_io.h
index b55af5d..e187c89 100644
--- a/src/iceberg/arrow/arrow_fs_file_io.h
+++ b/src/iceberg/arrow/arrow_fs_file_io.h
@@ -46,6 +46,9 @@ class ICEBERG_BUNDLE_EXPORT ArrowFileSystemFileIO : public FileIO {
/// \brief Delete a file at the given location.
Status DeleteFile(const std::string& file_location) override;
+ /// \brief Get the Arrow file system.
+ const std::shared_ptr<::arrow::fs::FileSystem>& fs() const { return arrow_fs_; }
+
private:
std::shared_ptr<::arrow::fs::FileSystem> arrow_fs_;
};
diff --git a/src/iceberg/avro/demo_avro.h b/src/iceberg/avro/avro_data_util.cc
similarity index 62%
copy from src/iceberg/avro/demo_avro.h
copy to src/iceberg/avro/avro_data_util.cc
index ca7dbdf..48ac7e6 100644
--- a/src/iceberg/avro/demo_avro.h
+++ b/src/iceberg/avro/avro_data_util.cc
@@ -17,29 +17,16 @@
* under the License.
*/
-#pragma once
-
-#include <string>
-
-#include "iceberg/avro.h"
-#include "iceberg/file_reader.h"
-#include "iceberg/iceberg_bundle_export.h"
+#include "iceberg/avro/avro_data_util_internal.h"
namespace iceberg::avro {
-class ICEBERG_BUNDLE_EXPORT DemoAvro : public Avro {
- public:
- DemoAvro() = default;
- ~DemoAvro() override = default;
- std::string print() const override;
-};
-
-class ICEBERG_BUNDLE_EXPORT DemoAvroReader : public Reader {
- public:
- DemoAvroReader() = default;
- ~DemoAvroReader() override = default;
- Result<Data> Next() override;
- DataLayout data_layout() const override;
-};
+Status AppendDatumToBuilder(const ::avro::NodePtr& avro_node,
+ const ::avro::GenericDatum& avro_datum,
+ const SchemaProjection& projection,
+ const Schema& arrow_schema,
+ ::arrow::ArrayBuilder* array_builder) {
+ return NotImplemented("AppendDatumToBuilder is not yet implemented");
+}
} // namespace iceberg::avro
diff --git a/src/iceberg/avro/demo_avro.h b/src/iceberg/avro/avro_data_util_internal.h
similarity index 63%
copy from src/iceberg/avro/demo_avro.h
copy to src/iceberg/avro/avro_data_util_internal.h
index ca7dbdf..4b96483 100644
--- a/src/iceberg/avro/demo_avro.h
+++ b/src/iceberg/avro/avro_data_util_internal.h
@@ -19,27 +19,17 @@
#pragma once
-#include <string>
+#include <arrow/array/builder_base.h>
+#include <avro/GenericDatum.hh>
-#include "iceberg/avro.h"
-#include "iceberg/file_reader.h"
-#include "iceberg/iceberg_bundle_export.h"
+#include "iceberg/schema_util.h"
namespace iceberg::avro {
-class ICEBERG_BUNDLE_EXPORT DemoAvro : public Avro {
- public:
- DemoAvro() = default;
- ~DemoAvro() override = default;
- std::string print() const override;
-};
-
-class ICEBERG_BUNDLE_EXPORT DemoAvroReader : public Reader {
- public:
- DemoAvroReader() = default;
- ~DemoAvroReader() override = default;
- Result<Data> Next() override;
- DataLayout data_layout() const override;
-};
+Status AppendDatumToBuilder(const ::avro::NodePtr& avro_node,
+ const ::avro::GenericDatum& avro_datum,
+ const SchemaProjection& projection,
+ const Schema& arrow_schema,
+ ::arrow::ArrayBuilder* array_builder);
} // namespace iceberg::avro
diff --git a/src/iceberg/avro/avro_reader.cc b/src/iceberg/avro/avro_reader.cc
new file mode 100644
index 0000000..5b66329
--- /dev/null
+++ b/src/iceberg/avro/avro_reader.cc
@@ -0,0 +1,222 @@
+/*
+ * 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_reader.h"
+
+#include <memory>
+
+#include <arrow/array/builder_base.h>
+#include <arrow/c/bridge.h>
+#include <arrow/filesystem/filesystem.h>
+#include <arrow/record_batch.h>
+#include <arrow/result.h>
+#include <arrow/type.h>
+#include <avro/DataFile.hh>
+#include <avro/Generic.hh>
+#include <avro/GenericDatum.hh>
+
+#include "iceberg/arrow/arrow_fs_file_io.h"
+#include "iceberg/avro/avro_data_util_internal.h"
+#include "iceberg/avro/avro_schema_util_internal.h"
+#include "iceberg/avro/avro_stream_internal.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<AvroInputStream>> CreateInputStream(const ReaderOptions& options,
+ int64_t buffer_size) {
+ ::arrow::fs::FileInfo file_info(options.path, ::arrow::fs::FileType::File);
+ if (options.length) {
+ file_info.set_size(options.length.value());
+ }
+
+ auto io = internal::checked_pointer_cast<arrow::ArrowFileSystemFileIO>(options.io);
+ auto result = io->fs()->OpenInputFile(file_info);
+ if (!result.ok()) {
+ return IOError("Failed to open file {} for {}", options.path,
+ result.status().message());
+ }
+
+ return std::make_unique<AvroInputStream>(result.MoveValueUnsafe(), buffer_size);
+}
+
+} // namespace
+
+// A stateful context to keep track of the reading progress.
+struct ReadContext {
+ // The datum to reuse for reading the data.
+ std::unique_ptr<::avro::GenericDatum> datum_;
+ // The arrow schema to build the record batch.
+ std::shared_ptr<::arrow::Schema> arrow_schema_;
+ // The builder to build the record batch.
+ std::shared_ptr<::arrow::ArrayBuilder> builder_;
+};
+
+// TODO(gang.wu): there are a lot to do to make this reader work.
+// 1. prune the reader schema based on the projection
+// 2. read key-value metadata from the avro file
+// 3. collect basic reader metrics
+class AvroBatchReader::Impl {
+ public:
+ Status Open(const ReaderOptions& options) {
+ batch_size_ = options.batch_size;
+ read_schema_ = options.projection;
+
+ // Open the input stream and adapt to the avro interface.
+ // TODO(gangwu): make this configurable
+ constexpr int64_t kDefaultBufferSize = 1024 * 1024;
+ ICEBERG_ASSIGN_OR_RAISE(auto input_stream,
+ CreateInputStream(options, kDefaultBufferSize));
+
+ // Create a base reader without setting reader schema to enable projection.
+ auto base_reader =
+ std::make_unique<::avro::DataFileReaderBase>(std::move(input_stream));
+ const ::avro::ValidSchema& file_schema = base_reader->dataSchema();
+
+ // Validate field ids in the file schema.
+ HasIdVisitor has_id_visitor;
+ ICEBERG_RETURN_UNEXPECTED(has_id_visitor.Visit(file_schema));
+ if (has_id_visitor.HasNoIds()) {
+ // TODO(gangwu): support applying field-ids based on name mapping
+ return NotImplemented("Avro file schema has no field IDs");
+ }
+ if (!has_id_visitor.AllHaveIds()) {
+ return InvalidSchema("Not all fields in the Avro file schema have field IDs");
+ }
+
+ // Project the read schema on top of the file schema.
+ // TODO(gangwu): support pruning source fields
+ ICEBERG_ASSIGN_OR_RAISE(projection_, Project(*options.projection, file_schema.root(),
+ /*prune_source=*/false));
+ base_reader->init(file_schema);
+ reader_ = std::make_unique<::avro::DataFileReader<::avro::GenericDatum>>(
+ std::move(base_reader));
+
+ if (options.split) {
+ reader_->sync(options.split->offset);
+ split_end_ = options.split->offset + options.split->length;
+ }
+ return {};
+ }
+
+ Result<Data> Next() {
+ if (!context_) {
+ ICEBERG_RETURN_UNEXPECTED(InitReadContext());
+ }
+
+ while (context_->builder_->length() < batch_size_) {
+ if (split_end_ && reader_->pastSync(split_end_.value())) {
+ break;
+ }
+ if (!reader_->read(*context_->datum_)) {
+ break;
+ }
+ ICEBERG_RETURN_UNEXPECTED(
+ AppendDatumToBuilder(reader_->readerSchema().root(), *context_->datum_,
+ projection_, *read_schema_, context_->builder_.get()));
+ }
+
+ return ConvertBuilderToArrowArray();
+ }
+
+ Status Close() {
+ if (reader_ != nullptr) {
+ reader_->close();
+ reader_.reset();
+ }
+ context_.reset();
+ return {};
+ }
+
+ private:
+ Status InitReadContext() {
+ context_ = std::make_unique<ReadContext>();
+ context_->datum_ = std::make_unique<::avro::GenericDatum>(reader_->readerSchema());
+
+ ArrowSchema arrow_schema;
+ ICEBERG_RETURN_UNEXPECTED(ToArrowSchema(*read_schema_, &arrow_schema));
+ auto import_result = ::arrow::ImportSchema(&arrow_schema);
+ if (!import_result.ok()) {
+ return InvalidSchema("Failed to import the arrow schema: {}",
+ import_result.status().message());
+ }
+ context_->arrow_schema_ = import_result.MoveValueUnsafe();
+
+ auto arrow_struct_type =
+ std::make_shared<::arrow::StructType>(context_->arrow_schema_->fields());
+ auto builder_result = ::arrow::MakeBuilder(arrow_struct_type);
+ if (!builder_result.ok()) {
+ return InvalidSchema("Failed to make the arrow builder: {}",
+ builder_result.status().message());
+ }
+ context_->builder_ = builder_result.MoveValueUnsafe();
+
+ return {};
+ }
+
+ Result<Data> ConvertBuilderToArrowArray() {
+ if (context_->builder_->length() == 0) {
+ return {};
+ }
+
+ auto builder_result = context_->builder_->Finish();
+ if (!builder_result.ok()) {
+ return InvalidArrowData("Failed to finish the arrow array builder: {}",
+ builder_result.status().message());
+ }
+
+ auto array = builder_result.MoveValueUnsafe();
+ ArrowArray arrow_array;
+ auto export_result = ::arrow::ExportArray(*array, &arrow_array);
+ if (!export_result.ok()) {
+ return InvalidArrowData("Failed to export the arrow array: {}",
+ export_result.message());
+ }
+ return arrow_array;
+ }
+
+ private:
+ // Max number of rows in the record batch to read.
+ int64_t batch_size_{};
+ // The end of the split to read and used to terminate the reading.
+ std::optional<int64_t> split_end_;
+ // The schema to read.
+ std::shared_ptr<Schema> read_schema_;
+ // The projection result to apply to the read schema.
+ SchemaProjection projection_;
+ // The avro reader to read the data into a datum.
+ std::unique_ptr<::avro::DataFileReader<::avro::GenericDatum>> reader_;
+ // The context to keep track of the reading progress.
+ std::unique_ptr<ReadContext> context_;
+};
+
+Result<Reader::Data> AvroBatchReader::Next() { return impl_->Next(); }
+
+Status AvroBatchReader::Open(const ReaderOptions& options) {
+ impl_ = std::make_unique<Impl>();
+ return impl_->Open(options);
+}
+
+Status AvroBatchReader::Close() { return impl_->Close(); }
+
+} // namespace iceberg::avro
diff --git a/src/iceberg/avro/demo_avro.h b/src/iceberg/avro/avro_reader.h
similarity index 69%
copy from src/iceberg/avro/demo_avro.h
copy to src/iceberg/avro/avro_reader.h
index ca7dbdf..dc4cf45 100644
--- a/src/iceberg/avro/demo_avro.h
+++ b/src/iceberg/avro/avro_reader.h
@@ -19,27 +19,29 @@
#pragma once
-#include <string>
-
-#include "iceberg/avro.h"
#include "iceberg/file_reader.h"
#include "iceberg/iceberg_bundle_export.h"
namespace iceberg::avro {
-class ICEBERG_BUNDLE_EXPORT DemoAvro : public Avro {
+/// \brief A reader that reads ArrowArray from Avro files.
+class ICEBERG_BUNDLE_EXPORT AvroBatchReader : public Reader {
public:
- DemoAvro() = default;
- ~DemoAvro() override = default;
- std::string print() const override;
-};
+ AvroBatchReader() = default;
-class ICEBERG_BUNDLE_EXPORT DemoAvroReader : public Reader {
- public:
- DemoAvroReader() = default;
- ~DemoAvroReader() override = default;
- Result<Data> Next() override;
- DataLayout data_layout() const override;
+ ~AvroBatchReader() override = default;
+
+ Status Open(const ReaderOptions& options) final;
+
+ Status Close() final;
+
+ Result<Data> Next() final;
+
+ DataLayout data_layout() const final { return DataLayout::kArrowArray; }
+
+ private:
+ class Impl;
+ std::unique_ptr<Impl> impl_;
};
} // namespace iceberg::avro
diff --git a/src/iceberg/avro/demo_avro.cc b/src/iceberg/avro/demo_avro.cc
index 1b7f7a2..679b496 100644
--- a/src/iceberg/avro/demo_avro.cc
+++ b/src/iceberg/avro/demo_avro.cc
@@ -55,9 +55,12 @@ Reader::DataLayout DemoAvroReader::data_layout() const {
return Reader::DataLayout::kStructLike;
}
-ICEBERG_REGISTER_READER_FACTORY(
- Avro, [](const ReaderOptions& options) -> Result<std::unique_ptr<Reader>> {
- return std::make_unique<DemoAvroReader>();
- });
+Status DemoAvroReader::Open(const ReaderOptions& options) { return {}; }
+
+Status DemoAvroReader::Close() { return {}; }
+
+ICEBERG_REGISTER_READER_FACTORY(Avro, []() -> Result<std::unique_ptr<Reader>> {
+ return std::make_unique<DemoAvroReader>();
+});
} // namespace iceberg::avro
diff --git a/src/iceberg/avro/demo_avro.h b/src/iceberg/avro/demo_avro.h
index ca7dbdf..14cb2fa 100644
--- a/src/iceberg/avro/demo_avro.h
+++ b/src/iceberg/avro/demo_avro.h
@@ -38,6 +38,8 @@ class ICEBERG_BUNDLE_EXPORT DemoAvroReader : public Reader {
public:
DemoAvroReader() = default;
~DemoAvroReader() override = default;
+ Status Open(const ReaderOptions& options) override;
+ Status Close() override;
Result<Data> Next() override;
DataLayout data_layout() const override;
};
diff --git a/src/iceberg/expected.h b/src/iceberg/expected.h
index 9d28d38..723b282 100644
--- a/src/iceberg/expected.h
+++ b/src/iceberg/expected.h
@@ -122,7 +122,7 @@ inline constexpr bool is_error_type_valid_v = is_error_type_valid<T>::value;
} // namespace expected_detail
template <class E>
-class ICEBERG_EXPORT [[nodiscard]] unexpected {
+class [[nodiscard]] unexpected {
public:
static_assert(expected_detail::is_error_type_valid_v<E>);
@@ -192,7 +192,7 @@ class ICEBERG_EXPORT [[nodiscard]] unexpected {
template <class E>
unexpected(E) -> unexpected<E>;
-struct ICEBERG_EXPORT unexpect_t {
+struct unexpect_t {
explicit unexpect_t() = default;
};
inline constexpr unexpect_t unexpect{};
@@ -201,7 +201,7 @@ template <class E>
class bad_expected_access;
template <>
-class ICEBERG_EXPORT bad_expected_access<void> : public std::exception {
+class bad_expected_access<void> : public std::exception {
public:
const char* what() const noexcept override { return "Bad expected access"; }
@@ -214,7 +214,7 @@ class ICEBERG_EXPORT bad_expected_access<void> : public std::exception {
};
template <class E>
-class ICEBERG_EXPORT bad_expected_access : public bad_expected_access<void> {
+class bad_expected_access : public bad_expected_access<void> {
public:
explicit bad_expected_access(E e) noexcept(std::is_nothrow_move_constructible_v<E>)
: m_val(std::move(e)) {}
@@ -1006,9 +1006,8 @@ struct default_ctor_base<T, E, false> {
/// tracked by the expected object.
template <class T, class E>
-class ICEBERG_EXPORT [[nodiscard]] expected
- : private expected_detail::move_assign_base<T, E>,
- private expected_detail::default_ctor_base<T, E> {
+class [[nodiscard]] expected : private expected_detail::move_assign_base<T, E>,
+ private expected_detail::default_ctor_base<T, E> {
static_assert(expected_detail::is_value_type_valid_v<T>);
static_assert(expected_detail::is_error_type_valid_v<E>);
@@ -1798,7 +1797,7 @@ constexpr void swap(expected<T, E>& lhs,
}
template <class E>
-class ICEBERG_EXPORT [[nodiscard]] expected<void, E>
+class [[nodiscard]] expected<void, E>
: private expected_detail::move_assign_base<void, E>,
private expected_detail::default_ctor_base<void, E> {
static_assert(expected_detail::is_error_type_valid_v<E>);
@@ -2349,8 +2348,8 @@ class ICEBERG_EXPORT [[nodiscard]] expected<void, E>
// standalone swap for void value type
template <class E, std::enable_if_t<std::is_move_constructible_v<E> &&
std::is_swappable_v<E>>* = nullptr>
-ICEBERG_EXPORT constexpr void swap(
- expected<void, E>& lhs, expected<void, E>& rhs) noexcept(noexcept(lhs.swap(rhs))) {
+constexpr void swap(expected<void, E>& lhs,
+ expected<void, E>& rhs) noexcept(noexcept(lhs.swap(rhs))) {
lhs.swap(rhs);
}
diff --git a/src/iceberg/file_reader.cc b/src/iceberg/file_reader.cc
index bf8035a..a56c30c 100644
--- a/src/iceberg/file_reader.cc
+++ b/src/iceberg/file_reader.cc
@@ -23,13 +23,14 @@
#include "iceberg/expected.h"
#include "iceberg/util/formatter.h"
+#include "iceberg/util/macros.h"
namespace iceberg {
namespace {
ReaderFactory GetNotImplementedFactory(FileFormatType format_type) {
- return [format_type](const ReaderOptions& options) -> Result<std::unique_ptr<Reader>> {
+ return [format_type]() -> Result<std::unique_ptr<Reader>> {
return NotImplemented("Missing reader factory for file format: {}", format_type);
};
}
@@ -51,9 +52,11 @@ ReaderFactoryRegistry::ReaderFactoryRegistry(FileFormatType format_type,
GetFactory(format_type) = std::move(factory);
}
-Result<std::unique_ptr<Reader>> ReaderFactoryRegistry::Create(
+Result<std::unique_ptr<Reader>> ReaderFactoryRegistry::Open(
FileFormatType format_type, const ReaderOptions& options) {
- return GetFactory(format_type)(options);
+ ICEBERG_ASSIGN_OR_RAISE(auto reader, GetFactory(format_type)());
+ ICEBERG_RETURN_UNEXPECTED(reader->Open(options));
+ return reader;
}
StructLikeReader::StructLikeReader(std::unique_ptr<Reader> reader)
diff --git a/src/iceberg/file_reader.h b/src/iceberg/file_reader.h
index c3faab5..ca46cdc 100644
--- a/src/iceberg/file_reader.h
+++ b/src/iceberg/file_reader.h
@@ -38,6 +38,15 @@ namespace iceberg {
class ICEBERG_EXPORT Reader {
public:
virtual ~Reader() = default;
+ Reader() = default;
+ Reader(const Reader&) = delete;
+ Reader& operator=(const Reader&) = delete;
+
+ /// \brief Open the reader.
+ virtual Status Open(const struct ReaderOptions& options) = 0;
+
+ /// \brief Close the reader.
+ virtual Status Close() = 0;
/// \brief Read next data from file.
///
@@ -61,11 +70,19 @@ class ICEBERG_EXPORT StructLikeReader : public Reader {
public:
explicit StructLikeReader(std::unique_ptr<Reader> reader);
+ ~StructLikeReader() override = default;
+
/// \brief Always read data into `StructLike` or monostate if no more data.
Result<Data> Next() final;
DataLayout data_layout() const final { return DataLayout::kStructLike; }
+ Status Open(const struct ReaderOptions& options) final {
+ return reader_->Open(options);
+ }
+
+ Status Close() final { return reader_->Close(); }
+
private:
std::unique_ptr<Reader> reader_;
};
@@ -78,11 +95,19 @@ class ICEBERG_EXPORT BatchReader : public Reader {
public:
explicit BatchReader(std::unique_ptr<Reader> reader);
+ ~BatchReader() override = default;
+
/// \brief Always read data into `ArrowArray` or monostate if no more data.
Result<Data> Next() final;
DataLayout data_layout() const final { return DataLayout::kArrowArray; }
+ Status Open(const struct ReaderOptions& options) final {
+ return reader_->Open(options);
+ }
+
+ Status Close() final { return reader_->Close(); }
+
private:
std::unique_ptr<Reader> reader_;
};
@@ -115,11 +140,12 @@ struct ICEBERG_EXPORT ReaderOptions {
/// \brief The filter to apply to the data. Reader implementations may ignore this if
/// the file format does not support filtering.
std::shared_ptr<class Expression> filter;
+ /// \brief Format-specific or implementation-specific properties.
+ std::unordered_map<std::string, std::string> properties;
};
/// \brief Factory function to create a reader of a specific file format.
-using ReaderFactory =
- std::function<Result<std::unique_ptr<Reader>>(const ReaderOptions&)>;
+using ReaderFactory = std::function<Result<std::unique_ptr<Reader>>()>;
/// \brief Registry of reader factories for different file formats.
struct ICEBERG_EXPORT ReaderFactoryRegistry {
@@ -129,9 +155,9 @@ struct ICEBERG_EXPORT ReaderFactoryRegistry {
/// \brief Get the factory function for a specific file format.
static ReaderFactory& GetFactory(FileFormatType format_type);
- /// \brief Create a reader for a specific file format.
- static Result<std::unique_ptr<Reader>> Create(FileFormatType format_type,
- const ReaderOptions& options);
+ /// \brief Open a reader for a specific file format.
+ static Result<std::unique_ptr<Reader>> Open(FileFormatType format_type,
+ const ReaderOptions& options);
};
/// \brief Macro to register a reader factory for a specific file format.
diff --git a/src/iceberg/result.h b/src/iceberg/result.h
index ecfcec4..e9f1dfb 100644
--- a/src/iceberg/result.h
+++ b/src/iceberg/result.h
@@ -33,6 +33,7 @@ enum class ErrorKind {
kCommitStateUnknown,
kDecompressError,
kInvalidArgument,
+ kInvalidArrowData,
kInvalidExpression,
kInvalidSchema,
kIOError,
@@ -77,6 +78,7 @@ DEFINE_ERROR_FUNCTION(AlreadyExists)
DEFINE_ERROR_FUNCTION(CommitStateUnknown)
DEFINE_ERROR_FUNCTION(DecompressError)
DEFINE_ERROR_FUNCTION(InvalidArgument)
+DEFINE_ERROR_FUNCTION(InvalidArrowData)
DEFINE_ERROR_FUNCTION(InvalidExpression)
DEFINE_ERROR_FUNCTION(InvalidSchema)
DEFINE_ERROR_FUNCTION(IOError)
diff --git a/src/iceberg/schema_internal.h b/src/iceberg/schema_internal.h
index 26be1c3..1a31e68 100644
--- a/src/iceberg/schema_internal.h
+++ b/src/iceberg/schema_internal.h
@@ -24,6 +24,7 @@
#include <nanoarrow/nanoarrow.h>
+#include "iceberg/iceberg_export.h"
#include "iceberg/result.h"
#include "iceberg/type_fwd.h"
@@ -39,7 +40,7 @@ constexpr std::string_view kFieldIdKey = "ICEBERG:field_id";
/// \param[in] schema The Iceberg schema to convert.
/// \param[out] out The Arrow schema to convert to.
/// \return An error if the conversion fails.
-Status ToArrowSchema(const Schema& schema, ArrowSchema* out);
+ICEBERG_EXPORT Status ToArrowSchema(const Schema& schema, ArrowSchema* out);
/// \brief Convert an Arrow schema to an Iceberg schema.
///
diff --git a/test/avro_test.cc b/test/avro_test.cc
index 5ffcbc0..01ac821 100644
--- a/test/avro_test.cc
+++ b/test/avro_test.cc
@@ -45,7 +45,7 @@ TEST(AVROTest, TestDemoAvro) {
}
TEST(AVROTest, TestDemoAvroReader) {
- auto result = ReaderFactoryRegistry::Create(FileFormatType::kAvro, {});
+ auto result = ReaderFactoryRegistry::Open(FileFormatType::kAvro, {});
ASSERT_THAT(result, IsOk());
auto reader = std::move(result.value());