Skip to content

Commit 7a3a846

Browse files
WZhuowgtmac
authored andcommitted
feat: implement RewriteFiles operation
1 parent c65a806 commit 7a3a846

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
@@ -108,6 +108,7 @@ set(ICEBERG_SOURCES
108108
update/merging_snapshot_update.cc
109109
update/overwrite_files.cc
110110
update/pending_update.cc
111+
update/rewrite_files.cc
111112
update/row_delta.cc
112113
update/set_snapshot.cc
113114
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
@@ -234,6 +234,7 @@ if(ICEBERG_BUILD_BUNDLE)
234234
merge_append_test.cc
235235
merging_snapshot_update_test.cc
236236
name_mapping_update_test.cc
237+
rewrite_files_test.cc
237238
row_delta_test.cc
238239
snapshot_manager_test.cc
239240
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
@@ -1450,6 +1450,43 @@ TEST_F(MergingSnapshotUpdateTest,
14501450
IsError(ErrorKind::kValidationFailed));
14511451
}
14521452

1453+
TEST_F(MergingSnapshotUpdateTest,
1454+
ValidateNoNewDeletesForDataFilesIgnoresEqualityDeletesWhenFlagIsTrue) {
1455+
// This tests the behavior that RewriteFiles::SetDataSequenceNumber() and
1456+
// RewriteFiles::RewriteDataFiles() enable: when a data sequence number is
1457+
// set for rewritten data files, concurrent equality deletes at higher
1458+
// sequence numbers still apply to the new files and are NOT a conflict.
1459+
// Only position deletes should still fail (tested separately by
1460+
// ValidateNoNewDeletesForDataFilesFailsOnPositionDeleteWhenIgnoringEqualityDeletes).
1461+
CommitFileA();
1462+
ICEBERG_UNWRAP_OR_FAIL(auto first_snapshot, table_->current_snapshot());
1463+
1464+
auto del_file = MakeEqualityDeleteFile("/delete/del_a.parquet", 1L);
1465+
ICEBERG_UNWRAP_OR_FAIL(auto op, NewOverwriteUpdate());
1466+
EXPECT_THAT(op->AddDelete(del_file), IsOk());
1467+
const int64_t second_snapshot_id = op->GeneratedSnapshotId();
1468+
ICEBERG_UNWRAP_OR_FAIL(auto manifests, op->Apply(*table_->metadata(), first_snapshot));
1469+
ICEBERG_UNWRAP_OR_FAIL(
1470+
auto second_snapshot,
1471+
MakeSyntheticSnapshot(DataOperation::kOverwrite, second_snapshot_id,
1472+
first_snapshot->snapshot_id,
1473+
first_snapshot->sequence_number + 1, manifests));
1474+
1475+
auto metadata = std::make_shared<TableMetadata>(*table_->metadata());
1476+
metadata->snapshots.push_back(second_snapshot);
1477+
metadata->current_snapshot_id = second_snapshot->snapshot_id;
1478+
metadata->last_sequence_number = second_snapshot->sequence_number;
1479+
1480+
DataFileSet replaced_files;
1481+
replaced_files.insert(file_a_);
1482+
// With ignore_equality_deletes=true, concurrently-added equality deletes
1483+
// should NOT cause a conflict.
1484+
EXPECT_THAT(TestMergeAppend::ValidateNoNewDeletesForDataFilesForTest(
1485+
*metadata, first_snapshot->snapshot_id, replaced_files, second_snapshot,
1486+
file_io_, /*ignore_equality_deletes=*/true),
1487+
IsOk());
1488+
}
1489+
14531490
TEST_F(MergingSnapshotUpdateTest,
14541491
ValidateNoNewDeletesForDataFilesUsesConfiguredCaseSensitivity) {
14551492
CommitFileA();

0 commit comments

Comments
 (0)