wgtmac commented on code in PR #801:
URL: https://github.com/apache/iceberg-cpp/pull/801#discussion_r3789268260
##########
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:
Agreed. Please add <tuple> or avoid std::ignore here so this file does not
depend on transitive includes.
##########
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 remains constructible, but the new common Scan() API always
returns NotSupported here. Java HistoryTable scans table.history() into rows.
Please implement a streaming history scan, or keep non-scannable tables out of
this interface.
##########
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:
This stream re-reads the mutable Table on every batch. If Table::Refresh()
runs between get_next calls, later batches can come from a different snapshot
list, skipping or truncating rows. Please capture the metadata or snapshot list
when Scan() creates the stream.
##########
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:
ArrowArrayStreamRelease is not declared by this repository's Arrow C Data
API; callers must invoke the stream's release callback. Please correct the
ownership documentation.
--
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]
