WZhuo commented on code in PR #801:
URL: https://github.com/apache/iceberg-cpp/pull/801#discussion_r3849726675
##########
src/iceberg/inspect/metadata_table.h:
##########
@@ -20,46 +20,108 @@
#pragma once
/// \file iceberg/inspect/metadata_table.h
-/// \brief Define base APIs for metadata tables.
+/// \brief Base APIs for inspecting Iceberg metadata tables.
+#include <concepts>
#include <memory>
+#include <string>
+#include <utility>
+#include <variant>
+#include "iceberg/arrow_c_data.h"
#include "iceberg/iceberg_export.h"
#include "iceberg/result.h"
-#include "iceberg/table_identifier.h"
#include "iceberg/type_fwd.h"
+#include "iceberg/util/timepoint.h"
namespace iceberg {
-/// \brief Base class for Iceberg metadata tables.
+/// \brief Base interface for an Iceberg metadata table.
class ICEBERG_EXPORT MetadataTable {
public:
+ /// \brief Supported metadata table kinds.
enum class Kind {
kSnapshots,
kHistory,
};
- static Result<std::unique_ptr<MetadataTable>> Make(std::shared_ptr<Table>
table,
- Kind kind);
+ /// \brief Maximum number of rows emitted in each Arrow batch.
+ static constexpr int64_t kBatchSize = 1024;
+
+ /// \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.
+ template <typename MetadataTableType>
+ requires std::derived_from<MetadataTableType, MetadataTable>
+ static Result<std::unique_ptr<MetadataTableType>>
Make(std::shared_ptr<Table> table) {
+ return MetadataTableType::Make(std::move(table));
+ }
virtual ~MetadataTable();
+ /// \brief Return this metadata table's kind.
virtual Kind kind() const noexcept = 0;
- const TableIdentifier& name() const { return identifier_; }
+ /// \brief Return the schema of rows emitted by scans.
+ 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<Schema>& schema() const { return schema_; }
+ /// \brief Return whether this metadata table supports time travel.
+ virtual bool supports_time_travel() const noexcept;
- const std::shared_ptr<Table>& source_table() const { return source_table_; }
+ /// \brief Scan the metadata table without time travel.
+ ///
+ /// The caller owns the returned stream and must release it with
+ /// ArrowArrayStreamRelease.
+ virtual Result<ArrowArrayStream> Scan() = 0;
protected:
- explicit MetadataTable(std::shared_ptr<Table> source_table, TableIdentifier
identifier,
- std::shared_ptr<Schema> schema);
+ explicit MetadataTable(std::shared_ptr<Table> source_table);
private:
- TableIdentifier identifier_;
- std::shared_ptr<Schema> schema_;
std::shared_ptr<Table> source_table_;
};
+/// \brief Snapshot selection parameters for a time-travel scan.
+struct SnapshotSelection {
+ /// \brief Select the current snapshot, a snapshot ID, or an as-of timestamp.
+ ///
+ /// std::monostate selects the current snapshot.
+ std::variant<std::monostate, int64_t, TimePointMs> snapshot;
+
+ /// \brief Resolve the snapshot relative to this branch or tag.
+ ///
+ /// An empty string uses the main branch.
+ std::string ref_name;
+};
+
+/// \brief Base interface for metadata tables that support time travel.
+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;
+
+ /// \brief Scan using the current snapshot on the main branch.
+ Result<ArrowArrayStream> Scan() final;
+
+ /// \brief Scan using the requested snapshot selection.
+ ///
+ /// \param snapshot_selection Snapshot ID, timestamp, and optional ref
selection.
+ /// \return An Arrow stream containing the metadata table rows, or an error.
+ Result<ArrowArrayStream> Scan(const SnapshotSelection& snapshot_selection);
Review Comment:
Keeping `TimeTravelMetadataTable` is intentional. It keeps `Scan(const
SnapshotSelection&)` out of `MetadataTable` and `SnapshotsTable`, so metadata
tables without time-travel support cannot expose or accept snapshot selectors.
`supports_time_travel()` remains available for capability introspection.
##########
src/iceberg/inspect/metadata_table.h:
##########
@@ -20,46 +20,108 @@
#pragma once
/// \file iceberg/inspect/metadata_table.h
-/// \brief Define base APIs for metadata tables.
+/// \brief Base APIs for inspecting Iceberg metadata tables.
+#include <concepts>
#include <memory>
+#include <string>
+#include <utility>
+#include <variant>
+#include "iceberg/arrow_c_data.h"
#include "iceberg/iceberg_export.h"
#include "iceberg/result.h"
-#include "iceberg/table_identifier.h"
#include "iceberg/type_fwd.h"
+#include "iceberg/util/timepoint.h"
namespace iceberg {
-/// \brief Base class for Iceberg metadata tables.
+/// \brief Base interface for an Iceberg metadata table.
class ICEBERG_EXPORT MetadataTable {
public:
+ /// \brief Supported metadata table kinds.
enum class Kind {
kSnapshots,
kHistory,
};
- static Result<std::unique_ptr<MetadataTable>> Make(std::shared_ptr<Table>
table,
- Kind kind);
+ /// \brief Maximum number of rows emitted in each Arrow batch.
+ static constexpr int64_t kBatchSize = 1024;
+
+ /// \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.
+ template <typename MetadataTableType>
+ requires std::derived_from<MetadataTableType, MetadataTable>
+ static Result<std::unique_ptr<MetadataTableType>>
Make(std::shared_ptr<Table> table) {
+ return MetadataTableType::Make(std::move(table));
+ }
virtual ~MetadataTable();
+ /// \brief Return this metadata table's kind.
virtual Kind kind() const noexcept = 0;
- const TableIdentifier& name() const { return identifier_; }
+ /// \brief Return the schema of rows emitted by scans.
+ 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<Schema>& schema() const { return schema_; }
+ /// \brief Return whether this metadata table supports time travel.
+ virtual bool supports_time_travel() const noexcept;
- const std::shared_ptr<Table>& source_table() const { return source_table_; }
+ /// \brief Scan the metadata table without time travel.
+ ///
+ /// The caller owns the returned stream and must release it with
+ /// ArrowArrayStreamRelease.
Review Comment:
Updated the ownership documentation to state that callers must invoke the
returned stream's `release` callback. `ArrowArrayStreamRelease` is not part of
this repository's Arrow C Data API.
##########
src/iceberg/inspect/snapshots_table.cc:
##########
@@ -19,21 +19,133 @@
#include "iceberg/inspect/snapshots_table.h"
+#include <chrono>
+#include <cstddef>
#include <memory>
+#include <optional>
#include <utility>
#include <vector>
+#include <nanoarrow/nanoarrow.h>
+
+#include "iceberg/arrow/nanoarrow_status_internal.h"
+#include "iceberg/arrow_c_data_util_internal.h"
+#include "iceberg/arrow_row_builder_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_identifier.h"
#include "iceberg/type.h"
+#include "iceberg/util/macros.h"
namespace iceberg {
namespace {
-std::shared_ptr<Schema> MakeSnapshotsTableSchema() {
- return std::make_shared<Schema>(std::vector<SchemaField>{
+Status AppendSnapshot(ArrowRowBuilder& builder, const Snapshot& snapshot) {
+ ICEBERG_RETURN_UNEXPECTED(
+ AppendInt(builder.column(0),
std::chrono::duration_cast<std::chrono::microseconds>(
+
snapshot.timestamp_ms.time_since_epoch())
+ .count()));
+ ICEBERG_RETURN_UNEXPECTED(AppendInt(builder.column(1),
snapshot.snapshot_id));
+
+ if (snapshot.parent_snapshot_id.has_value()) {
+ ICEBERG_RETURN_UNEXPECTED(AppendInt(builder.column(2),
*snapshot.parent_snapshot_id));
+ } else {
+ ICEBERG_RETURN_UNEXPECTED(AppendNull(builder.column(2)));
+ }
+
+ auto operation = snapshot.Operation();
+ if (operation.has_value()) {
+ ICEBERG_RETURN_UNEXPECTED(AppendString(builder.column(3), *operation));
+ } else {
+ ICEBERG_RETURN_UNEXPECTED(AppendNull(builder.column(3)));
+ }
+
+ ICEBERG_RETURN_UNEXPECTED(AppendString(builder.column(4),
snapshot.manifest_list));
+
+ auto summary = snapshot.summary;
+ summary.erase(SnapshotSummaryFields::kOperation);
+ if (summary.empty()) {
+ ICEBERG_RETURN_UNEXPECTED(AppendNull(builder.column(5)));
+ } else {
+ ICEBERG_RETURN_UNEXPECTED(AppendStringMap(builder.column(5), summary));
+ }
+
+ return builder.FinishRow();
+}
+
+class SnapshotsTableStream {
+ public:
+ static Result<std::unique_ptr<SnapshotsTableStream>> Make(
+ std::shared_ptr<Table> table, const iceberg::Schema& schema) {
+ ArrowSchema arrow_schema{};
+ ICEBERG_RETURN_UNEXPECTED(ToArrowSchema(schema, &arrow_schema));
+ return std::unique_ptr<SnapshotsTableStream>(
+ new SnapshotsTableStream(std::move(table), std::move(arrow_schema)));
+ }
+
+ ~SnapshotsTableStream() { std::ignore = Close(); }
+
+ Status Close() {
+ table_.reset();
+ if (arrow_schema_.release != nullptr) {
+ ArrowSchemaRelease(&arrow_schema_);
+ }
+ return {};
+ }
+
+ Result<std::optional<ArrowArray>> Next() {
+ const auto& snapshots = table_->snapshots();
Review Comment:
Fixed by capturing a `shared_ptr<const TableMetadata>` when `Scan()` creates
the stream. Subsequent batches now use the same metadata snapshot even if the
source `Table` is refreshed. Added a test covering refresh between batches.
##########
src/iceberg/inspect/history_table.cc:
##########
@@ -26,37 +26,32 @@
#include "iceberg/schema.h"
#include "iceberg/schema_field.h"
#include "iceberg/table.h"
-#include "iceberg/table_identifier.h"
#include "iceberg/type.h"
+#include "iceberg/util/macros.h"
namespace iceberg {
-namespace {
-std::shared_ptr<Schema> MakeHistoryTableSchema() {
- return std::make_shared<Schema>(std::vector<SchemaField>{
+HistoryTable::HistoryTable(std::shared_ptr<Table> table)
+ : MetadataTable(std::move(table)) {}
+
+HistoryTable::~HistoryTable() = default;
+
+const std::shared_ptr<Schema>& HistoryTable::schema() const {
+ 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()),
SchemaField::MakeOptional(3, "parent_id", int64()),
SchemaField::MakeRequired(4, "is_current_ancestor", boolean())});
+ return schema;
}
-TableIdentifier MakeHistoryTableName(const TableIdentifier& source_name) {
- return TableIdentifier{.ns = source_name.ns, .name = source_name.name +
".history"};
-}
-
-} // namespace
-
-HistoryTable::HistoryTable(std::shared_ptr<Table> table)
- : MetadataTable(table, MakeHistoryTableName(table->name()),
- MakeHistoryTableSchema()) {}
-
-HistoryTable::~HistoryTable() = default;
-
Result<std::unique_ptr<HistoryTable>>
HistoryTable::Make(std::shared_ptr<Table> table) {
- if (table == nullptr) [[unlikely]] {
- return InvalidArgument("Table cannot be null");
- }
+ ICEBERG_PRECHECK(table != nullptr, "Table cannot be null");
return std::unique_ptr<HistoryTable>(new HistoryTable(std::move(table)));
}
+Result<ArrowArrayStream> HistoryTable::Scan() {
Review Comment:
HistoryTable scanning is intentionally deferred to the next PR, as
requested. It remains constructible but returns `NotSupported` for `Scan()` in
this change.
##########
src/iceberg/inspect/snapshots_table.cc:
##########
@@ -19,21 +19,133 @@
#include "iceberg/inspect/snapshots_table.h"
+#include <chrono>
+#include <cstddef>
#include <memory>
+#include <optional>
#include <utility>
#include <vector>
+#include <nanoarrow/nanoarrow.h>
+
+#include "iceberg/arrow/nanoarrow_status_internal.h"
+#include "iceberg/arrow_c_data_util_internal.h"
+#include "iceberg/arrow_row_builder_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_identifier.h"
#include "iceberg/type.h"
+#include "iceberg/util/macros.h"
namespace iceberg {
namespace {
-std::shared_ptr<Schema> MakeSnapshotsTableSchema() {
- return std::make_shared<Schema>(std::vector<SchemaField>{
+Status AppendSnapshot(ArrowRowBuilder& builder, const Snapshot& snapshot) {
+ ICEBERG_RETURN_UNEXPECTED(
+ AppendInt(builder.column(0),
std::chrono::duration_cast<std::chrono::microseconds>(
+
snapshot.timestamp_ms.time_since_epoch())
+ .count()));
+ ICEBERG_RETURN_UNEXPECTED(AppendInt(builder.column(1),
snapshot.snapshot_id));
+
+ if (snapshot.parent_snapshot_id.has_value()) {
+ ICEBERG_RETURN_UNEXPECTED(AppendInt(builder.column(2),
*snapshot.parent_snapshot_id));
+ } else {
+ ICEBERG_RETURN_UNEXPECTED(AppendNull(builder.column(2)));
+ }
+
+ auto operation = snapshot.Operation();
+ if (operation.has_value()) {
+ ICEBERG_RETURN_UNEXPECTED(AppendString(builder.column(3), *operation));
+ } else {
+ ICEBERG_RETURN_UNEXPECTED(AppendNull(builder.column(3)));
+ }
+
+ ICEBERG_RETURN_UNEXPECTED(AppendString(builder.column(4),
snapshot.manifest_list));
+
+ auto summary = snapshot.summary;
+ summary.erase(SnapshotSummaryFields::kOperation);
+ if (summary.empty()) {
+ ICEBERG_RETURN_UNEXPECTED(AppendNull(builder.column(5)));
+ } else {
+ ICEBERG_RETURN_UNEXPECTED(AppendStringMap(builder.column(5), summary));
+ }
+
+ return builder.FinishRow();
+}
+
+class SnapshotsTableStream {
+ public:
+ static Result<std::unique_ptr<SnapshotsTableStream>> Make(
+ std::shared_ptr<Table> table, const iceberg::Schema& schema) {
+ ArrowSchema arrow_schema{};
+ ICEBERG_RETURN_UNEXPECTED(ToArrowSchema(schema, &arrow_schema));
+ return std::unique_ptr<SnapshotsTableStream>(
+ new SnapshotsTableStream(std::move(table), std::move(arrow_schema)));
+ }
+
+ ~SnapshotsTableStream() { std::ignore = Close(); }
Review Comment:
Kept `std::ignore = Close()` and added the direct `<tuple>` include.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
