github-actions[bot] commented on code in PR #69842:
URL: https://github.com/apache/doris/pull/69842#discussion_r4411927190
##########
be/src/exec/sink/writer/iceberg/viceberg_sort_writer.cpp:
##########
@@ -118,80 +118,35 @@ Status VIcebergSortWriter::close(const Status& status) {
}
Status VIcebergSortWriter::_close_locked(const Status& status) {
- // Track the actual internal status of operations performed during close.
- // This is important because if intermediate operations (like do_sort())
fail,
- // we need to propagate the actual error status to the underlying
partition writer's
- // close() call, rather than the original status parameter which could be
OK.
- Status internal_status = Status::OK();
- // Track the close status of the underlying partition writer.
- // If _iceberg_partition_writer->close() fails (e.g., Parquet file flush
error),
- // we must propagate this error to the caller to avoid silent data loss.
- Status close_status = Status::OK();
-
- // Defer ensures the underlying partition writer is always closed and
- // spill streams are cleaned up, regardless of whether intermediate
operations succeed.
- // Uses internal_status to propagate any errors that occurred during close
operations.
- Defer defer {[&]() {
- // If any intermediate operation failed, pass that error to the
partition writer;
- // otherwise, pass the original status from the caller.
- close_status =
- _iceberg_partition_writer->close(internal_status.ok() ? status
: internal_status);
- if (!close_status.ok()) {
- LOG(WARNING) << fmt::format("_iceberg_partition_writer close
failed, reason: {}",
- close_status.to_string());
- }
- _cleanup_spill_streams();
- }};
-
- // If the original status is already an error or the query is cancelled,
- // skip all close operations and propagate the original error
- if (!status.ok() || _runtime_state->is_cancelled()) {
- return status;
- }
-
- // If sorter was never initialized (e.g., no data was written), nothing to
do
- if (_sorter == nullptr) {
- return Status::OK();
- }
-
- // Check if there is any remaining data in the sorter (either unsorted or
already sorted blocks)
- if (!_sorter->merge_sort_state()->unsorted_block()->empty() ||
- !_sorter->merge_sort_state()->get_sorted_block().empty()) {
- if (_sorted_spill_files.empty()) {
- // No spill has occurred, all data is in memory.
- // Sort the remaining data, prepare for reading, and write to file.
- internal_status = _sorter->do_sort();
- if (!internal_status.ok()) {
- return internal_status;
- }
- internal_status = _sorter->prepare_for_read(false);
- if (!internal_status.ok()) {
- return internal_status;
+ Status internal_status = status;
+ if (status.ok() && !_runtime_state->is_cancelled()) {
+ internal_status = Status::OK();
+ if (_sorter != nullptr &&
(!_sorter->merge_sort_state()->unsorted_block()->empty() ||
+
!_sorter->merge_sort_state()->get_sorted_block().empty())) {
+ if (_sorted_spill_files.empty()) {
+ internal_status = _sorter->do_sort();
+ if (internal_status.ok()) {
+ internal_status = _sorter->prepare_for_read(false);
+ }
+ if (internal_status.ok()) {
+ internal_status = _write_sorted_data();
+ }
+ } else {
+ internal_status = _do_spill();
}
- internal_status = _write_sorted_data();
- return internal_status;
}
-
- // Some data has already been spilled to disk.
- // Spill the remaining in-memory data to a new spill stream.
- internal_status = _do_spill();
- if (!internal_status.ok()) {
- return internal_status;
+ if (internal_status.ok() && !_sorted_spill_files.empty()) {
+ internal_status = _combine_files_output();
}
}
- // Merge all spilled streams using multi-way merge sort and output final
sorted data to files
- if (!_sorted_spill_files.empty()) {
- internal_status = _combine_files_output();
- if (!internal_status.ok()) {
- return internal_status;
- }
+ // Form the return value only after the underlying close runs; a deferred
assignment is too late.
+ Status close_status =
Review Comment:
[P2] Keep partition finalization safe when sorting throws
For a spill-enabled sorted INSERT, EOS reaches this code from
`SpillIcebergTableSinkLocalState::_close_writer()`. `do_sort()`/merge/write can
throw, for example when `get_permutation()` or `permute()` allocation raises
`MEM_ALLOC_FAILED`. Removing the `Defer` lets that bypass partition close,
while `close()` still sets `_closed = true` during unwinding. The scheduler
converts the exception and retries outer close with the failure, but the sorted
writer now returns immediately: the current path never enters `_closed_files`
or BE commit data. Its stream is eventually closed during destruction, leaving
an orphan that neither BE nor FE rollback knows about (HDFS already creates it
at open). Convert intermediate exceptions into `internal_status` before
unconditional partition finalization, or retain exception-safe cleanup
ownership. Add an EOS allocation-failure case that checks physical file removal.
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/IcebergTransaction.java:
##########
@@ -690,9 +691,18 @@ public void commit() throws UserException {
}
transaction.commitTransaction();
});
+ } catch (CommitStateUnknownException e) {
+ // The metadata commit may have succeeded. A later rollback must
not delete files
+ // that could already be referenced by a published snapshot.
+ synchronized (this) {
+ commitDataList.clear();
+ }
+ throw e;
} catch (RuntimeException e) {
+ cleanupReportedFiles();
Review Comment:
[P1] Preserve files for indeterminate REST commit failures
A REST server can publish the snapshot and then lose or truncate the
response. In the pinned Iceberg 1.11.0,
[HTTPClient.execute](https://raw.githubusercontent.com/apache/iceberg/apache-iceberg-1.11.0/core/src/main/java/org/apache/iceberg/rest/HTTPClient.java#L296-L333)
converts response I/O/parsing failures to `RESTException`, so these bypass the
`CommitStateUnknownException` catch.
[BaseTransaction](https://raw.githubusercontent.com/apache/iceberg/apache-iceberg-1.11.0/core/src/main/java/org/apache/iceberg/BaseTransaction.java#L349-L360)
deliberately preserves files for such non-`CleanableFailure` exceptions under
strict cleanup. This new catch instead deletes every reported data/delete file,
leaving the published snapshot referencing missing objects. Gate deletion on a
definitive commit failure and preserve the files through the subsequent
`rollback()` too; the explicit unknown-state test does not cover this
response-loss case.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
