Skip to content

Commit 6c781c7

Browse files
authored
feat(io): add bulk delete API to FileIO. (#659)
## Summary Add a new `FileIO::DeleteFiles(...)` API as a bulk deletion entry point. The default implementation deletes files sequentially by calling the existing `DeleteFile(...)` method and returns the first deletion error encountered. This PR only adds the API and backward-compatible fallback behavior. It does not yet update `ExpireSnapshots` to use `DeleteFiles(...)`, and it does not introduce parallel deletion. Fixed: #658 ## Motivation `ExpireSnapshots` and other cleanup flows may need to delete many files. A bulk deletion API gives FileIO implementations a common extension point for future optimized deletion strategies, such as storage-native batch deletion or parallel fallback deletion.
1 parent 136b468 commit 6c781c7

9 files changed

Lines changed: 134 additions & 11 deletions

File tree

src/iceberg/arrow/arrow_io.cc

Lines changed: 13 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -22,6 +22,7 @@
2222
#include <limits>
2323
#include <mutex>
2424
#include <optional>
25+
#include <vector>
2526

2627
#include <arrow/buffer.h>
2728
#include <arrow/filesystem/localfs.h>
@@ -568,6 +569,18 @@ Status ArrowFileSystemFileIO::DeleteFile(const std::string& file_location) {
568569
return {};
569570
}
570571

572+
Status ArrowFileSystemFileIO::DeleteFiles(
573+
const std::vector<std::string>& file_locations) {
574+
std::vector<std::string> paths;
575+
paths.reserve(file_locations.size());
576+
for (const auto& file_location : file_locations) {
577+
ICEBERG_ASSIGN_OR_RAISE(auto path, ResolvePath(file_location));
578+
paths.push_back(std::move(path));
579+
}
580+
ICEBERG_ARROW_RETURN_NOT_OK(arrow_fs_->DeleteFiles(paths));
581+
return {};
582+
}
583+
571584
std::unique_ptr<FileIO> ArrowFileSystemFileIO::MakeMockFileIO() {
572585
return std::make_unique<ArrowFileSystemFileIO>(
573586
std::make_shared<::arrow::fs::internal::MockFileSystem>(

src/iceberg/arrow/arrow_io_internal.h

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -23,6 +23,7 @@
2323
#include <memory>
2424
#include <optional>
2525
#include <string>
26+
#include <vector>
2627

2728
#include <arrow/filesystem/type_fwd.h>
2829
#include <arrow/io/type_fwd.h>
@@ -77,6 +78,9 @@ class ICEBERG_BUNDLE_EXPORT ArrowFileSystemFileIO : public FileIO {
7778
/// \brief Delete a file at the given location.
7879
Status DeleteFile(const std::string& file_location) override;
7980

81+
/// \brief Delete files at the given locations.
82+
Status DeleteFiles(const std::vector<std::string>& file_locations) override;
83+
8084
/// \brief Get the Arrow file system.
8185
const std::shared_ptr<::arrow::fs::FileSystem>& fs() const { return arrow_fs_; }
8286

src/iceberg/file_io.cc

Lines changed: 7 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -100,4 +100,11 @@ Status FileIO::WriteFile(const std::string& file_location, std::string_view cont
100100
return FinishWithCloseStatus(std::move(status), stream->Close());
101101
}
102102

103+
Status FileIO::DeleteFiles(const std::vector<std::string>& file_locations) {
104+
for (const auto& file_location : file_locations) {
105+
ICEBERG_RETURN_UNEXPECTED(DeleteFile(file_location));
106+
}
107+
return {};
108+
}
109+
103110
} // namespace iceberg

src/iceberg/file_io.h

Lines changed: 11 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -26,6 +26,7 @@
2626
#include <span>
2727
#include <string>
2828
#include <string_view>
29+
#include <vector>
2930

3031
#include "iceberg/iceberg_export.h"
3132
#include "iceberg/result.h"
@@ -154,6 +155,16 @@ class ICEBERG_EXPORT FileIO {
154155
virtual Status DeleteFile(const std::string& file_location) {
155156
return NotImplemented("DeleteFile not implemented");
156157
}
158+
159+
/// \brief Delete files at the given locations.
160+
///
161+
/// Implementations that can delete multiple files efficiently should override this
162+
/// method. The default implementation deletes files sequentially using DeleteFile
163+
/// and returns the first error encountered.
164+
///
165+
/// \param file_locations The locations of the files to delete.
166+
/// \return void if all deletes succeed, or an error code if any delete fails.
167+
virtual Status DeleteFiles(const std::vector<std::string>& file_locations);
157168
};
158169

159170
} // namespace iceberg

src/iceberg/test/CMakeLists.txt

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -124,6 +124,7 @@ add_iceberg_test(util_test
124124
data_file_set_test.cc
125125
decimal_test.cc
126126
endian_test.cc
127+
file_io_test.cc
127128
formatter_test.cc
128129
lazy_test.cc
129130
location_util_test.cc

src/iceberg/test/arrow_io_test.cc

Lines changed: 15 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -21,6 +21,7 @@
2121
#include <array>
2222
#include <memory>
2323
#include <string>
24+
#include <vector>
2425

2526
#include <arrow/filesystem/localfs.h>
2627
#include <arrow/result.h>
@@ -341,6 +342,20 @@ TEST_F(LocalFileIOTest, DeleteFile) {
341342
EXPECT_THAT(del_res, HasErrorMessage("Cannot delete file"));
342343
}
343344

345+
TEST_F(LocalFileIOTest, DeleteFiles) {
346+
auto first_path = CreateNewTempFilePath();
347+
auto second_path = CreateNewTempFilePath();
348+
ASSERT_THAT(file_io_->WriteFile(first_path, "hello"), IsOk());
349+
ASSERT_THAT(file_io_->WriteFile(second_path, "world"), IsOk());
350+
351+
std::vector<std::string> paths = {first_path, second_path};
352+
EXPECT_THAT(file_io_->DeleteFiles(paths), IsOk());
353+
354+
EXPECT_THAT(file_io_->ReadFile(first_path, std::nullopt), IsError(ErrorKind::kIOError));
355+
EXPECT_THAT(file_io_->ReadFile(second_path, std::nullopt),
356+
IsError(ErrorKind::kIOError));
357+
}
358+
344359
void VerifyReadFullyReadsFromAbsolutePosition(const std::shared_ptr<FileIO>& file_io,
345360
const std::string& path) {
346361
ASSERT_THAT(file_io->WriteFile(path, "abcdef"), IsOk());

src/iceberg/test/file_io_test.cc

Lines changed: 75 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,75 @@
1+
/*
2+
* Licensed to the Apache Software Foundation (ASF) under one
3+
* or more contributor license agreements. See the NOTICE file
4+
* distributed with this work for additional information
5+
* regarding copyright ownership. The ASF licenses this file
6+
* to you under the Apache License, Version 2.0 (the
7+
* "License"); you may not use this file except in compliance
8+
* with the License. You may obtain a copy of the License at
9+
*
10+
* http://www.apache.org/licenses/LICENSE-2.0
11+
*
12+
* Unless required by applicable law or agreed to in writing,
13+
* software distributed under the License is distributed on an
14+
* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
15+
* KIND, either express or implied. See the License for the
16+
* specific language governing permissions and limitations
17+
* under the License.
18+
*/
19+
20+
#include "iceberg/file_io.h"
21+
22+
#include <string>
23+
#include <utility>
24+
#include <vector>
25+
26+
#include <gtest/gtest.h>
27+
28+
#include "iceberg/test/matchers.h"
29+
30+
namespace iceberg {
31+
namespace {
32+
33+
class RecordingFileIO : public FileIO {
34+
public:
35+
explicit RecordingFileIO(std::string failure_path = "")
36+
: failure_path_(std::move(failure_path)) {}
37+
38+
Status DeleteFile(const std::string& file_location) override {
39+
deleted_paths.push_back(file_location);
40+
if (file_location == failure_path_) {
41+
return IOError("failed to delete {}", file_location);
42+
}
43+
return {};
44+
}
45+
46+
std::vector<std::string> deleted_paths;
47+
48+
private:
49+
std::string failure_path_;
50+
};
51+
52+
} // namespace
53+
54+
TEST(FileIOTest, DeleteFilesFallsBackToDeleteFileForEachPath) {
55+
RecordingFileIO file_io;
56+
std::vector<std::string> paths = {"file-a.avro", "file-b.avro"};
57+
58+
EXPECT_THAT(file_io.DeleteFiles(paths), IsOk());
59+
EXPECT_THAT(file_io.deleted_paths,
60+
::testing::ElementsAre("file-a.avro", "file-b.avro"));
61+
}
62+
63+
TEST(FileIOTest, DeleteFilesReturnsFirstDeleteFileError) {
64+
RecordingFileIO file_io("file-b.avro");
65+
std::vector<std::string> paths = {"file-a.avro", "file-b.avro", "file-c.avro"};
66+
67+
auto status = file_io.DeleteFiles(paths);
68+
69+
EXPECT_THAT(status, IsError(ErrorKind::kIOError));
70+
EXPECT_THAT(status, HasErrorMessage("failed to delete file-b.avro"));
71+
EXPECT_THAT(file_io.deleted_paths,
72+
::testing::ElementsAre("file-a.avro", "file-b.avro"));
73+
}
74+
75+
} // namespace iceberg

src/iceberg/test/meson.build

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -88,6 +88,7 @@ iceberg_tests = {
8888
'data_file_set_test.cc',
8989
'decimal_test.cc',
9090
'endian_test.cc',
91+
'file_io_test.cc',
9192
'formatter_test.cc',
9293
'lazy_test.cc',
9394
'location_util_test.cc',

src/iceberg/update/expire_snapshots.cc

Lines changed: 7 additions & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -98,26 +98,22 @@ class FileCleanupStrategy {
9898
return expired;
9999
}
100100

101-
/// \brief Delete a single file
102-
void DeleteFile(const std::string& path) {
101+
/// \brief Delete files at the given locations.
102+
void DeleteFiles(const std::unordered_set<std::string>& paths) {
103103
try {
104104
if (delete_func_) {
105-
delete_func_(path);
105+
for (const auto& path : paths) {
106+
delete_func_(path);
107+
}
106108
} else {
107-
std::ignore = file_io_->DeleteFile(path);
109+
std::vector<std::string> path_list(paths.begin(), paths.end());
110+
std::ignore = file_io_->DeleteFiles(path_list);
108111
}
109112
} catch (...) {
110113
// TODO(shangxinli): add retry
111114
}
112115
}
113116

114-
// TODO(shangxinli): Add bulk deletion
115-
void DeleteFiles(const std::unordered_set<std::string>& paths) {
116-
for (const auto& path : paths) {
117-
DeleteFile(path);
118-
}
119-
}
120-
121117
bool HasAnyStatisticsFiles(const TableMetadata& metadata) const {
122118
return !metadata.statistics.empty() || !metadata.partition_statistics.empty();
123119
}

0 commit comments

Comments
 (0)