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;