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 49b84a06 feat(inspect): add task-based scans for snapshots and history
(#1066)
49b84a06 is described below
commit 49b84a0621ef952095aa8ba52bb08fe25ee85976
Author: Gang Wu <[email protected]>
AuthorDate: Tue Nov 3 22:32:45 2026 +0800
feat(inspect): add task-based scans for snapshots and history (#1066)
- Add Table::Inspect() and task-based metadata scans while retaining
column selection, projection, case sensitivity, and time-travel
builder configuration from the merged inspection API.
- Separate task planning from reading with reusable static tasks backed
by captured table metadata.
- Preserve scan configuration and metadata across builder changes and
table refreshes.
- Produce Arrow batches lazily through MetadataScanTaskReader, with
reader-controlled batch sizes and independent stream lifetimes.
- Keep canonical schemas and row conversion in each metadata table
implementation, sharing batch construction and projection validation.
- Consolidate projected-column append helpers with lazy batch construction.
- Preserve all merged metadata inspection test scenarios and cover planned
projections, task reuse, stream lifetimes, batching, and vector<bool>.
---
src/iceberg/CMakeLists.txt | 2 +
src/iceberg/cpp20_compatibility_internal.h | 2 +
src/iceberg/inspect/history_table.cc | 127 ++++++++++-
src/iceberg/inspect/history_table.h | 38 +++-
src/iceberg/inspect/metadata_scan.cc | 143 ++++++++++--
src/iceberg/inspect/metadata_scan.h | 135 ++++++------
src/iceberg/inspect/metadata_scan_task.cc | 51 +++++
src/iceberg/inspect/metadata_scan_task.h | 88 ++++++++
src/iceberg/inspect/metadata_scan_task_reader.cc | 68 ++++++
src/iceberg/inspect/metadata_scan_task_reader.h | 61 +++++
...ns_internal.h => metadata_scan_util_internal.h} | 87 +++++++-
src/iceberg/inspect/metadata_table.cc | 43 ++--
src/iceberg/inspect/metadata_table.h | 113 ++++------
src/iceberg/inspect/snapshots_table.cc | 245 +++++++++------------
src/iceberg/inspect/snapshots_table.h | 44 +++-
src/iceberg/table.cc | 5 +
src/iceberg/table.h | 3 +
src/iceberg/test/arrow_test_util.h | 87 ++++++++
src/iceberg/test/history_table_test.cc | 171 ++++++++++++--
src/iceberg/test/metadata_table_test.cc | 163 +++++---------
src/iceberg/test/metadata_table_test_base.h | 11 +-
src/iceberg/test/snapshots_table_test.cc | 70 +++++-
src/iceberg/test/stream_test.cc | 37 ++++
src/iceberg/type_fwd.h | 7 +
src/iceberg/util/stream.h | 28 +++
25 files changed, 1342 insertions(+), 487 deletions(-)
diff --git a/src/iceberg/CMakeLists.txt b/src/iceberg/CMakeLists.txt
index cf465e5f..eee72061 100644
--- a/src/iceberg/CMakeLists.txt
+++ b/src/iceberg/CMakeLists.txt
@@ -61,6 +61,8 @@ set(ICEBERG_SOURCES
inspect/history_table.cc
inspect/metadata_scan.cc
inspect/metadata_table.cc
+ inspect/metadata_scan_task.cc
+ inspect/metadata_scan_task_reader.cc
inspect/snapshots_table.cc
inheritable_metadata.cc
json_serde.cc
diff --git a/src/iceberg/cpp20_compatibility_internal.h
b/src/iceberg/cpp20_compatibility_internal.h
index d186e5bb..6bcc5b74 100644
--- a/src/iceberg/cpp20_compatibility_internal.h
+++ b/src/iceberg/cpp20_compatibility_internal.h
@@ -82,6 +82,8 @@
#include "iceberg/inheritable_metadata.h"
#include "iceberg/inspect/history_table.h"
#include "iceberg/inspect/metadata_scan.h"
+#include "iceberg/inspect/metadata_scan_task.h"
+#include "iceberg/inspect/metadata_scan_task_reader.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 518073ca..a6f616fd 100644
--- a/src/iceberg/inspect/history_table.cc
+++ b/src/iceberg/inspect/history_table.cc
@@ -19,24 +19,67 @@
#include "iceberg/inspect/history_table.h"
+#include <array>
#include <memory>
+#include <unordered_map>
+#include <unordered_set>
#include <utility>
-#include <vector>
+#include "iceberg/arrow_row_builder_internal.h"
+#include "iceberg/constants.h"
+#include "iceberg/inspect/metadata_scan_task.h"
+#include "iceberg/inspect/metadata_scan_util_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 {
-HistoryTable::HistoryTable(std::shared_ptr<Table> table)
+struct HistoryRow {
+ const SnapshotLogEntry& entry;
+ const Snapshot* snapshot;
+ bool is_current_ancestor;
+};
+
+constexpr std::array<internal::MetadataTableColumn<HistoryRow>, 4>
kHistoryColumns = {
+ {{.field_id = 1,
+ .append =
+ [](ArrowArray* column, const HistoryRow& row) {
+ return AppendInt(column,
+ UnixMsFromTimePointMs(row.entry.timestamp_ms) *
1000);
+ }},
+ {.field_id = 2,
+ .append =
+ [](ArrowArray* column, const HistoryRow& row) {
+ return AppendInt(column, row.entry.snapshot_id);
+ }},
+ {.field_id = 3,
+ .append =
+ [](ArrowArray* column, const HistoryRow& row) {
+ return row.snapshot != nullptr && row.snapshot->parent_snapshot_id
+ ? AppendInt(column, *row.snapshot->parent_snapshot_id)
+ : AppendNull(column);
+ }},
+ {.field_id = 4, .append = [](ArrowArray* column, const HistoryRow& row) {
+ return AppendBoolean(column, row.is_current_ancestor);
+ }}}};
+
+} // namespace
+
+// HistoryTable implementation
+HistoryTable::HistoryTable(std::shared_ptr<const Table> table)
: MetadataTable(std::move(table)) {}
HistoryTable::~HistoryTable() = default;
-const std::shared_ptr<Schema>& HistoryTable::schema() const {
+const std::shared_ptr<Schema>& HistoryTable::schema() const { return
CanonicalSchema(); }
+
+const std::shared_ptr<Schema>& HistoryTable::CanonicalSchema() {
static const auto schema = std::make_shared<Schema>(std::vector<SchemaField>{
SchemaField::MakeRequired(1, "made_current_at", timestamp_tz()),
SchemaField::MakeRequired(2, "snapshot_id", int64()),
@@ -45,14 +88,84 @@ const std::shared_ptr<Schema>& HistoryTable::schema() const
{
return schema;
}
-Result<std::unique_ptr<HistoryTable>>
HistoryTable::Make(std::shared_ptr<Table> table) {
+Result<std::unique_ptr<HistoryTable>> HistoryTable::Make(
+ std::shared_ptr<const Table> table) {
ICEBERG_PRECHECK(table != nullptr, "Table cannot be null");
return std::unique_ptr<HistoryTable>(new HistoryTable(std::move(table)));
}
-Result<ArrowArrayStream> HistoryTable::ScanProjection(
- const Schema& /*projected_schema*/) const {
- return NotSupported("Scan is not supported for the history table");
+Result<std::unique_ptr<MetadataScan>> HistoryTable::MakeScan(
+ internal::MetadataScanContext context) const {
+ return HistoryScan::Make(std::move(context));
+}
+
+Result<ArrowArrayStream> HistoryTable::ReadRows(
+ std::shared_ptr<const TableMetadata> metadata,
+ const std::shared_ptr<Schema>& projected_schema, int64_t batch_size) {
+ std::unordered_map<int64_t, const Snapshot*> snapshots_by_id;
+ for (const auto& snapshot : metadata->snapshots) {
+ if (snapshot) {
+ snapshots_by_id.emplace(snapshot->snapshot_id, snapshot.get());
+ }
+ }
+
+ // Walk the current snapshot's ancestry; the insert check stops on cycles.
+ std::unordered_set<int64_t> ancestors;
+ for (auto current = metadata->current_snapshot_id; current !=
kInvalidSnapshotId;) {
+ auto it = snapshots_by_id.find(current);
+ // Stop at a missing snapshot without adding its ID to the ancestors.
+ if (it == snapshots_by_id.end() || !ancestors.insert(current).second) {
+ break;
+ }
+ if (!it->second->parent_snapshot_id) {
+ break;
+ }
+ current = *it->second->parent_snapshot_id;
+ }
+
+ const size_t num_rows = metadata->snapshot_log.size();
+ return internal::MakeMetadataBatchStream<HistoryRow>(
+ CanonicalSchema(), projected_schema, kHistoryColumns, num_rows,
batch_size,
+ [metadata = std::move(metadata), snapshots_by_id =
std::move(snapshots_by_id),
+ ancestors = std::move(ancestors)](size_t index) -> HistoryRow {
+ const auto& entry = metadata->snapshot_log[index];
+ auto it = snapshots_by_id.find(entry.snapshot_id);
+ return {.entry = entry,
+ .snapshot = it != snapshots_by_id.end() ? it->second : nullptr,
+ .is_current_ancestor = ancestors.contains(entry.snapshot_id)};
+ });
+}
+
+// HistoryScan implementation
+HistoryScan::HistoryScan(internal::MetadataScanContext context,
+ std::shared_ptr<MetadataScanTask> task)
+ : MetadataScan(std::move(context)), task_(std::move(task)) {}
+
+HistoryScan::~HistoryScan() = default;
+
+Result<std::unique_ptr<HistoryScan>> HistoryScan::Make(
+ internal::MetadataScanContext context) {
+ ICEBERG_PRECHECK(context.kind == MetadataTableKind::kHistory,
+ "Unexpected metadata table kind for history scan");
+ if (context.snapshot_id.has_value() || !context.ref_name.empty() ||
+ context.as_of_time_in_millis.has_value()) {
+ return NotSupported("Metadata table does not support time travel");
+ }
+ ICEBERG_PRECHECK(context.table_metadata != nullptr, "Table metadata cannot
be null");
+ ICEBERG_PRECHECK(context.schema != nullptr, "Metadata table schema cannot be
null");
+ ICEBERG_PRECHECK(context.projected_schema != nullptr,
+ "Projected schema cannot be null");
+ ICEBERG_RETURN_UNEXPECTED(internal::SelectAppendFns<HistoryRow>(
+ *HistoryTable::CanonicalSchema(), *context.projected_schema,
kHistoryColumns));
+ ICEBERG_ASSIGN_OR_RAISE(auto task,
+ StaticMetadataTask::Make(context.kind,
context.table_metadata,
+ context.projected_schema));
+ return std::unique_ptr<HistoryScan>(
+ new HistoryScan(std::move(context), std::move(task)));
+}
+
+Result<MetadataScanTaskStreamPtr> HistoryScan::PlanTasksStream() const {
+ return MetadataScanTaskStream::FromVector({task_});
}
} // namespace iceberg
diff --git a/src/iceberg/inspect/history_table.h
b/src/iceberg/inspect/history_table.h
index 6f07bcd0..ec18e36a 100644
--- a/src/iceberg/inspect/history_table.h
+++ b/src/iceberg/inspect/history_table.h
@@ -22,9 +22,12 @@
/// \file iceberg/inspect/history_table.h
/// \brief Define the history metadata table.
+#include <cstdint>
#include <memory>
+#include "iceberg/arrow_c_data.h"
#include "iceberg/iceberg_export.h"
+#include "iceberg/inspect/metadata_scan.h"
#include "iceberg/inspect/metadata_table.h"
#include "iceberg/result.h"
#include "iceberg/type_fwd.h"
@@ -34,19 +37,46 @@ namespace iceberg {
/// \brief History metadata table.
class ICEBERG_EXPORT HistoryTable : public MetadataTable {
public:
- static Result<std::unique_ptr<HistoryTable>> Make(std::shared_ptr<Table>
table);
+ static Result<std::unique_ptr<HistoryTable>> Make(std::shared_ptr<const
Table> table);
+
+ /// \brief Return the canonical schema of the history metadata table.
+ static const std::shared_ptr<Schema>& CanonicalSchema();
~HistoryTable() override;
- Kind kind() const noexcept override { return Kind::kHistory; }
+ MetadataTableKind kind() const noexcept override { return
MetadataTableKind::kHistory; }
const std::shared_ptr<Schema>& schema() const override;
protected:
- Result<ArrowArrayStream> ScanProjection(const Schema& projected_schema)
const override;
+ Result<std::unique_ptr<MetadataScan>> MakeScan(
+ internal::MetadataScanContext context) const override;
private:
- explicit HistoryTable(std::shared_ptr<Table> table);
+ friend class MetadataScanTaskReader;
+
+ static Result<ArrowArrayStream> ReadRows(
+ std::shared_ptr<const TableMetadata> metadata,
+ const std::shared_ptr<Schema>& projected_schema, int64_t batch_size);
+
+ explicit HistoryTable(std::shared_ptr<const Table> table);
+};
+
+/// \brief Scan the history metadata captured from a table.
+class ICEBERG_EXPORT HistoryScan final : public MetadataScan {
+ public:
+ /// \brief Create a scan from captured metadata table settings.
+ static Result<std::unique_ptr<HistoryScan>>
Make(internal::MetadataScanContext context);
+
+ ~HistoryScan() override;
+
+ Result<MetadataScanTaskStreamPtr> PlanTasksStream() const override;
+
+ private:
+ HistoryScan(internal::MetadataScanContext context,
+ std::shared_ptr<MetadataScanTask> task);
+
+ const std::shared_ptr<MetadataScanTask> task_;
};
} // namespace iceberg
diff --git a/src/iceberg/inspect/metadata_scan.cc
b/src/iceberg/inspect/metadata_scan.cc
index 6117c28c..2957a2ba 100644
--- a/src/iceberg/inspect/metadata_scan.cc
+++ b/src/iceberg/inspect/metadata_scan.cc
@@ -19,25 +19,115 @@
#include "iceberg/inspect/metadata_scan.h"
-#include <memory>
+#include <optional>
#include <utility>
+#include "iceberg/arrow_c_data_util_internal.h"
+#include "iceberg/inspect/metadata_scan_task_reader.h"
#include "iceberg/inspect/metadata_table.h"
#include "iceberg/schema.h"
+#include "iceberg/schema_internal.h"
+#include "iceberg/table.h"
#include "iceberg/util/macros.h"
+#include "iceberg/util/stream.h"
namespace iceberg {
+namespace {
+
+class TaskArrowStreamSource {
+ public:
+ TaskArrowStreamSource(std::vector<std::shared_ptr<MetadataScanTask>> tasks,
+ std::unique_ptr<MetadataScanTaskReader> reader,
+ std::shared_ptr<iceberg::Schema> schema)
+ : tasks_(std::move(tasks)),
+ reader_(std::move(reader)),
+ schema_(std::move(schema)) {}
+
+ ~TaskArrowStreamSource() { std::ignore = Close(); }
+
+ Status Close() {
+ if (current_.release != nullptr) {
+ current_.release(¤t_);
+ }
+ return {};
+ }
+
+ Result<ArrowSchema> Schema() {
+ ArrowSchema schema{};
+ ICEBERG_RETURN_UNEXPECTED(ToArrowSchema(*schema_, &schema));
+ return schema;
+ }
+
+ Result<std::optional<ArrowArray>> Next() {
+ if (error_) {
+ return iceberg::unexpected(*error_);
+ }
+ auto result = NextBatch();
+ if (!result) {
+ error_ = result.error();
+ }
+ return result;
+ }
+
+ Result<std::optional<ArrowArray>> NextBatch() {
+ while (true) {
+ if (current_.release == nullptr) {
+ if (next_task_ == tasks_.size()) {
+ return std::nullopt;
+ }
+ const auto& task = tasks_[next_task_++];
+ ICEBERG_CHECK(task != nullptr, "Metadata scan task cannot be null");
+ ICEBERG_ASSIGN_OR_RAISE(current_, reader_->Open(*task));
+ }
+
+ ArrowArray array{};
+ const int error = current_.get_next(¤t_, &array);
+ if (error != 0) {
+ std::string message = current_.get_last_error != nullptr &&
+ current_.get_last_error(¤t_) !=
nullptr
+ ? current_.get_last_error(¤t_)
+ : "Metadata task stream failed";
+ current_.release(¤t_);
+ return IOError("{}", message);
+ }
+ if (array.release != nullptr) {
+ return array;
+ }
+ current_.release(¤t_);
+ }
+ }
+
+ const std::vector<std::shared_ptr<MetadataScanTask>> tasks_;
+ const std::unique_ptr<MetadataScanTaskReader> reader_;
+ const std::shared_ptr<iceberg::Schema> schema_;
+ size_t next_task_ = 0;
+ ArrowArrayStream current_{};
+ std::optional<Error> error_;
+};
+
+} // namespace
MetadataScanBuilder::MetadataScanBuilder(const MetadataTable& table)
- : table_(&table), projected_schema_(table.schema()) {}
+ : table_(&table),
+ context_{.kind = table.kind(),
+ .table_metadata = table.source_table()->metadata(),
+ .schema = table.schema(),
+ .projected_schema = table.schema()} {}
MetadataScanBuilder::~MetadataScanBuilder() = default;
+Result<std::unique_ptr<MetadataScanBuilder>> MetadataScanBuilder::Make(
+ const MetadataTable& table) {
+ ICEBERG_PRECHECK(table.source_table() != nullptr, "Source table cannot be
null");
+ ICEBERG_PRECHECK(table.schema() != nullptr, "Metadata table schema cannot be
null");
+ return std::unique_ptr<MetadataScanBuilder>(new MetadataScanBuilder(table));
+}
+
MetadataScanBuilder& MetadataScanBuilder::Select(
const std::vector<std::string>& columns) {
- auto projected = table_->schema()->Select(columns, case_sensitive_);
+ auto projected = context_.schema->Select(columns, context_.case_sensitive);
if (projected) {
- projected_schema_ = std::move(projected).value();
+ context_.projected_schema = std::move(projected).value();
} else {
AddError(projected.error());
}
@@ -45,44 +135,61 @@ MetadataScanBuilder& MetadataScanBuilder::Select(
}
MetadataScanBuilder& MetadataScanBuilder::Project(std::shared_ptr<Schema>
schema) {
- projected_schema_ = std::move(schema);
+ context_.projected_schema = std::move(schema);
return *this;
}
MetadataScanBuilder& MetadataScanBuilder::CaseSensitive(bool value) {
- case_sensitive_ = value;
+ context_.case_sensitive = value;
return *this;
}
MetadataScanBuilder& MetadataScanBuilder::UseSnapshot(int64_t snapshot_id) {
ICEBERG_BUILDER_CHECK(
- std::holds_alternative<std::monostate>(snapshot_selection_.snapshot),
+ !context_.snapshot_id.has_value() &&
!context_.as_of_time_in_millis.has_value(),
"Cannot override snapshot, a snapshot or timestamp is already set");
- snapshot_selection_.snapshot = snapshot_id;
+ context_.snapshot_id = 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(context_.ref_name.empty(),
+ "Cannot override ref, already set ref={}",
context_.ref_name);
ICEBERG_BUILDER_CHECK(!ref.empty(), "Ref name cannot be empty");
- snapshot_selection_.ref_name = ref;
+ context_.ref_name = ref;
return *this;
}
-MetadataScanBuilder& MetadataScanBuilder::AsOfTime(int64_t timestamp_millis) {
+MetadataScanBuilder& MetadataScanBuilder::AsOfTime(int64_t
as_of_time_in_millis) {
ICEBERG_BUILDER_CHECK(
- std::holds_alternative<std::monostate>(snapshot_selection_.snapshot),
+ !context_.snapshot_id.has_value() &&
!context_.as_of_time_in_millis.has_value(),
"Cannot override snapshot, a snapshot or timestamp is already set");
- snapshot_selection_.snapshot = TimePointMsFromUnixMs(timestamp_millis);
+ context_.as_of_time_in_millis = as_of_time_in_millis;
return *this;
}
-Result<ArrowArrayStream> MetadataScanBuilder::Build() {
+Result<std::unique_ptr<MetadataScan>> MetadataScanBuilder::Build() {
ICEBERG_RETURN_UNEXPECTED(CheckErrors());
- ICEBERG_PRECHECK(projected_schema_ != nullptr, "Projected schema cannot be
null");
- return table_->ScanSnapshot(snapshot_selection_, *projected_schema_);
+ ICEBERG_PRECHECK(context_.projected_schema != nullptr,
+ "Projected schema cannot be null");
+ return table_->MakeScan(context_);
+}
+
+MetadataScan::MetadataScan(internal::MetadataScanContext context)
+ : context_(std::move(context)) {}
+
+MetadataScan::~MetadataScan() = default;
+
+Result<std::vector<std::shared_ptr<MetadataScanTask>>>
MetadataScan::PlanTasks() const {
+ ICEBERG_ASSIGN_OR_RAISE(auto stream, PlanTasksStream());
+ return stream->ToVector();
+}
+
+Result<ArrowArrayStream> MetadataScan::ToArrowArrayStream() const {
+ ICEBERG_ASSIGN_OR_RAISE(auto tasks, PlanTasks());
+ ICEBERG_ASSIGN_OR_RAISE(auto reader, MetadataScanTaskReader::Make({}));
+ return MakeArrowArrayStream(std::make_unique<TaskArrowStreamSource>(
+ std::move(tasks), std::move(reader), context_.projected_schema));
}
} // namespace iceberg
diff --git a/src/iceberg/inspect/metadata_scan.h
b/src/iceberg/inspect/metadata_scan.h
index f487bce4..5df4cc03 100644
--- a/src/iceberg/inspect/metadata_scan.h
+++ b/src/iceberg/inspect/metadata_scan.h
@@ -20,12 +20,12 @@
#pragma once
/// \file iceberg/inspect/metadata_scan.h
-/// \brief Scan configuration for Iceberg metadata tables.
+/// \brief Configure metadata scans, plan reusable tasks, and read Arrow
batches.
#include <cstdint>
#include <memory>
+#include <optional>
#include <string>
-#include <variant>
#include <vector>
#include "iceberg/arrow_c_data.h"
@@ -33,104 +33,101 @@
#include "iceberg/result.h"
#include "iceberg/type_fwd.h"
#include "iceberg/util/error_collector.h"
-#include "iceberg/util/timepoint.h"
+#include "iceberg/util/stream.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;
+using MetadataScanTaskStream = Stream<std::shared_ptr<MetadataScanTask>>;
+using MetadataScanTaskStreamPtr = std::unique_ptr<MetadataScanTaskStream>;
- /// \brief Resolve the snapshot relative to this branch or tag.
- ///
- /// An empty string uses the main branch.
+namespace internal {
+
+/// \brief Captured metadata and logical settings shared by a builder and its
scans.
+struct MetadataScanContext {
+ MetadataTableKind kind;
+ std::shared_ptr<const TableMetadata> table_metadata;
+ std::shared_ptr<Schema> schema;
+ std::shared_ptr<Schema> projected_schema;
+ bool case_sensitive = true;
+ std::optional<int64_t> snapshot_id;
std::string ref_name;
+ std::optional<int64_t> as_of_time_in_millis;
};
-/// \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
+} // namespace internal
+
+/// \brief Configure a scan over captured metadata; the metadata table must
outlive it.
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.
+ /// \brief Capture the source metadata and schema of a metadata table.
+ static Result<std::unique_ptr<MetadataScanBuilder>> Make(const
MetadataTable& table);
+
+ /// \brief Select names using the current CaseSensitive() setting.
///
- /// \param columns Column names from the metadata table's schema.
+ /// Columns are returned in schema order; unknown names are ignored. An empty
+ /// list selects no columns and `*` selects all columns. Nested names use
dots;
+ /// selecting a map value selects the whole map, while selecting only its
key fails.
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.
+ /// \brief Project whole columns by field ID, optionally reordering them.
///
- /// \param schema The output schema.
+ /// Types must match and nullable columns cannot become required.
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.
+ /// \brief Match column names case-sensitively; enabled by default.
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.
+ /// \brief Select a snapshot ID; Build rejects tables without time travel.
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.
+ /// \brief Select a branch or tag; Build rejects tables without time travel.
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.
+ /// \brief Select the latest snapshot at or before this timestamp on the
selected ref.
///
- /// \param timestamp_millis A timestamp in milliseconds since the Unix epoch.
- MetadataScanBuilder& AsOfTime(int64_t timestamp_millis);
+ /// The timestamp is milliseconds since the Unix epoch. Build rejects tables
+ /// without time travel.
+ MetadataScanBuilder& AsOfTime(int64_t as_of_time_in_millis);
- /// \brief Validate the settings and scan the metadata table.
+ /// \brief Validate the settings and build a scan with its own copy of them.
///
- /// 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();
+ /// Later builder changes and normal source table refreshes do not affect it.
+ Result<std::unique_ptr<MetadataScan>> 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_;
+ internal::MetadataScanContext context_;
+};
+
+/// \brief A metadata scan that plans tasks or reads all of them as Arrow
batches.
+class ICEBERG_EXPORT MetadataScan {
+ public:
+ virtual ~MetadataScan();
+
+ /// \brief Return a fresh task stream; consuming it may report planning
errors.
+ virtual Result<MetadataScanTaskStreamPtr> PlanTasksStream() const = 0;
+
+ /// \brief Collect every task from PlanTasksStream(), preserving planning
order.
+ virtual Result<std::vector<std::shared_ptr<MetadataScanTask>>> PlanTasks()
const;
+
+ /// \brief Read every task in planning order as one Arrow stream.
+ ///
+ /// The stream owns its read state and can outlive this scan. Release the
stream
+ /// and each returned schema or array independently. To set batch size, open
the
+ /// planned tasks with MetadataScanTaskReader; each task retains its
projection.
+ virtual Result<ArrowArrayStream> ToArrowArrayStream() const;
+
+ protected:
+ explicit MetadataScan(internal::MetadataScanContext context);
+
+ const internal::MetadataScanContext& context() const { return context_; }
+
+ private:
+ const internal::MetadataScanContext context_;
};
} // namespace iceberg
diff --git a/src/iceberg/inspect/metadata_scan_task.cc
b/src/iceberg/inspect/metadata_scan_task.cc
new file mode 100644
index 00000000..afb3eb10
--- /dev/null
+++ b/src/iceberg/inspect/metadata_scan_task.cc
@@ -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.
+ */
+
+#include "iceberg/inspect/metadata_scan_task.h"
+
+#include <utility>
+
+#include "iceberg/inspect/metadata_table.h"
+#include "iceberg/util/macros.h"
+
+namespace iceberg {
+
+// StaticMetadataTask
+
+Result<std::shared_ptr<StaticMetadataTask>> StaticMetadataTask::Make(
+ MetadataTableKind table_kind, std::shared_ptr<const TableMetadata>
metadata,
+ std::shared_ptr<Schema> projected_schema) {
+ ICEBERG_PRECHECK(metadata != nullptr, "Table metadata cannot be null");
+ ICEBERG_PRECHECK(projected_schema != nullptr, "Projected schema cannot be
null");
+ if (table_kind != MetadataTableKind::kSnapshots &&
+ table_kind != MetadataTableKind::kHistory) {
+ return NotImplemented("Static metadata table kind is not implemented");
+ }
+ return std::shared_ptr<StaticMetadataTask>(new StaticMetadataTask(
+ table_kind, std::move(metadata), std::move(projected_schema)));
+}
+
+StaticMetadataTask::StaticMetadataTask(MetadataTableKind table_kind,
+ std::shared_ptr<const TableMetadata>
metadata,
+ std::shared_ptr<Schema>
projected_schema)
+ : table_kind_(table_kind),
+ metadata_(std::move(metadata)),
+ projected_schema_(std::move(projected_schema)) {}
+
+} // namespace iceberg
diff --git a/src/iceberg/inspect/metadata_scan_task.h
b/src/iceberg/inspect/metadata_scan_task.h
new file mode 100644
index 00000000..b3511e94
--- /dev/null
+++ b/src/iceberg/inspect/metadata_scan_task.h
@@ -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.
+ */
+
+#pragma once
+
+#include <memory>
+
+#include "iceberg/iceberg_export.h"
+#include "iceberg/result.h"
+#include "iceberg/type_fwd.h"
+
+namespace iceberg {
+
+/// \brief Base class for metadata table scan tasks.
+///
+/// Metadata tables expose table metadata (snapshots, history, etc.) as
queryable
+/// Arrow tables. A task is a unit of work opened by MetadataScanTaskReader.
+class ICEBERG_EXPORT MetadataScanTask {
+ public:
+ /// \brief Task category, determines how the reader opens the task.
+ enum class TaskKind {
+ /// StaticMetadataTask reads in-memory table metadata (snapshots, history).
+ kStatic,
+ };
+
+ virtual ~MetadataScanTask() = default;
+
+ /// \brief Return how this task is read.
+ virtual TaskKind task_kind() const = 0;
+
+ /// \brief Return the metadata table whose rows this task produces.
+ virtual MetadataTableKind table_kind() const = 0;
+};
+
+/// \brief Task for static metadata tables (snapshots, history).
+///
+/// Holds the metadata captured for a table scan; readers produce Arrow batches
+/// on demand. The metadata and its nested objects must not be modified while
+/// this task or its streams are in use.
+class ICEBERG_EXPORT StaticMetadataTask : public MetadataScanTask {
+ public:
+ /// \brief Create a static metadata task.
+ ///
+ /// \param table_kind The metadata table kind (kSnapshots or kHistory)
+ /// \param metadata The table metadata to scan
+ /// \param projected_schema The planned output schema
+ /// \return A task, or an error for null metadata or an unsupported table
kind.
+ static Result<std::shared_ptr<StaticMetadataTask>> Make(
+ MetadataTableKind table_kind, std::shared_ptr<const TableMetadata>
metadata,
+ std::shared_ptr<Schema> projected_schema);
+
+ TaskKind task_kind() const override { return TaskKind::kStatic; }
+
+ MetadataTableKind table_kind() const override { return table_kind_; }
+
+ /// \brief Return the captured metadata and its shared ownership.
+ const std::shared_ptr<const TableMetadata>& metadata() const { return
metadata_; }
+
+ /// \brief Return the output schema captured when this task was planned.
+ const std::shared_ptr<Schema>& projected_schema() const { return
projected_schema_; }
+
+ private:
+ StaticMetadataTask(MetadataTableKind table_kind,
+ std::shared_ptr<const TableMetadata> metadata,
+ std::shared_ptr<Schema> projected_schema);
+
+ const MetadataTableKind table_kind_;
+ const std::shared_ptr<const TableMetadata> metadata_;
+ const std::shared_ptr<Schema> projected_schema_;
+};
+
+} // namespace iceberg
diff --git a/src/iceberg/inspect/metadata_scan_task_reader.cc
b/src/iceberg/inspect/metadata_scan_task_reader.cc
new file mode 100644
index 00000000..1ce5b3c1
--- /dev/null
+++ b/src/iceberg/inspect/metadata_scan_task_reader.cc
@@ -0,0 +1,68 @@
+/*
+ * 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_task_reader.h"
+
+#include <memory>
+#include <utility>
+
+#include "iceberg/inspect/history_table.h"
+#include "iceberg/inspect/metadata_scan_task.h"
+#include "iceberg/inspect/metadata_table.h"
+#include "iceberg/inspect/snapshots_table.h"
+#include "iceberg/util/checked_cast.h"
+#include "iceberg/util/macros.h"
+
+namespace iceberg {
+
+MetadataScanTaskReader::MetadataScanTaskReader(Options options)
+ : options_(std::move(options)) {}
+
+MetadataScanTaskReader::~MetadataScanTaskReader() = default;
+
+Result<std::unique_ptr<MetadataScanTaskReader>> MetadataScanTaskReader::Make(
+ Options options) {
+ ICEBERG_PRECHECK(options.batch_size > 0, "batch size must be positive");
+ return std::unique_ptr<MetadataScanTaskReader>(
+ new MetadataScanTaskReader(std::move(options)));
+}
+
+Result<ArrowArrayStream> MetadataScanTaskReader::Open(
+ const MetadataScanTask& task) const {
+ switch (task.task_kind()) {
+ case MetadataScanTask::TaskKind::kStatic: {
+ const auto& static_task = internal::checked_cast<const
StaticMetadataTask&>(task);
+ switch (static_task.table_kind()) {
+ case MetadataTableKind::kSnapshots:
+ return SnapshotsTable::ReadRows(static_task.metadata(),
+ static_task.projected_schema(),
+ options_.batch_size);
+ case MetadataTableKind::kHistory:
+ return HistoryTable::ReadRows(static_task.metadata(),
+ static_task.projected_schema(),
+ options_.batch_size);
+ default:
+ return NotImplemented("Static metadata table kind is not
implemented");
+ }
+ }
+ }
+ return NotImplemented("Metadata scan task kind is not implemented");
+}
+
+} // namespace iceberg
diff --git a/src/iceberg/inspect/metadata_scan_task_reader.h
b/src/iceberg/inspect/metadata_scan_task_reader.h
new file mode 100644
index 00000000..8003276b
--- /dev/null
+++ b/src/iceberg/inspect/metadata_scan_task_reader.h
@@ -0,0 +1,61 @@
+/*
+ * 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 <cstdint>
+#include <memory>
+
+#include "iceberg/arrow_c_data.h"
+#include "iceberg/iceberg_export.h"
+#include "iceberg/result.h"
+#include "iceberg/type_fwd.h"
+
+namespace iceberg {
+
+/// \brief Opens metadata scan tasks as Arrow streams.
+class ICEBERG_EXPORT MetadataScanTaskReader {
+ public:
+ /// \brief Options shared by all tasks opened by this reader.
+ struct Options {
+ /// Positive maximum number of rows per Arrow batch.
+ int64_t batch_size = 1024;
+ };
+
+ /// \brief Create a reusable task reader from shared read options.
+ static Result<std::unique_ptr<MetadataScanTaskReader>> Make(Options options);
+
+ ~MetadataScanTaskReader();
+
+ /// \brief Open a task and return an Arrow C stream for its projected rows.
+ ///
+ /// Batches are built on demand. The stream retains its input data and can
+ /// outlive this reader and the task; each call starts an independent read.
+ Result<ArrowArrayStream> Open(const MetadataScanTask& task) const;
+
+ MetadataScanTaskReader(const MetadataScanTaskReader&) = delete;
+ MetadataScanTaskReader& operator=(const MetadataScanTaskReader&) = delete;
+
+ private:
+ explicit MetadataScanTaskReader(Options options);
+
+ const Options options_;
+};
+
+} // namespace iceberg
diff --git a/src/iceberg/inspect/metadata_table_columns_internal.h
b/src/iceberg/inspect/metadata_scan_util_internal.h
similarity index 56%
rename from src/iceberg/inspect/metadata_table_columns_internal.h
rename to src/iceberg/inspect/metadata_scan_util_internal.h
index 596ffa2f..5cd2058f 100644
--- a/src/iceberg/inspect/metadata_table_columns_internal.h
+++ b/src/iceberg/inspect/metadata_scan_util_internal.h
@@ -19,20 +19,25 @@
#pragma once
-/// \file iceberg/inspect/metadata_table_columns_internal.h
-/// \brief Build only the selected columns of metadata table rows.
+/// \file iceberg/inspect/metadata_scan_util_internal.h
+/// \brief Read in-memory metadata as Arrow batches produced on demand.
#include <algorithm>
-#include <cstddef>
#include <cstdint>
+#include <memory>
+#include <optional>
#include <span>
+#include <type_traits>
+#include <utility>
#include <vector>
#include "iceberg/arrow_c_data.h"
+#include "iceberg/arrow_c_data_util_internal.h"
#include "iceberg/arrow_row_builder_internal.h"
#include "iceberg/result.h"
#include "iceberg/schema.h"
#include "iceberg/schema_field.h"
+#include "iceberg/schema_internal.h"
#include "iceberg/type.h"
#include "iceberg/util/macros.h"
@@ -100,4 +105,80 @@ Status AppendRow(ArrowRowBuilder& builder,
return builder.FinishRow();
}
+/// \brief Build batches lazily, retaining the row producer and its input
owners.
+template <typename Row, typename GetRow>
+class MetadataBatchSource {
+ public:
+ MetadataBatchSource(std::shared_ptr<iceberg::Schema> schema,
+ std::vector<AppendColumnFn<Row>> append_fns, size_t
num_rows,
+ int64_t batch_size, GetRow get_row)
+ : schema_(std::move(schema)),
+ append_fns_(std::move(append_fns)),
+ num_rows_(num_rows),
+ batch_size_(static_cast<size_t>(batch_size)),
+ get_row_(std::move(get_row)) {}
+
+ Status Close() { return {}; }
+
+ Result<ArrowSchema> Schema() {
+ ArrowSchema schema{};
+ ICEBERG_RETURN_UNEXPECTED(ToArrowSchema(*schema_, &schema));
+ return schema;
+ }
+
+ Result<std::optional<ArrowArray>> Next() {
+ if (error_) {
+ return unexpected(*error_);
+ }
+ auto result = NextBatch();
+ if (!result) {
+ error_ = result.error();
+ }
+ return result;
+ }
+
+ private:
+ Result<std::optional<ArrowArray>> NextBatch() {
+ if (index_ == num_rows_) {
+ return std::nullopt;
+ }
+ const size_t end = index_ + std::min(batch_size_, num_rows_ - index_);
+ ICEBERG_ASSIGN_OR_RAISE(auto builder, ArrowRowBuilder::Make(*schema_));
+ while (index_ < end) {
+ ICEBERG_RETURN_UNEXPECTED(AppendRow<Row>(builder, append_fns_,
get_row_(index_)));
+ ++index_;
+ }
+ ICEBERG_ASSIGN_OR_RAISE(auto batch, std::move(builder).Finish());
+ return batch;
+ }
+
+ const std::shared_ptr<iceberg::Schema> schema_;
+ const std::vector<AppendColumnFn<Row>> append_fns_;
+ const size_t num_rows_;
+ const size_t batch_size_;
+ GetRow get_row_;
+ size_t index_ = 0;
+ std::optional<Error> error_;
+};
+
+/// \brief Read only projected columns, constructing at most batch_size rows
per batch.
+///
+/// get_row must retain shared ownership of its inputs. Each stream has
independent
+/// read state, and rows are accessed only when a batch is requested.
+template <typename Row, typename GetRow>
+Result<ArrowArrayStream> MakeMetadataBatchStream(
+ const std::shared_ptr<Schema>& canonical_schema,
+ const std::shared_ptr<Schema>& projected_schema,
+ std::span<const MetadataTableColumn<Row>> columns, size_t num_rows,
+ int64_t batch_size, GetRow&& get_row) {
+ ICEBERG_PRECHECK(batch_size > 0, "Batch size must be positive");
+ ICEBERG_ASSIGN_OR_RAISE(
+ auto append_fns,
+ SelectAppendFns<Row>(*canonical_schema, *projected_schema, columns));
+ using Source = MetadataBatchSource<Row, std::decay_t<GetRow>>;
+ return MakeArrowArrayStream(
+ std::make_unique<Source>(projected_schema, std::move(append_fns),
num_rows,
+ batch_size, std::forward<GetRow>(get_row)));
+}
+
} // namespace iceberg::internal
diff --git a/src/iceberg/inspect/metadata_table.cc
b/src/iceberg/inspect/metadata_table.cc
index 3e21688f..6bd17644 100644
--- a/src/iceberg/inspect/metadata_table.cc
+++ b/src/iceberg/inspect/metadata_table.cc
@@ -21,46 +21,37 @@
#include <memory>
#include <utility>
-#include <variant>
+#include "iceberg/inspect/history_table.h"
#include "iceberg/inspect/metadata_scan.h"
+#include "iceberg/inspect/snapshots_table.h"
namespace iceberg {
-MetadataTable::MetadataTable(std::shared_ptr<Table> source_table)
+MetadataTable::MetadataTable(std::shared_ptr<const Table> source_table)
: source_table_(std::move(source_table)) {}
MetadataTable::~MetadataTable() = default;
-bool MetadataTable::supports_time_travel() const noexcept { return false; }
+Result<std::unique_ptr<MetadataTable>> MetadataTable::Make(
+ std::shared_ptr<const Table> table, MetadataTableKind kind) {
+ switch (kind) {
+ case MetadataTableKind::kSnapshots:
+ return SnapshotsTable::Make(std::move(table));
+ case MetadataTableKind::kHistory:
+ return HistoryTable::Make(std::move(table));
+ default:
+ break;
+ }
+ return NotImplemented("Metadata table kind is not implemented");
+}
-const std::shared_ptr<Table>& MetadataTable::source_table() const {
+const std::shared_ptr<const 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::ScanProjection(
- const Schema& projected_schema) const {
- return ScanSnapshot(SnapshotSelection{}, projected_schema);
+ return MetadataScanBuilder::Make(*this);
}
} // namespace iceberg
diff --git a/src/iceberg/inspect/metadata_table.h
b/src/iceberg/inspect/metadata_table.h
index 7717ce58..83f44f5c 100644
--- a/src/iceberg/inspect/metadata_table.h
+++ b/src/iceberg/inspect/metadata_table.h
@@ -20,40 +20,56 @@
#pragma once
/// \file iceberg/inspect/metadata_table.h
-/// \brief Base APIs for inspecting Iceberg metadata tables.
+/// \brief Metadata table factories, schemas, and scan entry points.
#include <concepts>
+#include <cstdint>
#include <memory>
#include <utility>
-#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"
namespace iceberg {
+namespace internal {
+struct MetadataScanContext;
+} // namespace internal
+
+/// \brief Iceberg metadata tables exposed by Table::Inspect().
+enum class MetadataTableKind : uint8_t {
+ kEntries,
+ kFiles,
+ kDataFiles,
+ kDeleteFiles,
+ kHistory,
+ kMetadataLogEntries,
+ kSnapshots,
+ kRefs,
+ kManifests,
+ kPartitions,
+ kPositionDeletes,
+ kAllEntries,
+ kAllFiles,
+ kAllDataFiles,
+ kAllDeleteFiles,
+ kAllManifests,
+};
-/// \brief Base interface for an Iceberg metadata table.
+/// \brief A metadata table backed by a source Iceberg table.
class ICEBERG_EXPORT MetadataTable {
public:
- /// \brief Supported metadata table kinds.
- enum class Kind {
- kSnapshots,
- kHistory,
- };
+ using Kind = MetadataTableKind;
- /// \brief Maximum number of rows emitted in each Arrow batch.
- static constexpr int64_t kBatchSize = 1024;
+ /// \brief Create a metadata table; unsupported kinds return NotImplemented.
+ static Result<std::unique_ptr<MetadataTable>> Make(std::shared_ptr<const
Table> table,
+ Kind kind);
- /// \brief Create a metadata table of the requested concrete type.
- ///
- /// \tparam MetadataTableType Concrete class derived from MetadataTable.
- /// \param table Source table whose metadata will be exposed.
- /// \return The constructed metadata table, or an error.
+ /// \brief Create a specific metadata table from its concrete class.
template <typename MetadataTableType>
requires std::derived_from<MetadataTableType, MetadataTable>
- static Result<std::unique_ptr<MetadataTableType>>
Make(std::shared_ptr<Table> table) {
+ static Result<std::unique_ptr<MetadataTableType>> Make(
+ std::shared_ptr<const Table> table) {
return MetadataTableType::Make(std::move(table));
}
@@ -66,70 +82,25 @@ class ICEBERG_EXPORT MetadataTable {
virtual const std::shared_ptr<Schema>& schema() const = 0;
/// \brief Return the source table whose metadata is exposed.
- const std::shared_ptr<Table>& source_table() const;
+ const std::shared_ptr<const Table>& source_table() const;
- /// \brief Return whether this metadata table supports time travel.
- virtual bool supports_time_travel() const noexcept;
-
- /// \brief Create a builder for configuring a metadata table scan.
+ /// \brief Capture the source metadata and create a scan builder.
///
- /// A builder with no options scans all columns of the current snapshot on
the main
- /// branch. This table must outlive the builder.
+ /// This metadata table must outlive the builder. Built scans retain their
own
+ /// inputs and can outlive both tables.
virtual Result<std::unique_ptr<MetadataScanBuilder>> NewScan() const;
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) const = 0;
+ explicit MetadataTable(std::shared_ptr<const Table> source_table);
- /// \brief Implement a scan for the requested snapshot selection.
- ///
- /// 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;
+ /// \brief Plan a scan using the captured metadata, projection, and snapshot
selection.
+ virtual Result<std::unique_ptr<MetadataScan>> MakeScan(
+ internal::MetadataScanContext context) const = 0;
private:
friend class MetadataScanBuilder;
- std::shared_ptr<Table> source_table_;
-};
-
-/// \brief Base interface for metadata tables that 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;
-
- /// \brief Return true because this interface supports time travel.
- bool supports_time_travel() const noexcept final;
-
- 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.
- ///
- /// \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;
+ const std::shared_ptr<const Table> source_table_;
};
} // namespace iceberg
diff --git a/src/iceberg/inspect/snapshots_table.cc
b/src/iceberg/inspect/snapshots_table.cc
index 0dede12c..7eb204b4 100644
--- a/src/iceberg/inspect/snapshots_table.cc
+++ b/src/iceberg/inspect/snapshots_table.cc
@@ -20,22 +20,15 @@
#include "iceberg/inspect/snapshots_table.h"
#include <array>
-#include <chrono>
-#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_columns_internal.h"
-#include "iceberg/nanoarrow_status_internal.h"
+#include "iceberg/inspect/metadata_scan_task.h"
+#include "iceberg/inspect/metadata_scan_util_internal.h"
#include "iceberg/schema.h"
#include "iceberg/schema_field.h"
-#include "iceberg/schema_internal.h"
#include "iceberg/snapshot.h"
#include "iceberg/table.h"
#include "iceberg/table_metadata.h"
@@ -45,165 +38,127 @@
namespace iceberg {
namespace {
-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()) {
- 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()) {
- return AppendString(column, *operation);
- }
- return AppendNull(column);
-}
-
-Status AppendManifestList(ArrowArray* column, const Snapshot& snapshot) {
- return AppendString(column, snapshot.manifest_list);
-}
-
-Status AppendSummary(ArrowArray* column, const Snapshot& snapshot) {
- if (snapshot.summary.empty()) {
- return AppendNull(column);
- }
- 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& projected_schema,
- std::vector<internal::AppendColumnFn<Snapshot>> append_fns) {
- ArrowSchema 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;
-
- Status Close() {
- metadata_.reset();
- if (arrow_schema_.release != nullptr) {
- ArrowSchemaRelease(&arrow_schema_);
- }
- return {};
- }
-
- Result<std::optional<ArrowArray>> Next() {
- 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;
- }
-
- ICEBERG_ASSIGN_OR_RAISE(auto builder,
ArrowRowBuilder::Make(&arrow_schema_));
- while (next_snapshot_ < snapshots.size() &&
- builder.num_rows() < MetadataTable::kBatchSize) {
- const auto& snapshot = snapshots[next_snapshot_++];
- if (snapshot == nullptr) [[unlikely]] {
- continue;
- }
- ICEBERG_RETURN_UNEXPECTED(
- internal::AppendRow<Snapshot>(builder, append_fns_, *snapshot));
- }
- if (builder.num_rows() == 0) {
- return std::nullopt;
- }
-
- ICEBERG_ASSIGN_OR_RAISE(auto array, std::move(builder).Finish());
- return array;
- }
-
- Result<ArrowSchema> Schema() {
- if (arrow_schema_.release == nullptr) [[unlikely]] {
- return InvalidArgument("Cannot read schema from a closed snapshots table
stream");
- }
- ArrowSchema schema_copy{};
- ICEBERG_NANOARROW_RETURN_UNEXPECTED(
- ArrowSchemaDeepCopy(&arrow_schema_, &schema_copy));
- return schema_copy;
- }
-
- private:
- SnapshotsTableStream(std::shared_ptr<const TableMetadata> metadata,
- std::vector<internal::AppendColumnFn<Snapshot>>
append_fns,
- ArrowSchema 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;
-};
+constexpr std::array<internal::MetadataTableColumn<Snapshot>, 6>
kSnapshotColumns = {
+ {{.field_id = 1,
+ .append =
+ [](ArrowArray* column, const Snapshot& row) {
+ return AppendInt(column, UnixMsFromTimePointMs(row.timestamp_ms) *
1000);
+ }},
+ {.field_id = 2,
+ .append = [](ArrowArray* column,
+ const Snapshot& row) { return AppendInt(column,
row.snapshot_id); }},
+ {.field_id = 3,
+ .append =
+ [](ArrowArray* column, const Snapshot& row) {
+ return row.parent_snapshot_id ? AppendInt(column,
*row.parent_snapshot_id)
+ : AppendNull(column);
+ }},
+ {.field_id = 4,
+ .append =
+ [](ArrowArray* column, const Snapshot& row) {
+ auto operation = row.Operation();
+ return operation ? AppendString(column, *operation) :
AppendNull(column);
+ }},
+ {.field_id = 5,
+ .append =
+ [](ArrowArray* column, const Snapshot& row) {
+ return row.manifest_list.empty() ? AppendNull(column)
+ : AppendString(column,
row.manifest_list);
+ }},
+ {.field_id = 6, .append = [](ArrowArray* column, const Snapshot& row) {
+ if (row.summary.empty()) {
+ return AppendNull(column);
+ }
+ auto summary = row.summary;
+ summary.erase(SnapshotSummaryFields::kOperation);
+ return AppendStringMap(column, summary);
+ }}}};
} // namespace
-SnapshotsTable::SnapshotsTable(std::shared_ptr<Table> table)
+// SnapshotsTable implementation
+SnapshotsTable::SnapshotsTable(std::shared_ptr<const Table> table)
: MetadataTable(std::move(table)) {}
SnapshotsTable::~SnapshotsTable() = default;
const std::shared_ptr<Schema>& SnapshotsTable::schema() const {
+ return CanonicalSchema();
+}
+
+const std::shared_ptr<Schema>& SnapshotsTable::CanonicalSchema() {
static const auto schema = std::make_shared<Schema>(std::vector<SchemaField>{
SchemaField::MakeRequired(1, "committed_at", timestamp_tz()),
SchemaField::MakeRequired(2, "snapshot_id", int64()),
SchemaField::MakeOptional(3, "parent_id", int64()),
SchemaField::MakeOptional(4, "operation", string()),
SchemaField::MakeOptional(5, "manifest_list", string()),
- SchemaField::MakeOptional(6, "summary",
- std::make_shared<iceberg::MapType>(
- SchemaField::MakeRequired(7, "key",
string()),
+ SchemaField::MakeOptional(
+ 6, "summary",
+ std::make_shared<MapType>(SchemaField::MakeRequired(7, "key",
string()),
SchemaField::MakeRequired(8, "value",
string())))});
return schema;
}
Result<std::unique_ptr<SnapshotsTable>> SnapshotsTable::Make(
- std::shared_ptr<Table> table) {
+ std::shared_ptr<const Table> table) {
ICEBERG_PRECHECK(table != nullptr, "Table cannot be null");
return std::unique_ptr<SnapshotsTable>(new SnapshotsTable(std::move(table)));
}
-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(),
- projected_schema,
std::move(append_fns)));
- return MakeArrowArrayStream(std::move(stream));
+Result<std::unique_ptr<MetadataScan>> SnapshotsTable::MakeScan(
+ internal::MetadataScanContext context) const {
+ return SnapshotsScan::Make(std::move(context));
+}
+
+Result<ArrowArrayStream> SnapshotsTable::ReadRows(
+ std::shared_ptr<const TableMetadata> metadata,
+ const std::shared_ptr<Schema>& projected_schema, int64_t batch_size) {
+ std::vector<size_t> snapshot_indices;
+ for (size_t i = 0; i < metadata->snapshots.size(); ++i) {
+ if (metadata->snapshots[i]) {
+ snapshot_indices.push_back(i);
+ }
+ }
+ const size_t num_rows = snapshot_indices.size();
+ return internal::MakeMetadataBatchStream<Snapshot>(
+ CanonicalSchema(), projected_schema, kSnapshotColumns, num_rows,
batch_size,
+ [metadata = std::move(metadata),
+ snapshot_indices = std::move(snapshot_indices)](size_t index) -> const
Snapshot& {
+ return *metadata->snapshots[snapshot_indices[index]];
+ });
+}
+
+// SnapshotsScan implementation
+SnapshotsScan::SnapshotsScan(internal::MetadataScanContext context,
+ std::shared_ptr<MetadataScanTask> task)
+ : MetadataScan(std::move(context)), task_(std::move(task)) {}
+
+SnapshotsScan::~SnapshotsScan() = default;
+
+Result<std::unique_ptr<SnapshotsScan>> SnapshotsScan::Make(
+ internal::MetadataScanContext context) {
+ ICEBERG_PRECHECK(context.kind == MetadataTableKind::kSnapshots,
+ "Unexpected metadata table kind for snapshots scan");
+ if (context.snapshot_id.has_value() || !context.ref_name.empty() ||
+ context.as_of_time_in_millis.has_value()) {
+ return NotSupported("Metadata table does not support time travel");
+ }
+ ICEBERG_PRECHECK(context.table_metadata != nullptr, "Table metadata cannot
be null");
+ ICEBERG_PRECHECK(context.schema != nullptr, "Metadata table schema cannot be
null");
+ ICEBERG_PRECHECK(context.projected_schema != nullptr,
+ "Projected schema cannot be null");
+ ICEBERG_RETURN_UNEXPECTED(internal::SelectAppendFns<Snapshot>(
+ *SnapshotsTable::CanonicalSchema(), *context.projected_schema,
kSnapshotColumns));
+ ICEBERG_ASSIGN_OR_RAISE(auto task,
+ StaticMetadataTask::Make(context.kind,
context.table_metadata,
+ context.projected_schema));
+ return std::unique_ptr<SnapshotsScan>(
+ new SnapshotsScan(std::move(context), std::move(task)));
+}
+
+Result<MetadataScanTaskStreamPtr> SnapshotsScan::PlanTasksStream() const {
+ return MetadataScanTaskStream::FromVector({task_});
}
} // namespace iceberg
diff --git a/src/iceberg/inspect/snapshots_table.h
b/src/iceberg/inspect/snapshots_table.h
index ae1434a5..4bfaa554 100644
--- a/src/iceberg/inspect/snapshots_table.h
+++ b/src/iceberg/inspect/snapshots_table.h
@@ -22,9 +22,12 @@
/// \file iceberg/inspect/snapshots_table.h
/// \brief Define the snapshots metadata table.
+#include <cstdint>
#include <memory>
+#include "iceberg/arrow_c_data.h"
#include "iceberg/iceberg_export.h"
+#include "iceberg/inspect/metadata_scan.h"
#include "iceberg/inspect/metadata_table.h"
#include "iceberg/result.h"
#include "iceberg/type_fwd.h"
@@ -34,22 +37,49 @@ namespace iceberg {
/// \brief Snapshots metadata table.
class ICEBERG_EXPORT SnapshotsTable : public MetadataTable {
public:
- static Result<std::unique_ptr<SnapshotsTable>> Make(std::shared_ptr<Table>
table);
+ static Result<std::unique_ptr<SnapshotsTable>> Make(std::shared_ptr<const
Table> table);
+
+ /// \brief Return the canonical schema of the snapshots metadata table.
+ static const std::shared_ptr<Schema>& CanonicalSchema();
~SnapshotsTable() override;
- Kind kind() const noexcept override { return Kind::kSnapshots; }
+ MetadataTableKind kind() const noexcept override {
+ return MetadataTableKind::kSnapshots;
+ }
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> ScanProjection(const Schema& projected_schema)
const override;
+ Result<std::unique_ptr<MetadataScan>> MakeScan(
+ internal::MetadataScanContext context) const override;
private:
- explicit SnapshotsTable(std::shared_ptr<Table> table);
+ friend class MetadataScanTaskReader;
+
+ static Result<ArrowArrayStream> ReadRows(
+ std::shared_ptr<const TableMetadata> metadata,
+ const std::shared_ptr<Schema>& projected_schema, int64_t batch_size);
+
+ explicit SnapshotsTable(std::shared_ptr<const Table> table);
+};
+
+/// \brief Scan the snapshots metadata captured from a table.
+class ICEBERG_EXPORT SnapshotsScan final : public MetadataScan {
+ public:
+ /// \brief Create a scan from captured metadata table settings.
+ static Result<std::unique_ptr<SnapshotsScan>> Make(
+ internal::MetadataScanContext context);
+
+ ~SnapshotsScan() override;
+
+ Result<MetadataScanTaskStreamPtr> PlanTasksStream() const override;
+
+ private:
+ SnapshotsScan(internal::MetadataScanContext context,
+ std::shared_ptr<MetadataScanTask> task);
+
+ const std::shared_ptr<MetadataScanTask> task_;
};
} // namespace iceberg
diff --git a/src/iceberg/table.cc b/src/iceberg/table.cc
index e56746e3..0fa6bf40 100644
--- a/src/iceberg/table.cc
+++ b/src/iceberg/table.cc
@@ -25,6 +25,7 @@
#include "iceberg/data_table_scan.h"
#include "iceberg/incremental_append_scan.h"
#include "iceberg/incremental_changelog_scan.h"
+#include "iceberg/inspect/metadata_table.h"
#include "iceberg/labels.h"
#include "iceberg/location_provider.h"
#include "iceberg/partition_spec.h"
@@ -294,6 +295,10 @@ Result<std::shared_ptr<SnapshotManager>>
Table::NewSnapshotManager() {
return SnapshotManager::Make(shared_from_this());
}
+Result<std::unique_ptr<MetadataTable>> Table::Inspect(MetadataTableKind kind)
const {
+ return MetadataTable::Make(shared_from_this(), kind);
+}
+
Result<std::shared_ptr<StagedTable>> StagedTable::Make(
TableIdentifier identifier, std::shared_ptr<TableMetadata> metadata,
std::string metadata_location, std::shared_ptr<FileIO> io,
diff --git a/src/iceberg/table.h b/src/iceberg/table.h
index 97b1a43a..50e6aa8f 100644
--- a/src/iceberg/table.h
+++ b/src/iceberg/table.h
@@ -221,6 +221,9 @@ class ICEBERG_EXPORT Table : public
std::enable_shared_from_this<Table> {
/// \brief Create a new SnapshotManager to manage snapshots and snapshot
references.
virtual Result<std::shared_ptr<SnapshotManager>> NewSnapshotManager();
+ /// \brief Construct a metadata table inspector for this table.
+ Result<std::unique_ptr<MetadataTable>> Inspect(MetadataTableKind kind) const;
+
protected:
Table(TableIdentifier identifier, std::shared_ptr<TableMetadata> metadata,
std::string metadata_location, std::shared_ptr<FileIO> io,
diff --git a/src/iceberg/test/arrow_test_util.h
b/src/iceberg/test/arrow_test_util.h
new file mode 100644
index 00000000..3f3d58c3
--- /dev/null
+++ b/src/iceberg/test/arrow_test_util.h
@@ -0,0 +1,87 @@
+/*
+ * 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 <memory>
+#include <string_view>
+#include <utility>
+
+#include <arrow/array.h>
+#include <arrow/c/bridge.h>
+#include <arrow/json/from_string.h>
+#include <arrow/record_batch.h>
+#include <arrow/table.h>
+#include <gtest/gtest.h>
+
+#include "iceberg/arrow/arrow_status_internal.h"
+#include "iceberg/arrow_c_data.h"
+#include "iceberg/result.h"
+#include "iceberg/schema.h"
+#include "iceberg/schema_internal.h"
+#include "iceberg/util/macros.h"
+
+namespace iceberg {
+
+/// \brief Consume an Arrow C stream into a table, retaining the stream's
schema.
+inline Result<std::shared_ptr<::arrow::Table>>
ReadArrowTable(ArrowArrayStream&& stream) {
+ auto reader = ::arrow::ImportRecordBatchReader(&stream);
+ if (!reader.ok()) {
+ if (stream.release != nullptr) {
+ stream.release(&stream);
+ }
+ return InvalidArrowData(reader.status().ToString());
+ }
+ ICEBERG_ARROW_ASSIGN_OR_RETURN(auto table, reader.ValueUnsafe()->ToTable());
+ return table;
+}
+
+/// \brief Build an expected Arrow table from an Iceberg schema and JSON rows.
+inline Result<std::shared_ptr<::arrow::Table>> TableFromJson(const Schema&
schema,
+ std::string_view
rows) {
+ ArrowSchema c_schema{};
+ ICEBERG_RETURN_UNEXPECTED(ToArrowSchema(schema, &c_schema));
+ ICEBERG_ARROW_ASSIGN_OR_RETURN(auto arrow_schema,
::arrow::ImportSchema(&c_schema));
+ ICEBERG_ARROW_ASSIGN_OR_RETURN(
+ auto array,
+
::arrow::json::ArrayFromJSONString(::arrow::struct_(arrow_schema->fields()),
rows));
+ const auto& struct_array = static_cast<const ::arrow::StructArray&>(*array);
+ auto batch =
+ ::arrow::RecordBatch::Make(arrow_schema, array->length(),
struct_array.fields());
+ ICEBERG_ARROW_ASSIGN_OR_RETURN(
+ auto table, ::arrow::Table::FromRecordBatches(arrow_schema,
{std::move(batch)}));
+ return table;
+}
+
+/// \brief Compare schemas and values independently of Arrow chunk boundaries.
+inline ::testing::AssertionResult TablesEqual(const ::arrow::Table& expected,
+ const ::arrow::Table& actual) {
+ const auto status = actual.ValidateFull();
+ if (!status.ok()) {
+ return ::testing::AssertionFailure() << status.ToString();
+ }
+ if (expected.Equals(actual, /*check_metadata=*/true)) {
+ return ::testing::AssertionSuccess();
+ }
+ return ::testing::AssertionFailure() << "Expected:\n"
+ << expected.ToString() << "\nActual:\n"
+ << actual.ToString();
+}
+
+} // namespace iceberg
diff --git a/src/iceberg/test/history_table_test.cc
b/src/iceberg/test/history_table_test.cc
index 8da311c0..1dbcdefb 100644
--- a/src/iceberg/test/history_table_test.cc
+++ b/src/iceberg/test/history_table_test.cc
@@ -17,39 +17,174 @@
* under the License.
*/
-/// \file history_table_test.cc
-/// Unit tests for HistoryTable.
-
#include "iceberg/inspect/history_table.h"
-#include <gmock/gmock.h>
+#include <cstdint>
+#include <memory>
+#include <utility>
+#include <vector>
+
#include <gtest/gtest.h>
-#include "iceberg/inspect/metadata_table.h"
+#include "iceberg/inspect/metadata_scan_task_reader.h"
#include "iceberg/schema.h"
#include "iceberg/schema_field.h"
+#include "iceberg/snapshot.h"
+#include "iceberg/table_metadata.h"
+#include "iceberg/test/arrow_test_util.h"
#include "iceberg/test/matchers.h"
#include "iceberg/test/metadata_table_test_base.h"
#include "iceberg/type.h"
+#include "iceberg/util/timepoint.h"
namespace iceberg {
-namespace {
-
-std::shared_ptr<Schema> MakeHistorySchema() {
- return std::make_shared<Schema>(std::vector<SchemaField>{
- SchemaField::MakeRequired(1, "made_current_at", timestamp_tz()),
- SchemaField::MakeRequired(2, "snapshot_id", int64()),
- SchemaField::MakeOptional(3, "parent_id", int64()),
- SchemaField::MakeRequired(4, "is_current_ancestor", boolean())});
+
+class HistoryTableTest : public MetadataTableTestBase {
+ protected:
+ void SetUp() override {
+ MetadataTableTestBase::SetUp();
+ metadata_->snapshots = {
+ std::make_shared<Snapshot>(Snapshot{.snapshot_id = 1}),
+ std::make_shared<Snapshot>(Snapshot{.snapshot_id = 2,
.parent_snapshot_id = 1}),
+ std::make_shared<Snapshot>(Snapshot{.snapshot_id = 3,
.parent_snapshot_id = 1})};
+ metadata_->current_snapshot_id = 3;
+ // Roll back from 2 to 1, include an expired snapshot (4), then move to 3.
+ metadata_->snapshot_log = {
+ {.timestamp_ms = TimePointMsFromUnixMs(1000), .snapshot_id = 1},
+ {.timestamp_ms = TimePointMsFromUnixMs(2000), .snapshot_id = 2},
+ {.timestamp_ms = TimePointMsFromUnixMs(2500), .snapshot_id = 1},
+ {.timestamp_ms = TimePointMsFromUnixMs(2700), .snapshot_id = 4},
+ {.timestamp_ms = TimePointMsFromUnixMs(3000), .snapshot_id = 3}};
+ ICEBERG_UNWRAP_OR_FAIL(metadata_table_,
table_->Inspect(MetadataTableKind::kHistory));
+ }
+
+ static std::shared_ptr<Schema> ExpectedSchema() {
+ return std::make_shared<Schema>(std::vector<SchemaField>{
+ SchemaField::MakeRequired(1, "made_current_at", timestamp_tz()),
+ SchemaField::MakeRequired(2, "snapshot_id", int64()),
+ SchemaField::MakeOptional(3, "parent_id", int64()),
+ SchemaField::MakeRequired(4, "is_current_ancestor", boolean())});
+ }
+
+ std::unique_ptr<MetadataTable> metadata_table_;
+};
+
+TEST_F(HistoryTableTest, CanonicalSchema) {
+ const auto expected = ExpectedSchema();
+ EXPECT_EQ(*HistoryTable::CanonicalSchema(), *expected);
+ EXPECT_EQ(*metadata_table_->schema(), *expected);
}
-} // namespace
+TEST_F(HistoryTableTest, Read) {
+ ICEBERG_UNWRAP_OR_FAIL(auto builder, metadata_table_->NewScan());
+ ICEBERG_UNWRAP_OR_FAIL(auto scan, builder->Build());
+ ICEBERG_UNWRAP_OR_FAIL(auto tasks, scan->PlanTasks());
+ ASSERT_EQ(tasks.size(), 1);
+ ICEBERG_UNWRAP_OR_FAIL(auto expected, TableFromJson(*ExpectedSchema(), R"([
+
{"made_current_at":1000000,"snapshot_id":1,"parent_id":null,"is_current_ancestor":true},
+
{"made_current_at":2000000,"snapshot_id":2,"parent_id":1,"is_current_ancestor":false},
+
{"made_current_at":2500000,"snapshot_id":1,"parent_id":null,"is_current_ancestor":true},
+
{"made_current_at":2700000,"snapshot_id":4,"parent_id":null,"is_current_ancestor":false},
+
{"made_current_at":3000000,"snapshot_id":3,"parent_id":1,"is_current_ancestor":true}
+ ])"));
+ for (int64_t batch_size : {1, 2}) {
+ SCOPED_TRACE(batch_size);
+ ICEBERG_UNWRAP_OR_FAIL(auto reader,
+ MetadataScanTaskReader::Make({.batch_size =
batch_size}));
+ ICEBERG_UNWRAP_OR_FAIL(auto stream, reader->Open(*tasks.front()));
+ ICEBERG_UNWRAP_OR_FAIL(auto actual, ReadArrowTable(std::move(stream)));
+ EXPECT_TRUE(TablesEqual(*expected, *actual));
+ }
+}
-class HistoryTableTest : public MetadataTableTestBase {};
+TEST_F(HistoryTableTest, Project) {
+ ICEBERG_UNWRAP_OR_FAIL(auto builder, metadata_table_->NewScan());
+ builder->Select({"snapshot_id", "is_current_ancestor"});
+ ICEBERG_UNWRAP_OR_FAIL(auto selected_scan, builder->Build());
+
+ auto projection = std::make_shared<Schema>(
+ std::vector<SchemaField>{SchemaField::MakeOptional(3, "parent_id",
int64()),
+ SchemaField::MakeRequired(2, "snapshot_id",
int64())});
+ builder->Project(projection);
+ ICEBERG_UNWRAP_OR_FAIL(auto projected_scan, builder->Build());
+
+ Schema selection({SchemaField::MakeRequired(2, "snapshot_id", int64()),
+ SchemaField::MakeRequired(4, "is_current_ancestor",
boolean())});
+ ICEBERG_UNWRAP_OR_FAIL(auto expected, TableFromJson(selection, R"([
+ {"snapshot_id":1,"is_current_ancestor":true},
+ {"snapshot_id":2,"is_current_ancestor":false},
+ {"snapshot_id":1,"is_current_ancestor":true},
+ {"snapshot_id":4,"is_current_ancestor":false},
+ {"snapshot_id":3,"is_current_ancestor":true}
+ ])"));
+ ICEBERG_UNWRAP_OR_FAIL(auto stream, selected_scan->ToArrowArrayStream());
+ ICEBERG_UNWRAP_OR_FAIL(auto actual, ReadArrowTable(std::move(stream)));
+ EXPECT_TRUE(TablesEqual(*expected, *actual));
+
+ ICEBERG_UNWRAP_OR_FAIL(auto reordered, TableFromJson(*projection, R"([
+ {"parent_id":null,"snapshot_id":1},
+ {"parent_id":1,"snapshot_id":2},
+ {"parent_id":null,"snapshot_id":1},
+ {"parent_id":null,"snapshot_id":4},
+ {"parent_id":1,"snapshot_id":3}
+ ])"));
+ ICEBERG_UNWRAP_OR_FAIL(auto tasks, projected_scan->PlanTasks());
+ ASSERT_EQ(tasks.size(), 1);
+ ICEBERG_UNWRAP_OR_FAIL(auto reader,
MetadataScanTaskReader::Make({.batch_size = 2}));
+ ICEBERG_UNWRAP_OR_FAIL(auto projected_stream, reader->Open(*tasks.front()));
+ ICEBERG_UNWRAP_OR_FAIL(auto projected,
ReadArrowTable(std::move(projected_stream)));
+ EXPECT_TRUE(TablesEqual(*reordered, *projected));
+}
+
+TEST_F(HistoryTableTest, MissingSnapshotsAreNotAncestors) {
+ // Snapshot 4 is in the history log but no longer exists in snapshots.
+ for (bool missing_current : {true, false}) {
+ SCOPED_TRACE(missing_current);
+ metadata_->current_snapshot_id = missing_current ? 4 : 3;
+ metadata_->snapshots.back()->parent_snapshot_id = 4;
+ ICEBERG_UNWRAP_OR_FAIL(auto builder, metadata_table_->NewScan());
+ builder->Select({"snapshot_id", "is_current_ancestor"});
+ ICEBERG_UNWRAP_OR_FAIL(auto scan, builder->Build());
+ ICEBERG_UNWRAP_OR_FAIL(auto stream, scan->ToArrowArrayStream());
+ ICEBERG_UNWRAP_OR_FAIL(auto actual, ReadArrowTable(std::move(stream)));
+ Schema schema({SchemaField::MakeRequired(2, "snapshot_id", int64()),
+ SchemaField::MakeRequired(4, "is_current_ancestor",
boolean())});
+ const auto rows = missing_current ? R"([
+ {"snapshot_id":1,"is_current_ancestor":false},
+ {"snapshot_id":2,"is_current_ancestor":false},
+ {"snapshot_id":1,"is_current_ancestor":false},
+ {"snapshot_id":4,"is_current_ancestor":false},
+ {"snapshot_id":3,"is_current_ancestor":false}
+ ])"
+ : R"([
+ {"snapshot_id":1,"is_current_ancestor":false},
+ {"snapshot_id":2,"is_current_ancestor":false},
+ {"snapshot_id":1,"is_current_ancestor":false},
+ {"snapshot_id":4,"is_current_ancestor":false},
+ {"snapshot_id":3,"is_current_ancestor":true}
+ ])";
+ ICEBERG_UNWRAP_OR_FAIL(auto expected, TableFromJson(schema, rows));
+ EXPECT_TRUE(TablesEqual(*expected, *actual));
+ }
+}
-TEST_F(HistoryTableTest, SchemaMatchesIcebergSchema) {
- ICEBERG_UNWRAP_OR_FAIL(auto history_table,
MetadataTable::Make<HistoryTable>(table_));
- EXPECT_TRUE(*history_table->schema() == *MakeHistorySchema());
+TEST_F(HistoryTableTest, CyclicAncestryTerminates) {
+ metadata_->snapshots.front()->parent_snapshot_id = 3;
+ ICEBERG_UNWRAP_OR_FAIL(auto builder, metadata_table_->NewScan());
+ builder->Select({"snapshot_id", "is_current_ancestor"});
+ ICEBERG_UNWRAP_OR_FAIL(auto scan, builder->Build());
+ ICEBERG_UNWRAP_OR_FAIL(auto stream, scan->ToArrowArrayStream());
+ ICEBERG_UNWRAP_OR_FAIL(auto actual, ReadArrowTable(std::move(stream)));
+ Schema schema({SchemaField::MakeRequired(2, "snapshot_id", int64()),
+ SchemaField::MakeRequired(4, "is_current_ancestor",
boolean())});
+ ICEBERG_UNWRAP_OR_FAIL(auto expected, TableFromJson(schema, R"([
+ {"snapshot_id":1,"is_current_ancestor":true},
+ {"snapshot_id":2,"is_current_ancestor":false},
+ {"snapshot_id":1,"is_current_ancestor":true},
+ {"snapshot_id":4,"is_current_ancestor":false},
+ {"snapshot_id":3,"is_current_ancestor":true}
+ ])"));
+ EXPECT_TRUE(TablesEqual(*expected, *actual));
}
} // namespace iceberg
diff --git a/src/iceberg/test/metadata_table_test.cc
b/src/iceberg/test/metadata_table_test.cc
index 187646c9..7babd917 100644
--- a/src/iceberg/test/metadata_table_test.cc
+++ b/src/iceberg/test/metadata_table_test.cc
@@ -23,7 +23,6 @@
#include <memory>
#include <string>
#include <utility>
-#include <variant>
#include <vector>
#include <arrow/c/bridge.h>
@@ -32,78 +31,40 @@
#include <gmock/gmock.h>
#include <gtest/gtest.h>
-#include "iceberg/constants.h"
#include "iceberg/inspect/history_table.h"
+#include "iceberg/inspect/metadata_scan.h"
+#include "iceberg/inspect/metadata_scan_task.h"
#include "iceberg/inspect/snapshots_table.h"
#include "iceberg/schema.h"
#include "iceberg/schema_field.h"
#include "iceberg/table.h"
-#include "iceberg/table_identifier.h"
#include "iceberg/table_metadata.h"
+#include "iceberg/test/arrow_test_util.h"
#include "iceberg/test/matchers.h"
-#include "iceberg/test/mock_catalog.h"
-#include "iceberg/test/mock_io.h"
+#include "iceberg/test/metadata_table_test_base.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 {
+class PlannedMetadataScan : public MetadataScan {
public:
- explicit RecordingTimeTravelTable(std::shared_ptr<Table> table)
- : TimeTravelMetadataTable(std::move(table)) {}
+ PlannedMetadataScan(internal::MetadataScanContext context,
+ std::vector<std::shared_ptr<MetadataScanTask>> tasks)
+ : MetadataScan(std::move(context)), tasks_(std::move(tasks)) {}
- 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");
+ Result<MetadataScanTaskStreamPtr> PlanTasksStream() const override {
+ return MetadataScanTaskStream::FromVector(tasks_);
}
private:
- mutable SnapshotSelection scanned_snapshot_;
- mutable std::vector<std::string> scanned_columns_;
+ const std::vector<std::shared_ptr<MetadataScanTask>> tasks_;
};
} // namespace
-class MetadataTableTest : public ::testing::Test {
+class MetadataTableTest : public MetadataTableTestBase {
protected:
- void SetUp() override {
- auto schema = std::make_shared<Schema>(
- std::vector<SchemaField>{SchemaField::MakeRequired(1, "id", int64()),
- SchemaField::MakeOptional(2, "name",
string())},
- 1);
- auto metadata = std::make_shared<TableMetadata>(
- TableMetadata{.format_version = 2,
- .schemas = {schema},
- .current_schema_id = 1,
- .current_snapshot_id = kInvalidSnapshotId});
-
- TableIdentifier ident{.ns = Namespace{.levels = {"db"}}, .name =
"source_table"};
- ICEBERG_UNWRAP_OR_FAIL(table_, Table::Make(ident, metadata,
"s3://bucket/meta.json",
- std::make_shared<MockFileIO>(),
-
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);
@@ -115,8 +76,6 @@ class MetadataTableTest : public ::testing::Test {
}
return reader_result.ValueUnsafe()->schema();
}
-
- std::shared_ptr<Table> table_;
};
TEST_F(MetadataTableTest, FactoryRejectsNullSourceTable) {
@@ -131,15 +90,15 @@ TEST_F(MetadataTableTest,
ScanBuilderSelectResolvesNamesImmediately) {
// 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 stream,
BuildStream(builder->Select({"SNAPSHOT_ID"}).CaseSensitive(false)));
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());
+
BuildStream(insensitive_builder->CaseSensitive(false).Select({"SNAPSHOT_ID"})));
ICEBERG_UNWRAP_OR_FAIL(auto insensitive_schema,
ReadSchema(std::move(insensitive_stream)));
EXPECT_THAT(insensitive_schema->field_names(),
::testing::ElementsAre("snapshot_id"));
@@ -160,8 +119,8 @@ TEST_F(MetadataTableTest, ScanBuilderProject) {
// 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 stream,
BuildStream(builder->Select({"snapshot_id"}).Project(projection)));
ICEBERG_UNWRAP_OR_FAIL(auto schema, ReadSchema(std::move(stream)));
EXPECT_THAT(schema->field_names(), ::testing::ElementsAre("operation",
"snapshot_id"));
@@ -170,73 +129,30 @@ TEST_F(MetadataTableTest, ScanBuilderProject) {
IsError(ErrorKind::kInvalidArgument));
}
-TEST_F(MetadataTableTest, SupportsTimeTravel) {
- ICEBERG_UNWRAP_OR_FAIL(auto snapshots_table,
- MetadataTable::Make<SnapshotsTable>(table_));
- EXPECT_FALSE(snapshots_table->supports_time_travel());
-
- 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 table,
MetadataTable::Make<SnapshotsTable>(table_));
- ICEBERG_UNWRAP_OR_FAIL(auto snapshot_builder, table.NewScan());
+ 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());
+ 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());
+ ICEBERG_UNWRAP_OR_FAIL(auto empty_ref_builder, table->NewScan());
EXPECT_THAT(empty_ref_builder->UseRef("").Build(),
IsError(ErrorKind::kValidationFailed));
}
-TEST_F(MetadataTableTest, HistoryTableScanIsNotSupported) {
+TEST_F(MetadataTableTest, HistoryTableScanAndTimeTravelSelection) {
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));
+ ICEBERG_UNWRAP_OR_FAIL(auto stream,
BuildStream(builder->Select({"snapshot_id"})));
+ ICEBERG_UNWRAP_OR_FAIL(auto schema, ReadSchema(std::move(stream)));
+ EXPECT_THAT(schema->field_names(), ::testing::ElementsAre("snapshot_id"));
// Tables without time travel reject snapshot selections.
ICEBERG_UNWRAP_OR_FAIL(auto snapshot_builder, history_table->NewScan());
@@ -245,4 +161,33 @@ TEST_F(MetadataTableTest, HistoryTableScanIsNotSupported) {
EXPECT_THAT(snapshot_result, HasErrorMessage("does not support time
travel"));
}
+TEST_F(MetadataTableTest, ScanReadsEveryTaskInPlanningOrder) {
+ auto projection = std::make_shared<Schema>(
+ std::vector<SchemaField>{SchemaField::MakeRequired(2, "snapshot_id",
int64())});
+ auto metadata = std::make_shared<TableMetadata>();
+ metadata->snapshots = {std::make_shared<Snapshot>(Snapshot{.snapshot_id =
2})};
+ ICEBERG_UNWRAP_OR_FAIL(
+ auto first,
+ StaticMetadataTask::Make(MetadataTableKind::kSnapshots, metadata,
projection));
+ auto next_metadata = std::make_shared<TableMetadata>();
+ next_metadata->snapshots = {std::make_shared<Snapshot>(Snapshot{.snapshot_id
= 1})};
+ ICEBERG_UNWRAP_OR_FAIL(
+ auto second,
+ StaticMetadataTask::Make(MetadataTableKind::kSnapshots, next_metadata,
projection));
+ ICEBERG_UNWRAP_OR_FAIL(auto empty, StaticMetadataTask::Make(
+ MetadataTableKind::kSnapshots,
+ std::make_shared<TableMetadata>(),
projection));
+ PlannedMetadataScan scan({.kind = MetadataTableKind::kSnapshots,
+ .table_metadata = metadata,
+ .schema = SnapshotsTable::CanonicalSchema(),
+ .projected_schema = projection},
+ {empty, first, empty, second});
+ ICEBERG_UNWRAP_OR_FAIL(auto stream, scan.ToArrowArrayStream());
+ ICEBERG_UNWRAP_OR_FAIL(auto actual, ReadArrowTable(std::move(stream)));
+ ICEBERG_UNWRAP_OR_FAIL(
+ auto expected,
+ TableFromJson(*projection, R"([{"snapshot_id":2},{"snapshot_id":1}])"));
+ EXPECT_TRUE(TablesEqual(*expected, *actual));
+}
+
} // namespace iceberg
diff --git a/src/iceberg/test/metadata_table_test_base.h
b/src/iceberg/test/metadata_table_test_base.h
index cefdb5cb..de44c1d1 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_scan.h"
#include "iceberg/inspect/metadata_table.h"
#include "iceberg/schema.h"
#include "iceberg/schema_field.h"
@@ -81,10 +82,16 @@ class MetadataTableTestBase : public ::testing::Test {
"s3://bucket/meta.json", io_,
catalog_));
}
+ /// \brief Build a scan and open its projected rows.
+ static Result<ArrowArrayStream> BuildStream(MetadataScanBuilder& builder) {
+ ICEBERG_ASSIGN_OR_RAISE(auto scan, builder.Build());
+ return scan->ToArrowArrayStream();
+ }
+
/// \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();
+ return BuildStream(*builder);
}
/// \brief Scan the selected columns of a metadata table.
@@ -92,7 +99,7 @@ class MetadataTableTestBase : public ::testing::Test {
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();
+ return BuildStream(builder->CaseSensitive(case_sensitive).Select(columns));
}
/// \brief Import and consume a scan-produced ArrowArrayStream.
diff --git a/src/iceberg/test/snapshots_table_test.cc
b/src/iceberg/test/snapshots_table_test.cc
index 9a62c6b6..dea10a01 100644
--- a/src/iceberg/test/snapshots_table_test.cc
+++ b/src/iceberg/test/snapshots_table_test.cc
@@ -32,7 +32,9 @@
#include <gtest/gtest.h>
#include "iceberg/constants.h"
+#include "iceberg/inspect/metadata_scan_task_reader.h"
#include "iceberg/inspect/metadata_table.h"
+#include "iceberg/test/arrow_test_util.h"
#include "iceberg/test/matchers.h"
#include "iceberg/test/metadata_table_test_base.h"
@@ -194,7 +196,7 @@ TEST_F(SnapshotsTableTest, ScanBuildsRepeatedly) {
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 stream, BuildStream(*builder));
ICEBERG_UNWRAP_OR_FAIL(auto batches, ReadAllBatches(std::move(stream)));
ASSERT_EQ(batches.size(), 1);
EXPECT_EQ(batches.front()->num_rows(), 2);
@@ -223,7 +225,7 @@ TEST_F(SnapshotsTableTest, ScanProjectedColumns) {
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 stream,
BuildStream(builder->Project(projection)));
ICEBERG_UNWRAP_OR_FAIL(auto batches, ReadAllBatches(std::move(stream)));
ASSERT_EQ(batches.size(), 1);
const auto& batch = batches.front();
@@ -320,6 +322,7 @@ TEST_F(SnapshotsTableTest, ScanSkipsNullSnapshots) {
TEST_F(SnapshotsTableTest, ScanEmptySummary) {
auto [missing_summary, operation_only_summary] = MakeTestSnapshots();
missing_summary->summary.clear();
+ missing_summary->manifest_list.clear();
operation_only_summary->summary = {
{SnapshotSummaryFields::kOperation, DataOperation::kAppend}};
ICEBERG_UNWRAP_OR_FAIL(auto table,
@@ -334,14 +337,18 @@ TEST_F(SnapshotsTableTest, ScanEmptySummary) {
auto summaries =
std::static_pointer_cast<::arrow::MapArray>(batches.front()->column(5));
EXPECT_TRUE(summaries->IsNull(0));
+ auto manifests =
+
std::static_pointer_cast<::arrow::StringArray>(batches.front()->column(4));
+ EXPECT_TRUE(manifests->IsNull(0));
+ EXPECT_FALSE(manifests->IsNull(1));
EXPECT_FALSE(summaries->IsNull(1));
EXPECT_EQ(summaries->value_length(1), 0);
}
TEST_F(SnapshotsTableTest, ScanReturnsMultipleBatches) {
auto snapshot = MakeTestSnapshots().first;
- std::vector<std::shared_ptr<Snapshot>> snapshots(MetadataTable::kBatchSize +
1,
- snapshot);
+ std::vector<std::shared_ptr<Snapshot>> snapshots(
+ MetadataScanTaskReader::Options{}.batch_size + 1, snapshot);
ICEBERG_UNWRAP_OR_FAIL(auto table,
MakeTableWithSnapshots(std::move(snapshots),
/*current_snapshot_id=*/1));
ICEBERG_UNWRAP_OR_FAIL(auto snapshots_table,
@@ -350,14 +357,14 @@ TEST_F(SnapshotsTableTest, ScanReturnsMultipleBatches) {
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);
+ EXPECT_EQ(batches[0]->num_rows(),
MetadataScanTaskReader::Options{}.batch_size);
EXPECT_EQ(batches[1]->num_rows(), 1);
}
TEST_F(SnapshotsTableTest, ScanUsesMetadataCapturedAtCreation) {
auto snapshot = MakeTestSnapshots().first;
- std::vector<std::shared_ptr<Snapshot>> snapshots(MetadataTable::kBatchSize +
1,
- snapshot);
+ std::vector<std::shared_ptr<Snapshot>> snapshots(
+ MetadataScanTaskReader::Options{}.batch_size + 1, snapshot);
ICEBERG_UNWRAP_OR_FAIL(auto table,
MakeTableWithSnapshots(std::move(snapshots),
/*current_snapshot_id=*/1));
ICEBERG_UNWRAP_OR_FAIL(auto snapshots_table,
@@ -371,7 +378,8 @@ TEST_F(SnapshotsTableTest,
ScanUsesMetadataCapturedAtCreation) {
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);
+ EXPECT_EQ((*first_batch_result)->num_rows(),
+ MetadataScanTaskReader::Options{}.batch_size);
ICEBERG_UNWRAP_OR_FAIL(
auto refreshed_table,
@@ -387,4 +395,50 @@ TEST_F(SnapshotsTableTest,
ScanUsesMetadataCapturedAtCreation) {
EXPECT_EQ((*second_batch_result)->num_rows(), 1);
}
+TEST_F(SnapshotsTableTest, CanonicalSchema) {
+ auto expected = std::make_shared<Schema>(std::vector<SchemaField>{
+ SchemaField::MakeRequired(1, "committed_at", timestamp_tz()),
+ SchemaField::MakeRequired(2, "snapshot_id", int64()),
+ SchemaField::MakeOptional(3, "parent_id", int64()),
+ SchemaField::MakeOptional(4, "operation", string()),
+ SchemaField::MakeOptional(5, "manifest_list", string()),
+ SchemaField::MakeOptional(
+ 6, "summary",
+ std::make_shared<MapType>(SchemaField::MakeRequired(7, "key",
string()),
+ SchemaField::MakeRequired(8, "value",
string())))});
+ EXPECT_EQ(*SnapshotsTable::CanonicalSchema(), *expected);
+ EXPECT_EQ(*snapshots_table_->schema(), *expected);
+}
+
+TEST_F(SnapshotsTableTest,
PlannedTaskRetainsProjectionAndIndependentReadState) {
+ ICEBERG_UNWRAP_OR_FAIL(auto builder, snapshots_table_->NewScan());
+ builder->Select({"snapshot_id", "operation"});
+ ICEBERG_UNWRAP_OR_FAIL(auto scan, builder->Build());
+ ICEBERG_UNWRAP_OR_FAIL(auto tasks, scan->PlanTasks());
+ ASSERT_EQ(tasks.size(), 1);
+ ICEBERG_UNWRAP_OR_FAIL(auto reader,
MetadataScanTaskReader::Make({.batch_size = 1}));
+ ICEBERG_UNWRAP_OR_FAIL(auto first, reader->Open(*tasks.front()));
+ ICEBERG_UNWRAP_OR_FAIL(auto second, reader->Open(*tasks.front()));
+ // Streams retain their own input and cursors after the planning objects are
gone.
+ tasks.clear();
+ scan.reset();
+ reader.reset();
+ builder.reset();
+ snapshots_table_.reset();
+ table_.reset();
+ Schema selected({SchemaField::MakeRequired(2, "snapshot_id", int64()),
+ SchemaField::MakeOptional(4, "operation", string())});
+ ICEBERG_UNWRAP_OR_FAIL(auto expected, TableFromJson(selected, R"([
+ {"snapshot_id":1,"operation":"append"},
+ {"snapshot_id":2,"operation":"append"}
+ ])"));
+ ICEBERG_UNWRAP_OR_FAIL(auto actual, ReadArrowTable(std::move(first)));
+ EXPECT_TRUE(TablesEqual(*expected, *actual));
+ ASSERT_EQ(actual->column(0)->num_chunks(), 2);
+ EXPECT_EQ(actual->column(0)->chunk(0)->length(), 1);
+ EXPECT_EQ(actual->column(0)->chunk(1)->length(), 1);
+ ICEBERG_UNWRAP_OR_FAIL(auto repeated, ReadArrowTable(std::move(second)));
+ EXPECT_TRUE(TablesEqual(*expected, *repeated));
+}
+
} // namespace iceberg
diff --git a/src/iceberg/test/stream_test.cc b/src/iceberg/test/stream_test.cc
index 91b35216..dbb9248a 100644
--- a/src/iceberg/test/stream_test.cc
+++ b/src/iceberg/test/stream_test.cc
@@ -104,6 +104,43 @@ static_assert(std::is_move_assignable_v<CopyOnlyStream>);
static_assert(std::is_move_constructible_v<MoveOnlyStream>);
static_assert(std::is_move_assignable_v<MoveOnlyStream>);
+TEST(StreamTest, FromVectorPreservesOrderAndEndOfStream) {
+ auto stream = Stream<int>::FromVector({3, 1, 2});
+ ICEBERG_UNWRAP_OR_FAIL(auto first, stream->Next());
+ ASSERT_TRUE(first.has_value());
+ EXPECT_EQ(*first, 3);
+ ICEBERG_UNWRAP_OR_FAIL(auto remaining, stream->ToVector());
+ EXPECT_EQ(remaining, (std::vector<int>{1, 2}));
+ for (int i = 0; i < 2; ++i) {
+ ICEBERG_UNWRAP_OR_FAIL(auto end, stream->Next());
+ EXPECT_FALSE(end.has_value());
+ }
+ auto empty = Stream<int>::FromVector({});
+ ICEBERG_UNWRAP_OR_FAIL(auto end, empty->Next());
+ EXPECT_FALSE(end.has_value());
+}
+
+TEST(StreamTest, FromVectorSupportsBooleans) {
+ auto stream = Stream<bool>::FromVector({true, false, true});
+ ICEBERG_UNWRAP_OR_FAIL(auto values, stream->ToVector());
+ EXPECT_EQ(values, (std::vector<bool>{true, false, true}));
+}
+
+TEST(StreamTest, FromVectorSupportsMoveAndCopyOnlyValues) {
+ std::vector<std::unique_ptr<int>> values;
+ values.push_back(std::make_unique<int>(42));
+ auto stream = Stream<std::unique_ptr<int>>::FromVector(std::move(values));
+ ICEBERG_UNWRAP_OR_FAIL(auto result, stream->ToVector());
+ ASSERT_EQ(result.size(), 1);
+ EXPECT_EQ(*result.front(), 42);
+
+ std::vector<CopyOnly> copy_values(1, CopyOnly(42));
+ auto copy_stream = Stream<CopyOnly>::FromVector(std::move(copy_values));
+ ICEBERG_UNWRAP_OR_FAIL(auto copied, copy_stream->ToVector());
+ ASSERT_EQ(copied.size(), 1);
+ EXPECT_EQ(copied.front().value(), 42);
+}
+
TEST(StreamTest, SupportsIncompleteSharedPointerValue) {
class IncompleteType;
std::unique_ptr<Stream<std::shared_ptr<IncompleteType>>> stream;
diff --git a/src/iceberg/type_fwd.h b/src/iceberg/type_fwd.h
index 9386c9dd..c0230feb 100644
--- a/src/iceberg/type_fwd.h
+++ b/src/iceberg/type_fwd.h
@@ -19,6 +19,8 @@
#pragma once
+#include <cstdint>
+
/// \file iceberg/type_fwd.h
/// Forward declarations and enum definitions. When writing your own headers,
/// you can include this instead of the "full" headers to help reduce compile
@@ -280,9 +282,14 @@ class DeleteLoader;
class PositionDeleteIndex;
/// \brief Metadata tables.
+enum class MetadataTableKind : uint8_t;
+
class HistoryTable;
class MetadataScanBuilder;
class MetadataTable;
+class MetadataScan;
+class MetadataScanTask;
+class MetadataScanTaskReader;
class SnapshotsTable;
/// \brief Table encryption
diff --git a/src/iceberg/util/stream.h b/src/iceberg/util/stream.h
index fc01e98c..1dc3055a 100644
--- a/src/iceberg/util/stream.h
+++ b/src/iceberg/util/stream.h
@@ -22,7 +22,9 @@
/// \file iceberg/util/stream.h
/// \brief Pull-based stream interface for fallible, lazily produced values.
+#include <cstddef>
#include <deque>
+#include <memory>
#include <optional>
#include <type_traits>
#include <utility>
@@ -62,6 +64,9 @@ class Stream {
/// \brief Move-assign a stream and its terminal state.
Stream& operator=(Stream&&) noexcept = default;
+ /// \brief Create a stream that owns and consumes the given values in order.
+ static std::unique_ptr<Stream<T>> FromVector(std::vector<T> values);
+
/// \brief Return the next value, or std::nullopt when the stream is
exhausted.
///
/// After this method returns an error or std::nullopt, subsequent calls
return the same
@@ -136,4 +141,27 @@ class Stream {
std::optional<Error> error_;
};
+template <typename T>
+std::unique_ptr<Stream<T>> Stream<T>::FromVector(std::vector<T> values) {
+ class VectorStream final : public Stream<T> {
+ public:
+ explicit VectorStream(std::vector<T> values) : values_(std::move(values))
{}
+
+ protected:
+ Result<std::optional<T>> NextImpl() override {
+ if (index_ == values_.size()) {
+ return std::nullopt;
+ }
+ auto&& value = values_[index_++];
+ return Result<std::optional<T>>(std::in_place, std::in_place,
+ std::move_if_noexcept(value));
+ }
+
+ private:
+ std::vector<T> values_;
+ size_t index_ = 0;
+ };
+ return std::make_unique<VectorStream>(std::move(values));
+}
+
} // namespace iceberg
