github-actions[bot] commented on code in PR #66825:
URL: https://github.com/apache/doris/pull/66825#discussion_r3844888590
##########
be/src/io/fs/s3_file_writer.cpp:
##########
@@ -106,6 +107,22 @@ Status S3FileWriter::_create_multi_upload_request() {
return {resp.resp.status.code, std::move(resp.resp.status.msg)};
}
+std::function<void()> S3FileWriter::rejected_report_cleanup() const {
+ auto client_holder = _obj_client;
+ auto path_opts = _obj_storage_path_opts;
+ return [client_holder = std::move(client_holder), path_opts =
std::move(path_opts)]() {
+ auto client = client_holder->get();
+ if (client == nullptr || !path_opts.upload_id.has_value()) {
+ return;
+ }
+ auto response = client->abort_multipart_upload(path_opts);
Review Comment:
[P1] Retain cleanup ownership when the multipart abort fails
This is the only cleanup owner for uploads rejected before the final report
reaches FE, but it makes one abort attempt and discards the callback even on
failure. In particular, the new rate-limited decorator can return
`EXCEEDED_LIMIT` without calling S3; this path only logs it, `RuntimeState` has
already consumed the terminal finalizer, and the writer destructor deliberately
does not abort. The MPU therefore remains indefinitely when no lifecycle rule
is configured. Please keep/retry terminal cleanup until abort succeeds or
reaches an authoritative already-gone result (and avoid dropping it solely
because the ordinary PUT limiter is exhausted).
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/IcebergTransaction.java:
##########
@@ -552,23 +579,46 @@ private void updateManifestAfterInsert(TUpdateMode
updateMode) {
@Override
public void commit() throws UserException {
- // commit the iceberg transaction
- transaction.commitTransaction();
+ stopAcceptingCommitData();
+ try {
+ transaction.commitTransaction();
+ } finally {
+ releaseCommitFence();
+ }
}
@Override
public void rollback() {
- if (isRewriteMode) {
- // Clear the collected files for rewrite mode
- synchronized (filesToDelete) {
- filesToDelete.clear();
- }
- synchronized (filesToAdd) {
- filesToAdd.clear();
+ stopAcceptingCommitData();
+ try {
+ if (isRewriteMode) {
+ // Clear the collected files for rewrite mode
+ synchronized (filesToDelete) {
+ filesToDelete.clear();
+ }
+ synchronized (filesToAdd) {
+ filesToAdd.clear();
+ }
+ LOG.info("Rewrite transaction rolled back");
}
- LOG.info("Rewrite transaction rolled back");
+ // For insert mode, do nothing as original implementation
Review Comment:
[P1] Clean up acknowledged files when the transaction rolls back
The final-report ACK only means FE accepted the commit metadata; it can
happen before the statement is durable. For example, `executeSingleInsert`
joins the fragments and then rejects filtered rows in strict mode. By then BE
has terminalized its file callbacks as `ACKNOWLEDGED`, but this rollback branch
does nothing for insert/delete/merge paths (and merely clears rewrite
additions), so the newly staged objects have no cleanup owner. Please retain
the acknowledged paths on FE and delete the new files on rollback outcomes
known not to have published a snapshot, while preserving them for genuinely
ambiguous commit outcomes.
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/physical/PhysicalIcebergMergeSink.java:
##########
@@ -339,22 +340,100 @@ private InsertPartitionFieldResult
buildInsertPartitionFields(
insertPartitionFields.clear();
return new InsertPartitionFieldResult(false, hasNonIdentity,
spec.specId());
}
- ExprId exprId = columnExprIdMap.get(sourceField.name());
+ Column rootColumn = findSourceRootColumn(cols, field.sourceId());
+ // Nested routing must bind through stable Iceberg IDs; a name
fallback can target a
+ // different field after rename/drop-and-add schema evolution.
+ ExprId exprId = rootColumn == null || rootColumn.getUniqueId() < 0
+ ? null : columnIdToExprId.get(rootColumn.getUniqueId());
if (exprId == null) {
insertPartitionFields.clear();
return new InsertPartitionFieldResult(false, hasNonIdentity,
spec.specId());
}
String transform = field.transform().toString();
Integer param = parseTransformParam(transform);
+ List<Integer> sourceFieldPath = resolveSourceFieldPath(rootColumn,
field.sourceId());
+ if (sourceFieldPath == null) {
+ insertPartitionFields.clear();
+ return new InsertPartitionFieldResult(false, hasNonIdentity,
spec.specId());
+ }
insertPartitionFields.add(new
DistributionSpecMerge.IcebergPartitionField(
- transform, exprId, param, field.name(), field.sourceId()));
+ transform, exprId, param, field.name(), field.sourceId(),
sourceFieldPath));
}
if (insertPartitionFields.isEmpty()) {
return new InsertPartitionFieldResult(false, hasNonIdentity,
spec.specId());
}
return new InsertPartitionFieldResult(true, hasNonIdentity,
spec.specId());
}
+ private Map<Integer, ExprId> buildColumnIdExprIdMap(List<Slot>
outputSlots) {
+ Map<Integer, ExprId> result = new java.util.HashMap<>();
+ List<Column> visibleColumns = new ArrayList<>();
+ for (Column column : cols) {
+ if (column.isVisible()) {
+ visibleColumns.add(column);
+ }
+ }
+ List<Slot> dataSlots = getDataSlots(outputSlots);
+ if (visibleColumns.size() != dataSlots.size()) {
Review Comment:
[P1] Exclude V3 lineage slots from the positional ID map
On every V3 UPDATE/MERGE projection, `getDataSlots()` still contains
`_row_id` and `_last_updated_sequence_number`, while `visibleColumns` excludes
them. The size check therefore returns an empty map, so transformed partitions
(and nested identity sources) fall back to `insertRandom` even though merge
partitioning is enabled. Round-robin fans each logical partition across
writers; with more than 128 partitions a writer can hit `Too many open
partitions`, and smaller writes still amplify files. Please exclude the two
lineage slots here or bind IDs from column metadata directly, and add V3
transformed/nested plan coverage that asserts non-random routing.
##########
be/src/runtime/runtime_state.cpp:
##########
@@ -53,13 +53,128 @@
#include "runtime/thread_context.h"
#include "storage/id_manager.h"
#include "storage/storage_engine.h"
+#include "util/thrift_util.h"
#include "util/timezone_utils.h"
#include "util/uid_util.h"
namespace doris {
#include "common/compile_check_begin.h"
using namespace ErrorCode;
+Status RuntimeState::add_iceberg_commit_datas(TIcebergCommitData
iceberg_commit_data) {
+ ThriftSerializer serializer(false, 256);
+ uint32_t serialized_size = 0;
+ uint8_t* buffer = nullptr;
+ RETURN_IF_ERROR(serializer.serialize(&iceberg_commit_data,
&serialized_size, &buffer));
+
+ // This is an early per-vector guard only; the assembled RPC is measured
again before send.
+ constexpr size_t report_envelope_headroom = 1024 * 1024;
+ const size_t thrift_limit = coordinator_thrift_message_limit();
+ const size_t commit_data_limit =
+ thrift_limit > report_envelope_headroom ? thrift_limit -
report_envelope_headroom : 0;
+ std::lock_guard<std::mutex>
budget_lock(_external_file_report_state->mutex);
+ // Parallel task states share this budget because FE receives their
vectors in one fragment report.
+ if (_external_file_report_state->iceberg_serialized_bytes +
serialized_size + sizeof(uint32_t) >
+ commit_data_limit) {
+ return Status::InternalError(
+ "Iceberg commit metadata exceeds the Thrift report limit;
reduce output file "
+ "count");
+ }
+ std::lock_guard<std::mutex> data_lock(_iceberg_commit_datas_mutex);
+ _external_file_report_state->iceberg_serialized_bytes += serialized_size +
sizeof(uint32_t);
+ _iceberg_commit_datas.emplace_back(std::move(iceberg_commit_data));
+ return Status::OK();
+}
+
+size_t RuntimeState::coordinator_thrift_message_limit() const {
+ int32_t effective_thrift_limit = std::max(config::thrift_max_message_size,
0);
+ if (_query_options.__isset.coordinator_thrift_max_message_size &&
+ _query_options.coordinator_thrift_max_message_size > 0) {
+ // An older FE omits this field; otherwise the receiver's smaller
limit is authoritative.
+ effective_thrift_limit = std::min(effective_thrift_limit,
+
_query_options.coordinator_thrift_max_message_size);
+ }
+ return static_cast<size_t>(effective_thrift_limit);
+}
+
+void RuntimeState::append_external_file_commit_data(TReportExecStatusParams*
params,
+ bool final_report) const {
+ if (!final_report) {
+ // Ownership-bearing commit vectors must only appear in the final
report that transfers them.
+ return;
+ }
+ if (auto updates = hive_partition_updates(); !updates.empty()) {
+ params->__isset.hive_partition_updates = true;
+
params->hive_partition_updates.insert(params->hive_partition_updates.end(),
updates.begin(),
+ updates.end());
+ }
+ append_iceberg_commit_datas(¶ms->iceberg_commit_datas);
+ if (!params->iceberg_commit_datas.empty()) {
+ params->__isset.iceberg_commit_datas = true;
+ }
+ if (auto commit_datas = mc_commit_datas(); !commit_datas.empty()) {
+ params->__isset.mc_commit_datas = true;
+ params->mc_commit_datas.insert(params->mc_commit_datas.end(),
commit_datas.begin(),
+ commit_datas.end());
+ }
+ if (auto commit_messages = paimon_commit_messages();
!commit_messages.empty()) {
+ // branch-4.1 still carries Paimon commit messages in the shared
external-file report.
+ params->__isset.paimon_commit_messages = true;
+
params->paimon_commit_messages.insert(params->paimon_commit_messages.end(),
+ commit_messages.begin(),
commit_messages.end());
+ }
+}
+
+void
RuntimeState::add_rejected_external_file_report_cleanup(std::function<void()>
cleanup) {
+ add_external_file_report_finalizer(
+ [cleanup = std::move(cleanup)](ExternalFileReportOutcome outcome) {
+ if (outcome == ExternalFileReportOutcome::REJECTED) {
+ cleanup();
+ }
+ });
+}
+
+void RuntimeState::add_external_file_report_finalizer(
+ std::function<void(ExternalFileReportOutcome)> finalizer) {
+ std::optional<ExternalFileReportOutcome> terminal_outcome;
+ {
+ std::lock_guard lock(_external_file_report_state->mutex);
+ terminal_outcome = _external_file_report_state->terminal_outcome;
+ if (!terminal_outcome.has_value()) {
+
_external_file_report_state->report_finalizers.emplace_back(std::move(finalizer));
Review Comment:
[P1] Do not retain owners created after an ambiguous report snapshot
An async Hive close can reach this registration after the final report has
already copied its commit vectors. If that earlier RPC is marked `AMBIGUOUS`,
there is no terminal outcome here, so this new callback is queued even though
its MPU identity was not in the report and could not have transferred to FE.
The global ambiguity bit then suppresses a later rejection, no second report
includes the upload, and the S3 writer destructor does not abort it. Please
associate cleanup owners with the report generation that contains them, or
quiesce async writers before taking the final snapshot; owners created after an
ambiguous snapshot must still be rejected and cleaned up.
--
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]
