Skip to content

Commit 34e09ef

Browse files
committed
feat: implement RewriteFiles operation
1 parent 3c9d13d commit 34e09ef

11 files changed

Lines changed: 831 additions & 0 deletions

src/iceberg/CMakeLists.txt

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -107,6 +107,7 @@ set(ICEBERG_SOURCES
107107
update/merging_snapshot_update.cc
108108
update/overwrite_files.cc
109109
update/pending_update.cc
110+
update/rewrite_files.cc
110111
update/row_delta.cc
111112
update/set_snapshot.cc
112113
update/snapshot_manager.cc

src/iceberg/table.cc

Lines changed: 7 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -36,6 +36,7 @@
3636
#include "iceberg/update/fast_append.h"
3737
#include "iceberg/update/merge_append.h"
3838
#include "iceberg/update/overwrite_files.h"
39+
#include "iceberg/update/rewrite_files.h"
3940
#include "iceberg/update/row_delta.h"
4041
#include "iceberg/update/set_snapshot.h"
4142
#include "iceberg/update/snapshot_manager.h"
@@ -245,6 +246,12 @@ Result<std::shared_ptr<OverwriteFiles>> Table::NewOverwrite() {
245246
return OverwriteFiles::Make(name().name, std::move(ctx));
246247
}
247248

249+
Result<std::shared_ptr<RewriteFiles>> Table::NewRewriteFiles() {
250+
ICEBERG_ASSIGN_OR_RAISE(
251+
auto ctx, TransactionContext::Make(shared_from_this(), TransactionKind::kUpdate));
252+
return RewriteFiles::Make(name().name, std::move(ctx));
253+
}
254+
248255
Result<std::shared_ptr<UpdateStatistics>> Table::NewUpdateStatistics() {
249256
ICEBERG_ASSIGN_OR_RAISE(
250257
auto ctx, TransactionContext::Make(shared_from_this(), TransactionKind::kUpdate));

src/iceberg/table.h

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -188,6 +188,10 @@ class ICEBERG_EXPORT Table : public std::enable_shared_from_this<Table> {
188188
/// \brief Create a new OverwriteFiles to overwrite data files and commit the changes.
189189
virtual Result<std::shared_ptr<OverwriteFiles>> NewOverwrite();
190190

191+
/// \brief Create a new RewriteFiles to replace files in this table and commit the
192+
/// changes.
193+
virtual Result<std::shared_ptr<RewriteFiles>> NewRewriteFiles();
194+
191195
/// \brief Create a new SnapshotManager to manage snapshots and snapshot references.
192196
virtual Result<std::shared_ptr<SnapshotManager>> NewSnapshotManager();
193197

src/iceberg/test/CMakeLists.txt

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -230,6 +230,7 @@ if(ICEBERG_BUILD_BUNDLE)
230230
merge_append_test.cc
231231
merging_snapshot_update_test.cc
232232
name_mapping_update_test.cc
233+
rewrite_files_test.cc
233234
row_delta_test.cc
234235
snapshot_manager_test.cc
235236
transaction_test.cc

src/iceberg/test/merging_snapshot_update_test.cc

Lines changed: 37 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1405,6 +1405,43 @@ TEST_F(MergingSnapshotUpdateTest,
14051405
IsError(ErrorKind::kValidationFailed));
14061406
}
14071407

1408+
TEST_F(MergingSnapshotUpdateTest,
1409+
ValidateNoNewDeletesForDataFilesIgnoresEqualityDeletesWhenFlagIsTrue) {
1410+
// This tests the behavior that RewriteFiles::SetDataSequenceNumber() and
1411+
// RewriteFiles::RewriteDataFiles() enable: when a data sequence number is
1412+
// set for rewritten data files, concurrent equality deletes at higher
1413+
// sequence numbers still apply to the new files and are NOT a conflict.
1414+
// Only position deletes should still fail (tested separately by
1415+
// ValidateNoNewDeletesForDataFilesFailsOnPositionDeleteWhenIgnoringEqualityDeletes).
1416+
CommitFileA();
1417+
ICEBERG_UNWRAP_OR_FAIL(auto first_snapshot, table_->current_snapshot());
1418+
1419+
auto del_file = MakeEqualityDeleteFile("/delete/del_a.parquet", 1L);
1420+
ICEBERG_UNWRAP_OR_FAIL(auto op, NewOverwriteUpdate());
1421+
EXPECT_THAT(op->AddDelete(del_file), IsOk());
1422+
const int64_t second_snapshot_id = op->GeneratedSnapshotId();
1423+
ICEBERG_UNWRAP_OR_FAIL(auto manifests, op->Apply(*table_->metadata(), first_snapshot));
1424+
ICEBERG_UNWRAP_OR_FAIL(
1425+
auto second_snapshot,
1426+
MakeSyntheticSnapshot(DataOperation::kOverwrite, second_snapshot_id,
1427+
first_snapshot->snapshot_id,
1428+
first_snapshot->sequence_number + 1, manifests));
1429+
1430+
auto metadata = std::make_shared<TableMetadata>(*table_->metadata());
1431+
metadata->snapshots.push_back(second_snapshot);
1432+
metadata->current_snapshot_id = second_snapshot->snapshot_id;
1433+
metadata->last_sequence_number = second_snapshot->sequence_number;
1434+
1435+
DataFileSet replaced_files;
1436+
replaced_files.insert(file_a_);
1437+
// With ignore_equality_deletes=true, concurrently-added equality deletes
1438+
// should NOT cause a conflict.
1439+
EXPECT_THAT(TestMergeAppend::ValidateNoNewDeletesForDataFilesForTest(
1440+
*metadata, first_snapshot->snapshot_id, replaced_files, second_snapshot,
1441+
file_io_, /*ignore_equality_deletes=*/true),
1442+
IsOk());
1443+
}
1444+
14081445
TEST_F(MergingSnapshotUpdateTest,
14091446
ValidateNoNewDeletesForDataFilesUsesConfiguredCaseSensitivity) {
14101447
CommitFileA();

0 commit comments

Comments
 (0)