Skip to content

Commit ae973f6

Browse files
committed
feat: add ReplacePartitions core class and tests (PR1)
Adds iceberg::ReplacePartitions — the dynamic partition overwrite operation. Each AddFile() registers the file's partition for replacement of all existing data files in that partition. Produces an overwrite snapshot with "replace-partitions=true" in the summary. Unpartitioned tables replace all existing data files. Extends MergingSnapshotUpdate (matching Java's BaseReplacePartitions) rather than SnapshotUpdate. This is load-bearing: MergingSnapshotUpdate handles both data and delete manifest pipelines, preserves custom summary properties across commit retry, and exposes the validator helpers needed for conflict detection. An earlier draft extended SnapshotUpdate directly and reimplemented the manifest pipeline by hand; that draft dropped parent delete manifests and lost custom summary entries — both bugs disappear under the merging-update design. Validation is partition-scoped via DropPartition(spec_id, partition): unpartitioned specs naturally use an empty partition value, so the filter manager matches every file in that spec without needing a separate AlwaysTrue path. Validate() uses the PartitionSet overloads of ValidateAddedDataFiles / ValidateNoNewDeleteFiles when partitions are tracked and skips entirely when nothing was staged. Tests (replace_partitions_test.cc, parameterized over format v2/v3 and main/feature branch): * AddFile null / no-spec-id rejection * operation() == "overwrite" * Empty-table replace writes overwrite snapshot with replace-partitions=true * Replace overwrites only the touched partition (existing file in untouched partition survives; existing files in touched partition marked deleted; new file added) * ValidateAppendOnly fails when target partition has data, succeeds when target partition is empty API surface added: * Table::NewReplacePartitions() / StaticTable rejects * Transaction::NewReplacePartitions() * SnapshotSummaryFields::kReplacePartitions = "replace-partitions" * MergingSnapshotUpdate::SetSummaryProperty promoted from private to protected so subclasses can stash custom summary entries through the cached-rebuild path. Closes #637. Tracking: #775
1 parent ed051a7 commit ae973f6

12 files changed

Lines changed: 554 additions & 2 deletions

src/iceberg/CMakeLists.txt

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -104,6 +104,7 @@ set(ICEBERG_SOURCES
104104
update/expire_snapshots.cc
105105
update/fast_append.cc
106106
update/merge_append.cc
107+
update/replace_partitions.cc
107108
update/merging_snapshot_update.cc
108109
update/pending_update.cc
109110
update/row_delta.cc

src/iceberg/snapshot.h

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -251,6 +251,8 @@ struct ICEBERG_EXPORT SnapshotSummaryFields {
251251
inline static const std::string kEngineName = "engine-name";
252252
/// \brief Version of the engine that created the snapshot
253253
inline static const std::string kEngineVersion = "engine-version";
254+
/// \brief Whether this is a replace-partitions operation
255+
inline static const std::string kReplacePartitions = "replace-partitions";
254256
};
255257

256258
/// \brief Helper class for building snapshot summaries.

src/iceberg/table.cc

Lines changed: 11 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -35,6 +35,7 @@
3535
#include "iceberg/update/expire_snapshots.h"
3636
#include "iceberg/update/fast_append.h"
3737
#include "iceberg/update/merge_append.h"
38+
#include "iceberg/update/replace_partitions.h"
3839
#include "iceberg/update/row_delta.h"
3940
#include "iceberg/update/set_snapshot.h"
4041
#include "iceberg/update/snapshot_manager.h"
@@ -238,6 +239,12 @@ Result<std::shared_ptr<RowDelta>> Table::NewRowDelta() {
238239
return RowDelta::Make(name().name, std::move(ctx));
239240
}
240241

242+
Result<std::shared_ptr<ReplacePartitions>> Table::NewReplacePartitions() {
243+
ICEBERG_ASSIGN_OR_RAISE(
244+
auto ctx, TransactionContext::Make(shared_from_this(), TransactionKind::kUpdate));
245+
return ReplacePartitions::Make(name().name, std::move(ctx));
246+
}
247+
241248
Result<std::shared_ptr<UpdateStatistics>> Table::NewUpdateStatistics() {
242249
ICEBERG_ASSIGN_OR_RAISE(
243250
auto ctx, TransactionContext::Make(shared_from_this(), TransactionKind::kUpdate));
@@ -349,6 +356,10 @@ Result<std::shared_ptr<RowDelta>> StaticTable::NewRowDelta() {
349356
return NotSupported("Cannot create a row delta for a static table");
350357
}
351358

359+
Result<std::shared_ptr<ReplacePartitions>> StaticTable::NewReplacePartitions() {
360+
return NotSupported("Cannot create a replace partitions for a static table");
361+
}
362+
352363
Result<std::shared_ptr<SnapshotManager>> StaticTable::NewSnapshotManager() {
353364
return NotSupported("Cannot create a snapshot manager for a static table");
354365
}

src/iceberg/table.h

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -185,6 +185,10 @@ class ICEBERG_EXPORT Table : public std::enable_shared_from_this<Table> {
185185
/// \brief Create a new RowDelta to add rows and row-level deletes.
186186
virtual Result<std::shared_ptr<RowDelta>> NewRowDelta();
187187

188+
/// \brief Create a new ReplacePartitions to dynamically overwrite partitions
189+
/// touched by added data files.
190+
virtual Result<std::shared_ptr<ReplacePartitions>> NewReplacePartitions();
191+
188192
/// \brief Create a new SnapshotManager to manage snapshots and snapshot references.
189193
virtual Result<std::shared_ptr<SnapshotManager>> NewSnapshotManager();
190194

@@ -257,6 +261,7 @@ class ICEBERG_EXPORT StaticTable : public Table {
257261
Result<std::shared_ptr<DeleteFiles>> NewDeleteFiles() override;
258262

259263
Result<std::shared_ptr<RowDelta>> NewRowDelta() override;
264+
Result<std::shared_ptr<ReplacePartitions>> NewReplacePartitions() override;
260265

261266
Result<std::shared_ptr<SnapshotManager>> NewSnapshotManager() override;
262267

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+
replace_partitions_test.cc
233234
row_delta_test.cc
234235
snapshot_manager_test.cc
235236
transaction_test.cc
Lines changed: 275 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,275 @@
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/update/replace_partitions.h"
21+
22+
#include <format>
23+
#include <memory>
24+
#include <string>
25+
#include <tuple>
26+
#include <vector>
27+
28+
#include <gmock/gmock.h>
29+
#include <gtest/gtest.h>
30+
31+
#include "iceberg/avro/avro_register.h"
32+
#include "iceberg/manifest/manifest_entry.h"
33+
#include "iceberg/manifest/manifest_list.h"
34+
#include "iceberg/manifest/manifest_reader.h"
35+
#include "iceberg/partition_spec.h"
36+
#include "iceberg/schema.h"
37+
#include "iceberg/snapshot.h"
38+
#include "iceberg/table.h"
39+
#include "iceberg/table_metadata.h"
40+
#include "iceberg/test/matchers.h"
41+
#include "iceberg/test/test_resource.h"
42+
#include "iceberg/test/update_test_base.h"
43+
#include "iceberg/transaction.h"
44+
#include "iceberg/update/fast_append.h"
45+
#include "iceberg/util/uuid.h"
46+
47+
namespace iceberg {
48+
49+
class ReplacePartitionsTestBase : public UpdateTestBase {
50+
protected:
51+
static void SetUpTestSuite() { avro::RegisterAll(); }
52+
53+
std::string TableName() const override { return "minimal_table"; }
54+
55+
void SetUp() override {
56+
table_ident_ = TableIdentifier{.name = TableName()};
57+
table_location_ = "/warehouse/" + TableName();
58+
59+
InitializeFileIO();
60+
RegisterMinimalTable(format_version());
61+
62+
ICEBERG_UNWRAP_OR_FAIL(spec_, table_->spec());
63+
ICEBERG_UNWRAP_OR_FAIL(schema_, table_->schema());
64+
65+
file_p1_a_ = MakeDataFile("/data/p1_a.parquet", /*partition_x=*/1L);
66+
file_p1_b_ = MakeDataFile("/data/p1_b.parquet", /*partition_x=*/1L);
67+
file_p2_a_ = MakeDataFile("/data/p2_a.parquet", /*partition_x=*/2L);
68+
file_p1_new_ = MakeDataFile("/data/p1_new.parquet", /*partition_x=*/1L);
69+
}
70+
71+
void RegisterMinimalTable(int8_t format_version) {
72+
auto metadata_location = std::format("{}/metadata/00001-{}.metadata.json",
73+
table_location_, Uuid::GenerateV7().ToString());
74+
ICEBERG_UNWRAP_OR_FAIL(
75+
auto metadata, ReadTableMetadataFromResource("TableMetadataV2ValidMinimal.json"));
76+
metadata->format_version = format_version;
77+
metadata->location = table_location_;
78+
metadata->next_row_id = TableMetadata::kInitialRowId;
79+
80+
ASSERT_THAT(TableMetadataUtil::Write(*file_io_, metadata_location, *metadata),
81+
IsOk());
82+
ICEBERG_UNWRAP_OR_FAIL(table_,
83+
catalog_->RegisterTable(table_ident_, metadata_location));
84+
}
85+
86+
virtual int8_t format_version() const {
87+
return TableMetadata::kDefaultTableFormatVersion;
88+
}
89+
90+
virtual std::string branch() const { return std::string(SnapshotRef::kMainBranch); }
91+
92+
std::shared_ptr<DataFile> MakeDataFile(const std::string& path, int64_t partition_x) {
93+
return MakeDataFile(path, spec_, {Literal::Long(partition_x)});
94+
}
95+
96+
std::shared_ptr<DataFile> MakeDataFile(const std::string& path,
97+
std::shared_ptr<PartitionSpec> spec,
98+
std::vector<Literal> partition_values) {
99+
auto file = std::make_shared<DataFile>();
100+
file->content = DataFile::Content::kData;
101+
file->file_path = table_location_ + path;
102+
file->file_format = FileFormatType::kParquet;
103+
file->partition = PartitionValues(std::move(partition_values));
104+
file->file_size_in_bytes = 1024;
105+
file->record_count = 100;
106+
file->partition_spec_id = spec->spec_id();
107+
return file;
108+
}
109+
110+
Result<std::shared_ptr<ReplacePartitions>> NewBranchReplacePartitions() {
111+
ICEBERG_ASSIGN_OR_RAISE(auto replace, table_->NewReplacePartitions());
112+
replace->SetTargetBranch(branch());
113+
return replace;
114+
}
115+
116+
Result<std::shared_ptr<FastAppend>> NewBranchFastAppend() {
117+
ICEBERG_ASSIGN_OR_RAISE(auto append, table_->NewFastAppend());
118+
append->SetTargetBranch(branch());
119+
return append;
120+
}
121+
122+
Result<std::shared_ptr<Snapshot>> CurrentSnapshot() {
123+
const auto& refs = table_->metadata()->refs;
124+
auto it = refs.find(branch());
125+
if (it == refs.end()) {
126+
return NotFound("No snapshot ref for branch '{}'", branch());
127+
}
128+
return table_->SnapshotById(it->second->snapshot_id);
129+
}
130+
131+
Result<std::vector<ManifestFile>> CurrentDataManifests() {
132+
ICEBERG_ASSIGN_OR_RAISE(auto snapshot, CurrentSnapshot());
133+
SnapshotCache snapshot_cache(snapshot.get());
134+
ICEBERG_ASSIGN_OR_RAISE(auto manifests, snapshot_cache.DataManifests(file_io_));
135+
return std::vector<ManifestFile>(manifests.begin(), manifests.end());
136+
}
137+
138+
std::shared_ptr<PartitionSpec> spec_;
139+
std::shared_ptr<Schema> schema_;
140+
std::shared_ptr<DataFile> file_p1_a_;
141+
std::shared_ptr<DataFile> file_p1_b_;
142+
std::shared_ptr<DataFile> file_p2_a_;
143+
std::shared_ptr<DataFile> file_p1_new_;
144+
};
145+
146+
class ReplacePartitionsTest
147+
: public ReplacePartitionsTestBase,
148+
public ::testing::WithParamInterface<std::tuple<int8_t, std::string>> {
149+
protected:
150+
int8_t format_version() const override { return std::get<0>(GetParam()); }
151+
std::string branch() const override { return std::get<1>(GetParam()); }
152+
};
153+
154+
TEST_F(ReplacePartitionsTestBase, AddNullFile) {
155+
ICEBERG_UNWRAP_OR_FAIL(auto replace, NewBranchReplacePartitions());
156+
replace->AddFile(nullptr);
157+
158+
auto result = replace->Commit();
159+
EXPECT_THAT(result, IsError(ErrorKind::kValidationFailed));
160+
}
161+
162+
TEST_F(ReplacePartitionsTestBase, AddFileWithoutSpecIdRejected) {
163+
auto file = MakeDataFile("/data/no_spec.parquet", /*partition_x=*/1L);
164+
file->partition_spec_id = std::nullopt;
165+
166+
ICEBERG_UNWRAP_OR_FAIL(auto replace, NewBranchReplacePartitions());
167+
replace->AddFile(file);
168+
169+
auto result = replace->Commit();
170+
EXPECT_THAT(result, IsError(ErrorKind::kValidationFailed));
171+
}
172+
173+
TEST_P(ReplacePartitionsTest, OperationIsOverwrite) {
174+
ICEBERG_UNWRAP_OR_FAIL(auto replace, NewBranchReplacePartitions());
175+
EXPECT_EQ(replace->operation(), DataOperation::kOverwrite);
176+
}
177+
178+
TEST_P(ReplacePartitionsTest, EmptyTableReplaceWritesOverwriteSnapshot) {
179+
EXPECT_THAT(CurrentSnapshot(), IsError(ErrorKind::kNotFound));
180+
181+
ICEBERG_UNWRAP_OR_FAIL(auto replace, NewBranchReplacePartitions());
182+
replace->AddFile(file_p1_a_).AddFile(file_p2_a_);
183+
EXPECT_THAT(replace->Commit(), IsOk());
184+
185+
EXPECT_THAT(table_->Refresh(), IsOk());
186+
ICEBERG_UNWRAP_OR_FAIL(auto snapshot, CurrentSnapshot());
187+
EXPECT_EQ(snapshot->Operation(), DataOperation::kOverwrite);
188+
EXPECT_EQ(snapshot->summary.at(SnapshotSummaryFields::kReplacePartitions), "true");
189+
EXPECT_EQ(snapshot->summary.at(SnapshotSummaryFields::kAddedDataFiles), "2");
190+
191+
ICEBERG_UNWRAP_OR_FAIL(auto manifests, CurrentDataManifests());
192+
ASSERT_EQ(manifests.size(), 1U);
193+
}
194+
195+
TEST_P(ReplacePartitionsTest, ReplaceOverwritesOnlyTouchedPartitions) {
196+
// Seed the table with two partitions worth of data.
197+
{
198+
ICEBERG_UNWRAP_OR_FAIL(auto append, NewBranchFastAppend());
199+
append->AppendFile(file_p1_a_).AppendFile(file_p1_b_).AppendFile(file_p2_a_);
200+
EXPECT_THAT(append->Commit(), IsOk());
201+
EXPECT_THAT(table_->Refresh(), IsOk());
202+
}
203+
204+
// Now replace partition 1 only.
205+
ICEBERG_UNWRAP_OR_FAIL(auto replace, NewBranchReplacePartitions());
206+
replace->AddFile(file_p1_new_);
207+
EXPECT_THAT(replace->Commit(), IsOk());
208+
EXPECT_THAT(table_->Refresh(), IsOk());
209+
210+
ICEBERG_UNWRAP_OR_FAIL(auto snapshot, CurrentSnapshot());
211+
EXPECT_EQ(snapshot->Operation(), DataOperation::kOverwrite);
212+
EXPECT_EQ(snapshot->summary.at(SnapshotSummaryFields::kReplacePartitions), "true");
213+
EXPECT_EQ(snapshot->summary.at(SnapshotSummaryFields::kAddedDataFiles), "1");
214+
// file_p1_a_ and file_p1_b_ should be marked deleted; file_p2_a_ should not.
215+
EXPECT_EQ(snapshot->summary.at(SnapshotSummaryFields::kDeletedDataFiles), "2");
216+
217+
// Verify the surviving live data file set: file_p2_a_ + file_p1_new_.
218+
ICEBERG_UNWRAP_OR_FAIL(auto manifests, CurrentDataManifests());
219+
std::unordered_set<std::string> live_paths;
220+
for (const auto& manifest : manifests) {
221+
ICEBERG_UNWRAP_OR_FAIL(
222+
auto spec, table_->metadata()->PartitionSpecById(manifest.partition_spec_id));
223+
ICEBERG_UNWRAP_OR_FAIL(auto reader,
224+
ManifestReader::Make(manifest, file_io_, schema_, spec));
225+
ICEBERG_UNWRAP_OR_FAIL(auto entries, reader->Entries());
226+
for (const auto& entry : entries) {
227+
if (entry.status != ManifestStatus::kDeleted) {
228+
live_paths.insert(entry.data_file->file_path);
229+
}
230+
}
231+
}
232+
EXPECT_THAT(live_paths, ::testing::UnorderedElementsAre(file_p2_a_->file_path,
233+
file_p1_new_->file_path));
234+
}
235+
236+
TEST_P(ReplacePartitionsTest, ValidateAppendOnlyFailsWhenPartitionHasData) {
237+
// Seed partition 1 with a file.
238+
{
239+
ICEBERG_UNWRAP_OR_FAIL(auto append, NewBranchFastAppend());
240+
append->AppendFile(file_p1_a_);
241+
EXPECT_THAT(append->Commit(), IsOk());
242+
EXPECT_THAT(table_->Refresh(), IsOk());
243+
}
244+
245+
// ValidateAppendOnly should reject replacing partition 1 because it already
246+
// contains data.
247+
ICEBERG_UNWRAP_OR_FAIL(auto replace, NewBranchReplacePartitions());
248+
replace->ValidateAppendOnly();
249+
replace->AddFile(file_p1_new_);
250+
EXPECT_THAT(replace->Commit(), IsError(ErrorKind::kValidationFailed));
251+
}
252+
253+
TEST_P(ReplacePartitionsTest, ValidateAppendOnlyAllowsAddToEmptyPartition) {
254+
// Seed partition 1 only.
255+
{
256+
ICEBERG_UNWRAP_OR_FAIL(auto append, NewBranchFastAppend());
257+
append->AppendFile(file_p1_a_);
258+
EXPECT_THAT(append->Commit(), IsOk());
259+
EXPECT_THAT(table_->Refresh(), IsOk());
260+
}
261+
262+
// Replacing the empty partition 2 with append-only validation should succeed.
263+
ICEBERG_UNWRAP_OR_FAIL(auto replace, NewBranchReplacePartitions());
264+
replace->ValidateAppendOnly();
265+
replace->AddFile(file_p2_a_);
266+
EXPECT_THAT(replace->Commit(), IsOk());
267+
}
268+
269+
INSTANTIATE_TEST_SUITE_P(
270+
FormatVersionAndBranch, ReplacePartitionsTest,
271+
::testing::Combine(::testing::Values(2, 3),
272+
::testing::Values(std::string(SnapshotRef::kMainBranch),
273+
std::string("feature"))));
274+
275+
} // namespace iceberg

src/iceberg/transaction.cc

Lines changed: 8 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -37,6 +37,7 @@
3737
#include "iceberg/update/fast_append.h"
3838
#include "iceberg/update/merge_append.h"
3939
#include "iceberg/update/pending_update.h"
40+
#include "iceberg/update/replace_partitions.h"
4041
#include "iceberg/update/row_delta.h"
4142
#include "iceberg/update/set_snapshot.h"
4243
#include "iceberg/update/snapshot_manager.h"
@@ -513,6 +514,13 @@ Result<std::shared_ptr<RowDelta>> Transaction::NewRowDelta() {
513514
return row_delta;
514515
}
515516

517+
Result<std::shared_ptr<ReplacePartitions>> Transaction::NewReplacePartitions() {
518+
ICEBERG_ASSIGN_OR_RAISE(std::shared_ptr<ReplacePartitions> replace_partitions,
519+
ReplacePartitions::Make(ctx_->table->name().name, ctx_));
520+
ICEBERG_RETURN_UNEXPECTED(AddUpdate(replace_partitions));
521+
return replace_partitions;
522+
}
523+
516524
Result<std::shared_ptr<UpdateStatistics>> Transaction::NewUpdateStatistics() {
517525
ICEBERG_ASSIGN_OR_RAISE(std::shared_ptr<UpdateStatistics> update_statistics,
518526
UpdateStatistics::Make(ctx_));

src/iceberg/transaction.h

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -115,6 +115,10 @@ class ICEBERG_EXPORT Transaction : public std::enable_shared_from_this<Transacti
115115
/// \brief Create a new RowDelta to add rows and row-level deletes.
116116
Result<std::shared_ptr<RowDelta>> NewRowDelta();
117117

118+
/// \brief Create a new ReplacePartitions to dynamically overwrite partitions
119+
/// touched by added data files.
120+
Result<std::shared_ptr<ReplacePartitions>> NewReplacePartitions();
121+
118122
/// \brief Create a new SnapshotManager to manage snapshots.
119123
Result<std::shared_ptr<SnapshotManager>> NewSnapshotManager();
120124

src/iceberg/type_fwd.h

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -241,6 +241,7 @@ class TransactionContext;
241241
class DeleteFiles;
242242
class ExpireSnapshots;
243243
class FastAppend;
244+
class ReplacePartitions;
244245
class MergeAppend;
245246
class PendingUpdate;
246247
class RowDelta;

0 commit comments

Comments
 (0)