Skip to content

Commit 868d58c

Browse files
wgtmacwgtmac
committed
refactor code to align more with java style
1 parent 3f492a6 commit 868d58c

5 files changed

Lines changed: 377 additions & 250 deletions

File tree

‎src/iceberg/manifest/v3_metadata.cc‎

Lines changed: 11 additions & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -196,21 +196,21 @@ Status ManifestFileAdapterV3::Init() {
196196
}
197197

198198
Status ManifestFileAdapterV3::Append(const ManifestFile& file) {
199-
if (WrapFirstRowId(file)) {
200-
if (!next_row_id_.has_value()) {
201-
return InvalidManifestList("Missing next-row-id for file: {}", file.manifest_path);
202-
}
203-
if (!file.existing_rows_count.has_value() || !file.added_rows_count.has_value()) {
204-
return InvalidManifestList("Missing row counts for file: {}", file.manifest_path);
205-
}
199+
if (!WrapFirstRowId(file)) {
200+
return AppendInternal(file);
201+
}
202+
203+
if (!next_row_id_.has_value()) {
204+
return InvalidManifestList("Missing next-row-id for file: {}", file.manifest_path);
205+
}
206+
if (!file.existing_rows_count.has_value() || !file.added_rows_count.has_value()) {
207+
return InvalidManifestList("Missing row counts for file: {}", file.manifest_path);
206208
}
207209

208210
ICEBERG_RETURN_UNEXPECTED(AppendInternal(file));
209211

210-
if (WrapFirstRowId(file)) {
211-
next_row_id_ = next_row_id_.value() + file.existing_rows_count.value() +
212-
file.added_rows_count.value();
213-
}
212+
next_row_id_ = next_row_id_.value() + file.existing_rows_count.value() +
213+
file.added_rows_count.value();
214214
return {};
215215
}
216216

‎src/iceberg/test/rewrite_manifests_test.cc‎

Lines changed: 108 additions & 28 deletions
Original file line numberDiff line numberDiff line change
@@ -53,6 +53,7 @@
5353
#include "iceberg/update/delete_files.h"
5454
#include "iceberg/update/fast_append.h"
5555
#include "iceberg/update/row_delta.h"
56+
#include "iceberg/update/snapshot_manager.h"
5657
#include "iceberg/update/update_partition_spec.h"
5758
#include "iceberg/update/update_properties.h"
5859
#include "iceberg/update/update_schema.h"
@@ -180,6 +181,51 @@ class ScopedManifestReadTracking {
180181
ReaderFactory previous_;
181182
};
182183

184+
class FailingManifestFlushWriter : public Writer {
185+
public:
186+
explicit FailingManifestFlushWriter(std::unique_ptr<Writer> delegate)
187+
: delegate_(std::move(delegate)) {}
188+
189+
Status Open(const WriterOptions& options) override { return delegate_->Open(options); }
190+
191+
Status Close() override { return delegate_->Close(); }
192+
193+
Status Write(ArrowArray* data) override {
194+
if (data->release != nullptr) {
195+
data->release(data);
196+
}
197+
return IOError("Injected rewrite manifest flush failure");
198+
}
199+
200+
Result<Metrics> metrics() override { return delegate_->metrics(); }
201+
202+
Result<int64_t> length() override { return delegate_->length(); }
203+
204+
std::vector<int64_t> split_offsets() override { return delegate_->split_offsets(); }
205+
206+
private:
207+
std::unique_ptr<Writer> delegate_;
208+
};
209+
210+
class ScopedManifestFlushFailure {
211+
public:
212+
ScopedManifestFlushFailure()
213+
: previous_(WriterFactoryRegistry::GetFactory(FileFormatType::kAvro)) {
214+
WriterFactoryRegistry::GetFactory(FileFormatType::kAvro) =
215+
[factory = previous_]() -> Result<std::unique_ptr<Writer>> {
216+
ICEBERG_ASSIGN_OR_RAISE(auto writer, factory());
217+
return std::make_unique<FailingManifestFlushWriter>(std::move(writer));
218+
};
219+
}
220+
221+
~ScopedManifestFlushFailure() {
222+
WriterFactoryRegistry::GetFactory(FileFormatType::kAvro) = std::move(previous_);
223+
}
224+
225+
private:
226+
WriterFactory previous_;
227+
};
228+
183229
struct ManifestWriteFailureState {
184230
int manifest_output_count = 0;
185231
int fail_on_manifest_output = 0;
@@ -1284,29 +1330,47 @@ TEST_P(RewriteManifestsTest, ClosesOpenRewriteWritersAfterApplyFailure) {
12841330

12851331
TEST_P(RewriteManifestsTest, ClosesAllRewriteWritersAfterCloseFailure) {
12861332
ASSERT_THAT(AppendFiles({file_a_, file_b_, file_c_}), IsOk());
1333+
ICEBERG_UNWRAP_OR_FAIL(auto before, CurrentManifests());
12871334

1288-
auto failure_state = std::make_shared<ManifestWriteFailureState>();
1289-
failure_state->fail_on_manifest_close = 1;
1290-
BindTableWithFileIO(std::make_shared<ManifestWriteFailureFileIO>(
1291-
file_io_, table_location_ + "/metadata/", failure_state));
1292-
1293-
ICEBERG_UNWRAP_OR_FAIL(auto rewrite, table_->NewRewriteManifests());
1294-
rewrite->ClusterBy([](const DataFile& file) { return file.file_path; });
1295-
1296-
auto result = rewrite->Commit();
1297-
EXPECT_THAT(result, IsError(ErrorKind::kIOError));
1298-
EXPECT_THAT(result, HasErrorMessage("Injected rewrite manifest close failure"));
1335+
enum class Failure { kFlush, kClose, kException };
1336+
for (auto failure : {Failure::kFlush, Failure::kClose, Failure::kException}) {
1337+
SCOPED_TRACE(static_cast<int>(failure));
1338+
auto state = std::make_shared<ManifestWriteFailureState>();
1339+
state->fail_on_manifest_close = failure == Failure::kClose ? 1 : 0;
1340+
state->throw_on_manifest_close = failure == Failure::kException ? 1 : 0;
1341+
BindTableWithFileIO(std::make_shared<ManifestWriteFailureFileIO>(
1342+
file_io_, table_location_ + "/metadata/", state));
1343+
std::unique_ptr<ScopedManifestFlushFailure> flush_failure;
1344+
if (failure == Failure::kFlush) {
1345+
flush_failure = std::make_unique<ScopedManifestFlushFailure>();
1346+
}
12991347

1300-
ASSERT_EQ(failure_state->manifest_outputs.size(), 3U);
1301-
ASSERT_EQ(failure_state->manifest_close_attempts.size(), 3U);
1302-
EXPECT_THAT(failure_state->manifest_close_attempts,
1303-
::testing::UnorderedElementsAre(failure_state->manifest_outputs[0],
1304-
failure_state->manifest_outputs[1],
1305-
failure_state->manifest_outputs[2]));
1306-
for (size_t i = 0; i < failure_state->manifest_close_attempts.size(); ++i) {
1307-
const auto& closed_manifest_path = failure_state->manifest_close_attempts[i];
1308-
EXPECT_THAT(failure_state->deleted_files, ::testing::Contains(closed_manifest_path));
1309-
EXPECT_FALSE(FileExists(closed_manifest_path));
1348+
ICEBERG_UNWRAP_OR_FAIL(auto rewrite, table_->NewRewriteManifests());
1349+
rewrite->ClusterBy([](const DataFile& file) { return file.file_path; });
1350+
1351+
auto result = rewrite->Commit();
1352+
EXPECT_THAT(result,
1353+
IsError(failure == Failure::kException ? ErrorKind::kValidationFailed
1354+
: ErrorKind::kIOError));
1355+
EXPECT_THAT(result,
1356+
HasErrorMessage(failure == Failure::kFlush
1357+
? "Injected rewrite manifest flush failure"
1358+
: failure == Failure::kClose
1359+
? "Injected rewrite manifest close failure"
1360+
: "Injected rewrite manifest close exception"));
1361+
1362+
ASSERT_EQ(state->manifest_outputs.size(), 3U);
1363+
EXPECT_THAT(state->manifest_close_attempts,
1364+
::testing::UnorderedElementsAreArray(state->manifest_outputs));
1365+
// The update is still alive: cleanup must not rely on its destructor.
1366+
EXPECT_TRUE(state->open_manifest_paths.empty());
1367+
EXPECT_TRUE(state->deleted_open_files.empty());
1368+
for (const auto& path : state->manifest_outputs) {
1369+
EXPECT_THAT(state->deleted_files, ::testing::Contains(path));
1370+
EXPECT_FALSE(FileExists(path));
1371+
}
1372+
ICEBERG_UNWRAP_OR_FAIL(auto after, CurrentManifests());
1373+
EXPECT_EQ(after, before);
13101374
}
13111375
}
13121376

@@ -2393,19 +2457,35 @@ TEST_P(RewriteManifestsTest, ManifestReplacementFailureWithSnapshotIdInheritance
23932457
EXPECT_TRUE(FileExists(new_manifest.manifest_path));
23942458
}
23952459

2396-
TEST_P(RewriteManifestsTest, RewriteManifestsOnBranchUnsupported) {
2460+
TEST_P(RewriteManifestsTest, RewriteManifestsOnBranch) {
23972461
ASSERT_THAT(AppendFiles({file_a_, file_b_}), IsOk());
23982462

23992463
ICEBERG_UNWRAP_OR_FAIL(auto manifests, CurrentManifests());
24002464
ASSERT_EQ(manifests.size(), 1U);
2465+
ICEBERG_UNWRAP_OR_FAIL(auto main_snapshot, table_->current_snapshot());
2466+
const auto main_snapshot_id = main_snapshot->snapshot_id;
2467+
2468+
ICEBERG_UNWRAP_OR_FAIL(auto manager, table_->NewSnapshotManager());
2469+
manager->CreateBranch("audit", main_snapshot_id);
2470+
ASSERT_THAT(manager->Commit(), IsOk());
2471+
ASSERT_THAT(table_->Refresh(), IsOk());
24012472

24022473
ICEBERG_UNWRAP_OR_FAIL(auto rewrite, table_->NewRewriteManifests());
2403-
rewrite->ToBranch("someBranch");
2404-
auto result = rewrite->Commit();
2405-
EXPECT_THAT(result, IsError(ErrorKind::kValidationFailed));
2406-
EXPECT_THAT(result,
2407-
HasErrorMessage("Cannot commit to branch someBranch: RewriteManifests "
2408-
"does not support branch commits"));
2474+
rewrite->ToBranch("audit");
2475+
rewrite->ClusterBy([](const DataFile&) { return "all"; });
2476+
ASSERT_THAT(rewrite->Commit(), IsOk());
2477+
2478+
ASSERT_THAT(table_->Refresh(), IsOk());
2479+
ICEBERG_UNWRAP_OR_FAIL(auto current_main, table_->current_snapshot());
2480+
EXPECT_EQ(current_main->snapshot_id, main_snapshot_id);
2481+
auto audit_it = table_->metadata()->refs.find("audit");
2482+
ASSERT_NE(audit_it, table_->metadata()->refs.end());
2483+
ASSERT_NE(audit_it->second->snapshot_id, main_snapshot_id);
2484+
ICEBERG_UNWRAP_OR_FAIL(auto audit_snapshot,
2485+
table_->metadata()->SnapshotById(audit_it->second->snapshot_id));
2486+
SnapshotReader audit_reader(audit_snapshot.get());
2487+
ICEBERG_UNWRAP_OR_FAIL(auto audit_manifests, audit_reader.DataManifests(file_io_));
2488+
EXPECT_EQ(std::ranges::distance(audit_manifests), 1);
24092489
}
24102490

24112491
TEST_P(RewriteManifestsTest, RewriteDataManifestsPreservesDeletes) {

0 commit comments

Comments
 (0)