| /* |
| * 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/delete_files.h" |
| |
| #include <memory> |
| #include <string> |
| #include <vector> |
| |
| #include <gmock/gmock.h> |
| #include <gtest/gtest.h> |
| |
| #include "iceberg/avro/avro_register.h" |
| #include "iceberg/expression/expressions.h" |
| #include "iceberg/expression/literal.h" |
| #include "iceberg/manifest/manifest_entry.h" |
| #include "iceberg/partition_spec.h" |
| #include "iceberg/row/partition_values.h" |
| #include "iceberg/schema.h" |
| #include "iceberg/snapshot.h" |
| #include "iceberg/table.h" |
| #include "iceberg/test/matchers.h" |
| #include "iceberg/test/update_test_base.h" |
| #include "iceberg/update/fast_append.h" |
| |
| namespace iceberg { |
| |
| class DeleteFilesTest : public MinimalUpdateTestBase { |
| 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("/data/file_a.parquet", /*partition_x=*/1L); |
| file_b_ = MakeDataFile("/data/file_b.parquet", /*partition_x=*/2L); |
| } |
| |
| std::shared_ptr<DataFile> MakeDataFile(const std::string& path, int64_t partition_x) { |
| auto file = std::make_shared<DataFile>(); |
| file->content = DataFile::Content::kData; |
| file->file_path = table_location_ + path; |
| file->file_format = FileFormatType::kParquet; |
| file->partition = PartitionValues(std::vector<Literal>{Literal::Long(partition_x)}); |
| file->file_size_in_bytes = 1024; |
| file->record_count = 100; |
| file->partition_spec_id = spec_->spec_id(); |
| return file; |
| } |
| |
| void SetLongBounds(const std::shared_ptr<DataFile>& file, int32_t field_id, |
| int64_t lower, int64_t upper) { |
| ASSERT_NE(file, nullptr); |
| ICEBERG_UNWRAP_OR_FAIL(auto lower_bound, Literal::Long(lower).Serialize()); |
| ICEBERG_UNWRAP_OR_FAIL(auto upper_bound, Literal::Long(upper).Serialize()); |
| file->value_counts[field_id] = file->record_count; |
| file->null_value_counts[field_id] = 0; |
| file->lower_bounds[field_id] = lower_bound; |
| file->upper_bounds[field_id] = upper_bound; |
| } |
| |
| void CommitFiles(const std::vector<std::shared_ptr<DataFile>>& files) { |
| ICEBERG_UNWRAP_OR_FAIL(auto append, table_->NewFastAppend()); |
| for (const auto& file : files) { |
| append->AppendFile(file); |
| } |
| ASSERT_THAT(append->Commit(), IsOk()); |
| ASSERT_THAT(table_->Refresh(), IsOk()); |
| } |
| |
| void CommitInitialFiles() { CommitFiles({file_a_, file_b_}); } |
| |
| void ExpectOneFileDeleted() { |
| ASSERT_THAT(table_->Refresh(), IsOk()); |
| ICEBERG_UNWRAP_OR_FAIL(auto snapshot, table_->current_snapshot()); |
| EXPECT_EQ(snapshot->summary.at(SnapshotSummaryFields::kOperation), |
| DataOperation::kDelete); |
| EXPECT_EQ(snapshot->summary.at(SnapshotSummaryFields::kDeletedDataFiles), "1"); |
| EXPECT_EQ(snapshot->summary.at(SnapshotSummaryFields::kDeletedRecords), "100"); |
| EXPECT_EQ(snapshot->summary.at(SnapshotSummaryFields::kRemovedFileSize), "1024"); |
| } |
| |
| std::shared_ptr<PartitionSpec> spec_; |
| std::shared_ptr<Schema> schema_; |
| std::shared_ptr<DataFile> file_a_; |
| std::shared_ptr<DataFile> file_b_; |
| |
| static constexpr int32_t kYFieldId = 2; |
| }; |
| |
| TEST_F(DeleteFilesTest, DeleteFileByPath) { |
| CommitInitialFiles(); |
| |
| ICEBERG_UNWRAP_OR_FAIL(auto delete_files, table_->NewDeleteFiles()); |
| delete_files->DeleteFile(file_a_->file_path); |
| |
| EXPECT_THAT(delete_files->Commit(), IsOk()); |
| ExpectOneFileDeleted(); |
| } |
| |
| TEST_F(DeleteFilesTest, DeleteFileByDataFile) { |
| CommitInitialFiles(); |
| |
| ICEBERG_UNWRAP_OR_FAIL(auto delete_files, table_->NewDeleteFiles()); |
| delete_files->DeleteFile(file_a_); |
| |
| EXPECT_THAT(delete_files->Commit(), IsOk()); |
| ExpectOneFileDeleted(); |
| } |
| |
| TEST_F(DeleteFilesTest, DeleteFromRowFilterCaseInsensitive) { |
| CommitInitialFiles(); |
| |
| ICEBERG_UNWRAP_OR_FAIL(auto delete_files, table_->NewDeleteFiles()); |
| delete_files->CaseSensitive(false).DeleteFromRowFilter( |
| Expressions::Equal("X", Literal::Long(1L))); |
| |
| EXPECT_THAT(delete_files->Commit(), IsOk()); |
| ExpectOneFileDeleted(); |
| } |
| |
| TEST_F(DeleteFilesTest, EmptyDeleteCommit) { |
| CommitInitialFiles(); |
| ICEBERG_UNWRAP_OR_FAIL(auto previous_snapshot, table_->current_snapshot()); |
| |
| ICEBERG_UNWRAP_OR_FAIL(auto delete_files, table_->NewDeleteFiles()); |
| |
| EXPECT_THAT(delete_files->Commit(), IsOk()); |
| |
| ASSERT_THAT(table_->Refresh(), IsOk()); |
| ICEBERG_UNWRAP_OR_FAIL(auto snapshot, table_->current_snapshot()); |
| ASSERT_TRUE(snapshot->parent_snapshot_id.has_value()); |
| EXPECT_EQ(snapshot->parent_snapshot_id.value(), previous_snapshot->snapshot_id); |
| EXPECT_EQ(snapshot->summary.at(SnapshotSummaryFields::kOperation), |
| DataOperation::kDelete); |
| EXPECT_EQ(snapshot->summary.count(SnapshotSummaryFields::kDeletedDataFiles), 0U); |
| EXPECT_EQ(snapshot->summary.count(SnapshotSummaryFields::kDeletedRecords), 0U); |
| EXPECT_EQ(snapshot->summary.count(SnapshotSummaryFields::kRemovedFileSize), 0U); |
| } |
| |
| TEST_F(DeleteFilesTest, DeleteFromRowFilter) { |
| CommitInitialFiles(); |
| |
| ICEBERG_UNWRAP_OR_FAIL(auto delete_files, table_->NewDeleteFiles()); |
| delete_files->DeleteFromRowFilter(Expressions::Equal("x", Literal::Long(1L))); |
| |
| EXPECT_THAT(delete_files->Commit(), IsOk()); |
| ExpectOneFileDeleted(); |
| } |
| |
| TEST_F(DeleteFilesTest, DeleteFromRowFilterRejectsPartialMatchFile) { |
| auto partial_match_file = MakeDataFile("/data/partial_match.parquet", |
| /*partition_x=*/1L); |
| SetLongBounds(partial_match_file, kYFieldId, /*lower=*/0L, /*upper=*/10L); |
| CommitFiles({partial_match_file}); |
| ICEBERG_UNWRAP_OR_FAIL(auto previous_snapshot, table_->current_snapshot()); |
| |
| ICEBERG_UNWRAP_OR_FAIL(auto delete_files, table_->NewDeleteFiles()); |
| delete_files->DeleteFromRowFilter(Expressions::Equal("y", Literal::Long(5L))); |
| |
| auto status = delete_files->Commit(); |
| EXPECT_THAT(status, IsError(ErrorKind::kValidationFailed)); |
| EXPECT_THAT(status, |
| HasErrorMessage("Cannot delete file where some, but not all, rows match " |
| "filter")); |
| EXPECT_THAT(status, HasErrorMessage(partial_match_file->file_path)); |
| |
| ASSERT_THAT(table_->Refresh(), IsOk()); |
| ICEBERG_UNWRAP_OR_FAIL(auto snapshot, table_->current_snapshot()); |
| EXPECT_EQ(snapshot->snapshot_id, previous_snapshot->snapshot_id); |
| } |
| |
| TEST_F(DeleteFilesTest, ValidateFilesExistRejectsMissingPath) { |
| CommitInitialFiles(); |
| |
| ICEBERG_UNWRAP_OR_FAIL(auto delete_files, table_->NewDeleteFiles()); |
| delete_files->DeleteFile(table_location_ + "/data/missing.parquet") |
| .ValidateFilesExist(); |
| |
| auto status = delete_files->Commit(); |
| EXPECT_THAT(status, IsError(ErrorKind::kValidationFailed)); |
| EXPECT_THAT(status, HasErrorMessage("Missing required files to delete")); |
| } |
| |
| } // namespace iceberg |