feat: add projection and filtering to manifest reader

- Consolidate `ManifestReader` implementation into `manifest_reader.cc`
  and remove `manifest_reader_internal.cc`.
- Implement fluent API for column selection, partition filtering, and
  row filtering.
- Support lazy initialization of the underlying Avro reader.
- Add various filtering support for entries.

Author: wgtmac

src/iceberg/CMakeLists.txt

@@ -42,7 +42,6 @@ set(ICEBERG_SOURCES
     manifest/manifest_entry.cc
     manifest/manifest_list.cc
     manifest/manifest_reader.cc
-    manifest/manifest_reader_internal.cc
     manifest/manifest_writer.cc
     manifest/v1_metadata.cc
     manifest/v2_metadata.cc

src/iceberg/manifest/manifest_list.cc

@@ -19,22 +19,24 @@
 
 #include "iceberg/manifest/manifest_list.h"
 
-#include "iceberg/schema.h"
+#include <memory>
+
+#include "iceberg/type.h"
 
 namespace iceberg {
 
-const StructType& PartitionFieldSummary::Type() {
-  static const StructType kInstance{{
+const std::shared_ptr<StructType>& PartitionFieldSummary::Type() {
+  static const auto kInstance = std::make_shared<StructType>(std::vector<SchemaField>{
       PartitionFieldSummary::kContainsNull,
       PartitionFieldSummary::kContainsNaN,
       PartitionFieldSummary::kLowerBound,
       PartitionFieldSummary::kUpperBound,
-  }};
+  });
   return kInstance;
 }
 
-const std::shared_ptr<Schema>& ManifestFile::Type() {
-  static const auto kInstance = std::make_shared<Schema>(std::vector<SchemaField>{
+const std::shared_ptr<StructType>& ManifestFile::Type() {
+  static const auto kInstance = std::make_shared<StructType>(std::vector<SchemaField>{
       kManifestPath,
       kManifestLength,
       kPartitionSpecId,

src/iceberg/manifest/manifest_list.h

@@ -69,7 +69,7 @@ struct ICEBERG_EXPORT PartitionFieldSummary {
 
   bool operator==(const PartitionFieldSummary& other) const = default;
 
-  static const StructType& Type();
+  static const std::shared_ptr<StructType>& Type();
 };
 
 /// \brief The type of files tracked by the manifest, either data or delete files; 0 for
@@ -197,7 +197,7 @@ struct ICEBERG_EXPORT ManifestFile {
 
   bool operator==(const ManifestFile& other) const = default;
 
-  static const std::shared_ptr<Schema>& Type();
+  static const std::shared_ptr<StructType>& Type();
 };
 
 /// Snapshots are embedded in table metadata, but the list of manifests for a snapshot are

src/iceberg/manifest/manifest_reader.cc

@@ -23,6 +23,7 @@
 /// Data reader interface for manifest files.
 
 #include <memory>
+#include <string>
 #include <vector>
 
 #include "iceberg/iceberg_export.h"
@@ -32,34 +33,96 @@
 
 namespace iceberg {
 
+class Expression;
+class PartitionSet;
+
 /// \brief Read manifest entries from a manifest file.
+///
+/// This class provides a fluent builder-style API to configure how manifest entries
+/// are read, including column selection, partition filtering, and row filtering.
+/// The underlying Avro reader is lazily created when Entries() or LiveEntries() is
+/// called.
 class ICEBERG_EXPORT ManifestReader {
  public:
+  /// \brief Special value to select all columns from manifest files.
+  inline static const std::vector<std::string> kAllColumns{"*"};
+
   virtual ~ManifestReader() = default;
 
   /// \brief Read all manifest entries in the manifest file.
-  virtual Result<std::vector<ManifestEntry>> Entries() const = 0;
+  ///
+  /// This method returns all entries including deleted entries. Filtering is applied
+  /// based on the configuration set via Select(), FilterPartitions(), and FilterRows().
+  virtual Result<std::vector<ManifestEntry>> Entries() = 0;
 
-  /// \brief Get the metadata of the manifest file.
-  virtual Result<std::unordered_map<std::string, std::string>> Metadata() const = 0;
+  /// \brief Read only live (non-deleted) manifest entries.
+  ///
+  /// This method returns only entries with status ADDED or EXISTING.
+  /// Filtering is applied based on the configuration.
+  virtual Result<std::vector<ManifestEntry>> LiveEntries() = 0;
+
+  /// \brief Select specific columns to read from the manifest entries.
+  ///
+  /// When filtering with FilterRows(), stats columns (value_counts, null_value_counts,
+  /// nan_value_counts, lower_bounds, upper_bounds, record_count) are automatically
+  /// included for filtering evaluation even if not specified in the select list.
+  ///
+  /// \param columns List of column names to read.
+  /// \return Reference to this reader for method chaining.
+  virtual ManifestReader& Select(const std::vector<std::string>& columns) = 0;
+
+  /// \brief Filter manifest entries by partition expression.
+  ///
+  /// \param expr Expression to filter partitions.
+  /// \return Reference to this reader for method chaining.
+  virtual ManifestReader& FilterPartitions(std::shared_ptr<Expression> expr) = 0;
+
+  /// \brief Filter manifest entries to a specific set of partitions.
+  ///
+  /// \param partition_set Set of partitions to include.
+  /// \return Reference to this reader for method chaining.
+  virtual ManifestReader& FilterPartitions(
+      std::shared_ptr<PartitionSet> partition_set) = 0;
+
+  /// \brief Filter manifest entries by row-level expression.
+  ///
+  /// This filter uses file-level statistics (lower_bounds, upper_bounds, null_counts,
+  /// etc.) to skip files that cannot contain matching rows. Stats columns are
+  /// automatically projected for evaluation.
+  ///
+  /// \param expr Expression to filter rows.
+  /// \return Reference to this reader for method chaining.
+  virtual ManifestReader& FilterRows(std::shared_ptr<Expression> expr) = 0;
+
+  /// \brief Set case sensitivity for column name matching.
+  ///
+  /// \param case_sensitive Whether column name matching is case-sensitive.
+  /// \return Reference to this reader for method chaining.
+  virtual ManifestReader& CaseSensitive(bool case_sensitive) = 0;
 
   /// \brief Creates a reader for a manifest file.
   /// \param manifest A ManifestFile object containing metadata about the manifest.
   /// \param file_io File IO implementation to use.
-  /// \param partition_type Schema for the partition.
+  /// \param schema Schema used to bind the partition type.
+  /// \param spec Partition spec used for this manifest file.
   /// \return A Result containing the reader or an error.
   static Result<std::unique_ptr<ManifestReader>> Make(
       const ManifestFile& manifest, std::shared_ptr<FileIO> file_io,
-      std::shared_ptr<StructType> partition_type);
+      std::shared_ptr<Schema> schema, std::shared_ptr<PartitionSpec> spec);
 
   /// \brief Creates a reader for a manifest file.
   /// \param manifest_location Path to the manifest file.
   /// \param file_io File IO implementation to use.
-  /// \param partition_type Schema for the partition.
+  /// \param schema Schema used to bind the partition type.
+  /// \param spec Partition spec used for this manifest file.
   /// \return A Result containing the reader or an error.
   static Result<std::unique_ptr<ManifestReader>> Make(
       std::string_view manifest_location, std::shared_ptr<FileIO> file_io,
-      std::shared_ptr<StructType> partition_type);
+      std::shared_ptr<Schema> schema, std::shared_ptr<PartitionSpec> spec);
+
+  /// \brief Add stats columns to the column list if needed.
+  static std::vector<std::string> WithStatsColumns(
+      const std::vector<std::string>& columns);
 };
 
 /// \brief Read manifest files from a manifest list file.