fix: retain partition values during scan planning

Preserve partitions for delete matching with narrow projections and cover equality and position deletes. Document the intentional stream API migrations.

Co-authored-by: Codex <codex@openai.com>

Author: manuzhang

src/iceberg/manifest/manifest_group.cc

@@ -645,6 +645,13 @@ ManifestGroup::StatsProjection ManifestGroup::PrepareStatsProjection(
   // here ensures eager and stream planning use identical semantics.
   StatsProjection result{.columns = columns_,
                          .drop_stats = ManifestReader::ShouldDropStats(columns_)};
+  // Delete matching and residual evaluation require partition values even when the
+  // caller does not select them. Do not narrow an empty or select-all projection.
+  if (!result.columns.empty() &&
+      !std::ranges::contains(result.columns, Schema::kAllColumns) &&
+      !std::ranges::contains(result.columns, DataFile::kPartitionField)) {
+    result.columns.emplace_back(DataFile::kPartitionField);
+  }
   if (has_equality_deletes) {
     result.columns = ManifestReader::WithStatsColumns(result.columns);
   }

src/iceberg/manifest/manifest_group.h

@@ -114,6 +114,9 @@ class ICEBERG_EXPORT ManifestGroup : public ErrorCollector {
 
   /// \brief Select specific columns from manifest entries.
   ///
+  /// Task planning also reads partition values for delete matching and residuals, and
+  /// may temporarily read statistics needed for equality-delete matching.
+  ///
   /// \param columns Column names to select from manifest entries.
   ManifestGroup& Select(std::vector<std::string> columns);
 
@@ -140,6 +143,9 @@ class ICEBERG_EXPORT ManifestGroup : public ErrorCollector {
   /// \brief Plan scan tasks for all matching data files.
   ///
   /// Consumes this group and collects PlanFilesStream() into a vector.
+  /// Callers must use std::move(group).PlanFiles(), or std::move(*group).PlanFiles()
+  /// for an owning pointer, instead of calling on an lvalue. Do not reuse the consumed
+  /// group; construct a new group to plan again.
   Result<std::vector<std::shared_ptr<FileScanTask>>> PlanFiles() &&;
 
   /// \brief Lazily plan scan tasks for matching data files.
@@ -154,7 +160,9 @@ class ICEBERG_EXPORT ManifestGroup : public ErrorCollector {
   /// manifest at a time. Delete manifests are still read eagerly when creating the
   /// stream because delete files must be indexed before data-file planning can begin.
   /// Creating the stream consumes this group's configuration, so this method may only
-  /// be called on an rvalue.
+  /// be called on an rvalue. Use std::move(group).PlanFilesStream(), or
+  /// std::move(*group).PlanFilesStream() for an owning pointer, and do not reuse the
+  /// group.
   Result<FileScanTaskStreamPtr> PlanFilesStream() &&;
 
   /// \brief Get all matching manifest entries.

src/iceberg/manifest/manifest_reader.h

@@ -45,6 +45,12 @@ using ManifestEntryStream = Stream<ManifestEntry>;
 using ManifestEntryStreamPtr = std::unique_ptr<ManifestEntryStream>;
 
 /// \brief Read manifest entries from a manifest file.
+///
+/// Implementations must override EntriesStream() and LiveEntriesStream(), returning
+/// self-contained streams that may outlive the reader. These are the extension points;
+/// Entries() and LiveEntries() are non-virtual eager collection helpers. Custom readers
+/// that previously overrode the eager methods must migrate to the stream methods;
+/// no eager-to-stream fallback is provided.
 class ICEBERG_EXPORT ManifestReader {
  public:
   virtual ~ManifestReader() = default;

src/iceberg/test/manifest_group_test.cc

@@ -369,18 +369,77 @@ TEST_P(ManifestGroupTest, PlanFilesDropsUnselectedStatsWithEqualityDeletes) {
                                      /*sequence_number=*/2, std::move(equality_delete))},
                           partitioned_spec_);
 
-  ICEBERG_UNWRAP_OR_FAIL(
-      auto group, ManifestGroup::Make(file_io_, schema_, GetSpecsById(), {data_manifest},
-                                      {delete_manifest}));
-  group->Select({"file_path"});
+  for (bool use_executor : {false, true}) {
+    SCOPED_TRACE(std::format("use_executor={}", use_executor));
+    for (bool use_stream : {false, true}) {
+      SCOPED_TRACE(std::format("use_stream={}", use_stream));
+      test::ThreadExecutor executor;
+      ICEBERG_UNWRAP_OR_FAIL(auto group,
+                             ManifestGroup::Make(file_io_, schema_, GetSpecsById(),
+                                                 {data_manifest}, {delete_manifest}));
+      group->Select({"file_path"});
+      if (use_executor) {
+        group->PlanWith(std::ref(executor));
+      }
+
+      std::vector<std::shared_ptr<FileScanTask>> tasks;
+      if (use_stream) {
+        ICEBERG_UNWRAP_OR_FAIL(auto stream, std::move(*group).PlanFilesStream());
+        ICEBERG_UNWRAP_OR_FAIL(tasks, stream->ToVector());
+      } else {
+        ICEBERG_UNWRAP_OR_FAIL(tasks, std::move(*group).PlanFiles());
+      }
+      ASSERT_EQ(tasks.size(), 1);
+      EXPECT_EQ(tasks.front()->data_file()->partition, part_value);
+      ASSERT_EQ(tasks.front()->delete_files().size(), 1);
+      EXPECT_EQ(tasks.front()->delete_files().front()->file_path,
+                "/path/to/equality-delete.parquet");
+      EXPECT_TRUE(tasks.front()->data_file()->lower_bounds.empty());
+      EXPECT_TRUE(tasks.front()->data_file()->upper_bounds.empty());
+    }
+  }
+}
 
-  ICEBERG_UNWRAP_OR_FAIL(auto tasks, std::move(*group).PlanFiles());
-  ASSERT_EQ(tasks.size(), 1);
-  ASSERT_EQ(tasks.front()->delete_files().size(), 1);
-  EXPECT_EQ(tasks.front()->delete_files().front()->file_path,
-            "/path/to/equality-delete.parquet");
-  EXPECT_TRUE(tasks.front()->data_file()->lower_bounds.empty());
-  EXPECT_TRUE(tasks.front()->data_file()->upper_bounds.empty());
+TEST_P(ManifestGroupTest, PlanFilesStreamPreservesPartitionWithPositionDeletes) {
+  auto version = GetParam();
+  if (version < 2) {
+    GTEST_SKIP() << "Position deletes only supported in V2+";
+  }
+
+  constexpr int64_t kSnapshotId = 1000L;
+  const auto part_value = PartitionValues({Literal::Int(0)});
+  auto data_manifest = WriteDataManifest(
+      version, kSnapshotId,
+      {MakeEntry(ManifestStatus::kAdded, kSnapshotId, /*sequence_number=*/1,
+                 MakeDataFile("/path/to/data.parquet", part_value,
+                              partitioned_spec_->spec_id()))},
+      partitioned_spec_);
+  auto delete_manifest = WriteDeleteManifest(
+      version, kSnapshotId,
+      {MakeEntry(ManifestStatus::kAdded, kSnapshotId, /*sequence_number=*/2,
+                 MakePositionDeleteFile("/path/to/position-delete.parquet", part_value,
+                                        partitioned_spec_->spec_id()))},
+      partitioned_spec_);
+
+  for (bool use_executor : {false, true}) {
+    SCOPED_TRACE(std::format("use_executor={}", use_executor));
+    test::ThreadExecutor executor;
+    ICEBERG_UNWRAP_OR_FAIL(auto group,
+                           ManifestGroup::Make(file_io_, schema_, GetSpecsById(),
+                                               {data_manifest}, {delete_manifest}));
+    group->Select({"file_path"});
+    if (use_executor) {
+      group->PlanWith(std::ref(executor));
+    }
+
+    ICEBERG_UNWRAP_OR_FAIL(auto stream, std::move(*group).PlanFilesStream());
+    ICEBERG_UNWRAP_OR_FAIL(auto tasks, stream->ToVector());
+    ASSERT_EQ(tasks.size(), 1);
+    EXPECT_EQ(tasks.front()->data_file()->partition, part_value);
+    ASSERT_EQ(tasks.front()->delete_files().size(), 1);
+    EXPECT_EQ(tasks.front()->delete_files().front()->file_path,
+              "/path/to/position-delete.parquet");
+  }
 }
 
 TEST_P(ManifestGroupTest, IgnoreDeleted) {

src/iceberg/util/stream.h

@@ -40,6 +40,10 @@ namespace iceberg {
 /// Once Next() returns an error or std::nullopt, the stream is terminal. Subsequent
 /// calls return the same terminal result without invoking the implementation again.
 ///
+/// This interface replaces Iterator<T>. Include iceberg/util/stream.h instead of
+/// iceberg/util/iterator.h and derive from Stream<T>, retaining the NextImpl() override.
+/// The former header and type alias are not provided.
+///
 /// \tparam T Value returned by the stream.
 template <typename T>
 class Stream {