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