This is an automated email from the ASF dual-hosted git repository.
wgtmac pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/iceberg-cpp.git
The following commit(s) were added to refs/heads/main by this push:
new d0ca49fc feat(inspect): support selecting columns in metadata table
scans (#1008)
d0ca49fc is described below
commit d0ca49fc0c089d0ed4c16b4c4eadcc22faba983f
Author: Manu Zhang <[email protected]>
AuthorDate: Fri Oct 30 22:29:43 2026 +0800
feat(inspect): support selecting columns in metadata table scans (#1008)
---
src/iceberg/CMakeLists.txt | 1 +
src/iceberg/cpp20_compatibility_internal.h | 1 +
src/iceberg/inspect/history_table.cc | 3 +-
src/iceberg/inspect/history_table.h | 3 +-
src/iceberg/inspect/metadata_scan.cc | 88 +++++++++++
src/iceberg/inspect/metadata_scan.h | 136 ++++++++++++++++
src/iceberg/inspect/metadata_table.cc | 26 +++-
src/iceberg/inspect/metadata_table.h | 71 +++++----
.../inspect/metadata_table_columns_internal.h | 103 +++++++++++++
src/iceberg/inspect/snapshots_table.cc | 93 +++++++----
src/iceberg/inspect/snapshots_table.h | 3 +-
src/iceberg/test/metadata_table_test.cc | 171 +++++++++++++++++++++
src/iceberg/test/metadata_table_test_base.h | 17 +-
src/iceberg/test/snapshots_table_test.cc | 170 +++++++++++++++++++-
src/iceberg/type_fwd.h | 1 +
15 files changed, 804 insertions(+), 83 deletions(-)
diff --git a/src/iceberg/CMakeLists.txt b/src/iceberg/CMakeLists.txt
index 06e59988..cf465e5f 100644
--- a/src/iceberg/CMakeLists.txt
+++ b/src/iceberg/CMakeLists.txt
@@ -59,6 +59,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
diff --git a/src/iceberg/cpp20_compatibility_internal.h
b/src/iceberg/cpp20_compatibility_internal.h
index 12114d8f..d186e5bb 100644
--- a/src/iceberg/cpp20_compatibility_internal.h
+++ b/src/iceberg/cpp20_compatibility_internal.h
@@ -81,6 +81,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/labels.h"
diff --git a/src/iceberg/inspect/history_table.cc
b/src/iceberg/inspect/history_table.cc
index 03bbcca9..518073ca 100644
--- a/src/iceberg/inspect/history_table.cc
+++ b/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");
}
diff --git a/src/iceberg/inspect/history_table.h
b/src/iceberg/inspect/history_table.h
index 7d863fc2..6f07bcd0 100644
--- a/src/iceberg/inspect/history_table.h
+++ b/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);
diff --git a/src/iceberg/inspect/metadata_scan.cc
b/src/iceberg/inspect/metadata_scan.cc
new file mode 100644
index 00000000..6117c28c
--- /dev/null
+++ b/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
diff --git a/src/iceberg/inspect/metadata_scan.h
b/src/iceberg/inspect/metadata_scan.h
new file mode 100644
index 00000000..f487bce4
--- /dev/null
+++ b/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
diff --git a/src/iceberg/inspect/metadata_table.cc
b/src/iceberg/inspect/metadata_table.cc
index 7bc94c51..3e21688f 100644
--- a/src/iceberg/inspect/metadata_table.cc
+++ b/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,6 +38,19 @@ 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)) {}
@@ -42,13 +58,9 @@ 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
diff --git a/src/iceberg/inspect/metadata_table.h
b/src/iceberg/inspect/metadata_table.h
index 85000ee1..7717ce58 100644
--- a/src/iceberg/inspect/metadata_table.h
+++ b/src/iceberg/inspect/metadata_table.h
@@ -24,15 +24,13 @@
#include <concepts>
#include <memory>
-#include <string>
#include <utility>
-#include <variant>
#include "iceberg/arrow_c_data.h"
#include "iceberg/iceberg_export.h"
+#include "iceberg/inspect/metadata_scan.h"
#include "iceberg/result.h"
#include "iceberg/type_fwd.h"
-#include "iceberg/util/timepoint.h"
namespace iceberg {
@@ -64,7 +62,7 @@ class ICEBERG_EXPORT MetadataTable {
/// \brief Return this metadata table's kind.
virtual Kind kind() const noexcept = 0;
- /// \brief Return the schema of rows emitted by scans.
+ /// \brief Return the complete canonical schema of this metadata table.
virtual const std::shared_ptr<Schema>& schema() const = 0;
/// \brief Return the source table whose metadata is exposed.
@@ -73,36 +71,44 @@ 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 Create a builder for configuring a metadata table scan.
///
- /// The caller owns the returned stream and must invoke its `release`
callback
- /// when the stream is no longer needed.
- virtual Result<ArrowArrayStream> Scan() = 0;
+ /// A builder with no options scans all columns of the current snapshot on
the main
+ /// branch. This table must outlive the builder.
+ virtual Result<std::unique_ptr<MetadataScanBuilder>> NewScan() const;
protected:
explicit MetadataTable(std::shared_ptr<Table> source_table);
- private:
- std::shared_ptr<Table> source_table_;
-};
-
-/// \brief Snapshot selection parameters for a time-travel scan.
-struct SnapshotSelection {
- /// \brief Select the current snapshot, a snapshot ID, or an as-of timestamp.
+ /// \brief Implement a scan that returns the columns of a projected schema.
///
- /// std::monostate selects the current snapshot.
- std::variant<std::monostate, int64_t, TimePointMs> snapshot;
+ /// \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) const = 0;
- /// \brief Resolve the snapshot relative to this branch or tag.
+ /// \brief Implement a scan for the requested snapshot selection.
///
- /// An empty string uses the main branch.
- std::string ref_name;
+ /// The default implementation scans the current snapshot on the main branch
with
+ /// ScanProjection() and returns NotSupported for any other selection.
+ /// TimeTravelMetadataTable requires an implementation.
+ ///
+ /// \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, const Schema&
projected_schema) const;
+
+ private:
+ friend class MetadataScanBuilder;
+
+ std::shared_ptr<Table> source_table_;
};
/// \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.
+/// Scans of these tables accept MetadataScanBuilder::UseSnapshot(), UseRef(),
and
+/// AsOfTime(), which MetadataScanBuilder::Build() rejects for other tables.
class ICEBERG_EXPORT TimeTravelMetadataTable : public MetadataTable {
public:
~TimeTravelMetadataTable() override;
@@ -110,21 +116,20 @@ 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;
-
- /// \brief Scan 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);
-
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)
const final;
+
/// \brief Implement a scan for the requested snapshot selection.
- virtual Result<ArrowArrayStream> ScanSnapshot(
- const SnapshotSelection& snapshot_selection) = 0;
+ ///
+ /// \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.
+ Result<ArrowArrayStream> ScanSnapshot(const SnapshotSelection&
snapshot_selection,
+ const Schema& projected_schema) const
override =
+ 0;
};
} // namespace iceberg
diff --git a/src/iceberg/inspect/metadata_table_columns_internal.h
b/src/iceberg/inspect/metadata_table_columns_internal.h
new file mode 100644
index 00000000..596ffa2f
--- /dev/null
+++ b/src/iceberg/inspect/metadata_table_columns_internal.h
@@ -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.
+ */
+
+#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.
+/// Each projected field must select a whole column of `schema` by field id,
with the
+/// same type and without tightening its nullability.
+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()) {
+ // Only top-level columns have append functions, so nested field ids are
rejected.
+ ICEBERG_ASSIGN_OR_RAISE(auto field,
schema.GetFieldById(projected_field.field_id()));
+ if (!field.has_value()) {
+ return InvalidArgument("Unknown metadata column: {}",
projected_field.name());
+ }
+ if (*field->get().type() != *projected_field.type()) {
+ return NotSupported("Partial projection of metadata column is not
supported: {}",
+ projected_field.name());
+ }
+ if (field->get().optional() && !projected_field.optional()) {
+ return InvalidArgument(
+ "Nullable metadata column cannot be projected as required: {}",
+ projected_field.name());
+ }
+ auto column = std::ranges::find(columns, projected_field.field_id(),
+ &MetadataTableColumn<Row>::field_id);
+ if (column == columns.end()) {
+ return InvalidArgument("Metadata 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
diff --git a/src/iceberg/inspect/snapshots_table.cc
b/src/iceberg/inspect/snapshots_table.cc
index 9e9d7906..0dede12c 100644
--- a/src/iceberg/inspect/snapshots_table.cc
+++ b/src/iceberg/inspect/snapshots_table.cc
@@ -19,6 +19,7 @@
#include "iceberg/inspect/snapshots_table.h"
+#include <array>
#include <chrono>
#include <cstddef>
#include <memory>
@@ -30,6 +31,7 @@
#include "iceberg/arrow_c_data_util_internal.h"
#include "iceberg/arrow_row_builder_internal.h"
+#include "iceberg/inspect/metadata_table_columns_internal.h"
#include "iceberg/nanoarrow_status_internal.h"
#include "iceberg/schema.h"
#include "iceberg/schema_field.h"
@@ -43,51 +45,66 @@
namespace iceberg {
namespace {
-Status AppendSnapshot(ArrowRowBuilder& builder, const Snapshot& snapshot) {
- ICEBERG_RETURN_UNEXPECTED(
- AppendInt(builder.column(0),
std::chrono::duration_cast<std::chrono::microseconds>(
-
snapshot.timestamp_ms.time_since_epoch())
- .count()));
- ICEBERG_RETURN_UNEXPECTED(AppendInt(builder.column(1),
snapshot.snapshot_id));
+Status AppendCommittedAt(ArrowArray* column, const Snapshot& snapshot) {
+ return AppendInt(column,
std::chrono::duration_cast<std::chrono::microseconds>(
+ snapshot.timestamp_ms.time_since_epoch())
+ .count());
+}
+
+Status AppendSnapshotId(ArrowArray* column, const Snapshot& snapshot) {
+ return AppendInt(column, snapshot.snapshot_id);
+}
+Status AppendParentId(ArrowArray* column, const Snapshot& snapshot) {
if (snapshot.parent_snapshot_id.has_value()) {
- ICEBERG_RETURN_UNEXPECTED(AppendInt(builder.column(2),
*snapshot.parent_snapshot_id));
- } else {
- ICEBERG_RETURN_UNEXPECTED(AppendNull(builder.column(2)));
+ return AppendInt(column, *snapshot.parent_snapshot_id);
}
+ return AppendNull(column);
+}
+Status AppendOperation(ArrowArray* column, const Snapshot& snapshot) {
auto operation = snapshot.Operation();
if (operation.has_value()) {
- ICEBERG_RETURN_UNEXPECTED(AppendString(builder.column(3), *operation));
- } else {
- ICEBERG_RETURN_UNEXPECTED(AppendNull(builder.column(3)));
+ return AppendString(column, *operation);
}
+ return AppendNull(column);
+}
- ICEBERG_RETURN_UNEXPECTED(AppendString(builder.column(4),
snapshot.manifest_list));
+Status AppendManifestList(ArrowArray* column, const Snapshot& snapshot) {
+ return AppendString(column, snapshot.manifest_list);
+}
- if (!snapshot.summary.empty()) {
- if (snapshot.summary.contains(SnapshotSummaryFields::kOperation)) {
- auto summary = snapshot.summary;
- summary.erase(SnapshotSummaryFields::kOperation);
- ICEBERG_RETURN_UNEXPECTED(AppendStringMap(builder.column(5), summary));
- } else {
- ICEBERG_RETURN_UNEXPECTED(AppendStringMap(builder.column(5),
snapshot.summary));
- }
- } else {
- ICEBERG_RETURN_UNEXPECTED(AppendNull(builder.column(5)));
+Status AppendSummary(ArrowArray* column, const Snapshot& snapshot) {
+ if (snapshot.summary.empty()) {
+ return AppendNull(column);
}
-
- return builder.FinishRow();
+ if (snapshot.summary.contains(SnapshotSummaryFields::kOperation)) {
+ auto summary = snapshot.summary;
+ summary.erase(SnapshotSummaryFields::kOperation);
+ return AppendStringMap(column, summary);
+ }
+ return AppendStringMap(column, snapshot.summary);
}
+constexpr std::array<internal::MetadataTableColumn<Snapshot>, 6>
kSnapshotColumns{{
+ {.field_id = 1, .append = AppendCommittedAt},
+ {.field_id = 2, .append = AppendSnapshotId},
+ {.field_id = 3, .append = AppendParentId},
+ {.field_id = 4, .append = AppendOperation},
+ {.field_id = 5, .append = AppendManifestList},
+ {.field_id = 6, .append = AppendSummary},
+}};
+
class SnapshotsTableStream {
public:
static Result<std::unique_ptr<SnapshotsTableStream>> Make(
- std::shared_ptr<const TableMetadata> metadata, const iceberg::Schema&
schema) {
+ std::shared_ptr<const TableMetadata> metadata,
+ const iceberg::Schema& projected_schema,
+ std::vector<internal::AppendColumnFn<Snapshot>> append_fns) {
ArrowSchema arrow_schema{};
- ICEBERG_RETURN_UNEXPECTED(ToArrowSchema(schema, &arrow_schema));
- return std::unique_ptr<SnapshotsTableStream>(
- new SnapshotsTableStream(std::move(metadata),
std::move(arrow_schema)));
+ ICEBERG_RETURN_UNEXPECTED(ToArrowSchema(projected_schema, &arrow_schema));
+ return std::unique_ptr<SnapshotsTableStream>(new SnapshotsTableStream(
+ std::move(metadata), std::move(append_fns), std::move(arrow_schema)));
}
~SnapshotsTableStream() = default;
@@ -116,7 +133,8 @@ class SnapshotsTableStream {
if (snapshot == nullptr) [[unlikely]] {
continue;
}
- ICEBERG_RETURN_UNEXPECTED(AppendSnapshot(builder, *snapshot));
+ ICEBERG_RETURN_UNEXPECTED(
+ internal::AppendRow<Snapshot>(builder, append_fns_, *snapshot));
}
if (builder.num_rows() == 0) {
return std::nullopt;
@@ -138,10 +156,14 @@ class SnapshotsTableStream {
private:
SnapshotsTableStream(std::shared_ptr<const TableMetadata> metadata,
+ std::vector<internal::AppendColumnFn<Snapshot>>
append_fns,
ArrowSchema arrow_schema)
- : metadata_(std::move(metadata)), arrow_schema_(std::move(arrow_schema))
{}
+ : metadata_(std::move(metadata)),
+ append_fns_(std::move(append_fns)),
+ arrow_schema_(std::move(arrow_schema)) {}
std::shared_ptr<const TableMetadata> metadata_;
+ std::vector<internal::AppendColumnFn<Snapshot>> append_fns_;
ArrowSchema arrow_schema_{};
size_t next_snapshot_ = 0;
};
@@ -173,9 +195,14 @@ Result<std::unique_ptr<SnapshotsTable>>
SnapshotsTable::Make(
return std::unique_ptr<SnapshotsTable>(new SnapshotsTable(std::move(table)));
}
-Result<ArrowArrayStream> SnapshotsTable::Scan() {
+Result<ArrowArrayStream> SnapshotsTable::ScanProjection(
+ const Schema& projected_schema) const {
+ ICEBERG_ASSIGN_OR_RAISE(
+ auto append_fns,
+ internal::SelectAppendFns<Snapshot>(*schema(), projected_schema,
kSnapshotColumns));
ICEBERG_ASSIGN_OR_RAISE(
- auto stream, SnapshotsTableStream::Make(source_table()->metadata(),
*schema()));
+ auto stream, SnapshotsTableStream::Make(source_table()->metadata(),
+ projected_schema,
std::move(append_fns)));
return MakeArrowArrayStream(std::move(stream));
}
diff --git a/src/iceberg/inspect/snapshots_table.h
b/src/iceberg/inspect/snapshots_table.h
index d2f0ddf9..ae1434a5 100644
--- a/src/iceberg/inspect/snapshots_table.h
+++ b/src/iceberg/inspect/snapshots_table.h
@@ -42,10 +42,11 @@ class ICEBERG_EXPORT SnapshotsTable : public MetadataTable {
const std::shared_ptr<Schema>& schema() const override;
+ protected:
/// \brief Scan all snapshots as rows.
///
/// The snapshots table always returns every known snapshot.
- Result<ArrowArrayStream> Scan() override;
+ Result<ArrowArrayStream> ScanProjection(const Schema& projected_schema)
const override;
private:
explicit SnapshotsTable(std::shared_ptr<Table> table);
diff --git a/src/iceberg/test/metadata_table_test.cc
b/src/iceberg/test/metadata_table_test.cc
index b014ef96..187646c9 100644
--- a/src/iceberg/test/metadata_table_test.cc
+++ b/src/iceberg/test/metadata_table_test.cc
@@ -19,6 +19,16 @@
#include "iceberg/inspect/metadata_table.h"
+#include <cstdint>
+#include <memory>
+#include <string>
+#include <utility>
+#include <variant>
+#include <vector>
+
+#include <arrow/c/bridge.h>
+#include <arrow/record_batch.h>
+#include <arrow/type.h>
#include <gmock/gmock.h>
#include <gtest/gtest.h>
@@ -34,8 +44,46 @@
#include "iceberg/test/mock_catalog.h"
#include "iceberg/test/mock_io.h"
#include "iceberg/type.h"
+#include "iceberg/util/timepoint.h"
namespace iceberg {
+namespace {
+
+/// \brief Time-travel table that records the arguments of each scan.
+class RecordingTimeTravelTable : public TimeTravelMetadataTable {
+ public:
+ explicit RecordingTimeTravelTable(std::shared_ptr<Table> table)
+ : TimeTravelMetadataTable(std::move(table)) {}
+
+ Kind kind() const noexcept override { return Kind::kSnapshots; }
+
+ const std::shared_ptr<Schema>& schema() const override {
+ static const auto schema = std::make_shared<Schema>(
+ std::vector<SchemaField>{SchemaField::MakeRequired(1, "id", int64()),
+ SchemaField::MakeOptional(2, "name",
string())});
+ return schema;
+ }
+
+ const SnapshotSelection& scanned_snapshot() const { return
scanned_snapshot_; }
+ const std::vector<std::string>& scanned_columns() const { return
scanned_columns_; }
+
+ protected:
+ Result<ArrowArrayStream> ScanSnapshot(const SnapshotSelection&
snapshot_selection,
+ const Schema& projected_schema) const
override {
+ scanned_snapshot_ = snapshot_selection;
+ scanned_columns_.clear();
+ for (const auto& field : projected_schema.fields()) {
+ scanned_columns_.emplace_back(field.name());
+ }
+ return NotSupported("Recording table does not return rows");
+ }
+
+ private:
+ mutable SnapshotSelection scanned_snapshot_;
+ mutable std::vector<std::string> scanned_columns_;
+};
+
+} // namespace
class MetadataTableTest : public ::testing::Test {
protected:
@@ -56,6 +104,18 @@ class MetadataTableTest : public ::testing::Test {
std::make_shared<MockCatalog>()));
}
+ /// \brief Import the schema of a stream and release the stream.
+ static Result<std::shared_ptr<::arrow::Schema>>
ReadSchema(ArrowArrayStream&& stream) {
+ auto reader_result = ::arrow::ImportRecordBatchReader(&stream);
+ if (!reader_result.ok()) {
+ if (stream.release != nullptr) {
+ stream.release(&stream);
+ }
+ return InvalidArrowData(reader_result.status().ToString());
+ }
+ return reader_result.ValueUnsafe()->schema();
+ }
+
std::shared_ptr<Table> table_;
};
@@ -65,6 +125,51 @@ TEST_F(MetadataTableTest, FactoryRejectsNullSourceTable) {
EXPECT_THAT(result, HasErrorMessage("Table cannot be null"));
}
+TEST_F(MetadataTableTest, ScanBuilderSelectResolvesNamesImmediately) {
+ ICEBERG_UNWRAP_OR_FAIL(auto snapshots_table,
+ MetadataTable::Make<SnapshotsTable>(table_));
+
+ // Select() uses the case sensitivity in effect when it is called.
+ ICEBERG_UNWRAP_OR_FAIL(auto builder, snapshots_table->NewScan());
+ ICEBERG_UNWRAP_OR_FAIL(auto stream,
+
builder->Select({"SNAPSHOT_ID"}).CaseSensitive(false).Build());
+ ICEBERG_UNWRAP_OR_FAIL(auto schema, ReadSchema(std::move(stream)));
+ EXPECT_EQ(schema->num_fields(), 0);
+
+ ICEBERG_UNWRAP_OR_FAIL(auto insensitive_builder, snapshots_table->NewScan());
+ ICEBERG_UNWRAP_OR_FAIL(
+ auto insensitive_stream,
+
insensitive_builder->CaseSensitive(false).Select({"SNAPSHOT_ID"}).Build());
+ ICEBERG_UNWRAP_OR_FAIL(auto insensitive_schema,
+ ReadSchema(std::move(insensitive_stream)));
+ EXPECT_THAT(insensitive_schema->field_names(),
::testing::ElementsAre("snapshot_id"));
+
+ // An invalid selection is reported when the scan is built.
+ ICEBERG_UNWRAP_OR_FAIL(auto invalid_builder, snapshots_table->NewScan());
+ auto invalid = invalid_builder->Select({"summary.key"}).Build();
+ EXPECT_THAT(invalid, IsError(ErrorKind::kValidationFailed));
+ EXPECT_THAT(invalid, HasErrorMessage("Cannot project Map without value
field"));
+}
+
+TEST_F(MetadataTableTest, ScanBuilderProject) {
+ ICEBERG_UNWRAP_OR_FAIL(auto snapshots_table,
+ MetadataTable::Make<SnapshotsTable>(table_));
+ auto projection = std::make_shared<Schema>(
+ std::vector<SchemaField>{SchemaField::MakeOptional(4, "operation",
string()),
+ SchemaField::MakeRequired(2, "snapshot_id",
int64())});
+
+ // Project() replaces an earlier selection.
+ ICEBERG_UNWRAP_OR_FAIL(auto builder, snapshots_table->NewScan());
+ ICEBERG_UNWRAP_OR_FAIL(auto stream,
+
builder->Select({"snapshot_id"}).Project(projection).Build());
+ ICEBERG_UNWRAP_OR_FAIL(auto schema, ReadSchema(std::move(stream)));
+ EXPECT_THAT(schema->field_names(), ::testing::ElementsAre("operation",
"snapshot_id"));
+
+ ICEBERG_UNWRAP_OR_FAIL(auto null_builder, snapshots_table->NewScan());
+ EXPECT_THAT(null_builder->Project(nullptr).Build(),
+ IsError(ErrorKind::kInvalidArgument));
+}
+
TEST_F(MetadataTableTest, SupportsTimeTravel) {
ICEBERG_UNWRAP_OR_FAIL(auto snapshots_table,
MetadataTable::Make<SnapshotsTable>(table_));
@@ -72,6 +177,72 @@ TEST_F(MetadataTableTest, SupportsTimeTravel) {
ICEBERG_UNWRAP_OR_FAIL(auto history_table,
MetadataTable::Make<HistoryTable>(table_));
EXPECT_FALSE(history_table->supports_time_travel());
+
+ RecordingTimeTravelTable time_travel_table(table_);
+ EXPECT_TRUE(time_travel_table.supports_time_travel());
+}
+
+TEST_F(MetadataTableTest, TimeTravelScanDefaults) {
+ RecordingTimeTravelTable table(table_);
+ ICEBERG_UNWRAP_OR_FAIL(auto builder, table.NewScan());
+ EXPECT_THAT(builder->Build(), IsError(ErrorKind::kNotSupported));
+
EXPECT_TRUE(std::holds_alternative<std::monostate>(table.scanned_snapshot().snapshot));
+ EXPECT_TRUE(table.scanned_snapshot().ref_name.empty());
+ EXPECT_THAT(table.scanned_columns(), ::testing::ElementsAre("id", "name"));
+}
+
+TEST_F(MetadataTableTest, TimeTravelScanSelectsColumnsAndSnapshot) {
+ RecordingTimeTravelTable table(table_);
+ ICEBERG_UNWRAP_OR_FAIL(auto builder, table.NewScan());
+ EXPECT_THAT(builder->Select({"name"}).UseSnapshot(1).UseRef("dev").Build(),
+ IsError(ErrorKind::kNotSupported));
+
ASSERT_TRUE(std::holds_alternative<int64_t>(table.scanned_snapshot().snapshot));
+ EXPECT_EQ(std::get<int64_t>(table.scanned_snapshot().snapshot), 1);
+ EXPECT_EQ(table.scanned_snapshot().ref_name, "dev");
+ EXPECT_THAT(table.scanned_columns(), ::testing::ElementsAre("name"));
+}
+
+TEST_F(MetadataTableTest, TimeTravelScanAsOfTime) {
+ RecordingTimeTravelTable table(table_);
+ ICEBERG_UNWRAP_OR_FAIL(auto builder, table.NewScan());
+ EXPECT_THAT(builder->AsOfTime(1234567890000).Build(),
+ IsError(ErrorKind::kNotSupported));
+
ASSERT_TRUE(std::holds_alternative<TimePointMs>(table.scanned_snapshot().snapshot));
+ EXPECT_EQ(std::get<TimePointMs>(table.scanned_snapshot().snapshot),
+ TimePointMsFromUnixMs(1234567890000));
+ EXPECT_TRUE(table.scanned_snapshot().ref_name.empty());
+ EXPECT_THAT(table.scanned_columns(), ::testing::ElementsAre("id", "name"));
+}
+
+TEST_F(MetadataTableTest, ScanBuilderRejectsConflictingSnapshotOptions) {
+ RecordingTimeTravelTable table(table_);
+
+ ICEBERG_UNWRAP_OR_FAIL(auto snapshot_builder, table.NewScan());
+ auto snapshot_result = snapshot_builder->UseSnapshot(1).AsOfTime(2).Build();
+ EXPECT_THAT(snapshot_result, IsError(ErrorKind::kValidationFailed));
+ EXPECT_THAT(snapshot_result, HasErrorMessage("Cannot override snapshot"));
+
+ ICEBERG_UNWRAP_OR_FAIL(auto ref_builder, table.NewScan());
+ auto ref_result = ref_builder->UseRef("dev").UseRef("main").Build();
+ EXPECT_THAT(ref_result, IsError(ErrorKind::kValidationFailed));
+ EXPECT_THAT(ref_result, HasErrorMessage("Cannot override ref"));
+
+ ICEBERG_UNWRAP_OR_FAIL(auto empty_ref_builder, table.NewScan());
+ EXPECT_THAT(empty_ref_builder->UseRef("").Build(),
+ IsError(ErrorKind::kValidationFailed));
+}
+
+TEST_F(MetadataTableTest, HistoryTableScanIsNotSupported) {
+ ICEBERG_UNWRAP_OR_FAIL(auto history_table,
MetadataTable::Make<HistoryTable>(table_));
+ ICEBERG_UNWRAP_OR_FAIL(auto builder, history_table->NewScan());
+ EXPECT_THAT(builder->Select({"snapshot_id"}).Build(),
+ IsError(ErrorKind::kNotSupported));
+
+ // Tables without time travel reject snapshot selections.
+ ICEBERG_UNWRAP_OR_FAIL(auto snapshot_builder, history_table->NewScan());
+ auto snapshot_result = snapshot_builder->UseSnapshot(1).Build();
+ EXPECT_THAT(snapshot_result, IsError(ErrorKind::kNotSupported));
+ EXPECT_THAT(snapshot_result, HasErrorMessage("does not support time
travel"));
}
} // namespace iceberg
diff --git a/src/iceberg/test/metadata_table_test_base.h
b/src/iceberg/test/metadata_table_test_base.h
index b9f601cc..cefdb5cb 100644
--- a/src/iceberg/test/metadata_table_test_base.h
+++ b/src/iceberg/test/metadata_table_test_base.h
@@ -39,6 +39,7 @@
#include <gtest/gtest.h>
#include "iceberg/constants.h"
+#include "iceberg/inspect/metadata_table.h"
#include "iceberg/schema.h"
#include "iceberg/schema_field.h"
#include "iceberg/schema_internal.h"
@@ -80,7 +81,21 @@ class MetadataTableTestBase : public ::testing::Test {
"s3://bucket/meta.json", io_,
catalog_));
}
- /// \brief Import and consume a Scan()-produced ArrowArrayStream.
+ /// \brief Scan all columns of a metadata table.
+ static Result<ArrowArrayStream> Scan(MetadataTable& table) {
+ ICEBERG_ASSIGN_OR_RAISE(auto builder, table.NewScan());
+ return builder->Build();
+ }
+
+ /// \brief Scan the selected columns of a metadata table.
+ static Result<ArrowArrayStream> ScanColumns(MetadataTable& table,
+ const std::vector<std::string>&
columns,
+ bool case_sensitive = true) {
+ ICEBERG_ASSIGN_OR_RAISE(auto builder, table.NewScan());
+ return builder->CaseSensitive(case_sensitive).Select(columns).Build();
+ }
+
+ /// \brief Import and consume a scan-produced ArrowArrayStream.
static Result<std::vector<std::shared_ptr<::arrow::RecordBatch>>>
ReadAllBatches(
ArrowArrayStream&& stream) {
auto reader_result = ::arrow::ImportRecordBatchReader(&stream);
diff --git a/src/iceberg/test/snapshots_table_test.cc
b/src/iceberg/test/snapshots_table_test.cc
index cddfce69..9a62c6b6 100644
--- a/src/iceberg/test/snapshots_table_test.cc
+++ b/src/iceberg/test/snapshots_table_test.cc
@@ -77,7 +77,7 @@ TEST_F(SnapshotsTableTest, Construct) {
TEST_F(SnapshotsTableTest, Scan) {
// Scan the snapshots table once and verify all columns of the result.
- ICEBERG_UNWRAP_OR_FAIL(auto stream, snapshots_table_->Scan());
+ ICEBERG_UNWRAP_OR_FAIL(auto stream, Scan(*snapshots_table_));
ICEBERG_UNWRAP_OR_FAIL(auto batches, ReadAllBatches(std::move(stream)));
ASSERT_EQ(batches.size(), 1);
const auto& batch = batches.front();
@@ -132,6 +132,164 @@ TEST_F(SnapshotsTableTest, Scan) {
EXPECT_THAT(second_summary,
::testing::Contains(::testing::Pair("total-records", "2")));
}
+TEST_F(SnapshotsTableTest, ScanSelectedColumns) {
+ ICEBERG_UNWRAP_OR_FAIL(auto stream,
+ ScanColumns(*snapshots_table_, {"operation",
"snapshot_id"}));
+ ICEBERG_UNWRAP_OR_FAIL(auto batches, ReadAllBatches(std::move(stream)));
+ ASSERT_EQ(batches.size(), 1);
+ const auto& batch = batches.front();
+ EXPECT_EQ(batch->num_rows(), 2);
+ ASSERT_EQ(batch->num_columns(), 2);
+ EXPECT_EQ(batch->column_name(0), "snapshot_id");
+ EXPECT_EQ(batch->column_name(1), "operation");
+ ASSERT_EQ(batch->column(0)->type_id(), ::arrow::Type::INT64);
+ ASSERT_EQ(batch->column(1)->type_id(), ::arrow::Type::STRING);
+
+ auto snapshot_ids =
std::static_pointer_cast<::arrow::Int64Array>(batch->column(0));
+ EXPECT_EQ(snapshot_ids->Value(0), 1);
+ EXPECT_EQ(snapshot_ids->Value(1), 2);
+ auto operations =
std::static_pointer_cast<::arrow::StringArray>(batch->column(1));
+ EXPECT_EQ(operations->GetString(0), "append");
+ EXPECT_EQ(operations->GetString(1), "append");
+}
+
+TEST_F(SnapshotsTableTest, ScanSelectedColumnsIgnoresUnknownNames) {
+ ICEBERG_UNWRAP_OR_FAIL(auto stream,
+ ScanColumns(*snapshots_table_, {"parent_id",
"missing"}));
+ ICEBERG_UNWRAP_OR_FAIL(auto batches, ReadAllBatches(std::move(stream)));
+ ASSERT_EQ(batches.size(), 1);
+ ASSERT_EQ(batches.front()->num_columns(), 1);
+ EXPECT_EQ(batches.front()->column_name(0), "parent_id");
+}
+
+TEST_F(SnapshotsTableTest, ScanSelectedColumnsCaseSensitivity) {
+ ICEBERG_UNWRAP_OR_FAIL(
+ auto insensitive_stream,
+ ScanColumns(*snapshots_table_, {"SNAPSHOT_ID"},
/*case_sensitive=*/false));
+ ICEBERG_UNWRAP_OR_FAIL(auto insensitive_batches,
+ ReadAllBatches(std::move(insensitive_stream)));
+ ASSERT_EQ(insensitive_batches.size(), 1);
+ ASSERT_EQ(insensitive_batches.front()->num_columns(), 1);
+ EXPECT_EQ(insensitive_batches.front()->column_name(0), "snapshot_id");
+
+ ICEBERG_UNWRAP_OR_FAIL(auto sensitive_stream,
+ ScanColumns(*snapshots_table_, {"SNAPSHOT_ID"}));
+ ICEBERG_UNWRAP_OR_FAIL(auto sensitive_batches,
+ ReadAllBatches(std::move(sensitive_stream)));
+ ASSERT_EQ(sensitive_batches.size(), 1);
+ EXPECT_EQ(sensitive_batches.front()->num_columns(), 0);
+}
+
+TEST_F(SnapshotsTableTest, ScanAllColumnsSelection) {
+ // Selecting `*` returns the same columns as a scan without a selection.
+ ICEBERG_UNWRAP_OR_FAIL(auto stream, ScanColumns(*snapshots_table_, {"*"}));
+ ICEBERG_UNWRAP_OR_FAIL(auto batches, ReadAllBatches(std::move(stream)));
+ ASSERT_EQ(batches.size(), 1);
+ EXPECT_EQ(batches.front()->num_rows(), 2);
+ EXPECT_EQ(batches.front()->num_columns(), 6);
+}
+
+TEST_F(SnapshotsTableTest, ScanBuildsRepeatedly) {
+ // A builder can be built more than once; each build returns a new stream.
+ ICEBERG_UNWRAP_OR_FAIL(auto builder, snapshots_table_->NewScan());
+ builder->Select({"snapshot_id"});
+ for (int read = 0; read < 2; ++read) {
+ ICEBERG_UNWRAP_OR_FAIL(auto stream, builder->Build());
+ ICEBERG_UNWRAP_OR_FAIL(auto batches, ReadAllBatches(std::move(stream)));
+ ASSERT_EQ(batches.size(), 1);
+ EXPECT_EQ(batches.front()->num_rows(), 2);
+ ASSERT_EQ(batches.front()->num_columns(), 1);
+ EXPECT_EQ(batches.front()->column_name(0), "snapshot_id");
+ }
+}
+
+TEST_F(SnapshotsTableTest, ScanRejectsTimeTravel) {
+ ICEBERG_UNWRAP_OR_FAIL(auto builder, snapshots_table_->NewScan());
+ EXPECT_THAT(builder->UseSnapshot(1).Build(),
IsError(ErrorKind::kNotSupported));
+
+ ICEBERG_UNWRAP_OR_FAIL(auto ref_builder, snapshots_table_->NewScan());
+ EXPECT_THAT(ref_builder->UseRef("dev").Build(),
IsError(ErrorKind::kNotSupported));
+
+ ICEBERG_UNWRAP_OR_FAIL(auto time_builder, snapshots_table_->NewScan());
+ EXPECT_THAT(time_builder->AsOfTime(1).Build(),
IsError(ErrorKind::kNotSupported));
+}
+
+TEST_F(SnapshotsTableTest, ScanProjectedColumns) {
+ // Project() reorders columns and keeps the whole summary map.
+ auto projection = std::make_shared<Schema>(std::vector<SchemaField>{
+ SchemaField::MakeOptional(
+ 6, "summary",
+ std::make_shared<MapType>(SchemaField::MakeRequired(7, "key",
string()),
+ SchemaField::MakeRequired(8, "value",
string()))),
+ SchemaField::MakeRequired(2, "snapshot_id", int64())});
+ ICEBERG_UNWRAP_OR_FAIL(auto builder, snapshots_table_->NewScan());
+ ICEBERG_UNWRAP_OR_FAIL(auto stream, builder->Project(projection).Build());
+ ICEBERG_UNWRAP_OR_FAIL(auto batches, ReadAllBatches(std::move(stream)));
+ ASSERT_EQ(batches.size(), 1);
+ const auto& batch = batches.front();
+ EXPECT_EQ(batch->num_rows(), 2);
+ ASSERT_EQ(batch->num_columns(), 2);
+ EXPECT_EQ(batch->column_name(0), "summary");
+ EXPECT_EQ(batch->column_name(1), "snapshot_id");
+ auto summaries =
std::static_pointer_cast<::arrow::MapArray>(batch->column(0));
+ EXPECT_THAT(GetMapEntries(summaries, 1),
+ ::testing::Contains(::testing::Pair("total-records", "2")));
+ auto snapshot_ids =
std::static_pointer_cast<::arrow::Int64Array>(batch->column(1));
+ EXPECT_EQ(snapshot_ids->Value(0), 1);
+ EXPECT_EQ(snapshot_ids->Value(1), 2);
+}
+
+TEST_F(SnapshotsTableTest, ScanRejectsInvalidProjection) {
+ // An optional column cannot be projected as required.
+ auto required_parent = std::make_shared<Schema>(
+ std::vector<SchemaField>{SchemaField::MakeRequired(3, "parent_id",
int64())});
+ ICEBERG_UNWRAP_OR_FAIL(auto builder, snapshots_table_->NewScan());
+ EXPECT_THAT(builder->Project(required_parent).Build(),
+ IsError(ErrorKind::kInvalidArgument));
+
+ // A column must keep its type.
+ auto string_id = std::make_shared<Schema>(
+ std::vector<SchemaField>{SchemaField::MakeRequired(2, "snapshot_id",
string())});
+ ICEBERG_UNWRAP_OR_FAIL(auto type_builder, snapshots_table_->NewScan());
+ EXPECT_THAT(type_builder->Project(string_id).Build(),
+ IsError(ErrorKind::kNotSupported));
+
+ // Nested fields are not top-level columns.
+ auto summary_key = std::make_shared<Schema>(
+ std::vector<SchemaField>{SchemaField::MakeRequired(7, "key", string())});
+ ICEBERG_UNWRAP_OR_FAIL(auto nested_builder, snapshots_table_->NewScan());
+ EXPECT_THAT(nested_builder->Project(summary_key).Build(),
+ IsError(ErrorKind::kInvalidArgument));
+}
+
+TEST_F(SnapshotsTableTest, ScanNoColumns) {
+ ICEBERG_UNWRAP_OR_FAIL(auto stream, ScanColumns(*snapshots_table_, {}));
+ ICEBERG_UNWRAP_OR_FAIL(auto batches, ReadAllBatches(std::move(stream)));
+ ASSERT_EQ(batches.size(), 1);
+ EXPECT_EQ(batches.front()->num_rows(), 2);
+ EXPECT_EQ(batches.front()->num_columns(), 0);
+}
+
+TEST_F(SnapshotsTableTest, ScanSelectedMapColumn) {
+ ICEBERG_UNWRAP_OR_FAIL(auto stream, ScanColumns(*snapshots_table_,
{"summary.value"}));
+ ICEBERG_UNWRAP_OR_FAIL(auto batches, ReadAllBatches(std::move(stream)));
+ ASSERT_EQ(batches.size(), 1);
+ const auto& batch = batches.front();
+ ASSERT_EQ(batch->num_columns(), 1);
+ EXPECT_EQ(batch->column_name(0), "summary");
+ ASSERT_EQ(batch->column(0)->type_id(), ::arrow::Type::MAP);
+ auto summaries =
std::static_pointer_cast<::arrow::MapArray>(batch->column(0));
+ EXPECT_THAT(GetMapEntries(summaries, 0),
+ ::testing::Contains(::testing::Pair("total-records", "1")));
+ EXPECT_THAT(GetMapEntries(summaries, 1),
+ ::testing::Contains(::testing::Pair("total-records", "2")));
+
+ // A map cannot be projected without its values.
+ ICEBERG_UNWRAP_OR_FAIL(auto builder, snapshots_table_->NewScan());
+ EXPECT_THAT(builder->Select({"summary.key"}).Build(),
+ IsError(ErrorKind::kValidationFailed));
+}
+
TEST_F(SnapshotsTableTest, ScanEmptySnapshotList) {
// A table with zero snapshots should return zero rows.
ICEBERG_UNWRAP_OR_FAIL(
@@ -141,7 +299,7 @@ TEST_F(SnapshotsTableTest, ScanEmptySnapshotList) {
ICEBERG_UNWRAP_OR_FAIL(snapshots_table_,
MetadataTable::Make<SnapshotsTable>(empty_table));
- ICEBERG_UNWRAP_OR_FAIL(auto stream, snapshots_table_->Scan());
+ ICEBERG_UNWRAP_OR_FAIL(auto stream, Scan(*snapshots_table_));
ICEBERG_UNWRAP_OR_FAIL(auto batches, ReadAllBatches(std::move(stream)));
EXPECT_TRUE(batches.empty());
}
@@ -153,7 +311,7 @@ TEST_F(SnapshotsTableTest, ScanSkipsNullSnapshots) {
ICEBERG_UNWRAP_OR_FAIL(auto snapshots_table,
MetadataTable::Make<SnapshotsTable>(table));
- ICEBERG_UNWRAP_OR_FAIL(auto stream, snapshots_table->Scan());
+ ICEBERG_UNWRAP_OR_FAIL(auto stream, Scan(*snapshots_table));
ICEBERG_UNWRAP_OR_FAIL(auto batches, ReadAllBatches(std::move(stream)));
ASSERT_EQ(batches.size(), 1);
EXPECT_EQ(batches.front()->num_rows(), 2);
@@ -170,7 +328,7 @@ TEST_F(SnapshotsTableTest, ScanEmptySummary) {
ICEBERG_UNWRAP_OR_FAIL(auto snapshots_table,
MetadataTable::Make<SnapshotsTable>(table));
- ICEBERG_UNWRAP_OR_FAIL(auto stream, snapshots_table->Scan());
+ ICEBERG_UNWRAP_OR_FAIL(auto stream, Scan(*snapshots_table));
ICEBERG_UNWRAP_OR_FAIL(auto batches, ReadAllBatches(std::move(stream)));
ASSERT_EQ(batches.size(), 1);
auto summaries =
@@ -189,7 +347,7 @@ TEST_F(SnapshotsTableTest, ScanReturnsMultipleBatches) {
ICEBERG_UNWRAP_OR_FAIL(auto snapshots_table,
MetadataTable::Make<SnapshotsTable>(table));
- ICEBERG_UNWRAP_OR_FAIL(auto stream, snapshots_table->Scan());
+ ICEBERG_UNWRAP_OR_FAIL(auto stream, Scan(*snapshots_table));
ICEBERG_UNWRAP_OR_FAIL(auto batches, ReadAllBatches(std::move(stream)));
ASSERT_EQ(batches.size(), 2);
EXPECT_EQ(batches[0]->num_rows(), MetadataTable::kBatchSize);
@@ -205,7 +363,7 @@ TEST_F(SnapshotsTableTest,
ScanUsesMetadataCapturedAtCreation) {
ICEBERG_UNWRAP_OR_FAIL(auto snapshots_table,
MetadataTable::Make<SnapshotsTable>(table));
- ICEBERG_UNWRAP_OR_FAIL(auto stream, snapshots_table->Scan());
+ ICEBERG_UNWRAP_OR_FAIL(auto stream, Scan(*snapshots_table));
auto reader_result = ::arrow::ImportRecordBatchReader(&stream);
ASSERT_TRUE(reader_result.ok()) << reader_result.status().ToString();
auto reader = std::move(reader_result).ValueUnsafe();
diff --git a/src/iceberg/type_fwd.h b/src/iceberg/type_fwd.h
index 477774a1..9386c9dd 100644
--- a/src/iceberg/type_fwd.h
+++ b/src/iceberg/type_fwd.h
@@ -281,6 +281,7 @@ class PositionDeleteIndex;
/// \brief Metadata tables.
class HistoryTable;
+class MetadataScanBuilder;
class MetadataTable;
class SnapshotsTable;
