[fix](iceberg) Harden write lifecycle and cleanup - #66348
Conversation
Problem Summary: Iceberg writes could bypass cleanup, miss overwrite conflicts, undercount partition sorter memory, hide close errors, exceed report limits, or leave incomplete remote files. Partition dispatch also allocated full-block filters for every partition. Solution: Unify failure handling, preserve the durable commit point, validate overwrite conflicts, account for all active sort writers, propagate close failures, bound commit metadata, abort incomplete uploads, add orphan cleanup, and use compact row permutations.
|
Thank you for your contribution to Apache Doris. Please clearly describe your PR:
|
|
run buildall |
|
/review |
![github-actions[bot]](https://avatars.githubusercontent.com/in/15368?s=60&v=4){kind=small}
There was a problem hiding this comment.
Requesting changes: the static review found nine actionable issues (six P1, three P2), including three paths that can delete or reference missing Iceberg data, a concurrent use-after-free, an aggregate Thrift limit bypass, pathological spill reservation, and two incomplete object-storage cleanup paths.
Critical checkpoint conclusions:
- Goal and proof: The PR partially hardens write/cleanup/OCC behavior, and the overwrite and external-insert lifecycle changes are credible, but the nine inline findings prevent the stated lifecycle and cleanup goals from being met. Added tests prove only narrow single-state, AWS-abort, permutation, and executor cases.
- Scope and clarity: The edits are generally focused, but the new destructive action reimplements Iceberg safety policy incompletely, and the provider-neutral abort surface is implemented with AWS-specific state.
- Concurrency: The async writer publishes active-writer snapshots while a separately scheduled revocation task reads them. Per-writer
_sorter_mutexlocking is consistent and no deadlock was found, but it cannot protect the snapshot container because the range-for drops its temporary owner before iteration (MAIN-BE-003). - Lifecycle: Sort-writer close/cancel/deferred cleanup and external commit/listener ordering were traced end to end and otherwise preserve error ownership. S3/Azure abort lifecycle is incomplete: a rejected abort is never retried, and Azure reports successful abandonment without provider cleanup (MAIN-IO-001/002).
- Configuration: No new configuration was added. Existing
thrift_max_message_sizeand S3 limiter settings are read dynamically, but their consumers do not enforce the correct aggregate/cleanup contracts. - Compatibility: No storage-format or wire-field change was introduced. Default virtual methods preserve source-level parallel implementations, but that default is exactly why Azure silently lacks the new cleanup behavior.
- Parallel paths: Static/dynamic/branch/empty overwrites, data/delete writers, sync/async/destructor close, S3/Azure providers, and all insert executors were checked. The accepted findings identify the paths where parity is missing.
- Conditional checks: Existing overwrite conflict conditions are sound. The orphan action is missing the required
gc.enabledfence, safe retention interval, canonical URI identity handling, and version-hint reachability. - Test coverage: Coverage is not comprehensive. Missing cases include multi-task aggregate reports, concurrent snapshot publication/traversal, many-partition reserve accounting, hard-limit abort rejection, Azure staged-block abort, and execution-level orphan deletion tests for GC, retention, URI aliases, and the version hint.
- Test results: The new assertions are logically consistent within their narrow scopes; no result files changed. Builds and tests were intentionally not run because the authoritative review prompt forbids them, so this conclusion is static-only.
- Observability: Existing timers/logs cover writer/spill/commit paths, and abort failures are warned. No additional standalone observability defect was found, but logging does not substitute for retained cleanup ownership or retry.
- Transaction and persistence: No Doris EditLog state is added. Iceberg overwrite OCC anchors for main/branch/empty/static/dynamic cases are sound, and the durable external commit marker precedes refresh/listener suppression. The orphan action can nevertheless race a concurrent external commit (MAIN-FE-004).
- Data writes and atomicity: The aggregate-report failure can strand uncommitted objects; unsafe orphan deletion can corrupt shared, aliased, or concurrent writes; and multipart cleanup can leak provider state. These are blocking write-lifecycle defects.
- FE/BE propagation: No new FE-BE variable or Thrift field is introduced. Existing commit metadata is propagated on every producer path, but its budget is enforced per task rather than on the final aggregated RPC (MAIN-BE-001).
- Performance: Compact permutation dispatch is semantically sound, and bounded snapshot-copy cost is not independently material. Summing one full-batch reserve estimate per active partition can request roughly 128 blocks and repeatedly spill tiny partitions (MAIN-BE-002).
- Other issues: Error precedence, close idempotence, spill cleanup, overwrite validation, permission checks, location containment, and post-commit listener behavior produced no additional substantiated defects.
User focus: review_focus.txt contained no additional focus guidance, so the whole PR was reviewed without a narrower focus.
Review status: complete and converged in round 3. All final full-scope and risk-focused reviewers returned NO_NEW_VALUABLE_FINDINGS for this exact comment set. Static review only; no builds or tests were run.
| } else if (!req.runtime_states.empty()) { | ||
| for (auto* rs : req.runtime_states) { | ||
| if (auto rs_icd = rs->iceberg_commit_datas(); !rs_icd.empty()) { | ||
| rs->append_iceberg_commit_datas(¶ms.iceberg_commit_datas); |
There was a problem hiding this comment.
[P1] Enforce one budget for the aggregated report
The new guard is per task RuntimeState, but this loop concatenates every task's accepted vector into one TReportExecStatusParams. With a 100 MiB limit, two parallel sink tasks can each accept about 60 MiB and then send a roughly 120 MiB RPC, which FE rejects. By this point successful writers have cleared their local cleanup lists, so the failed report also strands the uncommitted objects. Please enforce a shared fragment/report budget before cleanup ownership is released and add a multi-task aggregate-limit test.
| if (!sort_writer) { | ||
| return 0; | ||
| size_t reserve_size = 0; | ||
| for (const auto& writer : *_writer->active_writers()) { |
There was a problem hiding this comment.
[P1] Do not reserve a full input batch per partition
Each FullSorter::get_reserve_mem_size assumes that sorter may receive state->batch_size() rows, but this loop sums that full-batch estimate across every active partition even though there is only one incoming block and its rows are divided among them. At the default 128-partition limit this can request memory for roughly 128 blocks; when that false reservation fails, revoke_memory spills only one small partition and the same block repeatedly pauses. Please bound the aggregate to the actual one-block input (or reserve after dispatch using each selected row count) and test the many-partition case.
| if (!sort_writer) { | ||
| return 0; | ||
| size_t revocable_size = 0; | ||
| for (const auto& writer : *_writer->active_writers()) { |
There was a problem hiding this comment.
[P1] Retain the snapshot owner while iterating
active_writers() returns a shared_ptr by value, but this C++20 range-for dereferences that temporary directly. The temporary owner is destroyed before iterator initialization; if the async writer then publishes a replacement snapshot, the old vector can be freed while the revocation thread is traversing it. The same lifetime bug appears in the loops at lines 59 and 87. Please first store the snapshot in a named shared_ptr and iterate through that retained owner, and cover publication versus traversal concurrency.
| } | ||
|
|
||
| @Override | ||
| protected List<String> executeAction(Table table, ConnectorSession session) { |
There was a problem hiding this comment.
[P1] Honor Iceberg's GC-disabled safety fence
This destructive path never checks gc.enabled. Iceberg uses gc.enabled=false specifically when files may be shared with another table, and its 1.10.1 orphan-file action refuses to run in that state because deletion can corrupt the other table. With dry_run=false, this implementation proceeds directly to deleteFile. Please reject execution when GC is disabled (using the Iceberg property/default) and add an execution test proving no deletion occurs.
| // from being treated as children of "table". | ||
| String listingPrefix = scanLocation.endsWith("/") ? scanLocation : scanLocation + "/"; | ||
| for (FileInfo file : ((SupportsPrefixOperations) table.io()).listPrefix(listingPrefix)) { | ||
| if (file.createdAtMillis() < olderThan && !reachable.contains(file.location())) { |
There was a problem hiding this comment.
[P1] Compare canonical file identities before deleting
The membership test compares raw locations. If retained metadata references s3a://bucket/key while prefix listing returns s3://bucket/key for the same object (or an equivalent authority differs), this branch treats the live file as orphaned and deletes it. Iceberg's 1.10.1 action canonicalizes both sides, treats s3/s3a/s3n as equivalent by default, and fails closed on unresolved prefix mismatches. Please adopt equivalent identity handling and add alias/mismatch tests before enabling deletion.
| Set<String> reachable = collectReachableFiles(table); | ||
| long orphanCount = 0; | ||
| long deletedCount = 0; | ||
| long olderThan = namedArguments.getLong(OLDER_THAN); |
There was a problem hiding this comment.
[P1] Reject unsafe recent orphan cutoffs
older_than is accepted as any nonnegative value, including now or a future timestamp. A concurrent writer's newly uploaded file is not yet reachable from table snapshots, so this action can list and delete it and the writer can then commit metadata pointing to the missing object. Iceberg's 1.10.1 SQL procedure rejects intervals shorter than 24 hours for exactly this race, reserving arbitrary cutoffs for its expert Action API. Please enforce the SQL safety interval (or an explicit expert-only override) and test a recent uncommitted file.
|
|
||
| ObjectStorageResponse RateLimitedObjStorageClient::abort_multipart_upload( | ||
| const ObjectStoragePathOptions& opts) { | ||
| S3RateLimitGuard guard(S3RateLimitType::PUT, 0); |
There was a problem hiding this comment.
[P2] Ensure cleanup bypasses or reserves the PUT limit
This abort uses the same hard PUT request-count bucket as create/upload/complete. When that bucket is what caused the write failure, the immediately following abort is rejected before _inner is called; S3FileWriter::close only warns, transitions to its terminal close state, and the destructor no longer retries, leaving the multipart upload behind. Please give mandatory abort cleanup guaranteed provider admission (or retain a retryable cleanup state) and cover the exhausted-limit path rather than only an unwrapped successful abort.
| Status S3FileWriter::_abort_impl() { | ||
| _wait_until_finish( | ||
| fmt::format("wait s3 file {} before abort", _obj_storage_path_opts.path.native())); | ||
| _pending_buf.reset(); |
There was a problem hiding this comment.
[P2] Implement abort for Azure staged blocks
Azure uses this writer too, but create_multipart_upload intentionally returns no upload ID and upload_part stages blocks. This guard therefore skips provider cleanup, returns OK, and lets abort() mark the writer CLOSED even after Azure blocks were staged; generic upload/destructor failures have no later path deletion, so those blocks remain until provider GC. Please make abandonment provider-aware (for example, delete/discard the uncommitted blob through an Azure override), preserve cleanup failures, and add an Azure failure test.
| } | ||
|
|
||
| private Set<String> collectReachableFiles(Table table) throws IOException { | ||
| Set<String> reachable = new HashSet<>(ReachableFileUtil.metadataFileLocations(table, true)); |
There was a problem hiding this comment.
[P2] Keep the live version hint reachable
metadataFileLocations(table, true) does not include Hadoop Iceberg's metadata/version-hint.text; Iceberg exposes that path separately via ReachableFileUtil.versionHintLocation(table) and its own orphan action explicitly adds it. HadoopFileIO supports prefix listing, so a table-root run with a cutoff newer than an old hint will classify and delete this live metadata file. Please add the hint to the reachable identities and cover it in a non-dry-run test.
FE UT Coverage ReportIncrement line coverage |
TPC-H: Total hot run time: 29107 ms |
TPC-DS: Total hot run time: 169793 ms |
ClickBench: Total hot run time: 23.92 s |
BE UT Coverage ReportIncrement line coverage Increment coverage report
|
BE Regression && UT Coverage ReportIncrement line coverage Increment coverage report
|
FE Regression Coverage ReportIncrement line coverage |
### What problem does this PR solve? Issue Number: None Related PR: apache#66348 Problem Summary: Parallel Iceberg sink tasks could exceed the shared report limit, partition writers could over-reserve memory or outlive a temporary snapshot, failed multipart uploads could skip mandatory cleanup, and orphan-file deletion lacked Iceberg safety fences. Share the report budget, retain writer snapshots, bound reservations, make abort cleanup provider-aware, and enforce safe canonical orphan detection. ### Release note Improve Iceberg write failure cleanup and orphan-file deletion safety. ### Check List (For Author) - Test: Unit Test - BE ASAN focused tests with Azure support enabled - Full fe-connector-iceberg unit test suite - Behavior changed: Yes, unsafe orphan cutoffs and GC-disabled deletion are rejected, and failed multipart uploads are cleaned up reliably. - Does this need documentation: No
|
run buildall |
|
/review |
There was a problem hiding this comment.
Request changes: two distinct issues remain after the current fixes: Azure abort can delete a valid pre-existing Put Blob object, and orphan-file reachability repeatedly scans inherited manifests with quadratic cost. I did not duplicate the existing review threads.
Critical checkpoint conclusions:
- Goal and proof: The PR broadly hardens Iceberg write/cleanup lifecycle, OCC, reporting limits, and orphan cleanup, but the Azure data-preservation goal is not yet met. The added tests cover nominal cleanup and OCC paths, not the two cases below.
- Scope and clarity: The changes are cohesive for the stated lifecycle work. The orphan action's hand-built reachability scan is the one area that should reuse Iceberg's unique-manifest approach.
- Concurrency: Async writer publication is protected by retained shared snapshots and per-sorter locking; the shared report budget has a consistent lock order. Azure's committed-block test is not a valid ownership fence for an existing object.
- Lifecycle: FE before-execution cleanup and post-commit error boundaries are sound, as are the reviewed nested writer close paths. Azure abort still destroys state it does not own.
- Configuration: No new configuration behavior requires dynamic-reload handling.
- Compatibility: No FE-BE or storage-format incompatibility was found. The changed Azure block-ID format has a bounded retry impact and did not survive as a separate finding.
- Parallel paths: S3, Azure, single-shot, multipart, committer, overwrite, branch, static/dynamic overwrite, and delete/merge paths were traced; the surviving provider-specific issue is inline.
- Conditional checks: GC, retention, URI containment, identity ambiguity, and overwrite validation checks are fail-closed.
CommittedBlocks.empty()is the invalid special case. - Test coverage: Unit coverage is substantial but misses preservation of an existing Put Blob target and an append-heavy multi-snapshot reachability scan. No end-to-end result covers either.
- Test results: No tests or builds were run because the authoritative review task is static-only; no generated result files were changed.
- Observability: Existing status/log paths are adequate for the reviewed failures; no additional metric was required for a correctness conclusion.
- Transactions and persistence: Iceberg overwrite OCC, including empty-table and branch anchors, is sound. No EditLog or master-failover persistence change is involved.
- Data writes: Commit/report cleanup ownership is sound outside the Azure abort branch; that branch can lose previously committed data and blocks approval.
- FE-BE variables: No new cross-version transmitted variable or missing propagation path was found.
- Performance: Per-snapshot
planFiles()causes the P2 quadratic manifest-read issue below; no other distinct hot-path issue survived duplicate suppression. - Other issues: The sorter aggregate-reservation concern remains covered by the existing thread, and cleanup admission/terminal-state concerns remain covered by the existing Azure/rate-limit threads, so they are not repeated here.
User focus: no additional user-provided focus was present; the full PR was reviewed.
| GetBlockListOptions get_options; | ||
| get_options.ListType = Models::BlockListType::All; | ||
| auto block_list = client.GetBlockList(get_options); | ||
| if (block_list.Value.CommittedBlocks.empty()) { |
There was a problem hiding this comment.
[P1] Preserve a pre-existing Put Blob object during abort
CommittedBlocks.empty() does not prove that the target key has no committed object. Azure defines this list as blocks committed through Put Block List, while this client itself writes small objects with UploadFrom / Put Blob. Because S3FileSystem::create_file_impl allows an existing key, a multipart replacement can stage blocks over such an object and, on failure, this branch deletes the valid old blob; IfMatch, when present, merely authorizes deleting the ETag that already existed. Please preserve committed content when abandoning this writer (leaving its unique staged blocks to expire if Azure cannot selectively unstage them) and add a test that seeds the key through put_object, stages a multipart block, aborts, and verifies the original bytes remain. See Azure's Get Block List contract.
| deletes.forEach(delete -> reachable.add(delete.location())); | ||
| } | ||
| } | ||
| try (CloseableIterable<FileScanTask> tasks = table.newScan() |
There was a problem hiding this comment.
[P2] Read each retained manifest only once
This calls planFiles() for every retained snapshot. In an append-heavy history, later snapshots inherit earlier data manifests, so the same remote manifests and entries are reopened across snapshots and total work grows quadratically before the HashSet removes duplicate locations. This synchronous FE action can therefore become impractical on a table with a long retained history. Iceberg 1.10.1's own action deduplicates manifest paths before reading their entries. Please collect unique data manifests and read each once, and add a recording-FileIO test whose manifest-open count is bounded by unique manifests rather than snapshots.
FE UT Coverage ReportIncrement line coverage |
TPC-H: Total hot run time: 29004 ms |
TPC-DS: Total hot run time: 169560 ms |
ClickBench: Total hot run time: 23.98 s |
BE UT Coverage ReportIncrement line coverage Increment coverage report
|
BE Regression && UT Coverage ReportIncrement line coverage Increment coverage report
|
FE Regression Coverage ReportIncrement line coverage |
2 participants
What problem does this PR solve?
Issue Number: None
Related PR: None
Problem Summary:
Iceberg writes had several correctness and resource-safety gaps across the insert lifecycle:
This change unifies lifecycle cleanup, makes post-commit refresh/listener work best-effort, adds overwrite OCC validation, publishes all active partition writers for memory accounting, propagates close failures, bounds commit metadata before reporting, adds explicit writer abort and orphan cleanup support, and dispatches rows with compact permutations.
Release note
Harden Iceberg write correctness, memory accounting, and orphan-file cleanup.
Check List (For Author)
Test
Behavior changed:
Does this need documentation?
Check List (For Reviewer who merge this PR)