diff --git a/src/iceberg/CMakeLists.txt b/src/iceberg/CMakeLists.txt index 8a98274fff..fbb73a22d9 100644 --- a/src/iceberg/CMakeLists.txt +++ b/src/iceberg/CMakeLists.txt @@ -242,6 +242,7 @@ set(ICEBERG_DATA_SOURCES data/delete_loader.cc data/equality_delete_writer.cc data/file_scan_task_reader.cc + data/position_delete_update.cc data/position_delete_writer.cc data/writer.cc) diff --git a/src/iceberg/arrow/arrow_io.cc b/src/iceberg/arrow/arrow_io.cc index 4c795badfb..11f905da67 100644 --- a/src/iceberg/arrow/arrow_io.cc +++ b/src/iceberg/arrow/arrow_io.cc @@ -591,7 +591,14 @@ Result> ArrowFileSystemFileIO::NewOutputFile( /// \brief Delete a file at the given location. Status ArrowFileSystemFileIO::DeleteFile(const std::string& file_location) { ICEBERG_ASSIGN_OR_RAISE(auto path, ResolvePath(file_location)); - ICEBERG_ARROW_RETURN_NOT_OK(arrow_fs_->DeleteFile(path)); + auto status = arrow_fs_->DeleteFile(path); + if (!status.ok()) { + auto info = arrow_fs_->GetFileInfo(path); + if (info.ok() && info->type() == ::arrow::fs::FileType::NotFound) { + return {}; + } + } + ICEBERG_ARROW_RETURN_NOT_OK(status); return {}; } diff --git a/src/iceberg/data/output_file_cleanup_internal.h b/src/iceberg/data/output_file_cleanup_internal.h new file mode 100644 index 0000000000..c84b5f91eb --- /dev/null +++ b/src/iceberg/data/output_file_cleanup_internal.h @@ -0,0 +1,56 @@ +/* + * 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 "iceberg/file_io.h" +#include "iceberg/result.h" + +namespace iceberg::internal { + +// Try every owned output, keeping failed paths for a subsequent cleanup attempt. +inline Status CleanupOutputFiles(FileIO& io, std::set& paths) { + Status first_error; + for (auto it = paths.begin(); it != paths.end();) { + auto status = io.DeleteFile(*it); + if (status.has_value()) { + it = paths.erase(it); + } else { + if (first_error.has_value()) first_error = std::move(status); + ++it; + } + } + return first_error; +} + +// Preserve the operation's error kind while reporting a cleanup failure, if any. +inline Status FailWithOutputCleanup(Error error, FileIO& io, + std::set& paths) { + auto cleanup = CleanupOutputFiles(io, paths); + if (!cleanup.has_value()) { + error.message += "; output cleanup failed: " + cleanup.error().message; + } + return std::unexpected(std::move(error)); +} + +} // namespace iceberg::internal diff --git a/src/iceberg/data/position_delete_update.cc b/src/iceberg/data/position_delete_update.cc new file mode 100644 index 0000000000..93e463bb05 --- /dev/null +++ b/src/iceberg/data/position_delete_update.cc @@ -0,0 +1,264 @@ +/* + * 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/data/position_delete_update.h" + +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +#include "iceberg/data/delete_loader.h" +#include "iceberg/data/output_file_cleanup_internal.h" +#include "iceberg/data/position_delete_writer.h" +#include "iceberg/deletes/dv_writer.h" +#include "iceberg/file_format.h" +#include "iceberg/file_io.h" +#include "iceberg/location_provider.h" +#include "iceberg/manifest/manifest_entry.h" +#include "iceberg/partition_spec.h" +#include "iceberg/schema.h" +#include "iceberg/table.h" +#include "iceberg/table_metadata.h" +#include "iceberg/table_properties.h" +#include "iceberg/table_scan.h" +#include "iceberg/update/row_delta.h" +#include "iceberg/util/macros.h" +#include "iceberg/util/string_util.h" +#include "iceberg/util/uuid.h" + +namespace iceberg { + +namespace { + +struct TargetFile { + std::shared_ptr data_file; + std::shared_ptr spec; + std::vector> position_delete_files; +}; + +} // namespace + +class PositionDeleteUpdate::Impl { + public: + explicit Impl(std::shared_ptr table) : table_(std::move(table)) {} + + Status Delete(std::string_view data_file_path, int64_t pos) { + ICEBERG_PRECHECK(!terminal_, "Position delete update is no longer usable"); + ICEBERG_PRECHECK(!data_file_path.empty(), "Data file path cannot be empty"); + ICEBERG_PRECHECK(pos >= 0, "Position delete must be non-negative: {}", pos); + deletes_[std::string(data_file_path)].push_back(pos); + return {}; + } + + Status Commit() { + ICEBERG_PRECHECK(!terminal_, "Position delete update is no longer usable"); + ICEBERG_PRECHECK(!deletes_.empty(), "Position delete update is empty"); + ICEBERG_PRECHECK(table_->metadata()->format_version >= 2, + "Position deletes require table format version 2 or later"); + ICEBERG_RETURN_UNEXPECTED(internal::CleanupOutputFiles(*table_->io(), output_paths_)); + auto status = CommitDeletes(); + if (!status.has_value() && status.error().kind != ErrorKind::kCommitStateUnknown) { + return internal::FailWithOutputCleanup(std::move(status.error()), *table_->io(), + output_paths_); + } + terminal_ = true; + output_paths_.clear(); + return status; + } + + private: + Status CommitDeletes() { + ICEBERG_ASSIGN_OR_RAISE(auto snapshot, table_->current_snapshot()); + ICEBERG_ASSIGN_OR_RAISE(auto targets, ResolveTargets()); + + ICEBERG_ASSIGN_OR_RAISE(auto written, table_->metadata()->format_version >= 3 + ? WriteDeletionVectors(targets) + : WriteParquetDeletes(targets)); + ICEBERG_ASSIGN_OR_RAISE(auto row_delta, table_->NewRowDelta()); + row_delta->ValidateFromSnapshot(snapshot->snapshot_id) + .ValidateDataFilesExist(written.referenced_data_files) + .ValidateDeletedFiles(); + for (const auto& file : written.data_files) { + row_delta->AddDeletes(file); + } + for (const auto& file : written.rewritten_delete_files) { + row_delta->RemoveDeletes(file); + } + + return row_delta->Commit(); + } + + Result> ResolveTargets() const { + ICEBERG_ASSIGN_OR_RAISE(auto scan_builder, table_->NewScan()); + ICEBERG_ASSIGN_OR_RAISE(auto scan, scan_builder->Build()); + ICEBERG_ASSIGN_OR_RAISE(auto tasks, scan->PlanFilesStream()); + + std::unordered_map targets; + while (targets.size() < deletes_.size()) { + ICEBERG_ASSIGN_OR_RAISE(auto task, tasks->Next()); + if (!task.has_value()) break; + const auto& data_file = (*task)->data_file(); + if (!deletes_.contains(data_file->file_path)) { + continue; + } + + ICEBERG_PRECHECK(data_file->partition_spec_id.has_value(), + "Data file is missing partition spec ID: {}", + data_file->file_path); + ICEBERG_ASSIGN_OR_RAISE(auto spec, table_->metadata()->PartitionSpecById( + *data_file->partition_spec_id)); + + TargetFile target{.data_file = data_file, .spec = std::move(spec)}; + for (const auto& delete_file : (*task)->delete_files()) { + if (delete_file->content == DataFile::Content::kPositionDeletes) { + target.position_delete_files.push_back(delete_file); + } + } + + targets.emplace(data_file->file_path, std::move(target)); + } + + for (const auto& [path, _] : deletes_) { + ICEBERG_PRECHECK(targets.contains(path), "Cannot find live data file: {}", path); + } + return targets; + } + + Result WriteDeletionVectors( + const std::unordered_map& targets) { + ICEBERG_ASSIGN_OR_RAISE(auto location_provider, table_->location_provider()); + auto output_path = location_provider->NewDataLocation( + std::format("position-deletes-{}.puffin", Uuid::GenerateV7().ToString())); + output_paths_.insert(output_path); + + DeleteLoader loader(table_->io()); + ICEBERG_ASSIGN_OR_RAISE( + auto writer, + DVWriter::Make(DVWriterOptions{ + .path = output_path, + .io = table_->io(), + .load_previous_deletes = [&targets, &loader](std::string_view path) + -> Result> { + const auto& previous = targets.at(std::string(path)).position_delete_files; + if (previous.empty()) return std::nullopt; + ICEBERG_ASSIGN_OR_RAISE(auto index, + loader.LoadPositionDeletes(previous, path)); + return std::optional(std::move(index)); + }, + })); + + for (const auto& [path, positions] : deletes_) { + const auto& target = targets.at(path); + for (int64_t pos : positions) { + ICEBERG_RETURN_UNEXPECTED( + writer->Delete(path, pos, target.spec, target.data_file->partition)); + } + } + ICEBERG_RETURN_UNEXPECTED(writer->Close()); + return writer->Metadata(); + } + + Result WriteParquetDeletes( + const std::unordered_map& targets) { + ICEBERG_ASSIGN_OR_RAISE(auto location_provider, table_->location_provider()); + ICEBERG_ASSIGN_OR_RAISE(auto schema, table_->schema()); + + DeleteWriteResult result; + result.data_files.reserve(deletes_.size()); + result.referenced_data_files.reserve(deletes_.size()); + auto properties = table_->properties().configs(); + properties[TableProperties::kParquetCompression.key()] = + table_->properties().Get(TableProperties::kDeleteParquetCompression); + properties[TableProperties::kParquetCompressionLevel.key()] = + table_->properties().Get(TableProperties::kDeleteParquetCompressionLevel); + const auto write_uuid = Uuid::GenerateV7().ToString(); + size_t file_number = 0; + + for (auto& [path, positions] : deletes_) { + std::ranges::sort(positions); + + const auto& target = targets.at(path); + const auto filename = + std::format("position-deletes-{}-{}.parquet", write_uuid, file_number++); + auto output_path = location_provider->NewDataLocation(filename); + output_paths_.insert(output_path); + + ICEBERG_ASSIGN_OR_RAISE(auto writer, + PositionDeleteWriter::Make(PositionDeleteWriterOptions{ + .path = output_path, + .schema = schema, + .spec = target.spec, + .partition = target.data_file->partition, + .format = FileFormatType::kParquet, + .io = table_->io(), + .properties = properties, + })); + for (int64_t pos : positions) { + ICEBERG_RETURN_UNEXPECTED(writer->WriteDelete(path, pos)); + } + ICEBERG_RETURN_UNEXPECTED(writer->Close()); + ICEBERG_ASSIGN_OR_RAISE(auto metadata, writer->Metadata()); + result.data_files.insert(result.data_files.end(), + std::make_move_iterator(metadata.data_files.begin()), + std::make_move_iterator(metadata.data_files.end())); + result.referenced_data_files.push_back(path); + } + return result; + } + + std::shared_ptr
table_; + std::map, StringLess> deletes_; + std::set output_paths_; + bool terminal_ = false; +}; + +PositionDeleteUpdate::PositionDeleteUpdate(std::unique_ptr impl) + : impl_(std::move(impl)) {} + +PositionDeleteUpdate::~PositionDeleteUpdate() = default; + +Result> PositionDeleteUpdate::Make( + std::shared_ptr
table) { + ICEBERG_PRECHECK(table != nullptr, + "Cannot create position delete update without table"); + return std::unique_ptr( + new PositionDeleteUpdate(std::make_unique(std::move(table)))); +} + +PositionDeleteUpdate& PositionDeleteUpdate::Delete(std::string_view data_file_path, + int64_t pos) { + ICEBERG_BUILDER_RETURN_IF_ERROR(impl_->Delete(data_file_path, pos)); + return *this; +} + +Status PositionDeleteUpdate::Commit() { + ICEBERG_RETURN_UNEXPECTED(CheckErrors()); + return impl_->Commit(); +} + +} // namespace iceberg diff --git a/src/iceberg/data/position_delete_update.h b/src/iceberg/data/position_delete_update.h new file mode 100644 index 0000000000..3b531d8fab --- /dev/null +++ b/src/iceberg/data/position_delete_update.h @@ -0,0 +1,60 @@ +/* + * 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/data/position_delete_update.h +/// Table-aware position delete writing and commit. + +#include +#include +#include + +#include "iceberg/iceberg_data_export.h" +#include "iceberg/result.h" +#include "iceberg/type_fwd.h" +#include "iceberg/util/error_collector.h" + +namespace iceberg { + +/// \brief Writes and commits position deletes using the table format. +/// +/// Format v2 tables produce Parquet position delete files. Format v3 tables +/// produce deletion vectors and merge previous file-scoped position deletes. +class ICEBERG_DATA_EXPORT PositionDeleteUpdate : public ErrorCollector { + public: + ~PositionDeleteUpdate() override; + + /// \brief Create a position delete update for a table. + static Result> Make(std::shared_ptr
table); + + /// \brief Add a deleted row position for a live data file. + PositionDeleteUpdate& Delete(std::string_view data_file_path, int64_t pos); + + /// \brief Write and commit all added position deletes. + Status Commit(); + + private: + class Impl; + std::unique_ptr impl_; + + explicit PositionDeleteUpdate(std::unique_ptr impl); +}; + +} // namespace iceberg diff --git a/src/iceberg/delete_file_index.cc b/src/iceberg/delete_file_index.cc index a8c4ef1261..2bd6fc7336 100644 --- a/src/iceberg/delete_file_index.cc +++ b/src/iceberg/delete_file_index.cc @@ -41,6 +41,7 @@ #include "iceberg/util/content_file_util.h" #include "iceberg/util/executor_util_internal.h" #include "iceberg/util/macros.h" +#include "iceberg/util/struct_like_set.h" namespace iceberg { @@ -458,6 +459,15 @@ Result> DeleteFileIndex::FindDV( "sequence number {}", it->second.sequence_number.value(), seq); + const auto& dv = *it->second.data_file; + ICEBERG_CHECK(dv.partition_spec_id == data_file.partition_spec_id, + "DV and data file have mismatched partition specs: {}", + data_file.file_path); + ICEBERG_ASSIGN_OR_RAISE(auto partitions_match, + StructLikeEqual(dv.partition, data_file.partition)); + ICEBERG_CHECK(partitions_match, "DV and data file have mismatched partitions: {}", + data_file.file_path); + return it->second.data_file; } diff --git a/src/iceberg/file_io.h b/src/iceberg/file_io.h index e22e5cf210..791133fcf3 100644 --- a/src/iceberg/file_io.h +++ b/src/iceberg/file_io.h @@ -162,6 +162,9 @@ class ICEBERG_EXPORT FileIO { /// \brief Delete a file at the given location. /// + /// Deletion is idempotent: implementations must return success when the file does + /// not exist. + /// /// \param file_location The location of the file to delete. /// \return void if the delete succeeded, an error code if the delete failed. virtual Status DeleteFile(const std::string& file_location) { diff --git a/src/iceberg/test/CMakeLists.txt b/src/iceberg/test/CMakeLists.txt index 6f6ff76031..c4bdd0733b 100644 --- a/src/iceberg/test/CMakeLists.txt +++ b/src/iceberg/test/CMakeLists.txt @@ -265,6 +265,7 @@ if(ICEBERG_BUILD_BUNDLE) delete_loader_test.cc dv_writer_test.cc file_scan_task_reader_test.cc + position_delete_update_test.cc literal_util_test.cc) endif() diff --git a/src/iceberg/test/arrow_io_test.cc b/src/iceberg/test/arrow_io_test.cc index 7bc9ebba53..6b3f6e4dd4 100644 --- a/src/iceberg/test/arrow_io_test.cc +++ b/src/iceberg/test/arrow_io_test.cc @@ -364,8 +364,7 @@ TEST_F(LocalFileIOTest, DeleteFile) { EXPECT_THAT(del_res, IsOk()); del_res = file_io_->DeleteFile(temp_filepath_); - EXPECT_THAT(del_res, IsError(ErrorKind::kIOError)); - EXPECT_THAT(del_res, HasErrorMessage("Cannot delete file")); + EXPECT_THAT(del_res, IsOk()); } TEST_F(LocalFileIOTest, DeleteFiles) { @@ -414,6 +413,13 @@ TEST_F(LocalFileIOTest, StdReadFullyReadsFromAbsolutePosition) { VerifyReadFullyReadsFromAbsolutePosition(file_io, temp_filepath_)); } +TEST_F(LocalFileIOTest, StdDeleteFileIsIdempotent) { + auto file_io = std::make_shared(); + ASSERT_THAT(file_io->WriteFile(temp_filepath_, "abc"), IsOk()); + EXPECT_THAT(file_io->DeleteFile(temp_filepath_), IsOk()); + EXPECT_THAT(file_io->DeleteFile(temp_filepath_), IsOk()); +} + TEST_F(LocalFileIOTest, StdReadKeepsPositionAvailableAtEof) { auto file_io = std::make_shared(); ASSERT_THAT(file_io->WriteFile(temp_filepath_, "abc"), IsOk()); diff --git a/src/iceberg/test/delete_file_index_test.cc b/src/iceberg/test/delete_file_index_test.cc index 75c82bc39c..48693a9b49 100644 --- a/src/iceberg/test/delete_file_index_test.cc +++ b/src/iceberg/test/delete_file_index_test.cc @@ -1113,6 +1113,81 @@ TEST_P(DeleteFileIndexTest, TestMultipleDVs) { EXPECT_THAT(index_result, HasErrorMessage(file_a_->file_path)); } +TEST_P(DeleteFileIndexTest, TestDVApplicability) { + auto version = GetParam(); + if (version < 3) { + GTEST_SKIP() << "DVs only supported in V3+"; + } + + const auto null_partition = PartitionValues({Literal::Null(int32())}); + auto null_partition_file = MakeDataFile("/path/to/data-null.parquet", null_partition, + partitioned_spec_->spec_id()); + + struct TestCase { + std::string name; + PartitionValues dv_partition; + std::shared_ptr dv_spec; + std::shared_ptr data_file; + bool applies; + }; + const std::vector cases = { + { + .name = "equal-partition", + .dv_partition = file_a_->partition, + .dv_spec = partitioned_spec_, + .data_file = file_a_, + .applies = true, + }, + { + .name = "different-spec", + .dv_partition = PartitionValues{}, + .dv_spec = unpartitioned_spec_, + .data_file = file_a_, + .applies = false, + }, + { + .name = "different-partition-value", + .dv_partition = file_b_->partition, + .dv_spec = partitioned_spec_, + .data_file = file_a_, + .applies = false, + }, + { + .name = "equal-null-partition", + .dv_partition = null_partition, + .dv_spec = partitioned_spec_, + .data_file = null_partition_file, + .applies = true, + }, + { + .name = "null-partition-mismatch", + .dv_partition = file_a_->partition, + .dv_spec = partitioned_spec_, + .data_file = null_partition_file, + .applies = false, + }, + }; + + for (const auto& test_case : cases) { + SCOPED_TRACE(test_case.name); + auto dv = MakeDV("/path/to/" + test_case.name + ".puffin", test_case.dv_partition, + test_case.dv_spec->spec_id(), test_case.data_file->file_path); + std::vector entries; + entries.push_back(MakeDeleteEntry(/*snapshot_id=*/1000L, /*sequence_number=*/2, dv)); + auto manifest = WriteDeleteManifest(version, /*snapshot_id=*/1000L, + std::move(entries), test_case.dv_spec); + ICEBERG_UNWRAP_OR_FAIL(auto index, BuildIndex({manifest})); + auto result = index->ForDataFile(1, *test_case.data_file); + if (test_case.applies) { + ICEBERG_UNWRAP_OR_FAIL(auto deletes, std::move(result)); + ASSERT_EQ(deletes.size(), 1); + EXPECT_EQ(deletes[0]->file_path, dv->file_path); + } else { + EXPECT_THAT(result, IsError(ErrorKind::kValidationFailed)); + } + } +} + TEST_P(DeleteFileIndexTest, TestInvalidDVSequenceNumber) { auto version = GetParam(); if (version < 3) { diff --git a/src/iceberg/test/position_delete_update_test.cc b/src/iceberg/test/position_delete_update_test.cc new file mode 100644 index 0000000000..f94e5684ef --- /dev/null +++ b/src/iceberg/test/position_delete_update_test.cc @@ -0,0 +1,327 @@ +/* + * 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/data/position_delete_update.h" + +#include +#include +#include + +#include +#include +#include +#include +#include +#include +#include + +#include "iceberg/arrow/arrow_io_internal.h" +#include "iceberg/avro/avro_register.h" +#include "iceberg/data/data_writer.h" +#include "iceberg/data/file_scan_task_reader.h" +#include "iceberg/file_reader.h" +#include "iceberg/metadata_columns.h" +#include "iceberg/parquet/parquet_register.h" +#include "iceberg/partition_spec.h" +#include "iceberg/schema.h" +#include "iceberg/schema_internal.h" +#include "iceberg/table.h" +#include "iceberg/table_metadata.h" +#include "iceberg/table_properties.h" +#include "iceberg/table_scan.h" +#include "iceberg/test/matchers.h" +#include "iceberg/test/mock_catalog.h" +#include "iceberg/test/update_test_base.h" +#include "iceberg/update/delete_files.h" +#include "iceberg/update/fast_append.h" +#include "iceberg/update/update_partition_spec.h" +#include "iceberg/update/update_properties.h" + +namespace iceberg { +namespace { + +class FailOncePuffinDeleteFileIO : public arrow::ArrowFileSystemFileIO { + public: + explicit FailOncePuffinDeleteFileIO(std::shared_ptr<::arrow::fs::FileSystem> fs) + : ArrowFileSystemFileIO(std::move(fs)) {} + + Status DeleteFile(const std::string& path) override { + if (path.ends_with(".puffin") && delete_attempts++ == 0) { + return IOError("injected cleanup failure"); + } + return ArrowFileSystemFileIO::DeleteFile(path); + } + + int delete_attempts = 0; +}; + +class PositionDeleteV3Test : public MinimalUpdateTestBase { + protected: + static void SetUpTestSuite() { + avro::RegisterAll(); + parquet::RegisterAll(); + } + + int8_t format_version() const override { return 3; } + virtual bool unpartitioned() const { return false; } + + void SetRetries(int retries) { + ICEBERG_UNWRAP_OR_FAIL(auto properties, table_->NewUpdateProperties()); + properties->Set(TableProperties::kCommitNumRetries.key(), std::to_string(retries)) + .Set(TableProperties::kCommitMinRetryWaitMs.key(), "1") + .Set(TableProperties::kCommitMaxRetryWaitMs.key(), "1") + .Set(TableProperties::kDeleteParquetCompression.key(), "uncompressed"); + ASSERT_THAT(properties->Commit(), IsOk()); + ASSERT_THAT(table_->Refresh(), IsOk()); + } + + void SetUp() override { + MinimalUpdateTestBase::SetUp(); + SetRetries(0); + if (unpartitioned()) { + ICEBERG_UNWRAP_OR_FAIL(auto evolution, table_->NewUpdatePartitionSpec()); + evolution->RemoveField("x"); + ASSERT_THAT(evolution->Commit(), IsOk()); + ASSERT_THAT(table_->Refresh(), IsOk()); + } + ICEBERG_UNWRAP_OR_FAIL(auto schema, table_->schema()); + ICEBERG_UNWRAP_OR_FAIL(auto spec, table_->spec()); + ICEBERG_UNWRAP_OR_FAIL( + auto writer, + DataWriter::Make({ + .path = table_location_ + "/data/file.parquet", + .schema = schema, + .spec = spec, + .partition = + unpartitioned() ? PartitionValues{} : PartitionValues(Literal::Long(10)), + .format = FileFormatType::kParquet, + .io = file_io_, + .properties = {{"write.parquet.compression-codec", "uncompressed"}}, + })); + ArrowSchema c_schema{}; + ASSERT_THAT(ToArrowSchema(*schema, &c_schema), IsOk()); + auto type = ::arrow::ImportType(&c_schema).ValueOrDie(); + auto array = ::arrow::json::ArrayFromJSONString( + type, "[[10,0,0],[10,1,10],[10,2,20],[10,3,30],[10,4,40]]") + .ValueOrDie(); + ArrowArray c_array{}; + ASSERT_TRUE(::arrow::ExportArray(*array, &c_array).ok()); + ASSERT_THAT(writer->Write(&c_array), IsOk()); + ASSERT_THAT(writer->Close(), IsOk()); + ICEBERG_UNWRAP_OR_FAIL(auto result, writer->Metadata()); + data_file_ = result.data_files.front(); + ICEBERG_UNWRAP_OR_FAIL(auto append, table_->NewFastAppend()); + append->AppendFile(data_file_); + ASSERT_THAT(append->Commit(), IsOk()); + ASSERT_THAT(table_->Refresh(), IsOk()); + } + + std::vector> Tasks() { + ICEBERG_THROW_NOT_OK(table_->Refresh()); + auto builder = table_->NewScan().value(); + return builder->Build().value()->PlanFiles().value(); + } + + std::vector ReadRows() { + table_ = catalog_->LoadTable(table_ident_).value(); + auto tasks = Tasks(); + auto schema = table_->schema().value(); + auto reader = FileScanTaskReader::Make({ + .io = file_io_, + .table_schema = schema, + .schemas = table_->metadata()->schemas, + .projected_schema = schema, + }) + .value(); + auto stream = reader->Open(*tasks.at(0)).value(); + auto batches = ::arrow::ImportRecordBatchReader(&stream).ValueOrDie(); + std::vector rows; + while (auto batch = batches->Next().ValueOrDie()) { + const auto& values = static_cast(*batch->column(1)); + for (int64_t i = 0; i < values.length(); ++i) rows.push_back(values.Value(i)); + } + return rows; + } + + std::vector PuffinFiles() { + auto io = std::static_pointer_cast(file_io_); + ::arrow::fs::FileSelector selector; + selector.base_dir = table_location_ + "/data"; + std::vector paths; + for (const auto& info : io->fs()->GetFileInfo(selector).ValueOrDie()) { + if (info.path().ends_with(".puffin")) paths.push_back(info.path()); + } + return paths; + } + + std::shared_ptr data_file_; +}; + +struct RoutingCase { + int8_t format_version; + bool unpartitioned; +}; + +class PositionDeleteUpdateTest : public PositionDeleteV3Test, + public ::testing::WithParamInterface { + protected: + int8_t format_version() const override { return GetParam().format_version; } + bool unpartitioned() const override { return GetParam().unpartitioned; } +}; + +TEST_P(PositionDeleteUpdateTest, DeleteCommitReloadAndMerge) { + ICEBERG_UNWRAP_OR_FAIL(auto first, PositionDeleteUpdate::Make(table_)); + first->Delete(data_file_->file_path, 3).Delete(data_file_->file_path, 1); + ASSERT_THAT(first->Commit(), IsOk()); + EXPECT_THAT(ReadRows(), ::testing::ElementsAre(0, 2, 4)); + auto task = Tasks().at(0); + ASSERT_EQ(task->delete_files().size(), 1); + const auto old_delete = task->delete_files().front(); + EXPECT_EQ(old_delete->file_format, + format_version() == 2 ? FileFormatType::kParquet : FileFormatType::kPuffin); + + if (format_version() == 2) { + auto projection = std::make_shared(std::vector{ + MetadataColumns::kDeleteFilePath, MetadataColumns::kDeleteFilePos}); + ICEBERG_UNWRAP_OR_FAIL( + auto reader, + ReaderFactoryRegistry::Open( + FileFormatType::kParquet, + {.path = old_delete->file_path, .io = file_io_, .projection = projection})); + ICEBERG_UNWRAP_OR_FAIL(auto batch, reader->Next()); + ASSERT_TRUE(batch.has_value()); + ArrowSchema schema{}; + ASSERT_THAT(ToArrowSchema(*projection, &schema), IsOk()); + auto rows = ::arrow::ImportArray(&*batch, ::arrow::ImportType(&schema).ValueOrDie()) + .ValueOrDie(); + const auto& positions = static_cast( + *static_cast(*rows).field(1)); + ASSERT_EQ(positions.length(), 2); + EXPECT_EQ(positions.Value(0), 1); + EXPECT_EQ(positions.Value(1), 3); + ICEBERG_UNWRAP_OR_FAIL(auto properties, table_->NewUpdateProperties()); + properties->Set(TableProperties::kFormatVersion.key(), "3"); + ASSERT_THAT(properties->Commit(), IsOk()); + ASSERT_THAT(table_->Refresh(), IsOk()); + } + if (!unpartitioned()) { + ICEBERG_UNWRAP_OR_FAIL(auto evolution, table_->NewUpdatePartitionSpec()); + evolution->RemoveField("x"); + ASSERT_THAT(evolution->Commit(), IsOk()); + ASSERT_THAT(table_->Refresh(), IsOk()); + ASSERT_NE(table_->metadata()->default_spec_id, data_file_->partition_spec_id); + } + + ICEBERG_UNWRAP_OR_FAIL(auto second, PositionDeleteUpdate::Make(table_)); + second->Delete(data_file_->file_path, 2); + ASSERT_THAT(second->Commit(), IsOk()); + EXPECT_THAT(ReadRows(), ::testing::ElementsAre(0, 4)); + task = Tasks().at(0); + ASSERT_EQ(task->delete_files().size(), 1); + const auto& dv = task->delete_files().front(); + EXPECT_EQ(dv->file_format, FileFormatType::kPuffin); + EXPECT_NE(dv->file_path, old_delete->file_path); + EXPECT_EQ(dv->partition_spec_id, data_file_->partition_spec_id); + EXPECT_EQ(dv->partition, data_file_->partition); +} + +INSTANTIATE_TEST_SUITE_P(FormatAndPartitioning, PositionDeleteUpdateTest, + ::testing::Values(RoutingCase{2, false}, RoutingCase{3, false}, + RoutingCase{3, true})); + +TEST_F(PositionDeleteV3Test, ConcurrentRemovalRejectsPreparedDeletes) { + SetRetries(1); + auto mock = std::make_shared<::testing::NiceMock>(); + ON_CALL(*mock, LoadTable(::testing::_)).WillByDefault([this](const auto& name) { + return catalog_->LoadTable(name); + }); + EXPECT_CALL(*mock, UpdateTable(::testing::_, ::testing::_, ::testing::_)) + .WillOnce([this](const auto&, const auto&, + const auto&) -> Result> { + ICEBERG_ASSIGN_OR_RAISE(auto removal, table_->NewDeleteFiles()); + removal->DeleteFile(data_file_); + ICEBERG_RETURN_UNEXPECTED(removal->Commit()); + return CommitFailed("concurrent removal committed"); + }); + ICEBERG_UNWRAP_OR_FAIL( + auto competing, + Table::Make(table_->name(), table_->metadata(), + std::string(table_->metadata_file_location()), file_io_, mock)); + ICEBERG_UNWRAP_OR_FAIL(auto update, PositionDeleteUpdate::Make(competing)); + update->Delete(data_file_->file_path, 1); + EXPECT_THAT(update->Commit(), IsError(ErrorKind::kValidationFailed)); + EXPECT_TRUE(Tasks().empty()); + EXPECT_TRUE(PuffinFiles().empty()); +} + +TEST_F(PositionDeleteV3Test, CommitRetryReusesWrittenDV) { + SetRetries(1); + FailCommits(1); + ICEBERG_UNWRAP_OR_FAIL(auto update, PositionDeleteUpdate::Make(table_)); + update->Delete(data_file_->file_path, 1); + ASSERT_THAT(update->Commit(), IsOk()); + EXPECT_EQ(PuffinFiles().size(), 1); +} + +TEST_F(PositionDeleteV3Test, CleanupFailureIsReportedAndRetried) { + auto mock = std::make_shared<::testing::NiceMock>(); + EXPECT_CALL(*mock, UpdateTable(::testing::_, ::testing::_, ::testing::_)) + .WillOnce(::testing::Return(CommitFailed("injected commit failure"))) + .WillOnce([this](const auto& name, const auto& requirements, const auto& updates) { + return catalog_->UpdateTable(name, requirements, updates); + }); + auto io = std::static_pointer_cast(file_io_); + auto failing_io = std::make_shared(io->fs()); + ICEBERG_UNWRAP_OR_FAIL( + auto failing, + Table::Make(table_->name(), table_->metadata(), + std::string(table_->metadata_file_location()), failing_io, mock)); + ICEBERG_UNWRAP_OR_FAIL(auto update, PositionDeleteUpdate::Make(failing)); + update->Delete(data_file_->file_path, 1); + auto status = update->Commit(); + EXPECT_THAT(status, IsError(ErrorKind::kCommitFailed)); + EXPECT_THAT(status, HasErrorMessage("injected cleanup failure")); + const auto retained = PuffinFiles(); + ASSERT_EQ(retained.size(), 1); + ASSERT_THAT(update->Commit(), IsOk()); + const auto committed = PuffinFiles(); + ASSERT_EQ(committed.size(), 1); + EXPECT_NE(committed.front(), retained.front()); + EXPECT_EQ(failing_io->delete_attempts, 2); +} + +TEST_F(PositionDeleteV3Test, CommitStateUnknownPreservesOutput) { + auto mock = std::make_shared<::testing::NiceMock>(); + EXPECT_CALL(*mock, UpdateTable(::testing::_, ::testing::_, ::testing::_)) + .WillOnce(::testing::Return(CommitStateUnknown("injected unknown state"))); + ICEBERG_UNWRAP_OR_FAIL( + auto unknown, + Table::Make(table_->name(), table_->metadata(), + std::string(table_->metadata_file_location()), file_io_, mock)); + ICEBERG_UNWRAP_OR_FAIL(auto update, PositionDeleteUpdate::Make(unknown)); + update->Delete(data_file_->file_path, 1); + EXPECT_THAT(update->Commit(), IsError(ErrorKind::kCommitStateUnknown)); + const auto retained = PuffinFiles(); + ASSERT_EQ(retained.size(), 1); + EXPECT_THAT(update->Commit(), IsError(ErrorKind::kInvalidArgument)); + EXPECT_EQ(PuffinFiles(), retained); +} + +} // namespace +} // namespace iceberg diff --git a/src/iceberg/test/std_io.h b/src/iceberg/test/std_io.h index 725fc7ba52..c7d5b312b2 100644 --- a/src/iceberg/test/std_io.h +++ b/src/iceberg/test/std_io.h @@ -319,11 +319,9 @@ class StdFileIO : public FileIO { Status DeleteFile(const std::string& file_location) override { std::error_code ec; - if (!std::filesystem::remove(file_location, ec)) { - if (ec) { - return IOError("Failed to delete file {}: {}", file_location, ec.message()); - } - return IOError("File does not exist: {}", file_location); + std::filesystem::remove(file_location, ec); + if (ec) { + return IOError("Failed to delete file {}: {}", file_location, ec.message()); } return {}; }