wgtmac commented on code in PR #873:
URL: https://github.com/apache/iceberg-cpp/pull/873#discussion_r3828350097
##########
src/iceberg/manifest/manifest_group.cc:
##########
@@ -252,6 +475,12 @@ Result<std::vector<std::shared_ptr<FileScanTask>>>
ManifestGroup::PlanFiles() {
return file_tasks;
}
+Result<std::unique_ptr<Iterator<std::shared_ptr<FileScanTask>>>>
+ManifestGroup::PlanFilesIterator() {
+ auto group = std::make_unique<ManifestGroup>(std::move(*this));
Review Comment:
Can we make this `PlanFilesIterator() &&`? It moves from `*this`, but
callers can invoke it on an lvalue and accidentally reuse a consumed
`ManifestGroup`.
##########
src/iceberg/util/iterator.h:
##########
@@ -0,0 +1,67 @@
+/*
+ * 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
+
+/// \file iceberg/util/iterator.h
+/// \brief Pull-based iterator interface for fallible, lazily produced values.
+
+#include <memory>
+#include <optional>
+#include <utility>
+#include <vector>
+
+#include "iceberg/result.h"
+#include "iceberg/util/macros.h"
+
+namespace iceberg {
+
+/// \brief A pull-based iterator whose reads may fail.
+///
+/// Iterator implementations own any resources needed to produce values.
Destroying an
+/// iterator releases those resources, including when iteration stops before
reaching the
+/// end. Iterators are not thread-safe unless an implementation explicitly
says otherwise.
+///
+/// \tparam T Value returned by the iterator.
+template <typename T>
+class Iterator {
+ public:
+ virtual ~Iterator() = default;
+
+ Iterator() = default;
+ Iterator(const Iterator&) = delete;
+ Iterator& operator=(const Iterator&) = delete;
+
+ /// \brief Return the next value, or std::nullopt when the iterator is
exhausted.
+ virtual Result<std::optional<T>> Next() = 0;
+
+ /// \brief Consume the remaining values into a vector.
+ Result<std::vector<T>> ToVector() {
+ std::vector<T> values;
+ while (true) {
+ ICEBERG_ASSIGN_OR_RAISE(auto value, Next());
+ if (!value.has_value()) {
+ return values;
+ }
+ values.push_back(std::move(value).value());
Review Comment:
This fails to compile for a copy-only `T`, because `push_back` selects the
deleted move constructor. Please either constrain `Iterator<T>` to movable
types or use a copy fallback, and test the chosen contract.
##########
src/iceberg/table_scan.cc:
##########
@@ -64,6 +65,93 @@ const std::vector<std::string> kScanColumnsWithStats = [] {
return cols;
}();
+template <typename T>
+class EmptyIterator final : public Iterator<T> {
+ public:
+ Result<std::optional<T>> Next() override { return std::nullopt; }
+};
+
+Result<ScanReport> MakeScanReport(const DataTableScan& scan, const Snapshot&
snapshot,
+ ScanMetricsResult scan_metrics) {
+ ICEBERG_ASSIGN_OR_RAISE(auto schema_ptr, scan.schema());
+
+ ICEBERG_ASSIGN_OR_RAISE(
+ auto projected_id_set,
+ GetProjectedIdsVisitor::GetProjectedIds(*schema_ptr,
/*include_struct_ids=*/true));
+ std::vector<int32_t> projected_field_ids(projected_id_set.begin(),
+ projected_id_set.end());
+ std::ranges::sort(projected_field_ids);
+
+ std::vector<std::string> projected_field_names;
+ projected_field_names.reserve(projected_field_ids.size());
+ for (int32_t field_id : projected_field_ids) {
+ ICEBERG_ASSIGN_OR_RAISE(auto field_name,
schema_ptr->FindColumnNameById(field_id));
+ ICEBERG_CHECK(field_name.has_value(), "Projected field {} not found in
schema",
+ field_id);
+ projected_field_names.emplace_back(*field_name);
+ }
+
+ ICEBERG_ASSIGN_OR_RAISE(auto sanitized_filter,
+ SanitizeExpression::Sanitize(*schema_ptr,
scan.filter(),
+
scan.context().case_sensitive));
+
+ return ScanReport{
+ .table_name = scan.context().table_name,
+ .snapshot_id = snapshot.snapshot_id,
+ .filter = std::move(sanitized_filter),
+ .schema_id = schema_ptr->schema_id(),
+ .projected_field_ids = std::move(projected_field_ids),
+ .projected_field_names = std::move(projected_field_names),
+ .scan_metrics = std::move(scan_metrics),
+ .metadata = scan.context().options,
+ };
+}
+
+class ReportingFileTaskIterator final : public
Iterator<std::shared_ptr<FileScanTask>> {
+ public:
+ ReportingFileTaskIterator(
+ std::unique_ptr<Iterator<std::shared_ptr<FileScanTask>>> iterator,
+ std::shared_ptr<ScanMetrics> scan_metrics,
+ std::chrono::nanoseconds planning_duration,
+ std::shared_ptr<MetricsReporter> reporter, ScanReport report)
+ : iterator_(std::move(iterator)),
+ scan_metrics_(std::move(scan_metrics)),
+ planning_duration_(std::move(planning_duration)),
+ reporter_(std::move(reporter)),
+ report_(std::move(report)) {}
+
+ ~ReportingFileTaskIterator() override { Finalize(); }
+
+ Result<std::optional<std::shared_ptr<FileScanTask>>> Next() override {
+ auto start = std::chrono::steady_clock::now();
+ auto result = iterator_->Next();
+ planning_duration_ += std::chrono::duration_cast<std::chrono::nanoseconds>(
+ std::chrono::steady_clock::now() - start);
+ if (result.has_value() && !result.value().has_value()) {
Review Comment:
Please handle `Next()` errors here. Right now reporting waits until
destruction and records the failed plan as normal partial consumption. Finalize
immediately or skip the report, and cover both error and early-destruction
paths.
##########
src/iceberg/manifest/manifest_reader.h:
##########
@@ -134,6 +145,19 @@ class ICEBERG_EXPORT ManifestReader {
const std::vector<std::string>& columns);
};
+/// \brief Optional mix-in for ManifestReader implementations that support
lazy entry
+/// iteration.
+class ICEBERG_EXPORT SupportsManifestEntryIteration {
+ public:
+ virtual ~SupportsManifestEntryIteration() = default;
+
+ /// \brief Lazily read manifest entries.
+ virtual Result<std::unique_ptr<Iterator<ManifestEntry>>> EntriesIterator() =
0;
Review Comment:
Please document that these methods must return self-contained iterators. The
planning path destroys the `ManifestReader` immediately after obtaining the
iterator, so a third-party implementation that keeps a pointer to `this` will
dangle.
##########
src/iceberg/manifest/manifest_group.h:
##########
@@ -136,6 +137,15 @@ class ICEBERG_EXPORT ManifestGroup : public ErrorCollector
{
/// \brief Plan scan tasks for all matching data files.
Result<std::vector<std::shared_ptr<FileScanTask>>> PlanFiles();
+ /// \brief Lazily plan scan tasks for matching data files.
+ ///
+ /// The returned iterator owns the planning state and may outlive this
ManifestGroup.
+ /// It reads one manifest batch at a time instead of materializing all
manifest entries
+ /// and scan tasks. Creating the iterator consumes this group's
configuration. Streaming
+ /// planning is pull-based and does not eagerly submit manifests to the
executor set by
Review Comment:
`PlanFilesIterator()` still receives `plan_executor`, but this path ignores
it and reads data manifests serially. Was that slowdown measured? Please
preserve executor-backed planning or add benchmark data and document the
trade-off.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
