3434#include " iceberg/nanoarrow_status_internal.h"
3535#include " iceberg/schema.h"
3636#include " iceberg/schema_internal.h"
37+ #include " iceberg/util/iterator.h"
3738#include " iceberg/util/macros.h"
3839
3940namespace iceberg ::internal {
4041
41- // / \brief Arrow stream backed by a fixed set of metadata-table rows .
42+ // / \brief Arrow stream backed by a metadata-table row iterator .
4243template <typename Row>
4344class MetadataTableRowsStream {
4445 public:
4546 using AppendRow = std::function<Status(ArrowRowBuilder&, const Row&)>;
4647
47- static Result<std::unique_ptr<MetadataTableRowsStream>> Make (const Schema& schema,
48- std::vector<Row> rows,
49- AppendRow append_row) {
48+ static Result<std::unique_ptr<MetadataTableRowsStream>> Make (
49+ const Schema& schema, std::unique_ptr<Iterator<Row>> rows, AppendRow append_row) {
5050 ArrowSchema arrow_schema{};
5151 ICEBERG_RETURN_UNEXPECTED (ToArrowSchema (schema, &arrow_schema));
5252 return std::unique_ptr<MetadataTableRowsStream>(new MetadataTableRowsStream (
@@ -59,7 +59,7 @@ class MetadataTableRowsStream {
5959 }
6060
6161 Status Close () {
62- rows_.clear ();
62+ rows_.reset ();
6363 append_row_ = nullptr ;
6464 if (arrow_schema_.release != nullptr ) {
6565 ArrowSchemaRelease (&arrow_schema_);
@@ -70,13 +70,16 @@ class MetadataTableRowsStream {
7070 Result<std::optional<ArrowArray>> Next () {
7171 ICEBERG_PRECHECK (arrow_schema_.release != nullptr ,
7272 " Cannot read from a closed metadata table stream" );
73- if (next_row_ == rows_.size ()) {
74- return std::nullopt ;
75- }
76-
7773 ICEBERG_ASSIGN_OR_RAISE (auto builder, ArrowRowBuilder::Make (&arrow_schema_));
78- while (next_row_ < rows_.size () && builder.num_rows () < MetadataTable::kBatchSize ) {
79- ICEBERG_RETURN_UNEXPECTED (append_row_ (builder, rows_[next_row_++]));
74+ while (builder.num_rows () < MetadataTable::kBatchSize ) {
75+ ICEBERG_ASSIGN_OR_RAISE (auto row, rows_->Next ());
76+ if (!row.has_value ()) {
77+ break ;
78+ }
79+ ICEBERG_RETURN_UNEXPECTED (append_row_ (builder, *row));
80+ }
81+ if (builder.num_rows () == 0 ) {
82+ return std::nullopt ;
8083 }
8184
8285 ICEBERG_ASSIGN_OR_RAISE (auto array, std::move (builder).Finish ());
@@ -93,21 +96,20 @@ class MetadataTableRowsStream {
9396 }
9497
9598 private:
96- MetadataTableRowsStream (std::vector< Row> rows, AppendRow append_row,
99+ MetadataTableRowsStream (std::unique_ptr<Iterator< Row> > rows, AppendRow append_row,
97100 ArrowSchema arrow_schema)
98101 : rows_(std::move(rows)),
99102 append_row_ (std::move(append_row)),
100103 arrow_schema_(std::move(arrow_schema)) {}
101104
102- std::vector< Row> rows_;
105+ std::unique_ptr<Iterator< Row> > rows_;
103106 AppendRow append_row_;
104107 ArrowSchema arrow_schema_{};
105- size_t next_row_ = 0 ;
106108};
107109
108110template <typename Row, typename AppendRow>
109111Result<ArrowArrayStream> MakeMetadataTableStream (const Schema& schema,
110- std::vector< Row> rows,
112+ std::unique_ptr<Iterator< Row> > rows,
111113 AppendRow append_row) {
112114 ICEBERG_ASSIGN_OR_RAISE (
113115 auto stream,
@@ -117,4 +119,31 @@ Result<ArrowArrayStream> MakeMetadataTableStream(const Schema& schema,
117119 return MakeArrowArrayStream (std::move (stream));
118120}
119121
122+ template <typename Row>
123+ class MetadataTableVectorIterator : public Iterator <Row> {
124+ public:
125+ explicit MetadataTableVectorIterator (std::vector<Row> rows) : rows_(std::move(rows)) {}
126+
127+ protected:
128+ Result<std::optional<Row>> NextImpl () override {
129+ if (next_row_ == rows_.size ()) {
130+ return std::nullopt ;
131+ }
132+ return std::move (rows_[next_row_++]);
133+ }
134+
135+ private:
136+ std::vector<Row> rows_;
137+ size_t next_row_ = 0 ;
138+ };
139+
140+ template <typename Row, typename AppendRow>
141+ Result<ArrowArrayStream> MakeMetadataTableStream (const Schema& schema,
142+ std::vector<Row> rows,
143+ AppendRow append_row) {
144+ std::unique_ptr<Iterator<Row>> iterator =
145+ std::make_unique<MetadataTableVectorIterator<Row>>(std::move (rows));
146+ return MakeMetadataTableStream (schema, std::move (iterator), std::move (append_row));
147+ }
148+
120149} // namespace iceberg::internal
0 commit comments