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