From 6b12224f74905a921174f5690f999f20a48e4ca2 Mon Sep 17 00:00:00 2001 From: Manu Zhang Date: Sun, 15 Nov 2026 00:21:41 +0800 Subject: [PATCH] feat(inspect): add files metadata table Add the files metadata table and its focused tests. Depends on the shared snapshot metadata scan support prerequisite. Keep readable metric schema and Arrow serialization local to this table; convert promoted date bounds and reject out-of-range timestamp values. Co-authored-by: Codex --- src/iceberg/CMakeLists.txt | 2 + src/iceberg/cpp20_compatibility_internal.h | 1 + src/iceberg/inspect/files_table.cc | 94 ++++ src/iceberg/inspect/files_table.h | 67 +++ .../inspect/files_table_util_internal.cc | 330 +++++++++++++ .../inspect/files_table_util_internal.h | 35 ++ src/iceberg/inspect/metadata_table.cc | 3 + src/iceberg/test/CMakeLists.txt | 3 +- src/iceberg/test/inspect/files_table_test.cc | 453 ++++++++++++++++++ src/iceberg/type_fwd.h | 1 + 10 files changed, 988 insertions(+), 1 deletion(-) create mode 100644 src/iceberg/inspect/files_table.cc create mode 100644 src/iceberg/inspect/files_table.h create mode 100644 src/iceberg/inspect/files_table_util_internal.cc create mode 100644 src/iceberg/inspect/files_table_util_internal.h create mode 100644 src/iceberg/test/inspect/files_table_test.cc diff --git a/src/iceberg/CMakeLists.txt b/src/iceberg/CMakeLists.txt index 4cde4fb23..bda072875 100644 --- a/src/iceberg/CMakeLists.txt +++ b/src/iceberg/CMakeLists.txt @@ -58,6 +58,8 @@ set(ICEBERG_SOURCES geospatial.cc incremental_append_scan.cc incremental_changelog_scan.cc + inspect/files_table.cc + inspect/files_table_util_internal.cc inspect/history_table.cc inspect/metadata_scan.cc inspect/metadata_table.cc diff --git a/src/iceberg/cpp20_compatibility_internal.h b/src/iceberg/cpp20_compatibility_internal.h index d9c64d36a..7a8f85ccd 100644 --- a/src/iceberg/cpp20_compatibility_internal.h +++ b/src/iceberg/cpp20_compatibility_internal.h @@ -80,6 +80,7 @@ #include "iceberg/incremental_append_scan.h" #include "iceberg/incremental_changelog_scan.h" #include "iceberg/inheritable_metadata.h" +#include "iceberg/inspect/files_table.h" #include "iceberg/inspect/history_table.h" #include "iceberg/inspect/metadata_scan.h" #include "iceberg/inspect/metadata_scan_task.h" diff --git a/src/iceberg/inspect/files_table.cc b/src/iceberg/inspect/files_table.cc new file mode 100644 index 000000000..3a9336487 --- /dev/null +++ b/src/iceberg/inspect/files_table.cc @@ -0,0 +1,94 @@ +/* + * 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 +#include +#include + +#include "iceberg/inspect/files_table_util_internal.h" +#include "iceberg/inspect/metadata_scan_util_internal.h" +#include "iceberg/inspect/metadata_table_util_internal.h" +#include "iceberg/schema.h" +#include "iceberg/table.h" +#include "iceberg/table_metadata.h" +#include "iceberg/type.h" +#include "iceberg/util/macros.h" + +namespace iceberg { + +FilesTable::FilesTable(std::shared_ptr table) + : MetadataTable(std::move(table)) {} + +FilesTable::~FilesTable() = default; + +std::shared_ptr FilesTable::schema() const { + std::lock_guard lock(schema_mutex_); + return schema_; +} + +Result> FilesTable::Make(std::shared_ptr table) { + ICEBERG_PRECHECK(table != nullptr, "Table cannot be null"); + auto files = std::unique_ptr(new FilesTable(std::move(table))); + ICEBERG_RETURN_UNEXPECTED(files->SchemaForScan(*files->source_table()->metadata())); + return files; +} + +Result> FilesTable::SchemaForScan( + const TableMetadata& metadata) const { + ICEBERG_ASSIGN_OR_RAISE(auto table_schema, metadata.Schema()); + ICEBERG_ASSIGN_OR_RAISE(auto partition_type, internal::UnifiedPartitionType(metadata)); + ICEBERG_ASSIGN_OR_RAISE(auto schema, + internal::FilesTableSchema(*table_schema, partition_type)); + { + std::lock_guard lock(schema_mutex_); + schema_ = schema; + } + return schema; +} + +Result> FilesTable::MakeScan( + internal::MetadataScanContext context) const { + return internal::MakeSnapshotMetadataScan(std::move(context), ReadRows); +} + +Result FilesTable::ReadRows( + const internal::MetadataScanContext& context, const std::shared_ptr& io, + int64_t batch_size) { + ICEBERG_ASSIGN_OR_RAISE(auto partition_type, + internal::UnifiedPartitionType(*context.table_metadata)); + ICEBERG_ASSIGN_OR_RAISE(auto table_schema, context.table_metadata->Schema()); + ICEBERG_ASSIGN_OR_RAISE(auto snapshot, internal::ResolveMetadataTableSnapshot( + *context.table_metadata, context)); + ICEBERG_ASSIGN_OR_RAISE(auto files, + internal::LiveFiles(*context.table_metadata, io, snapshot)); + auto schema = context.projected_schema; + const int32_t readable_metrics_id = context.schema->fields().back().field_id(); + return internal::MakeMetadataBatchStream( + schema, std::move(files), batch_size, + [schema, readable_metrics_id, 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, + readable_metrics_id, file); + }); +} + +} // namespace iceberg diff --git a/src/iceberg/inspect/files_table.h b/src/iceberg/inspect/files_table.h new file mode 100644 index 000000000..5e6b2d7e8 --- /dev/null +++ b/src/iceberg/inspect/files_table.h @@ -0,0 +1,67 @@ +/* + * 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. + */ + +#pragma once + +/// \file iceberg/inspect/files_table.h +/// \brief Define the files metadata table. + +#include +#include + +#include "iceberg/arrow_c_data.h" +#include "iceberg/iceberg_export.h" +#include "iceberg/inspect/metadata_table.h" +#include "iceberg/result.h" +#include "iceberg/type_fwd.h" + +namespace iceberg { + +/// \brief Metadata table containing live data and delete files in a snapshot. +class ICEBERG_EXPORT FilesTable : public MetadataTable { + public: + static Result> Make(std::shared_ptr table); + + ~FilesTable() override; + + Kind kind() const noexcept override { return Kind::kFiles; } + + /// \brief Schema from construction or the most recent scan builder. + /// Existing scans and tasks retain their captured schema. + std::shared_ptr schema() const override; + + protected: + Result> SchemaForScan( + const TableMetadata& metadata) const override; + + Result> MakeScan( + internal::MetadataScanContext context) const override; + + private: + static Result ReadRows(const internal::MetadataScanContext& context, + const std::shared_ptr& io, + int64_t batch_size); + + explicit FilesTable(std::shared_ptr table); + + mutable std::mutex schema_mutex_; + mutable std::shared_ptr schema_; +}; + +} // namespace iceberg diff --git a/src/iceberg/inspect/files_table_util_internal.cc b/src/iceberg/inspect/files_table_util_internal.cc new file mode 100644 index 000000000..c2a1d4cd7 --- /dev/null +++ b/src/iceberg/inspect/files_table_util_internal.cc @@ -0,0 +1,330 @@ +/* + * 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_util_internal.h" + +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +#include + +#include "iceberg/arrow_row_builder_internal.h" +#include "iceberg/file_format.h" +#include "iceberg/inspect/metadata_table_util_internal.h" +#include "iceberg/nanoarrow_status_internal.h" +#include "iceberg/partition_spec.h" +#include "iceberg/schema.h" +#include "iceberg/schema_field.h" +#include "iceberg/type.h" +#include "iceberg/util/checked_cast.h" +#include "iceberg/util/conversions.h" +#include "iceberg/util/macros.h" + +namespace iceberg::internal { +namespace { + +template +Status AppendOptionalInt(ArrowArray* array, const std::optional& value) { + if (!value.has_value()) { + return AppendNull(array); + } + return AppendInt(array, static_cast(*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>& 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(field.type()), + primitive_fields); + } + } +} + +Result ReadableMetricsField(const Schema& table_schema, + int32_t highest_metadata_field_id) { + std::vector> primitive_fields; + CollectPrimitiveFields(table_schema, primitive_fields); + + int32_t next_id = highest_metadata_field_id; + std::vector 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 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& 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>& bounds, + const SchemaField& field) { + auto bound = bounds.find(field.field_id()); + if (bound == bounds.end()) { + return AppendNull(array); + } + auto primitive = checked_pointer_cast(field.type()); + // Four-byte bounds under a promoted timestamp still contain date values in days. + if (bound->second.size() == sizeof(int32_t) && + (primitive->type_id() == TypeId::kTimestamp || + primitive->type_id() == TypeId::kTimestampNs)) { + ICEBERG_ASSIGN_OR_RAISE(auto date_value, + Conversions::FromBytes(date(), bound->second)); + const int64_t days = std::get(date_value.value()); + const int64_t units_per_day = primitive->type_id() == TypeId::kTimestampNs + ? 86'400'000'000'000LL + : 86'400'000'000LL; + ICEBERG_PRECHECK(days >= std::numeric_limits::min() / units_per_day && + days <= std::numeric_limits::max() / units_per_day, + "Date bound {} is outside the {} range", days, + primitive->ToString()); + return AppendInt(array, days * units_per_day); + } + 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(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> FilesTableSchema( + const Schema& table_schema, const std::shared_ptr& partition_type) { + ICEBERG_PRECHECK(partition_type != nullptr, "Partition type cannot be null"); + auto data_file_type = DataFile::Type(partition_type); + std::vector 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(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(file_schema->fields().begin(), + file_schema->fields().end()); + fields.push_back(std::move(readable_metrics)); + return std::make_shared(std::move(fields)); +} + +Status AppendDataFile(ArrowRowBuilder& builder, const Schema& schema, + const Schema& table_schema, const StructType& partition_type, + int32_t readable_metrics_id, const LiveFile& live_file) { + ICEBERG_PRECHECK(live_file.file != nullptr, "Data file cannot be null"); + ICEBERG_PRECHECK(live_file.spec != nullptr, "Partition spec cannot be null"); + const auto& file = *live_file.file; + + for (size_t index = 0; index < schema.fields().size(); ++index) { + const auto& field = schema.fields()[index]; + auto* array = builder.column(index); + switch (field.field_id()) { + case DataFile::kContentFieldId: + ICEBERG_RETURN_UNEXPECTED(AppendInt(array, static_cast(file.content))); + break; + case DataFile::kFilePathFieldId: + ICEBERG_RETURN_UNEXPECTED(AppendString(array, file.file_path)); + break; + case DataFile::kFileFormatFieldId: + ICEBERG_RETURN_UNEXPECTED( + AppendString(array, MetadataFileFormat(file.file_format))); + break; + case DataFile::kSpecIdFieldId: + ICEBERG_RETURN_UNEXPECTED(AppendInt(array, live_file.spec->spec_id())); + break; + case DataFile::kPartitionFieldId: { + ICEBERG_ASSIGN_OR_RAISE( + auto projected, + ProjectPartitionValues(partition_type, *live_file.spec, file.partition)); + ICEBERG_RETURN_UNEXPECTED( + AppendPartitionValues(array, partition_type, projected)); + break; + } + case DataFile::kRecordCountFieldId: + ICEBERG_RETURN_UNEXPECTED(AppendInt(array, file.record_count)); + break; + case DataFile::kFileSizeFieldId: + ICEBERG_RETURN_UNEXPECTED(AppendInt(array, file.file_size_in_bytes)); + break; + case DataFile::kColumnSizesFieldId: + ICEBERG_RETURN_UNEXPECTED(AppendIntMap(array, file.column_sizes)); + break; + case DataFile::kValueCountsFieldId: + ICEBERG_RETURN_UNEXPECTED(AppendIntMap(array, file.value_counts)); + break; + case DataFile::kNullValueCountsFieldId: + ICEBERG_RETURN_UNEXPECTED(AppendIntMap(array, file.null_value_counts)); + break; + case DataFile::kNanValueCountsFieldId: + ICEBERG_RETURN_UNEXPECTED(AppendIntMap(array, file.nan_value_counts)); + break; + case DataFile::kLowerBoundsFieldId: + ICEBERG_RETURN_UNEXPECTED(AppendBinaryMap(array, file.lower_bounds)); + break; + case DataFile::kUpperBoundsFieldId: + ICEBERG_RETURN_UNEXPECTED(AppendBinaryMap(array, file.upper_bounds)); + break; + case DataFile::kKeyMetadataFieldId: + if (file.key_metadata.empty()) { + ICEBERG_RETURN_UNEXPECTED(AppendNull(array)); + } else { + ICEBERG_RETURN_UNEXPECTED(AppendBytes(array, file.key_metadata)); + } + break; + case DataFile::kSplitOffsetsFieldId: + ICEBERG_RETURN_UNEXPECTED(AppendIntList(array, file.split_offsets)); + break; + case DataFile::kEqualityIdsFieldId: + ICEBERG_RETURN_UNEXPECTED(AppendIntList(array, file.equality_ids)); + break; + case DataFile::kSortOrderIdFieldId: + ICEBERG_RETURN_UNEXPECTED(AppendOptionalInt(array, file.sort_order_id)); + break; + case DataFile::kFirstRowIdFieldId: + ICEBERG_RETURN_UNEXPECTED(AppendOptionalInt(array, file.first_row_id)); + break; + case DataFile::kReferencedDataFileFieldId: + if (file.referenced_data_file.has_value()) { + ICEBERG_RETURN_UNEXPECTED(AppendString(array, *file.referenced_data_file)); + } else { + ICEBERG_RETURN_UNEXPECTED(AppendNull(array)); + } + break; + case DataFile::kContentOffsetFieldId: + ICEBERG_RETURN_UNEXPECTED(AppendOptionalInt(array, file.content_offset)); + break; + case DataFile::kContentSizeFieldId: + ICEBERG_RETURN_UNEXPECTED(AppendOptionalInt(array, file.content_size_in_bytes)); + break; + default: + if (field.field_id() == readable_metrics_id) { + auto readable_type = checked_pointer_cast(field.type()); + ICEBERG_RETURN_UNEXPECTED( + AppendReadableMetrics(array, *readable_type, table_schema, file)); + } else { + return InvalidSchema("Unsupported files metadata field {}", field.field_id()); + } + } + } + return builder.FinishRow(); +} + +} // namespace iceberg::internal diff --git a/src/iceberg/inspect/files_table_util_internal.h b/src/iceberg/inspect/files_table_util_internal.h new file mode 100644 index 000000000..bab5481e6 --- /dev/null +++ b/src/iceberg/inspect/files_table_util_internal.h @@ -0,0 +1,35 @@ +/* + * 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. + */ + +#pragma once + +#include "iceberg/inspect/metadata_table_util_internal.h" + +namespace iceberg::internal { + +/// \brief Build the files metadata table schema for a table. +Result> FilesTableSchema( + const Schema& table_schema, const std::shared_ptr& partition_type); + +/// \brief Append a data-file row using the files metadata table schema. +Status AppendDataFile(ArrowRowBuilder& builder, const Schema& schema, + const Schema& table_schema, const StructType& partition_type, + int32_t readable_metrics_id, const LiveFile& live_file); + +} // namespace iceberg::internal diff --git a/src/iceberg/inspect/metadata_table.cc b/src/iceberg/inspect/metadata_table.cc index 012ffaa99..7c1ae9a04 100644 --- a/src/iceberg/inspect/metadata_table.cc +++ b/src/iceberg/inspect/metadata_table.cc @@ -22,6 +22,7 @@ #include #include +#include "iceberg/inspect/files_table.h" #include "iceberg/inspect/history_table.h" #include "iceberg/inspect/metadata_scan.h" #include "iceberg/inspect/refs_table.h" @@ -43,6 +44,8 @@ Result> MetadataTable::Make( return RefsTable::Make(std::move(table)); case MetadataTableKind::kHistory: return HistoryTable::Make(std::move(table)); + case MetadataTableKind::kFiles: + return FilesTable::Make(std::move(table)); default: break; } diff --git a/src/iceberg/test/CMakeLists.txt b/src/iceberg/test/CMakeLists.txt index 1874fc25e..bd387998a 100644 --- a/src/iceberg/test/CMakeLists.txt +++ b/src/iceberg/test/CMakeLists.txt @@ -209,7 +209,8 @@ if(ICEBERG_BUILD_BUNDLE) inspect/history_table_test.cc inspect/metadata_table_test.cc inspect/refs_table_test.cc - inspect/snapshots_table_test.cc) + inspect/snapshots_table_test.cc + inspect/files_table_test.cc) add_iceberg_test(eval_expr_test USE_BUNDLE diff --git a/src/iceberg/test/inspect/files_table_test.cc b/src/iceberg/test/inspect/files_table_test.cc new file mode 100644 index 000000000..01510dd26 --- /dev/null +++ b/src/iceberg/test/inspect/files_table_test.cc @@ -0,0 +1,453 @@ +/* + * 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 +#include +#include +#include +#include +#include +#include +#include +#include + +#include +#include +#include +#include + +#include "iceberg/constants.h" +#include "iceberg/inspect/metadata_table.h" +#include "iceberg/partition_spec.h" +#include "iceberg/row/partition_values.h" +#include "iceberg/snapshot.h" +#include "iceberg/table.h" +#include "iceberg/table_identifier.h" +#include "iceberg/test/inspect/metadata_table_scan_test_base.h" +#include "iceberg/test/matchers.h" +#include "iceberg/transform.h" +#include "iceberg/util/macros.h" + +namespace iceberg { +namespace { + +class FilesTableTest : public MetadataTableScanTestBase {}; + +TEST_P(FilesTableTest, Scan) { + auto snapshot = MakeAppendSnapshotWithPartitionValues( + GetParam(), 10, std::nullopt, 1, + {{"s3://bucket/data.parquet", PartitionValues(Literal::Int(7))}}, + partitioned_spec_); + ICEBERG_UNWRAP_OR_FAIL(auto table, MakeTable({snapshot}, 10, {}, partitioned_spec_)); + + ICEBERG_UNWRAP_OR_FAIL(auto files, MetadataTable::Make(table)); + ICEBERG_UNWRAP_OR_FAIL(auto files_stream, Scan(files)); + ICEBERG_UNWRAP_OR_FAIL(auto files_batches, ReadAllBatches(std::move(files_stream))); + ASSERT_EQ(files_batches.size(), 1); + ASSERT_EQ(files_batches[0]->num_rows(), 1); + auto paths = std::static_pointer_cast<::arrow::StringArray>( + files_batches[0]->GetColumnByName("file_path")); + auto formats = std::static_pointer_cast<::arrow::StringArray>( + files_batches[0]->GetColumnByName("file_format")); + auto spec_ids = std::static_pointer_cast<::arrow::Int32Array>( + files_batches[0]->GetColumnByName("spec_id")); + auto file_partitions = std::static_pointer_cast<::arrow::StructArray>( + files_batches[0]->GetColumnByName("partition")); + auto partition_values = + std::static_pointer_cast<::arrow::Int32Array>(file_partitions->field(0)); + EXPECT_EQ(paths->GetString(0), "s3://bucket/data.parquet"); + EXPECT_EQ(formats->GetString(0), "PARQUET"); + EXPECT_EQ(spec_ids->Value(0), partitioned_spec_->spec_id()); + EXPECT_EQ(partition_values->Value(0), 7); +} + +TEST_P(FilesTableTest, SupportsTimeTravelForSnapshotScopedTables) { + auto first = + MakeAppendSnapshot(GetParam(), 1, std::nullopt, 1, {"s3://bucket/first.parquet"}); + auto second = MakeAppendSnapshot(GetParam(), 2, 1, 2, {"s3://bucket/second.parquet"}); + ICEBERG_UNWRAP_OR_FAIL(auto table, MakeTable({first, second}, 2)); + ICEBERG_UNWRAP_OR_FAIL(auto files, MetadataTable::Make(table)); + + ICEBERG_UNWRAP_OR_FAIL(auto stream, + Scan(files, SnapshotSelection{.snapshot = int64_t{1}})); + ICEBERG_UNWRAP_OR_FAIL(auto batches, ReadAllBatches(std::move(stream))); + ASSERT_EQ(batches.size(), 1); + auto paths = std::static_pointer_cast<::arrow::StringArray>( + batches[0]->GetColumnByName("file_path")); + ASSERT_EQ(paths->length(), 1); + EXPECT_EQ(paths->GetString(0), "s3://bucket/first.parquet"); +} + +TEST_P(FilesTableTest, RejectsCombiningNamedRefsWithTimeTravel) { + auto first = MakeAppendSnapshot(GetParam(), 1, std::nullopt, 1, {"first.parquet"}); + auto second = MakeAppendSnapshot(GetParam(), 2, 1, 2, {"second.parquet"}); + ICEBERG_UNWRAP_OR_FAIL(auto branch, SnapshotRef::MakeBranch(1)); + ICEBERG_UNWRAP_OR_FAIL(auto tag, SnapshotRef::MakeTag(1)); + std::unordered_map> refs; + refs.emplace("dev", std::move(branch)); + refs.emplace("release", std::move(tag)); + ICEBERG_UNWRAP_OR_FAIL(auto table, MakeTable({first, second}, 2, std::move(refs))); + ICEBERG_UNWRAP_OR_FAIL(auto metadata_table, MetadataTable::Make(table)); + for (const std::string ref_name : {"dev", "release"}) { + for (const SnapshotSelection selection : + {SnapshotSelection{.snapshot = int64_t{1}, .ref_name = ref_name}, + SnapshotSelection{.snapshot = first->timestamp_ms, .ref_name = ref_name}}) { + auto result = Scan(metadata_table, selection); + ASSERT_FALSE(result.has_value()); + EXPECT_EQ(result.error().kind, ErrorKind::kInvalidArgument); + } + } + + // A named ref alone selects its head; main is a no-op in Java's useRef. + for (const std::string ref_name : {"dev", "release", "main"}) { + SnapshotSelection selection{.ref_name = ref_name}; + if (ref_name == "main") { + selection.snapshot = int64_t{1}; + } + ICEBERG_UNWRAP_OR_FAIL(auto stream, Scan(metadata_table, selection)); + ICEBERG_UNWRAP_OR_FAIL(auto batches, ReadAllBatches(std::move(stream))); + ASSERT_EQ(batches.size(), 1); + ASSERT_EQ(batches[0]->num_rows(), 1); + auto paths = std::static_pointer_cast<::arrow::StringArray>( + batches[0]->GetColumnByName("file_path")); + EXPECT_EQ(paths->GetString(0), "first.parquet"); + } +} + +TEST_P(FilesTableTest, RejectsUnknownReference) { + ICEBERG_UNWRAP_OR_FAIL(auto table, MakeTable({}, kInvalidSnapshotId)); + ICEBERG_UNWRAP_OR_FAIL(auto metadata_table, MetadataTable::Make(table)); + EXPECT_THAT(Scan(metadata_table, SnapshotSelection{.ref_name = "missing"}), + IsError(ErrorKind::kInvalidArgument)); +} + +TEST_P(FilesTableTest, ScansRefreshedSchemaAndPartitionSpec) { + auto first = MakeAppendSnapshotWithPartitionValues( + GetParam(), 10, std::nullopt, 1, + {{"old.parquet", PartitionValues(Literal::Int(7))}}, partitioned_spec_); + ICEBERG_UNWRAP_OR_FAIL(auto table, MakeTable({first}, 10, {}, partitioned_spec_)); + ICEBERG_UNWRAP_OR_FAIL(auto files, MetadataTable::Make(table)); + auto old_files_schema = files->schema(); + ICEBERG_UNWRAP_OR_FAIL(auto old_stream, Scan(files)); + auto old_reader = ::arrow::ImportRecordBatchReader(&old_stream); + ASSERT_TRUE(old_reader.ok()) << old_reader.status().ToString(); + + ASSERT_THAT(RefreshTableSchemaAndPartitionSpec(table, first), IsOk()); + + // The public builder factory must refresh the schema just like NewScan(). + ICEBERG_UNWRAP_OR_FAIL(auto builder, MetadataScanBuilder::Make(*files)); + ICEBERG_UNWRAP_OR_FAIL(auto scan, builder->Build()); + ICEBERG_UNWRAP_OR_FAIL(auto files_stream, scan->ToArrowArrayStream()); + ICEBERG_UNWRAP_OR_FAIL(auto files_batches, ReadAllBatches(std::move(files_stream))); + ASSERT_EQ(files_batches.size(), 1); + ASSERT_EQ(files_batches[0]->num_rows(), 2); + auto file_partition = std::static_pointer_cast<::arrow::StructArray>( + files_batches[0]->GetColumnByName("partition")); + ASSERT_EQ(file_partition->num_fields(), 2); + auto ids = std::static_pointer_cast<::arrow::Int32Array>(file_partition->field(1)); + EXPECT_EQ(ids->Value(0), 1); + EXPECT_EQ(ids->Value(1), 2); + auto metrics = std::static_pointer_cast<::arrow::StructArray>( + files_batches[0]->GetColumnByName("readable_metrics")); + EXPECT_NE(metrics->GetFieldByName("extra"), nullptr); + ICEBERG_UNWRAP_OR_FAIL(auto extra, + files->schema()->FindFieldByName("readable_metrics.extra")); + ASSERT_TRUE(extra.has_value()); + EXPECT_NE(files->schema(), old_files_schema); + + // Streams created before refresh retain their original metadata and schema. + auto batches = (*old_reader)->ToRecordBatches(); + ASSERT_TRUE(batches.ok()) << batches.status().ToString(); + ASSERT_EQ(batches->size(), 1); + EXPECT_EQ((*batches)[0]->num_rows(), 1); + auto old_partition = std::static_pointer_cast<::arrow::StructArray>( + (*batches)[0]->GetColumnByName("partition")); + EXPECT_EQ(old_partition->num_fields(), 1); +} + +TEST_P(FilesTableTest, MainTimestampSelectionUsesSnapshotLogAfterRollback) { + auto first = + MakeAppendSnapshot(GetParam(), 1, std::nullopt, 1, {"s3://bucket/first.parquet"}); + auto second = MakeAppendSnapshot(GetParam(), 2, 1, 2, {"s3://bucket/second.parquet"}); + auto third = MakeAppendSnapshot(GetParam(), 3, 2, 3, {"s3://bucket/third.parquet"}); + ICEBERG_UNWRAP_OR_FAIL(auto table, MakeTable({first, second, third}, 1)); + table->metadata()->snapshot_log = { + SnapshotLogEntry{.timestamp_ms = first->timestamp_ms, .snapshot_id = 1}, + SnapshotLogEntry{.timestamp_ms = second->timestamp_ms, .snapshot_id = 2}, + SnapshotLogEntry{.timestamp_ms = third->timestamp_ms, .snapshot_id = 3}, + SnapshotLogEntry{.timestamp_ms = third->timestamp_ms + std::chrono::milliseconds(1), + .snapshot_id = 1}, + }; + ICEBERG_UNWRAP_OR_FAIL(auto files, MetadataTable::Make(table)); + + ICEBERG_UNWRAP_OR_FAIL(auto stream, + Scan(files, SnapshotSelection{.snapshot = third->timestamp_ms})); + ICEBERG_UNWRAP_OR_FAIL(auto batches, ReadAllBatches(std::move(stream))); + ASSERT_EQ(batches.size(), 1); + auto paths = std::static_pointer_cast<::arrow::StringArray>( + batches[0]->GetColumnByName("file_path")); + ASSERT_EQ(paths->length(), 1); + EXPECT_EQ(paths->GetString(0), "s3://bucket/third.parquet"); +} + +TEST_P(FilesTableTest, FilesSchemaIncludesReadableMetrics) { + ICEBERG_UNWRAP_OR_FAIL(auto table, + MakeTable({}, kInvalidSnapshotId, {}, partitioned_spec_)); + ICEBERG_UNWRAP_OR_FAIL(auto files, MetadataTable::Make(table)); + + ICEBERG_UNWRAP_OR_FAIL(auto spec_id_field, + files->schema()->FindFieldById(DataFile::kSpecIdFieldId)); + ASSERT_TRUE(spec_id_field.has_value()); + EXPECT_TRUE(spec_id_field->get().optional()); + + ICEBERG_UNWRAP_OR_FAIL(auto readable_metrics, + files->schema()->FindFieldByName("readable_metrics")); + ASSERT_TRUE(readable_metrics.has_value()); + EXPECT_TRUE(readable_metrics->get().optional()); + auto metrics_type = + std::static_pointer_cast(readable_metrics->get().type()); + ASSERT_EQ(metrics_type->fields().size(), 2); + EXPECT_EQ(metrics_type->fields()[0].name(), "data"); + EXPECT_EQ(metrics_type->fields()[1].name(), "id"); + for (const auto& field : metrics_type->fields()) { + auto column_metrics = std::static_pointer_cast(field.type()); + EXPECT_EQ(column_metrics->fields().size(), 6); + } +} + +TEST_P(FilesTableTest, SupportsLegacyVoidPartitionEvolution) { + ICEBERG_UNWRAP_OR_FAIL(auto older_spec, + PartitionSpec::Make(1, {PartitionField(2, 1000, "old_bucket", + Transform::Bucket(16))})); + ICEBERG_UNWRAP_OR_FAIL( + auto latest_spec, + PartitionSpec::Make(2, {PartitionField(2, 1000, "new_name", Transform::Void())})); + auto metadata = MakeTableMetadata({}, kInvalidSnapshotId); + metadata->partition_specs = { + std::shared_ptr(std::move(older_spec)), + std::shared_ptr(std::move(latest_spec)), + }; + metadata->default_spec_id = 2; + ICEBERG_UNWRAP_OR_FAIL( + auto table, + Table::Make(TableIdentifier{.ns = Namespace{.levels = {"db"}}, .name = "table"}, + std::move(metadata), "s3://bucket/metadata.json", file_io_, catalog_)); + + ICEBERG_UNWRAP_OR_FAIL(auto files, MetadataTable::Make(table)); + ICEBERG_UNWRAP_OR_FAIL(auto partition_field, + files->schema()->FindFieldById(DataFile::kPartitionFieldId)); + ASSERT_TRUE(partition_field.has_value()); + auto partition_type = + std::static_pointer_cast(partition_field->get().type()); + ASSERT_EQ(partition_type->fields().size(), 1); + EXPECT_EQ(partition_type->fields()[0].name(), "new_name"); + EXPECT_EQ(partition_type->fields()[0].type()->type_id(), TypeId::kInt); +} + +TEST_P(FilesTableTest, FilesScanReadsManifestsOnDemand) { + std::vector entries; + for (int64_t index = 0; index < 1024; ++index) { + entries.push_back(MakeEntry(ManifestStatus::kAdded, 10, 1, + MakeDataFile(std::format("data-{}.parquet", index)))); + } + auto first_manifest = WriteDataManifest(GetParam(), 10, std::move(entries)); + auto second_manifest = WriteDataManifest( + GetParam(), 10, + {MakeEntry(ManifestStatus::kAdded, 10, 1, MakeDataFile("last.parquet"))}); + auto snapshot = std::make_shared(Snapshot{ + .snapshot_id = 10, + .sequence_number = 1, + .timestamp_ms = TimePointMsFromUnixMs(1000), + .manifest_list = + WriteManifestList(GetParam(), 10, 0, 1, {first_manifest, second_manifest}), + }); + ASSERT_THAT(file_io_->DeleteFile(second_manifest.manifest_path), IsOk()); + ICEBERG_UNWRAP_OR_FAIL(auto table, MakeTable({snapshot}, 10)); + ICEBERG_UNWRAP_OR_FAIL(auto files, MetadataTable::Make(table)); + ICEBERG_UNWRAP_OR_FAIL(auto stream, Scan(files)); + auto imported = ::arrow::ImportRecordBatchReader(&stream); + ASSERT_TRUE(imported.ok()) << imported.status().ToString(); + auto reader = *imported; + // The stream owns everything it needs, independent of the source table's lifetime. + files.reset(); + table.reset(); + auto first = reader->Next(); + ASSERT_TRUE(first.ok()) << first.status().ToString(); + ASSERT_NE(*first, nullptr); + EXPECT_EQ((*first)->num_rows(), 1024); + // Opening the absent second manifest is deferred until the next batch. + const auto error = reader->Next(); + ASSERT_FALSE(error.ok()); + const auto repeated_error = reader->Next(); + ASSERT_FALSE(repeated_error.ok()); + EXPECT_EQ(repeated_error.status().ToString(), error.status().ToString()); +} + +TEST_P(FilesTableTest, FilesScanSkipsEmptyAndDeletedEntries) { + auto first = WriteDataManifest( + GetParam(), 10, + {MakeEntry(ManifestStatus::kAdded, 10, 1, MakeDataFile("first.parquet"))}); + auto empty = WriteDataManifest(GetParam(), 10, {}); + auto last = WriteDataManifest( + GetParam(), 10, + {MakeEntry(ManifestStatus::kDeleted, 10, 1, MakeDataFile("deleted.parquet")), + MakeEntry(ManifestStatus::kExisting, 10, 1, MakeDataFile("last.parquet"))}); + auto snapshot = std::make_shared(Snapshot{ + .snapshot_id = 10, + .sequence_number = 1, + .timestamp_ms = TimePointMsFromUnixMs(1000), + .manifest_list = WriteManifestList(GetParam(), 10, 0, 1, {first, empty, last}), + }); + ICEBERG_UNWRAP_OR_FAIL(auto table, MakeTable({snapshot}, 10)); + ICEBERG_UNWRAP_OR_FAIL(auto files, MetadataTable::Make(table)); + ICEBERG_UNWRAP_OR_FAIL(auto stream, Scan(files)); + ICEBERG_UNWRAP_OR_FAIL(auto batches, ReadAllBatches(std::move(stream))); + ASSERT_EQ(batches.size(), 1); + ASSERT_EQ(batches[0]->num_rows(), 2); + auto paths = std::static_pointer_cast<::arrow::StringArray>( + batches[0]->GetColumnByName("file_path")); + EXPECT_EQ(paths->GetString(0), "first.parquet"); + EXPECT_EQ(paths->GetString(1), "last.parquet"); +} + +TEST_P(FilesTableTest, ScansDecimalPartitionsAndReadableBounds) { + const int128_t large = (static_cast(1) << 80) + 123; + const std::vector values{123, -456, large, -large, 0}; + ICEBERG_UNWRAP_OR_FAIL(auto table, MakeDecimalPartitionTable(values)); + ICEBERG_UNWRAP_OR_FAIL(auto files, MetadataTable::Make(table)); + ICEBERG_UNWRAP_OR_FAIL(auto stream, Scan(files)); + ICEBERG_UNWRAP_OR_FAIL(auto batches, ReadAllBatches(std::move(stream))); + ASSERT_EQ(batches.size(), 1); + ASSERT_EQ(batches[0]->num_rows(), values.size()); + auto partition = std::static_pointer_cast<::arrow::StructArray>( + batches[0]->GetColumnByName("partition")); + auto amounts = std::static_pointer_cast<::arrow::Decimal128Array>(partition->field(0)); + auto metrics = std::static_pointer_cast<::arrow::StructArray>( + batches[0]->GetColumnByName("readable_metrics")); + auto amount_metrics = + std::static_pointer_cast<::arrow::StructArray>(metrics->GetFieldByName("amount")); + auto lower = std::static_pointer_cast<::arrow::Decimal128Array>( + amount_metrics->GetFieldByName("lower_bound")); + auto upper = std::static_pointer_cast<::arrow::Decimal128Array>( + amount_metrics->GetFieldByName("upper_bound")); + for (size_t index = 0; index < values.size(); ++index) { + ICEBERG_UNWRAP_OR_FAIL(auto expected, Decimal(values[index]).ToString(2)); + EXPECT_EQ(amounts->FormatValue(index), expected); + EXPECT_EQ(lower->FormatValue(index), expected); + EXPECT_EQ(upper->FormatValue(index), expected); + } +} + +TEST_P(FilesTableTest, ScanEmpty) { + ICEBERG_UNWRAP_OR_FAIL(auto table, MakeTable({}, kInvalidSnapshotId)); + ICEBERG_UNWRAP_OR_FAIL(auto metadata_table, MetadataTable::Make(table)); + ICEBERG_UNWRAP_OR_FAIL(auto stream, Scan(metadata_table)); + ICEBERG_UNWRAP_OR_FAIL(auto batches, ReadAllBatches(std::move(stream))); + EXPECT_TRUE(batches.empty()); +} + +TEST_P(FilesTableTest, TasksRetainProjectionAndMetadataWithCustomBatchSize) { + CheckTaskProjectionAndLifetime(MetadataTableKind::kFiles); +} + +TEST_P(FilesTableTest, ConcurrentBuildersRetainTheirOwnSchemas) { + CheckConcurrentBuilders(MetadataTableKind::kFiles); +} + +TEST_P(FilesTableTest, ProjectsReadableMetricsByFieldIdWithRenamedColumn) { + auto snapshot = MakeAppendSnapshot(GetParam(), 10, std::nullopt, 1, {"data.parquet"}); + ICEBERG_UNWRAP_OR_FAIL(auto table, MakeTable({snapshot}, 10)); + ICEBERG_UNWRAP_OR_FAIL(auto files, MetadataTable::Make(table)); + ICEBERG_UNWRAP_OR_FAIL(auto builder, files->NewScan()); + const auto& metrics = files->schema()->fields().back(); + auto projection = std::make_shared(std::vector{ + SchemaField::MakeOptional(metrics.field_id(), "metrics", metrics.type())}); + ICEBERG_UNWRAP_OR_FAIL(auto scan, builder->Project(projection).Build()); + ICEBERG_UNWRAP_OR_FAIL(auto stream, scan->ToArrowArrayStream()); + ICEBERG_UNWRAP_OR_FAIL(auto batches, ReadAllBatches(std::move(stream))); + ASSERT_EQ(batches.size(), 1); + EXPECT_EQ(batches[0]->num_rows(), 1); + ASSERT_EQ(batches[0]->num_columns(), 1); + EXPECT_EQ(batches[0]->schema()->field(0)->name(), "metrics"); +} + +TEST_P(FilesTableTest, ReadableDateBoundsAreConvertedAfterTimestampPromotion) { + if (GetParam() < 3) { + GTEST_SKIP() << "Date promotion requires format version 3"; + } + schema_ = std::make_shared( + std::vector{SchemaField::MakeOptional(1, "event_date", date())}); + for (int32_t days : {7, std::numeric_limits::max()}) { + auto file = MakeDataFile("date.parquet"); + ICEBERG_UNWRAP_OR_FAIL(auto bytes, Conversions::ToBytes(Literal::Date(days))); + file->lower_bounds.emplace(1, bytes); + file->upper_bounds.emplace(1, bytes); + auto manifest = WriteDataManifest(GetParam(), 10, + {MakeEntry(ManifestStatus::kAdded, 10, 1, file)}); + auto snapshot = std::make_shared( + Snapshot{.snapshot_id = 10, + .sequence_number = 1, + .timestamp_ms = TimePointMsFromUnixMs(1000), + .manifest_list = WriteManifestList(GetParam(), 10, 0, 1, {manifest})}); + for (bool nanos : {false, true}) { + SCOPED_TRACE(nanos ? "timestamp_ns" : "timestamp"); + std::shared_ptr promoted_type = timestamp(); + if (nanos) { + promoted_type = timestamp_ns(); + } + auto metadata = MakeTableMetadata({snapshot}, 10); + auto promoted = + std::make_shared(std::vector{SchemaField::MakeOptional( + 1, "event_date", promoted_type)}, + 1); + metadata->schemas = {schema_, promoted}; + metadata->current_schema_id = 1; + metadata->partition_specs = {unpartitioned_spec_}; + ICEBERG_UNWRAP_OR_FAIL( + auto table, Table::Make(TableIdentifier{.name = "table"}, metadata, + "s3://bucket/metadata.json", file_io_, catalog_)); + ICEBERG_UNWRAP_OR_FAIL(auto files, FilesTable::Make(table)); + ICEBERG_UNWRAP_OR_FAIL(auto stream, Scan(files)); + auto batches = ReadAllBatches(std::move(stream)); + if (days == std::numeric_limits::max()) { + EXPECT_THAT(batches, HasErrorMessage("outside")); + continue; + } + ASSERT_THAT(batches, IsOk()); + ASSERT_EQ(batches->size(), 1); + auto metrics_column = (*batches)[0]->GetColumnByName("readable_metrics"); + ASSERT_EQ(metrics_column->type_id(), ::arrow::Type::STRUCT); + auto metrics = std::static_pointer_cast<::arrow::StructArray>(metrics_column); + auto column = std::static_pointer_cast<::arrow::StructArray>(metrics->field(0)); + for (const auto* name : {"lower_bound", "upper_bound"}) { + auto bound = column->GetFieldByName(name); + ASSERT_EQ(bound->type_id(), ::arrow::Type::TIMESTAMP); + auto timestamps = std::static_pointer_cast<::arrow::TimestampArray>(bound); + EXPECT_EQ(timestamps->Value(0), nanos ? 604800000000000LL : 604800000000LL); + } + } + } +} + +INSTANTIATE_TEST_SUITE_P(FormatVersions, FilesTableTest, ::testing::Values(2, 3)); + +} // namespace +} // namespace iceberg diff --git a/src/iceberg/type_fwd.h b/src/iceberg/type_fwd.h index 8e441d874..a42b79acf 100644 --- a/src/iceberg/type_fwd.h +++ b/src/iceberg/type_fwd.h @@ -285,6 +285,7 @@ class PositionDeleteIndex; /// \brief Metadata tables. enum class MetadataTableKind : uint8_t; +class FilesTable; class HistoryTable; class MetadataScanBuilder; class MetadataTable;