diff --git a/src/iceberg/CMakeLists.txt b/src/iceberg/CMakeLists.txt index 1cd1d4467..6a93950ee 100644 --- a/src/iceberg/CMakeLists.txt +++ b/src/iceberg/CMakeLists.txt @@ -45,9 +45,15 @@ set(ICEBERG_SOURCES file_io_registry.cc file_reader.cc file_writer.cc + inspect/branches_table.cc + inspect/files_table.cc inspect/history_table.cc + inspect/manifests_table.cc inspect/metadata_table.cc + inspect/metadata_table_util_internal.cc + inspect/partitions_table.cc inspect/snapshots_table.cc + inspect/tags_table.cc inheritable_metadata.cc json_serde.cc location_provider.cc diff --git a/src/iceberg/arrow_row_builder.cc b/src/iceberg/arrow_row_builder.cc index 26e7cb4a2..9b07b4f4f 100644 --- a/src/iceberg/arrow_row_builder.cc +++ b/src/iceberg/arrow_row_builder.cc @@ -75,6 +75,8 @@ ArrowRowBuilder::~ArrowRowBuilder() { int64_t ArrowRowBuilder::num_columns() const { return array_.n_children; } +int64_t ArrowRowBuilder::num_rows() const { return array_.length; } + ArrowArray* ArrowRowBuilder::column(int64_t index) { if (index < 0 || index >= array_.n_children) { return nullptr; diff --git a/src/iceberg/arrow_row_builder_internal.h b/src/iceberg/arrow_row_builder_internal.h index db1b66f63..e9c55f07a 100644 --- a/src/iceberg/arrow_row_builder_internal.h +++ b/src/iceberg/arrow_row_builder_internal.h @@ -85,6 +85,9 @@ class ICEBERG_EXPORT ArrowRowBuilder { /// \brief The number of top-level columns in the batch. int64_t num_columns() const; + /// \brief The number of completed rows in the batch. + int64_t num_rows() const; + /// \brief Access the nanoarrow child builder for a top-level column. /// /// \param index Zero-based column index. Returns nullptr if out of range. diff --git a/src/iceberg/inspect/branches_table.cc b/src/iceberg/inspect/branches_table.cc new file mode 100644 index 000000000..80e879fe0 --- /dev/null +++ b/src/iceberg/inspect/branches_table.cc @@ -0,0 +1,108 @@ +/* + * 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/branches_table.h" + +#include +#include +#include +#include +#include +#include + +#include "iceberg/arrow_row_builder_internal.h" +#include "iceberg/inspect/metadata_table_stream_internal.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/type.h" +#include "iceberg/util/macros.h" + +namespace iceberg { +namespace { + +struct BranchRow { + std::string name; + int64_t snapshot_id; + std::optional max_ref_age_ms; + std::optional min_snapshots_to_keep; + std::optional max_snapshot_age_ms; +}; + +Status AppendOptional(ArrowArray* array, const auto& value) { + if (!value.has_value()) { + return AppendNull(array); + } + return AppendInt(array, static_cast(*value)); +} + +Status AppendBranch(ArrowRowBuilder& builder, const BranchRow& branch) { + ICEBERG_RETURN_UNEXPECTED(AppendString(builder.column(0), branch.name)); + ICEBERG_RETURN_UNEXPECTED(AppendInt(builder.column(1), branch.snapshot_id)); + ICEBERG_RETURN_UNEXPECTED(AppendOptional(builder.column(2), branch.max_ref_age_ms)); + ICEBERG_RETURN_UNEXPECTED( + AppendOptional(builder.column(3), branch.min_snapshots_to_keep)); + ICEBERG_RETURN_UNEXPECTED( + AppendOptional(builder.column(4), branch.max_snapshot_age_ms)); + return builder.FinishRow(); +} + +} // namespace + +BranchesTable::BranchesTable(std::shared_ptr table) + : MetadataTable(std::move(table)) {} + +BranchesTable::~BranchesTable() = default; + +const std::shared_ptr& BranchesTable::schema() const { + static const auto schema = std::make_shared(std::vector{ + SchemaField::MakeRequired(1, "name", string()), + SchemaField::MakeRequired(2, "snapshot_id", int64()), + SchemaField::MakeOptional(3, "max_reference_age_in_ms", int64()), + SchemaField::MakeOptional(4, "min_snapshots_to_keep", int32()), + SchemaField::MakeOptional(5, "max_snapshot_age_in_ms", int64()), + }); + return schema; +} + +Result> BranchesTable::Make(std::shared_ptr
table) { + ICEBERG_PRECHECK(table != nullptr, "Table cannot be null"); + return std::unique_ptr(new BranchesTable(std::move(table))); +} + +Result BranchesTable::Scan() { + std::vector rows; + for (const auto& [name, ref] : source_table()->metadata()->refs) { + if (ref == nullptr || ref->type() != SnapshotRefType::kBranch) { + continue; + } + const auto& retention = std::get(ref->retention); + rows.push_back(BranchRow{.name = name, + .snapshot_id = ref->snapshot_id, + .max_ref_age_ms = retention.max_ref_age_ms, + .min_snapshots_to_keep = retention.min_snapshots_to_keep, + .max_snapshot_age_ms = retention.max_snapshot_age_ms}); + } + std::ranges::sort(rows, {}, &BranchRow::name); + return internal::MakeMetadataTableStream(*schema(), std::move(rows), AppendBranch); +} + +} // namespace iceberg diff --git a/src/iceberg/inspect/branches_table.h b/src/iceberg/inspect/branches_table.h new file mode 100644 index 000000000..74e12c34e --- /dev/null +++ b/src/iceberg/inspect/branches_table.h @@ -0,0 +1,51 @@ +/* + * 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/branches_table.h +/// \brief Define the branches metadata table. + +#include + +#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 the table's branch snapshot references. +class ICEBERG_EXPORT BranchesTable : public MetadataTable { + public: + static Result> Make(std::shared_ptr
table); + + ~BranchesTable() override; + + Kind kind() const noexcept override { return Kind::kBranches; } + + const std::shared_ptr& schema() const override; + + Result Scan() override; + + private: + explicit BranchesTable(std::shared_ptr
table); +}; + +} // namespace iceberg diff --git a/src/iceberg/inspect/files_table.cc b/src/iceberg/inspect/files_table.cc new file mode 100644 index 000000000..3405e0f60 --- /dev/null +++ b/src/iceberg/inspect/files_table.cc @@ -0,0 +1,75 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +#include "iceberg/inspect/files_table.h" + +#include +#include + +#include "iceberg/inspect/metadata_table_stream_internal.h" +#include "iceberg/inspect/metadata_table_util_internal.h" +#include "iceberg/schema.h" +#include "iceberg/table.h" +#include "iceberg/type.h" +#include "iceberg/util/macros.h" + +namespace iceberg { + +FilesTable::FilesTable(std::shared_ptr
table, std::shared_ptr schema, + std::shared_ptr table_schema, + std::shared_ptr partition_type) + : TimeTravelMetadataTable(std::move(table)), + schema_(std::move(schema)), + table_schema_(std::move(table_schema)), + partition_type_(std::move(partition_type)) {} + +FilesTable::~FilesTable() = default; + +const std::shared_ptr& FilesTable::schema() const { return schema_; } + +Result> FilesTable::Make(std::shared_ptr
table) { + ICEBERG_PRECHECK(table != nullptr, "Table cannot be null"); + ICEBERG_ASSIGN_OR_RAISE(auto table_schema, table->schema()); + ICEBERG_ASSIGN_OR_RAISE(auto partition_type, internal::UnifiedPartitionType(*table)); + ICEBERG_ASSIGN_OR_RAISE(auto schema, + internal::FilesTableSchema(*table_schema, partition_type)); + return std::unique_ptr(new FilesTable(std::move(table), std::move(schema), + std::move(table_schema), + std::move(partition_type))); +} + +Result FilesTable::ScanSnapshot( + const SnapshotSelection& snapshot_selection) { + ICEBERG_ASSIGN_OR_RAISE(auto snapshot, internal::ResolveMetadataTableSnapshot( + *source_table(), snapshot_selection)); + ICEBERG_ASSIGN_OR_RAISE(auto files, internal::LoadLiveFiles(*source_table(), snapshot)); + auto schema = schema_; + auto table_schema = table_schema_; + auto partition_type = partition_type_; + return internal::MakeMetadataTableStream( + *schema_, std::move(files), + [schema = std::move(schema), 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, + 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..79fd5dbc0 --- /dev/null +++ b/src/iceberg/inspect/files_table.h @@ -0,0 +1,59 @@ +/* + * 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 "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 TimeTravelMetadataTable { + public: + static Result> Make(std::shared_ptr
table); + + ~FilesTable() override; + + Kind kind() const noexcept override { return Kind::kFiles; } + + const std::shared_ptr& schema() const override; + + protected: + Result ScanSnapshot( + const SnapshotSelection& snapshot_selection) override; + + private: + FilesTable(std::shared_ptr
table, std::shared_ptr schema, + std::shared_ptr table_schema, + std::shared_ptr partition_type); + + std::shared_ptr schema_; + std::shared_ptr table_schema_; + std::shared_ptr partition_type_; +}; + +} // namespace iceberg diff --git a/src/iceberg/inspect/history_table.cc b/src/iceberg/inspect/history_table.cc index 7fa840043..03bbcca97 100644 --- a/src/iceberg/inspect/history_table.cc +++ b/src/iceberg/inspect/history_table.cc @@ -26,37 +26,32 @@ #include "iceberg/schema.h" #include "iceberg/schema_field.h" #include "iceberg/table.h" -#include "iceberg/table_identifier.h" #include "iceberg/type.h" +#include "iceberg/util/macros.h" namespace iceberg { -namespace { -std::shared_ptr MakeHistoryTableSchema() { - return std::make_shared(std::vector{ +HistoryTable::HistoryTable(std::shared_ptr
table) + : MetadataTable(std::move(table)) {} + +HistoryTable::~HistoryTable() = default; + +const std::shared_ptr& HistoryTable::schema() const { + static const auto schema = std::make_shared(std::vector{ SchemaField::MakeRequired(1, "made_current_at", timestamp_tz()), SchemaField::MakeRequired(2, "snapshot_id", int64()), SchemaField::MakeOptional(3, "parent_id", int64()), SchemaField::MakeRequired(4, "is_current_ancestor", boolean())}); + return schema; } -TableIdentifier MakeHistoryTableName(const TableIdentifier& source_name) { - return TableIdentifier{.ns = source_name.ns, .name = source_name.name + ".history"}; -} - -} // namespace - -HistoryTable::HistoryTable(std::shared_ptr
table) - : MetadataTable(table, MakeHistoryTableName(table->name()), - MakeHistoryTableSchema()) {} - -HistoryTable::~HistoryTable() = default; - Result> HistoryTable::Make(std::shared_ptr
table) { - if (table == nullptr) [[unlikely]] { - return InvalidArgument("Table cannot be null"); - } + ICEBERG_PRECHECK(table != nullptr, "Table cannot be null"); return std::unique_ptr(new HistoryTable(std::move(table))); } +Result HistoryTable::Scan() { + return NotSupported("Scan is not supported for the history table"); +} + } // namespace iceberg diff --git a/src/iceberg/inspect/history_table.h b/src/iceberg/inspect/history_table.h index 21f1f8002..7d863fc2c 100644 --- a/src/iceberg/inspect/history_table.h +++ b/src/iceberg/inspect/history_table.h @@ -40,6 +40,10 @@ class ICEBERG_EXPORT HistoryTable : public MetadataTable { Kind kind() const noexcept override { return Kind::kHistory; } + const std::shared_ptr& schema() const override; + + Result Scan() override; + private: explicit HistoryTable(std::shared_ptr
table); }; diff --git a/src/iceberg/inspect/manifests_table.cc b/src/iceberg/inspect/manifests_table.cc new file mode 100644 index 000000000..7e8c5c358 --- /dev/null +++ b/src/iceberg/inspect/manifests_table.cc @@ -0,0 +1,208 @@ +/* + * 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 "iceberg/arrow/nanoarrow_status_internal.h" +#include "iceberg/arrow_row_builder_internal.h" +#include "iceberg/inspect/metadata_table_stream_internal.h" +#include "iceberg/inspect/metadata_table_util_internal.h" +#include "iceberg/manifest/manifest_list.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/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; +}; + +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(AppendNull(entries->children[2])); + } + 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(AppendNull(entries->children[3])); + } + ICEBERG_NANOARROW_RETURN_UNEXPECTED(ArrowArrayFinishElement(entries)); + } + ICEBERG_NANOARROW_RETURN_UNEXPECTED(ArrowArrayFinishElement(array)); + return {}; +} + +Status AppendManifest(ArrowRowBuilder& builder, const ManifestRow& row, + const std::shared_ptr& table_schema) { + const auto& manifest = row.manifest; + ICEBERG_RETURN_UNEXPECTED( + AppendInt(builder.column(0), static_cast(manifest.content))); + ICEBERG_RETURN_UNEXPECTED(AppendString(builder.column(1), manifest.manifest_path)); + ICEBERG_RETURN_UNEXPECTED(AppendInt(builder.column(2), manifest.manifest_length)); + ICEBERG_RETURN_UNEXPECTED(AppendInt(builder.column(3), manifest.partition_spec_id)); + ICEBERG_RETURN_UNEXPECTED(AppendInt(builder.column(4), manifest.added_snapshot_id)); + + const bool data = manifest.content == ManifestContent::kData; + ICEBERG_RETURN_UNEXPECTED( + AppendInt(builder.column(5), data ? manifest.added_files_count.value_or(0) : 0)); + ICEBERG_RETURN_UNEXPECTED( + AppendInt(builder.column(6), data ? manifest.existing_files_count.value_or(0) : 0)); + ICEBERG_RETURN_UNEXPECTED( + AppendInt(builder.column(7), data ? manifest.deleted_files_count.value_or(0) : 0)); + ICEBERG_RETURN_UNEXPECTED( + AppendInt(builder.column(8), data ? 0 : manifest.added_files_count.value_or(0))); + ICEBERG_RETURN_UNEXPECTED( + AppendInt(builder.column(9), data ? 0 : manifest.existing_files_count.value_or(0))); + ICEBERG_RETURN_UNEXPECTED( + AppendInt(builder.column(10), data ? 0 : manifest.deleted_files_count.value_or(0))); + + ICEBERG_ASSIGN_OR_RAISE(auto partition_type, row.spec->PartitionType(*table_schema)); + ICEBERG_RETURN_UNEXPECTED(AppendPartitionSummaries( + builder.column(11), row, std::shared_ptr(std::move(partition_type)))); + return builder.FinishRow(); +} + +} // namespace + +ManifestsTable::ManifestsTable(std::shared_ptr
table) + : TimeTravelMetadataTable(std::move(table)) {} + +ManifestsTable::~ManifestsTable() = default; + +const 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()), + SchemaField::MakeRequired(5, "added_data_files_count", int32()), + SchemaField::MakeRequired(6, "existing_data_files_count", int32()), + SchemaField::MakeRequired(7, "deleted_data_files_count", int32()), + SchemaField::MakeRequired(15, "added_delete_files_count", int32()), + SchemaField::MakeRequired(16, "existing_delete_files_count", int32()), + SchemaField::MakeRequired(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::ScanSnapshot( + const SnapshotSelection& snapshot_selection) { + ICEBERG_ASSIGN_OR_RAISE(auto snapshot, internal::ResolveMetadataTableSnapshot( + *source_table(), snapshot_selection)); + std::vector rows; + if (snapshot != nullptr) { + ICEBERG_ASSIGN_OR_RAISE(auto specs_ref, source_table()->specs()); + SnapshotCache snapshot_cache(snapshot.get()); + ICEBERG_ASSIGN_OR_RAISE(auto manifests, + snapshot_cache.Manifests(source_table()->io())); + rows.reserve(manifests.size()); + for (const auto& manifest : manifests) { + auto spec = specs_ref.get().find(manifest.partition_spec_id); + ICEBERG_CHECK(spec != specs_ref.get().end(), + "Cannot find partition spec {} for manifest '{}'", + manifest.partition_spec_id, manifest.manifest_path); + ICEBERG_PRECHECK(spec->second != nullptr, "Partition spec {} is null", + manifest.partition_spec_id); + rows.push_back(ManifestRow{.manifest = manifest, .spec = spec->second}); + } + } + + ICEBERG_ASSIGN_OR_RAISE(auto table_schema, source_table()->schema()); + return internal::MakeMetadataTableStream( + *schema(), std::move(rows), + [table_schema = std::move(table_schema)](ArrowRowBuilder& builder, + const ManifestRow& row) { + return AppendManifest(builder, row, table_schema); + }); +} + +} // namespace iceberg diff --git a/src/iceberg/inspect/manifests_table.h b/src/iceberg/inspect/manifests_table.h new file mode 100644 index 000000000..383945003 --- /dev/null +++ b/src/iceberg/inspect/manifests_table.h @@ -0,0 +1,53 @@ +/* + * 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/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 TimeTravelMetadataTable { + public: + static Result> Make(std::shared_ptr
table); + + ~ManifestsTable() override; + + Kind kind() const noexcept override { return Kind::kManifests; } + + const std::shared_ptr& schema() const override; + + protected: + Result ScanSnapshot( + const SnapshotSelection& snapshot_selection) override; + + private: + explicit ManifestsTable(std::shared_ptr
table); +}; + +} // namespace iceberg diff --git a/src/iceberg/inspect/meson.build b/src/iceberg/inspect/meson.build index 5c738008a..3d543f057 100644 --- a/src/iceberg/inspect/meson.build +++ b/src/iceberg/inspect/meson.build @@ -16,6 +16,15 @@ # under the License. install_headers( - ['history_table.h', 'metadata_table.h', 'snapshots_table.h'], + [ + 'branches_table.h', + 'files_table.h', + 'history_table.h', + 'manifests_table.h', + 'metadata_table.h', + 'partitions_table.h', + 'snapshots_table.h', + 'tags_table.h', + ], subdir: 'iceberg/inspect', ) diff --git a/src/iceberg/inspect/metadata_table.cc b/src/iceberg/inspect/metadata_table.cc index 5e9504003..7bc94c511 100644 --- a/src/iceberg/inspect/metadata_table.cc +++ b/src/iceberg/inspect/metadata_table.cc @@ -22,33 +22,33 @@ #include #include -#include "iceberg/inspect/history_table.h" -#include "iceberg/inspect/snapshots_table.h" - namespace iceberg { -MetadataTable::MetadataTable(std::shared_ptr
source_table, - TableIdentifier identifier, std::shared_ptr schema) - : identifier_(std::move(identifier)), - schema_(std::move(schema)), - source_table_(std::move(source_table)) {} +MetadataTable::MetadataTable(std::shared_ptr
source_table) + : source_table_(std::move(source_table)) {} MetadataTable::~MetadataTable() = default; -Result> MetadataTable::Make(std::shared_ptr
table, - Kind kind) { - if (table == nullptr) [[unlikely]] { - return InvalidArgument("Table cannot be null"); - } +bool MetadataTable::supports_time_travel() const noexcept { return false; } + +const std::shared_ptr
& MetadataTable::source_table() const { + return source_table_; +} + +TimeTravelMetadataTable::TimeTravelMetadataTable(std::shared_ptr
source_table) + : MetadataTable(std::move(source_table)) {} - switch (kind) { - case Kind::kSnapshots: - return SnapshotsTable::Make(table); - case Kind::kHistory: - return HistoryTable::Make(table); - } +TimeTravelMetadataTable::~TimeTravelMetadataTable() = default; + +bool TimeTravelMetadataTable::supports_time_travel() const noexcept { return true; } + +Result TimeTravelMetadataTable::Scan() { + return ScanSnapshot(SnapshotSelection{}); +} - return NotSupported("Unsupported metadata table type"); +Result TimeTravelMetadataTable::Scan( + const SnapshotSelection& snapshot_selection) { + return ScanSnapshot(snapshot_selection); } } // namespace iceberg diff --git a/src/iceberg/inspect/metadata_table.h b/src/iceberg/inspect/metadata_table.h index 51c5f7920..7a91ca292 100644 --- a/src/iceberg/inspect/metadata_table.h +++ b/src/iceberg/inspect/metadata_table.h @@ -20,46 +20,113 @@ #pragma once /// \file iceberg/inspect/metadata_table.h -/// \brief Define base APIs for metadata tables. +/// \brief Base APIs for inspecting Iceberg metadata tables. +#include #include +#include +#include +#include +#include "iceberg/arrow_c_data.h" #include "iceberg/iceberg_export.h" #include "iceberg/result.h" -#include "iceberg/table_identifier.h" #include "iceberg/type_fwd.h" +#include "iceberg/util/timepoint.h" namespace iceberg { -/// \brief Base class for Iceberg metadata tables. +/// \brief Base interface for an Iceberg metadata table. class ICEBERG_EXPORT MetadataTable { public: + /// \brief Supported metadata table kinds. enum class Kind { kSnapshots, kHistory, + kBranches, + kTags, + kFiles, + kPartitions, + kManifests, }; - static Result> Make(std::shared_ptr
table, - Kind kind); + /// \brief Maximum number of rows emitted in each Arrow batch. + static constexpr int64_t kBatchSize = 1024; + + /// \brief Create a metadata table of the requested concrete type. + /// + /// \tparam MetadataTableType Concrete class derived from MetadataTable. + /// \param table Source table whose metadata will be exposed. + /// \return The constructed metadata table, or an error. + template + requires std::derived_from + static Result> Make(std::shared_ptr
table) { + return MetadataTableType::Make(std::move(table)); + } virtual ~MetadataTable(); + /// \brief Return this metadata table's kind. virtual Kind kind() const noexcept = 0; - const TableIdentifier& name() const { return identifier_; } + /// \brief Return the schema of rows emitted by scans. + virtual const std::shared_ptr& schema() const = 0; + + /// \brief Return the source table whose metadata is exposed. + const std::shared_ptr
& source_table() const; - const std::shared_ptr& schema() const { return schema_; } + /// \brief Return whether this metadata table supports time travel. + virtual bool supports_time_travel() const noexcept; - const std::shared_ptr
& source_table() const { return source_table_; } + /// \brief Scan the metadata table without time travel. + /// + /// The caller owns the returned stream and must release it with + /// ArrowArrayStreamRelease. + virtual Result Scan() = 0; protected: - explicit MetadataTable(std::shared_ptr
source_table, TableIdentifier identifier, - std::shared_ptr schema); + explicit MetadataTable(std::shared_ptr
source_table); private: - TableIdentifier identifier_; - std::shared_ptr schema_; std::shared_ptr
source_table_; }; +/// \brief Snapshot selection parameters for a time-travel scan. +struct SnapshotSelection { + /// \brief Select the current snapshot, a snapshot ID, or an as-of timestamp. + /// + /// std::monostate selects the current snapshot. + std::variant snapshot; + + /// \brief Resolve the snapshot relative to this branch or tag. + /// + /// An empty string uses the main branch. + std::string ref_name; +}; + +/// \brief Base interface for metadata tables that support time travel. +class ICEBERG_EXPORT TimeTravelMetadataTable : public MetadataTable { + public: + ~TimeTravelMetadataTable() override; + + /// \brief Return true because this interface supports time travel. + bool supports_time_travel() const noexcept final; + + /// \brief Scan using the current snapshot on the main branch. + Result Scan() final; + + /// \brief Scan using the requested snapshot selection. + /// + /// \param snapshot_selection Snapshot ID, timestamp, and optional ref selection. + /// \return An Arrow stream containing the metadata table rows, or an error. + Result Scan(const SnapshotSelection& snapshot_selection); + + protected: + explicit TimeTravelMetadataTable(std::shared_ptr
source_table); + + /// \brief Implement a scan for the requested snapshot selection. + virtual Result ScanSnapshot( + const SnapshotSelection& snapshot_selection) = 0; +}; + } // namespace iceberg diff --git a/src/iceberg/inspect/metadata_table_stream_internal.h b/src/iceberg/inspect/metadata_table_stream_internal.h new file mode 100644 index 000000000..887f56f9d --- /dev/null +++ b/src/iceberg/inspect/metadata_table_stream_internal.h @@ -0,0 +1,120 @@ +/* + * 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 "iceberg/arrow/nanoarrow_status_internal.h" +#include "iceberg/arrow_c_data_util_internal.h" +#include "iceberg/arrow_row_builder_internal.h" +#include "iceberg/inspect/metadata_table.h" +#include "iceberg/schema.h" +#include "iceberg/schema_internal.h" +#include "iceberg/util/macros.h" + +namespace iceberg::internal { + +/// \brief Arrow stream backed by a fixed set of metadata-table rows. +template +class MetadataTableRowsStream { + public: + using AppendRow = std::function; + + static Result> Make(const Schema& schema, + std::vector rows, + AppendRow append_row) { + ArrowSchema arrow_schema{}; + ICEBERG_RETURN_UNEXPECTED(ToArrowSchema(schema, &arrow_schema)); + return std::unique_ptr(new MetadataTableRowsStream( + std::move(rows), std::move(append_row), std::move(arrow_schema))); + } + + ~MetadataTableRowsStream() { + auto status = Close(); + static_cast(status); + } + + Status Close() { + rows_.clear(); + append_row_ = nullptr; + if (arrow_schema_.release != nullptr) { + ArrowSchemaRelease(&arrow_schema_); + } + return {}; + } + + Result> Next() { + ICEBERG_PRECHECK(arrow_schema_.release != nullptr, + "Cannot read from a closed metadata table stream"); + if (next_row_ == rows_.size()) { + return std::nullopt; + } + + ICEBERG_ASSIGN_OR_RAISE(auto builder, ArrowRowBuilder::Make(&arrow_schema_)); + while (next_row_ < rows_.size() && builder.num_rows() < MetadataTable::kBatchSize) { + ICEBERG_RETURN_UNEXPECTED(append_row_(builder, rows_[next_row_++])); + } + + ICEBERG_ASSIGN_OR_RAISE(auto array, std::move(builder).Finish()); + return array; + } + + Result Schema() { + ICEBERG_PRECHECK(arrow_schema_.release != nullptr, + "Cannot read schema from a closed metadata table stream"); + ArrowSchema schema_copy{}; + ICEBERG_NANOARROW_RETURN_UNEXPECTED( + ArrowSchemaDeepCopy(&arrow_schema_, &schema_copy)); + return schema_copy; + } + + private: + MetadataTableRowsStream(std::vector rows, AppendRow append_row, + ArrowSchema arrow_schema) + : rows_(std::move(rows)), + append_row_(std::move(append_row)), + arrow_schema_(std::move(arrow_schema)) {} + + std::vector rows_; + AppendRow append_row_; + ArrowSchema arrow_schema_{}; + size_t next_row_ = 0; +}; + +template +Result MakeMetadataTableStream(const Schema& schema, + std::vector rows, + AppendRow append_row) { + ICEBERG_ASSIGN_OR_RAISE( + auto stream, + MetadataTableRowsStream::Make( + schema, std::move(rows), + typename MetadataTableRowsStream::AppendRow(std::move(append_row)))); + return MakeArrowArrayStream(std::move(stream)); +} + +} // namespace iceberg::internal diff --git a/src/iceberg/inspect/metadata_table_util_internal.cc b/src/iceberg/inspect/metadata_table_util_internal.cc new file mode 100644 index 000000000..021f2a720 --- /dev/null +++ b/src/iceberg/inspect/metadata_table_util_internal.cc @@ -0,0 +1,642 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +#include "iceberg/inspect/metadata_table_util_internal.h" + +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +#include + +#include "iceberg/arrow/nanoarrow_status_internal.h" +#include "iceberg/arrow_row_builder_internal.h" +#include "iceberg/constants.h" +#include "iceberg/file_format.h" +#include "iceberg/manifest/manifest_list.h" +#include "iceberg/manifest/manifest_reader.h" +#include "iceberg/partition_spec.h" +#include "iceberg/schema.h" +#include "iceberg/schema_field.h" +#include "iceberg/snapshot.h" +#include "iceberg/table.h" +#include "iceberg/table_metadata.h" +#include "iceberg/transform.h" +#include "iceberg/type.h" +#include "iceberg/util/checked_cast.h" +#include "iceberg/util/conversions.h" +#include "iceberg/util/macros.h" +#include "iceberg/util/snapshot_util_internal.h" + +namespace iceberg::internal { +namespace { + +Result> SnapshotAtRef(const Table& table, + std::string_view ref_name) { + const auto& metadata = table.metadata(); + ICEBERG_PRECHECK(metadata != nullptr, "Table metadata cannot be null"); + + if (ref_name.empty() || ref_name == SnapshotRef::kMainBranch) { + if (metadata->current_snapshot_id == kInvalidSnapshotId) { + return std::shared_ptr{nullptr}; + } + return metadata->SnapshotById(metadata->current_snapshot_id); + } + + auto ref = metadata->refs.find(std::string(ref_name)); + ICEBERG_CHECK(ref != metadata->refs.end(), "Cannot find snapshot reference '{}'", + ref_name); + ICEBERG_PRECHECK(ref->second != nullptr, "Snapshot reference '{}' is null", ref_name); + return metadata->SnapshotById(ref->second->snapshot_id); +} + +Result IsAncestorOf(const Table& table, int64_t ancestor_id, + const std::shared_ptr& head) { + std::unordered_set visited; + auto current = head; + while (current != nullptr) { + if (!visited.insert(current->snapshot_id).second) { + return Invalid("Cycle detected in snapshot ancestry at {}", current->snapshot_id); + } + if (current->snapshot_id == ancestor_id) { + return true; + } + if (!current->parent_snapshot_id.has_value()) { + break; + } + auto parent = table.SnapshotById(*current->parent_snapshot_id); + if (!parent.has_value()) { + if (parent.error().kind == ErrorKind::kNotFound) { + break; + } + return std::unexpected(parent.error()); + } + current = std::move(parent).value(); + } + return false; +} + +Status AppendLiteral(ArrowArray* array, const Literal& literal) { + if (literal.IsNull()) { + return AppendNull(array); + } + if (literal.IsAboveMax() || literal.IsBelowMin()) { + return InvalidArgument("Cannot append non-value partition literal {}", + literal.ToString()); + } + + switch (literal.type()->type_id()) { + case TypeId::kBoolean: + return AppendBoolean(array, std::get(literal.value())); + case TypeId::kInt: + case TypeId::kDate: + return AppendInt(array, std::get(literal.value())); + case TypeId::kLong: + case TypeId::kTime: + case TypeId::kTimestamp: + case TypeId::kTimestampTz: + case TypeId::kTimestampNs: + case TypeId::kTimestampTzNs: + return AppendInt(array, std::get(literal.value())); + case TypeId::kFloat: + return AppendDouble(array, std::get(literal.value())); + case TypeId::kDouble: + return AppendDouble(array, std::get(literal.value())); + case TypeId::kString: + return AppendString(array, std::get(literal.value())); + case TypeId::kBinary: + case TypeId::kFixed: + return AppendBytes(array, std::get>(literal.value())); + case TypeId::kDecimal: + return AppendBytes(array, std::get(literal.value()).ToBytes()); + case TypeId::kUuid: + return AppendBytes(array, std::get(literal.value()).bytes()); + case TypeId::kUnknown: + case TypeId::kStruct: + case TypeId::kList: + case TypeId::kMap: + case TypeId::kVariant: + case TypeId::kGeometry: + case TypeId::kGeography: + return NotSupported("Cannot append partition literal of type {}", + literal.type()->ToString()); + } + std::unreachable(); +} + +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()); + 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> ResolveMetadataTableSnapshot( + const Table& table, const SnapshotSelection& selection) { + ICEBERG_ASSIGN_OR_RAISE(auto head, SnapshotAtRef(table, selection.ref_name)); + + if (std::holds_alternative(selection.snapshot)) { + return head; + } + + if (const auto* snapshot_id = std::get_if(&selection.snapshot)) { + ICEBERG_ASSIGN_OR_RAISE(auto selected, table.SnapshotById(*snapshot_id)); + if (!selection.ref_name.empty()) { + ICEBERG_ASSIGN_OR_RAISE(auto is_ancestor, IsAncestorOf(table, *snapshot_id, head)); + ICEBERG_CHECK(is_ancestor, "Snapshot {} is not reachable from reference '{}'", + *snapshot_id, selection.ref_name); + } + return selected; + } + + const auto timestamp = std::get(selection.snapshot); + if (selection.ref_name.empty() || selection.ref_name == SnapshotRef::kMainBranch) { + ICEBERG_ASSIGN_OR_RAISE(auto snapshot_id, + SnapshotUtil::SnapshotIdAsOfTime(table, timestamp)); + return table.SnapshotById(snapshot_id); + } + + std::shared_ptr selected; + std::unordered_set visited; + auto current = head; + while (current != nullptr) { + if (!visited.insert(current->snapshot_id).second) { + return Invalid("Cycle detected in snapshot ancestry at {}", current->snapshot_id); + } + if (current->timestamp_ms <= timestamp && + (selected == nullptr || current->timestamp_ms > selected->timestamp_ms)) { + selected = current; + } + if (!current->parent_snapshot_id.has_value()) { + break; + } + auto parent = table.SnapshotById(*current->parent_snapshot_id); + if (!parent.has_value()) { + if (parent.error().kind == ErrorKind::kNotFound) { + break; + } + return std::unexpected(parent.error()); + } + current = std::move(parent).value(); + } + ICEBERG_CHECK(selected != nullptr, "Cannot find a snapshot at or before the timestamp"); + return selected; +} + +Result> UnifiedPartitionType(const Table& table) { + ICEBERG_ASSIGN_OR_RAISE(auto schema, table.schema()); + ICEBERG_ASSIGN_OR_RAISE(auto specs_ref, table.specs()); + + std::vector> specs; + specs.reserve(specs_ref.get().size()); + for (const auto& [_, spec] : specs_ref.get()) { + ICEBERG_PRECHECK(spec != nullptr, "Partition spec cannot be null"); + specs.push_back(spec); + } + std::ranges::sort(specs, std::greater{}, &PartitionSpec::spec_id); + + std::unordered_set active_field_ids; + for (const auto& spec : specs) { + for (const auto& field : spec->fields()) { + ICEBERG_PRECHECK(field.transform() != nullptr, + "Partition field {} has a null transform", field.field_id()); + ICEBERG_CHECK(field.transform()->transform_type() != TransformType::kUnknown, + "Cannot build table partition type with unknown transform '{}'", + field.transform()->ToString()); + ICEBERG_ASSIGN_OR_RAISE(auto source_field, + schema->FindFieldById(field.source_id())); + if (source_field.has_value()) { + active_field_ids.insert(field.field_id()); + } + } + } + + struct ProjectedField { + const PartitionField* definition; + std::string name; + std::shared_ptr type; + }; + std::map fields_by_id; + for (const auto& spec : specs) { + ICEBERG_ASSIGN_OR_RAISE(auto spec_type, spec->PartitionType(*schema)); + ICEBERG_PRECHECK(spec_type->fields().size() == spec->fields().size(), + "Partition spec {} has mismatched field and type counts", + spec->spec_id()); + for (size_t index = 0; index < spec->fields().size(); ++index) { + const auto& partition_field = spec->fields()[index]; + if (!active_field_ids.contains(partition_field.field_id())) { + continue; + } + + const auto& spec_field = spec_type->fields()[index]; + auto [iter, inserted] = + fields_by_id.try_emplace(partition_field.field_id(), + ProjectedField{.definition = &partition_field, + .name = std::string(spec_field.name()), + .type = spec_field.type()}); + if (!inserted) { + const auto& existing = *iter->second.definition; + const auto current_transform = partition_field.transform()->transform_type(); + const auto existing_transform = existing.transform()->transform_type(); + const bool compatible_transform = + *partition_field.transform() == *existing.transform() || + current_transform == TransformType::kVoid || + existing_transform == TransformType::kVoid; + ICEBERG_CHECK( + partition_field.source_id() == existing.source_id() && compatible_transform, + "Conflicting partition fields with ID {}: '{}' and '{}'", + partition_field.field_id(), partition_field.ToString(), existing.ToString()); + + if (existing_transform == TransformType::kVoid && + current_transform != TransformType::kVoid) { + iter->second.definition = &partition_field; + iter->second.type = spec_field.type(); + } + } + } + } + + std::vector fields; + fields.reserve(fields_by_id.size()); + for (auto& [_, field] : fields_by_id) { + fields.emplace_back(field.definition->field_id(), std::move(field.name), + std::move(field.type), /*optional=*/true); + } + return std::make_shared(std::move(fields)); +} + +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)); +} + +Result ProjectPartitionValues(const StructType& partition_type, + const PartitionSpec& spec, + const PartitionValues& values) { + ICEBERG_PRECHECK(values.num_fields() == spec.fields().size(), + "Partition has {} values but spec {} has {} fields", + values.num_fields(), spec.spec_id(), spec.fields().size()); + + std::unordered_map positions; + positions.reserve(spec.fields().size()); + for (size_t index = 0; index < spec.fields().size(); ++index) { + positions.emplace(spec.fields()[index].field_id(), index); + } + + std::vector projected; + projected.reserve(partition_type.fields().size()); + for (const auto& field : partition_type.fields()) { + auto target_type = checked_pointer_cast(field.type()); + auto position = positions.find(field.field_id()); + if (position == positions.end()) { + projected.push_back(Literal::Null(std::move(target_type))); + continue; + } + ICEBERG_ASSIGN_OR_RAISE(auto value, values.ValueAt(position->second)); + if (value.get().IsNull()) { + projected.push_back(Literal::Null(std::move(target_type))); + } else if (value.get().type()->type_id() == target_type->type_id()) { + projected.push_back(value.get()); + } else { + ICEBERG_ASSIGN_OR_RAISE(auto coerced, value.get().CastTo(target_type)); + projected.push_back(std::move(coerced)); + } + } + return PartitionValues(std::move(projected)); +} + +Status AppendPartitionValues(ArrowArray* array, const StructType& partition_type, + const PartitionValues& values) { + ICEBERG_PRECHECK(array != nullptr, "Partition Arrow array cannot be null"); + ICEBERG_PRECHECK( + array->n_children == static_cast(partition_type.fields().size()), + "Partition Arrow array has {} fields but schema has {}", array->n_children, + partition_type.fields().size()); + ICEBERG_PRECHECK(values.num_fields() == partition_type.fields().size(), + "Partition has {} values but schema has {} fields", + values.num_fields(), partition_type.fields().size()); + + for (size_t index = 0; index < values.num_fields(); ++index) { + ICEBERG_ASSIGN_OR_RAISE(auto value, values.ValueAt(index)); + ICEBERG_RETURN_UNEXPECTED(AppendLiteral(array->children[index], value.get())); + } + ICEBERG_NANOARROW_RETURN_UNEXPECTED(ArrowArrayFinishElement(array)); + return {}; +} + +Status AppendDataFile(ArrowRowBuilder& builder, const Schema& schema, + const Schema& table_schema, const StructType& partition_type, + 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.name() == "readable_metrics") { + 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(); +} + +Result> LoadLiveFiles(const Table& table, + const std::shared_ptr& snapshot) { + if (snapshot == nullptr) { + return std::vector{}; + } + + ICEBERG_ASSIGN_OR_RAISE(auto schema, table.schema()); + ICEBERG_ASSIGN_OR_RAISE(auto specs_ref, table.specs()); + SnapshotCache snapshot_cache(snapshot.get()); + ICEBERG_ASSIGN_OR_RAISE(auto manifests, snapshot_cache.Manifests(table.io())); + + std::vector files; + for (const auto& manifest : manifests) { + auto spec = specs_ref.get().find(manifest.partition_spec_id); + ICEBERG_CHECK(spec != specs_ref.get().end(), + "Cannot find partition spec {} for manifest '{}'", + manifest.partition_spec_id, manifest.manifest_path); + ICEBERG_PRECHECK(spec->second != nullptr, "Partition spec {} is null", + manifest.partition_spec_id); + + ICEBERG_ASSIGN_OR_RAISE( + auto reader, ManifestReader::Make(manifest, table.io(), schema, specs_ref.get())); + ICEBERG_ASSIGN_OR_RAISE(auto entries, reader->LiveEntries()); + files.reserve(files.size() + entries.size()); + for (auto& entry : entries) { + ICEBERG_PRECHECK(entry.data_file != nullptr, + "Manifest '{}' contains an entry with no data file", + manifest.manifest_path); + files.push_back(LiveFile{.file = std::move(entry.data_file), + .spec = spec->second, + .snapshot_id = entry.snapshot_id}); + } + } + return files; +} + +} // namespace iceberg::internal diff --git a/src/iceberg/inspect/metadata_table_util_internal.h b/src/iceberg/inspect/metadata_table_util_internal.h new file mode 100644 index 000000000..dea8c894a --- /dev/null +++ b/src/iceberg/inspect/metadata_table_util_internal.h @@ -0,0 +1,74 @@ +/* + * 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 "iceberg/inspect/metadata_table.h" +#include "iceberg/manifest/manifest_entry.h" +#include "iceberg/result.h" +#include "iceberg/row/partition_values.h" +#include "iceberg/type_fwd.h" + +namespace iceberg { +class ArrowRowBuilder; +} + +namespace iceberg::internal { + +struct LiveFile { + std::shared_ptr file; + std::shared_ptr spec; + std::optional snapshot_id; +}; + +/// \brief Resolve a time-travel selection to a snapshot. +Result> ResolveMetadataTableSnapshot( + const Table& table, const SnapshotSelection& selection); + +/// \brief Build the Java-compatible union of active partition fields across all specs. +Result> UnifiedPartitionType(const Table& table); + +/// \brief Build the files metadata table schema for a table. +Result> FilesTableSchema( + const Schema& table_schema, const std::shared_ptr& partition_type); + +/// \brief Project values written with one spec into the table-wide partition type. +Result ProjectPartitionValues(const StructType& partition_type, + const PartitionSpec& spec, + const PartitionValues& values); + +/// \brief Append partition values to an Arrow struct builder. +Status AppendPartitionValues(ArrowArray* array, const StructType& partition_type, + const PartitionValues& values); + +/// \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, + const LiveFile& live_file); + +/// \brief Read all live files in the selected snapshot. +Result> LoadLiveFiles(const Table& table, + const std::shared_ptr& snapshot); + +} // namespace iceberg::internal diff --git a/src/iceberg/inspect/partitions_table.cc b/src/iceberg/inspect/partitions_table.cc new file mode 100644 index 000000000..603847673 --- /dev/null +++ b/src/iceberg/inspect/partitions_table.cc @@ -0,0 +1,317 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +#include "iceberg/inspect/partitions_table.h" + +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +#include "iceberg/arrow_row_builder_internal.h" +#include "iceberg/expression/literal.h" +#include "iceberg/inspect/metadata_table_stream_internal.h" +#include "iceberg/inspect/metadata_table_util_internal.h" +#include "iceberg/manifest/manifest_entry.h" +#include "iceberg/partition_spec.h" +#include "iceberg/row/partition_values.h" +#include "iceberg/schema.h" +#include "iceberg/schema_field.h" +#include "iceberg/snapshot.h" +#include "iceberg/table.h" +#include "iceberg/type.h" +#include "iceberg/util/macros.h" + +namespace iceberg { +namespace { + +constexpr int32_t kPartitionFieldId = 1; +constexpr int32_t kRecordCountFieldId = 2; +constexpr int32_t kFileCountFieldId = 3; +constexpr int32_t kSpecIdFieldId = 4; +constexpr int32_t kPositionDeleteRecordCountFieldId = 5; +constexpr int32_t kPositionDeleteFileCountFieldId = 6; +constexpr int32_t kEqualityDeleteRecordCountFieldId = 7; +constexpr int32_t kEqualityDeleteFileCountFieldId = 8; +constexpr int32_t kLastUpdatedAtFieldId = 9; +constexpr int32_t kLastUpdatedSnapshotIdFieldId = 10; +constexpr int32_t kTotalDataFileSizeFieldId = 11; + +struct PartitionKey { + PartitionValues values; + size_t projected_fields; + + bool operator==(const PartitionKey& other) const { + if (projected_fields != other.projected_fields || + values.num_fields() != other.values.num_fields()) { + return false; + } + for (size_t index = 0; index < values.num_fields(); ++index) { + const auto& lhs = values.values()[index]; + const auto& rhs = other.values.values()[index]; + if (lhs.IsNull() || rhs.IsNull()) { + if (lhs.IsNull() != rhs.IsNull()) { + return false; + } + } else if (lhs != rhs) { + return false; + } + } + return true; + } +}; + +struct PartitionKeyHash { + size_t operator()(const PartitionKey& key) const noexcept { + size_t result = 17; + for (const auto& value : key.values.values()) { + size_t value_hash; + if (value.IsNaN()) { + const bool negative = std::holds_alternative(value.value()) + ? std::signbit(std::get(value.value())) + : std::signbit(std::get(value.value())); + value_hash = negative ? 0x9e3779b97f4a7c15ULL : 0x7ff8000000000000ULL; + } else { + value_hash = LiteralHash{}(value); + } + result = result * 37 + value_hash; + } + return result * 37 + key.projected_fields; + } +}; + +size_t ProjectedFieldCount(const StructType& partition_type, const PartitionSpec& spec) { + size_t count = 0; + for (const auto& field : partition_type.fields()) { + count += std::ranges::any_of( + spec.fields(), [field_id = field.field_id()](const PartitionField& spec_field) { + return spec_field.field_id() == field_id; + }); + } + return count; +} + +struct PartitionStats { + explicit PartitionStats(PartitionValues values) : partition(std::move(values)) {} + + PartitionValues partition; + int32_t spec_id = PartitionSpec::kInitialSpecId; + int64_t data_record_count = 0; + int32_t data_file_count = 0; + int64_t data_file_size = 0; + int64_t position_delete_record_count = 0; + int32_t position_delete_file_count = 0; + int64_t equality_delete_record_count = 0; + int32_t equality_delete_file_count = 0; + std::optional last_updated_at; + std::optional last_updated_snapshot_id; +}; + +Status AppendPartition(ArrowRowBuilder& builder, const Schema& schema, + const StructType& partition_type, + const PartitionStats& partition) { + for (size_t index = 0; index < schema.fields().size(); ++index) { + auto* array = builder.column(index); + switch (schema.fields()[index].field_id()) { + case kPartitionFieldId: + ICEBERG_RETURN_UNEXPECTED( + internal::AppendPartitionValues(array, partition_type, partition.partition)); + break; + case kSpecIdFieldId: + ICEBERG_RETURN_UNEXPECTED(AppendInt(array, partition.spec_id)); + break; + case kRecordCountFieldId: + ICEBERG_RETURN_UNEXPECTED(AppendInt(array, partition.data_record_count)); + break; + case kFileCountFieldId: + ICEBERG_RETURN_UNEXPECTED(AppendInt(array, partition.data_file_count)); + break; + case kTotalDataFileSizeFieldId: + ICEBERG_RETURN_UNEXPECTED(AppendInt(array, partition.data_file_size)); + break; + case kPositionDeleteRecordCountFieldId: + ICEBERG_RETURN_UNEXPECTED( + AppendInt(array, partition.position_delete_record_count)); + break; + case kPositionDeleteFileCountFieldId: + ICEBERG_RETURN_UNEXPECTED(AppendInt(array, partition.position_delete_file_count)); + break; + case kEqualityDeleteRecordCountFieldId: + ICEBERG_RETURN_UNEXPECTED( + AppendInt(array, partition.equality_delete_record_count)); + break; + case kEqualityDeleteFileCountFieldId: + ICEBERG_RETURN_UNEXPECTED(AppendInt(array, partition.equality_delete_file_count)); + break; + case kLastUpdatedAtFieldId: + if (partition.last_updated_at.has_value()) { + ICEBERG_RETURN_UNEXPECTED( + AppendInt(array, std::chrono::duration_cast( + partition.last_updated_at->time_since_epoch()) + .count())); + } else { + ICEBERG_RETURN_UNEXPECTED(AppendNull(array)); + } + break; + case kLastUpdatedSnapshotIdFieldId: + if (partition.last_updated_snapshot_id.has_value()) { + ICEBERG_RETURN_UNEXPECTED( + AppendInt(array, *partition.last_updated_snapshot_id)); + } else { + ICEBERG_RETURN_UNEXPECTED(AppendNull(array)); + } + break; + default: + return InvalidSchema("Unsupported partitions metadata field {}", + schema.fields()[index].field_id()); + } + } + return builder.FinishRow(); +} + +void UpdateCounts(PartitionStats& partition, const DataFile& file) { + switch (file.content) { + case DataFile::Content::kData: + partition.data_record_count += file.record_count; + ++partition.data_file_count; + partition.data_file_size += file.file_size_in_bytes; + break; + case DataFile::Content::kPositionDeletes: + partition.position_delete_record_count += file.record_count; + ++partition.position_delete_file_count; + break; + case DataFile::Content::kEqualityDeletes: + partition.equality_delete_record_count += file.record_count; + ++partition.equality_delete_file_count; + break; + } +} + +} // namespace + +PartitionsTable::PartitionsTable(std::shared_ptr
table, + std::shared_ptr schema, + std::shared_ptr partition_type) + : TimeTravelMetadataTable(std::move(table)), + schema_(std::move(schema)), + partition_type_(std::move(partition_type)) {} + +PartitionsTable::~PartitionsTable() = default; + +const std::shared_ptr& PartitionsTable::schema() const { return schema_; } + +Result> PartitionsTable::Make( + std::shared_ptr
table) { + ICEBERG_PRECHECK(table != nullptr, "Table cannot be null"); + ICEBERG_ASSIGN_OR_RAISE(auto partition_type, internal::UnifiedPartitionType(*table)); + + std::vector fields; + if (!partition_type->fields().empty()) { + fields.push_back( + SchemaField::MakeRequired(kPartitionFieldId, "partition", partition_type)); + fields.push_back(SchemaField::MakeRequired(kSpecIdFieldId, "spec_id", int32())); + } + fields.push_back( + SchemaField::MakeRequired(kRecordCountFieldId, "record_count", int64())); + fields.push_back(SchemaField::MakeRequired(kFileCountFieldId, "file_count", int32())); + fields.push_back(SchemaField::MakeRequired(kTotalDataFileSizeFieldId, + "total_data_file_size_in_bytes", int64())); + fields.push_back(SchemaField::MakeRequired(kPositionDeleteRecordCountFieldId, + "position_delete_record_count", int64())); + fields.push_back(SchemaField::MakeRequired(kPositionDeleteFileCountFieldId, + "position_delete_file_count", int32())); + fields.push_back(SchemaField::MakeRequired(kEqualityDeleteRecordCountFieldId, + "equality_delete_record_count", int64())); + fields.push_back(SchemaField::MakeRequired(kEqualityDeleteFileCountFieldId, + "equality_delete_file_count", int32())); + fields.push_back(SchemaField::MakeOptional(kLastUpdatedAtFieldId, "last_updated_at", + timestamp_tz())); + fields.push_back(SchemaField::MakeOptional(kLastUpdatedSnapshotIdFieldId, + "last_updated_snapshot_id", int64())); + + auto schema = std::make_shared(std::move(fields)); + return std::unique_ptr(new PartitionsTable( + std::move(table), std::move(schema), std::move(partition_type))); +} + +Result PartitionsTable::ScanSnapshot( + const SnapshotSelection& snapshot_selection) { + ICEBERG_ASSIGN_OR_RAISE(auto snapshot, internal::ResolveMetadataTableSnapshot( + *source_table(), snapshot_selection)); + ICEBERG_ASSIGN_OR_RAISE(auto files, internal::LoadLiveFiles(*source_table(), snapshot)); + + std::vector partitions; + std::unordered_map positions; + std::unordered_map> snapshots; + for (const auto& live_file : files) { + ICEBERG_ASSIGN_OR_RAISE(auto partition_values, internal::ProjectPartitionValues( + *partition_type_, *live_file.spec, + live_file.file->partition)); + PartitionKey key{ + .values = std::move(partition_values), + .projected_fields = ProjectedFieldCount(*partition_type_, *live_file.spec)}; + auto [position, inserted] = positions.try_emplace(key, partitions.size()); + if (inserted) { + partitions.emplace_back(std::move(key.values)); + } + auto& partition = partitions[position->second]; + UpdateCounts(partition, *live_file.file); + + if (live_file.snapshot_id.has_value()) { + auto snapshot_iter = snapshots.find(*live_file.snapshot_id); + if (snapshot_iter == snapshots.end()) { + auto file_snapshot = source_table()->SnapshotById(*live_file.snapshot_id); + if (!file_snapshot.has_value() && + file_snapshot.error().kind != ErrorKind::kNotFound) { + return std::unexpected(file_snapshot.error()); + } + snapshot_iter = + snapshots.emplace(*live_file.snapshot_id, file_snapshot.value_or(nullptr)) + .first; + } + const auto& file_snapshot = snapshot_iter->second; + if (file_snapshot != nullptr && + (!partition.last_updated_at.has_value() || + file_snapshot->timestamp_ms > *partition.last_updated_at)) { + partition.spec_id = live_file.spec->spec_id(); + partition.last_updated_at = file_snapshot->timestamp_ms; + partition.last_updated_snapshot_id = file_snapshot->snapshot_id; + } + } + } + + auto schema = schema_; + auto partition_type = partition_type_; + return internal::MakeMetadataTableStream( + *schema_, std::move(partitions), + [schema = std::move(schema), partition_type = std::move(partition_type)]( + ArrowRowBuilder& builder, const PartitionStats& partition) { + return AppendPartition(builder, *schema, *partition_type, partition); + }); +} + +} // namespace iceberg diff --git a/src/iceberg/inspect/partitions_table.h b/src/iceberg/inspect/partitions_table.h new file mode 100644 index 000000000..4c10b390a --- /dev/null +++ b/src/iceberg/inspect/partitions_table.h @@ -0,0 +1,57 @@ +/* + * 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/partitions_table.h +/// \brief Define the partitions metadata table. + +#include + +#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 aggregate file statistics by partition. +class ICEBERG_EXPORT PartitionsTable : public TimeTravelMetadataTable { + public: + static Result> Make(std::shared_ptr
table); + + ~PartitionsTable() override; + + Kind kind() const noexcept override { return Kind::kPartitions; } + + const std::shared_ptr& schema() const override; + + protected: + Result ScanSnapshot( + const SnapshotSelection& snapshot_selection) override; + + private: + PartitionsTable(std::shared_ptr
table, std::shared_ptr schema, + std::shared_ptr partition_type); + + std::shared_ptr schema_; + std::shared_ptr partition_type_; +}; + +} // namespace iceberg diff --git a/src/iceberg/inspect/snapshots_table.cc b/src/iceberg/inspect/snapshots_table.cc index 4b0c3ce9f..f0a37843b 100644 --- a/src/iceberg/inspect/snapshots_table.cc +++ b/src/iceberg/inspect/snapshots_table.cc @@ -19,21 +19,133 @@ #include "iceberg/inspect/snapshots_table.h" +#include +#include #include +#include #include #include +#include + +#include "iceberg/arrow/nanoarrow_status_internal.h" +#include "iceberg/arrow_c_data_util_internal.h" +#include "iceberg/arrow_row_builder_internal.h" #include "iceberg/schema.h" #include "iceberg/schema_field.h" +#include "iceberg/schema_internal.h" +#include "iceberg/snapshot.h" #include "iceberg/table.h" -#include "iceberg/table_identifier.h" #include "iceberg/type.h" +#include "iceberg/util/macros.h" namespace iceberg { namespace { -std::shared_ptr MakeSnapshotsTableSchema() { - return std::make_shared(std::vector{ +Status AppendSnapshot(ArrowRowBuilder& builder, const Snapshot& snapshot) { + ICEBERG_RETURN_UNEXPECTED( + AppendInt(builder.column(0), std::chrono::duration_cast( + snapshot.timestamp_ms.time_since_epoch()) + .count())); + ICEBERG_RETURN_UNEXPECTED(AppendInt(builder.column(1), snapshot.snapshot_id)); + + if (snapshot.parent_snapshot_id.has_value()) { + ICEBERG_RETURN_UNEXPECTED(AppendInt(builder.column(2), *snapshot.parent_snapshot_id)); + } else { + ICEBERG_RETURN_UNEXPECTED(AppendNull(builder.column(2))); + } + + auto operation = snapshot.Operation(); + if (operation.has_value()) { + ICEBERG_RETURN_UNEXPECTED(AppendString(builder.column(3), *operation)); + } else { + ICEBERG_RETURN_UNEXPECTED(AppendNull(builder.column(3))); + } + + ICEBERG_RETURN_UNEXPECTED(AppendString(builder.column(4), snapshot.manifest_list)); + + auto summary = snapshot.summary; + summary.erase(SnapshotSummaryFields::kOperation); + if (summary.empty()) { + ICEBERG_RETURN_UNEXPECTED(AppendNull(builder.column(5))); + } else { + ICEBERG_RETURN_UNEXPECTED(AppendStringMap(builder.column(5), summary)); + } + + return builder.FinishRow(); +} + +class SnapshotsTableStream { + public: + static Result> Make( + std::shared_ptr
table, const iceberg::Schema& schema) { + ArrowSchema arrow_schema{}; + ICEBERG_RETURN_UNEXPECTED(ToArrowSchema(schema, &arrow_schema)); + return std::unique_ptr( + new SnapshotsTableStream(std::move(table), std::move(arrow_schema))); + } + + ~SnapshotsTableStream() { std::ignore = Close(); } + + Status Close() { + table_.reset(); + if (arrow_schema_.release != nullptr) { + ArrowSchemaRelease(&arrow_schema_); + } + return {}; + } + + Result> Next() { + const auto& snapshots = table_->snapshots(); + if (next_snapshot_ == snapshots.size()) { + return std::nullopt; + } + + ICEBERG_ASSIGN_OR_RAISE(auto builder, ArrowRowBuilder::Make(&arrow_schema_)); + while (next_snapshot_ < snapshots.size() && + builder.num_rows() < MetadataTable::kBatchSize) { + const auto& snapshot = snapshots[next_snapshot_++]; + if (snapshot == nullptr) [[unlikely]] { + continue; + } + ICEBERG_RETURN_UNEXPECTED(AppendSnapshot(builder, *snapshot)); + } + if (builder.num_rows() == 0) { + return std::nullopt; + } + + ICEBERG_ASSIGN_OR_RAISE(auto array, std::move(builder).Finish()); + return array; + } + + Result Schema() { + if (arrow_schema_.release == nullptr) [[unlikely]] { + return InvalidArgument("Cannot read schema from a closed snapshots table stream"); + } + ArrowSchema schema_copy{}; + ICEBERG_NANOARROW_RETURN_UNEXPECTED( + ArrowSchemaDeepCopy(&arrow_schema_, &schema_copy)); + return schema_copy; + } + + private: + SnapshotsTableStream(std::shared_ptr
table, ArrowSchema arrow_schema) + : table_(std::move(table)), arrow_schema_(std::move(arrow_schema)) {} + + std::shared_ptr
table_; + ArrowSchema arrow_schema_{}; + size_t next_snapshot_ = 0; +}; + +} // namespace + +SnapshotsTable::SnapshotsTable(std::shared_ptr
table) + : MetadataTable(std::move(table)) {} + +SnapshotsTable::~SnapshotsTable() = default; + +const std::shared_ptr& SnapshotsTable::schema() const { + static const auto schema = std::make_shared(std::vector{ SchemaField::MakeRequired(1, "committed_at", timestamp_tz()), SchemaField::MakeRequired(2, "snapshot_id", int64()), SchemaField::MakeOptional(3, "parent_id", int64()), @@ -43,26 +155,19 @@ std::shared_ptr MakeSnapshotsTableSchema() { std::make_shared( SchemaField::MakeRequired(7, "key", string()), SchemaField::MakeRequired(8, "value", string())))}); + return schema; } -TableIdentifier MakeSnapshotsTableName(const TableIdentifier& source_name) { - return TableIdentifier{.ns = source_name.ns, .name = source_name.name + ".snapshots"}; -} - -} // namespace - -SnapshotsTable::SnapshotsTable(std::shared_ptr
table) - : MetadataTable(table, MakeSnapshotsTableName(table->name()), - MakeSnapshotsTableSchema()) {} - -SnapshotsTable::~SnapshotsTable() = default; - Result> SnapshotsTable::Make( std::shared_ptr
table) { - if (table == nullptr) [[unlikely]] { - return InvalidArgument("Table cannot be null"); - } + ICEBERG_PRECHECK(table != nullptr, "Table cannot be null"); return std::unique_ptr(new SnapshotsTable(std::move(table))); } +Result SnapshotsTable::Scan() { + ICEBERG_ASSIGN_OR_RAISE(auto stream, + SnapshotsTableStream::Make(source_table(), *schema())); + return MakeArrowArrayStream(std::move(stream)); +} + } // namespace iceberg diff --git a/src/iceberg/inspect/snapshots_table.h b/src/iceberg/inspect/snapshots_table.h index 9af1bcacb..d2f0ddf90 100644 --- a/src/iceberg/inspect/snapshots_table.h +++ b/src/iceberg/inspect/snapshots_table.h @@ -40,6 +40,13 @@ class ICEBERG_EXPORT SnapshotsTable : public MetadataTable { Kind kind() const noexcept override { return Kind::kSnapshots; } + const std::shared_ptr& schema() const override; + + /// \brief Scan all snapshots as rows. + /// + /// The snapshots table always returns every known snapshot. + Result Scan() override; + private: explicit SnapshotsTable(std::shared_ptr
table); }; diff --git a/src/iceberg/inspect/tags_table.cc b/src/iceberg/inspect/tags_table.cc new file mode 100644 index 000000000..940a8a538 --- /dev/null +++ b/src/iceberg/inspect/tags_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/tags_table.h" + +#include +#include +#include +#include +#include +#include + +#include "iceberg/arrow_row_builder_internal.h" +#include "iceberg/inspect/metadata_table_stream_internal.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/type.h" +#include "iceberg/util/macros.h" + +namespace iceberg { +namespace { + +struct TagRow { + std::string name; + int64_t snapshot_id; + std::optional max_ref_age_ms; +}; + +Status AppendTag(ArrowRowBuilder& builder, const TagRow& tag) { + ICEBERG_RETURN_UNEXPECTED(AppendString(builder.column(0), tag.name)); + ICEBERG_RETURN_UNEXPECTED(AppendInt(builder.column(1), tag.snapshot_id)); + if (tag.max_ref_age_ms.has_value()) { + ICEBERG_RETURN_UNEXPECTED(AppendInt(builder.column(2), *tag.max_ref_age_ms)); + } else { + ICEBERG_RETURN_UNEXPECTED(AppendNull(builder.column(2))); + } + return builder.FinishRow(); +} + +} // namespace + +TagsTable::TagsTable(std::shared_ptr
table) : MetadataTable(std::move(table)) {} + +TagsTable::~TagsTable() = default; + +const std::shared_ptr& TagsTable::schema() const { + static const auto schema = std::make_shared(std::vector{ + SchemaField::MakeRequired(1, "name", string()), + SchemaField::MakeRequired(2, "snapshot_id", int64()), + SchemaField::MakeOptional(3, "max_reference_age_in_ms", int64()), + }); + return schema; +} + +Result> TagsTable::Make(std::shared_ptr
table) { + ICEBERG_PRECHECK(table != nullptr, "Table cannot be null"); + return std::unique_ptr(new TagsTable(std::move(table))); +} + +Result TagsTable::Scan() { + std::vector rows; + for (const auto& [name, ref] : source_table()->metadata()->refs) { + if (ref == nullptr || ref->type() != SnapshotRefType::kTag) { + continue; + } + const auto& retention = std::get(ref->retention); + rows.push_back(TagRow{.name = name, + .snapshot_id = ref->snapshot_id, + .max_ref_age_ms = retention.max_ref_age_ms}); + } + std::ranges::sort(rows, {}, &TagRow::name); + return internal::MakeMetadataTableStream(*schema(), std::move(rows), AppendTag); +} + +} // namespace iceberg diff --git a/src/iceberg/inspect/tags_table.h b/src/iceberg/inspect/tags_table.h new file mode 100644 index 000000000..f8edc7c84 --- /dev/null +++ b/src/iceberg/inspect/tags_table.h @@ -0,0 +1,51 @@ +/* + * 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/tags_table.h +/// \brief Define the tags metadata table. + +#include + +#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 the table's tag snapshot references. +class ICEBERG_EXPORT TagsTable : public MetadataTable { + public: + static Result> Make(std::shared_ptr
table); + + ~TagsTable() override; + + Kind kind() const noexcept override { return Kind::kTags; } + + const std::shared_ptr& schema() const override; + + Result Scan() override; + + private: + explicit TagsTable(std::shared_ptr
table); +}; + +} // namespace iceberg diff --git a/src/iceberg/meson.build b/src/iceberg/meson.build index 1de293cb5..9766afcf3 100644 --- a/src/iceberg/meson.build +++ b/src/iceberg/meson.build @@ -94,9 +94,15 @@ iceberg_sources = files( 'file_reader.cc', 'file_writer.cc', 'inheritable_metadata.cc', + 'inspect/branches_table.cc', + 'inspect/files_table.cc', 'inspect/history_table.cc', + 'inspect/manifests_table.cc', 'inspect/metadata_table.cc', + 'inspect/metadata_table_util_internal.cc', + 'inspect/partitions_table.cc', 'inspect/snapshots_table.cc', + 'inspect/tags_table.cc', 'json_serde.cc', 'location_provider.cc', 'logging/cerr_logger.cc', diff --git a/src/iceberg/test/CMakeLists.txt b/src/iceberg/test/CMakeLists.txt index 98129a6d2..2358bf0fe 100644 --- a/src/iceberg/test/CMakeLists.txt +++ b/src/iceberg/test/CMakeLists.txt @@ -186,7 +186,13 @@ if(ICEBERG_BUILD_BUNDLE) add_iceberg_test(catalog_test USE_BUNDLE SOURCES in_memory_catalog_test.cc) - add_iceberg_test(metadata_table_test USE_BUNDLE SOURCES metadata_table_test.cc) + add_iceberg_test(metadata_table_test + USE_BUNDLE + SOURCES + history_table_test.cc + metadata_table_test.cc + snapshots_table_test.cc + system_metadata_tables_test.cc) add_iceberg_test(eval_expr_test USE_BUNDLE diff --git a/src/iceberg/test/arrow_row_builder_test.cc b/src/iceberg/test/arrow_row_builder_test.cc index 45fb3b787..d37fe3458 100644 --- a/src/iceberg/test/arrow_row_builder_test.cc +++ b/src/iceberg/test/arrow_row_builder_test.cc @@ -73,6 +73,7 @@ TEST(ArrowRowBuilderTest, BuildsRowsWithTypedValues) { ICEBERG_UNWRAP_OR_FAIL(auto builder, ArrowRowBuilder::Make(*schema)); ASSERT_EQ(builder.num_columns(), 5); + ASSERT_EQ(builder.num_rows(), 0); // Row 0 ASSERT_THAT(AppendInt(builder.column(0), 1), IsOk()); @@ -81,6 +82,7 @@ TEST(ArrowRowBuilderTest, BuildsRowsWithTypedValues) { ASSERT_THAT(AppendBoolean(builder.column(3), true), IsOk()); ASSERT_THAT(AppendStringMap(builder.column(4), {{"k", "v"}}), IsOk()); ASSERT_THAT(builder.FinishRow(), IsOk()); + ASSERT_EQ(builder.num_rows(), 1); // Row 1 ASSERT_THAT(AppendInt(builder.column(0), 2), IsOk()); @@ -89,6 +91,7 @@ TEST(ArrowRowBuilderTest, BuildsRowsWithTypedValues) { ASSERT_THAT(AppendBoolean(builder.column(3), false), IsOk()); ASSERT_THAT(AppendStringMap(builder.column(4), {}), IsOk()); ASSERT_THAT(builder.FinishRow(), IsOk()); + ASSERT_EQ(builder.num_rows(), 2); auto batch = FinishAndImport(std::move(builder), *schema); ASSERT_EQ(batch->num_rows(), 2); diff --git a/src/iceberg/test/history_table_test.cc b/src/iceberg/test/history_table_test.cc new file mode 100644 index 000000000..8da311c02 --- /dev/null +++ b/src/iceberg/test/history_table_test.cc @@ -0,0 +1,55 @@ +/* + * 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. + */ + +/// \file history_table_test.cc +/// Unit tests for HistoryTable. + +#include "iceberg/inspect/history_table.h" + +#include +#include + +#include "iceberg/inspect/metadata_table.h" +#include "iceberg/schema.h" +#include "iceberg/schema_field.h" +#include "iceberg/test/matchers.h" +#include "iceberg/test/metadata_table_test_base.h" +#include "iceberg/type.h" + +namespace iceberg { +namespace { + +std::shared_ptr MakeHistorySchema() { + return std::make_shared(std::vector{ + SchemaField::MakeRequired(1, "made_current_at", timestamp_tz()), + SchemaField::MakeRequired(2, "snapshot_id", int64()), + SchemaField::MakeOptional(3, "parent_id", int64()), + SchemaField::MakeRequired(4, "is_current_ancestor", boolean())}); +} + +} // namespace + +class HistoryTableTest : public MetadataTableTestBase {}; + +TEST_F(HistoryTableTest, SchemaMatchesIcebergSchema) { + ICEBERG_UNWRAP_OR_FAIL(auto history_table, MetadataTable::Make(table_)); + EXPECT_TRUE(*history_table->schema() == *MakeHistorySchema()); +} + +} // namespace iceberg diff --git a/src/iceberg/test/meson.build b/src/iceberg/test/meson.build index 0e5399347..8ef30f7b4 100644 --- a/src/iceberg/test/meson.build +++ b/src/iceberg/test/meson.build @@ -54,6 +54,7 @@ iceberg_tests = { 'metrics_test.cc', 'snapshot_test.cc', 'snapshot_util_test.cc', + 'system_metadata_tables_test.cc', 'table_metadata_builder_test.cc', 'table_requirement_test.cc', 'table_requirements_test.cc', diff --git a/src/iceberg/test/metadata_table_test.cc b/src/iceberg/test/metadata_table_test.cc index 1e0a664c3..b014ef962 100644 --- a/src/iceberg/test/metadata_table_test.cc +++ b/src/iceberg/test/metadata_table_test.cc @@ -22,6 +22,9 @@ #include #include +#include "iceberg/constants.h" +#include "iceberg/inspect/history_table.h" +#include "iceberg/inspect/snapshots_table.h" #include "iceberg/schema.h" #include "iceberg/schema_field.h" #include "iceberg/table.h" @@ -33,88 +36,42 @@ #include "iceberg/type.h" namespace iceberg { -namespace { - -std::shared_ptr MakeSnapshotsSchema() { - return std::make_shared(std::vector{ - SchemaField::MakeRequired(1, "committed_at", timestamp_tz()), - SchemaField::MakeRequired(2, "snapshot_id", int64()), - SchemaField::MakeOptional(3, "parent_id", int64()), - SchemaField::MakeOptional(4, "operation", string()), - SchemaField::MakeOptional(5, "manifest_list", string()), - SchemaField::MakeOptional( - 6, "summary", - std::make_shared(SchemaField::MakeRequired(7, "key", string()), - SchemaField::MakeRequired(8, "value", string())))}); -} - -std::shared_ptr MakeHistorySchema() { - return std::make_shared(std::vector{ - SchemaField::MakeRequired(1, "made_current_at", timestamp_tz()), - SchemaField::MakeRequired(2, "snapshot_id", int64()), - SchemaField::MakeOptional(3, "parent_id", int64()), - SchemaField::MakeRequired(4, "is_current_ancestor", boolean())}); -} - -} // namespace class MetadataTableTest : public ::testing::Test { protected: void SetUp() override { - io_ = std::make_shared(); - catalog_ = std::make_shared(); - auto schema = std::make_shared( std::vector{SchemaField::MakeRequired(1, "id", int64()), SchemaField::MakeOptional(2, "name", string())}, 1); - metadata_ = std::make_shared( - TableMetadata{.format_version = 2, .schemas = {schema}, .current_schema_id = 1}); - - TableIdentifier source_ident{.ns = Namespace{.levels = {"db"}}, - .name = "source_table"}; - auto source_table_result = - Table::Make(source_ident, metadata_, "s3://bucket/meta.json", io_, catalog_); - EXPECT_THAT(source_table_result, IsOk()); - source_table_ = *source_table_result; - - auto snapshots_table_result = - MetadataTable::Make(source_table_, MetadataTable::Kind::kSnapshots); - EXPECT_THAT(snapshots_table_result, IsOk()); - snapshots_table_ = std::move(*snapshots_table_result); + auto metadata = std::make_shared( + TableMetadata{.format_version = 2, + .schemas = {schema}, + .current_schema_id = 1, + .current_snapshot_id = kInvalidSnapshotId}); + + TableIdentifier ident{.ns = Namespace{.levels = {"db"}}, .name = "source_table"}; + ICEBERG_UNWRAP_OR_FAIL(table_, Table::Make(ident, metadata, "s3://bucket/meta.json", + std::make_shared(), + std::make_shared())); } - std::shared_ptr io_; - std::shared_ptr catalog_; - std::shared_ptr metadata_; - std::shared_ptr
source_table_; - std::unique_ptr snapshots_table_; + std::shared_ptr
table_; }; -TEST_F(MetadataTableTest, Constructor) { - EXPECT_EQ(snapshots_table_->kind(), MetadataTable::Kind::kSnapshots); - EXPECT_EQ(snapshots_table_->source_table(), source_table_); - EXPECT_EQ(snapshots_table_->name().name, "source_table.snapshots"); - EXPECT_EQ(snapshots_table_->name().ns.levels, (std::vector{"db"})); - EXPECT_NE(snapshots_table_->schema(), nullptr); -} - -TEST_F(MetadataTableTest, SnapshotsSchemaMatchesIcebergSchema) { - EXPECT_TRUE(*snapshots_table_->schema() == *MakeSnapshotsSchema()); -} - -TEST_F(MetadataTableTest, HistorySchemaMatchesIcebergSchema) { - auto history_table_result = - MetadataTable::Make(source_table_, MetadataTable::Kind::kHistory); - ASSERT_THAT(history_table_result, IsOk()); - - EXPECT_TRUE(*(*history_table_result)->schema() == *MakeHistorySchema()); -} - TEST_F(MetadataTableTest, FactoryRejectsNullSourceTable) { - auto result = MetadataTable::Make(nullptr, MetadataTable::Kind::kSnapshots); + auto result = MetadataTable::Make(nullptr); EXPECT_THAT(result, IsError(ErrorKind::kInvalidArgument)); EXPECT_THAT(result, HasErrorMessage("Table cannot be null")); } +TEST_F(MetadataTableTest, SupportsTimeTravel) { + ICEBERG_UNWRAP_OR_FAIL(auto snapshots_table, + MetadataTable::Make(table_)); + EXPECT_FALSE(snapshots_table->supports_time_travel()); + + ICEBERG_UNWRAP_OR_FAIL(auto history_table, MetadataTable::Make(table_)); + EXPECT_FALSE(history_table->supports_time_travel()); +} + } // namespace iceberg diff --git a/src/iceberg/test/metadata_table_test_base.h b/src/iceberg/test/metadata_table_test_base.h new file mode 100644 index 000000000..d15163ba8 --- /dev/null +++ b/src/iceberg/test/metadata_table_test_base.h @@ -0,0 +1,169 @@ +/* + * 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. + */ + +/// \file metadata_table_test_base.h +/// Shared test base for all metadata table tests. +/// +/// Provides common helpers (ReadAllBatches, MakeTestSnapshots, +/// MakeTableWithSnapshots) and the MockFileIO + MockCatalog fixture that +/// every metadata table test needs. + +#pragma once + +#include +#include +#include +#include + +#include +#include +#include +#include +#include + +#include "iceberg/constants.h" +#include "iceberg/schema.h" +#include "iceberg/schema_field.h" +#include "iceberg/schema_internal.h" +#include "iceberg/snapshot.h" +#include "iceberg/table.h" +#include "iceberg/table_identifier.h" +#include "iceberg/table_metadata.h" +#include "iceberg/test/matchers.h" +#include "iceberg/test/mock_catalog.h" +#include "iceberg/test/mock_io.h" +#include "iceberg/type.h" +#include "iceberg/util/timepoint.h" + +namespace iceberg { + +/// \brief Base class for all metadata table tests. +/// +/// Provides MockFileIO and MockCatalog instances plus helpers shared across +/// metadata table tests (SnapshotsTable, HistoryTable, RefsTable, ...). +class MetadataTableTestBase : public ::testing::Test { + protected: + void SetUp() override { + io_ = std::make_shared(); + catalog_ = std::make_shared(); + + auto schema = std::make_shared( + std::vector{SchemaField::MakeRequired(1, "id", int64()), + SchemaField::MakeOptional(2, "name", string())}, + 1); + metadata_ = std::make_shared( + TableMetadata{.format_version = 2, + .schemas = {schema}, + .current_schema_id = 1, + .current_snapshot_id = kInvalidSnapshotId}); + + TableIdentifier source_ident{.ns = Namespace{.levels = {"db"}}, + .name = "source_table"}; + ICEBERG_UNWRAP_OR_FAIL(table_, Table::Make(source_ident, metadata_, + "s3://bucket/meta.json", io_, catalog_)); + } + + /// \brief Import and consume a Scan()-produced ArrowArrayStream. + static Result>> ReadAllBatches( + ArrowArrayStream&& stream) { + auto reader_result = ::arrow::ImportRecordBatchReader(&stream); + if (!reader_result.ok()) { + return InvalidArrowData(reader_result.status().ToString()); + } + + auto batches_result = reader_result.ValueUnsafe()->ToRecordBatches(); + if (!batches_result.ok()) { + return InvalidArrowData(batches_result.status().ToString()); + } + return std::move(batches_result).MoveValueUnsafe(); + } + + /// \brief Create two snapshots matching the Java TestDataTaskParser test data. + /// + /// Snapshot 1: id=1, no parent, timestamp=1234567890000, operation="append" + /// Snapshot 2: id=2, parent=1, timestamp=9876543210000, operation="append" + static std::pair, std::shared_ptr> + MakeTestSnapshots() { + std::unordered_map summary1{ + {"added-data-files", "1"}, {"added-records", "1"}, + {"added-files-size", "10"}, {"changed-partition-count", "1"}, + {"total-records", "1"}, {"total-files-size", "10"}, + {"total-data-files", "1"}, {"total-delete-files", "0"}, + {"total-position-deletes", "0"}, {"total-equality-deletes", "0"}, + {"operation", "append"}, + }; + + std::unordered_map summary2{ + {"added-data-files", "1"}, {"added-records", "1"}, + {"added-files-size", "10"}, {"changed-partition-count", "1"}, + {"total-records", "2"}, {"total-files-size", "20"}, + {"total-data-files", "2"}, {"total-delete-files", "0"}, + {"total-position-deletes", "0"}, {"total-equality-deletes", "0"}, + {"operation", "append"}, + }; + + auto snap1 = std::make_shared(Snapshot{ + .snapshot_id = 1, + .parent_snapshot_id = std::nullopt, + .sequence_number = 1, + .timestamp_ms = TimePointMsFromUnixMs(1234567890000), + .manifest_list = "file:/tmp/manifest1.avro", + .summary = std::move(summary1), + .schema_id = 1, + }); + + auto snap2 = std::make_shared(Snapshot{ + .snapshot_id = 2, + .parent_snapshot_id = 1, + .sequence_number = 2, + .timestamp_ms = TimePointMsFromUnixMs(9876543210000), + .manifest_list = "file:/tmp/manifest2.avro", + .summary = std::move(summary2), + .schema_id = 1, + }); + + return {snap1, snap2}; + } + + /// \brief Create a Table with the given snapshots. + Result> MakeTableWithSnapshots( + std::vector> snapshots, int64_t current_snapshot_id) { + auto schema = std::make_shared( + std::vector{SchemaField::MakeRequired(1, "id", int64()), + SchemaField::MakeOptional(2, "name", string())}, + 1); + auto metadata = std::make_shared(TableMetadata{ + .format_version = 2, + .schemas = {schema}, + .current_schema_id = 1, + .current_snapshot_id = current_snapshot_id, + .snapshots = std::move(snapshots), + }); + + TableIdentifier ident{.ns = Namespace{.levels = {"db"}}, .name = "test_table"}; + return Table::Make(ident, metadata, "s3://bucket/meta.json", io_, catalog_); + } + + std::shared_ptr io_; + std::shared_ptr catalog_; + std::shared_ptr metadata_; + std::shared_ptr
table_; +}; + +} // namespace iceberg diff --git a/src/iceberg/test/snapshots_table_test.cc b/src/iceberg/test/snapshots_table_test.cc new file mode 100644 index 000000000..5cda84052 --- /dev/null +++ b/src/iceberg/test/snapshots_table_test.cc @@ -0,0 +1,197 @@ +/* + * 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/snapshots_table.h" + +#include +#include +#include +#include +#include + +#include +#include +#include +#include +#include + +#include "iceberg/constants.h" +#include "iceberg/inspect/metadata_table.h" +#include "iceberg/test/matchers.h" +#include "iceberg/test/metadata_table_test_base.h" + +namespace iceberg { +namespace { + +std::vector> GetMapEntries( + const std::shared_ptr<::arrow::MapArray>& map_array, int64_t row) { + auto keys = std::static_pointer_cast<::arrow::StringArray>(map_array->keys()); + auto values = std::static_pointer_cast<::arrow::StringArray>(map_array->items()); + std::vector> entries; + entries.reserve(map_array->value_length(row)); + const auto offset = map_array->value_offset(row); + for (int64_t index = offset; index < offset + map_array->value_length(row); ++index) { + entries.emplace_back(keys->GetString(index), values->GetString(index)); + } + return entries; +} + +} // namespace + +class SnapshotsTableTest : public MetadataTableTestBase { + protected: + void SetUp() override { + MetadataTableTestBase::SetUp(); + + auto [snap1, snap2] = MakeTestSnapshots(); + ICEBERG_UNWRAP_OR_FAIL( + table_, MakeTableWithSnapshots({snap1, snap2}, /*current_snapshot_id=*/2)); + + ICEBERG_UNWRAP_OR_FAIL(snapshots_table_, MetadataTable::Make(table_)); + } + + std::unique_ptr snapshots_table_; +}; + +TEST_F(SnapshotsTableTest, Construct) { + EXPECT_EQ(snapshots_table_->kind(), MetadataTable::Kind::kSnapshots); + EXPECT_EQ(snapshots_table_->source_table(), table_); + EXPECT_NE(snapshots_table_->schema(), nullptr); +} + +TEST_F(SnapshotsTableTest, Scan) { + // Scan the snapshots table once and verify all columns of the result. + ICEBERG_UNWRAP_OR_FAIL(auto stream, snapshots_table_->Scan()); + ICEBERG_UNWRAP_OR_FAIL(auto batches, ReadAllBatches(std::move(stream))); + ASSERT_EQ(batches.size(), 1); + const auto& batch = batches.front(); + + // Row and column counts. + EXPECT_EQ(batch->num_rows(), 2); + EXPECT_EQ(batch->num_columns(), 6); + + // Column 0: committed_at (timestamptz) — microseconds since epoch. + auto committed_at = std::static_pointer_cast<::arrow::TimestampArray>(batch->column(0)); + EXPECT_EQ(committed_at->Value(0), 1234567890000 * 1000); + EXPECT_EQ(committed_at->Value(1), 9876543210000 * 1000); + + // Column 1: snapshot_id (long) — returned in storage order. + auto snapshot_ids = std::static_pointer_cast<::arrow::Int64Array>(batch->column(1)); + EXPECT_EQ(snapshot_ids->Value(0), 1); + EXPECT_EQ(snapshot_ids->Value(1), 2); + + // Column 2: parent_id (long) — first snapshot has no parent. + auto parent_ids = std::static_pointer_cast<::arrow::Int64Array>(batch->column(2)); + EXPECT_TRUE(parent_ids->IsNull(0)); + EXPECT_FALSE(parent_ids->IsNull(1)); + EXPECT_EQ(parent_ids->Value(1), 1); + + // Column 3: operation (string). + auto operations = std::static_pointer_cast<::arrow::StringArray>(batch->column(3)); + EXPECT_EQ(operations->GetString(0), "append"); + EXPECT_EQ(operations->GetString(1), "append"); + + // Column 4: manifest_list (string). + auto manifest_lists = std::static_pointer_cast<::arrow::StringArray>(batch->column(4)); + EXPECT_EQ(manifest_lists->GetString(0), "file:/tmp/manifest1.avro"); + EXPECT_EQ(manifest_lists->GetString(1), "file:/tmp/manifest2.avro"); + + // Column 5: summary (map) excludes the separate operation field. + auto summaries = std::static_pointer_cast<::arrow::MapArray>(batch->column(5)); + EXPECT_FALSE(summaries->IsNull(0)); + EXPECT_FALSE(summaries->IsNull(1)); + EXPECT_EQ(summaries->value_length(0), 10); + EXPECT_EQ(summaries->value_length(1), 10); + + auto first_summary = GetMapEntries(summaries, 0); + EXPECT_THAT( + first_summary, + ::testing::Not(::testing::Contains(::testing::Pair("operation", "append")))); + EXPECT_THAT(first_summary, ::testing::Contains(::testing::Pair("total-records", "1"))); + + auto second_summary = GetMapEntries(summaries, 1); + EXPECT_THAT( + second_summary, + ::testing::Not(::testing::Contains(::testing::Pair("operation", "append")))); + EXPECT_THAT(second_summary, ::testing::Contains(::testing::Pair("total-records", "2"))); +} + +TEST_F(SnapshotsTableTest, ScanEmptySnapshotList) { + // A table with zero snapshots should return zero rows. + ICEBERG_UNWRAP_OR_FAIL( + auto empty_table, + MakeTableWithSnapshots({}, /*current_snapshot_id=*/kInvalidSnapshotId)); + + ICEBERG_UNWRAP_OR_FAIL(snapshots_table_, + MetadataTable::Make(empty_table)); + + ICEBERG_UNWRAP_OR_FAIL(auto stream, snapshots_table_->Scan()); + ICEBERG_UNWRAP_OR_FAIL(auto batches, ReadAllBatches(std::move(stream))); + EXPECT_TRUE(batches.empty()); +} + +TEST_F(SnapshotsTableTest, ScanSkipsNullSnapshots) { + auto [snap1, snap2] = MakeTestSnapshots(); + ICEBERG_UNWRAP_OR_FAIL(auto table, MakeTableWithSnapshots({snap1, nullptr, snap2}, + /*current_snapshot_id=*/2)); + ICEBERG_UNWRAP_OR_FAIL(auto snapshots_table, + MetadataTable::Make(table)); + + ICEBERG_UNWRAP_OR_FAIL(auto stream, snapshots_table->Scan()); + ICEBERG_UNWRAP_OR_FAIL(auto batches, ReadAllBatches(std::move(stream))); + ASSERT_EQ(batches.size(), 1); + EXPECT_EQ(batches.front()->num_rows(), 2); +} + +TEST_F(SnapshotsTableTest, ScanTreatsEmptySummaryAsNull) { + auto [missing_summary, operation_only_summary] = MakeTestSnapshots(); + missing_summary->summary.clear(); + operation_only_summary->summary = { + {SnapshotSummaryFields::kOperation, DataOperation::kAppend}}; + ICEBERG_UNWRAP_OR_FAIL(auto table, + MakeTableWithSnapshots({missing_summary, operation_only_summary}, + /*current_snapshot_id=*/2)); + ICEBERG_UNWRAP_OR_FAIL(auto snapshots_table, + MetadataTable::Make(table)); + + ICEBERG_UNWRAP_OR_FAIL(auto stream, snapshots_table->Scan()); + ICEBERG_UNWRAP_OR_FAIL(auto batches, ReadAllBatches(std::move(stream))); + ASSERT_EQ(batches.size(), 1); + auto summaries = + std::static_pointer_cast<::arrow::MapArray>(batches.front()->column(5)); + EXPECT_TRUE(summaries->IsNull(0)); + EXPECT_TRUE(summaries->IsNull(1)); +} + +TEST_F(SnapshotsTableTest, ScanReturnsMultipleBatches) { + auto snapshot = MakeTestSnapshots().first; + std::vector> snapshots(1025, snapshot); + ICEBERG_UNWRAP_OR_FAIL(auto table, MakeTableWithSnapshots(std::move(snapshots), + /*current_snapshot_id=*/1)); + ICEBERG_UNWRAP_OR_FAIL(auto snapshots_table, + MetadataTable::Make(table)); + + ICEBERG_UNWRAP_OR_FAIL(auto stream, snapshots_table->Scan()); + ICEBERG_UNWRAP_OR_FAIL(auto batches, ReadAllBatches(std::move(stream))); + ASSERT_EQ(batches.size(), 2); + EXPECT_EQ(batches[0]->num_rows(), 1024); + EXPECT_EQ(batches[1]->num_rows(), 1); +} + +} // namespace iceberg diff --git a/src/iceberg/test/system_metadata_tables_test.cc b/src/iceberg/test/system_metadata_tables_test.cc new file mode 100644 index 000000000..bd0ed67a4 --- /dev/null +++ b/src/iceberg/test/system_metadata_tables_test.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 +#include +#include +#include +#include +#include +#include + +#include +#include +#include +#include + +#include "iceberg/constants.h" +#include "iceberg/inspect/branches_table.h" +#include "iceberg/inspect/files_table.h" +#include "iceberg/inspect/manifests_table.h" +#include "iceberg/inspect/metadata_table.h" +#include "iceberg/inspect/partitions_table.h" +#include "iceberg/inspect/tags_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/matchers.h" +#include "iceberg/test/mock_catalog.h" +#include "iceberg/test/scan_test_base.h" +#include "iceberg/transform.h" +#include "iceberg/util/macros.h" + +namespace iceberg { +namespace { + +class SystemMetadataTablesTest : public ScanTestBase { + protected: + 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(); + } + + std::shared_ptr catalog_; +}; + +TEST_P(SystemMetadataTablesTest, ScansBranchesAndTagsSeparately) { + ICEBERG_UNWRAP_OR_FAIL(auto main_ref, SnapshotRef::MakeBranch(2)); + ICEBERG_UNWRAP_OR_FAIL(auto dev_ref, SnapshotRef::MakeBranch(1, 3, 2000, 1000)); + ICEBERG_UNWRAP_OR_FAIL(auto release_ref, SnapshotRef::MakeTag(1, 5000)); + std::unordered_map> refs; + refs.emplace("main", std::move(main_ref)); + refs.emplace("dev", std::move(dev_ref)); + refs.emplace("release", std::move(release_ref)); + + auto first = std::make_shared(Snapshot{ + .snapshot_id = 1, + .sequence_number = 1, + .timestamp_ms = TimePointMsFromUnixMs(1000), + .manifest_list = "unused-1.avro", + }); + auto second = std::make_shared(Snapshot{ + .snapshot_id = 2, + .parent_snapshot_id = 1, + .sequence_number = 2, + .timestamp_ms = TimePointMsFromUnixMs(2000), + .manifest_list = "unused-2.avro", + }); + ICEBERG_UNWRAP_OR_FAIL(auto table, MakeTable({first, second}, 2, std::move(refs))); + + ICEBERG_UNWRAP_OR_FAIL(auto branches, MetadataTable::Make(table)); + ICEBERG_UNWRAP_OR_FAIL(auto branch_stream, branches->Scan()); + ICEBERG_UNWRAP_OR_FAIL(auto branch_batches, ReadAllBatches(std::move(branch_stream))); + ASSERT_EQ(branch_batches.size(), 1); + ASSERT_EQ(branch_batches[0]->num_rows(), 2); + auto branch_names = std::static_pointer_cast<::arrow::StringArray>( + branch_batches[0]->GetColumnByName("name")); + auto branch_ids = std::static_pointer_cast<::arrow::Int64Array>( + branch_batches[0]->GetColumnByName("snapshot_id")); + EXPECT_EQ(branch_names->GetString(0), "dev"); + EXPECT_EQ(branch_ids->Value(0), 1); + EXPECT_EQ(branch_names->GetString(1), "main"); + EXPECT_EQ(branch_ids->Value(1), 2); + + ICEBERG_UNWRAP_OR_FAIL(auto tags, MetadataTable::Make(table)); + ICEBERG_UNWRAP_OR_FAIL(auto tag_stream, tags->Scan()); + ICEBERG_UNWRAP_OR_FAIL(auto tag_batches, ReadAllBatches(std::move(tag_stream))); + ASSERT_EQ(tag_batches.size(), 1); + ASSERT_EQ(tag_batches[0]->num_rows(), 1); + auto tag_names = std::static_pointer_cast<::arrow::StringArray>( + tag_batches[0]->GetColumnByName("name")); + auto max_ref_age = std::static_pointer_cast<::arrow::Int64Array>( + tag_batches[0]->GetColumnByName("max_reference_age_in_ms")); + EXPECT_EQ(tag_names->GetString(0), "release"); + EXPECT_EQ(max_ref_age->Value(0), 5000); +} + +TEST_P(SystemMetadataTablesTest, ScansFilesManifestsAndPartitions) { + 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, files->Scan()); + 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); + + ICEBERG_UNWRAP_OR_FAIL(auto manifests, MetadataTable::Make(table)); + ICEBERG_UNWRAP_OR_FAIL(auto manifests_stream, manifests->Scan()); + 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); + + ICEBERG_UNWRAP_OR_FAIL(auto partitions, MetadataTable::Make(table)); + ICEBERG_UNWRAP_OR_FAIL(auto partitions_stream, partitions->Scan()); + ICEBERG_UNWRAP_OR_FAIL(auto partition_batches, + ReadAllBatches(std::move(partitions_stream))); + ASSERT_EQ(partition_batches.size(), 1); + ASSERT_EQ(partition_batches[0]->num_rows(), 1); + auto records = std::static_pointer_cast<::arrow::Int64Array>( + partition_batches[0]->GetColumnByName("record_count")); + auto file_counts = std::static_pointer_cast<::arrow::Int32Array>( + partition_batches[0]->GetColumnByName("file_count")); + auto updated_snapshot_ids = std::static_pointer_cast<::arrow::Int64Array>( + partition_batches[0]->GetColumnByName("last_updated_snapshot_id")); + EXPECT_EQ(records->Value(0), 1); + EXPECT_EQ(file_counts->Value(0), 1); + EXPECT_EQ(updated_snapshot_ids->Value(0), 10); +} + +TEST_P(SystemMetadataTablesTest, 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, + files->Scan(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(SystemMetadataTablesTest, StopsTimestampTraversalAtExpiredParent) { + auto snapshot = + MakeAppendSnapshot(GetParam(), 2, 1, 2, {"s3://bucket/current.parquet"}); + ICEBERG_UNWRAP_OR_FAIL(auto dev_ref, SnapshotRef::MakeBranch(2)); + std::unordered_map> refs; + refs.emplace("dev", std::move(dev_ref)); + ICEBERG_UNWRAP_OR_FAIL(auto table, MakeTable({snapshot}, 2, std::move(refs))); + ICEBERG_UNWRAP_OR_FAIL(auto files, MetadataTable::Make(table)); + + ICEBERG_UNWRAP_OR_FAIL(auto stream, + files->Scan(SnapshotSelection{.snapshot = snapshot->timestamp_ms, + .ref_name = "dev"})); + ICEBERG_UNWRAP_OR_FAIL(auto batches, ReadAllBatches(std::move(stream))); + ASSERT_EQ(batches.size(), 1); + ASSERT_EQ(batches[0]->num_rows(), 1); +} + +TEST_P(SystemMetadataTablesTest, 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, + files->Scan(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(SystemMetadataTablesTest, 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(SystemMetadataTablesTest, 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(SystemMetadataTablesTest, GroupsNullPartitionValues) { + auto snapshot = MakeAppendSnapshotWithPartitionValues( + GetParam(), 10, std::nullopt, 1, + {{"s3://bucket/first.parquet", PartitionValues(Literal::Null(int32()))}, + {"s3://bucket/second.parquet", PartitionValues(Literal::Null(int32()))}}, + partitioned_spec_); + ICEBERG_UNWRAP_OR_FAIL(auto table, MakeTable({snapshot}, 10, {}, partitioned_spec_)); + ICEBERG_UNWRAP_OR_FAIL(auto partitions, MetadataTable::Make(table)); + + ICEBERG_UNWRAP_OR_FAIL(auto stream, partitions->Scan()); + ICEBERG_UNWRAP_OR_FAIL(auto batches, ReadAllBatches(std::move(stream))); + ASSERT_EQ(batches.size(), 1); + ASSERT_EQ(batches[0]->num_rows(), 1); + auto file_counts = std::static_pointer_cast<::arrow::Int32Array>( + batches[0]->GetColumnByName("file_count")); + EXPECT_EQ(file_counts->Value(0), 2); +} + +INSTANTIATE_TEST_SUITE_P(FormatVersions, SystemMetadataTablesTest, + ::testing::Values(2, 3)); + +} // namespace +} // namespace iceberg diff --git a/src/iceberg/type_fwd.h b/src/iceberg/type_fwd.h index 91e18b24c..293b597e6 100644 --- a/src/iceberg/type_fwd.h +++ b/src/iceberg/type_fwd.h @@ -265,9 +265,14 @@ class DeleteLoader; class PositionDeleteIndex; /// \brief Metadata tables. +class BranchesTable; +class FilesTable; class HistoryTable; +class ManifestsTable; class MetadataTable; +class PartitionsTable; class SnapshotsTable; +class TagsTable; /// \brief Table encryption struct EncryptedKey;