zhjwpku commented on code in PR #774: URL: https://github.com/apache/iceberg-cpp/pull/774#discussion_r4443736116
########## src/iceberg/test/rewrite_manifests_test.cc: ########## @@ -0,0 +1,2802 @@ +/* + * 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/update/rewrite_manifests.h" + +#include <algorithm> +#include <format> +#include <memory> +#include <mutex> +#include <optional> +#include <stdexcept> +#include <string> +#include <unordered_map> +#include <unordered_set> +#include <utility> +#include <vector> + +#include <gmock/gmock.h> +#include <gtest/gtest.h> + +#include "iceberg/avro/avro_register.h" +#include "iceberg/constants.h" +#include "iceberg/expression/expressions.h" +#include "iceberg/file_io.h" +#include "iceberg/file_reader.h" +#include "iceberg/partition_spec.h" +#include "iceberg/schema.h" +#include "iceberg/table_metadata.h" +#include "iceberg/table_properties.h" +#include "iceberg/test/executor.h" +#include "iceberg/test/mock_catalog.h" +#include "iceberg/test/update_test_base.h" +#include "iceberg/transaction.h" +#include "iceberg/transform.h" +#include "iceberg/update/delete_files.h" +#include "iceberg/update/fast_append.h" +#include "iceberg/update/row_delta.h" +#include "iceberg/update/update_partition_spec.h" +#include "iceberg/update/update_properties.h" +#include "iceberg/util/macros.h" + +namespace iceberg { + +namespace { + +class TestRewriteManifests : public RewriteManifests { + public: + static Result<std::shared_ptr<TestRewriteManifests>> Make( + const std::shared_ptr<Table>& table) { + ICEBERG_ASSIGN_OR_RAISE(auto ctx, + TransactionContext::Make(table, TransactionKind::kUpdate)); + return std::shared_ptr<TestRewriteManifests>( + new TestRewriteManifests(table->full_name(), std::move(ctx))); + } + + Result<std::vector<ManifestFile>> ApplyForTest( + const TableMetadata& metadata, const std::shared_ptr<Snapshot>& snapshot) { + return RewriteManifests::Apply(metadata, snapshot); + } + + private: + TestRewriteManifests(std::string table_name, std::shared_ptr<TransactionContext> ctx) + : RewriteManifests(std::move(table_name), std::move(ctx)) {} +}; + +struct ManifestReadState { + std::unordered_set<std::string> source_paths; + std::mutex mutex; + int open_readers = 0; + int max_open_readers = 0; + int opened_readers = 0; + int64_t unread_entries = 0; + int64_t max_unread_entries = 0; + int64_t processed_entries = 0; + bool fail_after_first_batch = false; +}; + +class TrackingManifestReader : public Reader { + public: + TrackingManifestReader(std::unique_ptr<Reader> delegate, + std::shared_ptr<ManifestReadState> state) + : delegate_(std::move(delegate)), state_(std::move(state)) {} + + ~TrackingManifestReader() override { Release(); } + + Status Open(const ReaderOptions& options) override { + auto read_options = options; + const bool track = state_->source_paths.contains(options.path); + if (track) { + read_options.properties.Set(ReaderProperties::kBatchSize, int64_t{2}); + } + ICEBERG_RETURN_UNEXPECTED(delegate_->Open(read_options)); + if (track) { + std::lock_guard lock(state_->mutex); + tracked_ = true; + ++state_->opened_readers; + state_->max_open_readers = + std::max(state_->max_open_readers, ++state_->open_readers); + } + return {}; + } + + Status Close() override { + Release(); + return delegate_->Close(); + } + + Result<std::optional<ArrowArray>> Next() override { + if (tracked_ && state_->fail_after_first_batch && read_batch_) { + return IOError("Injected streaming manifest read failure"); + } + ICEBERG_ASSIGN_OR_RAISE(auto batch, delegate_->Next()); + if (tracked_ && batch.has_value()) { + read_batch_ = true; + std::lock_guard lock(state_->mutex); + state_->unread_entries += batch->length; + state_->max_unread_entries = + std::max(state_->max_unread_entries, state_->unread_entries); + } + return batch; + } + + Result<ArrowSchema> Schema() override { return delegate_->Schema(); } + + Result<std::unordered_map<std::string, std::string>> Metadata() override { + return delegate_->Metadata(); + } + + private: + void Release() { + if (tracked_) { + std::lock_guard lock(state_->mutex); + --state_->open_readers; + tracked_ = false; + } + } + + std::unique_ptr<Reader> delegate_; + std::shared_ptr<ManifestReadState> state_; + bool tracked_ = false; + bool read_batch_ = false; +}; + +class ScopedManifestReadTracking { + public: + explicit ScopedManifestReadTracking(std::shared_ptr<ManifestReadState> state) + : previous_(ReaderFactoryRegistry::GetFactory(FileFormatType::kAvro)) { + ReaderFactoryRegistry::GetFactory(FileFormatType::kAvro) = + [factory = previous_, + state = std::move(state)]() -> Result<std::unique_ptr<Reader>> { + ICEBERG_ASSIGN_OR_RAISE(auto reader, factory()); + return std::make_unique<TrackingManifestReader>(std::move(reader), state); + }; + } + + ~ScopedManifestReadTracking() { + ReaderFactoryRegistry::GetFactory(FileFormatType::kAvro) = std::move(previous_); + } + + private: + ReaderFactory previous_; +}; + +struct ManifestWriteFailureState { + int manifest_output_count = 0; + int fail_on_manifest_output = 0; + int manifest_close_count = 0; + int fail_on_manifest_close = 0; + int throw_on_manifest_close = 0; + std::vector<std::string> manifest_outputs; + std::vector<std::string> manifest_close_attempts; + std::vector<std::string> deleted_files; + std::unordered_set<std::string> open_manifest_paths; + std::vector<std::string> deleted_open_files; +}; + +class ManifestWriteFailureOutputStream : public PositionOutputStream { + public: + ManifestWriteFailureOutputStream(std::unique_ptr<PositionOutputStream> delegate, + std::string location, + std::shared_ptr<ManifestWriteFailureState> state) + : delegate_(std::move(delegate)), + location_(std::move(location)), + state_(std::move(state)) { + state_->open_manifest_paths.insert(location_); + } + + Result<int64_t> Position() const override { return delegate_->Position(); } + + Result<int64_t> StoredLength() const override { return delegate_->StoredLength(); } + + Status Write(std::span<const std::byte> data) override { + return delegate_->Write(data); + } + + Status Flush() override { return delegate_->Flush(); } + + Status Close() override { + if (closed_) { + return {}; + } + closed_ = true; + state_->manifest_close_attempts.push_back(location_); + ++state_->manifest_close_count; + ICEBERG_RETURN_UNEXPECTED(delegate_->Close()); + state_->open_manifest_paths.erase(location_); + if (state_->manifest_close_count == state_->throw_on_manifest_close) { + throw std::runtime_error("Injected rewrite manifest close exception"); + } + if (state_->manifest_close_count == state_->fail_on_manifest_close) { + return IOError("Injected rewrite manifest close failure"); + } + return {}; + } + + private: + std::unique_ptr<PositionOutputStream> delegate_; + std::string location_; + std::shared_ptr<ManifestWriteFailureState> state_; + bool closed_{false}; +}; + +class ManifestWriteFailureOutputFile : public OutputFile { + public: + ManifestWriteFailureOutputFile(std::unique_ptr<OutputFile> delegate, + std::string location, + std::shared_ptr<ManifestWriteFailureState> state) + : delegate_(std::move(delegate)), + location_(std::move(location)), + state_(std::move(state)) {} + + std::string_view location() const override { return location_; } + + Result<std::unique_ptr<PositionOutputStream>> Create() override { + ICEBERG_ASSIGN_OR_RAISE(auto stream, delegate_->Create()); + return std::make_unique<ManifestWriteFailureOutputStream>(std::move(stream), + location_, state_); + } + + Result<std::unique_ptr<PositionOutputStream>> CreateOrOverwrite() override { + ICEBERG_ASSIGN_OR_RAISE(auto stream, delegate_->CreateOrOverwrite()); + return std::make_unique<ManifestWriteFailureOutputStream>(std::move(stream), + location_, state_); + } + + private: + std::unique_ptr<OutputFile> delegate_; + std::string location_; + std::shared_ptr<ManifestWriteFailureState> state_; +}; + +class ManifestWriteFailureFileIO : public FileIO { + public: + ManifestWriteFailureFileIO(std::shared_ptr<FileIO> delegate, + std::string metadata_location, + std::shared_ptr<ManifestWriteFailureState> state) + : delegate_(std::move(delegate)), + metadata_location_(std::move(metadata_location)), + state_(std::move(state)) {} + + Result<std::unique_ptr<InputFile>> NewInputFile(std::string file_location) override { + return delegate_->NewInputFile(std::move(file_location)); + } + + Result<std::unique_ptr<InputFile>> NewInputFile(std::string file_location, + size_t length) override { + return delegate_->NewInputFile(std::move(file_location), length); + } + + Result<std::unique_ptr<OutputFile>> NewOutputFile(std::string file_location) override { + const bool is_manifest_output = IsManifestOutput(file_location); + if (is_manifest_output) { + state_->manifest_outputs.push_back(file_location); + ++state_->manifest_output_count; + if (state_->manifest_output_count == state_->fail_on_manifest_output) { + return IOError("Injected rewrite manifest writer failure"); + } + } + ICEBERG_ASSIGN_OR_RAISE(auto output_file, delegate_->NewOutputFile(file_location)); + if (is_manifest_output) { + return std::make_unique<ManifestWriteFailureOutputFile>( + std::move(output_file), std::move(file_location), state_); + } + return output_file; + } + + Status DeleteFile(const std::string& file_location) override { + if (state_->open_manifest_paths.contains(file_location)) { + state_->deleted_open_files.push_back(file_location); + } + state_->deleted_files.push_back(file_location); + return delegate_->DeleteFile(file_location); + } + + Status DeleteFiles(const std::vector<std::string>& file_locations) override { + state_->deleted_files.insert(state_->deleted_files.end(), file_locations.begin(), + file_locations.end()); + return delegate_->DeleteFiles(file_locations); + } + + private: + bool IsManifestOutput(const std::string& file_location) const { + return file_location.starts_with(metadata_location_) && + file_location.ends_with(".avro"); + } + + std::shared_ptr<FileIO> delegate_; + std::string metadata_location_; + std::shared_ptr<ManifestWriteFailureState> state_; +}; + +} // namespace + +class RewriteManifestsTest : public MinimalUpdateTestBase, + public ::testing::WithParamInterface<int8_t> { + protected: + static void SetUpTestSuite() { avro::RegisterAll(); } + + void SetUp() override { + MinimalUpdateTestBase::SetUp(); + + ICEBERG_UNWRAP_OR_FAIL(spec_, table_->spec()); + ICEBERG_UNWRAP_OR_FAIL(schema_, table_->schema()); + file_a_ = MakeDataFile("a", 10, 100, 1); + file_b_ = MakeDataFile("b", 20, 200, 2); + file_c_ = MakeDataFile("c", 30, 300, 3); + file_d_ = MakeDataFile("d", 40, 400, 4); + } + + int8_t format_version() const override { return GetParam(); } + + std::shared_ptr<DataFile> MakeDataFile(const std::string& name, int64_t record_count, + int64_t size, int64_t partition_value) { + return MakeDataFile(name, spec_, record_count, size, + {Literal::Long(partition_value)}); + } + + std::shared_ptr<DataFile> MakeDataFile(const std::string& name, + std::shared_ptr<PartitionSpec> spec, + int64_t record_count, int64_t size, + std::vector<Literal> partition_values) { + auto file = std::make_shared<DataFile>(); + file->content = DataFile::Content::kData; + file->file_path = std::format("{}/data/{}.parquet", table_location_, name); + file->file_format = FileFormatType::kParquet; + file->partition = PartitionValues(std::move(partition_values)); + file->record_count = record_count; + file->file_size_in_bytes = size; + file->partition_spec_id = spec->spec_id(); + return file; + } + + std::shared_ptr<DataFile> MakeDeleteFile( + const std::string& name, int64_t partition_value, + std::optional<std::string> referenced_data_file = std::nullopt) { + auto file = MakeDataFile(name, 7, 50, partition_value); + file->content = DataFile::Content::kPositionDeletes; + file->file_path = std::format("{}/delete/{}.parquet", table_location_, name); + if (table_->metadata()->format_version >= 3) { + file->file_format = FileFormatType::kPuffin; + file->referenced_data_file = referenced_data_file.value_or(file_a_->file_path); + file->content_offset = 0; + file->content_size_in_bytes = 10; + } + return file; + } + + std::shared_ptr<DataFile> MakeEqualityDeleteFile(const std::string& name, + int64_t partition_value) { + auto file = MakeDataFile(name, /*record_count=*/1, /*size=*/10, partition_value); + file->content = DataFile::Content::kEqualityDeletes; + file->file_path = std::format("{}/delete/{}.parquet", table_location_, name); + file->equality_ids = {1}; + return file; + } + + Status AppendFiles(std::vector<std::shared_ptr<DataFile>> files) { + ICEBERG_ASSIGN_OR_RAISE(auto append, table_->NewFastAppend()); + for (const auto& file : files) { + append->AppendFile(file); + } + ICEBERG_RETURN_UNEXPECTED(append->Commit()); + return table_->Refresh(); + } + + Status AppendDeleteFiles(std::vector<std::shared_ptr<DataFile>> files) { + ICEBERG_ASSIGN_OR_RAISE(auto row_delta, table_->NewRowDelta()); + for (const auto& file : files) { + row_delta->AddDeletes(file); + } + ICEBERG_RETURN_UNEXPECTED(row_delta->Commit()); + return table_->Refresh(); + } + + Status RemoveDeleteFiles(std::vector<std::shared_ptr<DataFile>> files) { + ICEBERG_ASSIGN_OR_RAISE(auto row_delta, table_->NewRowDelta()); + for (const auto& file : files) { + row_delta->RemoveDeletes(file); + } + ICEBERG_RETURN_UNEXPECTED(row_delta->Commit()); + return table_->Refresh(); + } + + void SetSnapshotIdInheritanceEnabled() { + ICEBERG_UNWRAP_OR_FAIL(auto props, table_->NewUpdateProperties()); + props->Set(std::string(TableProperties::kSnapshotIdInheritanceEnabled.key()), "true"); + EXPECT_THAT(props->Commit(), IsOk()); + EXPECT_THAT(table_->Refresh(), IsOk()); + } + + void ReplayV1ReplacementAfterUpgrade(bool inherit_snapshot_id) { + if (format_version() != 1) { + GTEST_SKIP() << "This regression starts from a v1 table"; + } + + ASSERT_THAT(AppendFiles({file_a_, file_b_}), IsOk()); + if (inherit_snapshot_id) { + SetSnapshotIdInheritanceEnabled(); + } + ICEBERG_UNWRAP_OR_FAIL(auto snapshot, table_->current_snapshot()); + ICEBERG_UNWRAP_OR_FAIL(auto before, CurrentManifests()); + ASSERT_EQ(before.size(), 1U); + ICEBERG_UNWRAP_OR_FAIL(auto replacement, WriteExistingManifest("legacy-replacement", + snapshot->snapshot_id, + {file_a_, file_b_})); + if (inherit_snapshot_id) { + replacement.added_snapshot_id.reset(); + } + + ICEBERG_UNWRAP_OR_FAIL(auto txn, table_->NewTransaction()); + ICEBERG_UNWRAP_OR_FAIL(auto rewrite, txn->NewRewriteManifests()); + rewrite->DeleteManifest(before[0]).AddManifest(replacement); + ASSERT_THAT(rewrite->Commit(), IsOk()); + + ICEBERG_UNWRAP_OR_FAIL(auto props, table_->NewUpdateProperties()); + props->Set(std::string(TableProperties::kFormatVersion.key()), "2"); + ASSERT_THAT(props->Commit(), IsOk()); + ASSERT_THAT(table_->Refresh(), IsOk()); + ASSERT_THAT(txn->Commit(), IsOk()); + + ICEBERG_UNWRAP_OR_FAIL(auto after, CurrentManifests()); + ASSERT_EQ(after.size(), 1U); + EXPECT_NE(after[0].manifest_path, replacement.manifest_path); + EXPECT_TRUE(FileExists(replacement.manifest_path)); + ICEBERG_UNWRAP_OR_FAIL(auto entries, ReadManifestEntries(after[0])); + ASSERT_EQ(entries.size(), 2U); + for (const auto& entry : entries) { + EXPECT_EQ(entry.status, ManifestStatus::kExisting); + EXPECT_EQ(entry.snapshot_id, snapshot->snapshot_id); + EXPECT_EQ(entry.sequence_number, std::make_optional<int64_t>(0)); + EXPECT_EQ(entry.file_sequence_number, std::make_optional<int64_t>(0)); + } + ICEBERG_UNWRAP_OR_FAIL(auto rewritten_snapshot, table_->current_snapshot()); + ValidateSummary(rewritten_snapshot, /*replaced=*/1, /*kept=*/0, /*created=*/1, + /*entry_count=*/0); + + ICEBERG_UNWRAP_OR_FAIL(auto remove, table_->NewDeleteFiles()); + remove->DeleteFile(file_a_); + ASSERT_THAT(remove->Commit(), IsOk()); + } + + void SetManifestMergeEnabled(bool enabled) { + ICEBERG_UNWRAP_OR_FAIL(auto props, table_->NewUpdateProperties()); + props->Set(std::string(TableProperties::kManifestMergeEnabled.key()), + enabled ? "true" : "false"); + EXPECT_THAT(props->Commit(), IsOk()); + EXPECT_THAT(table_->Refresh(), IsOk()); + } + + void SetCommitRetryProperties(int32_t retries) { + ICEBERG_UNWRAP_OR_FAIL(auto props, table_->NewUpdateProperties()); + props->Set(std::string(TableProperties::kCommitNumRetries.key()), + std::to_string(retries)); + props->Set(std::string(TableProperties::kCommitMinRetryWaitMs.key()), "1"); + props->Set(std::string(TableProperties::kCommitMaxRetryWaitMs.key()), "1"); + props->Set(std::string(TableProperties::kCommitTotalRetryTimeMs.key()), "1000"); + EXPECT_THAT(props->Commit(), IsOk()); + EXPECT_THAT(table_->Refresh(), IsOk()); + } + + void BindTableWithFailingCommits(int failures) { + auto mock_catalog = std::make_shared<::testing::NiceMock<MockCatalog>>(); + std::weak_ptr<::testing::NiceMock<MockCatalog>> weak_catalog = mock_catalog; + + ON_CALL(*mock_catalog, LoadTable(::testing::_)) + .WillByDefault([this, weak_catalog](const TableIdentifier& identifier) + -> Result<std::shared_ptr<Table>> { + ICEBERG_ASSIGN_OR_RAISE(auto loaded, catalog_->LoadTable(identifier)); + auto catalog = weak_catalog.lock(); + ICEBERG_PRECHECK(catalog != nullptr, "Mock catalog expired"); + return Table::Make(loaded->name(), loaded->metadata(), + std::string(loaded->metadata_file_location()), loaded->io(), + catalog); + }); + + ON_CALL(*mock_catalog, UpdateTable(::testing::_, ::testing::_, ::testing::_)) + .WillByDefault( + [this, weak_catalog, failures]( + const TableIdentifier& identifier, + const std::vector<std::unique_ptr<TableRequirement>>& requirements, + const std::vector<std::unique_ptr<TableUpdate>>& updates) mutable + -> Result<std::shared_ptr<Table>> { + if (failures-- > 0) { + return CommitFailed("Injected failure"); + } + ICEBERG_ASSIGN_OR_RAISE( + auto updated, catalog_->UpdateTable(identifier, requirements, updates)); + auto catalog = weak_catalog.lock(); + ICEBERG_PRECHECK(catalog != nullptr, "Mock catalog expired"); + return Table::Make(updated->name(), updated->metadata(), + std::string(updated->metadata_file_location()), + updated->io(), catalog); + }); + + ICEBERG_UNWRAP_OR_FAIL(auto bound_table, + Table::Make(table_->name(), table_->metadata(), + std::string(table_->metadata_file_location()), + table_->io(), mock_catalog)); + table_ = std::move(bound_table); + mock_catalog_ = std::move(mock_catalog); + } + + void BindTableWithFileIO(std::shared_ptr<FileIO> file_io) { + ICEBERG_UNWRAP_OR_FAIL(auto bound_table, + Table::Make(table_->name(), table_->metadata(), + std::string(table_->metadata_file_location()), + std::move(file_io), catalog_)); + table_ = std::move(bound_table); + } + + void ValidateSummary(const std::shared_ptr<Snapshot>& snapshot, int replaced, int kept, + int created, int entry_count) { + ASSERT_NE(snapshot, nullptr); + const auto& summary = snapshot->summary; + EXPECT_THAT(summary, ::testing::Contains(::testing::Pair( + std::string(SnapshotSummaryFields::kManifestsReplaced), + std::to_string(replaced)))); + EXPECT_THAT(summary, ::testing::Contains(::testing::Pair( + std::string(SnapshotSummaryFields::kManifestsKept), + std::to_string(kept)))); + EXPECT_THAT(summary, ::testing::Contains(::testing::Pair( + std::string(SnapshotSummaryFields::kManifestsCreated), + std::to_string(created)))); + EXPECT_THAT(summary, ::testing::Contains(::testing::Pair( + std::string(SnapshotSummaryFields::kEntriesProcessed), + std::to_string(entry_count)))); + } + + void MatchNumberOfManifestFileWithSpecId(std::span<const ManifestFile> manifests, + int32_t partition_spec_id, + int expected_manifest_count) { + const auto matched_manifest_count = std::ranges::count_if( + manifests, [partition_spec_id](const ManifestFile& manifest) { + return manifest.partition_spec_id == partition_spec_id; + }); + EXPECT_EQ(matched_manifest_count, expected_manifest_count); + } + + bool FileExists(const std::string& path) { + auto input_file = file_io_->NewInputFile(path); + if (!input_file.has_value()) { + return false; + } + return input_file.value()->Size().has_value(); + } + + Result<const ManifestFile*> ManifestContainingPath( + std::span<const ManifestFile> manifests, const std::string& path) { + for (const auto& manifest : manifests) { + ICEBERG_ASSIGN_OR_RAISE(auto entries, ReadManifestEntries(manifest)); + for (const auto& entry : entries) { + if (entry.data_file != nullptr && entry.data_file->file_path == path) { + return &manifest; + } + } + } + return NotFound("Manifest containing path {} is not found", path); + } + + void ExpectManifestEntryContents(const ManifestFile& manifest, + std::vector<std::string> expected_paths, + std::vector<DataFile::Content> expected_contents) { + ASSERT_EQ(expected_contents.size(), expected_paths.size()); + + ICEBERG_UNWRAP_OR_FAIL(auto entries, ReadManifestEntries(manifest)); + ASSERT_EQ(entries.size(), expected_paths.size()); + + std::vector<std::pair<std::string, DataFile::Content>> actual; + actual.reserve(entries.size()); + for (const auto& entry : entries) { + ASSERT_NE(entry.data_file, nullptr); + actual.emplace_back(entry.data_file->file_path, entry.data_file->content); + } + + std::vector<std::pair<std::string, DataFile::Content>> expected; + expected.reserve(expected_paths.size()); + for (size_t i = 0; i < expected_paths.size(); ++i) { + expected.emplace_back(std::move(expected_paths[i]), expected_contents[i]); + } + EXPECT_THAT(actual, ::testing::UnorderedElementsAreArray(expected)); + } + + Result<std::vector<ManifestFile>> CurrentManifests() { + ICEBERG_RETURN_UNEXPECTED(table_->Refresh()); + ICEBERG_ASSIGN_OR_RAISE(auto snapshot, table_->current_snapshot()); + SnapshotReader reader(snapshot.get()); + ICEBERG_ASSIGN_OR_RAISE(auto manifests, reader.DataManifests(file_io_)); + return std::vector<ManifestFile>(manifests.begin(), manifests.end()); + } + + Result<std::vector<ManifestFile>> CurrentDeleteManifests() { + ICEBERG_RETURN_UNEXPECTED(table_->Refresh()); + ICEBERG_ASSIGN_OR_RAISE(auto snapshot, table_->current_snapshot()); + SnapshotReader reader(snapshot.get()); + ICEBERG_ASSIGN_OR_RAISE(auto manifests, reader.DeleteManifests(file_io_)); + return std::vector<ManifestFile>(manifests.begin(), manifests.end()); + } + + Result<ManifestFile> WriteExistingManifest( + const std::string& name, int64_t snapshot_id, + const std::vector<std::shared_ptr<DataFile>>& files, + ManifestContent content = ManifestContent::kData, + std::shared_ptr<PartitionSpec> manifest_spec = nullptr, + int64_t data_sequence_number = TableMetadata::kInitialSequenceNumber, + std::optional<int64_t> file_sequence_number = std::nullopt) { Review Comment: I'll do that, I'll also update the helper overload that currently hard-codes zero. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
