diff --git a/src/iceberg/CMakeLists.txt b/src/iceberg/CMakeLists.txt index 4cde4fb23..5bba6e330 100644 --- a/src/iceberg/CMakeLists.txt +++ b/src/iceberg/CMakeLists.txt @@ -58,6 +58,7 @@ set(ICEBERG_SOURCES geospatial.cc incremental_append_scan.cc incremental_changelog_scan.cc + inspect/manifests_table.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..ef7ac765d 100644 --- a/src/iceberg/cpp20_compatibility_internal.h +++ b/src/iceberg/cpp20_compatibility_internal.h @@ -81,6 +81,7 @@ #include "iceberg/incremental_changelog_scan.h" #include "iceberg/inheritable_metadata.h" #include "iceberg/inspect/history_table.h" +#include "iceberg/inspect/manifests_table.h" #include "iceberg/inspect/metadata_scan.h" #include "iceberg/inspect/metadata_scan_task.h" #include "iceberg/inspect/metadata_scan_task_reader.h" diff --git a/src/iceberg/inspect/manifests_table.cc b/src/iceberg/inspect/manifests_table.cc new file mode 100644 index 000000000..93bc91991 --- /dev/null +++ b/src/iceberg/inspect/manifests_table.cc @@ -0,0 +1,292 @@ +/* + * 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/manifests_table.h" + +#include +#include +#include +#include +#include +#include +#include +#include + +#include + +#include "iceberg/arrow_row_builder_internal.h" +#include "iceberg/inspect/metadata_scan_util_internal.h" +#include "iceberg/inspect/metadata_table_util_internal.h" +#include "iceberg/manifest/manifest_list.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" + +namespace iceberg { +namespace { + +struct ManifestRow { + ManifestFile manifest; + std::shared_ptr spec; + std::shared_ptr partition_type; +}; + +Result> ManifestPartitionType(const TableMetadata& metadata, + const PartitionSpec& spec) { + ICEBERG_ASSIGN_OR_RAISE(auto current_schema, metadata.Schema()); + std::vector fields; + fields.reserve(spec.fields().size()); + for (const auto& partition_field : spec.fields()) { + ICEBERG_ASSIGN_OR_RAISE(auto source, + current_schema->FindFieldById(partition_field.source_id())); + // A retained spec can reference a dropped column. Recover its most recent + // known type by field ID, including any promotion before the column was dropped. + if (!source) { + int32_t latest_schema_id = -1; + for (const auto& schema : metadata.schemas) { + ICEBERG_PRECHECK(schema != nullptr, "Table schema cannot be null"); + if (schema->schema_id() > latest_schema_id) { + ICEBERG_ASSIGN_OR_RAISE(auto historical, + schema->FindFieldById(partition_field.source_id())); + if (historical) { + source = historical; + latest_schema_id = schema->schema_id(); + } + } + } + } + auto source_type = source ? source->get().type() : unknown(); + fields.push_back(SchemaField::MakeOptional( + partition_field.field_id(), std::string(partition_field.name()), + partition_field.transform()->ResultType(source_type))); + } + return std::make_shared(std::move(fields)); +} + +Result HumanReadableBound(const PartitionSpec& spec, + const StructType& partition_type, size_t index, + const std::vector& bytes) { + ICEBERG_PRECHECK(index < spec.fields().size() && index < partition_type.fields().size(), + "Partition summary index {} is out of range", index); + auto primitive = internal::checked_pointer_cast( + partition_type.fields()[index].type()); + ICEBERG_ASSIGN_OR_RAISE(auto literal, + Conversions::FromBytes(std::move(primitive), bytes)); + return spec.fields()[index].transform()->ToHumanString(literal); +} + +Status AppendPartitionSummaries(ArrowArray* array, const ManifestRow& row, + const std::shared_ptr& partition_type) { + ICEBERG_PRECHECK(row.manifest.partitions.size() <= row.spec->fields().size(), + "Manifest '{}' has more partition summaries than spec {} fields", + row.manifest.manifest_path, row.spec->spec_id()); + auto* entries = array->children[0]; + ICEBERG_PRECHECK(entries != nullptr && entries->n_children == 4, + "Partition summaries must contain four fields"); + + for (size_t index = 0; index < row.manifest.partitions.size(); ++index) { + const auto& summary = row.manifest.partitions[index]; + ICEBERG_RETURN_UNEXPECTED(AppendBoolean(entries->children[0], summary.contains_null)); + if (summary.contains_nan.has_value()) { + ICEBERG_RETURN_UNEXPECTED( + AppendBoolean(entries->children[1], *summary.contains_nan)); + } else { + ICEBERG_RETURN_UNEXPECTED(AppendNull(entries->children[1])); + } + + if (summary.lower_bound.has_value()) { + ICEBERG_ASSIGN_OR_RAISE( + auto lower, + HumanReadableBound(*row.spec, *partition_type, index, *summary.lower_bound)); + ICEBERG_RETURN_UNEXPECTED(AppendString(entries->children[2], lower)); + } else { + ICEBERG_RETURN_UNEXPECTED(AppendString(entries->children[2], "null")); + } + if (summary.upper_bound.has_value()) { + ICEBERG_ASSIGN_OR_RAISE( + auto upper, + HumanReadableBound(*row.spec, *partition_type, index, *summary.upper_bound)); + ICEBERG_RETURN_UNEXPECTED(AppendString(entries->children[3], upper)); + } else { + ICEBERG_RETURN_UNEXPECTED(AppendString(entries->children[3], "null")); + } + ICEBERG_NANOARROW_RETURN_UNEXPECTED(ArrowArrayFinishElement(entries)); + } + ICEBERG_NANOARROW_RETURN_UNEXPECTED(ArrowArrayFinishElement(array)); + return {}; +} + +Status AppendManifestCount(ArrowArray* array, const std::optional& count, + bool applicable) { + if (!applicable) { + return AppendInt(array, 0); + } + return count.has_value() ? AppendInt(array, *count) : AppendNull(array); +} + +Status AppendManifest(ArrowRowBuilder& builder, const Schema& schema, + const ManifestRow& row) { + const auto& manifest = row.manifest; + const bool data = manifest.content == ManifestContent::kData; + for (size_t index = 0; index < schema.fields().size(); ++index) { + auto* array = builder.column(index); + switch (schema.fields()[index].field_id()) { + case 14: + ICEBERG_RETURN_UNEXPECTED( + AppendInt(array, static_cast(manifest.content))); + break; + case 1: + ICEBERG_RETURN_UNEXPECTED(AppendString(array, manifest.manifest_path)); + break; + case 2: + ICEBERG_RETURN_UNEXPECTED(AppendInt(array, manifest.manifest_length)); + break; + case 3: + ICEBERG_RETURN_UNEXPECTED(AppendInt(array, manifest.partition_spec_id)); + break; + case 4: + ICEBERG_PRECHECK(manifest.added_snapshot_id.has_value(), + "Manifest '{}' has no added snapshot ID", + manifest.manifest_path); + ICEBERG_RETURN_UNEXPECTED(AppendInt(array, *manifest.added_snapshot_id)); + break; + case 5: + ICEBERG_RETURN_UNEXPECTED( + AppendManifestCount(array, manifest.added_files_count, data)); + break; + case 6: + ICEBERG_RETURN_UNEXPECTED( + AppendManifestCount(array, manifest.existing_files_count, data)); + break; + case 7: + ICEBERG_RETURN_UNEXPECTED( + AppendManifestCount(array, manifest.deleted_files_count, data)); + break; + case 15: + ICEBERG_RETURN_UNEXPECTED( + AppendManifestCount(array, manifest.added_files_count, !data)); + break; + case 16: + ICEBERG_RETURN_UNEXPECTED( + AppendManifestCount(array, manifest.existing_files_count, !data)); + break; + case 17: + ICEBERG_RETURN_UNEXPECTED( + AppendManifestCount(array, manifest.deleted_files_count, !data)); + break; + case 8: { + ICEBERG_RETURN_UNEXPECTED( + AppendPartitionSummaries(array, row, row.partition_type)); + break; + } + default: + return InvalidSchema("Unsupported manifests metadata field {}", + schema.fields()[index].field_id()); + } + } + return builder.FinishRow(); +} + +} // namespace + +ManifestsTable::ManifestsTable(std::shared_ptr table) + : MetadataTable(std::move(table)) {} + +ManifestsTable::~ManifestsTable() = default; + +std::shared_ptr ManifestsTable::schema() const { + static const auto schema = std::make_shared(std::vector{ + SchemaField::MakeRequired(14, "content", int32()), + SchemaField::MakeRequired(1, "path", string()), + SchemaField::MakeRequired(2, "length", int64()), + SchemaField::MakeRequired(3, "partition_spec_id", int32()), + SchemaField::MakeRequired(4, "added_snapshot_id", int64()), + // Legacy manifests can omit counts; unknown counts must stay null. + SchemaField::MakeOptional(5, "added_data_files_count", int32()), + SchemaField::MakeOptional(6, "existing_data_files_count", int32()), + SchemaField::MakeOptional(7, "deleted_data_files_count", int32()), + SchemaField::MakeOptional(15, "added_delete_files_count", int32()), + SchemaField::MakeOptional(16, "existing_delete_files_count", int32()), + SchemaField::MakeOptional(17, "deleted_delete_files_count", int32()), + SchemaField::MakeRequired( + 8, "partition_summaries", + list(SchemaField::MakeRequired( + 9, std::string(ListType::kElementName), + struct_({SchemaField::MakeRequired(10, "contains_null", boolean()), + SchemaField::MakeOptional(11, "contains_nan", boolean()), + SchemaField::MakeOptional(12, "lower_bound", string()), + SchemaField::MakeOptional(13, "upper_bound", string())})))), + }); + return schema; +} + +Result> ManifestsTable::Make( + std::shared_ptr table) { + ICEBERG_PRECHECK(table != nullptr, "Table cannot be null"); + return std::unique_ptr(new ManifestsTable(std::move(table))); +} + +Result> ManifestsTable::MakeScan( + internal::MetadataScanContext context) const { + return internal::MakeSnapshotMetadataScan(std::move(context), ReadRows); +} + +Result ManifestsTable::ReadRows( + const internal::MetadataScanContext& context, const std::shared_ptr& io, + int64_t batch_size) { + ICEBERG_ASSIGN_OR_RAISE(auto snapshot, internal::ResolveMetadataTableSnapshot( + *context.table_metadata, context)); + std::vector rows; + std::unordered_map> partition_types; + if (snapshot != nullptr) { + SnapshotReader snapshot_reader(snapshot.get()); + ICEBERG_ASSIGN_OR_RAISE(auto manifests, snapshot_reader.Manifests(io)); + rows.reserve(manifests.size()); + for (const auto& manifest : manifests) { + ICEBERG_ASSIGN_OR_RAISE(auto spec, context.table_metadata->PartitionSpecById( + manifest.partition_spec_id)); + auto [entry, inserted] = partition_types.try_emplace(spec->spec_id()); + if (inserted) { + ICEBERG_ASSIGN_OR_RAISE(entry->second, + ManifestPartitionType(*context.table_metadata, *spec)); + } + rows.push_back(ManifestRow{.manifest = manifest, + .spec = std::move(spec), + .partition_type = entry->second}); + } + } + + return internal::MakeMetadataBatchStream( + context.projected_schema, std::move(rows), batch_size, + [schema = context.projected_schema](ArrowRowBuilder& builder, + const ManifestRow& row) { + return AppendManifest(builder, *schema, row); + }); +} + +} // namespace iceberg diff --git a/src/iceberg/inspect/manifests_table.h b/src/iceberg/inspect/manifests_table.h new file mode 100644 index 000000000..ef81b3b57 --- /dev/null +++ b/src/iceberg/inspect/manifests_table.h @@ -0,0 +1,58 @@ +/* + * 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/manifests_table.h +/// \brief Define the manifests metadata table. + +#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 manifest-list entries for a snapshot. +class ICEBERG_EXPORT ManifestsTable : public MetadataTable { + public: + static Result> Make(std::shared_ptr table); + + ~ManifestsTable() override; + + Kind kind() const noexcept override { return Kind::kManifests; } + + std::shared_ptr schema() const override; + + protected: + 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 ManifestsTable(std::shared_ptr table); +}; + +} // namespace iceberg diff --git a/src/iceberg/inspect/metadata_table.cc b/src/iceberg/inspect/metadata_table.cc index 012ffaa99..e313aa8b3 100644 --- a/src/iceberg/inspect/metadata_table.cc +++ b/src/iceberg/inspect/metadata_table.cc @@ -23,6 +23,7 @@ #include #include "iceberg/inspect/history_table.h" +#include "iceberg/inspect/manifests_table.h" #include "iceberg/inspect/metadata_scan.h" #include "iceberg/inspect/refs_table.h" #include "iceberg/inspect/snapshots_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::kManifests: + return ManifestsTable::Make(std::move(table)); default: break; } diff --git a/src/iceberg/test/CMakeLists.txt b/src/iceberg/test/CMakeLists.txt index 1874fc25e..677813402 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/manifests_table_test.cc) add_iceberg_test(eval_expr_test USE_BUNDLE diff --git a/src/iceberg/test/inspect/manifests_table_test.cc b/src/iceberg/test/inspect/manifests_table_test.cc new file mode 100644 index 000000000..97645eae9 --- /dev/null +++ b/src/iceberg/test/inspect/manifests_table_test.cc @@ -0,0 +1,240 @@ +/* + * 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/manifests_table.h" + +#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 ManifestsTableTest : public MetadataTableScanTestBase {}; + +TEST_P(ManifestsTableTest, 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 manifests, MetadataTable::Make(table)); + ICEBERG_UNWRAP_OR_FAIL(auto manifests_stream, Scan(manifests)); + ICEBERG_UNWRAP_OR_FAIL(auto manifests_batches, + ReadAllBatches(std::move(manifests_stream))); + ASSERT_EQ(manifests_batches.size(), 1); + ASSERT_EQ(manifests_batches[0]->num_rows(), 1); + auto manifest_paths = std::static_pointer_cast<::arrow::StringArray>( + manifests_batches[0]->GetColumnByName("path")); + auto added_files = std::static_pointer_cast<::arrow::Int32Array>( + manifests_batches[0]->GetColumnByName("added_data_files_count")); + EXPECT_FALSE(manifest_paths->GetString(0).empty()); + EXPECT_EQ(added_files->Value(0), 1); +} + +TEST_P(ManifestsTableTest, 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); + } + } +} + +TEST_P(ManifestsTableTest, 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(ManifestsTableTest, PreservesUnknownManifestCounts) { + // Upgraded tables can still reference v1 manifest lists, where counts are optional. + auto unknown = WriteDataManifest( + 1, 10, {MakeEntry(ManifestStatus::kAdded, 10, 0, MakeDataFile("data.parquet"))}); + unknown.added_files_count.reset(); + unknown.existing_files_count.reset(); + unknown.deleted_files_count.reset(); + auto empty = WriteDataManifest(1, 10, {}); + auto snapshot = std::make_shared(Snapshot{ + .snapshot_id = 10, + .sequence_number = 0, + .timestamp_ms = TimePointMsFromUnixMs(1000), + .manifest_list = WriteManifestList(1, 10, 0, 0, {unknown, empty}), + }); + ICEBERG_UNWRAP_OR_FAIL(auto table, MakeTable({snapshot}, 10)); + ICEBERG_UNWRAP_OR_FAIL(auto manifests, MetadataTable::Make(table)); + ICEBERG_UNWRAP_OR_FAIL(auto stream, Scan(manifests)); + ICEBERG_UNWRAP_OR_FAIL(auto batches, ReadAllBatches(std::move(stream))); + ASSERT_EQ(batches.size(), 1); + ASSERT_EQ(batches[0]->num_rows(), 2); + for (int index = 5; index <= 10; ++index) { + auto counts = + std::static_pointer_cast<::arrow::Int32Array>(batches[0]->column(index)); + if (index < 8) { + EXPECT_TRUE(counts->IsNull(0)); + } else { + // Delete counts are definitively zero in a data manifest. + EXPECT_FALSE(counts->IsNull(0)); + EXPECT_EQ(counts->Value(0), 0); + } + // A known zero must remain distinguishable from an unknown count. + EXPECT_FALSE(counts->IsNull(1)); + EXPECT_EQ(counts->Value(1), 0); + EXPECT_TRUE(batches[0]->schema()->field(index)->nullable()); + EXPECT_TRUE(manifests->schema()->fields()[index].optional()); + } +} + +TEST_P(ManifestsTableTest, ManifestNullBoundsUseJavaRendering) { + auto snapshot = MakeAppendSnapshotWithPartitionValues( + GetParam(), 10, std::nullopt, 1, + {{"null.parquet", PartitionValues(Literal::Null(int32()))}}, partitioned_spec_); + ICEBERG_UNWRAP_OR_FAIL(auto table, MakeTable({snapshot}, 10, {}, partitioned_spec_)); + ICEBERG_UNWRAP_OR_FAIL(auto manifests, MetadataTable::Make(table)); + ICEBERG_UNWRAP_OR_FAIL(auto stream, Scan(manifests)); + ICEBERG_UNWRAP_OR_FAIL(auto batches, ReadAllBatches(std::move(stream))); + ASSERT_EQ(batches.size(), 1); + auto summaries = std::static_pointer_cast<::arrow::ListArray>( + batches[0]->GetColumnByName("partition_summaries")); + auto values = std::static_pointer_cast<::arrow::StructArray>(summaries->values()); + ASSERT_EQ(values->length(), 1); + for (int index : {2, 3}) { + auto bound = std::static_pointer_cast<::arrow::StringArray>(values->field(index)); + ASSERT_FALSE(bound->IsNull(0)); + EXPECT_EQ(bound->GetString(0), "null"); + } +} + +TEST_P(ManifestsTableTest, ManifestBoundsSurviveDroppedBucketSource) { + auto snapshot = MakeAppendSnapshotWithPartitionValues( + GetParam(), 10, std::nullopt, 1, + {{"data.parquet", PartitionValues(Literal::Int(7))}}, partitioned_spec_); + auto metadata = MakeTableMetadata({snapshot}, 10); + metadata->schemas.push_back(std::make_shared( + std::vector{SchemaField::MakeRequired(1, "id", int32())}, 1)); + metadata->current_schema_id = 1; + ICEBERG_UNWRAP_OR_FAIL( + auto table, Table::Make(TableIdentifier{.name = "table"}, std::move(metadata), + "s3://bucket/metadata.json", file_io_, catalog_)); + ICEBERG_UNWRAP_OR_FAIL(auto manifests, MetadataTable::Make(table)); + ICEBERG_UNWRAP_OR_FAIL(auto stream, Scan(manifests)); + ICEBERG_UNWRAP_OR_FAIL(auto batches, ReadAllBatches(std::move(stream))); + ASSERT_EQ(batches.size(), 1); + auto summaries = std::static_pointer_cast<::arrow::ListArray>( + batches[0]->GetColumnByName("partition_summaries")); + auto values = std::static_pointer_cast<::arrow::StructArray>(summaries->values()); + ASSERT_EQ(values->length(), 1); + for (int index : {2, 3}) { + auto bound = std::static_pointer_cast<::arrow::StringArray>(values->field(index)); + ASSERT_FALSE(bound->IsNull(0)); + EXPECT_EQ(bound->GetString(0), "7"); + } +} + +TEST_P(ManifestsTableTest, ManifestBoundsUseHistoricalTypesForDroppedSources) { + std::shared_ptr spec; + ICEBERG_UNWRAP_OR_FAIL( + spec, PartitionSpec::Make( + 1, {PartitionField(1, 1000, "id", Transform::Identity()), + PartitionField(2, 1001, "data_prefix", Transform::Truncate(3))})); + auto snapshot = MakeAppendSnapshotWithPartitionValues( + GetParam(), 10, std::nullopt, 1, + {{"data.parquet", PartitionValues({Literal::Int(7), Literal::String("abc")})}}, + spec); + auto metadata = MakeTableMetadata({snapshot}, 10); + auto promoted = std::make_shared( + std::vector{SchemaField::MakeRequired(1, "id", int64()), + SchemaField::MakeRequired(2, "data", string())}, + 1); + auto current = std::make_shared( + std::vector{SchemaField::MakeOptional(3, "extra", string())}, 2); + // Schema history need not be in ID order. The bounds were written before promotion. + metadata->schemas = {promoted, current, schema_}; + metadata->current_schema_id = 2; + metadata->last_column_id = 3; + metadata->partition_specs = {unpartitioned_spec_, spec}; + metadata->last_partition_id = 1001; + ICEBERG_UNWRAP_OR_FAIL( + auto table, Table::Make(TableIdentifier{.name = "table"}, std::move(metadata), + "s3://bucket/metadata.json", file_io_, catalog_)); + ICEBERG_UNWRAP_OR_FAIL(auto manifests, ManifestsTable::Make(table)); + ICEBERG_UNWRAP_OR_FAIL(auto stream, Scan(manifests)); + ICEBERG_UNWRAP_OR_FAIL(auto batches, ReadAllBatches(std::move(stream))); + ASSERT_EQ(batches.size(), 1); + auto summaries = std::static_pointer_cast<::arrow::ListArray>( + batches[0]->GetColumnByName("partition_summaries")); + auto values = std::static_pointer_cast<::arrow::StructArray>(summaries->values()); + ASSERT_EQ(values->length(), 2); + for (int index : {2, 3}) { + auto bounds = std::static_pointer_cast<::arrow::StringArray>(values->field(index)); + EXPECT_EQ(bounds->GetString(0), "7"); + EXPECT_EQ(bounds->GetString(1), "abc"); + } +} + +TEST_P(ManifestsTableTest, 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(ManifestsTableTest, TasksRetainProjectionAndMetadataWithCustomBatchSize) { + CheckTaskProjectionAndLifetime(MetadataTableKind::kManifests); +} + +INSTANTIATE_TEST_SUITE_P(FormatVersions, ManifestsTableTest, ::testing::Values(2, 3)); + +} // namespace +} // namespace iceberg diff --git a/src/iceberg/test/inspect/metadata_table_scan_test_base.h b/src/iceberg/test/inspect/metadata_table_scan_test_base.h new file mode 100644 index 000000000..218fdc325 --- /dev/null +++ b/src/iceberg/test/inspect/metadata_table_scan_test_base.h @@ -0,0 +1,270 @@ +/* + * 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 +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +#include +#include +#include + +#include "iceberg/inspect/metadata_scan.h" +#include "iceberg/inspect/metadata_scan_task_reader.h" +#include "iceberg/inspect/metadata_table.h" +#include "iceberg/row/partition_values.h" +#include "iceberg/table_identifier.h" +#include "iceberg/test/mock_catalog.h" +#include "iceberg/test/scan_test_base.h" +#include "iceberg/util/conversions.h" + +namespace iceberg { + +struct SnapshotSelection { + std::variant snapshot; + std::string ref_name; +}; + +class MetadataTableScanTestBase : public ScanTestBase { + protected: + void CheckConcurrentBuilders(MetadataTableKind kind) { + ICEBERG_UNWRAP_OR_FAIL(auto table, MakeTable({}, kInvalidSnapshotId)); + ICEBERG_UNWRAP_OR_FAIL(auto metadata_table, table->Inspect(kind)); + const auto held_schema = metadata_table->schema(); + const size_t num_fields = held_schema->fields().size(); + std::atomic failures = 0; + std::barrier start(4); + std::vector threads; + for (int thread = 0; thread < 4; ++thread) { + threads.emplace_back([&] { + start.arrive_and_wait(); + for (int iteration = 0; iteration < 1000; ++iteration) { + auto builder = iteration % 2 == 0 ? metadata_table->NewScan() + : MetadataScanBuilder::Make(*metadata_table); + if (!builder || !(*builder)->Build()) { + ++failures; + } + auto schema = metadata_table->schema(); + if (!schema || schema->fields().size() != num_fields || + held_schema->fields().size() != num_fields) { + ++failures; + } + } + }); + } + for (auto& thread : threads) { + thread.join(); + } + EXPECT_EQ(failures.load(), 0); + } + + void CheckTaskProjectionAndLifetime(MetadataTableKind kind) { + std::vector manifests; + for (int32_t index = 0; index < 3; ++index) { + manifests.push_back(WriteDataManifest( + GetParam(), 10, + {MakeEntry( + ManifestStatus::kAdded, 10, 1, + MakeDataFile(std::format("file-{}.parquet", index), + PartitionValues(Literal::Int(index)), partitioned_spec_))}, + partitioned_spec_)); + } + auto first = std::make_shared( + Snapshot{.snapshot_id = 10, + .sequence_number = 1, + .timestamp_ms = TimePointMsFromUnixMs(1000), + .manifest_list = WriteManifestList(GetParam(), 10, 0, 1, manifests)}); + ICEBERG_UNWRAP_OR_FAIL(auto table, MakeTable({first}, 10, {}, partitioned_spec_)); + ICEBERG_UNWRAP_OR_FAIL(auto metadata_table, table->Inspect(kind)); + const auto canonical_schema = metadata_table->schema(); + const auto& fields = canonical_schema->fields(); + auto projection = + std::make_shared(std::vector{fields[1], fields[0]}); + const std::string first_column(projection->fields()[0].name()); + const std::string second_column(projection->fields()[1].name()); + ICEBERG_UNWRAP_OR_FAIL(auto builder, metadata_table->NewScan()); + ICEBERG_UNWRAP_OR_FAIL(auto empty_builder, metadata_table->NewScan()); + ICEBERG_UNWRAP_OR_FAIL(auto invalid_builder, metadata_table->NewScan()); + auto invalid_projection = std::make_shared(std::vector{ + SchemaField::MakeRequired(fields[0].field_id(), "wrong_type", int64())}); + EXPECT_FALSE(invalid_builder->Project(invalid_projection).Build().has_value()); + + // Builders retain metadata even when the source refreshes before Build(). + ASSERT_THAT(RefreshTableSchemaAndPartitionSpec(table, first, arrow::MakeMockFileIO()), + IsOk()); + ICEBERG_UNWRAP_OR_FAIL(auto scan, builder->Project(projection).Build()); + ICEBERG_UNWRAP_OR_FAIL(auto empty_scan, empty_builder->Select({}).Build()); + ICEBERG_UNWRAP_OR_FAIL(auto tasks, scan->PlanTasks()); + ICEBERG_UNWRAP_OR_FAIL(auto empty_tasks, empty_scan->PlanTasks()); + ASSERT_EQ(tasks.size(), 1); + ASSERT_EQ(empty_tasks.size(), 1); + builder.reset(); + empty_builder.reset(); + invalid_builder.reset(); + scan.reset(); + empty_scan.reset(); + metadata_table.reset(); + table.reset(); + + ICEBERG_UNWRAP_OR_FAIL(auto reader, MetadataScanTaskReader::Make({.batch_size = 2})); + ICEBERG_UNWRAP_OR_FAIL(auto stream, reader->Open(*tasks[0])); + ICEBERG_UNWRAP_OR_FAIL(auto repeat_stream, reader->Open(*tasks[0])); + ICEBERG_UNWRAP_OR_FAIL(auto empty_stream, reader->Open(*empty_tasks[0])); + reader.reset(); + tasks.clear(); + empty_tasks.clear(); + for (auto* output : {&stream, &repeat_stream}) { + ICEBERG_UNWRAP_OR_FAIL(auto batches, ReadAllBatches(std::move(*output))); + ASSERT_EQ(batches.size(), 2); + EXPECT_EQ(batches[0]->num_rows(), 2); + EXPECT_EQ(batches[1]->num_rows(), 1); + ASSERT_EQ(batches[0]->num_columns(), 2); + EXPECT_EQ(batches[0]->schema()->field(0)->name(), first_column); + EXPECT_EQ(batches[0]->schema()->field(1)->name(), second_column); + } + ICEBERG_UNWRAP_OR_FAIL(auto empty_batches, ReadAllBatches(std::move(empty_stream))); + ASSERT_EQ(empty_batches.size(), 2); + EXPECT_EQ(empty_batches[0]->num_rows(), 2); + EXPECT_EQ(empty_batches[1]->num_rows(), 1); + EXPECT_EQ(empty_batches[0]->num_columns(), 0); + } + + template + static Result Scan(const std::unique_ptr& table, + const SnapshotSelection& selection = {}) { + ICEBERG_ASSIGN_OR_RAISE(auto builder, table->NewScan()); + if (auto id = std::get_if(&selection.snapshot)) { + builder->UseSnapshot(*id); + } else if (auto time = std::get_if(&selection.snapshot)) { + builder->AsOfTime(time->time_since_epoch().count()); + } + if (!selection.ref_name.empty()) { + builder->UseRef(selection.ref_name); + } + ICEBERG_ASSIGN_OR_RAISE(auto scan, builder->Build()); + return scan->ToArrowArrayStream(); + } + + void SetUp() override { + ScanTestBase::SetUp(); + catalog_ = std::make_shared(); + } + + Result> MakeTable( + std::vector> snapshots, int64_t current_snapshot_id, + std::unordered_map> refs = {}, + std::shared_ptr spec = nullptr) { + auto metadata = MakeTableMetadata(snapshots, current_snapshot_id, refs, spec); + return Table::Make( + TableIdentifier{.ns = Namespace{.levels = {"db"}}, .name = "table"}, + std::move(metadata), "s3://bucket/metadata.json", file_io_, catalog_); + } + + static Result>> ReadAllBatches( + ArrowArrayStream&& stream) { + auto reader = ::arrow::ImportRecordBatchReader(&stream); + if (!reader.ok()) { + return InvalidArrowData(reader.status().ToString()); + } + auto batches = reader.ValueUnsafe()->ToRecordBatches(); + if (!batches.ok()) { + return InvalidArrowData(batches.status().ToString()); + } + return std::move(batches).MoveValueUnsafe(); + } + + Status RefreshTableSchemaAndPartitionSpec( + const std::shared_ptr& table, const std::shared_ptr& first, + std::shared_ptr refreshed_io = nullptr) { + auto old_schema = schema_; + schema_ = std::make_shared( + std::vector{SchemaField::MakeRequired(1, "id", int32()), + SchemaField::MakeRequired(2, "data", string()), + SchemaField::MakeOptional(3, "extra", int64())}, + 1); + std::shared_ptr new_spec; + ICEBERG_ASSIGN_OR_RAISE( + new_spec, + PartitionSpec::Make( + 2, {PartitionField(2, 1000, "data_bucket_16_2", Transform::Bucket(16)), + PartitionField(1, 1001, "id", Transform::Identity())})); + auto second = MakeAppendSnapshotWithPartitionValues( + GetParam(), 20, 10, 2, + {{"one.parquet", PartitionValues({Literal::Int(7), Literal::Int(1)})}, + {"two.parquet", PartitionValues({Literal::Int(7), Literal::Int(2)})}}, + new_spec); + auto metadata = MakeTableMetadata({first, second}, 20, {}, new_spec); + metadata->schemas.insert(metadata->schemas.begin(), old_schema); + metadata->partition_specs.push_back(new_spec); + metadata->last_column_id = 3; + metadata->last_partition_id = 1001; + ICEBERG_ASSIGN_OR_RAISE( + auto refreshed, + Table::Make(table->name(), std::move(metadata), "s3://bucket/refreshed.json", + refreshed_io != nullptr ? std::move(refreshed_io) : file_io_, + catalog_)); + EXPECT_CALL(*catalog_, LoadTable(::testing::_)) + .WillOnce(::testing::Return(refreshed)); + return table->Refresh(); + } + + Result> MakeDecimalPartitionTable( + const std::vector& values) { + schema_ = std::make_shared( + std::vector{SchemaField::MakeRequired(1, "id", int32()), + SchemaField::MakeRequired(2, "amount", decimal(38, 2))}); + ICEBERG_ASSIGN_OR_RAISE( + partitioned_spec_, + PartitionSpec::Make(1, + {PartitionField(2, 1000, "amount", Transform::Identity())})); + std::vector entries; + for (size_t index = 0; index < values.size(); ++index) { + auto literal = Literal::Decimal(values[index], 38, 2); + auto file = MakeDataFile(std::format("decimal-{}.parquet", index), + PartitionValues(literal), partitioned_spec_); + ICEBERG_ASSIGN_OR_RAISE(auto bound, Conversions::ToBytes(literal)); + file->lower_bounds.emplace(2, bound); + file->upper_bounds.emplace(2, bound); + entries.push_back(MakeEntry(ManifestStatus::kAdded, 10, 1, std::move(file))); + } + auto manifest = + WriteDataManifest(GetParam(), 10, std::move(entries), partitioned_spec_); + auto snapshot = std::make_shared(Snapshot{ + .snapshot_id = 10, + .sequence_number = 1, + .timestamp_ms = TimePointMsFromUnixMs(1000), + .manifest_list = WriteManifestList(GetParam(), 10, 0, 1, {manifest}), + }); + return MakeTable({snapshot}, 10, {}, partitioned_spec_); + } + + std::shared_ptr catalog_; +}; + +} // namespace iceberg diff --git a/src/iceberg/type_fwd.h b/src/iceberg/type_fwd.h index 8e441d874..354495e38 100644 --- a/src/iceberg/type_fwd.h +++ b/src/iceberg/type_fwd.h @@ -286,6 +286,7 @@ class PositionDeleteIndex; enum class MetadataTableKind : uint8_t; class HistoryTable; +class ManifestsTable; class MetadataScanBuilder; class MetadataTable; class MetadataScan;