shangxinli commented on code in PR #1066:
URL: https://github.com/apache/iceberg-cpp/pull/1066#discussion_r4404193373
##########
src/iceberg/inspect/snapshots_table.cc:
##########
@@ -19,164 +19,136 @@
#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_c_data_util_internal.h"
#include "iceberg/arrow_row_builder_internal.h"
-#include "iceberg/nanoarrow_status_internal.h"
+#include "iceberg/inspect/metadata_scan_task.h"
+#include "iceberg/inspect/metadata_scan_task_reader.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"
#include "iceberg/type.h"
#include "iceberg/util/macros.h"
namespace iceberg {
-namespace {
-
-Status AppendSnapshot(ArrowRowBuilder& builder, const Snapshot& snapshot) {
- ICEBERG_RETURN_UNEXPECTED(
- AppendInt(builder.column(0),
std::chrono::duration_cast<std::chrono::microseconds>(
-
snapshot.timestamp_ms.time_since_epoch())
- .count()));
- ICEBERG_RETURN_UNEXPECTED(AppendInt(builder.column(1),
snapshot.snapshot_id));
-
- 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));
-
- if (!snapshot.summary.empty()) {
- if (snapshot.summary.contains(SnapshotSummaryFields::kOperation)) {
- auto summary = snapshot.summary;
- summary.erase(SnapshotSummaryFields::kOperation);
- ICEBERG_RETURN_UNEXPECTED(AppendStringMap(builder.column(5), summary));
- } else {
- ICEBERG_RETURN_UNEXPECTED(AppendStringMap(builder.column(5),
snapshot.summary));
- }
- } else {
- ICEBERG_RETURN_UNEXPECTED(AppendNull(builder.column(5)));
- }
-
- return builder.FinishRow();
-}
-
-class SnapshotsTableStream {
- public:
- static Result<std::unique_ptr<SnapshotsTableStream>> Make(
- std::shared_ptr<const TableMetadata> metadata, const iceberg::Schema&
schema) {
- ArrowSchema arrow_schema{};
- ICEBERG_RETURN_UNEXPECTED(ToArrowSchema(schema, &arrow_schema));
- return std::unique_ptr<SnapshotsTableStream>(
- new SnapshotsTableStream(std::move(metadata),
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(AppendSnapshot(builder, *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,
- ArrowSchema arrow_schema)
- : metadata_(std::move(metadata)), arrow_schema_(std::move(arrow_schema))
{}
-
- std::shared_ptr<const TableMetadata> metadata_;
- ArrowSchema arrow_schema_{};
- size_t next_snapshot_ = 0;
-};
-
-} // 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::Scan() {
+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]) {
Review Comment:
nit: no test exercises this skip when an entry in metadata->snapshots is
null.
##########
src/iceberg/inspect/snapshots_table.cc:
##########
@@ -19,164 +19,136 @@
#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_c_data_util_internal.h"
#include "iceberg/arrow_row_builder_internal.h"
-#include "iceberg/nanoarrow_status_internal.h"
+#include "iceberg/inspect/metadata_scan_task.h"
+#include "iceberg/inspect/metadata_scan_task_reader.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"
#include "iceberg/type.h"
#include "iceberg/util/macros.h"
namespace iceberg {
-namespace {
-
-Status AppendSnapshot(ArrowRowBuilder& builder, const Snapshot& snapshot) {
- ICEBERG_RETURN_UNEXPECTED(
- AppendInt(builder.column(0),
std::chrono::duration_cast<std::chrono::microseconds>(
-
snapshot.timestamp_ms.time_since_epoch())
- .count()));
- ICEBERG_RETURN_UNEXPECTED(AppendInt(builder.column(1),
snapshot.snapshot_id));
-
- 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));
-
- if (!snapshot.summary.empty()) {
- if (snapshot.summary.contains(SnapshotSummaryFields::kOperation)) {
- auto summary = snapshot.summary;
- summary.erase(SnapshotSummaryFields::kOperation);
- ICEBERG_RETURN_UNEXPECTED(AppendStringMap(builder.column(5), summary));
- } else {
- ICEBERG_RETURN_UNEXPECTED(AppendStringMap(builder.column(5),
snapshot.summary));
- }
- } else {
- ICEBERG_RETURN_UNEXPECTED(AppendNull(builder.column(5)));
- }
-
- return builder.FinishRow();
-}
-
-class SnapshotsTableStream {
- public:
- static Result<std::unique_ptr<SnapshotsTableStream>> Make(
- std::shared_ptr<const TableMetadata> metadata, const iceberg::Schema&
schema) {
- ArrowSchema arrow_schema{};
- ICEBERG_RETURN_UNEXPECTED(ToArrowSchema(schema, &arrow_schema));
- return std::unique_ptr<SnapshotsTableStream>(
- new SnapshotsTableStream(std::move(metadata),
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(AppendSnapshot(builder, *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,
- ArrowSchema arrow_schema)
- : metadata_(std::move(metadata)), arrow_schema_(std::move(arrow_schema))
{}
-
- std::shared_ptr<const TableMetadata> metadata_;
- ArrowSchema arrow_schema_{};
- size_t next_snapshot_ = 0;
-};
-
-} // 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::Scan() {
+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(
+ CanonicalSchema(), projected_schema, num_rows, batch_size,
+ [metadata = std::move(metadata), snapshot_indices =
std::move(snapshot_indices)](
+ ArrowArray* column, int32_t field_id, size_t row) -> Status {
+ const auto& snapshot = *metadata->snapshots[snapshot_indices[row]];
+ switch (field_id) {
+ case 1: // committed_at
+ return AppendInt(column,
UnixMsFromTimePointMs(snapshot.timestamp_ms) * 1000);
+ case 2: // snapshot_id
+ return AppendInt(column, snapshot.snapshot_id);
+ case 3: // parent_id
+ return snapshot.parent_snapshot_id
+ ? AppendInt(column, *snapshot.parent_snapshot_id)
+ : AppendNull(column);
+ case 4: { // operation
+ auto operation = snapshot.Operation();
+ return operation ? AppendString(column, *operation) :
AppendNull(column);
+ }
+ case 5: // manifest_list
+ return snapshot.manifest_list.empty()
Review Comment:
nit: no test exercises the empty manifest_list branch here.
##########
src/iceberg/inspect/history_table.cc:
##########
@@ -45,13 +56,94 @@ 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::Scan() {
- return NotSupported("Scan is not supported for the history table");
+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 && ancestors.insert(current).second;) {
Review Comment:
nit: the cycle guard here isn't tested against an actual cycle in the
snapshot graph, so it's defensive-only right now.
--
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]
