You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@iceberg.apache.org by "manuzhang (via GitHub)" <gi...@apache.org> on 2026/08/06 09:54:13 UTC
[PR] feat(inspect): add system metadata tables [iceberg-cpp]
manuzhang opened a new pull request, #876:
URL: https://github.com/apache/iceberg-cpp/pull/876
## Summary
Add streaming system metadata tables for branches, tags, files, partitions, and manifests.
## Changes
- expose typed metadata-table kinds and public table classes
- add shared Arrow stream and metadata-table utilities
- implement snapshot-aware files, partitions, and manifests scans
- implement branch and tag reference scans
- wire sources and tests into CMake and Meson
- add coverage for schemas, values, partition evolution, and batching behavior
## Why
This makes Iceberg system metadata available through the C++ metadata-table API, using bounded Arrow batches and behavior aligned with Apache Iceberg's reference implementation.
## Impact
Consumers can inspect table references, live files, partition aggregates, and manifest metadata without implementing manifest traversal or Arrow materialization themselves.
## Dependency
This branch is based on #801, which provides the metadata-table streaming APIs used here. Because this PR targets `main`, its diff includes that prerequisite history until #801 is merged.
## Validation
- Independent pre-commit review completed with no actionable findings.
- `git diff --cached --check` passed before commit.
- Local tests were not run, per repository `AGENTS.md` instructions.
- `clang-format` validation was unavailable because the binary is not installed in the environment.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat(inspect): add system metadata tables [iceberg-cpp]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #876:
URL: https://github.com/apache/iceberg-cpp/pull/876#discussion_r4025858958
##########
src/iceberg/inspect/files_table.cc:
##########
@@ -0,0 +1,75 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+#include "iceberg/inspect/files_table.h"
+
+#include <memory>
+#include <utility>
+
+#include "iceberg/inspect/metadata_table_stream_internal.h"
+#include "iceberg/inspect/metadata_table_util_internal.h"
+#include "iceberg/schema.h"
+#include "iceberg/table.h"
+#include "iceberg/type.h"
+#include "iceberg/util/macros.h"
+
+namespace iceberg {
+
+FilesTable::FilesTable(std::shared_ptr<Table> table, std::shared_ptr<Schema> schema,
+ std::shared_ptr<Schema> table_schema,
+ std::shared_ptr<StructType> partition_type)
+ : TimeTravelMetadataTable(std::move(table)),
+ schema_(std::move(schema)),
+ table_schema_(std::move(table_schema)),
+ partition_type_(std::move(partition_type)) {}
+
+FilesTable::~FilesTable() = default;
+
+const std::shared_ptr<Schema>& FilesTable::schema() const { return schema_; }
+
+Result<std::unique_ptr<FilesTable>> FilesTable::Make(std::shared_ptr<Table> table) {
+ ICEBERG_PRECHECK(table != nullptr, "Table cannot be null");
+ ICEBERG_ASSIGN_OR_RAISE(auto table_schema, table->schema());
+ ICEBERG_ASSIGN_OR_RAISE(auto partition_type, internal::UnifiedPartitionType(*table));
+ ICEBERG_ASSIGN_OR_RAISE(auto schema,
+ internal::FilesTableSchema(*table_schema, partition_type));
+ return std::unique_ptr<FilesTable>(new FilesTable(std::move(table), std::move(schema),
Review Comment:
These derived schemas are computed only once, while `source_table()` is the same mutable `Table` object that `Table::Refresh()` updates in place. After a refresh, `ScanSnapshot` can resolve the new snapshot/manifests but still serializes rows with the old table schema and partition type (and `schema()` remains stale), which can omit new fields or misdecode changed partition/metric types. Please make these derived values refresh-aware (or rebuild them per scan) and cover a refresh before scanning.
##########
src/iceberg/inspect/partitions_table.cc:
##########
@@ -0,0 +1,316 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+#include "iceberg/inspect/partitions_table.h"
+
+#include <algorithm>
+#include <chrono>
+#include <cstddef>
+#include <cstdint>
+#include <memory>
+#include <optional>
+#include <ranges>
+#include <unordered_map>
+#include <utility>
+#include <vector>
+
+#include "iceberg/arrow_row_builder_internal.h"
+#include "iceberg/expression/literal.h"
+#include "iceberg/inspect/metadata_table_stream_internal.h"
+#include "iceberg/inspect/metadata_table_util_internal.h"
+#include "iceberg/manifest/manifest_entry.h"
+#include "iceberg/partition_spec.h"
+#include "iceberg/row/partition_values.h"
+#include "iceberg/schema.h"
+#include "iceberg/schema_field.h"
+#include "iceberg/snapshot.h"
+#include "iceberg/table.h"
+#include "iceberg/type.h"
+#include "iceberg/util/macros.h"
+
+namespace iceberg {
+namespace {
+
+constexpr int32_t kPartitionFieldId = 1;
+constexpr int32_t kRecordCountFieldId = 2;
+constexpr int32_t kFileCountFieldId = 3;
+constexpr int32_t kSpecIdFieldId = 4;
+constexpr int32_t kPositionDeleteRecordCountFieldId = 5;
+constexpr int32_t kPositionDeleteFileCountFieldId = 6;
+constexpr int32_t kEqualityDeleteRecordCountFieldId = 7;
+constexpr int32_t kEqualityDeleteFileCountFieldId = 8;
+constexpr int32_t kLastUpdatedAtFieldId = 9;
+constexpr int32_t kLastUpdatedSnapshotIdFieldId = 10;
+constexpr int32_t kTotalDataFileSizeFieldId = 11;
+
+struct PartitionKey {
+ PartitionValues values;
+ size_t projected_fields;
+
+ bool operator==(const PartitionKey& other) const {
+ if (projected_fields != other.projected_fields ||
+ values.num_fields() != other.values.num_fields()) {
+ return false;
+ }
+ for (size_t index = 0; index < values.num_fields(); ++index) {
+ const auto& lhs = values.values()[index];
+ const auto& rhs = other.values.values()[index];
+ if (lhs.IsNull() || rhs.IsNull()) {
+ if (lhs.IsNull() != rhs.IsNull()) {
+ return false;
+ }
+ } else if (lhs.IsNaN() && rhs.IsNaN()) {
+ // Java partition equality canonicalizes all NaN signs and payloads.
+ continue;
+ } else if (lhs != rhs) {
+ return false;
+ }
+ }
+ return true;
+ }
+};
+
+struct PartitionKeyHash {
+ size_t operator()(const PartitionKey& key) const noexcept {
+ size_t result = 17;
+ for (const auto& value : key.values.values()) {
+ const size_t value_hash =
+ value.IsNaN() ? 0x7ff8000000000000ULL : LiteralHash{}(value);
+ result = result * 37 + value_hash;
+ }
+ return result * 37 + key.projected_fields;
+ }
+};
+
+size_t ProjectedFieldCount(const StructType& partition_type, const PartitionSpec& spec) {
+ size_t count = 0;
+ for (const auto& field : partition_type.fields()) {
+ count += std::ranges::any_of(
+ spec.fields(), [field_id = field.field_id()](const PartitionField& spec_field) {
+ return spec_field.field_id() == field_id;
+ });
+ }
+ return count;
+}
+
+struct PartitionStats {
+ explicit PartitionStats(PartitionValues values) : partition(std::move(values)) {}
+
+ PartitionValues partition;
+ int32_t spec_id = PartitionSpec::kInitialSpecId;
+ int64_t data_record_count = 0;
+ int32_t data_file_count = 0;
+ int64_t data_file_size = 0;
+ int64_t position_delete_record_count = 0;
+ int32_t position_delete_file_count = 0;
+ int64_t equality_delete_record_count = 0;
+ int32_t equality_delete_file_count = 0;
+ std::optional<TimePointMs> last_updated_at;
+ std::optional<int64_t> last_updated_snapshot_id;
+};
+
+Status AppendPartition(ArrowRowBuilder& builder, const Schema& schema,
+ const StructType& partition_type,
+ const PartitionStats& partition) {
+ for (size_t index = 0; index < schema.fields().size(); ++index) {
+ auto* array = builder.column(index);
+ switch (schema.fields()[index].field_id()) {
+ case kPartitionFieldId:
+ ICEBERG_RETURN_UNEXPECTED(
+ internal::AppendPartitionValues(array, partition_type, partition.partition));
+ break;
+ case kSpecIdFieldId:
+ ICEBERG_RETURN_UNEXPECTED(AppendInt(array, partition.spec_id));
+ break;
+ case kRecordCountFieldId:
+ ICEBERG_RETURN_UNEXPECTED(AppendInt(array, partition.data_record_count));
+ break;
+ case kFileCountFieldId:
+ ICEBERG_RETURN_UNEXPECTED(AppendInt(array, partition.data_file_count));
+ break;
+ case kTotalDataFileSizeFieldId:
+ ICEBERG_RETURN_UNEXPECTED(AppendInt(array, partition.data_file_size));
+ break;
+ case kPositionDeleteRecordCountFieldId:
+ ICEBERG_RETURN_UNEXPECTED(
+ AppendInt(array, partition.position_delete_record_count));
+ break;
+ case kPositionDeleteFileCountFieldId:
+ ICEBERG_RETURN_UNEXPECTED(AppendInt(array, partition.position_delete_file_count));
+ break;
+ case kEqualityDeleteRecordCountFieldId:
+ ICEBERG_RETURN_UNEXPECTED(
+ AppendInt(array, partition.equality_delete_record_count));
+ break;
+ case kEqualityDeleteFileCountFieldId:
+ ICEBERG_RETURN_UNEXPECTED(AppendInt(array, partition.equality_delete_file_count));
+ break;
+ case kLastUpdatedAtFieldId:
+ if (partition.last_updated_at.has_value()) {
+ ICEBERG_RETURN_UNEXPECTED(
+ AppendInt(array, std::chrono::duration_cast<std::chrono::microseconds>(
+ partition.last_updated_at->time_since_epoch())
+ .count()));
+ } else {
+ ICEBERG_RETURN_UNEXPECTED(AppendNull(array));
+ }
+ break;
+ case kLastUpdatedSnapshotIdFieldId:
+ if (partition.last_updated_snapshot_id.has_value()) {
+ ICEBERG_RETURN_UNEXPECTED(
+ AppendInt(array, *partition.last_updated_snapshot_id));
+ } else {
+ ICEBERG_RETURN_UNEXPECTED(AppendNull(array));
+ }
+ break;
+ default:
+ return InvalidSchema("Unsupported partitions metadata field {}",
+ schema.fields()[index].field_id());
+ }
+ }
+ return builder.FinishRow();
+}
+
+void UpdateCounts(PartitionStats& partition, const DataFile& file) {
+ switch (file.content) {
+ case DataFile::Content::kData:
+ partition.data_record_count += file.record_count;
+ ++partition.data_file_count;
+ partition.data_file_size += file.file_size_in_bytes;
+ break;
+ case DataFile::Content::kPositionDeletes:
+ partition.position_delete_record_count += file.record_count;
+ ++partition.position_delete_file_count;
+ break;
+ case DataFile::Content::kEqualityDeletes:
+ partition.equality_delete_record_count += file.record_count;
+ ++partition.equality_delete_file_count;
+ break;
+ }
+}
+
+} // namespace
+
+PartitionsTable::PartitionsTable(std::shared_ptr<Table> table,
+ std::shared_ptr<Schema> schema,
+ std::shared_ptr<StructType> partition_type)
+ : TimeTravelMetadataTable(std::move(table)),
+ schema_(std::move(schema)),
+ partition_type_(std::move(partition_type)) {}
+
+PartitionsTable::~PartitionsTable() = default;
+
+const std::shared_ptr<Schema>& PartitionsTable::schema() const { return schema_; }
+
+Result<std::unique_ptr<PartitionsTable>> PartitionsTable::Make(
+ std::shared_ptr<Table> table) {
+ ICEBERG_PRECHECK(table != nullptr, "Table cannot be null");
+ ICEBERG_ASSIGN_OR_RAISE(auto partition_type, internal::UnifiedPartitionType(*table));
+
+ std::vector<SchemaField> fields;
+ if (!partition_type->fields().empty()) {
+ fields.push_back(
+ SchemaField::MakeRequired(kPartitionFieldId, "partition", partition_type));
Review Comment:
The partition schema and unified partition type are cached only during construction, even though scans use the refreshed `source_table()` metadata. If the caller refreshes the source table before scanning, new partition specs or source-type changes are processed by `LiveFiles` but grouped and emitted using this stale type/schema, producing incomplete or incompatible partition rows. Please make the derived state refresh-aware (or rebuild it per scan) and add a refresh regression test.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat(inspect): add system metadata tables [iceberg-cpp]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #876:
URL: https://github.com/apache/iceberg-cpp/pull/876#discussion_r4257083684
##########
src/iceberg/inspect/files_table.cc:
##########
@@ -0,0 +1,79 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+#include "iceberg/inspect/files_table.h"
+
+#include <memory>
+#include <utility>
+
+#include "iceberg/inspect/metadata_table_stream_internal.h"
+#include "iceberg/inspect/metadata_table_util_internal.h"
+#include "iceberg/schema.h"
+#include "iceberg/table.h"
+#include "iceberg/type.h"
+#include "iceberg/util/macros.h"
+
+namespace iceberg {
+
+FilesTable::FilesTable(std::shared_ptr<Table> table)
+ : TimeTravelMetadataTable(std::move(table)) {}
+
+FilesTable::~FilesTable() = default;
+
+const std::shared_ptr<Schema>& FilesTable::schema() const { return schema_; }
+
+Result<std::unique_ptr<FilesTable>> FilesTable::Make(std::shared_ptr<Table> table) {
+ ICEBERG_PRECHECK(table != nullptr, "Table cannot be null");
+ auto files = std::unique_ptr<FilesTable>(new FilesTable(std::move(table)));
+ ICEBERG_RETURN_UNEXPECTED(files->RefreshSchema());
+ return files;
+}
+
+Status FilesTable::RefreshSchema() {
+ ICEBERG_ASSIGN_OR_RAISE(auto table_schema, source_table()->schema());
+ ICEBERG_ASSIGN_OR_RAISE(auto partition_type,
+ internal::UnifiedPartitionType(*source_table()));
+ ICEBERG_ASSIGN_OR_RAISE(auto schema,
+ internal::FilesTableSchema(*table_schema, partition_type));
+ schema_ = std::move(schema);
+ table_schema_ = std::move(table_schema);
+ partition_type_ = std::move(partition_type);
+ return {};
+}
+
+Result<ArrowArrayStream> FilesTable::ScanSnapshot(
+ const SnapshotSelection& snapshot_selection) {
+ ICEBERG_ASSIGN_OR_RAISE(auto snapshot, internal::ResolveMetadataTableSnapshot(
+ *source_table(), snapshot_selection));
+ ICEBERG_RETURN_UNEXPECTED(RefreshSchema());
+ ICEBERG_ASSIGN_OR_RAISE(auto files, internal::LiveFiles(*source_table(), snapshot));
+ auto schema = schema_;
+ auto table_schema = table_schema_;
+ auto partition_type = partition_type_;
+ return internal::MakeMetadataTableStream(
+ *schema_, std::move(files),
+ [schema = std::move(schema), table_schema = std::move(table_schema),
+ partition_type = std::move(partition_type)](ArrowRowBuilder& builder,
+ const internal::LiveFile& file) {
+ return internal::AppendDataFile(builder, *schema, *table_schema, *partition_type,
+ file);
+ });
Review Comment:
The stream creation currently passes `*schema_` (the mutable member) into `MakeMetadataTableStream`, while the row appender captures the local `schema` copy. To ensure the stream schema and captured schema always stay consistent (especially if `Scan()` can be called concurrently and `RefreshSchema()` updates `schema_`), pass `*schema` instead of `*schema_` when constructing the stream.
##########
src/iceberg/inspect/partitions_table.cc:
##########
@@ -0,0 +1,324 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+#include "iceberg/inspect/partitions_table.h"
+
+#include <algorithm>
+#include <chrono>
+#include <cstddef>
+#include <cstdint>
+#include <memory>
+#include <optional>
+#include <ranges>
+#include <unordered_map>
+#include <utility>
+#include <vector>
+
+#include "iceberg/arrow_row_builder_internal.h"
+#include "iceberg/expression/literal.h"
+#include "iceberg/inspect/metadata_table_stream_internal.h"
+#include "iceberg/inspect/metadata_table_util_internal.h"
+#include "iceberg/manifest/manifest_entry.h"
+#include "iceberg/partition_spec.h"
+#include "iceberg/row/partition_values.h"
+#include "iceberg/schema.h"
+#include "iceberg/schema_field.h"
+#include "iceberg/snapshot.h"
+#include "iceberg/table.h"
+#include "iceberg/type.h"
+#include "iceberg/util/macros.h"
+
+namespace iceberg {
+namespace {
+
+constexpr int32_t kPartitionFieldId = 1;
+constexpr int32_t kRecordCountFieldId = 2;
+constexpr int32_t kFileCountFieldId = 3;
+constexpr int32_t kSpecIdFieldId = 4;
+constexpr int32_t kPositionDeleteRecordCountFieldId = 5;
+constexpr int32_t kPositionDeleteFileCountFieldId = 6;
+constexpr int32_t kEqualityDeleteRecordCountFieldId = 7;
+constexpr int32_t kEqualityDeleteFileCountFieldId = 8;
+constexpr int32_t kLastUpdatedAtFieldId = 9;
+constexpr int32_t kLastUpdatedSnapshotIdFieldId = 10;
+constexpr int32_t kTotalDataFileSizeFieldId = 11;
+
+struct PartitionKey {
+ PartitionValues values;
+ size_t projected_fields;
+
+ bool operator==(const PartitionKey& other) const {
+ if (projected_fields != other.projected_fields ||
+ values.num_fields() != other.values.num_fields()) {
+ return false;
+ }
+ for (size_t index = 0; index < values.num_fields(); ++index) {
+ const auto& lhs = values.values()[index];
+ const auto& rhs = other.values.values()[index];
+ if (lhs.IsNull() || rhs.IsNull()) {
+ if (lhs.IsNull() != rhs.IsNull()) {
+ return false;
+ }
+ } else if (lhs.IsNaN() && rhs.IsNaN()) {
+ // Java partition equality canonicalizes all NaN signs and payloads.
+ continue;
+ } else if (lhs != rhs) {
+ return false;
+ }
+ }
+ return true;
+ }
+};
+
+struct PartitionKeyHash {
+ size_t operator()(const PartitionKey& key) const noexcept {
+ size_t result = 17;
+ for (const auto& value : key.values.values()) {
+ const size_t value_hash =
+ value.IsNaN() ? 0x7ff8000000000000ULL : LiteralHash{}(value);
+ result = result * 37 + value_hash;
+ }
+ return result * 37 + key.projected_fields;
+ }
+};
+
+size_t ProjectedFieldCount(const StructType& partition_type, const PartitionSpec& spec) {
+ size_t count = 0;
+ for (const auto& field : partition_type.fields()) {
+ count += std::ranges::any_of(
+ spec.fields(), [field_id = field.field_id()](const PartitionField& spec_field) {
+ return spec_field.field_id() == field_id;
+ });
+ }
+ return count;
+}
Review Comment:
`ProjectedFieldCount()` is an `O(|union_fields| * |spec_fields|)` scan and is executed per live file; this can become a hotspot for large tables. Consider precomputing the projected field count per spec (e.g., keyed by `spec_id`) once per scan using a `unordered_set<int32_t>` of active union field IDs, then doing an `O(1)` lookup during the file loop. This reduces CPU overhead substantially without changing behavior.
##########
src/iceberg/inspect/refs_table.cc:
##########
@@ -0,0 +1,109 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+#include "iceberg/inspect/refs_table.h"
+
+#include <algorithm>
+#include <memory>
+#include <optional>
+#include <string>
+#include <utility>
+#include <vector>
+
+#include "iceberg/arrow_row_builder_internal.h"
+#include "iceberg/inspect/metadata_table_stream_internal.h"
+#include "iceberg/schema.h"
+#include "iceberg/schema_field.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 {
+
+struct RefRow {
+ std::string name;
+ std::string type;
+ int64_t snapshot_id;
+ std::optional<int64_t> max_ref_age_ms;
+ std::optional<int32_t> min_snapshots_to_keep;
+ std::optional<int64_t> max_snapshot_age_ms;
+};
+
+Status AppendOptional(ArrowArray* array, const auto& value) {
+ if (!value.has_value()) {
+ return AppendNull(array);
+ }
+ return AppendInt(array, static_cast<int64_t>(*value));
+}
Review Comment:
`AppendOptional` unconditionally funnels optionals through `AppendInt(... int64_t ...)`, but `min_snapshots_to_keep` is defined as an `int32()` field in the table schema. If `AppendInt` is not type-dispatching based on the `ArrowArray`’s concrete type, this can produce incorrect values or runtime failures when appending to a 32-bit integer column. Prefer a typed helper (e.g., a templated `AppendOptionalInt<T>` similar to the one already introduced in `metadata_table_util_internal.cc`) and use the correct-width append API for `int32` vs `int64`.
##########
src/iceberg/inspect/partitions_table.cc:
##########
@@ -0,0 +1,324 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+#include "iceberg/inspect/partitions_table.h"
+
+#include <algorithm>
+#include <chrono>
+#include <cstddef>
+#include <cstdint>
+#include <memory>
+#include <optional>
+#include <ranges>
+#include <unordered_map>
+#include <utility>
+#include <vector>
+
+#include "iceberg/arrow_row_builder_internal.h"
+#include "iceberg/expression/literal.h"
+#include "iceberg/inspect/metadata_table_stream_internal.h"
+#include "iceberg/inspect/metadata_table_util_internal.h"
+#include "iceberg/manifest/manifest_entry.h"
+#include "iceberg/partition_spec.h"
+#include "iceberg/row/partition_values.h"
+#include "iceberg/schema.h"
+#include "iceberg/schema_field.h"
+#include "iceberg/snapshot.h"
+#include "iceberg/table.h"
+#include "iceberg/type.h"
+#include "iceberg/util/macros.h"
+
+namespace iceberg {
+namespace {
+
+constexpr int32_t kPartitionFieldId = 1;
+constexpr int32_t kRecordCountFieldId = 2;
+constexpr int32_t kFileCountFieldId = 3;
+constexpr int32_t kSpecIdFieldId = 4;
+constexpr int32_t kPositionDeleteRecordCountFieldId = 5;
+constexpr int32_t kPositionDeleteFileCountFieldId = 6;
+constexpr int32_t kEqualityDeleteRecordCountFieldId = 7;
+constexpr int32_t kEqualityDeleteFileCountFieldId = 8;
+constexpr int32_t kLastUpdatedAtFieldId = 9;
+constexpr int32_t kLastUpdatedSnapshotIdFieldId = 10;
+constexpr int32_t kTotalDataFileSizeFieldId = 11;
+
+struct PartitionKey {
+ PartitionValues values;
+ size_t projected_fields;
+
+ bool operator==(const PartitionKey& other) const {
+ if (projected_fields != other.projected_fields ||
+ values.num_fields() != other.values.num_fields()) {
+ return false;
+ }
+ for (size_t index = 0; index < values.num_fields(); ++index) {
+ const auto& lhs = values.values()[index];
+ const auto& rhs = other.values.values()[index];
+ if (lhs.IsNull() || rhs.IsNull()) {
+ if (lhs.IsNull() != rhs.IsNull()) {
+ return false;
+ }
+ } else if (lhs.IsNaN() && rhs.IsNaN()) {
+ // Java partition equality canonicalizes all NaN signs and payloads.
+ continue;
+ } else if (lhs != rhs) {
+ return false;
+ }
+ }
+ return true;
+ }
+};
+
+struct PartitionKeyHash {
+ size_t operator()(const PartitionKey& key) const noexcept {
+ size_t result = 17;
+ for (const auto& value : key.values.values()) {
+ const size_t value_hash =
+ value.IsNaN() ? 0x7ff8000000000000ULL : LiteralHash{}(value);
+ result = result * 37 + value_hash;
+ }
+ return result * 37 + key.projected_fields;
+ }
+};
+
+size_t ProjectedFieldCount(const StructType& partition_type, const PartitionSpec& spec) {
+ size_t count = 0;
+ for (const auto& field : partition_type.fields()) {
+ count += std::ranges::any_of(
+ spec.fields(), [field_id = field.field_id()](const PartitionField& spec_field) {
+ return spec_field.field_id() == field_id;
+ });
+ }
+ return count;
+}
+
+struct PartitionStats {
+ explicit PartitionStats(PartitionValues values) : partition(std::move(values)) {}
+
+ PartitionValues partition;
+ int32_t spec_id = PartitionSpec::kInitialSpecId;
+ int64_t data_record_count = 0;
+ int32_t data_file_count = 0;
+ int64_t data_file_size = 0;
+ int64_t position_delete_record_count = 0;
+ int32_t position_delete_file_count = 0;
+ int64_t equality_delete_record_count = 0;
+ int32_t equality_delete_file_count = 0;
+ std::optional<TimePointMs> last_updated_at;
+ std::optional<int64_t> last_updated_snapshot_id;
+};
+
+Status AppendPartition(ArrowRowBuilder& builder, const Schema& schema,
+ const StructType& partition_type,
+ const PartitionStats& partition) {
+ for (size_t index = 0; index < schema.fields().size(); ++index) {
+ auto* array = builder.column(index);
+ switch (schema.fields()[index].field_id()) {
+ case kPartitionFieldId:
+ ICEBERG_RETURN_UNEXPECTED(
+ internal::AppendPartitionValues(array, partition_type, partition.partition));
+ break;
+ case kSpecIdFieldId:
+ ICEBERG_RETURN_UNEXPECTED(AppendInt(array, partition.spec_id));
+ break;
+ case kRecordCountFieldId:
+ ICEBERG_RETURN_UNEXPECTED(AppendInt(array, partition.data_record_count));
+ break;
+ case kFileCountFieldId:
+ ICEBERG_RETURN_UNEXPECTED(AppendInt(array, partition.data_file_count));
+ break;
+ case kTotalDataFileSizeFieldId:
+ ICEBERG_RETURN_UNEXPECTED(AppendInt(array, partition.data_file_size));
+ break;
+ case kPositionDeleteRecordCountFieldId:
+ ICEBERG_RETURN_UNEXPECTED(
+ AppendInt(array, partition.position_delete_record_count));
+ break;
+ case kPositionDeleteFileCountFieldId:
+ ICEBERG_RETURN_UNEXPECTED(AppendInt(array, partition.position_delete_file_count));
+ break;
+ case kEqualityDeleteRecordCountFieldId:
+ ICEBERG_RETURN_UNEXPECTED(
+ AppendInt(array, partition.equality_delete_record_count));
+ break;
+ case kEqualityDeleteFileCountFieldId:
+ ICEBERG_RETURN_UNEXPECTED(AppendInt(array, partition.equality_delete_file_count));
+ break;
+ case kLastUpdatedAtFieldId:
+ if (partition.last_updated_at.has_value()) {
+ ICEBERG_RETURN_UNEXPECTED(
+ AppendInt(array, std::chrono::duration_cast<std::chrono::microseconds>(
+ partition.last_updated_at->time_since_epoch())
+ .count()));
+ } else {
+ ICEBERG_RETURN_UNEXPECTED(AppendNull(array));
+ }
+ break;
+ case kLastUpdatedSnapshotIdFieldId:
+ if (partition.last_updated_snapshot_id.has_value()) {
+ ICEBERG_RETURN_UNEXPECTED(
+ AppendInt(array, *partition.last_updated_snapshot_id));
+ } else {
+ ICEBERG_RETURN_UNEXPECTED(AppendNull(array));
+ }
+ break;
+ default:
+ return InvalidSchema("Unsupported partitions metadata field {}",
+ schema.fields()[index].field_id());
+ }
+ }
+ return builder.FinishRow();
+}
+
+void UpdateCounts(PartitionStats& partition, const DataFile& file) {
+ switch (file.content) {
+ case DataFile::Content::kData:
+ partition.data_record_count += file.record_count;
+ ++partition.data_file_count;
+ partition.data_file_size += file.file_size_in_bytes;
+ break;
+ case DataFile::Content::kPositionDeletes:
+ partition.position_delete_record_count += file.record_count;
+ ++partition.position_delete_file_count;
+ break;
+ case DataFile::Content::kEqualityDeletes:
+ partition.equality_delete_record_count += file.record_count;
+ ++partition.equality_delete_file_count;
+ break;
+ }
+}
+
+} // namespace
+
+PartitionsTable::PartitionsTable(std::shared_ptr<Table> table)
+ : TimeTravelMetadataTable(std::move(table)) {}
+
+PartitionsTable::~PartitionsTable() = default;
+
+const std::shared_ptr<Schema>& PartitionsTable::schema() const { return schema_; }
+
+Result<std::unique_ptr<PartitionsTable>> PartitionsTable::Make(
+ std::shared_ptr<Table> table) {
+ ICEBERG_PRECHECK(table != nullptr, "Table cannot be null");
+ auto partitions =
+ std::unique_ptr<PartitionsTable>(new PartitionsTable(std::move(table)));
+ ICEBERG_RETURN_UNEXPECTED(partitions->RefreshSchema());
+ return partitions;
+}
+
+Status PartitionsTable::RefreshSchema() {
+ ICEBERG_ASSIGN_OR_RAISE(auto partition_type,
+ internal::UnifiedPartitionType(*source_table()));
+
+ std::vector<SchemaField> fields;
+ if (!partition_type->fields().empty()) {
+ fields.push_back(
+ SchemaField::MakeRequired(kPartitionFieldId, "partition", partition_type));
+ fields.push_back(SchemaField::MakeRequired(kSpecIdFieldId, "spec_id", int32()));
+ }
+ fields.push_back(
+ SchemaField::MakeRequired(kRecordCountFieldId, "record_count", int64()));
+ fields.push_back(SchemaField::MakeRequired(kFileCountFieldId, "file_count", int32()));
+ fields.push_back(SchemaField::MakeRequired(kTotalDataFileSizeFieldId,
+ "total_data_file_size_in_bytes", int64()));
+ fields.push_back(SchemaField::MakeRequired(kPositionDeleteRecordCountFieldId,
+ "position_delete_record_count", int64()));
+ fields.push_back(SchemaField::MakeRequired(kPositionDeleteFileCountFieldId,
+ "position_delete_file_count", int32()));
+ fields.push_back(SchemaField::MakeRequired(kEqualityDeleteRecordCountFieldId,
+ "equality_delete_record_count", int64()));
+ fields.push_back(SchemaField::MakeRequired(kEqualityDeleteFileCountFieldId,
+ "equality_delete_file_count", int32()));
+ fields.push_back(SchemaField::MakeOptional(kLastUpdatedAtFieldId, "last_updated_at",
+ timestamp_tz()));
+ fields.push_back(SchemaField::MakeOptional(kLastUpdatedSnapshotIdFieldId,
+ "last_updated_snapshot_id", int64()));
+
+ schema_ = std::make_shared<Schema>(std::move(fields));
+ partition_type_ = std::move(partition_type);
+ return {};
+}
+
+Result<ArrowArrayStream> PartitionsTable::ScanSnapshot(
+ const SnapshotSelection& snapshot_selection) {
+ ICEBERG_ASSIGN_OR_RAISE(auto snapshot, internal::ResolveMetadataTableSnapshot(
+ *source_table(), snapshot_selection));
+ ICEBERG_RETURN_UNEXPECTED(RefreshSchema());
+ ICEBERG_ASSIGN_OR_RAISE(
+ auto files, internal::LiveFiles(
+ *source_table(), snapshot,
+ {"content", "partition", "record_count", "file_size_in_bytes"}));
+
+ std::vector<PartitionStats> partitions;
+ std::unordered_map<PartitionKey, size_t, PartitionKeyHash> positions;
+ std::unordered_map<int64_t, std::shared_ptr<Snapshot>> snapshots;
+ while (true) {
+ ICEBERG_ASSIGN_OR_RAISE(auto next_file, files->Next());
+ if (!next_file.has_value()) {
+ break;
+ }
+ const auto& live_file = *next_file;
+ ICEBERG_ASSIGN_OR_RAISE(auto partition_values, internal::ProjectPartitionValues(
+ *partition_type_, *live_file.spec,
+ live_file.file->partition));
+ PartitionKey key{
+ .values = std::move(partition_values),
+ .projected_fields = ProjectedFieldCount(*partition_type_, *live_file.spec)};
Review Comment:
`ProjectedFieldCount()` is an `O(|union_fields| * |spec_fields|)` scan and is executed per live file; this can become a hotspot for large tables. Consider precomputing the projected field count per spec (e.g., keyed by `spec_id`) once per scan using a `unordered_set<int32_t>` of active union field IDs, then doing an `O(1)` lookup during the file loop. This reduces CPU overhead substantially without changing behavior.
##########
src/iceberg/inspect/metadata_table_util_internal.cc:
##########
@@ -0,0 +1,624 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+#include "iceberg/inspect/metadata_table_util_internal.h"
+
+#include <algorithm>
+#include <cstdint>
+#include <format>
+#include <functional>
+#include <map>
+#include <memory>
+#include <ranges>
+#include <string>
+#include <string_view>
+#include <unordered_map>
+#include <unordered_set>
+#include <utility>
+#include <variant>
+#include <vector>
+
+#include <nanoarrow/nanoarrow.h>
+
+#include "iceberg/arrow_row_builder_internal.h"
+#include "iceberg/constants.h"
+#include "iceberg/file_format.h"
+#include "iceberg/manifest/manifest_list.h"
+#include "iceberg/manifest/manifest_reader.h"
+#include "iceberg/nanoarrow_status_internal.h"
+#include "iceberg/partition_spec.h"
+#include "iceberg/schema.h"
+#include "iceberg/schema_field.h"
+#include "iceberg/snapshot.h"
+#include "iceberg/table.h"
+#include "iceberg/table_metadata.h"
+#include "iceberg/transform.h"
+#include "iceberg/type.h"
+#include "iceberg/util/checked_cast.h"
+#include "iceberg/util/conversions.h"
+#include "iceberg/util/macros.h"
+#include "iceberg/util/snapshot_util.h"
+
+namespace iceberg::internal {
+namespace {
+
+Result<std::shared_ptr<Snapshot>> SnapshotAtRef(const Table& table,
+ std::string_view ref_name) {
+ const auto& metadata = table.metadata();
+ ICEBERG_PRECHECK(metadata != nullptr, "Table metadata cannot be null");
+
+ if (ref_name.empty() || ref_name == SnapshotRef::kMainBranch) {
+ if (metadata->current_snapshot_id == kInvalidSnapshotId) {
+ return std::shared_ptr<Snapshot>{nullptr};
+ }
+ return metadata->SnapshotById(metadata->current_snapshot_id);
+ }
+
+ auto ref = metadata->refs.find(std::string(ref_name));
+ ICEBERG_PRECHECK(ref != metadata->refs.end(), "Cannot find snapshot reference '{}'",
+ ref_name);
+ ICEBERG_PRECHECK(ref->second != nullptr, "Snapshot reference '{}' is null", ref_name);
+ return metadata->SnapshotById(ref->second->snapshot_id);
+}
+
+Status AppendLiteral(ArrowArray* array, const Literal& literal) {
+ if (literal.IsNull()) {
+ return AppendNull(array);
+ }
+ if (literal.IsAboveMax() || literal.IsBelowMin()) {
+ return InvalidArgument("Cannot append non-value partition literal {}",
+ literal.ToString());
+ }
+
+ switch (literal.type()->type_id()) {
+ case TypeId::kBoolean:
+ return AppendBoolean(array, std::get<bool>(literal.value()));
+ case TypeId::kInt:
+ case TypeId::kDate:
+ return AppendInt(array, std::get<int32_t>(literal.value()));
+ case TypeId::kLong:
+ case TypeId::kTime:
+ case TypeId::kTimestamp:
+ case TypeId::kTimestampTz:
+ case TypeId::kTimestampNs:
+ case TypeId::kTimestampTzNs:
+ return AppendInt(array, std::get<int64_t>(literal.value()));
+ case TypeId::kFloat:
+ return AppendDouble(array, std::get<float>(literal.value()));
+ case TypeId::kDouble:
+ return AppendDouble(array, std::get<double>(literal.value()));
+ case TypeId::kString:
+ return AppendString(array, std::get<std::string>(literal.value()));
+ case TypeId::kBinary:
+ case TypeId::kFixed:
+ return AppendBytes(array, std::get<std::vector<uint8_t>>(literal.value()));
+ case TypeId::kDecimal:
+ return AppendDecimal(array, std::get<Decimal>(literal.value()));
+ case TypeId::kUuid:
+ return AppendBytes(array, std::get<Uuid>(literal.value()).bytes());
+ case TypeId::kUnknown:
+ case TypeId::kStruct:
+ case TypeId::kList:
+ case TypeId::kMap:
+ case TypeId::kVariant:
+ case TypeId::kGeometry:
+ case TypeId::kGeography:
+ return NotSupported("Cannot append partition literal of type {}",
+ literal.type()->ToString());
+ }
+ std::unreachable();
+}
+
+template <typename T>
+Status AppendOptionalInt(ArrowArray* array, const std::optional<T>& value) {
+ if (!value.has_value()) {
+ return AppendNull(array);
+ }
+ return AppendInt(array, static_cast<int64_t>(*value));
+}
+
+constexpr std::string_view MetadataFileFormat(FileFormatType format) {
+ switch (format) {
+ case FileFormatType::kParquet:
+ return "PARQUET";
+ case FileFormatType::kAvro:
+ return "AVRO";
+ case FileFormatType::kOrc:
+ return "ORC";
+ case FileFormatType::kPuffin:
+ return "PUFFIN";
+ }
+ std::unreachable();
+}
+
+void CollectPrimitiveFields(
+ const NestedType& type,
+ std::vector<std::reference_wrapper<const SchemaField>>& primitive_fields) {
+ for (const auto& field : type.fields()) {
+ if (field.type()->is_primitive()) {
+ primitive_fields.emplace_back(field);
+ } else if (field.type()->is_nested()) {
+ CollectPrimitiveFields(*std::static_pointer_cast<NestedType>(field.type()),
+ primitive_fields);
+ }
+ }
+}
+
+Result<SchemaField> ReadableMetricsField(const Schema& table_schema,
+ int32_t highest_metadata_field_id) {
+ std::vector<std::reference_wrapper<const SchemaField>> primitive_fields;
+ CollectPrimitiveFields(table_schema, primitive_fields);
+
+ int32_t next_id = highest_metadata_field_id;
+ std::vector<SchemaField> column_metrics;
+ column_metrics.reserve(primitive_fields.size());
+ for (const auto& field_ref : primitive_fields) {
+ const auto& field = field_ref.get();
+ ICEBERG_ASSIGN_OR_RAISE(auto column_name,
+ table_schema.FindColumnNameById(field.field_id()));
+ ICEBERG_PRECHECK(column_name.has_value(), "Cannot find name for field {}",
+ field.field_id());
+
+ const int32_t column_metrics_id = ++next_id;
+ std::vector<SchemaField> metrics{
+ SchemaField::MakeOptional(++next_id, "column_size", int64(),
+ "Total size on disk"),
+ SchemaField::MakeOptional(++next_id, "value_count", int64(),
+ "Total count, including null and NaN"),
+ SchemaField::MakeOptional(++next_id, "null_value_count", int64(),
+ "Null value count"),
+ SchemaField::MakeOptional(++next_id, "nan_value_count", int64(),
+ "NaN value count"),
+ SchemaField::MakeOptional(++next_id, "lower_bound", field.type(), "Lower bound"),
+ SchemaField::MakeOptional(++next_id, "upper_bound", field.type(), "Upper bound"),
+ };
+ column_metrics.emplace_back(
+ column_metrics_id, *column_name, struct_(std::move(metrics)),
+ /*optional=*/true, std::format("Metrics for column {}", *column_name));
+ }
+
+ std::ranges::sort(column_metrics, {},
+ [](const SchemaField& field) { return field.name(); });
+ return SchemaField::MakeOptional(++next_id, "readable_metrics",
+ struct_(std::move(column_metrics)),
+ "Column metrics in readable form");
+}
+
+Status AppendMetric(ArrowArray* array, const std::map<int32_t, int64_t>& metrics,
+ int32_t field_id) {
+ auto metric = metrics.find(field_id);
+ return metric == metrics.end() ? AppendNull(array) : AppendInt(array, metric->second);
+}
+
+Status AppendBound(ArrowArray* array,
+ const std::map<int32_t, std::vector<uint8_t>>& bounds,
+ const SchemaField& field) {
+ auto bound = bounds.find(field.field_id());
+ if (bound == bounds.end()) {
+ return AppendNull(array);
+ }
+ auto primitive = checked_pointer_cast<PrimitiveType>(field.type());
+ ICEBERG_ASSIGN_OR_RAISE(auto literal,
+ Conversions::FromBytes(std::move(primitive), bound->second));
+ return AppendLiteral(array, literal);
+}
+
+Status AppendReadableMetrics(ArrowArray* array, const StructType& readable_type,
+ const Schema& table_schema, const DataFile& file) {
+ ICEBERG_PRECHECK(array != nullptr, "Readable metrics Arrow array cannot be null");
+ ICEBERG_PRECHECK(
+ array->n_children == static_cast<int64_t>(readable_type.fields().size()),
+ "Readable metrics Arrow array has {} fields but schema has {}", array->n_children,
+ readable_type.fields().size());
+ for (int64_t index = 0; index < array->n_children; ++index) {
+ auto* column_metrics = array->children[index];
+ ICEBERG_PRECHECK(column_metrics != nullptr && column_metrics->n_children == 6,
+ "Readable column metrics must contain six fields");
+ const auto readable_name = readable_type.fields()[index].name();
+ ICEBERG_ASSIGN_OR_RAISE(auto source_field,
+ table_schema.FindFieldByName(readable_name));
+ ICEBERG_PRECHECK(source_field.has_value(),
+ "Cannot find readable metrics source field '{}'", readable_name);
Review Comment:
`AppendReadableMetrics()` performs a `FindFieldByName()` lookup for every `(file, metric-column)` pair. Since the readable metrics struct field order is stable for a given scan schema, you can precompute a vector of `field_id`s (or direct pointers/refs to `SchemaField`s) aligned with `readable_type.fields()` during schema construction, and then use those ids directly here. This avoids repeated name lookups and reduces per-row overhead on large scans.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat(inspect): add files, partitions, and manifests metadata tables [iceberg-cpp]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #876:
URL: https://github.com/apache/iceberg-cpp/pull/876#discussion_r4260950896
##########
src/iceberg/test/CMakeLists.txt:
##########
@@ -191,9 +191,12 @@ if(ICEBERG_BUILD_BUNDLE)
add_iceberg_test(metadata_table_test
USE_BUNDLE
SOURCES
- history_table_test.cc
- metadata_table_test.cc
- snapshots_table_test.cc)
+ inspect/files_table_test.cc
+ inspect/history_table_test.cc
+ inspect/manifests_table_test.cc
+ inspect/metadata_table_test.cc
+ inspect/partitions_table_test.cc
+ inspect/snapshots_table_test.cc)
Review Comment:
The test target now references `inspect/history_table_test.cc`, `inspect/snapshots_table_test.cc`, and `inspect/metadata_table_test.cc`, but in this diff set the edited history/snapshots tests are shown under `src/iceberg/test/*.cc` paths (and `metadata_table_test.cc` isn’t shown as moved/added under `inspect/`). If the corresponding file moves/renames aren’t part of this PR branch, this will break the test build. Ensure the sources are actually located under `src/iceberg/test/inspect/` (and added to the PR), or adjust the CMake source list to match the real paths.
##########
src/iceberg/inspect/partitions_table.cc:
##########
@@ -0,0 +1,324 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+#include "iceberg/inspect/partitions_table.h"
+
+#include <algorithm>
+#include <chrono>
+#include <cstddef>
+#include <cstdint>
+#include <memory>
+#include <optional>
+#include <ranges>
+#include <unordered_map>
+#include <utility>
+#include <vector>
+
+#include "iceberg/arrow_row_builder_internal.h"
+#include "iceberg/expression/literal.h"
+#include "iceberg/inspect/metadata_table_stream_internal.h"
+#include "iceberg/inspect/metadata_table_util_internal.h"
+#include "iceberg/manifest/manifest_entry.h"
+#include "iceberg/partition_spec.h"
+#include "iceberg/row/partition_values.h"
+#include "iceberg/schema.h"
+#include "iceberg/schema_field.h"
+#include "iceberg/snapshot.h"
+#include "iceberg/table.h"
+#include "iceberg/type.h"
+#include "iceberg/util/macros.h"
+
+namespace iceberg {
+namespace {
+
+constexpr int32_t kPartitionFieldId = 1;
+constexpr int32_t kRecordCountFieldId = 2;
+constexpr int32_t kFileCountFieldId = 3;
+constexpr int32_t kSpecIdFieldId = 4;
+constexpr int32_t kPositionDeleteRecordCountFieldId = 5;
+constexpr int32_t kPositionDeleteFileCountFieldId = 6;
+constexpr int32_t kEqualityDeleteRecordCountFieldId = 7;
+constexpr int32_t kEqualityDeleteFileCountFieldId = 8;
+constexpr int32_t kLastUpdatedAtFieldId = 9;
+constexpr int32_t kLastUpdatedSnapshotIdFieldId = 10;
+constexpr int32_t kTotalDataFileSizeFieldId = 11;
+
+struct PartitionKey {
+ PartitionValues values;
+ size_t projected_fields;
+
+ bool operator==(const PartitionKey& other) const {
+ if (projected_fields != other.projected_fields ||
+ values.num_fields() != other.values.num_fields()) {
+ return false;
+ }
+ for (size_t index = 0; index < values.num_fields(); ++index) {
+ const auto& lhs = values.values()[index];
+ const auto& rhs = other.values.values()[index];
+ if (lhs.IsNull() || rhs.IsNull()) {
+ if (lhs.IsNull() != rhs.IsNull()) {
+ return false;
+ }
+ } else if (lhs.IsNaN() && rhs.IsNaN()) {
+ // Java partition equality canonicalizes all NaN signs and payloads.
+ continue;
+ } else if (lhs != rhs) {
+ return false;
+ }
+ }
+ return true;
+ }
+};
+
+struct PartitionKeyHash {
+ size_t operator()(const PartitionKey& key) const noexcept {
+ size_t result = 17;
+ for (const auto& value : key.values.values()) {
+ const size_t value_hash =
+ value.IsNaN() ? 0x7ff8000000000000ULL : LiteralHash{}(value);
+ result = result * 37 + value_hash;
+ }
+ return result * 37 + key.projected_fields;
+ }
+};
+
+size_t ProjectedFieldCount(const StructType& partition_type, const PartitionSpec& spec) {
+ size_t count = 0;
+ for (const auto& field : partition_type.fields()) {
+ count += std::ranges::any_of(
+ spec.fields(), [field_id = field.field_id()](const PartitionField& spec_field) {
+ return spec_field.field_id() == field_id;
+ });
+ }
+ return count;
+}
Review Comment:
`ProjectedFieldCount` is O(|unified_fields| * |spec_fields|) and is called for every live file during `PartitionsTable::ScanSnapshot`. This can become a noticeable hotspot for tables with many partition fields/spec evolutions and large snapshots. Precompute per-spec lookup (e.g., `unordered_set<int32_t>` of spec field IDs, or a per-spec projected signature as suggested above) and reuse it inside the scan loop to reduce per-file cost to O(|unified_fields|) or better.
##########
src/iceberg/inspect/partitions_table.cc:
##########
@@ -0,0 +1,324 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+#include "iceberg/inspect/partitions_table.h"
+
+#include <algorithm>
+#include <chrono>
+#include <cstddef>
+#include <cstdint>
+#include <memory>
+#include <optional>
+#include <ranges>
+#include <unordered_map>
+#include <utility>
+#include <vector>
+
+#include "iceberg/arrow_row_builder_internal.h"
+#include "iceberg/expression/literal.h"
+#include "iceberg/inspect/metadata_table_stream_internal.h"
+#include "iceberg/inspect/metadata_table_util_internal.h"
+#include "iceberg/manifest/manifest_entry.h"
+#include "iceberg/partition_spec.h"
+#include "iceberg/row/partition_values.h"
+#include "iceberg/schema.h"
+#include "iceberg/schema_field.h"
+#include "iceberg/snapshot.h"
+#include "iceberg/table.h"
+#include "iceberg/type.h"
+#include "iceberg/util/macros.h"
+
+namespace iceberg {
+namespace {
+
+constexpr int32_t kPartitionFieldId = 1;
+constexpr int32_t kRecordCountFieldId = 2;
+constexpr int32_t kFileCountFieldId = 3;
+constexpr int32_t kSpecIdFieldId = 4;
+constexpr int32_t kPositionDeleteRecordCountFieldId = 5;
+constexpr int32_t kPositionDeleteFileCountFieldId = 6;
+constexpr int32_t kEqualityDeleteRecordCountFieldId = 7;
+constexpr int32_t kEqualityDeleteFileCountFieldId = 8;
+constexpr int32_t kLastUpdatedAtFieldId = 9;
+constexpr int32_t kLastUpdatedSnapshotIdFieldId = 10;
+constexpr int32_t kTotalDataFileSizeFieldId = 11;
+
+struct PartitionKey {
+ PartitionValues values;
+ size_t projected_fields;
+
+ bool operator==(const PartitionKey& other) const {
+ if (projected_fields != other.projected_fields ||
+ values.num_fields() != other.values.num_fields()) {
+ return false;
+ }
+ for (size_t index = 0; index < values.num_fields(); ++index) {
+ const auto& lhs = values.values()[index];
+ const auto& rhs = other.values.values()[index];
+ if (lhs.IsNull() || rhs.IsNull()) {
+ if (lhs.IsNull() != rhs.IsNull()) {
+ return false;
+ }
+ } else if (lhs.IsNaN() && rhs.IsNaN()) {
+ // Java partition equality canonicalizes all NaN signs and payloads.
+ continue;
+ } else if (lhs != rhs) {
+ return false;
+ }
+ }
+ return true;
+ }
+};
Review Comment:
`PartitionKey` uses `projected_fields` as a *count* to distinguish specs, but the count alone is not enough to uniquely identify *which* unified partition fields were present in the originating spec. Two different specs can have the same number of active fields but different field IDs; if their projected values line up (especially with nulls), they could be merged into one partition group unintentionally. Consider replacing `projected_fields` with a stable signature of the projected field-ID set (e.g., a sorted vector of field_ids, a bitset over unified field order, or a precomputed hash of present field_ids for the spec) so grouping can’t collide across distinct field sets.
##########
src/iceberg/inspect/partitions_table.cc:
##########
@@ -0,0 +1,324 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+#include "iceberg/inspect/partitions_table.h"
+
+#include <algorithm>
+#include <chrono>
+#include <cstddef>
+#include <cstdint>
+#include <memory>
+#include <optional>
+#include <ranges>
+#include <unordered_map>
+#include <utility>
+#include <vector>
+
+#include "iceberg/arrow_row_builder_internal.h"
+#include "iceberg/expression/literal.h"
+#include "iceberg/inspect/metadata_table_stream_internal.h"
+#include "iceberg/inspect/metadata_table_util_internal.h"
+#include "iceberg/manifest/manifest_entry.h"
+#include "iceberg/partition_spec.h"
+#include "iceberg/row/partition_values.h"
+#include "iceberg/schema.h"
+#include "iceberg/schema_field.h"
+#include "iceberg/snapshot.h"
+#include "iceberg/table.h"
+#include "iceberg/type.h"
+#include "iceberg/util/macros.h"
+
+namespace iceberg {
+namespace {
+
+constexpr int32_t kPartitionFieldId = 1;
+constexpr int32_t kRecordCountFieldId = 2;
+constexpr int32_t kFileCountFieldId = 3;
+constexpr int32_t kSpecIdFieldId = 4;
+constexpr int32_t kPositionDeleteRecordCountFieldId = 5;
+constexpr int32_t kPositionDeleteFileCountFieldId = 6;
+constexpr int32_t kEqualityDeleteRecordCountFieldId = 7;
+constexpr int32_t kEqualityDeleteFileCountFieldId = 8;
+constexpr int32_t kLastUpdatedAtFieldId = 9;
+constexpr int32_t kLastUpdatedSnapshotIdFieldId = 10;
+constexpr int32_t kTotalDataFileSizeFieldId = 11;
+
+struct PartitionKey {
+ PartitionValues values;
+ size_t projected_fields;
+
+ bool operator==(const PartitionKey& other) const {
+ if (projected_fields != other.projected_fields ||
+ values.num_fields() != other.values.num_fields()) {
+ return false;
+ }
+ for (size_t index = 0; index < values.num_fields(); ++index) {
+ const auto& lhs = values.values()[index];
+ const auto& rhs = other.values.values()[index];
+ if (lhs.IsNull() || rhs.IsNull()) {
+ if (lhs.IsNull() != rhs.IsNull()) {
+ return false;
+ }
+ } else if (lhs.IsNaN() && rhs.IsNaN()) {
+ // Java partition equality canonicalizes all NaN signs and payloads.
+ continue;
+ } else if (lhs != rhs) {
+ return false;
+ }
+ }
+ return true;
+ }
+};
+
+struct PartitionKeyHash {
+ size_t operator()(const PartitionKey& key) const noexcept {
+ size_t result = 17;
+ for (const auto& value : key.values.values()) {
+ const size_t value_hash =
+ value.IsNaN() ? 0x7ff8000000000000ULL : LiteralHash{}(value);
+ result = result * 37 + value_hash;
+ }
+ return result * 37 + key.projected_fields;
+ }
+};
+
+size_t ProjectedFieldCount(const StructType& partition_type, const PartitionSpec& spec) {
+ size_t count = 0;
+ for (const auto& field : partition_type.fields()) {
+ count += std::ranges::any_of(
+ spec.fields(), [field_id = field.field_id()](const PartitionField& spec_field) {
+ return spec_field.field_id() == field_id;
+ });
+ }
+ return count;
+}
Review Comment:
`PartitionKey` uses `projected_fields` as a *count* to distinguish specs, but the count alone is not enough to uniquely identify *which* unified partition fields were present in the originating spec. Two different specs can have the same number of active fields but different field IDs; if their projected values line up (especially with nulls), they could be merged into one partition group unintentionally. Consider replacing `projected_fields` with a stable signature of the projected field-ID set (e.g., a sorted vector of field_ids, a bitset over unified field order, or a precomputed hash of present field_ids for the spec) so grouping can’t collide across distinct field sets.
##########
src/iceberg/inspect/partitions_table.cc:
##########
@@ -0,0 +1,324 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+#include "iceberg/inspect/partitions_table.h"
+
+#include <algorithm>
+#include <chrono>
+#include <cstddef>
+#include <cstdint>
+#include <memory>
+#include <optional>
+#include <ranges>
+#include <unordered_map>
+#include <utility>
+#include <vector>
+
+#include "iceberg/arrow_row_builder_internal.h"
+#include "iceberg/expression/literal.h"
+#include "iceberg/inspect/metadata_table_stream_internal.h"
+#include "iceberg/inspect/metadata_table_util_internal.h"
+#include "iceberg/manifest/manifest_entry.h"
+#include "iceberg/partition_spec.h"
+#include "iceberg/row/partition_values.h"
+#include "iceberg/schema.h"
+#include "iceberg/schema_field.h"
+#include "iceberg/snapshot.h"
+#include "iceberg/table.h"
+#include "iceberg/type.h"
+#include "iceberg/util/macros.h"
+
+namespace iceberg {
+namespace {
+
+constexpr int32_t kPartitionFieldId = 1;
+constexpr int32_t kRecordCountFieldId = 2;
+constexpr int32_t kFileCountFieldId = 3;
+constexpr int32_t kSpecIdFieldId = 4;
+constexpr int32_t kPositionDeleteRecordCountFieldId = 5;
+constexpr int32_t kPositionDeleteFileCountFieldId = 6;
+constexpr int32_t kEqualityDeleteRecordCountFieldId = 7;
+constexpr int32_t kEqualityDeleteFileCountFieldId = 8;
+constexpr int32_t kLastUpdatedAtFieldId = 9;
+constexpr int32_t kLastUpdatedSnapshotIdFieldId = 10;
+constexpr int32_t kTotalDataFileSizeFieldId = 11;
+
+struct PartitionKey {
+ PartitionValues values;
+ size_t projected_fields;
+
+ bool operator==(const PartitionKey& other) const {
+ if (projected_fields != other.projected_fields ||
+ values.num_fields() != other.values.num_fields()) {
+ return false;
+ }
+ for (size_t index = 0; index < values.num_fields(); ++index) {
+ const auto& lhs = values.values()[index];
+ const auto& rhs = other.values.values()[index];
+ if (lhs.IsNull() || rhs.IsNull()) {
+ if (lhs.IsNull() != rhs.IsNull()) {
+ return false;
+ }
+ } else if (lhs.IsNaN() && rhs.IsNaN()) {
+ // Java partition equality canonicalizes all NaN signs and payloads.
+ continue;
+ } else if (lhs != rhs) {
+ return false;
+ }
+ }
+ return true;
+ }
+};
+
+struct PartitionKeyHash {
+ size_t operator()(const PartitionKey& key) const noexcept {
+ size_t result = 17;
+ for (const auto& value : key.values.values()) {
+ const size_t value_hash =
+ value.IsNaN() ? 0x7ff8000000000000ULL : LiteralHash{}(value);
+ result = result * 37 + value_hash;
+ }
+ return result * 37 + key.projected_fields;
+ }
+};
+
+size_t ProjectedFieldCount(const StructType& partition_type, const PartitionSpec& spec) {
+ size_t count = 0;
+ for (const auto& field : partition_type.fields()) {
+ count += std::ranges::any_of(
+ spec.fields(), [field_id = field.field_id()](const PartitionField& spec_field) {
+ return spec_field.field_id() == field_id;
+ });
+ }
+ return count;
+}
+
+struct PartitionStats {
+ explicit PartitionStats(PartitionValues values) : partition(std::move(values)) {}
+
+ PartitionValues partition;
+ int32_t spec_id = PartitionSpec::kInitialSpecId;
+ int64_t data_record_count = 0;
+ int32_t data_file_count = 0;
+ int64_t data_file_size = 0;
+ int64_t position_delete_record_count = 0;
+ int32_t position_delete_file_count = 0;
+ int64_t equality_delete_record_count = 0;
+ int32_t equality_delete_file_count = 0;
+ std::optional<TimePointMs> last_updated_at;
+ std::optional<int64_t> last_updated_snapshot_id;
+};
+
+Status AppendPartition(ArrowRowBuilder& builder, const Schema& schema,
+ const StructType& partition_type,
+ const PartitionStats& partition) {
+ for (size_t index = 0; index < schema.fields().size(); ++index) {
+ auto* array = builder.column(index);
+ switch (schema.fields()[index].field_id()) {
+ case kPartitionFieldId:
+ ICEBERG_RETURN_UNEXPECTED(
+ internal::AppendPartitionValues(array, partition_type, partition.partition));
+ break;
+ case kSpecIdFieldId:
+ ICEBERG_RETURN_UNEXPECTED(AppendInt(array, partition.spec_id));
+ break;
+ case kRecordCountFieldId:
+ ICEBERG_RETURN_UNEXPECTED(AppendInt(array, partition.data_record_count));
+ break;
+ case kFileCountFieldId:
+ ICEBERG_RETURN_UNEXPECTED(AppendInt(array, partition.data_file_count));
+ break;
+ case kTotalDataFileSizeFieldId:
+ ICEBERG_RETURN_UNEXPECTED(AppendInt(array, partition.data_file_size));
+ break;
+ case kPositionDeleteRecordCountFieldId:
+ ICEBERG_RETURN_UNEXPECTED(
+ AppendInt(array, partition.position_delete_record_count));
+ break;
+ case kPositionDeleteFileCountFieldId:
+ ICEBERG_RETURN_UNEXPECTED(AppendInt(array, partition.position_delete_file_count));
+ break;
+ case kEqualityDeleteRecordCountFieldId:
+ ICEBERG_RETURN_UNEXPECTED(
+ AppendInt(array, partition.equality_delete_record_count));
+ break;
+ case kEqualityDeleteFileCountFieldId:
+ ICEBERG_RETURN_UNEXPECTED(AppendInt(array, partition.equality_delete_file_count));
+ break;
+ case kLastUpdatedAtFieldId:
+ if (partition.last_updated_at.has_value()) {
+ ICEBERG_RETURN_UNEXPECTED(
+ AppendInt(array, std::chrono::duration_cast<std::chrono::microseconds>(
+ partition.last_updated_at->time_since_epoch())
+ .count()));
+ } else {
+ ICEBERG_RETURN_UNEXPECTED(AppendNull(array));
+ }
+ break;
+ case kLastUpdatedSnapshotIdFieldId:
+ if (partition.last_updated_snapshot_id.has_value()) {
+ ICEBERG_RETURN_UNEXPECTED(
+ AppendInt(array, *partition.last_updated_snapshot_id));
+ } else {
+ ICEBERG_RETURN_UNEXPECTED(AppendNull(array));
+ }
+ break;
+ default:
+ return InvalidSchema("Unsupported partitions metadata field {}",
+ schema.fields()[index].field_id());
+ }
+ }
+ return builder.FinishRow();
+}
+
+void UpdateCounts(PartitionStats& partition, const DataFile& file) {
+ switch (file.content) {
+ case DataFile::Content::kData:
+ partition.data_record_count += file.record_count;
+ ++partition.data_file_count;
+ partition.data_file_size += file.file_size_in_bytes;
+ break;
+ case DataFile::Content::kPositionDeletes:
+ partition.position_delete_record_count += file.record_count;
+ ++partition.position_delete_file_count;
+ break;
+ case DataFile::Content::kEqualityDeletes:
+ partition.equality_delete_record_count += file.record_count;
+ ++partition.equality_delete_file_count;
+ break;
+ }
+}
+
+} // namespace
+
+PartitionsTable::PartitionsTable(std::shared_ptr<Table> table)
+ : TimeTravelMetadataTable(std::move(table)) {}
+
+PartitionsTable::~PartitionsTable() = default;
+
+const std::shared_ptr<Schema>& PartitionsTable::schema() const { return schema_; }
+
+Result<std::unique_ptr<PartitionsTable>> PartitionsTable::Make(
+ std::shared_ptr<Table> table) {
+ ICEBERG_PRECHECK(table != nullptr, "Table cannot be null");
+ auto partitions =
+ std::unique_ptr<PartitionsTable>(new PartitionsTable(std::move(table)));
+ ICEBERG_RETURN_UNEXPECTED(partitions->RefreshSchema());
+ return partitions;
+}
+
+Status PartitionsTable::RefreshSchema() {
+ ICEBERG_ASSIGN_OR_RAISE(auto partition_type,
+ internal::UnifiedPartitionType(*source_table()));
+
+ std::vector<SchemaField> fields;
+ if (!partition_type->fields().empty()) {
+ fields.push_back(
+ SchemaField::MakeRequired(kPartitionFieldId, "partition", partition_type));
+ fields.push_back(SchemaField::MakeRequired(kSpecIdFieldId, "spec_id", int32()));
+ }
+ fields.push_back(
+ SchemaField::MakeRequired(kRecordCountFieldId, "record_count", int64()));
+ fields.push_back(SchemaField::MakeRequired(kFileCountFieldId, "file_count", int32()));
+ fields.push_back(SchemaField::MakeRequired(kTotalDataFileSizeFieldId,
+ "total_data_file_size_in_bytes", int64()));
+ fields.push_back(SchemaField::MakeRequired(kPositionDeleteRecordCountFieldId,
+ "position_delete_record_count", int64()));
+ fields.push_back(SchemaField::MakeRequired(kPositionDeleteFileCountFieldId,
+ "position_delete_file_count", int32()));
+ fields.push_back(SchemaField::MakeRequired(kEqualityDeleteRecordCountFieldId,
+ "equality_delete_record_count", int64()));
+ fields.push_back(SchemaField::MakeRequired(kEqualityDeleteFileCountFieldId,
+ "equality_delete_file_count", int32()));
+ fields.push_back(SchemaField::MakeOptional(kLastUpdatedAtFieldId, "last_updated_at",
+ timestamp_tz()));
+ fields.push_back(SchemaField::MakeOptional(kLastUpdatedSnapshotIdFieldId,
+ "last_updated_snapshot_id", int64()));
+
+ schema_ = std::make_shared<Schema>(std::move(fields));
+ partition_type_ = std::move(partition_type);
+ return {};
+}
+
+Result<ArrowArrayStream> PartitionsTable::ScanSnapshot(
+ const SnapshotSelection& snapshot_selection) {
+ ICEBERG_ASSIGN_OR_RAISE(auto snapshot, internal::ResolveMetadataTableSnapshot(
+ *source_table(), snapshot_selection));
+ ICEBERG_RETURN_UNEXPECTED(RefreshSchema());
+ ICEBERG_ASSIGN_OR_RAISE(
+ auto files, internal::LiveFiles(
+ *source_table(), snapshot,
+ {"content", "partition", "record_count", "file_size_in_bytes"}));
+
+ std::vector<PartitionStats> partitions;
+ std::unordered_map<PartitionKey, size_t, PartitionKeyHash> positions;
Review Comment:
This aggregation keeps a full `PartitionValues` copy in the unordered_map key *and* another in `PartitionStats` (the vector), which increases memory and copy/move work for large partition cardinalities. Consider restructuring so the partition values are stored only once (e.g., store `PartitionStats` directly in the map and keep an auxiliary order vector, or store keys in the map and have `PartitionStats` reference/move from the inserted map key) to reduce duplication and improve scan throughput.
##########
src/iceberg/inspect/partitions_table.cc:
##########
@@ -0,0 +1,324 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+#include "iceberg/inspect/partitions_table.h"
+
+#include <algorithm>
+#include <chrono>
+#include <cstddef>
+#include <cstdint>
+#include <memory>
+#include <optional>
+#include <ranges>
+#include <unordered_map>
+#include <utility>
+#include <vector>
+
+#include "iceberg/arrow_row_builder_internal.h"
+#include "iceberg/expression/literal.h"
+#include "iceberg/inspect/metadata_table_stream_internal.h"
+#include "iceberg/inspect/metadata_table_util_internal.h"
+#include "iceberg/manifest/manifest_entry.h"
+#include "iceberg/partition_spec.h"
+#include "iceberg/row/partition_values.h"
+#include "iceberg/schema.h"
+#include "iceberg/schema_field.h"
+#include "iceberg/snapshot.h"
+#include "iceberg/table.h"
+#include "iceberg/type.h"
+#include "iceberg/util/macros.h"
+
+namespace iceberg {
+namespace {
+
+constexpr int32_t kPartitionFieldId = 1;
+constexpr int32_t kRecordCountFieldId = 2;
+constexpr int32_t kFileCountFieldId = 3;
+constexpr int32_t kSpecIdFieldId = 4;
+constexpr int32_t kPositionDeleteRecordCountFieldId = 5;
+constexpr int32_t kPositionDeleteFileCountFieldId = 6;
+constexpr int32_t kEqualityDeleteRecordCountFieldId = 7;
+constexpr int32_t kEqualityDeleteFileCountFieldId = 8;
+constexpr int32_t kLastUpdatedAtFieldId = 9;
+constexpr int32_t kLastUpdatedSnapshotIdFieldId = 10;
+constexpr int32_t kTotalDataFileSizeFieldId = 11;
+
+struct PartitionKey {
+ PartitionValues values;
+ size_t projected_fields;
+
+ bool operator==(const PartitionKey& other) const {
+ if (projected_fields != other.projected_fields ||
+ values.num_fields() != other.values.num_fields()) {
+ return false;
+ }
+ for (size_t index = 0; index < values.num_fields(); ++index) {
+ const auto& lhs = values.values()[index];
+ const auto& rhs = other.values.values()[index];
+ if (lhs.IsNull() || rhs.IsNull()) {
+ if (lhs.IsNull() != rhs.IsNull()) {
+ return false;
+ }
+ } else if (lhs.IsNaN() && rhs.IsNaN()) {
+ // Java partition equality canonicalizes all NaN signs and payloads.
+ continue;
+ } else if (lhs != rhs) {
+ return false;
+ }
+ }
+ return true;
+ }
+};
+
+struct PartitionKeyHash {
+ size_t operator()(const PartitionKey& key) const noexcept {
+ size_t result = 17;
+ for (const auto& value : key.values.values()) {
+ const size_t value_hash =
+ value.IsNaN() ? 0x7ff8000000000000ULL : LiteralHash{}(value);
+ result = result * 37 + value_hash;
+ }
+ return result * 37 + key.projected_fields;
+ }
+};
+
+size_t ProjectedFieldCount(const StructType& partition_type, const PartitionSpec& spec) {
+ size_t count = 0;
+ for (const auto& field : partition_type.fields()) {
+ count += std::ranges::any_of(
+ spec.fields(), [field_id = field.field_id()](const PartitionField& spec_field) {
+ return spec_field.field_id() == field_id;
+ });
+ }
+ return count;
+}
+
+struct PartitionStats {
+ explicit PartitionStats(PartitionValues values) : partition(std::move(values)) {}
+
+ PartitionValues partition;
+ int32_t spec_id = PartitionSpec::kInitialSpecId;
+ int64_t data_record_count = 0;
+ int32_t data_file_count = 0;
+ int64_t data_file_size = 0;
+ int64_t position_delete_record_count = 0;
+ int32_t position_delete_file_count = 0;
+ int64_t equality_delete_record_count = 0;
+ int32_t equality_delete_file_count = 0;
+ std::optional<TimePointMs> last_updated_at;
+ std::optional<int64_t> last_updated_snapshot_id;
+};
+
+Status AppendPartition(ArrowRowBuilder& builder, const Schema& schema,
+ const StructType& partition_type,
+ const PartitionStats& partition) {
+ for (size_t index = 0; index < schema.fields().size(); ++index) {
+ auto* array = builder.column(index);
+ switch (schema.fields()[index].field_id()) {
+ case kPartitionFieldId:
+ ICEBERG_RETURN_UNEXPECTED(
+ internal::AppendPartitionValues(array, partition_type, partition.partition));
+ break;
+ case kSpecIdFieldId:
+ ICEBERG_RETURN_UNEXPECTED(AppendInt(array, partition.spec_id));
+ break;
+ case kRecordCountFieldId:
+ ICEBERG_RETURN_UNEXPECTED(AppendInt(array, partition.data_record_count));
+ break;
+ case kFileCountFieldId:
+ ICEBERG_RETURN_UNEXPECTED(AppendInt(array, partition.data_file_count));
+ break;
+ case kTotalDataFileSizeFieldId:
+ ICEBERG_RETURN_UNEXPECTED(AppendInt(array, partition.data_file_size));
+ break;
+ case kPositionDeleteRecordCountFieldId:
+ ICEBERG_RETURN_UNEXPECTED(
+ AppendInt(array, partition.position_delete_record_count));
+ break;
+ case kPositionDeleteFileCountFieldId:
+ ICEBERG_RETURN_UNEXPECTED(AppendInt(array, partition.position_delete_file_count));
+ break;
+ case kEqualityDeleteRecordCountFieldId:
+ ICEBERG_RETURN_UNEXPECTED(
+ AppendInt(array, partition.equality_delete_record_count));
+ break;
+ case kEqualityDeleteFileCountFieldId:
+ ICEBERG_RETURN_UNEXPECTED(AppendInt(array, partition.equality_delete_file_count));
+ break;
+ case kLastUpdatedAtFieldId:
+ if (partition.last_updated_at.has_value()) {
+ ICEBERG_RETURN_UNEXPECTED(
+ AppendInt(array, std::chrono::duration_cast<std::chrono::microseconds>(
+ partition.last_updated_at->time_since_epoch())
+ .count()));
+ } else {
+ ICEBERG_RETURN_UNEXPECTED(AppendNull(array));
+ }
+ break;
+ case kLastUpdatedSnapshotIdFieldId:
+ if (partition.last_updated_snapshot_id.has_value()) {
+ ICEBERG_RETURN_UNEXPECTED(
+ AppendInt(array, *partition.last_updated_snapshot_id));
+ } else {
+ ICEBERG_RETURN_UNEXPECTED(AppendNull(array));
+ }
+ break;
+ default:
+ return InvalidSchema("Unsupported partitions metadata field {}",
+ schema.fields()[index].field_id());
+ }
+ }
+ return builder.FinishRow();
+}
+
+void UpdateCounts(PartitionStats& partition, const DataFile& file) {
+ switch (file.content) {
+ case DataFile::Content::kData:
+ partition.data_record_count += file.record_count;
+ ++partition.data_file_count;
+ partition.data_file_size += file.file_size_in_bytes;
+ break;
+ case DataFile::Content::kPositionDeletes:
+ partition.position_delete_record_count += file.record_count;
+ ++partition.position_delete_file_count;
+ break;
+ case DataFile::Content::kEqualityDeletes:
+ partition.equality_delete_record_count += file.record_count;
+ ++partition.equality_delete_file_count;
+ break;
+ }
+}
+
+} // namespace
+
+PartitionsTable::PartitionsTable(std::shared_ptr<Table> table)
+ : TimeTravelMetadataTable(std::move(table)) {}
+
+PartitionsTable::~PartitionsTable() = default;
+
+const std::shared_ptr<Schema>& PartitionsTable::schema() const { return schema_; }
+
+Result<std::unique_ptr<PartitionsTable>> PartitionsTable::Make(
+ std::shared_ptr<Table> table) {
+ ICEBERG_PRECHECK(table != nullptr, "Table cannot be null");
+ auto partitions =
+ std::unique_ptr<PartitionsTable>(new PartitionsTable(std::move(table)));
+ ICEBERG_RETURN_UNEXPECTED(partitions->RefreshSchema());
+ return partitions;
+}
+
+Status PartitionsTable::RefreshSchema() {
+ ICEBERG_ASSIGN_OR_RAISE(auto partition_type,
+ internal::UnifiedPartitionType(*source_table()));
+
+ std::vector<SchemaField> fields;
+ if (!partition_type->fields().empty()) {
+ fields.push_back(
+ SchemaField::MakeRequired(kPartitionFieldId, "partition", partition_type));
+ fields.push_back(SchemaField::MakeRequired(kSpecIdFieldId, "spec_id", int32()));
+ }
+ fields.push_back(
+ SchemaField::MakeRequired(kRecordCountFieldId, "record_count", int64()));
+ fields.push_back(SchemaField::MakeRequired(kFileCountFieldId, "file_count", int32()));
+ fields.push_back(SchemaField::MakeRequired(kTotalDataFileSizeFieldId,
+ "total_data_file_size_in_bytes", int64()));
+ fields.push_back(SchemaField::MakeRequired(kPositionDeleteRecordCountFieldId,
+ "position_delete_record_count", int64()));
+ fields.push_back(SchemaField::MakeRequired(kPositionDeleteFileCountFieldId,
+ "position_delete_file_count", int32()));
+ fields.push_back(SchemaField::MakeRequired(kEqualityDeleteRecordCountFieldId,
+ "equality_delete_record_count", int64()));
+ fields.push_back(SchemaField::MakeRequired(kEqualityDeleteFileCountFieldId,
+ "equality_delete_file_count", int32()));
+ fields.push_back(SchemaField::MakeOptional(kLastUpdatedAtFieldId, "last_updated_at",
+ timestamp_tz()));
+ fields.push_back(SchemaField::MakeOptional(kLastUpdatedSnapshotIdFieldId,
+ "last_updated_snapshot_id", int64()));
+
+ schema_ = std::make_shared<Schema>(std::move(fields));
+ partition_type_ = std::move(partition_type);
+ return {};
+}
+
+Result<ArrowArrayStream> PartitionsTable::ScanSnapshot(
+ const SnapshotSelection& snapshot_selection) {
+ ICEBERG_ASSIGN_OR_RAISE(auto snapshot, internal::ResolveMetadataTableSnapshot(
+ *source_table(), snapshot_selection));
+ ICEBERG_RETURN_UNEXPECTED(RefreshSchema());
+ ICEBERG_ASSIGN_OR_RAISE(
+ auto files, internal::LiveFiles(
+ *source_table(), snapshot,
+ {"content", "partition", "record_count", "file_size_in_bytes"}));
+
+ std::vector<PartitionStats> partitions;
+ std::unordered_map<PartitionKey, size_t, PartitionKeyHash> positions;
+ std::unordered_map<int64_t, std::shared_ptr<Snapshot>> snapshots;
+ while (true) {
+ ICEBERG_ASSIGN_OR_RAISE(auto next_file, files->Next());
+ if (!next_file.has_value()) {
+ break;
+ }
+ const auto& live_file = *next_file;
+ ICEBERG_ASSIGN_OR_RAISE(auto partition_values, internal::ProjectPartitionValues(
+ *partition_type_, *live_file.spec,
+ live_file.file->partition));
+ PartitionKey key{
+ .values = std::move(partition_values),
+ .projected_fields = ProjectedFieldCount(*partition_type_, *live_file.spec)};
+ auto [position, inserted] = positions.try_emplace(key, partitions.size());
+ if (inserted) {
+ partitions.emplace_back(std::move(key.values));
+ }
Review Comment:
This aggregation keeps a full `PartitionValues` copy in the unordered_map key *and* another in `PartitionStats` (the vector), which increases memory and copy/move work for large partition cardinalities. Consider restructuring so the partition values are stored only once (e.g., store `PartitionStats` directly in the map and keep an auxiliary order vector, or store keys in the map and have `PartitionStats` reference/move from the inserted map key) to reduce duplication and improve scan throughput.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat(inspect): add system metadata tables [iceberg-cpp]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #876:
URL: https://github.com/apache/iceberg-cpp/pull/876#discussion_r4025461660
##########
src/iceberg/inspect/metadata_table_util_internal.cc:
##########
@@ -0,0 +1,642 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+#include "iceberg/inspect/metadata_table_util_internal.h"
+
+#include <algorithm>
+#include <cstdint>
+#include <format>
+#include <functional>
+#include <map>
+#include <memory>
+#include <ranges>
+#include <string>
+#include <string_view>
+#include <unordered_map>
+#include <unordered_set>
+#include <utility>
+#include <variant>
+#include <vector>
+
+#include <nanoarrow/nanoarrow.h>
+
+#include "iceberg/arrow_row_builder_internal.h"
+#include "iceberg/constants.h"
+#include "iceberg/file_format.h"
+#include "iceberg/manifest/manifest_list.h"
+#include "iceberg/manifest/manifest_reader.h"
+#include "iceberg/nanoarrow_status_internal.h"
+#include "iceberg/partition_spec.h"
+#include "iceberg/schema.h"
+#include "iceberg/schema_field.h"
+#include "iceberg/snapshot.h"
+#include "iceberg/table.h"
+#include "iceberg/table_metadata.h"
+#include "iceberg/transform.h"
+#include "iceberg/type.h"
+#include "iceberg/util/checked_cast.h"
+#include "iceberg/util/conversions.h"
+#include "iceberg/util/macros.h"
+#include "iceberg/util/snapshot_util.h"
+
+namespace iceberg::internal {
+namespace {
+
+Result<std::shared_ptr<Snapshot>> SnapshotAtRef(const Table& table,
+ std::string_view ref_name) {
+ const auto& metadata = table.metadata();
+ ICEBERG_PRECHECK(metadata != nullptr, "Table metadata cannot be null");
+
+ if (ref_name.empty() || ref_name == SnapshotRef::kMainBranch) {
+ if (metadata->current_snapshot_id == kInvalidSnapshotId) {
+ return std::shared_ptr<Snapshot>{nullptr};
+ }
+ return metadata->SnapshotById(metadata->current_snapshot_id);
+ }
+
+ auto ref = metadata->refs.find(std::string(ref_name));
+ ICEBERG_CHECK(ref != metadata->refs.end(), "Cannot find snapshot reference '{}'",
+ ref_name);
+ ICEBERG_PRECHECK(ref->second != nullptr, "Snapshot reference '{}' is null", ref_name);
+ return metadata->SnapshotById(ref->second->snapshot_id);
+}
+
+Result<bool> IsAncestorOf(const Table& table, int64_t ancestor_id,
+ const std::shared_ptr<Snapshot>& head) {
+ std::unordered_set<int64_t> visited;
+ auto current = head;
+ while (current != nullptr) {
+ if (!visited.insert(current->snapshot_id).second) {
+ return Invalid("Cycle detected in snapshot ancestry at {}", current->snapshot_id);
+ }
+ if (current->snapshot_id == ancestor_id) {
+ return true;
+ }
+ if (!current->parent_snapshot_id.has_value()) {
+ break;
+ }
+ auto parent = table.SnapshotById(*current->parent_snapshot_id);
+ if (!parent.has_value()) {
+ if (parent.error().kind == ErrorKind::kNotFound) {
+ break;
+ }
+ return std::unexpected<Error>(parent.error());
+ }
+ current = std::move(parent).value();
+ }
+ return false;
+}
+
+Status AppendLiteral(ArrowArray* array, const Literal& literal) {
+ if (literal.IsNull()) {
+ return AppendNull(array);
+ }
+ if (literal.IsAboveMax() || literal.IsBelowMin()) {
+ return InvalidArgument("Cannot append non-value partition literal {}",
+ literal.ToString());
+ }
+
+ switch (literal.type()->type_id()) {
+ case TypeId::kBoolean:
+ return AppendBoolean(array, std::get<bool>(literal.value()));
+ case TypeId::kInt:
+ case TypeId::kDate:
+ return AppendInt(array, std::get<int32_t>(literal.value()));
+ case TypeId::kLong:
+ case TypeId::kTime:
+ case TypeId::kTimestamp:
+ case TypeId::kTimestampTz:
+ case TypeId::kTimestampNs:
+ case TypeId::kTimestampTzNs:
+ return AppendInt(array, std::get<int64_t>(literal.value()));
+ case TypeId::kFloat:
+ return AppendDouble(array, std::get<float>(literal.value()));
+ case TypeId::kDouble:
+ return AppendDouble(array, std::get<double>(literal.value()));
+ case TypeId::kString:
+ return AppendString(array, std::get<std::string>(literal.value()));
+ case TypeId::kBinary:
+ case TypeId::kFixed:
+ return AppendBytes(array, std::get<std::vector<uint8_t>>(literal.value()));
+ case TypeId::kDecimal:
+ return AppendBytes(array, std::get<Decimal>(literal.value()).ToBytes());
Review Comment:
Decimal literals are sent through `AppendBytes`, but the generated Arrow field for `TypeId::kDecimal` is `decimal128`, not a binary array. A decimal partition value in the files/partitions tables or a decimal lower/upper metric bound therefore cannot be appended correctly, causing those scans to fail for tables with decimal columns. Add a decimal-aware append helper that writes the 128-bit value with the schema precision/scale and use it here (and for readable metrics).
##########
src/iceberg/inspect/files_table.cc:
##########
@@ -0,0 +1,75 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+#include "iceberg/inspect/files_table.h"
+
+#include <memory>
+#include <utility>
+
+#include "iceberg/inspect/metadata_table_stream_internal.h"
+#include "iceberg/inspect/metadata_table_util_internal.h"
+#include "iceberg/schema.h"
+#include "iceberg/table.h"
+#include "iceberg/type.h"
+#include "iceberg/util/macros.h"
+
+namespace iceberg {
+
+FilesTable::FilesTable(std::shared_ptr<Table> table, std::shared_ptr<Schema> schema,
+ std::shared_ptr<Schema> table_schema,
+ std::shared_ptr<StructType> partition_type)
+ : TimeTravelMetadataTable(std::move(table)),
+ schema_(std::move(schema)),
+ table_schema_(std::move(table_schema)),
+ partition_type_(std::move(partition_type)) {}
+
+FilesTable::~FilesTable() = default;
+
+const std::shared_ptr<Schema>& FilesTable::schema() const { return schema_; }
+
+Result<std::unique_ptr<FilesTable>> FilesTable::Make(std::shared_ptr<Table> table) {
+ ICEBERG_PRECHECK(table != nullptr, "Table cannot be null");
+ ICEBERG_ASSIGN_OR_RAISE(auto table_schema, table->schema());
+ ICEBERG_ASSIGN_OR_RAISE(auto partition_type, internal::UnifiedPartitionType(*table));
+ ICEBERG_ASSIGN_OR_RAISE(auto schema,
+ internal::FilesTableSchema(*table_schema, partition_type));
+ return std::unique_ptr<FilesTable>(new FilesTable(std::move(table), std::move(schema),
+ std::move(table_schema),
+ std::move(partition_type)));
+}
+
+Result<ArrowArrayStream> FilesTable::ScanSnapshot(
+ const SnapshotSelection& snapshot_selection) {
+ ICEBERG_ASSIGN_OR_RAISE(auto snapshot, internal::ResolveMetadataTableSnapshot(
+ *source_table(), snapshot_selection));
+ ICEBERG_ASSIGN_OR_RAISE(auto files, internal::LoadLiveFiles(*source_table(), snapshot));
Review Comment:
Although the returned Arrow batches are capped, this call eagerly reads every live manifest entry into one `files` vector before the stream is returned. A table with a large number of live files therefore performs all I/O and allocates O(number of files) memory before the first batch, which defeats the streaming path and can cause excessive memory use. Traverse manifests from the stream and materialize only the next bounded batch (or otherwise bound pending rows).
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat(inspect): add system metadata tables [iceberg-cpp]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #876:
URL: https://github.com/apache/iceberg-cpp/pull/876#discussion_r4238339079
##########
src/iceberg/inspect/partitions_table.cc:
##########
@@ -0,0 +1,324 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+#include "iceberg/inspect/partitions_table.h"
+
+#include <algorithm>
+#include <chrono>
+#include <cstddef>
+#include <cstdint>
+#include <memory>
+#include <optional>
+#include <ranges>
+#include <unordered_map>
+#include <utility>
+#include <vector>
+
+#include "iceberg/arrow_row_builder_internal.h"
+#include "iceberg/expression/literal.h"
+#include "iceberg/inspect/metadata_table_stream_internal.h"
+#include "iceberg/inspect/metadata_table_util_internal.h"
+#include "iceberg/manifest/manifest_entry.h"
+#include "iceberg/partition_spec.h"
+#include "iceberg/row/partition_values.h"
+#include "iceberg/schema.h"
+#include "iceberg/schema_field.h"
+#include "iceberg/snapshot.h"
+#include "iceberg/table.h"
+#include "iceberg/type.h"
+#include "iceberg/util/macros.h"
+
+namespace iceberg {
+namespace {
+
+constexpr int32_t kPartitionFieldId = 1;
+constexpr int32_t kRecordCountFieldId = 2;
+constexpr int32_t kFileCountFieldId = 3;
+constexpr int32_t kSpecIdFieldId = 4;
+constexpr int32_t kPositionDeleteRecordCountFieldId = 5;
+constexpr int32_t kPositionDeleteFileCountFieldId = 6;
+constexpr int32_t kEqualityDeleteRecordCountFieldId = 7;
+constexpr int32_t kEqualityDeleteFileCountFieldId = 8;
+constexpr int32_t kLastUpdatedAtFieldId = 9;
+constexpr int32_t kLastUpdatedSnapshotIdFieldId = 10;
+constexpr int32_t kTotalDataFileSizeFieldId = 11;
+
+struct PartitionKey {
+ PartitionValues values;
+ size_t projected_fields;
+
+ bool operator==(const PartitionKey& other) const {
+ if (projected_fields != other.projected_fields ||
+ values.num_fields() != other.values.num_fields()) {
Review Comment:
`PartitionKey` does not include the partition spec identity, so files from different specs can collapse into one aggregate whenever their projected field IDs and values match. The resulting row then combines counts from both specs while exposing only whichever `spec_id` happened to update `PartitionStats` last; keep the spec ID (or equivalent spec identity) in the grouping key so the `spec_id` column remains unambiguous.
##########
src/iceberg/inspect/metadata_table_util_internal.cc:
##########
@@ -0,0 +1,624 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+#include "iceberg/inspect/metadata_table_util_internal.h"
+
+#include <algorithm>
+#include <cstdint>
+#include <format>
+#include <functional>
+#include <map>
+#include <memory>
+#include <ranges>
+#include <string>
+#include <string_view>
+#include <unordered_map>
+#include <unordered_set>
+#include <utility>
+#include <variant>
+#include <vector>
+
+#include <nanoarrow/nanoarrow.h>
+
+#include "iceberg/arrow_row_builder_internal.h"
+#include "iceberg/constants.h"
+#include "iceberg/file_format.h"
+#include "iceberg/manifest/manifest_list.h"
+#include "iceberg/manifest/manifest_reader.h"
+#include "iceberg/nanoarrow_status_internal.h"
+#include "iceberg/partition_spec.h"
+#include "iceberg/schema.h"
+#include "iceberg/schema_field.h"
+#include "iceberg/snapshot.h"
+#include "iceberg/table.h"
+#include "iceberg/table_metadata.h"
+#include "iceberg/transform.h"
+#include "iceberg/type.h"
+#include "iceberg/util/checked_cast.h"
+#include "iceberg/util/conversions.h"
+#include "iceberg/util/macros.h"
+#include "iceberg/util/snapshot_util.h"
+
+namespace iceberg::internal {
+namespace {
+
+Result<std::shared_ptr<Snapshot>> SnapshotAtRef(const Table& table,
+ std::string_view ref_name) {
+ const auto& metadata = table.metadata();
+ ICEBERG_PRECHECK(metadata != nullptr, "Table metadata cannot be null");
+
+ if (ref_name.empty() || ref_name == SnapshotRef::kMainBranch) {
+ if (metadata->current_snapshot_id == kInvalidSnapshotId) {
+ return std::shared_ptr<Snapshot>{nullptr};
+ }
+ return metadata->SnapshotById(metadata->current_snapshot_id);
+ }
+
+ auto ref = metadata->refs.find(std::string(ref_name));
+ ICEBERG_PRECHECK(ref != metadata->refs.end(), "Cannot find snapshot reference '{}'",
+ ref_name);
+ ICEBERG_PRECHECK(ref->second != nullptr, "Snapshot reference '{}' is null", ref_name);
+ return metadata->SnapshotById(ref->second->snapshot_id);
+}
+
+Status AppendLiteral(ArrowArray* array, const Literal& literal) {
+ if (literal.IsNull()) {
+ return AppendNull(array);
+ }
+ if (literal.IsAboveMax() || literal.IsBelowMin()) {
+ return InvalidArgument("Cannot append non-value partition literal {}",
+ literal.ToString());
+ }
+
+ switch (literal.type()->type_id()) {
+ case TypeId::kBoolean:
+ return AppendBoolean(array, std::get<bool>(literal.value()));
+ case TypeId::kInt:
+ case TypeId::kDate:
+ return AppendInt(array, std::get<int32_t>(literal.value()));
+ case TypeId::kLong:
+ case TypeId::kTime:
+ case TypeId::kTimestamp:
+ case TypeId::kTimestampTz:
+ case TypeId::kTimestampNs:
+ case TypeId::kTimestampTzNs:
+ return AppendInt(array, std::get<int64_t>(literal.value()));
+ case TypeId::kFloat:
+ return AppendDouble(array, std::get<float>(literal.value()));
+ case TypeId::kDouble:
+ return AppendDouble(array, std::get<double>(literal.value()));
+ case TypeId::kString:
+ return AppendString(array, std::get<std::string>(literal.value()));
+ case TypeId::kBinary:
+ case TypeId::kFixed:
+ return AppendBytes(array, std::get<std::vector<uint8_t>>(literal.value()));
+ case TypeId::kDecimal:
+ return AppendDecimal(array, std::get<Decimal>(literal.value()));
+ case TypeId::kUuid:
+ return AppendBytes(array, std::get<Uuid>(literal.value()).bytes());
+ case TypeId::kUnknown:
+ case TypeId::kStruct:
+ case TypeId::kList:
+ case TypeId::kMap:
+ case TypeId::kVariant:
+ case TypeId::kGeometry:
+ case TypeId::kGeography:
+ return NotSupported("Cannot append partition literal of type {}",
+ literal.type()->ToString());
+ }
+ std::unreachable();
+}
+
+template <typename T>
+Status AppendOptionalInt(ArrowArray* array, const std::optional<T>& value) {
+ if (!value.has_value()) {
+ return AppendNull(array);
+ }
+ return AppendInt(array, static_cast<int64_t>(*value));
+}
+
+constexpr std::string_view MetadataFileFormat(FileFormatType format) {
+ switch (format) {
+ case FileFormatType::kParquet:
+ return "PARQUET";
+ case FileFormatType::kAvro:
+ return "AVRO";
+ case FileFormatType::kOrc:
+ return "ORC";
+ case FileFormatType::kPuffin:
+ return "PUFFIN";
+ }
+ std::unreachable();
+}
+
+void CollectPrimitiveFields(
+ const NestedType& type,
+ std::vector<std::reference_wrapper<const SchemaField>>& primitive_fields) {
+ for (const auto& field : type.fields()) {
+ if (field.type()->is_primitive()) {
+ primitive_fields.emplace_back(field);
+ } else if (field.type()->is_nested()) {
+ CollectPrimitiveFields(*std::static_pointer_cast<NestedType>(field.type()),
+ primitive_fields);
+ }
+ }
+}
+
+Result<SchemaField> ReadableMetricsField(const Schema& table_schema,
+ int32_t highest_metadata_field_id) {
+ std::vector<std::reference_wrapper<const SchemaField>> primitive_fields;
+ CollectPrimitiveFields(table_schema, primitive_fields);
+
+ int32_t next_id = highest_metadata_field_id;
+ std::vector<SchemaField> column_metrics;
+ column_metrics.reserve(primitive_fields.size());
+ for (const auto& field_ref : primitive_fields) {
+ const auto& field = field_ref.get();
+ ICEBERG_ASSIGN_OR_RAISE(auto column_name,
+ table_schema.FindColumnNameById(field.field_id()));
+ ICEBERG_PRECHECK(column_name.has_value(), "Cannot find name for field {}",
+ field.field_id());
+
+ const int32_t column_metrics_id = ++next_id;
+ std::vector<SchemaField> metrics{
+ SchemaField::MakeOptional(++next_id, "column_size", int64(),
+ "Total size on disk"),
+ SchemaField::MakeOptional(++next_id, "value_count", int64(),
+ "Total count, including null and NaN"),
+ SchemaField::MakeOptional(++next_id, "null_value_count", int64(),
+ "Null value count"),
+ SchemaField::MakeOptional(++next_id, "nan_value_count", int64(),
+ "NaN value count"),
+ SchemaField::MakeOptional(++next_id, "lower_bound", field.type(), "Lower bound"),
+ SchemaField::MakeOptional(++next_id, "upper_bound", field.type(), "Upper bound"),
+ };
+ column_metrics.emplace_back(
+ column_metrics_id, *column_name, struct_(std::move(metrics)),
+ /*optional=*/true, std::format("Metrics for column {}", *column_name));
+ }
+
+ std::ranges::sort(column_metrics, {},
+ [](const SchemaField& field) { return field.name(); });
+ return SchemaField::MakeOptional(++next_id, "readable_metrics",
+ struct_(std::move(column_metrics)),
+ "Column metrics in readable form");
+}
+
+Status AppendMetric(ArrowArray* array, const std::map<int32_t, int64_t>& metrics,
+ int32_t field_id) {
+ auto metric = metrics.find(field_id);
+ return metric == metrics.end() ? AppendNull(array) : AppendInt(array, metric->second);
+}
+
+Status AppendBound(ArrowArray* array,
+ const std::map<int32_t, std::vector<uint8_t>>& bounds,
+ const SchemaField& field) {
+ auto bound = bounds.find(field.field_id());
+ if (bound == bounds.end()) {
+ return AppendNull(array);
+ }
+ auto primitive = checked_pointer_cast<PrimitiveType>(field.type());
+ ICEBERG_ASSIGN_OR_RAISE(auto literal,
+ Conversions::FromBytes(std::move(primitive), bound->second));
+ return AppendLiteral(array, literal);
+}
+
+Status AppendReadableMetrics(ArrowArray* array, const StructType& readable_type,
+ const Schema& table_schema, const DataFile& file) {
+ ICEBERG_PRECHECK(array != nullptr, "Readable metrics Arrow array cannot be null");
+ ICEBERG_PRECHECK(
+ array->n_children == static_cast<int64_t>(readable_type.fields().size()),
+ "Readable metrics Arrow array has {} fields but schema has {}", array->n_children,
+ readable_type.fields().size());
+ for (int64_t index = 0; index < array->n_children; ++index) {
+ auto* column_metrics = array->children[index];
+ ICEBERG_PRECHECK(column_metrics != nullptr && column_metrics->n_children == 6,
+ "Readable column metrics must contain six fields");
+ const auto readable_name = readable_type.fields()[index].name();
+ ICEBERG_ASSIGN_OR_RAISE(auto source_field,
+ table_schema.FindFieldByName(readable_name));
+ ICEBERG_PRECHECK(source_field.has_value(),
+ "Cannot find readable metrics source field '{}'", readable_name);
+ const auto& field = source_field->get();
+ ICEBERG_PRECHECK(field.type()->is_primitive(),
+ "Readable metrics source field '{}' must be primitive",
+ readable_name);
+
+ ICEBERG_RETURN_UNEXPECTED(
+ AppendMetric(column_metrics->children[0], file.column_sizes, field.field_id()));
+ ICEBERG_RETURN_UNEXPECTED(
+ AppendMetric(column_metrics->children[1], file.value_counts, field.field_id()));
+ ICEBERG_RETURN_UNEXPECTED(AppendMetric(column_metrics->children[2],
+ file.null_value_counts, field.field_id()));
+ ICEBERG_RETURN_UNEXPECTED(AppendMetric(column_metrics->children[3],
+ file.nan_value_counts, field.field_id()));
+ ICEBERG_RETURN_UNEXPECTED(
+ AppendBound(column_metrics->children[4], file.lower_bounds, field));
+ ICEBERG_RETURN_UNEXPECTED(
+ AppendBound(column_metrics->children[5], file.upper_bounds, field));
+ ICEBERG_NANOARROW_RETURN_UNEXPECTED(ArrowArrayFinishElement(column_metrics));
+ }
+ ICEBERG_NANOARROW_RETURN_UNEXPECTED(ArrowArrayFinishElement(array));
+ return {};
+}
+
+} // namespace
+
+Result<std::shared_ptr<Snapshot>> ResolveMetadataTableSnapshot(
+ const Table& table, const SnapshotSelection& selection) {
+ if (!selection.ref_name.empty() && selection.ref_name != SnapshotRef::kMainBranch) {
+ ICEBERG_PRECHECK(
+ std::holds_alternative<std::monostate>(selection.snapshot),
+ "Cannot combine a snapshot reference with a snapshot ID or timestamp");
+ }
+
+ if (std::holds_alternative<std::monostate>(selection.snapshot)) {
+ return SnapshotAtRef(table, selection.ref_name);
+ }
+ if (const auto* snapshot_id = std::get_if<int64_t>(&selection.snapshot)) {
+ return table.SnapshotById(*snapshot_id);
+ }
+
+ ICEBERG_ASSIGN_OR_RAISE(
+ auto snapshot_id,
+ SnapshotUtil::SnapshotIdAsOfTime(table, std::get<TimePointMs>(selection.snapshot)));
+ return table.SnapshotById(snapshot_id);
+}
+
+Result<std::shared_ptr<StructType>> UnifiedPartitionType(const Table& table) {
+ ICEBERG_ASSIGN_OR_RAISE(auto schema, table.schema());
+ ICEBERG_ASSIGN_OR_RAISE(auto specs_ref, table.specs());
+
+ std::vector<std::shared_ptr<PartitionSpec>> specs;
+ specs.reserve(specs_ref.get().size());
+ for (const auto& [_, spec] : specs_ref.get()) {
+ ICEBERG_PRECHECK(spec != nullptr, "Partition spec cannot be null");
+ specs.push_back(spec);
+ }
+ std::ranges::sort(specs, std::greater{}, &PartitionSpec::spec_id);
+
+ std::unordered_set<int32_t> active_field_ids;
+ for (const auto& spec : specs) {
+ for (const auto& field : spec->fields()) {
+ ICEBERG_PRECHECK(field.transform() != nullptr,
+ "Partition field {} has a null transform", field.field_id());
+ ICEBERG_CHECK(field.transform()->transform_type() != TransformType::kUnknown,
+ "Cannot build table partition type with unknown transform '{}'",
+ field.transform()->ToString());
+ ICEBERG_ASSIGN_OR_RAISE(auto source_field,
+ schema->FindFieldById(field.source_id()));
+ if (source_field.has_value()) {
+ active_field_ids.insert(field.field_id());
+ }
+ }
+ }
+
+ struct ProjectedField {
+ const PartitionField* definition;
+ std::string name;
+ std::shared_ptr<Type> type;
+ };
+ std::map<int32_t, ProjectedField> fields_by_id;
+ for (const auto& spec : specs) {
+ ICEBERG_ASSIGN_OR_RAISE(auto spec_type, spec->PartitionType(*schema));
+ ICEBERG_PRECHECK(spec_type->fields().size() == spec->fields().size(),
+ "Partition spec {} has mismatched field and type counts",
+ spec->spec_id());
+ for (size_t index = 0; index < spec->fields().size(); ++index) {
+ const auto& partition_field = spec->fields()[index];
+ if (!active_field_ids.contains(partition_field.field_id())) {
+ continue;
+ }
+
+ const auto& spec_field = spec_type->fields()[index];
+ auto [iter, inserted] =
+ fields_by_id.try_emplace(partition_field.field_id(),
+ ProjectedField{.definition = &partition_field,
+ .name = std::string(spec_field.name()),
+ .type = spec_field.type()});
+ if (!inserted) {
+ const auto& existing = *iter->second.definition;
+ const auto current_transform = partition_field.transform()->transform_type();
+ const auto existing_transform = existing.transform()->transform_type();
+ const bool compatible_transform =
+ *partition_field.transform() == *existing.transform() ||
+ current_transform == TransformType::kVoid ||
+ existing_transform == TransformType::kVoid;
+ ICEBERG_CHECK(
+ partition_field.source_id() == existing.source_id() && compatible_transform,
+ "Conflicting partition fields with ID {}: '{}' and '{}'",
+ partition_field.field_id(), partition_field.ToString(), existing.ToString());
+
+ if (existing_transform == TransformType::kVoid &&
+ current_transform != TransformType::kVoid) {
+ iter->second.definition = &partition_field;
+ iter->second.type = spec_field.type();
+ }
Review Comment:
The specs are sorted newest-first above, so this branch replaces the newest `void` definition with the older non-void definition. In the legacy-void evolution covered by `SupportsLegacyVoidPartitionEvolution`, that makes the files schema expose `old_bucket` instead of the latest `new_name` (and can select the older type). Keep the existing definition when `existing_transform` is `kVoid`, or otherwise preserve the newest spec's field definition.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat(inspect): add files, partitions, and manifests metadata tables [iceberg-cpp]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #876:
URL: https://github.com/apache/iceberg-cpp/pull/876#discussion_r4259730722
##########
src/iceberg/inspect/metadata_table_util_internal.cc:
##########
@@ -0,0 +1,624 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+#include "iceberg/inspect/metadata_table_util_internal.h"
+
+#include <algorithm>
+#include <cstdint>
+#include <format>
+#include <functional>
+#include <map>
+#include <memory>
+#include <ranges>
+#include <string>
+#include <string_view>
+#include <unordered_map>
+#include <unordered_set>
+#include <utility>
+#include <variant>
+#include <vector>
+
+#include <nanoarrow/nanoarrow.h>
+
+#include "iceberg/arrow_row_builder_internal.h"
+#include "iceberg/constants.h"
+#include "iceberg/file_format.h"
+#include "iceberg/manifest/manifest_list.h"
+#include "iceberg/manifest/manifest_reader.h"
+#include "iceberg/nanoarrow_status_internal.h"
+#include "iceberg/partition_spec.h"
+#include "iceberg/schema.h"
+#include "iceberg/schema_field.h"
+#include "iceberg/snapshot.h"
+#include "iceberg/table.h"
+#include "iceberg/table_metadata.h"
+#include "iceberg/transform.h"
+#include "iceberg/type.h"
+#include "iceberg/util/checked_cast.h"
+#include "iceberg/util/conversions.h"
+#include "iceberg/util/macros.h"
+#include "iceberg/util/snapshot_util.h"
+
+namespace iceberg::internal {
+namespace {
+
+Result<std::shared_ptr<Snapshot>> SnapshotAtRef(const Table& table,
+ std::string_view ref_name) {
+ const auto& metadata = table.metadata();
+ ICEBERG_PRECHECK(metadata != nullptr, "Table metadata cannot be null");
+
+ if (ref_name.empty() || ref_name == SnapshotRef::kMainBranch) {
+ if (metadata->current_snapshot_id == kInvalidSnapshotId) {
+ return std::shared_ptr<Snapshot>{nullptr};
+ }
+ return metadata->SnapshotById(metadata->current_snapshot_id);
+ }
+
+ auto ref = metadata->refs.find(std::string(ref_name));
+ ICEBERG_PRECHECK(ref != metadata->refs.end(), "Cannot find snapshot reference '{}'",
+ ref_name);
+ ICEBERG_PRECHECK(ref->second != nullptr, "Snapshot reference '{}' is null", ref_name);
+ return metadata->SnapshotById(ref->second->snapshot_id);
+}
+
+Status AppendLiteral(ArrowArray* array, const Literal& literal) {
+ if (literal.IsNull()) {
+ return AppendNull(array);
+ }
+ if (literal.IsAboveMax() || literal.IsBelowMin()) {
+ return InvalidArgument("Cannot append non-value partition literal {}",
+ literal.ToString());
+ }
+
+ switch (literal.type()->type_id()) {
+ case TypeId::kBoolean:
+ return AppendBoolean(array, std::get<bool>(literal.value()));
+ case TypeId::kInt:
+ case TypeId::kDate:
+ return AppendInt(array, std::get<int32_t>(literal.value()));
+ case TypeId::kLong:
+ case TypeId::kTime:
+ case TypeId::kTimestamp:
+ case TypeId::kTimestampTz:
+ case TypeId::kTimestampNs:
+ case TypeId::kTimestampTzNs:
+ return AppendInt(array, std::get<int64_t>(literal.value()));
+ case TypeId::kFloat:
+ return AppendDouble(array, std::get<float>(literal.value()));
+ case TypeId::kDouble:
+ return AppendDouble(array, std::get<double>(literal.value()));
+ case TypeId::kString:
+ return AppendString(array, std::get<std::string>(literal.value()));
+ case TypeId::kBinary:
+ case TypeId::kFixed:
+ return AppendBytes(array, std::get<std::vector<uint8_t>>(literal.value()));
+ case TypeId::kDecimal:
+ return AppendDecimal(array, std::get<Decimal>(literal.value()));
+ case TypeId::kUuid:
+ return AppendBytes(array, std::get<Uuid>(literal.value()).bytes());
+ case TypeId::kUnknown:
+ case TypeId::kStruct:
+ case TypeId::kList:
+ case TypeId::kMap:
+ case TypeId::kVariant:
+ case TypeId::kGeometry:
+ case TypeId::kGeography:
+ return NotSupported("Cannot append partition literal of type {}",
+ literal.type()->ToString());
+ }
+ std::unreachable();
+}
+
+template <typename T>
+Status AppendOptionalInt(ArrowArray* array, const std::optional<T>& value) {
+ if (!value.has_value()) {
+ return AppendNull(array);
+ }
+ return AppendInt(array, static_cast<int64_t>(*value));
+}
+
+constexpr std::string_view MetadataFileFormat(FileFormatType format) {
+ switch (format) {
+ case FileFormatType::kParquet:
+ return "PARQUET";
+ case FileFormatType::kAvro:
+ return "AVRO";
+ case FileFormatType::kOrc:
+ return "ORC";
+ case FileFormatType::kPuffin:
+ return "PUFFIN";
+ }
+ std::unreachable();
+}
+
+void CollectPrimitiveFields(
+ const NestedType& type,
+ std::vector<std::reference_wrapper<const SchemaField>>& primitive_fields) {
+ for (const auto& field : type.fields()) {
+ if (field.type()->is_primitive()) {
+ primitive_fields.emplace_back(field);
+ } else if (field.type()->is_nested()) {
+ CollectPrimitiveFields(*std::static_pointer_cast<NestedType>(field.type()),
+ primitive_fields);
+ }
+ }
+}
+
+Result<SchemaField> ReadableMetricsField(const Schema& table_schema,
+ int32_t highest_metadata_field_id) {
+ std::vector<std::reference_wrapper<const SchemaField>> primitive_fields;
+ CollectPrimitiveFields(table_schema, primitive_fields);
+
+ int32_t next_id = highest_metadata_field_id;
+ std::vector<SchemaField> column_metrics;
+ column_metrics.reserve(primitive_fields.size());
+ for (const auto& field_ref : primitive_fields) {
+ const auto& field = field_ref.get();
+ ICEBERG_ASSIGN_OR_RAISE(auto column_name,
+ table_schema.FindColumnNameById(field.field_id()));
+ ICEBERG_PRECHECK(column_name.has_value(), "Cannot find name for field {}",
+ field.field_id());
+
+ const int32_t column_metrics_id = ++next_id;
+ std::vector<SchemaField> metrics{
+ SchemaField::MakeOptional(++next_id, "column_size", int64(),
+ "Total size on disk"),
+ SchemaField::MakeOptional(++next_id, "value_count", int64(),
+ "Total count, including null and NaN"),
+ SchemaField::MakeOptional(++next_id, "null_value_count", int64(),
+ "Null value count"),
+ SchemaField::MakeOptional(++next_id, "nan_value_count", int64(),
+ "NaN value count"),
+ SchemaField::MakeOptional(++next_id, "lower_bound", field.type(), "Lower bound"),
+ SchemaField::MakeOptional(++next_id, "upper_bound", field.type(), "Upper bound"),
+ };
+ column_metrics.emplace_back(
+ column_metrics_id, *column_name, struct_(std::move(metrics)),
+ /*optional=*/true, std::format("Metrics for column {}", *column_name));
+ }
+
+ std::ranges::sort(column_metrics, {},
+ [](const SchemaField& field) { return field.name(); });
+ return SchemaField::MakeOptional(++next_id, "readable_metrics",
+ struct_(std::move(column_metrics)),
+ "Column metrics in readable form");
+}
+
+Status AppendMetric(ArrowArray* array, const std::map<int32_t, int64_t>& metrics,
+ int32_t field_id) {
+ auto metric = metrics.find(field_id);
+ return metric == metrics.end() ? AppendNull(array) : AppendInt(array, metric->second);
+}
+
+Status AppendBound(ArrowArray* array,
+ const std::map<int32_t, std::vector<uint8_t>>& bounds,
+ const SchemaField& field) {
+ auto bound = bounds.find(field.field_id());
+ if (bound == bounds.end()) {
+ return AppendNull(array);
+ }
+ auto primitive = checked_pointer_cast<PrimitiveType>(field.type());
+ ICEBERG_ASSIGN_OR_RAISE(auto literal,
+ Conversions::FromBytes(std::move(primitive), bound->second));
+ return AppendLiteral(array, literal);
+}
+
+Status AppendReadableMetrics(ArrowArray* array, const StructType& readable_type,
+ const Schema& table_schema, const DataFile& file) {
+ ICEBERG_PRECHECK(array != nullptr, "Readable metrics Arrow array cannot be null");
+ ICEBERG_PRECHECK(
+ array->n_children == static_cast<int64_t>(readable_type.fields().size()),
+ "Readable metrics Arrow array has {} fields but schema has {}", array->n_children,
+ readable_type.fields().size());
+ for (int64_t index = 0; index < array->n_children; ++index) {
+ auto* column_metrics = array->children[index];
+ ICEBERG_PRECHECK(column_metrics != nullptr && column_metrics->n_children == 6,
+ "Readable column metrics must contain six fields");
+ const auto readable_name = readable_type.fields()[index].name();
+ ICEBERG_ASSIGN_OR_RAISE(auto source_field,
+ table_schema.FindFieldByName(readable_name));
+ ICEBERG_PRECHECK(source_field.has_value(),
+ "Cannot find readable metrics source field '{}'", readable_name);
+ const auto& field = source_field->get();
+ ICEBERG_PRECHECK(field.type()->is_primitive(),
+ "Readable metrics source field '{}' must be primitive",
+ readable_name);
+
+ ICEBERG_RETURN_UNEXPECTED(
+ AppendMetric(column_metrics->children[0], file.column_sizes, field.field_id()));
+ ICEBERG_RETURN_UNEXPECTED(
+ AppendMetric(column_metrics->children[1], file.value_counts, field.field_id()));
+ ICEBERG_RETURN_UNEXPECTED(AppendMetric(column_metrics->children[2],
+ file.null_value_counts, field.field_id()));
+ ICEBERG_RETURN_UNEXPECTED(AppendMetric(column_metrics->children[3],
+ file.nan_value_counts, field.field_id()));
+ ICEBERG_RETURN_UNEXPECTED(
+ AppendBound(column_metrics->children[4], file.lower_bounds, field));
+ ICEBERG_RETURN_UNEXPECTED(
+ AppendBound(column_metrics->children[5], file.upper_bounds, field));
+ ICEBERG_NANOARROW_RETURN_UNEXPECTED(ArrowArrayFinishElement(column_metrics));
+ }
+ ICEBERG_NANOARROW_RETURN_UNEXPECTED(ArrowArrayFinishElement(array));
+ return {};
+}
+
+} // namespace
+
+Result<std::shared_ptr<Snapshot>> ResolveMetadataTableSnapshot(
+ const Table& table, const SnapshotSelection& selection) {
+ if (!selection.ref_name.empty() && selection.ref_name != SnapshotRef::kMainBranch) {
+ ICEBERG_PRECHECK(
+ std::holds_alternative<std::monostate>(selection.snapshot),
+ "Cannot combine a snapshot reference with a snapshot ID or timestamp");
+ }
+
+ if (std::holds_alternative<std::monostate>(selection.snapshot)) {
+ return SnapshotAtRef(table, selection.ref_name);
+ }
+ if (const auto* snapshot_id = std::get_if<int64_t>(&selection.snapshot)) {
+ return table.SnapshotById(*snapshot_id);
+ }
+
+ ICEBERG_ASSIGN_OR_RAISE(
+ auto snapshot_id,
+ SnapshotUtil::SnapshotIdAsOfTime(table, std::get<TimePointMs>(selection.snapshot)));
+ return table.SnapshotById(snapshot_id);
+}
+
+Result<std::shared_ptr<StructType>> UnifiedPartitionType(const Table& table) {
+ ICEBERG_ASSIGN_OR_RAISE(auto schema, table.schema());
+ ICEBERG_ASSIGN_OR_RAISE(auto specs_ref, table.specs());
+
+ std::vector<std::shared_ptr<PartitionSpec>> specs;
+ specs.reserve(specs_ref.get().size());
+ for (const auto& [_, spec] : specs_ref.get()) {
+ ICEBERG_PRECHECK(spec != nullptr, "Partition spec cannot be null");
+ specs.push_back(spec);
+ }
+ std::ranges::sort(specs, std::greater{}, &PartitionSpec::spec_id);
+
+ std::unordered_set<int32_t> active_field_ids;
+ for (const auto& spec : specs) {
+ for (const auto& field : spec->fields()) {
+ ICEBERG_PRECHECK(field.transform() != nullptr,
+ "Partition field {} has a null transform", field.field_id());
+ ICEBERG_CHECK(field.transform()->transform_type() != TransformType::kUnknown,
+ "Cannot build table partition type with unknown transform '{}'",
+ field.transform()->ToString());
+ ICEBERG_ASSIGN_OR_RAISE(auto source_field,
+ schema->FindFieldById(field.source_id()));
+ if (source_field.has_value()) {
+ active_field_ids.insert(field.field_id());
+ }
+ }
+ }
+
+ struct ProjectedField {
+ const PartitionField* definition;
+ std::string name;
+ std::shared_ptr<Type> type;
+ };
+ std::map<int32_t, ProjectedField> fields_by_id;
+ for (const auto& spec : specs) {
+ ICEBERG_ASSIGN_OR_RAISE(auto spec_type, spec->PartitionType(*schema));
+ ICEBERG_PRECHECK(spec_type->fields().size() == spec->fields().size(),
+ "Partition spec {} has mismatched field and type counts",
+ spec->spec_id());
+ for (size_t index = 0; index < spec->fields().size(); ++index) {
+ const auto& partition_field = spec->fields()[index];
+ if (!active_field_ids.contains(partition_field.field_id())) {
+ continue;
+ }
+
+ const auto& spec_field = spec_type->fields()[index];
+ auto [iter, inserted] =
+ fields_by_id.try_emplace(partition_field.field_id(),
+ ProjectedField{.definition = &partition_field,
+ .name = std::string(spec_field.name()),
+ .type = spec_field.type()});
+ if (!inserted) {
+ const auto& existing = *iter->second.definition;
+ const auto current_transform = partition_field.transform()->transform_type();
+ const auto existing_transform = existing.transform()->transform_type();
+ const bool compatible_transform =
+ *partition_field.transform() == *existing.transform() ||
+ current_transform == TransformType::kVoid ||
+ existing_transform == TransformType::kVoid;
+ ICEBERG_CHECK(
+ partition_field.source_id() == existing.source_id() && compatible_transform,
+ "Conflicting partition fields with ID {}: '{}' and '{}'",
+ partition_field.field_id(), partition_field.ToString(), existing.ToString());
+
+ if (existing_transform == TransformType::kVoid &&
+ current_transform != TransformType::kVoid) {
+ iter->second.definition = &partition_field;
+ iter->second.type = spec_field.type();
+ }
+ }
+ }
+ }
+
+ std::vector<SchemaField> fields;
+ fields.reserve(fields_by_id.size());
+ for (auto& [_, field] : fields_by_id) {
+ fields.emplace_back(field.definition->field_id(), std::move(field.name),
+ std::move(field.type), /*optional=*/true);
+ }
+ return std::make_shared<StructType>(std::move(fields));
+}
+
+Result<std::shared_ptr<Schema>> FilesTableSchema(
+ const Schema& table_schema, const std::shared_ptr<StructType>& partition_type) {
+ ICEBERG_PRECHECK(partition_type != nullptr, "Partition type cannot be null");
+ auto data_file_type = DataFile::Type(partition_type);
+ std::vector<SchemaField> fields;
+ fields.reserve(data_file_type->fields().size() + 1);
+ for (const auto& field : data_file_type->fields()) {
+ fields.push_back(field);
+ if (field.field_id() == DataFile::kFileFormatFieldId) {
+ fields.push_back(DataFile::kSpecId);
+ }
+ }
+
+ if (partition_type->fields().empty()) {
+ std::erase_if(fields, [](const SchemaField& field) {
+ return field.field_id() == DataFile::kPartitionFieldId;
+ });
+ }
+ auto file_schema = std::make_shared<Schema>(std::move(fields));
+ ICEBERG_ASSIGN_OR_RAISE(auto highest_field_id, file_schema->HighestFieldId());
+ ICEBERG_ASSIGN_OR_RAISE(auto readable_metrics,
+ ReadableMetricsField(table_schema, highest_field_id));
+ fields = std::vector<SchemaField>(file_schema->fields().begin(),
+ file_schema->fields().end());
+ fields.push_back(std::move(readable_metrics));
+ return std::make_shared<Schema>(std::move(fields));
+}
+
+Result<PartitionValues> ProjectPartitionValues(const StructType& partition_type,
+ const PartitionSpec& spec,
+ const PartitionValues& values) {
+ ICEBERG_PRECHECK(values.num_fields() == spec.fields().size(),
+ "Partition has {} values but spec {} has {} fields",
+ values.num_fields(), spec.spec_id(), spec.fields().size());
+
+ std::unordered_map<int32_t, size_t> positions;
+ positions.reserve(spec.fields().size());
+ for (size_t index = 0; index < spec.fields().size(); ++index) {
+ positions.emplace(spec.fields()[index].field_id(), index);
+ }
+
+ std::vector<Literal> projected;
+ projected.reserve(partition_type.fields().size());
+ for (const auto& field : partition_type.fields()) {
+ auto target_type = checked_pointer_cast<PrimitiveType>(field.type());
+ auto position = positions.find(field.field_id());
+ if (position == positions.end()) {
+ projected.push_back(Literal::Null(std::move(target_type)));
+ continue;
+ }
+ ICEBERG_ASSIGN_OR_RAISE(auto value, values.ValueAt(position->second));
Review Comment:
`ProjectPartitionValues` rebuilds a `field_id -> index` hash map on every call. This runs once per live file (and can also be called indirectly from `AppendDataFile`), which can become a hotspot for large snapshots. Consider caching the per-spec lookup structure (e.g., a `spec_id -> positions` map built once per scan) and reusing it across calls, or restructuring the projection to avoid allocating a new `unordered_map` each time.
##########
src/iceberg/inspect/partitions_table.cc:
##########
@@ -0,0 +1,324 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+#include "iceberg/inspect/partitions_table.h"
+
+#include <algorithm>
+#include <chrono>
+#include <cstddef>
+#include <cstdint>
+#include <memory>
+#include <optional>
+#include <ranges>
+#include <unordered_map>
+#include <utility>
+#include <vector>
+
+#include "iceberg/arrow_row_builder_internal.h"
+#include "iceberg/expression/literal.h"
+#include "iceberg/inspect/metadata_table_stream_internal.h"
+#include "iceberg/inspect/metadata_table_util_internal.h"
+#include "iceberg/manifest/manifest_entry.h"
+#include "iceberg/partition_spec.h"
+#include "iceberg/row/partition_values.h"
+#include "iceberg/schema.h"
+#include "iceberg/schema_field.h"
+#include "iceberg/snapshot.h"
+#include "iceberg/table.h"
+#include "iceberg/type.h"
+#include "iceberg/util/macros.h"
+
+namespace iceberg {
+namespace {
+
+constexpr int32_t kPartitionFieldId = 1;
+constexpr int32_t kRecordCountFieldId = 2;
+constexpr int32_t kFileCountFieldId = 3;
+constexpr int32_t kSpecIdFieldId = 4;
+constexpr int32_t kPositionDeleteRecordCountFieldId = 5;
+constexpr int32_t kPositionDeleteFileCountFieldId = 6;
+constexpr int32_t kEqualityDeleteRecordCountFieldId = 7;
+constexpr int32_t kEqualityDeleteFileCountFieldId = 8;
+constexpr int32_t kLastUpdatedAtFieldId = 9;
+constexpr int32_t kLastUpdatedSnapshotIdFieldId = 10;
+constexpr int32_t kTotalDataFileSizeFieldId = 11;
+
+struct PartitionKey {
+ PartitionValues values;
+ size_t projected_fields;
+
+ bool operator==(const PartitionKey& other) const {
+ if (projected_fields != other.projected_fields ||
+ values.num_fields() != other.values.num_fields()) {
+ return false;
+ }
+ for (size_t index = 0; index < values.num_fields(); ++index) {
+ const auto& lhs = values.values()[index];
+ const auto& rhs = other.values.values()[index];
+ if (lhs.IsNull() || rhs.IsNull()) {
+ if (lhs.IsNull() != rhs.IsNull()) {
+ return false;
+ }
+ } else if (lhs.IsNaN() && rhs.IsNaN()) {
+ // Java partition equality canonicalizes all NaN signs and payloads.
+ continue;
+ } else if (lhs != rhs) {
+ return false;
+ }
+ }
+ return true;
+ }
+};
+
+struct PartitionKeyHash {
+ size_t operator()(const PartitionKey& key) const noexcept {
+ size_t result = 17;
+ for (const auto& value : key.values.values()) {
+ const size_t value_hash =
+ value.IsNaN() ? 0x7ff8000000000000ULL : LiteralHash{}(value);
+ result = result * 37 + value_hash;
+ }
+ return result * 37 + key.projected_fields;
+ }
+};
+
+size_t ProjectedFieldCount(const StructType& partition_type, const PartitionSpec& spec) {
+ size_t count = 0;
+ for (const auto& field : partition_type.fields()) {
+ count += std::ranges::any_of(
+ spec.fields(), [field_id = field.field_id()](const PartitionField& spec_field) {
+ return spec_field.field_id() == field_id;
+ });
+ }
+ return count;
+}
+
+struct PartitionStats {
+ explicit PartitionStats(PartitionValues values) : partition(std::move(values)) {}
+
+ PartitionValues partition;
+ int32_t spec_id = PartitionSpec::kInitialSpecId;
+ int64_t data_record_count = 0;
+ int32_t data_file_count = 0;
+ int64_t data_file_size = 0;
+ int64_t position_delete_record_count = 0;
+ int32_t position_delete_file_count = 0;
+ int64_t equality_delete_record_count = 0;
+ int32_t equality_delete_file_count = 0;
+ std::optional<TimePointMs> last_updated_at;
+ std::optional<int64_t> last_updated_snapshot_id;
+};
+
+Status AppendPartition(ArrowRowBuilder& builder, const Schema& schema,
+ const StructType& partition_type,
+ const PartitionStats& partition) {
+ for (size_t index = 0; index < schema.fields().size(); ++index) {
+ auto* array = builder.column(index);
+ switch (schema.fields()[index].field_id()) {
+ case kPartitionFieldId:
+ ICEBERG_RETURN_UNEXPECTED(
+ internal::AppendPartitionValues(array, partition_type, partition.partition));
+ break;
+ case kSpecIdFieldId:
+ ICEBERG_RETURN_UNEXPECTED(AppendInt(array, partition.spec_id));
+ break;
+ case kRecordCountFieldId:
+ ICEBERG_RETURN_UNEXPECTED(AppendInt(array, partition.data_record_count));
+ break;
+ case kFileCountFieldId:
+ ICEBERG_RETURN_UNEXPECTED(AppendInt(array, partition.data_file_count));
+ break;
+ case kTotalDataFileSizeFieldId:
+ ICEBERG_RETURN_UNEXPECTED(AppendInt(array, partition.data_file_size));
+ break;
+ case kPositionDeleteRecordCountFieldId:
+ ICEBERG_RETURN_UNEXPECTED(
+ AppendInt(array, partition.position_delete_record_count));
+ break;
+ case kPositionDeleteFileCountFieldId:
+ ICEBERG_RETURN_UNEXPECTED(AppendInt(array, partition.position_delete_file_count));
+ break;
+ case kEqualityDeleteRecordCountFieldId:
+ ICEBERG_RETURN_UNEXPECTED(
+ AppendInt(array, partition.equality_delete_record_count));
+ break;
+ case kEqualityDeleteFileCountFieldId:
+ ICEBERG_RETURN_UNEXPECTED(AppendInt(array, partition.equality_delete_file_count));
+ break;
+ case kLastUpdatedAtFieldId:
+ if (partition.last_updated_at.has_value()) {
+ ICEBERG_RETURN_UNEXPECTED(
+ AppendInt(array, std::chrono::duration_cast<std::chrono::microseconds>(
+ partition.last_updated_at->time_since_epoch())
+ .count()));
+ } else {
+ ICEBERG_RETURN_UNEXPECTED(AppendNull(array));
+ }
+ break;
+ case kLastUpdatedSnapshotIdFieldId:
+ if (partition.last_updated_snapshot_id.has_value()) {
+ ICEBERG_RETURN_UNEXPECTED(
+ AppendInt(array, *partition.last_updated_snapshot_id));
+ } else {
+ ICEBERG_RETURN_UNEXPECTED(AppendNull(array));
+ }
+ break;
+ default:
+ return InvalidSchema("Unsupported partitions metadata field {}",
+ schema.fields()[index].field_id());
+ }
+ }
+ return builder.FinishRow();
+}
+
+void UpdateCounts(PartitionStats& partition, const DataFile& file) {
+ switch (file.content) {
+ case DataFile::Content::kData:
+ partition.data_record_count += file.record_count;
+ ++partition.data_file_count;
+ partition.data_file_size += file.file_size_in_bytes;
+ break;
+ case DataFile::Content::kPositionDeletes:
+ partition.position_delete_record_count += file.record_count;
+ ++partition.position_delete_file_count;
+ break;
+ case DataFile::Content::kEqualityDeletes:
+ partition.equality_delete_record_count += file.record_count;
+ ++partition.equality_delete_file_count;
+ break;
+ }
+}
+
+} // namespace
+
+PartitionsTable::PartitionsTable(std::shared_ptr<Table> table)
+ : TimeTravelMetadataTable(std::move(table)) {}
+
+PartitionsTable::~PartitionsTable() = default;
+
+const std::shared_ptr<Schema>& PartitionsTable::schema() const { return schema_; }
+
+Result<std::unique_ptr<PartitionsTable>> PartitionsTable::Make(
+ std::shared_ptr<Table> table) {
+ ICEBERG_PRECHECK(table != nullptr, "Table cannot be null");
+ auto partitions =
+ std::unique_ptr<PartitionsTable>(new PartitionsTable(std::move(table)));
+ ICEBERG_RETURN_UNEXPECTED(partitions->RefreshSchema());
+ return partitions;
+}
+
+Status PartitionsTable::RefreshSchema() {
+ ICEBERG_ASSIGN_OR_RAISE(auto partition_type,
+ internal::UnifiedPartitionType(*source_table()));
+
+ std::vector<SchemaField> fields;
+ if (!partition_type->fields().empty()) {
+ fields.push_back(
+ SchemaField::MakeRequired(kPartitionFieldId, "partition", partition_type));
+ fields.push_back(SchemaField::MakeRequired(kSpecIdFieldId, "spec_id", int32()));
+ }
+ fields.push_back(
+ SchemaField::MakeRequired(kRecordCountFieldId, "record_count", int64()));
+ fields.push_back(SchemaField::MakeRequired(kFileCountFieldId, "file_count", int32()));
+ fields.push_back(SchemaField::MakeRequired(kTotalDataFileSizeFieldId,
+ "total_data_file_size_in_bytes", int64()));
+ fields.push_back(SchemaField::MakeRequired(kPositionDeleteRecordCountFieldId,
+ "position_delete_record_count", int64()));
+ fields.push_back(SchemaField::MakeRequired(kPositionDeleteFileCountFieldId,
+ "position_delete_file_count", int32()));
+ fields.push_back(SchemaField::MakeRequired(kEqualityDeleteRecordCountFieldId,
+ "equality_delete_record_count", int64()));
+ fields.push_back(SchemaField::MakeRequired(kEqualityDeleteFileCountFieldId,
+ "equality_delete_file_count", int32()));
+ fields.push_back(SchemaField::MakeOptional(kLastUpdatedAtFieldId, "last_updated_at",
+ timestamp_tz()));
+ fields.push_back(SchemaField::MakeOptional(kLastUpdatedSnapshotIdFieldId,
+ "last_updated_snapshot_id", int64()));
+
+ schema_ = std::make_shared<Schema>(std::move(fields));
+ partition_type_ = std::move(partition_type);
+ return {};
+}
+
+Result<ArrowArrayStream> PartitionsTable::ScanSnapshot(
+ const SnapshotSelection& snapshot_selection) {
+ ICEBERG_ASSIGN_OR_RAISE(auto snapshot, internal::ResolveMetadataTableSnapshot(
+ *source_table(), snapshot_selection));
+ ICEBERG_RETURN_UNEXPECTED(RefreshSchema());
+ ICEBERG_ASSIGN_OR_RAISE(
+ auto files, internal::LiveFiles(
+ *source_table(), snapshot,
+ {"content", "partition", "record_count", "file_size_in_bytes"}));
+
+ std::vector<PartitionStats> partitions;
+ std::unordered_map<PartitionKey, size_t, PartitionKeyHash> positions;
+ std::unordered_map<int64_t, std::shared_ptr<Snapshot>> snapshots;
+ while (true) {
+ ICEBERG_ASSIGN_OR_RAISE(auto next_file, files->Next());
+ if (!next_file.has_value()) {
+ break;
+ }
+ const auto& live_file = *next_file;
+ ICEBERG_ASSIGN_OR_RAISE(auto partition_values, internal::ProjectPartitionValues(
+ *partition_type_, *live_file.spec,
+ live_file.file->partition));
+ PartitionKey key{
+ .values = std::move(partition_values),
+ .projected_fields = ProjectedFieldCount(*partition_type_, *live_file.spec)};
+ auto [position, inserted] = positions.try_emplace(key, partitions.size());
+ if (inserted) {
+ partitions.emplace_back(std::move(key.values));
+ }
+ auto& partition = partitions[position->second];
+ UpdateCounts(partition, *live_file.file);
+
+ if (live_file.snapshot_id.has_value()) {
+ auto snapshot_iter = snapshots.find(*live_file.snapshot_id);
+ if (snapshot_iter == snapshots.end()) {
+ auto file_snapshot = source_table()->SnapshotById(*live_file.snapshot_id);
+ if (!file_snapshot.has_value() &&
+ file_snapshot.error().kind != ErrorKind::kNotFound) {
+ return std::unexpected<Error>(file_snapshot.error());
+ }
+ snapshot_iter =
+ snapshots.emplace(*live_file.snapshot_id, file_snapshot.value_or(nullptr))
+ .first;
+ }
+ const auto& file_snapshot = snapshot_iter->second;
+ if (file_snapshot != nullptr &&
+ (!partition.last_updated_at.has_value() ||
+ file_snapshot->timestamp_ms > *partition.last_updated_at)) {
+ partition.spec_id = live_file.spec->spec_id();
+ partition.last_updated_at = file_snapshot->timestamp_ms;
+ partition.last_updated_snapshot_id = file_snapshot->snapshot_id;
Review Comment:
The `last_updated_*` selection uses only `timestamp_ms` as a strict `>` comparison. If multiple snapshots share the same timestamp (possible with coarse clocks or synthetic test data), the chosen `spec_id` / `last_updated_snapshot_id` becomes order-dependent on file iteration, which can yield nondeterministic results. Consider adding a deterministic tie-breaker (e.g., when timestamps are equal, prefer the higher snapshot id or sequence number) so \"last updated\" selection is stable.
##########
src/iceberg/inspect/partitions_table.cc:
##########
@@ -0,0 +1,324 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+#include "iceberg/inspect/partitions_table.h"
+
+#include <algorithm>
+#include <chrono>
+#include <cstddef>
+#include <cstdint>
+#include <memory>
+#include <optional>
+#include <ranges>
+#include <unordered_map>
+#include <utility>
+#include <vector>
+
+#include "iceberg/arrow_row_builder_internal.h"
+#include "iceberg/expression/literal.h"
+#include "iceberg/inspect/metadata_table_stream_internal.h"
+#include "iceberg/inspect/metadata_table_util_internal.h"
+#include "iceberg/manifest/manifest_entry.h"
+#include "iceberg/partition_spec.h"
+#include "iceberg/row/partition_values.h"
+#include "iceberg/schema.h"
+#include "iceberg/schema_field.h"
+#include "iceberg/snapshot.h"
+#include "iceberg/table.h"
+#include "iceberg/type.h"
+#include "iceberg/util/macros.h"
+
+namespace iceberg {
+namespace {
+
+constexpr int32_t kPartitionFieldId = 1;
+constexpr int32_t kRecordCountFieldId = 2;
+constexpr int32_t kFileCountFieldId = 3;
+constexpr int32_t kSpecIdFieldId = 4;
+constexpr int32_t kPositionDeleteRecordCountFieldId = 5;
+constexpr int32_t kPositionDeleteFileCountFieldId = 6;
+constexpr int32_t kEqualityDeleteRecordCountFieldId = 7;
+constexpr int32_t kEqualityDeleteFileCountFieldId = 8;
+constexpr int32_t kLastUpdatedAtFieldId = 9;
+constexpr int32_t kLastUpdatedSnapshotIdFieldId = 10;
+constexpr int32_t kTotalDataFileSizeFieldId = 11;
+
+struct PartitionKey {
+ PartitionValues values;
+ size_t projected_fields;
+
+ bool operator==(const PartitionKey& other) const {
+ if (projected_fields != other.projected_fields ||
+ values.num_fields() != other.values.num_fields()) {
+ return false;
+ }
+ for (size_t index = 0; index < values.num_fields(); ++index) {
+ const auto& lhs = values.values()[index];
+ const auto& rhs = other.values.values()[index];
+ if (lhs.IsNull() || rhs.IsNull()) {
+ if (lhs.IsNull() != rhs.IsNull()) {
+ return false;
+ }
+ } else if (lhs.IsNaN() && rhs.IsNaN()) {
+ // Java partition equality canonicalizes all NaN signs and payloads.
+ continue;
+ } else if (lhs != rhs) {
+ return false;
+ }
+ }
+ return true;
+ }
+};
+
+struct PartitionKeyHash {
+ size_t operator()(const PartitionKey& key) const noexcept {
+ size_t result = 17;
+ for (const auto& value : key.values.values()) {
+ const size_t value_hash =
+ value.IsNaN() ? 0x7ff8000000000000ULL : LiteralHash{}(value);
+ result = result * 37 + value_hash;
+ }
+ return result * 37 + key.projected_fields;
+ }
+};
+
+size_t ProjectedFieldCount(const StructType& partition_type, const PartitionSpec& spec) {
+ size_t count = 0;
+ for (const auto& field : partition_type.fields()) {
+ count += std::ranges::any_of(
+ spec.fields(), [field_id = field.field_id()](const PartitionField& spec_field) {
+ return spec_field.field_id() == field_id;
+ });
+ }
+ return count;
+}
Review Comment:
`ProjectedFieldCount` is O(|partition_type| * |spec.fields|) and is invoked for every file during `PartitionsTable::ScanSnapshot`. That can add significant overhead for large tables. Consider precomputing the projected-field count per `spec_id` once per scan (e.g., build a `std::unordered_map<int32_t, size_t>` cache keyed by `live_file.spec->spec_id()`), or precomputing a fast membership structure for each spec to avoid repeated `any_of` scans.
##########
src/iceberg/inspect/manifests_table.cc:
##########
@@ -0,0 +1,219 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+#include "iceberg/inspect/manifests_table.h"
+
+#include <cstddef>
+#include <cstdint>
+#include <memory>
+#include <optional>
+#include <string>
+#include <utility>
+#include <vector>
+
+#include <nanoarrow/nanoarrow.h>
+
+#include "iceberg/arrow_row_builder_internal.h"
+#include "iceberg/inspect/metadata_table_stream_internal.h"
+#include "iceberg/inspect/metadata_table_util_internal.h"
+#include "iceberg/manifest/manifest_list.h"
+#include "iceberg/nanoarrow_status_internal.h"
+#include "iceberg/partition_spec.h"
+#include "iceberg/schema.h"
+#include "iceberg/schema_field.h"
+#include "iceberg/snapshot.h"
+#include "iceberg/table.h"
+#include "iceberg/transform.h"
+#include "iceberg/type.h"
+#include "iceberg/util/checked_cast.h"
+#include "iceberg/util/conversions.h"
+#include "iceberg/util/macros.h"
+
+namespace iceberg {
+namespace {
+
+struct ManifestRow {
+ ManifestFile manifest;
+ std::shared_ptr<PartitionSpec> spec;
+};
+
+Result<std::string> HumanReadableBound(const PartitionSpec& spec,
+ const StructType& partition_type, size_t index,
+ const std::vector<uint8_t>& bytes) {
+ ICEBERG_PRECHECK(index < spec.fields().size() && index < partition_type.fields().size(),
+ "Partition summary index {} is out of range", index);
+ auto primitive = internal::checked_pointer_cast<PrimitiveType>(
+ partition_type.fields()[index].type());
+ ICEBERG_ASSIGN_OR_RAISE(auto literal,
+ Conversions::FromBytes(std::move(primitive), bytes));
+ return spec.fields()[index].transform()->ToHumanString(literal);
+}
+
+Status AppendPartitionSummaries(ArrowArray* array, const ManifestRow& row,
+ const std::shared_ptr<StructType>& partition_type) {
+ ICEBERG_PRECHECK(row.manifest.partitions.size() <= row.spec->fields().size(),
+ "Manifest '{}' has more partition summaries than spec {} fields",
+ row.manifest.manifest_path, row.spec->spec_id());
+ auto* entries = array->children[0];
+ ICEBERG_PRECHECK(entries != nullptr && entries->n_children == 4,
+ "Partition summaries must contain four fields");
+
+ for (size_t index = 0; index < row.manifest.partitions.size(); ++index) {
+ const auto& summary = row.manifest.partitions[index];
+ ICEBERG_RETURN_UNEXPECTED(AppendBoolean(entries->children[0], summary.contains_null));
+ if (summary.contains_nan.has_value()) {
+ ICEBERG_RETURN_UNEXPECTED(
+ AppendBoolean(entries->children[1], *summary.contains_nan));
+ } else {
+ ICEBERG_RETURN_UNEXPECTED(AppendNull(entries->children[1]));
+ }
+
+ if (summary.lower_bound.has_value()) {
+ ICEBERG_ASSIGN_OR_RAISE(
+ auto lower,
+ HumanReadableBound(*row.spec, *partition_type, index, *summary.lower_bound));
+ ICEBERG_RETURN_UNEXPECTED(AppendString(entries->children[2], lower));
+ } else {
+ ICEBERG_RETURN_UNEXPECTED(AppendString(entries->children[2], "null"));
+ }
+ if (summary.upper_bound.has_value()) {
+ ICEBERG_ASSIGN_OR_RAISE(
+ auto upper,
+ HumanReadableBound(*row.spec, *partition_type, index, *summary.upper_bound));
+ ICEBERG_RETURN_UNEXPECTED(AppendString(entries->children[3], upper));
+ } else {
+ ICEBERG_RETURN_UNEXPECTED(AppendString(entries->children[3], "null"));
+ }
+ ICEBERG_NANOARROW_RETURN_UNEXPECTED(ArrowArrayFinishElement(entries));
+ }
+ ICEBERG_NANOARROW_RETURN_UNEXPECTED(ArrowArrayFinishElement(array));
+ return {};
+}
+
+Status AppendManifestCount(ArrowArray* array, const std::optional<int32_t>& count,
+ bool applicable) {
+ if (!applicable) {
+ return AppendInt(array, 0);
+ }
+ return count.has_value() ? AppendInt(array, *count) : AppendNull(array);
+}
+
+Status AppendManifest(ArrowRowBuilder& builder, const ManifestRow& row,
+ const std::shared_ptr<Schema>& table_schema) {
+ const auto& manifest = row.manifest;
+ ICEBERG_RETURN_UNEXPECTED(
+ AppendInt(builder.column(0), static_cast<int64_t>(manifest.content)));
+ ICEBERG_RETURN_UNEXPECTED(AppendString(builder.column(1), manifest.manifest_path));
+ ICEBERG_RETURN_UNEXPECTED(AppendInt(builder.column(2), manifest.manifest_length));
+ ICEBERG_RETURN_UNEXPECTED(AppendInt(builder.column(3), manifest.partition_spec_id));
+ ICEBERG_PRECHECK(manifest.added_snapshot_id.has_value(),
+ "Manifest '{}' has no added snapshot ID", manifest.manifest_path);
+ ICEBERG_RETURN_UNEXPECTED(AppendInt(builder.column(4), *manifest.added_snapshot_id));
+
+ const bool data = manifest.content == ManifestContent::kData;
+ ICEBERG_RETURN_UNEXPECTED(
+ AppendManifestCount(builder.column(5), manifest.added_files_count, data));
+ ICEBERG_RETURN_UNEXPECTED(
+ AppendManifestCount(builder.column(6), manifest.existing_files_count, data));
+ ICEBERG_RETURN_UNEXPECTED(
+ AppendManifestCount(builder.column(7), manifest.deleted_files_count, data));
+ ICEBERG_RETURN_UNEXPECTED(
+ AppendManifestCount(builder.column(8), manifest.added_files_count, !data));
+ ICEBERG_RETURN_UNEXPECTED(
+ AppendManifestCount(builder.column(9), manifest.existing_files_count, !data));
+ ICEBERG_RETURN_UNEXPECTED(
+ AppendManifestCount(builder.column(10), manifest.deleted_files_count, !data));
+
+ ICEBERG_ASSIGN_OR_RAISE(auto partition_type, row.spec->PartitionType(*table_schema));
+ ICEBERG_RETURN_UNEXPECTED(AppendPartitionSummaries(
+ builder.column(11), row, std::shared_ptr<StructType>(std::move(partition_type))));
+ return builder.FinishRow();
+}
+
+} // namespace
+
+ManifestsTable::ManifestsTable(std::shared_ptr<Table> table)
+ : TimeTravelMetadataTable(std::move(table)) {}
+
+ManifestsTable::~ManifestsTable() = default;
+
+const std::shared_ptr<Schema>& ManifestsTable::schema() const {
+ static const auto schema = std::make_shared<Schema>(std::vector<SchemaField>{
+ SchemaField::MakeRequired(14, "content", int32()),
+ SchemaField::MakeRequired(1, "path", string()),
+ SchemaField::MakeRequired(2, "length", int64()),
+ SchemaField::MakeRequired(3, "partition_spec_id", int32()),
+ SchemaField::MakeRequired(4, "added_snapshot_id", int64()),
+ // Legacy manifests can omit counts; unknown counts must stay null.
+ SchemaField::MakeOptional(5, "added_data_files_count", int32()),
+ SchemaField::MakeOptional(6, "existing_data_files_count", int32()),
+ SchemaField::MakeOptional(7, "deleted_data_files_count", int32()),
+ SchemaField::MakeOptional(15, "added_delete_files_count", int32()),
+ SchemaField::MakeOptional(16, "existing_delete_files_count", int32()),
+ SchemaField::MakeOptional(17, "deleted_delete_files_count", int32()),
+ SchemaField::MakeRequired(
+ 8, "partition_summaries",
+ list(SchemaField::MakeRequired(
+ 9, std::string(ListType::kElementName),
+ struct_({SchemaField::MakeRequired(10, "contains_null", boolean()),
+ SchemaField::MakeOptional(11, "contains_nan", boolean()),
+ SchemaField::MakeOptional(12, "lower_bound", string()),
+ SchemaField::MakeOptional(13, "upper_bound", string())})))),
+ });
+ return schema;
+}
+
+Result<std::unique_ptr<ManifestsTable>> ManifestsTable::Make(
+ std::shared_ptr<Table> table) {
+ ICEBERG_PRECHECK(table != nullptr, "Table cannot be null");
+ return std::unique_ptr<ManifestsTable>(new ManifestsTable(std::move(table)));
+}
+
+Result<ArrowArrayStream> ManifestsTable::ScanSnapshot(
+ const SnapshotSelection& snapshot_selection) {
+ ICEBERG_ASSIGN_OR_RAISE(auto snapshot, internal::ResolveMetadataTableSnapshot(
+ *source_table(), snapshot_selection));
+ std::vector<ManifestRow> rows;
+ if (snapshot != nullptr) {
+ ICEBERG_ASSIGN_OR_RAISE(auto specs_ref, source_table()->specs());
+ SnapshotReader snapshot_reader(snapshot.get());
+ ICEBERG_ASSIGN_OR_RAISE(auto manifests,
+ snapshot_reader.Manifests(source_table()->io()));
+ rows.reserve(manifests.size());
+ for (const auto& manifest : manifests) {
+ auto spec = specs_ref.get().find(manifest.partition_spec_id);
+ ICEBERG_CHECK(spec != specs_ref.get().end(),
+ "Cannot find partition spec {} for manifest '{}'",
+ manifest.partition_spec_id, manifest.manifest_path);
+ ICEBERG_PRECHECK(spec->second != nullptr, "Partition spec {} is null",
+ manifest.partition_spec_id);
+ rows.push_back(ManifestRow{.manifest = manifest, .spec = spec->second});
Review Comment:
This loop copies each `ManifestFile` into `rows` (`.manifest = manifest`). If `ManifestFile` contains non-trivial payloads (e.g., partition summaries), this introduces avoidable copying. Consider iterating `manifests` by non-const reference and moving into `rows` (or storing pointers/references when safe) to reduce memory churn during scans.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat(inspect): add system metadata tables [iceberg-cpp]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #876:
URL: https://github.com/apache/iceberg-cpp/pull/876#discussion_r3795799398
##########
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() {
+ return NotSupported("Scan is not supported for the history table");
+}
Review Comment:
HistoryTable is exposed as a concrete MetadataTable with a defined schema, but Scan() unconditionally returns NotSupported. This makes the table unusable through the public metadata-table API and is surprising for consumers expecting system metadata tables to be scannable.
##########
src/iceberg/test/meson.build:
##########
@@ -55,6 +55,7 @@ iceberg_tests = {
'metrics_test.cc',
'snapshot_test.cc',
'snapshot_util_test.cc',
+ 'system_metadata_tables_test.cc',
'table_metadata_builder_test.cc',
'table_requirement_test.cc',
Review Comment:
Meson test target 'table_test' doesn't compile the new metadata-table unit tests (history_table_test.cc, snapshots_table_test.cc). As a result, Meson CI/builds won't exercise these new table implementations even though CMake does.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org