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: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
