feat(inspect): iterate live files one manifest at a time

src/iceberg/CMakeLists.txt

@@ -58,6 +58,7 @@ set(ICEBERG_SOURCES
     geospatial.cc
     incremental_append_scan.cc
     incremental_changelog_scan.cc
+    inspect/metadata_manifest_util_internal.cc
     inspect/history_table.cc
     inspect/metadata_scan.cc
     inspect/metadata_table.cc

src/iceberg/inspect/metadata_manifest_util_internal.cc

@@ -0,0 +1,118 @@
+/*
+ * 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/inspect/metadata_manifest_util_internal.h"
+
+#include <memory>
+#include <optional>
+#include <string>
+#include <unordered_map>
+#include <utility>
+#include <vector>
+
+#include "iceberg/constants.h"
+#include "iceberg/manifest/manifest_entry.h"
+#include "iceberg/manifest/manifest_list.h"
+#include "iceberg/manifest/manifest_reader.h"
+#include "iceberg/partition_spec.h"
+#include "iceberg/snapshot.h"
+#include "iceberg/table.h"
+#include "iceberg/table_metadata.h"
+#include "iceberg/util/macros.h"
+
+namespace iceberg::internal {
+
+namespace {
+
+class LiveFilesIterator : public Stream<LiveFile> {
+ public:
+  LiveFilesIterator(std::shared_ptr<FileIO> io, std::shared_ptr<Schema> schema,
+                    std::unordered_map<int32_t, std::shared_ptr<PartitionSpec>> specs,
+                    std::vector<ManifestFile> manifests, std::vector<std::string> columns)
+      : io_(std::move(io)),
+        schema_(std::move(schema)),
+        specs_(std::move(specs)),
+        manifests_(std::move(manifests)),
+        columns_(std::move(columns)) {}
+
+ protected:
+  Result<std::optional<LiveFile>> NextImpl() override {
+    while (entry_index_ == entries_.size()) {
+      entries_.clear();
+      entry_index_ = 0;
+      if (manifest_index_ == manifests_.size()) {
+        return std::nullopt;
+      }
+      const auto& manifest = manifests_[manifest_index_++];
+      auto spec = specs_.find(manifest.partition_spec_id);
+      ICEBERG_CHECK(spec != specs_.end(),
+                    "Cannot find partition spec {} for manifest '{}'",
+                    manifest.partition_spec_id, manifest.manifest_path);
+      ICEBERG_PRECHECK(spec->second != nullptr, "Partition spec {} is null",
+                       manifest.partition_spec_id);
+      current_spec_ = spec->second;
+      ICEBERG_ASSIGN_OR_RAISE(auto reader,
+                              ManifestReader::Make(manifest, io_, schema_, specs_));
+      if (!columns_.empty()) {
+        reader->Select(columns_);
+      }
+      ICEBERG_ASSIGN_OR_RAISE(entries_, reader->LiveEntries());
+    }
+    auto& entry = entries_[entry_index_++];
+    ICEBERG_PRECHECK(entry.data_file != nullptr,
+                     "Manifest contains an entry with no data file");
+    return LiveFile{.file = std::move(entry.data_file),
+                    .spec = current_spec_,
+                    .snapshot_id = entry.snapshot_id};
+  }
+
+ private:
+  std::shared_ptr<FileIO> io_;
+  std::shared_ptr<Schema> schema_;
+  std::unordered_map<int32_t, std::shared_ptr<PartitionSpec>> specs_;
+  std::vector<ManifestFile> manifests_;
+  std::vector<std::string> columns_;
+  std::vector<ManifestEntry> entries_;
+  std::shared_ptr<PartitionSpec> current_spec_;
+  size_t manifest_index_ = 0;
+  size_t entry_index_ = 0;
+};
+
+}  // namespace
+
+Result<std::unique_ptr<Stream<LiveFile>>> LiveFiles(
+    const TableMetadata& metadata, const std::shared_ptr<FileIO>& io,
+    const std::shared_ptr<Snapshot>& snapshot, std::vector<std::string> columns) {
+  ICEBERG_ASSIGN_OR_RAISE(auto schema, metadata.Schema());
+  std::unordered_map<int32_t, std::shared_ptr<PartitionSpec>> specs;
+  for (const auto& spec : metadata.partition_specs) {
+    ICEBERG_PRECHECK(spec != nullptr, "Partition spec cannot be null");
+    specs.emplace(spec->spec_id(), spec);
+  }
+  std::vector<ManifestFile> manifests;
+  if (snapshot != nullptr) {
+    SnapshotReader snapshot_reader(snapshot.get());
+    ICEBERG_ASSIGN_OR_RAISE(auto snapshot_manifests, snapshot_reader.Manifests(io));
+    manifests.assign(snapshot_manifests.begin(), snapshot_manifests.end());
+  }
+  return std::make_unique<LiveFilesIterator>(io, std::move(schema), std::move(specs),
+                                             std::move(manifests), std::move(columns));
+}
+
+}  // namespace iceberg::internal

src/iceberg/inspect/metadata_manifest_util_internal.h

@@ -0,0 +1,48 @@
+/*
+ * 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
+
+#include <cstdint>
+#include <memory>
+#include <optional>
+#include <string>
+#include <vector>
+
+#include "iceberg/iceberg_export.h"
+#include "iceberg/result.h"
+#include "iceberg/type_fwd.h"
+#include "iceberg/util/stream.h"
+
+namespace iceberg::internal {
+
+struct LiveFile {
+  std::shared_ptr<DataFile> file;
+  std::shared_ptr<PartitionSpec> spec;
+  std::optional<int64_t> snapshot_id;
+};
+
+/// \brief Iterate over live files, reading one manifest at a time.
+/// An empty column selection reads all data-file fields. Each manifest's entries
+/// are materialized by ManifestReader, independently of the Arrow batch size.
+ICEBERG_EXPORT Result<std::unique_ptr<Stream<LiveFile>>> LiveFiles(
+    const TableMetadata& metadata, const std::shared_ptr<FileIO>& io,
+    const std::shared_ptr<Snapshot>& snapshot, std::vector<std::string> columns = {});
+
+}  // namespace iceberg::internal

src/iceberg/test/CMakeLists.txt

@@ -206,6 +206,7 @@ if(ICEBERG_BUILD_BUNDLE)
   add_iceberg_test(metadata_table_test
                    USE_BUNDLE
                    SOURCES
+                   inspect/metadata_manifest_util_test.cc
                    inspect/history_table_test.cc
                    inspect/metadata_table_test.cc
                    inspect/refs_table_test.cc

src/iceberg/test/inspect/metadata_manifest_util_test.cc

@@ -0,0 +1,103 @@
+/*
+ * 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 <cstdint>
+#include <memory>
+#include <optional>
+#include <string>
+#include <utility>
+#include <vector>
+
+#include <gtest/gtest.h>
+
+#include "iceberg/inspect/metadata_manifest_util_internal.h"
+#include "iceberg/manifest/manifest_entry.h"
+#include "iceberg/snapshot.h"
+#include "iceberg/table_metadata.h"
+#include "iceberg/test/matchers.h"
+#include "iceberg/test/scan_test_base.h"
+#include "iceberg/util/timepoint.h"
+
+namespace iceberg::internal {
+namespace {
+class MetadataManifestTest : public ScanTestBase {
+ protected:
+  std::shared_ptr<Snapshot> SnapshotWith(std::vector<ManifestFile> manifests) {
+    return std::make_shared<Snapshot>(
+        Snapshot{.snapshot_id = 10,
+                 .sequence_number = 1,
+                 .timestamp_ms = TimePointMsFromUnixMs(1000),
+                 .manifest_list = WriteManifestList(GetParam(), 10, 0, 1, manifests)});
+  }
+};
+
+TEST_P(MetadataManifestTest, ReadsLiveFilesAcrossManifestsAndRetainsInputs) {
+  auto first = WriteDataManifest(
+      GetParam(), 10,
+      {MakeEntry(ManifestStatus::kAdded, 10, 1, MakeDataFile("first.parquet")),
+       MakeEntry(ManifestStatus::kDeleted, 10, 1, MakeDataFile("deleted.parquet"))});
+  auto second = WriteDataManifest(
+      GetParam(), 10,
+      {MakeEntry(ManifestStatus::kAdded, 10, 1, MakeDataFile("second.parquet"))});
+  auto snapshot = SnapshotWith({first, second});
+  auto metadata = MakeTableMetadata({snapshot}, 10);
+  ICEBERG_UNWRAP_OR_FAIL(auto files, LiveFiles(*metadata, file_io_, snapshot));
+  metadata.reset();
+  snapshot.reset();
+  file_io_.reset();
+  for (const auto& path : {"first.parquet", "second.parquet"}) {
+    ICEBERG_UNWRAP_OR_FAIL(auto row, files->Next());
+    ASSERT_TRUE(row.has_value());
+    EXPECT_EQ(row->file->file_path, path);
+    ASSERT_NE(row->spec, nullptr);
+  }
+  for (int attempt = 0; attempt < 2; ++attempt) {
+    ICEBERG_UNWRAP_OR_FAIL(auto row, files->Next());
+    EXPECT_FALSE(row.has_value());
+  }
+}
+
+TEST_P(MetadataManifestTest, DefersManifestReadsAndKeepsFailuresTerminal) {
+  auto first = WriteDataManifest(
+      GetParam(), 10,
+      {MakeEntry(ManifestStatus::kAdded, 10, 1, MakeDataFile("first.parquet"))});
+  auto missing = first;
+  missing.manifest_path = "missing-manifest.avro";
+  auto snapshot = SnapshotWith({first, missing});
+  auto metadata = MakeTableMetadata({snapshot}, 10);
+  ICEBERG_UNWRAP_OR_FAIL(auto files, LiveFiles(*metadata, file_io_, snapshot));
+  ICEBERG_UNWRAP_OR_FAIL(auto row, files->Next());
+  ASSERT_TRUE(row.has_value());
+  EXPECT_EQ(row->file->file_path, "first.parquet");
+  for (int attempt = 0; attempt < 2; ++attempt) {
+    EXPECT_FALSE(files->Next().has_value());
+  }
+}
+
+TEST_P(MetadataManifestTest, EmptyTablesHaveNoLiveFiles) {
+  auto metadata = MakeTableMetadata({}, kInvalidSnapshotId);
+  ICEBERG_UNWRAP_OR_FAIL(auto files, LiveFiles(*metadata, file_io_, nullptr));
+  ICEBERG_UNWRAP_OR_FAIL(auto row, files->Next());
+  EXPECT_FALSE(row.has_value());
+}
+
+INSTANTIATE_TEST_SUITE_P(FormatVersions, MetadataManifestTest,
+                         ::testing::Values(1, 2, 3));
+}  // namespace
+}  // namespace iceberg::internal