fix: align manifest planning stats semantics

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

Author: manuzhang

src/iceberg/manifest/manifest_group.cc

@@ -142,10 +142,8 @@ class ManifestGroup::FilePlanningIterator final
     group->delete_index_builder_.WithScanMetrics(group->scan_metrics_);
     ICEBERG_ASSIGN_OR_RAISE(auto delete_index, group->delete_index_builder_.Build());
 
-    const bool drop_stats = ManifestReader::ShouldDropStats(group->columns_);
-    if (delete_index->has_equality_deletes()) {
-      group->columns_ = ManifestReader::WithStatsColumns(group->columns_);
-    }
+    const bool drop_stats =
+        group->PrepareStatsProjection(delete_index->has_equality_deletes());
 
     std::unique_ptr<Evaluator> data_file_evaluator;
     if (group->file_filter_ &&
@@ -426,6 +424,10 @@ class ManifestGroup::FilePlanningIterator final
   int32_t current_spec_id_ = 0;
   bool drop_stats_;
 
+  // Limit the number of manifest readers and iterators retained by executor-backed
+  // planning. The executor still controls actual task concurrency, while this fixed
+  // cap prevents resource use from scaling with the total manifest count. Entries
+  // within each manifest remain streamed, so this does not cap manifest size.
   static constexpr size_t kManifestReadBatchSize = 32;
 };
 
@@ -583,10 +585,7 @@ Result<std::vector<std::shared_ptr<ScanTask>>> ManifestGroup::Plan(
   delete_index_builder_.WithScanMetrics(scan_metrics_);
   ICEBERG_ASSIGN_OR_RAISE(auto delete_index, delete_index_builder_.Build());
 
-  bool drop_stats = ManifestReader::ShouldDropStats(columns_);
-  if (delete_index->has_equality_deletes()) {
-    columns_ = ManifestReader::WithStatsColumns(columns_);
-  }
+  const bool drop_stats = PrepareStatsProjection(delete_index->has_equality_deletes());
 
   std::unordered_map<int32_t, std::unique_ptr<TaskContext>> task_context_cache;
   auto get_task_context = [&](int32_t spec_id) -> Result<TaskContext*> {
@@ -680,6 +679,18 @@ Result<std::unique_ptr<ManifestReader>> ManifestGroup::MakeReader(
   return reader;
 }
 
+bool ManifestGroup::PrepareStatsProjection(bool has_equality_deletes) {
+  // The caller's projection records whether stats were requested. Equality-delete
+  // matching may add stats temporarily, but they should still be dropped from the
+  // result when the original projection did not request them. Keeping this decision
+  // here ensures eager and iterator planning use identical semantics.
+  const bool drop_stats = ManifestReader::ShouldDropStats(columns_);
+  if (has_equality_deletes) {
+    columns_ = ManifestReader::WithStatsColumns(columns_);
+  }
+  return drop_stats;
+}
+
 Result<std::unordered_map<int32_t, std::vector<ManifestEntry>>>
 ManifestGroup::ReadEntries() {
   // TODO(zehua): Replace with a thread-safe LRU cache.

src/iceberg/manifest/manifest_group.h

@@ -173,6 +173,8 @@ class ICEBERG_EXPORT ManifestGroup : public ErrorCollector {
 
   Result<std::unique_ptr<ManifestReader>> MakeReader(const ManifestFile& manifest);
 
+  bool PrepareStatsProjection(bool has_equality_deletes);
+
   std::shared_ptr<FileIO> io_;
   std::shared_ptr<Schema> schema_;
   std::unordered_map<int32_t, std::shared_ptr<PartitionSpec>> specs_by_id_;

src/iceberg/test/manifest_group_test.cc

@@ -329,6 +329,8 @@ TEST_P(ManifestGroupTest, PlanFilesIteratorPreservesSelectAllWithEqualityDeletes
   ASSERT_TRUE(task.has_value());
   EXPECT_EQ(task.value()->data_file()->file_path, "/path/to/data.parquet");
   EXPECT_EQ(task.value()->data_file()->record_count, 100);
+  EXPECT_TRUE(task.value()->data_file()->lower_bounds.contains(1));
+  EXPECT_TRUE(task.value()->data_file()->upper_bounds.contains(1));
   ASSERT_EQ(task.value()->delete_files().size(), 1);
   EXPECT_EQ(task.value()->delete_files().front()->file_path,
             "/path/to/equality-delete.parquet");