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: [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]
