WZhuo commented on code in PR #1066:
URL: https://github.com/apache/iceberg-cpp/pull/1066#discussion_r4408381252
##########
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()
+ ? AppendNull(column)
+ : AppendString(column, snapshot.manifest_list);
+ case 6: { // summary; operation is exposed as its own column
+ if (snapshot.summary.empty()) {
+ return AppendNull(column);
+ }
+ auto summary = snapshot.summary;
+ summary.erase(SnapshotSummaryFields::kOperation);
+ return AppendStringMap(column, summary);
+ }
+ default:
+ return InvalidArgument("Unknown snapshots metadata column id: {}",
field_id);
+ }
+ });
+}
+
+// SnapshotsScan implementation
+SnapshotsScan::SnapshotsScan(internal::MetadataScanContext context,
+ std::shared_ptr<StaticMetadataTask> 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");
+ 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));
Review Comment:
Could we pass `projected_schema` into `StaticMetadataTask::Make()` and
retain it in the task, rather than passing the entire scan context or requiring
the reader to receive the projection separately?
Currently, `Select({"snapshot_id"})` produces one column through
`ToArrowArrayStream()`, but `PlanTasks()` followed by
`MetadataScanTaskReader::Make({.batch_size = 1})` and `Open()` produces all six
columns. The task drops the projection, and the reader defaults to the
canonical schema. The history scan has the same issue.
The projection is part of the planned work, so the task should carry it.
`MetadataScanTaskReader::Open()` could then read the projection from the task,
and `Options::projected_schema` could be removed, leaving the reader
responsible only for execution settings such as batch size. This also avoids
requiring callers to supply the same projection twice.
Please add a regression test that selects a subset of columns, plans tasks,
and opens them with a reader configured only with a custom batch size,
verifying that the selected schema is preserved.
--
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]
