Skip to content

Commit 8e833d2

Browse files
committed
feat: add rewrite manifests
1 parent 131f976 commit 8e833d2

19 files changed

Lines changed: 2689 additions & 6 deletions

src/iceberg/CMakeLists.txt

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -110,6 +110,7 @@ set(ICEBERG_SOURCES
110110
update/overwrite_files.cc
111111
update/pending_update.cc
112112
update/rewrite_files.cc
113+
update/rewrite_manifests.cc
113114
update/row_delta.cc
114115
update/set_snapshot.cc
115116
update/snapshot_manager.cc

src/iceberg/manifest/rolling_manifest_writer.cc

Lines changed: 7 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -54,6 +54,13 @@ Status RollingManifestWriter::WriteExistingEntry(
5454
return {};
5555
}
5656

57+
Status RollingManifestWriter::WriteExistingEntry(const ManifestEntry& entry) {
58+
ICEBERG_ASSIGN_OR_RAISE(auto* writer, CurrentWriter());
59+
ICEBERG_RETURN_UNEXPECTED(writer->WriteExistingEntry(entry));
60+
current_file_rows_++;
61+
return {};
62+
}
63+
5764
Status RollingManifestWriter::WriteDeletedEntry(
5865
std::shared_ptr<DataFile> file, int64_t data_sequence_number,
5966
std::optional<int64_t> file_sequence_number) {

src/iceberg/manifest/rolling_manifest_writer.h

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -76,6 +76,9 @@ class ICEBERG_EXPORT RollingManifestWriter {
7676
int64_t data_sequence_number,
7777
std::optional<int64_t> file_sequence_number = std::nullopt);
7878

79+
/// \brief Add an existing entry while preserving snapshot and sequence fields.
80+
Status WriteExistingEntry(const ManifestEntry& entry);
81+
7982
/// \brief Add a delete entry for a file.
8083
///
8184
/// \param file a deleted data file

src/iceberg/manifest/v3_metadata.cc

Lines changed: 10 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -194,13 +194,20 @@ Status ManifestFileAdapterV3::Init() {
194194
}
195195

196196
Status ManifestFileAdapterV3::Append(const ManifestFile& file) {
197-
ICEBERG_RETURN_UNEXPECTED(AppendInternal(file));
198197
if (WrapFirstRowId(file)) {
199198
if (!next_row_id_.has_value()) {
200199
return InvalidManifestList("Missing next-row-id for file: {}", file.manifest_path);
201200
}
202-
next_row_id_ = next_row_id_.value() + file.existing_rows_count.value_or(0) +
203-
file.added_rows_count.value_or(0);
201+
if (!file.existing_rows_count.has_value() || !file.added_rows_count.has_value()) {
202+
return InvalidManifestList("Missing row counts for file: {}", file.manifest_path);
203+
}
204+
}
205+
206+
ICEBERG_RETURN_UNEXPECTED(AppendInternal(file));
207+
208+
if (WrapFirstRowId(file)) {
209+
next_row_id_ = next_row_id_.value() + file.existing_rows_count.value() +
210+
file.added_rows_count.value();
204211
}
205212
return {};
206213
}

src/iceberg/meson.build

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -135,6 +135,7 @@ iceberg_sources = files(
135135
'update/overwrite_files.cc',
136136
'update/pending_update.cc',
137137
'update/rewrite_files.cc',
138+
'update/rewrite_manifests.cc',
138139
'update/row_delta.cc',
139140
'update/set_snapshot.cc',
140141
'update/snapshot_manager.cc',

src/iceberg/table.cc

Lines changed: 11 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -37,6 +37,7 @@
3737
#include "iceberg/update/merge_append.h"
3838
#include "iceberg/update/overwrite_files.h"
3939
#include "iceberg/update/rewrite_files.h"
40+
#include "iceberg/update/rewrite_manifests.h"
4041
#include "iceberg/update/row_delta.h"
4142
#include "iceberg/update/set_snapshot.h"
4243
#include "iceberg/update/snapshot_manager.h"
@@ -252,6 +253,12 @@ Result<std::shared_ptr<RewriteFiles>> Table::NewRewriteFiles() {
252253
return RewriteFiles::Make(name().name, std::move(ctx));
253254
}
254255

256+
Result<std::shared_ptr<RewriteManifests>> Table::NewRewriteManifests() {
257+
ICEBERG_ASSIGN_OR_RAISE(
258+
auto ctx, TransactionContext::Make(shared_from_this(), TransactionKind::kUpdate));
259+
return RewriteManifests::Make(name().name, std::move(ctx));
260+
}
261+
255262
Result<std::shared_ptr<UpdateStatistics>> Table::NewUpdateStatistics() {
256263
ICEBERG_ASSIGN_OR_RAISE(
257264
auto ctx, TransactionContext::Make(shared_from_this(), TransactionKind::kUpdate));
@@ -371,6 +378,10 @@ Result<std::shared_ptr<RewriteFiles>> StaticTable::NewRewriteFiles() {
371378
return NotSupported("Cannot create a rewrite files for a static table");
372379
}
373380

381+
Result<std::shared_ptr<RewriteManifests>> StaticTable::NewRewriteManifests() {
382+
return NotSupported("Cannot create a rewrite manifests for a static table");
383+
}
384+
374385
Result<std::shared_ptr<SnapshotManager>> StaticTable::NewSnapshotManager() {
375386
return NotSupported("Cannot create a snapshot manager for a static table");
376387
}

src/iceberg/table.h

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -192,6 +192,9 @@ class ICEBERG_EXPORT Table : public std::enable_shared_from_this<Table> {
192192
/// changes.
193193
virtual Result<std::shared_ptr<RewriteFiles>> NewRewriteFiles();
194194

195+
/// \brief Create a new RewriteManifests to rewrite manifest layout.
196+
virtual Result<std::shared_ptr<RewriteManifests>> NewRewriteManifests();
197+
195198
/// \brief Create a new SnapshotManager to manage snapshots and snapshot references.
196199
virtual Result<std::shared_ptr<SnapshotManager>> NewSnapshotManager();
197200

@@ -269,6 +272,8 @@ class ICEBERG_EXPORT StaticTable : public Table {
269272

270273
Result<std::shared_ptr<RewriteFiles>> NewRewriteFiles() override;
271274

275+
Result<std::shared_ptr<RewriteManifests>> NewRewriteManifests() override;
276+
272277
Result<std::shared_ptr<SnapshotManager>> NewSnapshotManager() override;
273278

274279
private:

src/iceberg/test/CMakeLists.txt

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -236,6 +236,7 @@ if(ICEBERG_BUILD_BUNDLE)
236236
merging_snapshot_update_test.cc
237237
name_mapping_update_test.cc
238238
rewrite_files_test.cc
239+
rewrite_manifests_test.cc
239240
row_delta_test.cc
240241
snapshot_manager_test.cc
241242
transaction_test.cc

0 commit comments

Comments
 (0)