manuzhang commented on code in PR #1066:
URL: https://github.com/apache/iceberg-cpp/pull/1066#discussion_r4389385520
##########
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;) {
+ auto it = snapshots_by_id.find(current);
+ if (it == snapshots_by_id.end() || !it->second->parent_snapshot_id) {
+ break;
+ }
+ current = *it->second->parent_snapshot_id;
+ }
+
+ const size_t num_rows = metadata->snapshot_log.size();
+ return internal::MakeMetadataBatchStream(
+ CanonicalSchema(), projected_schema, num_rows, batch_size,
+ [metadata = std::move(metadata), snapshots_by_id =
std::move(snapshots_by_id),
+ ancestors = std::move(ancestors)](ArrowArray* column, int32_t field_id,
+ size_t row) -> Status {
+ const auto& entry = metadata->snapshot_log[row];
+ switch (field_id) {
+ case 1: // made_current_at
+ return AppendInt(column, UnixMsFromTimePointMs(entry.timestamp_ms)
* 1000);
+ case 2: // snapshot_id
+ return AppendInt(column, entry.snapshot_id);
+ case 3: { // parent_id; null when the snapshot has been expired
+ auto it = snapshots_by_id.find(entry.snapshot_id);
+ return it != snapshots_by_id.end() &&
it->second->parent_snapshot_id
+ ? AppendInt(column, *it->second->parent_snapshot_id)
+ : AppendNull(column);
+ }
+ case 4: // is_current_ancestor
+ return AppendBoolean(column,
ancestors.contains(entry.snapshot_id));
+ default:
+ return InvalidArgument("Unknown history metadata column id: {}",
field_id);
+ }
+ });
+}
+
+// HistoryScan implementation
+HistoryScan::HistoryScan(internal::MetadataScanContext context,
+ std::shared_ptr<StaticMetadataTask> 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");
+ 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_ASSIGN_OR_RAISE(auto task,
+ StaticMetadataTask::Make(context.kind,
context.table_metadata));
+ return std::unique_ptr<HistoryScan>(
+ new HistoryScan(std::move(context), std::move(task)));
+}
+
+Result<MetadataScanTaskStreamPtr> HistoryScan::PlanTasksStream() const {
Review Comment:
The follow methods look same as SnapshotTable's. Can they be moved to
superclass?
--
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]
