fix(inspect): address metadata table review feedback

Author: WZhuo

src/iceberg/inspect/metadata_table.h

@@ -75,8 +75,8 @@ class ICEBERG_EXPORT MetadataTable {
 
   /// \brief Scan the metadata table without time travel.
   ///
-  /// The caller owns the returned stream and must release it with
-  /// ArrowArrayStreamRelease.
+  /// The caller owns the returned stream and must invoke its `release` callback
+  /// when the stream is no longer needed.
   virtual Result<ArrowArrayStream> Scan() = 0;
 
  protected:
@@ -100,6 +100,9 @@ struct SnapshotSelection {
 };
 
 /// \brief Base interface for metadata tables that support time travel.
+///
+/// Keeping snapshot selection on this capability-specific interface prevents it from
+/// being exposed by metadata tables that do not support time travel.
 class ICEBERG_EXPORT TimeTravelMetadataTable : public MetadataTable {
  public:
   ~TimeTravelMetadataTable() override;

src/iceberg/inspect/snapshots_table.cc

@@ -23,6 +23,7 @@
 #include <cstddef>
 #include <memory>
 #include <optional>
+#include <tuple>
 #include <utility>
 #include <vector>
 
@@ -36,6 +37,7 @@
 #include "iceberg/schema_internal.h"
 #include "iceberg/snapshot.h"
 #include "iceberg/table.h"
+#include "iceberg/table_metadata.h"
 #include "iceberg/type.h"
 #include "iceberg/util/macros.h"
 
@@ -78,25 +80,28 @@ Status AppendSnapshot(ArrowRowBuilder& builder, const Snapshot& snapshot) {
 class SnapshotsTableStream {
  public:
   static Result<std::unique_ptr<SnapshotsTableStream>> Make(
-      std::shared_ptr<Table> table, const iceberg::Schema& schema) {
+      std::shared_ptr<const TableMetadata> metadata, const iceberg::Schema& schema) {
     ArrowSchema arrow_schema{};
     ICEBERG_RETURN_UNEXPECTED(ToArrowSchema(schema, &arrow_schema));
     return std::unique_ptr<SnapshotsTableStream>(
-        new SnapshotsTableStream(std::move(table), std::move(arrow_schema)));
+        new SnapshotsTableStream(std::move(metadata), std::move(arrow_schema)));
   }
 
   ~SnapshotsTableStream() { std::ignore = Close(); }
 
   Status Close() {
-    table_.reset();
+    metadata_.reset();
     if (arrow_schema_.release != nullptr) {
       ArrowSchemaRelease(&arrow_schema_);
     }
     return {};
   }
 
   Result<std::optional<ArrowArray>> Next() {
-    const auto& snapshots = table_->snapshots();
+    if (metadata_ == nullptr) [[unlikely]] {
+      return InvalidArgument("Cannot read from a closed snapshots table stream");
+    }
+    const auto& snapshots = metadata_->snapshots;
     if (next_snapshot_ == snapshots.size()) {
       return std::nullopt;
     }
@@ -129,10 +134,11 @@ class SnapshotsTableStream {
   }
 
  private:
-  SnapshotsTableStream(std::shared_ptr<Table> table, ArrowSchema arrow_schema)
-      : table_(std::move(table)), arrow_schema_(std::move(arrow_schema)) {}
+  SnapshotsTableStream(std::shared_ptr<const TableMetadata> metadata,
+                       ArrowSchema arrow_schema)
+      : metadata_(std::move(metadata)), arrow_schema_(std::move(arrow_schema)) {}
 
-  std::shared_ptr<Table> table_;
+  std::shared_ptr<const TableMetadata> metadata_;
   ArrowSchema arrow_schema_{};
   size_t next_snapshot_ = 0;
 };
@@ -165,8 +171,8 @@ Result<std::unique_ptr<SnapshotsTable>> SnapshotsTable::Make(
 }
 
 Result<ArrowArrayStream> SnapshotsTable::Scan() {
-  ICEBERG_ASSIGN_OR_RAISE(auto stream,
-                          SnapshotsTableStream::Make(source_table(), *schema()));
+  ICEBERG_ASSIGN_OR_RAISE(
+      auto stream, SnapshotsTableStream::Make(source_table()->metadata(), *schema()));
   return MakeArrowArrayStream(std::move(stream));
 }
 

src/iceberg/test/metadata_table_test_base.h

@@ -27,6 +27,7 @@
 #pragma once
 
 #include <memory>
+#include <string>
 #include <unordered_map>
 #include <utility>
 #include <vector>
@@ -143,7 +144,8 @@ class MetadataTableTestBase : public ::testing::Test {
 
   /// \brief Create a Table with the given snapshots.
   Result<std::shared_ptr<Table>> MakeTableWithSnapshots(
-      std::vector<std::shared_ptr<Snapshot>> snapshots, int64_t current_snapshot_id) {
+      std::vector<std::shared_ptr<Snapshot>> snapshots, int64_t current_snapshot_id,
+      std::string metadata_location = "s3://bucket/meta.json") {
     auto schema = std::make_shared<Schema>(
         std::vector<SchemaField>{SchemaField::MakeRequired(1, "id", int64()),
                                  SchemaField::MakeOptional(2, "name", string())},
@@ -157,7 +159,7 @@ class MetadataTableTestBase : public ::testing::Test {
     });
 
     TableIdentifier ident{.ns = Namespace{.levels = {"db"}}, .name = "test_table"};
-    return Table::Make(ident, metadata, "s3://bucket/meta.json", io_, catalog_);
+    return Table::Make(ident, metadata, std::move(metadata_location), io_, catalog_);
   }
 
   std::shared_ptr<MockFileIO> io_;

src/iceberg/test/snapshots_table_test.cc

@@ -194,4 +194,36 @@ TEST_F(SnapshotsTableTest, ScanReturnsMultipleBatches) {
   EXPECT_EQ(batches[1]->num_rows(), 1);
 }
 
+TEST_F(SnapshotsTableTest, ScanUsesMetadataCapturedAtCreation) {
+  auto snapshot = MakeTestSnapshots().first;
+  std::vector<std::shared_ptr<Snapshot>> snapshots(1025, snapshot);
+  ICEBERG_UNWRAP_OR_FAIL(auto table, MakeTableWithSnapshots(std::move(snapshots),
+                                                            /*current_snapshot_id=*/1));
+  ICEBERG_UNWRAP_OR_FAIL(auto snapshots_table,
+                         MetadataTable::Make<SnapshotsTable>(table));
+
+  ICEBERG_UNWRAP_OR_FAIL(auto stream, snapshots_table->Scan());
+  auto reader_result = ::arrow::ImportRecordBatchReader(&stream);
+  ASSERT_TRUE(reader_result.ok()) << reader_result.status().ToString();
+  auto reader = std::move(reader_result).ValueUnsafe();
+
+  auto first_batch_result = reader->Next();
+  ASSERT_TRUE(first_batch_result.ok()) << first_batch_result.status().ToString();
+  ASSERT_NE(*first_batch_result, nullptr);
+  EXPECT_EQ((*first_batch_result)->num_rows(), MetadataTable::kBatchSize);
+
+  ICEBERG_UNWRAP_OR_FAIL(
+      auto refreshed_table,
+      MakeTableWithSnapshots({}, /*current_snapshot_id=*/kInvalidSnapshotId,
+                             "s3://bucket/refreshed-meta.json"));
+  EXPECT_CALL(*catalog_, LoadTable(table->name()))
+      .WillOnce(::testing::Return(refreshed_table));
+  ASSERT_THAT(table->Refresh(), IsOk());
+
+  auto second_batch_result = reader->Next();
+  ASSERT_TRUE(second_batch_result.ok()) << second_batch_result.status().ToString();
+  ASSERT_NE(*second_batch_result, nullptr);
+  EXPECT_EQ((*second_batch_result)->num_rows(), 1);
+}
+
 }  // namespace iceberg