diff --git a/src/iceberg/CMakeLists.txt b/src/iceberg/CMakeLists.txt index 4cde4fb23..17edcc778 100644 --- a/src/iceberg/CMakeLists.txt +++ b/src/iceberg/CMakeLists.txt @@ -58,6 +58,7 @@ set(ICEBERG_SOURCES geospatial.cc incremental_append_scan.cc incremental_changelog_scan.cc + inspect/metadata_manifest_util_internal.cc inspect/history_table.cc inspect/metadata_scan.cc inspect/metadata_table.cc diff --git a/src/iceberg/inspect/metadata_manifest_util_internal.cc b/src/iceberg/inspect/metadata_manifest_util_internal.cc new file mode 100644 index 000000000..81bcb0418 --- /dev/null +++ b/src/iceberg/inspect/metadata_manifest_util_internal.cc @@ -0,0 +1,118 @@ +/* + * 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_manifest_util_internal.h" + +#include +#include +#include +#include +#include +#include + +#include "iceberg/constants.h" +#include "iceberg/manifest/manifest_entry.h" +#include "iceberg/manifest/manifest_list.h" +#include "iceberg/manifest/manifest_reader.h" +#include "iceberg/partition_spec.h" +#include "iceberg/snapshot.h" +#include "iceberg/table.h" +#include "iceberg/table_metadata.h" +#include "iceberg/util/macros.h" + +namespace iceberg::internal { + +namespace { + +class LiveFilesIterator : public Stream { + public: + LiveFilesIterator(std::shared_ptr io, std::shared_ptr schema, + std::unordered_map> specs, + std::vector manifests, std::vector columns) + : io_(std::move(io)), + schema_(std::move(schema)), + specs_(std::move(specs)), + manifests_(std::move(manifests)), + columns_(std::move(columns)) {} + + protected: + Result> NextImpl() override { + while (entry_index_ == entries_.size()) { + entries_.clear(); + entry_index_ = 0; + if (manifest_index_ == manifests_.size()) { + return std::nullopt; + } + const auto& manifest = manifests_[manifest_index_++]; + auto spec = specs_.find(manifest.partition_spec_id); + ICEBERG_CHECK(spec != specs_.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); + current_spec_ = spec->second; + ICEBERG_ASSIGN_OR_RAISE(auto reader, + ManifestReader::Make(manifest, io_, schema_, specs_)); + if (!columns_.empty()) { + reader->Select(columns_); + } + ICEBERG_ASSIGN_OR_RAISE(entries_, reader->LiveEntries()); + } + auto& entry = entries_[entry_index_++]; + ICEBERG_PRECHECK(entry.data_file != nullptr, + "Manifest contains an entry with no data file"); + return LiveFile{.file = std::move(entry.data_file), + .spec = current_spec_, + .snapshot_id = entry.snapshot_id}; + } + + private: + std::shared_ptr io_; + std::shared_ptr schema_; + std::unordered_map> specs_; + std::vector manifests_; + std::vector columns_; + std::vector entries_; + std::shared_ptr current_spec_; + size_t manifest_index_ = 0; + size_t entry_index_ = 0; +}; + +} // namespace + +Result>> LiveFiles( + const TableMetadata& metadata, const std::shared_ptr& io, + const std::shared_ptr& snapshot, std::vector columns) { + ICEBERG_ASSIGN_OR_RAISE(auto schema, metadata.Schema()); + std::unordered_map> specs; + for (const auto& spec : metadata.partition_specs) { + ICEBERG_PRECHECK(spec != nullptr, "Partition spec cannot be null"); + specs.emplace(spec->spec_id(), spec); + } + std::vector manifests; + if (snapshot != nullptr) { + SnapshotReader snapshot_reader(snapshot.get()); + ICEBERG_ASSIGN_OR_RAISE(auto snapshot_manifests, snapshot_reader.Manifests(io)); + manifests.assign(snapshot_manifests.begin(), snapshot_manifests.end()); + } + return std::make_unique(io, std::move(schema), std::move(specs), + std::move(manifests), std::move(columns)); +} + +} // namespace iceberg::internal diff --git a/src/iceberg/inspect/metadata_manifest_util_internal.h b/src/iceberg/inspect/metadata_manifest_util_internal.h new file mode 100644 index 000000000..4a522bddd --- /dev/null +++ b/src/iceberg/inspect/metadata_manifest_util_internal.h @@ -0,0 +1,48 @@ +/* + * 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 "iceberg/iceberg_export.h" +#include "iceberg/result.h" +#include "iceberg/type_fwd.h" +#include "iceberg/util/stream.h" + +namespace iceberg::internal { + +struct LiveFile { + std::shared_ptr file; + std::shared_ptr spec; + std::optional snapshot_id; +}; + +/// \brief Iterate over live files, reading one manifest at a time. +/// An empty column selection reads all data-file fields. Each manifest's entries +/// are materialized by ManifestReader, independently of the Arrow batch size. +ICEBERG_EXPORT Result>> LiveFiles( + const TableMetadata& metadata, const std::shared_ptr& io, + const std::shared_ptr& snapshot, std::vector columns = {}); + +} // namespace iceberg::internal diff --git a/src/iceberg/test/CMakeLists.txt b/src/iceberg/test/CMakeLists.txt index 1874fc25e..da9200a3f 100644 --- a/src/iceberg/test/CMakeLists.txt +++ b/src/iceberg/test/CMakeLists.txt @@ -206,6 +206,7 @@ if(ICEBERG_BUILD_BUNDLE) add_iceberg_test(metadata_table_test USE_BUNDLE SOURCES + inspect/metadata_manifest_util_test.cc inspect/history_table_test.cc inspect/metadata_table_test.cc inspect/refs_table_test.cc diff --git a/src/iceberg/test/inspect/metadata_manifest_util_test.cc b/src/iceberg/test/inspect/metadata_manifest_util_test.cc new file mode 100644 index 000000000..8361d783c --- /dev/null +++ b/src/iceberg/test/inspect/metadata_manifest_util_test.cc @@ -0,0 +1,103 @@ +/* + * 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 "iceberg/inspect/metadata_manifest_util_internal.h" +#include "iceberg/manifest/manifest_entry.h" +#include "iceberg/snapshot.h" +#include "iceberg/table_metadata.h" +#include "iceberg/test/matchers.h" +#include "iceberg/test/scan_test_base.h" +#include "iceberg/util/timepoint.h" + +namespace iceberg::internal { +namespace { +class MetadataManifestTest : public ScanTestBase { + protected: + std::shared_ptr SnapshotWith(std::vector manifests) { + return std::make_shared( + Snapshot{.snapshot_id = 10, + .sequence_number = 1, + .timestamp_ms = TimePointMsFromUnixMs(1000), + .manifest_list = WriteManifestList(GetParam(), 10, 0, 1, manifests)}); + } +}; + +TEST_P(MetadataManifestTest, ReadsLiveFilesAcrossManifestsAndRetainsInputs) { + auto first = WriteDataManifest( + GetParam(), 10, + {MakeEntry(ManifestStatus::kAdded, 10, 1, MakeDataFile("first.parquet")), + MakeEntry(ManifestStatus::kDeleted, 10, 1, MakeDataFile("deleted.parquet"))}); + auto second = WriteDataManifest( + GetParam(), 10, + {MakeEntry(ManifestStatus::kAdded, 10, 1, MakeDataFile("second.parquet"))}); + auto snapshot = SnapshotWith({first, second}); + auto metadata = MakeTableMetadata({snapshot}, 10); + ICEBERG_UNWRAP_OR_FAIL(auto files, LiveFiles(*metadata, file_io_, snapshot)); + metadata.reset(); + snapshot.reset(); + file_io_.reset(); + for (const auto& path : {"first.parquet", "second.parquet"}) { + ICEBERG_UNWRAP_OR_FAIL(auto row, files->Next()); + ASSERT_TRUE(row.has_value()); + EXPECT_EQ(row->file->file_path, path); + ASSERT_NE(row->spec, nullptr); + } + for (int attempt = 0; attempt < 2; ++attempt) { + ICEBERG_UNWRAP_OR_FAIL(auto row, files->Next()); + EXPECT_FALSE(row.has_value()); + } +} + +TEST_P(MetadataManifestTest, DefersManifestReadsAndKeepsFailuresTerminal) { + auto first = WriteDataManifest( + GetParam(), 10, + {MakeEntry(ManifestStatus::kAdded, 10, 1, MakeDataFile("first.parquet"))}); + auto missing = first; + missing.manifest_path = "missing-manifest.avro"; + auto snapshot = SnapshotWith({first, missing}); + auto metadata = MakeTableMetadata({snapshot}, 10); + ICEBERG_UNWRAP_OR_FAIL(auto files, LiveFiles(*metadata, file_io_, snapshot)); + ICEBERG_UNWRAP_OR_FAIL(auto row, files->Next()); + ASSERT_TRUE(row.has_value()); + EXPECT_EQ(row->file->file_path, "first.parquet"); + for (int attempt = 0; attempt < 2; ++attempt) { + EXPECT_FALSE(files->Next().has_value()); + } +} + +TEST_P(MetadataManifestTest, EmptyTablesHaveNoLiveFiles) { + auto metadata = MakeTableMetadata({}, kInvalidSnapshotId); + ICEBERG_UNWRAP_OR_FAIL(auto files, LiveFiles(*metadata, file_io_, nullptr)); + ICEBERG_UNWRAP_OR_FAIL(auto row, files->Next()); + EXPECT_FALSE(row.has_value()); +} + +INSTANTIATE_TEST_SUITE_P(FormatVersions, MetadataManifestTest, + ::testing::Values(1, 2, 3)); +} // namespace +} // namespace iceberg::internal