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

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*/) {
   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) override;
 
  private:
   explicit HistoryTable(std::shared_ptr<Table> table);

src/iceberg/inspect/metadata_table.cc

@@ -22,7 +22,23 @@
 #include <memory>
 #include <utility>
 
+#include "iceberg/schema.h"
+#include "iceberg/util/macros.h"
+
 namespace iceberg {
+namespace {
+
+Result<std::shared_ptr<Schema>> SelectColumns(const std::shared_ptr<Schema>& schema,
+                                              const ColumnSelection& columns) {
+  if (!columns.names.has_value()) {
+    return schema;
+  }
+  ICEBERG_ASSIGN_OR_RAISE(auto projected_schema,
+                          schema->Select(*columns.names, columns.case_sensitive));
+  return std::shared_ptr<Schema>(std::move(projected_schema));
+}
+
+}  // namespace
 
 MetadataTable::MetadataTable(std::shared_ptr<Table> source_table)
     : source_table_(std::move(source_table)) {}
@@ -35,20 +51,34 @@ const std::shared_ptr<Table>& MetadataTable::source_table() const {
   return source_table_;
 }
 
+Result<ArrowArrayStream> MetadataTable::Scan() { return ScanProjection(*schema()); }
+
+Result<ArrowArrayStream> MetadataTable::ScanColumns(const ColumnSelection& columns) {
+  ICEBERG_ASSIGN_OR_RAISE(auto projected_schema, SelectColumns(schema(), columns));
+  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);
+  return ScanSnapshot(snapshot_selection, *schema());
+}
+
+Result<ArrowArrayStream> TimeTravelMetadataTable::ScanColumns(
+    const SnapshotSelection& snapshot_selection, const ColumnSelection& columns) {
+  ICEBERG_ASSIGN_OR_RAISE(auto projected_schema, SelectColumns(schema(), columns));
+  return ScanSnapshot(snapshot_selection, *projected_schema);
+}
+
+Result<ArrowArrayStream> TimeTravelMetadataTable::ScanProjection(
+    const Schema& projected_schema) {
+  return ScanSnapshot(SnapshotSelection{}, projected_schema);
 }
 
 }  // namespace iceberg

src/iceberg/inspect/metadata_table.h

@@ -24,9 +24,11 @@
 
 #include <concepts>
 #include <memory>
+#include <optional>
 #include <string>
 #include <utility>
 #include <variant>
+#include <vector>
 
 #include "iceberg/arrow_c_data.h"
 #include "iceberg/iceberg_export.h"
@@ -36,6 +38,20 @@
 
 namespace iceberg {
 
+/// \brief Column selection for a metadata table scan.
+struct ColumnSelection {
+  /// \brief Names of the columns to return.
+  ///
+  /// Nested names are dot-concatenated. Columns are returned in schema order, and
+  /// names that match no column are ignored. std::nullopt or a list containing `*`
+  /// returns all columns. An empty list returns rows without columns. Selecting a
+  /// map's value returns the whole map; selecting only its key is an error.
+  std::optional<std::vector<std::string>> names;
+
+  /// \brief Whether names are matched case-sensitively.
+  bool case_sensitive = true;
+};
+
 /// \brief Base interface for an Iceberg metadata table.
 class ICEBERG_EXPORT MetadataTable {
  public:
@@ -73,15 +89,33 @@ class ICEBERG_EXPORT MetadataTable {
   /// \brief Return whether this metadata table supports time travel.
   virtual bool supports_time_travel() const noexcept;
 
-  /// \brief Scan the metadata table without time travel.
+  /// \brief Scan all columns of the metadata table without time travel.
   ///
   /// The caller owns the returned stream and must invoke its `release` callback
   /// when the stream is no longer needed.
-  virtual Result<ArrowArrayStream> Scan() = 0;
+  Result<ArrowArrayStream> Scan();
+
+  /// \brief Scan the selected columns of the metadata table without time travel.
+  ///
+  /// The caller owns the returned stream and must invoke its `release` callback
+  /// when the stream is no longer needed.
+  ///
+  /// This is not a Scan() overload so that `Scan({})` on a TimeTravelMetadataTable
+  /// still resolves to Scan(const SnapshotSelection&).
+  ///
+  /// \param columns Names of the columns to return.
+  /// \return An Arrow stream containing the selected columns, or an error.
+  Result<ArrowArrayStream> ScanColumns(const ColumnSelection& columns);
 
  protected:
   explicit MetadataTable(std::shared_ptr<Table> source_table);
 
+  /// \brief Implement a scan that returns the columns of a projected schema.
+  ///
+  /// \param projected_schema Projection of schema() with the columns to return. It
+  /// is only valid during this call.
+  virtual Result<ArrowArrayStream> ScanProjection(const Schema& projected_schema) = 0;
+
  private:
   std::shared_ptr<Table> source_table_;
 };
@@ -110,21 +144,37 @@ class ICEBERG_EXPORT TimeTravelMetadataTable : public MetadataTable {
   /// \brief Return true because this interface supports time travel.
   bool supports_time_travel() const noexcept final;
 
-  /// \brief Scan using the current snapshot on the main branch.
-  Result<ArrowArrayStream> Scan() final;
+  // Scans without a snapshot selection read the current snapshot on the main branch.
+  using MetadataTable::Scan;
+  using MetadataTable::ScanColumns;
 
-  /// \brief Scan using the requested snapshot selection.
+  /// \brief Scan all columns using the requested snapshot selection.
   ///
   /// \param snapshot_selection Snapshot ID, timestamp, and optional ref selection.
   /// \return An Arrow stream containing the metadata table rows, or an error.
   Result<ArrowArrayStream> Scan(const SnapshotSelection& snapshot_selection);
 
+  /// \brief Scan the selected columns using the requested snapshot selection.
+  ///
+  /// \param snapshot_selection Snapshot ID, timestamp, and optional ref selection.
+  /// \param columns Names of the columns to return.
+  /// \return An Arrow stream containing the selected columns, or an error.
+  Result<ArrowArrayStream> ScanColumns(const SnapshotSelection& snapshot_selection,
+                                       const ColumnSelection& columns);
+
  protected:
   explicit TimeTravelMetadataTable(std::shared_ptr<Table> source_table);
 
+  /// \brief Scan the current snapshot on the main branch.
+  Result<ArrowArrayStream> ScanProjection(const Schema& projected_schema) final;
+
   /// \brief Implement a scan for the requested snapshot selection.
+  ///
+  /// \param snapshot_selection Snapshot ID, timestamp, and optional ref selection.
+  /// \param projected_schema Projection of schema() with the columns to return. It
+  /// is only valid during this call.
   virtual Result<ArrowArrayStream> ScanSnapshot(
-      const SnapshotSelection& snapshot_selection) = 0;
+      const SnapshotSelection& snapshot_selection, const Schema& projected_schema) = 0;
 };
 
 }  // namespace iceberg

src/iceberg/inspect/metadata_table_columns_internal.h

@@ -0,0 +1,94 @@
+/*
+ * 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_table_columns_internal.h
+/// \brief Build only the selected columns of metadata table rows.
+
+#include <algorithm>
+#include <cstddef>
+#include <cstdint>
+#include <span>
+#include <vector>
+
+#include "iceberg/arrow_c_data.h"
+#include "iceberg/arrow_row_builder_internal.h"
+#include "iceberg/result.h"
+#include "iceberg/schema.h"
+#include "iceberg/schema_field.h"
+#include "iceberg/type.h"
+#include "iceberg/util/macros.h"
+
+namespace iceberg::internal {
+
+/// \brief Append the value of one metadata table column for a row.
+template <typename Row>
+using AppendColumnFn = Status (*)(ArrowArray* column, const Row& row);
+
+/// \brief A metadata table column and the function that appends its values.
+template <typename Row>
+struct MetadataTableColumn {
+  int32_t field_id;
+  AppendColumnFn<Row> append;
+};
+
+/// \brief Return the append functions for the columns of a projected schema.
+///
+/// \param schema The metadata table schema.
+/// \param projected_schema Projection of `schema` with the columns to return.
+/// \param columns The columns of `schema` with their append functions.
+/// \return One append function per column of `projected_schema`, in the same order,
+/// or NotSupported if only some fields of a nested column are selected.
+template <typename Row>
+Result<std::vector<AppendColumnFn<Row>>> SelectAppendFns(
+    const Schema& schema, const Schema& projected_schema,
+    std::span<const MetadataTableColumn<Row>> columns) {
+  std::vector<AppendColumnFn<Row>> append_fns;
+  append_fns.reserve(projected_schema.fields().size());
+  for (const auto& projected_field : projected_schema.fields()) {
+    ICEBERG_ASSIGN_OR_RAISE(auto field, schema.FindFieldById(projected_field.field_id()));
+    ICEBERG_CHECK(field.has_value(), "Metadata table column '{}' is not in the schema",
+                  projected_field.name());
+    if (*field->get().type() != *projected_field.type()) {
+      return NotSupported("Cannot select nested fields of metadata table column '{}'",
+                          projected_field.name());
+    }
+    auto column = std::ranges::find(columns, projected_field.field_id(),
+                                    &MetadataTableColumn<Row>::field_id);
+    ICEBERG_CHECK(column != columns.end(),
+                  "Metadata table column '{}' has no append function",
+                  projected_field.name());
+    append_fns.push_back(column->append);
+  }
+  return append_fns;
+}
+
+/// \brief Append the selected columns of a row and finish the row.
+template <typename Row>
+Status AppendRow(ArrowRowBuilder& builder,
+                 std::span<const AppendColumnFn<Row>> append_fns, const Row& row) {
+  for (size_t index = 0; index < append_fns.size(); ++index) {
+    ICEBERG_RETURN_UNEXPECTED(
+        append_fns[index](builder.column(static_cast<int64_t>(index)), row));
+  }
+  return builder.FinishRow();
+}
+
+}  // namespace iceberg::internal