Skip to content

Commit e76aa6e

Browse files
committed
feat: implement RewriteFiles operation
1 parent fc9781f commit e76aa6e

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
@@ -103,6 +103,7 @@ set(ICEBERG_SOURCES
103103
update/fast_append.cc
104104
update/merging_snapshot_update.cc
105105
update/pending_update.cc
106+
update/rewrite_files.cc
106107
update/set_snapshot.cc
107108
update/snapshot_manager.cc
108109
update/snapshot_update.cc

src/iceberg/table.cc

Lines changed: 7 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -33,6 +33,7 @@
3333
#include "iceberg/transaction.h"
3434
#include "iceberg/update/expire_snapshots.h"
3535
#include "iceberg/update/fast_append.h"
36+
#include "iceberg/update/rewrite_files.h"
3637
#include "iceberg/update/set_snapshot.h"
3738
#include "iceberg/update/snapshot_manager.h"
3839
#include "iceberg/update/update_location.h"
@@ -217,6 +218,12 @@ Result<std::shared_ptr<FastAppend>> Table::NewFastAppend() {
217218
return FastAppend::Make(name().name, std::move(ctx));
218219
}
219220

221+
Result<std::shared_ptr<RewriteFiles>> Table::NewRewriteFiles() {
222+
ICEBERG_ASSIGN_OR_RAISE(
223+
auto ctx, TransactionContext::Make(shared_from_this(), TransactionKind::kUpdate));
224+
return RewriteFiles::Make(name().name, std::move(ctx));
225+
}
226+
220227
Result<std::shared_ptr<UpdateStatistics>> Table::NewUpdateStatistics() {
221228
ICEBERG_ASSIGN_OR_RAISE(
222229
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
@@ -176,6 +176,10 @@ class ICEBERG_EXPORT Table : public std::enable_shared_from_this<Table> {
176176
/// \brief Create a new FastAppend to append data files and commit the changes.
177177
virtual Result<std::shared_ptr<FastAppend>> NewFastAppend();
178178

179+
/// \brief Create a new RewriteFiles to replace files in this table and commit the
180+
/// changes.
181+
virtual Result<std::shared_ptr<RewriteFiles>> NewRewriteFiles();
182+
179183
/// \brief Create a new SnapshotManager to manage snapshots and snapshot references.
180184
virtual Result<std::shared_ptr<SnapshotManager>> NewSnapshotManager();
181185

src/iceberg/test/CMakeLists.txt

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -227,6 +227,7 @@ if(ICEBERG_BUILD_BUNDLE)
227227
manifest_filter_manager_test.cc
228228
merging_snapshot_update_test.cc
229229
name_mapping_update_test.cc
230+
rewrite_files_test.cc
230231
snapshot_manager_test.cc
231232
transaction_test.cc
232233
update_location_test.cc

src/iceberg/test/merging_snapshot_update_test.cc

Lines changed: 37 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1373,6 +1373,43 @@ TEST_F(MergingSnapshotUpdateTest,
13731373
IsError(ErrorKind::kValidationFailed));
13741374
}
13751375

1376+
TEST_F(MergingSnapshotUpdateTest,
1377+
ValidateNoNewDeletesForDataFilesIgnoresEqualityDeletesWhenFlagIsTrue) {
1378+
// This tests the behavior that RewriteFiles::SetDataSequenceNumber() and
1379+
// RewriteFiles::RewriteDataFiles() enable: when a data sequence number is
1380+
// set for rewritten data files, concurrent equality deletes at higher
1381+
// sequence numbers still apply to the new files and are NOT a conflict.
1382+
// Only position deletes should still fail (tested separately by
1383+
// ValidateNoNewDeletesForDataFilesFailsOnPositionDeleteWhenIgnoringEqualityDeletes).
1384+
CommitFileA();
1385+
ICEBERG_UNWRAP_OR_FAIL(auto first_snapshot, table_->current_snapshot());
1386+
1387+
auto del_file = MakeEqualityDeleteFile("/delete/del_a.parquet", 1L);
1388+
ICEBERG_UNWRAP_OR_FAIL(auto op, NewOverwriteUpdate());
1389+
EXPECT_THAT(op->AddDelete(del_file), IsOk());
1390+
const int64_t second_snapshot_id = op->GeneratedSnapshotId();
1391+
ICEBERG_UNWRAP_OR_FAIL(auto manifests, op->Apply(*table_->metadata(), first_snapshot));
1392+
ICEBERG_UNWRAP_OR_FAIL(
1393+
auto second_snapshot,
1394+
MakeSyntheticSnapshot(DataOperation::kOverwrite, second_snapshot_id,
1395+
first_snapshot->snapshot_id,
1396+
first_snapshot->sequence_number + 1, manifests));
1397+
1398+
auto metadata = std::make_shared<TableMetadata>(*table_->metadata());
1399+
metadata->snapshots.push_back(second_snapshot);
1400+
metadata->current_snapshot_id = second_snapshot->snapshot_id;
1401+
metadata->last_sequence_number = second_snapshot->sequence_number;
1402+
1403+
DataFileSet replaced_files;
1404+
replaced_files.insert(file_a_);
1405+
// With ignore_equality_deletes=true, concurrently-added equality deletes
1406+
// should NOT cause a conflict.
1407+
EXPECT_THAT(TestMergeAppend::ValidateNoNewDeletesForDataFilesForTest(
1408+
*metadata, first_snapshot->snapshot_id, replaced_files, second_snapshot,
1409+
file_io_, /*ignore_equality_deletes=*/true),
1410+
IsOk());
1411+
}
1412+
13761413
TEST_F(MergingSnapshotUpdateTest,
13771414
ValidateNoNewDeletesForDataFilesUsesConfiguredCaseSensitivity) {
13781415
CommitFileA();

0 commit comments

Comments
 (0)