feat(inspect): support selecting columns in metadata table scans

src/iceberg/CMakeLists.txt

@@ -57,6 +57,7 @@ set(ICEBERG_SOURCES
     incremental_append_scan.cc
     incremental_changelog_scan.cc
     inspect/history_table.cc
+    inspect/metadata_scan.cc
     inspect/metadata_table.cc
     inspect/snapshots_table.cc
     inheritable_metadata.cc

src/iceberg/cpp20_compatibility_internal.h

@@ -79,6 +79,7 @@
 #include "iceberg/incremental_changelog_scan.h"
 #include "iceberg/inheritable_metadata.h"
 #include "iceberg/inspect/history_table.h"
+#include "iceberg/inspect/metadata_scan.h"
 #include "iceberg/inspect/metadata_table.h"
 #include "iceberg/inspect/snapshots_table.h"
 #include "iceberg/location_provider.h"

src/iceberg/inspect/history_table.cc

@@ -50,7 +50,8 @@ Result<std::unique_ptr<HistoryTable>> HistoryTable::Make(std::shared_ptr<Table>
   return std::unique_ptr<HistoryTable>(new HistoryTable(std::move(table)));
 }
 
-Result<ArrowArrayStream> HistoryTable::Scan() {
+Result<ArrowArrayStream> HistoryTable::ScanProjection(
+    const Schema& /*projected_schema*/) const {
   return NotSupported("Scan is not supported for the history table");
 }
 

src/iceberg/inspect/history_table.h

@@ -42,7 +42,8 @@ class ICEBERG_EXPORT HistoryTable : public MetadataTable {
 
   const std::shared_ptr<Schema>& schema() const override;
 
-  Result<ArrowArrayStream> Scan() override;
+ protected:
+  Result<ArrowArrayStream> ScanProjection(const Schema& projected_schema) const override;
 
  private:
   explicit HistoryTable(std::shared_ptr<Table> table);

src/iceberg/inspect/metadata_scan.cc

@@ -0,0 +1,88 @@
+/*
+ * 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_scan.h"
+
+#include <memory>
+#include <utility>
+
+#include "iceberg/inspect/metadata_table.h"
+#include "iceberg/schema.h"
+#include "iceberg/util/macros.h"
+
+namespace iceberg {
+
+MetadataScanBuilder::MetadataScanBuilder(const MetadataTable& table)
+    : table_(&table), projected_schema_(table.schema()) {}
+
+MetadataScanBuilder::~MetadataScanBuilder() = default;
+
+MetadataScanBuilder& MetadataScanBuilder::Select(
+    const std::vector<std::string>& columns) {
+  auto projected = table_->schema()->Select(columns, case_sensitive_);
+  if (projected) {
+    projected_schema_ = std::move(projected).value();
+  } else {
+    AddError(projected.error());
+  }
+  return *this;
+}
+
+MetadataScanBuilder& MetadataScanBuilder::Project(std::shared_ptr<Schema> schema) {
+  projected_schema_ = std::move(schema);
+  return *this;
+}
+
+MetadataScanBuilder& MetadataScanBuilder::CaseSensitive(bool value) {
+  case_sensitive_ = value;
+  return *this;
+}
+
+MetadataScanBuilder& MetadataScanBuilder::UseSnapshot(int64_t snapshot_id) {
+  ICEBERG_BUILDER_CHECK(
+      std::holds_alternative<std::monostate>(snapshot_selection_.snapshot),
+      "Cannot override snapshot, a snapshot or timestamp is already set");
+  snapshot_selection_.snapshot = snapshot_id;
+  return *this;
+}
+
+MetadataScanBuilder& MetadataScanBuilder::UseRef(const std::string& ref) {
+  ICEBERG_BUILDER_CHECK(snapshot_selection_.ref_name.empty(),
+                        "Cannot override ref, already set ref={}",
+                        snapshot_selection_.ref_name);
+  ICEBERG_BUILDER_CHECK(!ref.empty(), "Ref name cannot be empty");
+  snapshot_selection_.ref_name = ref;
+  return *this;
+}
+
+MetadataScanBuilder& MetadataScanBuilder::AsOfTime(int64_t timestamp_millis) {
+  ICEBERG_BUILDER_CHECK(
+      std::holds_alternative<std::monostate>(snapshot_selection_.snapshot),
+      "Cannot override snapshot, a snapshot or timestamp is already set");
+  snapshot_selection_.snapshot = TimePointMsFromUnixMs(timestamp_millis);
+  return *this;
+}
+
+Result<ArrowArrayStream> MetadataScanBuilder::Build() {
+  ICEBERG_RETURN_UNEXPECTED(CheckErrors());
+  ICEBERG_PRECHECK(projected_schema_ != nullptr, "Projected schema cannot be null");
+  return table_->ScanSnapshot(snapshot_selection_, *projected_schema_);
+}
+
+}  // namespace iceberg

src/iceberg/inspect/metadata_scan.h

@@ -0,0 +1,136 @@
+/*
+ * 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/metadata_scan.h
+/// \brief Scan configuration for Iceberg metadata tables.
+
+#include <cstdint>
+#include <memory>
+#include <string>
+#include <variant>
+#include <vector>
+
+#include "iceberg/arrow_c_data.h"
+#include "iceberg/iceberg_export.h"
+#include "iceberg/result.h"
+#include "iceberg/type_fwd.h"
+#include "iceberg/util/error_collector.h"
+#include "iceberg/util/timepoint.h"
+
+namespace iceberg {
+
+/// \brief Snapshot selection parameters for a time-travel scan.
+struct SnapshotSelection {
+  /// \brief Select the current snapshot, a snapshot ID, or an as-of timestamp.
+  ///
+  /// std::monostate selects the current snapshot.
+  std::variant<std::monostate, int64_t, TimePointMs> snapshot;
+
+  /// \brief Resolve the snapshot relative to this branch or tag.
+  ///
+  /// An empty string uses the main branch.
+  std::string ref_name;
+};
+
+/// \brief Configure a scan over a metadata table.
+///
+/// Builders are created by MetadataTable::NewScan(), and the metadata table must
+/// outlive the builder.
+///
+/// \code
+///   ICEBERG_ASSIGN_OR_RAISE(auto builder, snapshots_table->NewScan());
+///   ICEBERG_ASSIGN_OR_RAISE(auto stream, builder->Select({"snapshot_id"}).Build());
+/// \endcode
+class ICEBERG_EXPORT MetadataScanBuilder : public ErrorCollector {
+ public:
+  ~MetadataScanBuilder() override;
+
+  /// \brief Select columns by name; nested columns use dot-separated paths.
+  ///
+  /// Columns are returned in schema order, and names that match no column are
+  /// ignored. An empty list selects no columns; `*` selects all columns. Names are
+  /// resolved immediately, using the current CaseSensitive() setting. Selecting a
+  /// map's value returns the whole map; selecting only its key is an error.
+  ///
+  /// \param columns Column names from the metadata table's schema.
+  MetadataScanBuilder& Select(const std::vector<std::string>& columns);
+
+  /// \brief Set the output schema to a projection of the metadata table schema.
+  ///
+  /// Each field must select a whole column of the metadata table by field id, with
+  /// the same type and without tightening its nullability. Fields may be reordered.
+  ///
+  /// \param schema The output schema.
+  MetadataScanBuilder& Project(std::shared_ptr<Schema> schema);
+
+  /// \brief Use case-sensitive column names in Select(); enabled by default.
+  ///
+  /// \param value Whether column names are matched case-sensitively.
+  MetadataScanBuilder& CaseSensitive(bool value);
+
+  /// \brief Request this scan to use the given snapshot by ID.
+  ///
+  /// Only metadata tables that support time travel accept a snapshot; Build()
+  /// returns NotSupported otherwise. The table resolves the snapshot when the scan
+  /// is built.
+  ///
+  /// \param snapshot_id A snapshot ID.
+  MetadataScanBuilder& UseSnapshot(int64_t snapshot_id);
+
+  /// \brief Request this scan to resolve the snapshot on the given branch or tag.
+  ///
+  /// Only metadata tables that support time travel accept a reference; Build()
+  /// returns NotSupported otherwise. The table resolves the reference when the scan
+  /// is built.
+  ///
+  /// \param ref A branch or tag name.
+  MetadataScanBuilder& UseRef(const std::string& ref);
+
+  /// \brief Request this scan to use the most recent snapshot as of the given time.
+  ///
+  /// The snapshot is resolved on the branch set by UseRef(), or main if no branch
+  /// is set. Only metadata tables that support time travel accept a timestamp;
+  /// Build() returns NotSupported otherwise.
+  ///
+  /// \param timestamp_millis A timestamp in milliseconds since the Unix epoch.
+  MetadataScanBuilder& AsOfTime(int64_t timestamp_millis);
+
+  /// \brief Validate the settings and scan the metadata table.
+  ///
+  /// Each call returns a new stream with the output schema. The caller owns the
+  /// returned stream and must invoke its `release` callback when the stream is no
+  /// longer needed.
+  ///
+  /// \return An Arrow stream containing the selected columns, or an error.
+  Result<ArrowArrayStream> Build();
+
+ private:
+  friend class MetadataTable;
+
+  explicit MetadataScanBuilder(const MetadataTable& table);
+
+  const MetadataTable* table_;
+  std::shared_ptr<Schema> projected_schema_;
+  bool case_sensitive_ = true;
+  SnapshotSelection snapshot_selection_;
+};
+
+}  // namespace iceberg

src/iceberg/inspect/metadata_table.cc

@@ -21,6 +21,9 @@
 
 #include <memory>
 #include <utility>
+#include <variant>
+
+#include "iceberg/inspect/metadata_scan.h"
 
 namespace iceberg {
 
@@ -35,20 +38,29 @@ const std::shared_ptr<Table>& MetadataTable::source_table() const {
   return source_table_;
 }
 
+Result<std::unique_ptr<MetadataScanBuilder>> MetadataTable::NewScan() const {
+  return std::unique_ptr<MetadataScanBuilder>(new MetadataScanBuilder(*this));
+}
+
+Result<ArrowArrayStream> MetadataTable::ScanSnapshot(
+    const SnapshotSelection& snapshot_selection, const Schema& projected_schema) const {
+  if (!std::holds_alternative<std::monostate>(snapshot_selection.snapshot) ||
+      !snapshot_selection.ref_name.empty()) {
+    return NotSupported("Metadata table does not support time travel");
+  }
+  return ScanProjection(projected_schema);
+}
+
 TimeTravelMetadataTable::TimeTravelMetadataTable(std::shared_ptr<Table> source_table)
     : MetadataTable(std::move(source_table)) {}
 
 TimeTravelMetadataTable::~TimeTravelMetadataTable() = default;
 
 bool TimeTravelMetadataTable::supports_time_travel() const noexcept { return true; }
 
-Result<ArrowArrayStream> TimeTravelMetadataTable::Scan() {
-  return ScanSnapshot(SnapshotSelection{});
-}
-
-Result<ArrowArrayStream> TimeTravelMetadataTable::Scan(
-    const SnapshotSelection& snapshot_selection) {
-  return ScanSnapshot(snapshot_selection);
+Result<ArrowArrayStream> TimeTravelMetadataTable::ScanProjection(
+    const Schema& projected_schema) const {
+  return ScanSnapshot(SnapshotSelection{}, projected_schema);
 }
 
 }  // namespace iceberg