github-actions[bot] commented on code in PR #68032:
URL: https://github.com/apache/doris/pull/68032#discussion_r4241001557
##########
fe/fe-core/src/main/java/org/apache/doris/master/ReportHandler.java:
##########
@@ -185,6 +185,10 @@ public TMasterResult handleReport(TReportRequest request)
throws TException {
reportType = ReportType.DISK;
}
+ if (request.isSetRemoteSpillBytes()) {
+ backend.setRemoteSpillBytes(request.getRemoteSpillBytes());
Review Comment:
[P2] Reject remote-spill reports from a previous BE start. A disk report
sent just before a BE restart can arrive after `handleHbResponse()` has cleared
`remoteSpillBytes` for the new start time. This assignment restores the old
process count because `TReportRequest` has no start epoch or ordering field and
the report is matched only by host/port. SHOW DATA then exposes the stale value
until another disk report. Carry the BE epoch with the report and apply it only
when it matches the current heartbeat epoch; test the delayed-old-report
ordering. This is separate from the existing residual-object accounting thread.
##########
be/src/exec/spill/spill_file_writer.cpp:
##########
@@ -90,42 +146,208 @@ Status SpillFileWriter::_close_current_part(const
std::shared_ptr<SpillFile>& sp
_part_meta.append((const char*)&_part_max_sub_block_size,
sizeof(_part_max_sub_block_size));
_part_meta.append((const char*)&_part_written_blocks,
sizeof(_part_written_blocks));
- {
+ int64_t meta_size = _part_meta.size();
+ // The footer must always be written so that the part can be closed;
account it
+ // without checking the capacity limit.
+ Status status = _data_dir->try_reserve(meta_size, /*force=*/true);
+ if (status.ok()) {
SCOPED_TIMER(_write_file_timer);
- RETURN_IF_ERROR(_file_writer->append(_part_meta));
+ status = _append_to_part(Slice(_part_meta));
+ if (!status.ok()) {
+ _data_dir->release(meta_size);
+ }
}
- int64_t meta_size = _part_meta.size();
- _part_written_bytes += meta_size;
- COUNTER_UPDATE(_write_file_total_size, meta_size);
+ if (status.ok()) {
+ _part_written_bytes += meta_size;
+ COUNTER_UPDATE(_write_file_total_size, meta_size);
+ if (_resource_ctx) {
+ if (_data_dir->is_remote()) {
+
_resource_ctx->io_context()->update_spill_write_bytes_to_remote_storage(meta_size);
+ } else {
+
_resource_ctx->io_context()->update_spill_write_bytes_to_local_storage(meta_size);
+ }
+ }
+ if (_write_file_current_size) {
+ COUNTER_UPDATE(_write_file_current_size, meta_size);
+ }
+
ExecEnv::GetInstance()->spill_file_mgr()->update_spill_write_bytes(meta_size);
+ // Incrementally update SpillFile's accounting so gc() can always
+ // decrement the correct amount, even if close() is never called.
+ if (spill_file) {
+ spill_file->update_written_bytes(meta_size);
+ }
+ }
+
+ // Close synchronously. Spilling already runs on a thread that may block
on every append
+ // (the pipeline execution thread running the revoke task, or the async
result-writer
+ // thread for Iceberg), and with the default part size the wait at a part
boundary (the
+ // tail of the uploads plus one CompleteMultipartUpload) is negligible
against the part
+ // itself, so overlapping it with the next part is not worth an async
close pipeline.
+ std::unique_ptr<io::FileWriter> writer = std::move(_file_writer);
+ MultipartUploadId upload = _multipart_upload_id(writer.get());
+ if (status.ok()) {
+ MonotonicStopWatch close_watch;
+ close_watch.start();
+ status = writer->close();
Review Comment:
[P1] Return a spill error when post-upload verification fails. This new
close reaches `S3FileWriter::check_after_upload()` with
`enable_s3_object_check_after_upload` enabled by default. A failed HEAD or
mismatched size executes `DCHECK(false)`/`DCHECK_EQ` before returning the
intended `IOError`, so a transient object-store failure aborts a DCHECK-enabled
BE during ordinary spilling. The existing thread at line 218 addresses
stored-byte accounting after a failed close; it does not cover this process
abort. Return `Status` without a fatal assertion and fault-inject HEAD failure
and wrong size in the S3 spill test.
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/ShowDataCommand.java:
##########
@@ -585,8 +594,22 @@ public String toString() {
return toSql();
}
+ /**
+ * Bytes of query spill currently held in object storage, summed over the
alive BEs from their
+ * latest disk reports. Only the master FE keeps these values, so the
command is forwarded to it.
+ */
+ private long getRemoteSpillSize() throws AnalysisException {
+ long total = 0;
+ for (Backend backend :
Env.getCurrentSystemInfo().getAllBackendsByAllCluster().values()) {
+ if (backend.isAlive()) {
Review Comment:
[P2] Include fresh disk reports when heartbeat alone fails. Cloud
`REPORT_DISK_STATE` runs independently of the BE heartbeat endpoint, and
`ReportHandler` accepts `remote_spill_bytes` even after heartbeat failures mark
that BE dead. This new `isAlive()` filter then drops those successfully
reported bytes from the SHOW DATA spill row and total. The earlier poller
thread fixed its own heartbeat filter, but that poller is gone and this new
filter reintroduces the undercount. Sum accepted reports with an explicit
freshness rule for senders that stop reporting; test blocked heartbeat with
continuing disk reports.
##########
fe/fe-core/src/main/java/org/apache/doris/resource/workloadschedpolicy/WorkloadRuntimeStatusMgr.java:
##########
@@ -178,11 +180,11 @@ private List<AuditEvent> getQueryNeedAudit() {
if (waitTimeMs <= queryAuditLogTimeout) {
continue;
}
- Set<Long> expectedBackendIds =
externalDmlAuditBackendIds.get(ae);
+ Set<Long> expectedBackendIds = auditBackendIds.get(ae);
Review Comment:
[P2] Keep Flight audits pending until BE execution finishes. A BE-backed
Flight statement returns from `executeAndSendResult()` immediately after
dispatch, and `ConnectProcessor` submits its audit event while the client is
still consuming DoGet. With this newly generalized backend wait, a normal
S3-spilling Flight query running longer than the default 60-second fallback is
released before any final BE totals exist; the remote pair seeded as zero (or
interim totals) is persisted and never corrected. This differs from the older
completed-query late-report thread because the query itself has not finished.
Defer Flight audit until completion or retain its event while the query remains
active, and test a long Flight query.
--
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]
