feat(inspect): add refs metadata table

src/iceberg/CMakeLists.txt

@@ -58,6 +58,7 @@ set(ICEBERG_SOURCES
     incremental_changelog_scan.cc
     inspect/history_table.cc
     inspect/metadata_table.cc
+    inspect/refs_table.cc
     inspect/snapshots_table.cc
     inheritable_metadata.cc
     json_serde.cc

src/iceberg/cpp20_compatibility_internal.h

@@ -80,6 +80,7 @@
 #include "iceberg/inheritable_metadata.h"
 #include "iceberg/inspect/history_table.h"
 #include "iceberg/inspect/metadata_table.h"
+#include "iceberg/inspect/refs_table.h"
 #include "iceberg/inspect/snapshots_table.h"
 #include "iceberg/location_provider.h"
 #include "iceberg/logging/cerr_logger.h"

src/iceberg/inspect/metadata_table.h

@@ -43,6 +43,7 @@ class ICEBERG_EXPORT MetadataTable {
   enum class Kind {
     kSnapshots,
     kHistory,
+    kRefs,
   };
 
   /// \brief Maximum number of rows emitted in each Arrow batch.

src/iceberg/inspect/metadata_table_stream_internal.h

@@ -0,0 +1,153 @@
+/*
+ * 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 <cstddef>
+#include <memory>
+#include <optional>
+#include <utility>
+#include <vector>
+
+#include <nanoarrow/nanoarrow.h>
+
+#include "iceberg/arrow_c_data_util_internal.h"
+#include "iceberg/arrow_row_builder_internal.h"
+#include "iceberg/inspect/metadata_table.h"
+#include "iceberg/nanoarrow_status_internal.h"
+#include "iceberg/schema.h"
+#include "iceberg/schema_internal.h"
+#include "iceberg/util/macros.h"
+#include "iceberg/util/stream.h"
+
+namespace iceberg::internal {
+
+/// \brief Arrow stream backed by a metadata-table row iterator.
+template <typename Row, typename AppendRow>
+class MetadataTableRowsStream : public Stream<ArrowArray> {
+ public:
+  static Result<std::unique_ptr<MetadataTableRowsStream>> Make(
+      const iceberg::Schema& schema, std::unique_ptr<Stream<Row>> rows,
+      AppendRow append_row) {
+    ICEBERG_PRECHECK(rows != nullptr, "Metadata table row source cannot be null");
+    if constexpr (requires { static_cast<bool>(append_row); }) {
+      ICEBERG_PRECHECK(static_cast<bool>(append_row),
+                       "Metadata table row callback cannot be empty");
+    }
+    ArrowSchema arrow_schema{};
+    ICEBERG_RETURN_UNEXPECTED(ToArrowSchema(schema, &arrow_schema));
+    return std::unique_ptr<MetadataTableRowsStream>(new MetadataTableRowsStream(
+        std::move(rows), std::move(append_row), std::move(arrow_schema)));
+  }
+
+  ~MetadataTableRowsStream() override {
+    if (arrow_schema_.release != nullptr) {
+      ArrowSchemaRelease(&arrow_schema_);
+    }
+  }
+
+  Status Close() {
+    rows_.reset();
+    append_row_.reset();
+    if (arrow_schema_.release != nullptr) {
+      ArrowSchemaRelease(&arrow_schema_);
+    }
+    return {};
+  }
+
+  Result<ArrowSchema> Schema() {
+    ICEBERG_PRECHECK(arrow_schema_.release != nullptr,
+                     "Cannot read schema from a closed metadata table stream");
+    ArrowSchema schema_copy{};
+    ICEBERG_NANOARROW_RETURN_UNEXPECTED(
+        ArrowSchemaDeepCopy(&arrow_schema_, &schema_copy));
+    return schema_copy;
+  }
+
+ protected:
+  Result<std::optional<ArrowArray>> NextImpl() override {
+    ICEBERG_PRECHECK(rows_ != nullptr, "Cannot read from a closed metadata table stream");
+    // Check for exhaustion before allocating a builder, including after a partial batch.
+    ICEBERG_ASSIGN_OR_RAISE(auto row, rows_->Next());
+    if (!row.has_value()) {
+      return std::nullopt;
+    }
+    ICEBERG_ASSIGN_OR_RAISE(auto builder, ArrowRowBuilder::Make(&arrow_schema_));
+    ICEBERG_RETURN_UNEXPECTED((*append_row_)(builder, *row));
+    while (builder.num_rows() < MetadataTable::kBatchSize) {
+      ICEBERG_ASSIGN_OR_RAISE(auto next_row, rows_->Next());
+      if (!next_row.has_value()) {
+        break;
+      }
+      ICEBERG_RETURN_UNEXPECTED((*append_row_)(builder, *next_row));
+    }
+    ICEBERG_ASSIGN_OR_RAISE(auto array, std::move(builder).Finish());
+    return array;
+  }
+
+ private:
+  MetadataTableRowsStream(std::unique_ptr<Stream<Row>> rows, AppendRow append_row,
+                          ArrowSchema arrow_schema)
+      : rows_(std::move(rows)),
+        append_row_(std::move(append_row)),
+        arrow_schema_(std::move(arrow_schema)) {}
+
+  std::unique_ptr<Stream<Row>> rows_;
+  std::optional<AppendRow> append_row_;
+  ArrowSchema arrow_schema_{};
+};
+
+template <typename Row, typename AppendRow>
+Result<ArrowArrayStream> MakeMetadataTableStream(const Schema& schema,
+                                                 std::unique_ptr<Stream<Row>> rows,
+                                                 AppendRow append_row) {
+  using RowsStream = MetadataTableRowsStream<Row, AppendRow>;
+  ICEBERG_ASSIGN_OR_RAISE(
+      auto stream, RowsStream::Make(schema, std::move(rows), std::move(append_row)));
+  return MakeArrowArrayStream(std::move(stream));
+}
+
+template <typename Row>
+class MetadataTableVectorIterator : public Stream<Row> {
+ public:
+  explicit MetadataTableVectorIterator(std::vector<Row> rows) : rows_(std::move(rows)) {}
+
+ protected:
+  Result<std::optional<Row>> NextImpl() override {
+    if (next_row_ == rows_.size()) {
+      return std::nullopt;
+    }
+    return std::move(rows_[next_row_++]);
+  }
+
+ private:
+  std::vector<Row> rows_;
+  size_t next_row_ = 0;
+};
+
+template <typename Row, typename AppendRow>
+Result<ArrowArrayStream> MakeMetadataTableStream(const Schema& schema,
+                                                 std::vector<Row> rows,
+                                                 AppendRow append_row) {
+  std::unique_ptr<Stream<Row>> iterator =
+      std::make_unique<MetadataTableVectorIterator<Row>>(std::move(rows));
+  return MakeMetadataTableStream(schema, std::move(iterator), std::move(append_row));
+}
+
+}  // namespace iceberg::internal

src/iceberg/inspect/refs_table.cc

@@ -0,0 +1,122 @@
+/*
+ * 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/refs_table.h"
+
+#include <algorithm>
+#include <cstdint>
+#include <memory>
+#include <optional>
+#include <string>
+#include <utility>
+#include <variant>
+#include <vector>
+
+#include "iceberg/arrow_row_builder_internal.h"
+#include "iceberg/inspect/metadata_table_stream_internal.h"
+#include "iceberg/schema.h"
+#include "iceberg/schema_field.h"
+#include "iceberg/snapshot.h"
+#include "iceberg/table.h"
+#include "iceberg/table_metadata.h"
+#include "iceberg/type.h"
+#include "iceberg/util/macros.h"
+
+namespace iceberg {
+namespace {
+
+struct RefRow {
+  std::string name;
+  std::string type;
+  int64_t snapshot_id;
+  std::optional<int64_t> max_ref_age_ms;
+  std::optional<int32_t> min_snapshots_to_keep;
+  std::optional<int64_t> max_snapshot_age_ms;
+};
+
+Status AppendOptional(ArrowArray* array, const auto& value) {
+  if (!value.has_value()) {
+    return AppendNull(array);
+  }
+  return AppendInt(array, static_cast<int64_t>(*value));
+}
+
+Status AppendRef(ArrowRowBuilder& builder, const RefRow& ref) {
+  ICEBERG_RETURN_UNEXPECTED(AppendString(builder.column(0), ref.name));
+  ICEBERG_RETURN_UNEXPECTED(AppendString(builder.column(1), ref.type));
+  ICEBERG_RETURN_UNEXPECTED(AppendInt(builder.column(2), ref.snapshot_id));
+  ICEBERG_RETURN_UNEXPECTED(AppendOptional(builder.column(3), ref.max_ref_age_ms));
+  ICEBERG_RETURN_UNEXPECTED(AppendOptional(builder.column(4), ref.min_snapshots_to_keep));
+  ICEBERG_RETURN_UNEXPECTED(AppendOptional(builder.column(5), ref.max_snapshot_age_ms));
+  return builder.FinishRow();
+}
+
+}  // namespace
+
+RefsTable::RefsTable(std::shared_ptr<Table> table) : MetadataTable(std::move(table)) {}
+
+RefsTable::~RefsTable() = default;
+
+const std::shared_ptr<Schema>& RefsTable::schema() const {
+  static const auto schema = std::make_shared<Schema>(std::vector<SchemaField>{
+      SchemaField::MakeRequired(1, "name", string()),
+      SchemaField::MakeRequired(2, "type", string()),
+      SchemaField::MakeRequired(3, "snapshot_id", int64()),
+      SchemaField::MakeOptional(4, "max_reference_age_in_ms", int64()),
+      SchemaField::MakeOptional(5, "min_snapshots_to_keep", int32()),
+      SchemaField::MakeOptional(6, "max_snapshot_age_in_ms", int64()),
+  });
+  return schema;
+}
+
+Result<std::unique_ptr<RefsTable>> RefsTable::Make(std::shared_ptr<Table> table) {
+  ICEBERG_PRECHECK(table != nullptr, "Table cannot be null");
+  return std::unique_ptr<RefsTable>(new RefsTable(std::move(table)));
+}
+
+Result<ArrowArrayStream> RefsTable::Scan() {
+  std::vector<RefRow> rows;
+  const auto& refs = source_table()->metadata()->refs;
+  rows.reserve(refs.size());
+  for (const auto& [name, ref] : refs) {
+    if (ref == nullptr) {
+      return Invalid("Snapshot reference '{}' is null", name);
+    }
+    const auto* branch = std::get_if<SnapshotRef::Branch>(&ref->retention);
+    RefRow row{.name = name,
+               .type = branch != nullptr ? "BRANCH" : "TAG",
+               .snapshot_id = ref->snapshot_id,
+               .max_ref_age_ms =
+                   branch != nullptr
+                       ? branch->max_ref_age_ms
+                       : std::get<SnapshotRef::Tag>(ref->retention).max_ref_age_ms};
+    if (branch != nullptr) {
+      row.min_snapshots_to_keep = branch->min_snapshots_to_keep;
+      row.max_snapshot_age_ms = branch->max_snapshot_age_ms;
+    }
+    rows.push_back(std::move(row));
+  }
+  std::ranges::sort(rows, {}, &RefRow::name);
+  return internal::MakeMetadataTableStream(
+      *schema(), std::move(rows), [](ArrowRowBuilder& builder, const RefRow& ref) {
+        return AppendRef(builder, ref);
+      });
+}
+
+}  // namespace iceberg

src/iceberg/inspect/refs_table.h

@@ -0,0 +1,51 @@
+/*
+ * 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
+
+/// \file iceberg/inspect/refs_table.h
+/// \brief Define the snapshot references metadata table.
+
+#include <memory>
+
+#include "iceberg/iceberg_export.h"
+#include "iceberg/inspect/metadata_table.h"
+#include "iceberg/result.h"
+#include "iceberg/type_fwd.h"
+
+namespace iceberg {
+
+/// \brief Metadata table containing the table's branch and tag snapshot references.
+class ICEBERG_EXPORT RefsTable : public MetadataTable {
+ public:
+  static Result<std::unique_ptr<RefsTable>> Make(std::shared_ptr<Table> table);
+
+  ~RefsTable() override;
+
+  Kind kind() const noexcept override { return Kind::kRefs; }
+
+  const std::shared_ptr<Schema>& schema() const override;
+
+  Result<ArrowArrayStream> Scan() override;
+
+ private:
+  explicit RefsTable(std::shared_ptr<Table> table);
+};
+
+}  // namespace iceberg