diff --git a/src/iceberg/test/CMakeLists.txt b/src/iceberg/test/CMakeLists.txt
index 1181a722e..76f5452f8 100644
--- a/src/iceberg/test/CMakeLists.txt
+++ b/src/iceberg/test/CMakeLists.txt
@@ -133,6 +133,7 @@ add_iceberg_test(util_test
                  endian_test.cc
                  file_io_test.cc
                  formatter_test.cc
+                 iterator_test.cc
                  lazy_test.cc
                  location_util_test.cc
                  math_util_internal_test.cc
diff --git a/src/iceberg/test/iterator_test.cc b/src/iceberg/test/iterator_test.cc
new file mode 100644
index 000000000..49b338697
--- /dev/null
+++ b/src/iceberg/test/iterator_test.cc
@@ -0,0 +1,172 @@
+/*
+ * 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/util/iterator.h"
+
+#include <memory>
+#include <optional>
+#include <type_traits>
+#include <vector>
+
+#include <gtest/gtest.h>
+
+#include "iceberg/test/matchers.h"
+
+namespace iceberg {
+namespace {
+
+class CopyOnly {
+ public:
+  explicit CopyOnly(int value) : value_(value) {}
+
+  CopyOnly(const CopyOnly&) = default;
+  CopyOnly& operator=(const CopyOnly&) = default;
+  CopyOnly(CopyOnly&&) = delete;
+  CopyOnly& operator=(CopyOnly&&) = delete;
+
+  int value() const { return value_; }
+
+ private:
+  int value_;
+};
+
+static_assert(std::is_copy_constructible_v<CopyOnly>);
+static_assert(!std::is_move_constructible_v<CopyOnly>);
+
+// Exercises ToVector() with values that can be copied but not moved.
+class CopyOnlyIterator final : public Iterator<CopyOnly> {
+ private:
+  Result<std::optional<CopyOnly>> NextImpl() override {
+    if (next_ == 3) {
+      return Result<std::optional<CopyOnly>>(std::in_place, std::nullopt);
+    }
+    return Result<std::optional<CopyOnly>>(std::in_place, std::in_place, next_++);
+  }
+
+  int next_ = 0;
+};
+
+// Exercises ToVector() with values that can be moved but not copied.
+class MoveOnlyIterator final : public Iterator<std::unique_ptr<int>> {
+ public:
+  int calls() const { return calls_; }
+
+ private:
+  Result<std::optional<std::unique_ptr<int>>> NextImpl() override {
+    ++calls_;
+    if (next_ == 3) {
+      return Result<std::optional<std::unique_ptr<int>>>(std::in_place, std::nullopt);
+    }
+    return Result<std::optional<std::unique_ptr<int>>>(std::in_place, std::in_place,
+                                                       std::make_unique<int>(next_++));
+  }
+
+  int next_ = 0;
+  int calls_ = 0;
+};
+
+// Exercises ToVector() error propagation after some values have been consumed.
+class FailingIterator final : public Iterator<int> {
+ public:
+  int calls() const { return calls_; }
+
+ private:
+  Result<std::optional<int>> NextImpl() override {
+    ++calls_;
+    if (next_ < 2) {
+      return Result<std::optional<int>>(std::in_place, std::in_place, next_++);
+    }
+    return Invalid("iteration failed");
+  }
+
+  int next_ = 0;
+  int calls_ = 0;
+};
+
+static_assert(std::is_move_constructible_v<CopyOnlyIterator>);
+static_assert(std::is_move_assignable_v<CopyOnlyIterator>);
+static_assert(std::is_move_constructible_v<MoveOnlyIterator>);
+static_assert(std::is_move_assignable_v<MoveOnlyIterator>);
+
+TEST(IteratorTest, ToVectorSupportsCopyOnlyValues) {
+  CopyOnlyIterator iterator;
+
+  ICEBERG_UNWRAP_OR_FAIL(auto values, iterator.ToVector());
+
+  ASSERT_EQ(values.size(), 3);
+  EXPECT_EQ(values[0].value(), 0);
+  EXPECT_EQ(values[1].value(), 1);
+  EXPECT_EQ(values[2].value(), 2);
+}
+
+TEST(IteratorTest, ToVectorSupportsMoveOnlyValues) {
+  MoveOnlyIterator iterator;
+
+  ICEBERG_UNWRAP_OR_FAIL(auto values, iterator.ToVector());
+
+  ASSERT_EQ(values.size(), 3);
+  EXPECT_EQ(*values[0], 0);
+  EXPECT_EQ(*values[1], 1);
+  EXPECT_EQ(*values[2], 2);
+}
+
+TEST(IteratorTest, NextRemainsAtEndAfterExhaustion) {
+  MoveOnlyIterator iterator;
+  ICEBERG_UNWRAP_OR_FAIL(auto values, iterator.ToVector());
+  ASSERT_EQ(values.size(), 3);
+  EXPECT_EQ(iterator.calls(), 4);
+
+  for (int i = 0; i < 2; ++i) {
+    auto result = iterator.Next();
+    ASSERT_TRUE(result.has_value());
+    EXPECT_FALSE(result->has_value());
+  }
+  EXPECT_EQ(iterator.calls(), 4);
+}
+
+TEST(IteratorTest, ToVectorPropagatesErrorsAfterPartialConsumption) {
+  FailingIterator iterator;
+
+  ICEBERG_UNWRAP_OR_FAIL(auto first, iterator.Next());
+  ASSERT_TRUE(first.has_value());
+  EXPECT_EQ(first.value(), 0);
+
+  auto result = iterator.ToVector();
+
+  EXPECT_THAT(result, IsError(ErrorKind::kInvalid));
+  EXPECT_THAT(result, HasErrorMessage("iteration failed"));
+}
+
+TEST(IteratorTest, NextRepeatsErrorWithoutAdvancing) {
+  FailingIterator iterator;
+  auto first_error = iterator.ToVector();
+  EXPECT_THAT(first_error, IsError(ErrorKind::kInvalid));
+  EXPECT_THAT(first_error, HasErrorMessage("iteration failed"));
+  EXPECT_EQ(iterator.calls(), 3);
+
+  for (int i = 0; i < 2; ++i) {
+    auto result = iterator.Next();
+    EXPECT_THAT(result, IsError(ErrorKind::kInvalid));
+    EXPECT_THAT(result, HasErrorMessage("iteration failed"));
+  }
+  EXPECT_EQ(iterator.calls(), 3);
+}
+
+}  // namespace
+}  // namespace iceberg
diff --git a/src/iceberg/test/meson.build b/src/iceberg/test/meson.build
index 6dde45ee9..977b8041f 100644
--- a/src/iceberg/test/meson.build
+++ b/src/iceberg/test/meson.build
@@ -108,6 +108,7 @@ iceberg_tests = {
             'executor_util_test.cc',
             'file_io_test.cc',
             'formatter_test.cc',
+            'iterator_test.cc',
             'lazy_test.cc',
             'location_util_test.cc',
             'math_util_internal_test.cc',
diff --git a/src/iceberg/type_fwd.h b/src/iceberg/type_fwd.h
index 0d2b17fae..0b19adaf5 100644
--- a/src/iceberg/type_fwd.h
+++ b/src/iceberg/type_fwd.h
@@ -229,6 +229,8 @@ struct SessionContext;
 
 /// \brief Task execution.
 class Executor;
+template <typename T>
+class Iterator;
 
 /// \brief Metrics reporting.
 class MetricsReporter;
diff --git a/src/iceberg/util/iterator.h b/src/iceberg/util/iterator.h
new file mode 100644
index 000000000..16717499f
--- /dev/null
+++ b/src/iceberg/util/iterator.h
@@ -0,0 +1,129 @@
+/*
+ * 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 <deque>
+#include <optional>
+#include <type_traits>
+#include <utility>
+#include <vector>
+
+#include "iceberg/result.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.
+/// Once Next() returns an error or std::nullopt, the iterator is terminal. Subsequent
+/// calls return the same terminal result without invoking the implementation again.
+///
+/// \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;
+  Iterator(Iterator&&) noexcept = default;
+  Iterator& operator=(Iterator&&) noexcept = default;
+
+  /// \brief Return the next value, or std::nullopt when the iterator is exhausted.
+  ///
+  /// After this method returns an error or std::nullopt, subsequent calls return the same
+  /// terminal result without invoking NextImpl().
+  Result<std::optional<T>> Next() {
+    if (error_.has_value()) {
+      return std::unexpected(*error_);
+    }
+    if (finished_) {
+      return std::nullopt;
+    }
+
+    auto result = NextImpl();
+    if (!result.has_value()) {
+      error_ = result.error();
+    } else if (!result.value().has_value()) {
+      finished_ = true;
+    }
+    return result;
+  }
+
+  /// \brief Consume the remaining values into a vector.
+  Result<std::vector<T>> ToVector() {
+    auto collect = [this](auto& values, auto append,
+                          auto finish) -> Result<std::vector<T>> {
+      while (true) {
+        auto result = Next();
+        if (!result.has_value()) {
+          return std::unexpected(std::move(result.error()));
+        }
+        auto& value = result.value();
+        if (!value.has_value()) {
+          return finish(values);
+        }
+        append(values, value.value());
+      }
+    };
+
+    if constexpr (!std::is_move_constructible_v<T>) {
+      static_assert(std::is_copy_constructible_v<T>,
+                    "Iterator::ToVector requires T to be move- or copy-constructible");
+
+      // For strictly copy-only T, collecting directly into a vector can repeatedly copy
+      // previously collected elements during vector growth. Stage values in a deque,
+      // then copy once into an exactly sized vector.
+      std::deque<T> values;
+      return collect(
+          values, [](auto& destination, const T& value) { destination.push_back(value); },
+          [](const auto& source) {
+            return std::vector<T>(source.cbegin(), source.cend());
+          });
+    } else {
+      std::vector<T> values;
+      return collect(
+          values,
+          [](auto& destination, T& value) {
+            destination.push_back(std::move_if_noexcept(value));
+          },
+          [](auto& source) { return std::move(source); });
+    }
+  }
+
+ protected:
+  /// \brief Produce the next value for Next().
+  ///
+  /// Implementations must return std::nullopt when exhausted. Next() makes the
+  /// terminal state sticky, so implementations are not called after exhaustion or error.
+  virtual Result<std::optional<T>> NextImpl() = 0;
+
+ private:
+  bool finished_ = false;
+  std::optional<Error> error_;
+};
+
+}  // namespace iceberg
diff --git a/src/iceberg/util/meson.build b/src/iceberg/util/meson.build
index f9436e7ed..831cc5888 100644
--- a/src/iceberg/util/meson.build
+++ b/src/iceberg/util/meson.build
@@ -32,6 +32,7 @@ install_headers(
         'formatter.h',
         'functional.h',
         'int128.h',
+        'iterator.h',
         'lazy.h',
         'location_util.h',
         'macros.h',