github-actions[bot] commented on code in PR #68032:
URL: https://github.com/apache/doris/pull/68032#discussion_r4018558013
##########
cloud/src/meta-service/meta_service.cpp:
##########
@@ -3530,6 +3530,143 @@ void
MetaServiceImpl::get_tablet_stats(::google::protobuf::RpcController* contro
}
}
+void MetaServiceImpl::report_spill_stats(::google::protobuf::RpcController*
controller,
+ const ReportSpillStatsRequest*
request,
+ ReportSpillStatsResponse* response,
+ ::google::protobuf::Closure* done) {
+ RPC_PREPROCESS(report_spill_stats, put);
Review Comment:
`report_spill_stats` reads the existing record on every periodic report, but
this declaration publishes only the `put` side of the transaction.
`RPC_PREPROCESS` exports detailed KV counters only for the operation names
passed here, and the PR adds no report-spill `get` bvars, so this RPC's FDB
read QPS/bytes are invisible under `use_detailed_metrics`. Please add the get
counter/byte bvars and declare this as `RPC_PREPROCESS(report_spill_stats, get,
put)`.
##########
fe/fe-core/src/main/java/org/apache/doris/cloud/rpc/MetaServiceProxy.java:
##########
@@ -97,6 +97,32 @@ public boolean needReconn() {
}
}
+ public Cloud.GetSpillStatsResponse
getSpillStats(Cloud.GetSpillStatsRequest request)
Review Comment:
This new proxy method bypasses `MetaServiceClientWrapper.executeRequest`,
unlike comparable calls such as `getTabletStats`. One
`UNAVAILABLE`/`UNKNOWN`/retryable timeout or `MS_TOO_BUSY` response therefore
immediately fails `SHOW DATA PROPERTIES("entire_warehouse"="true")`, and a
failed client is not reconnected. Please implement this through
`executeWithMetrics("getSpillStats", client -> client.getSpillStats(request),
Cloud.GetSpillStatsResponse::getStatus)` so it retains the standard retry and
reconnection semantics.
##########
be/src/exec/spill/spill_file_reader.cpp:
##########
@@ -62,6 +66,26 @@ SpillFileReader::SpillFileReader(RuntimeState* state,
RuntimeProfile* profile,
_read_file_size = get_counter(profile::SPILL_READ_FILE_BYTES);
_read_rows_count = get_counter(profile::SPILL_READ_ROWS);
_read_file_count = get_counter(profile::SPILL_READ_FILE_COUNT);
+ // Optional: older profiles may not register it.
+ _remote_read_requests =
custom_profile->get_counter(profile::SPILL_REMOTE_READ_REQUESTS);
+}
+
+void SpillFileReader::_record_read(size_t bytes_read) {
+ COUNTER_UPDATE(_read_file_size, bytes_read);
+
ExecEnv::GetInstance()->spill_file_mgr()->update_spill_read_bytes(bytes_read);
+ if (_is_remote) {
+ // One read_at() is exactly one GET request on object storage.
Review Comment:
One logical `read_at()` is not necessarily one object-store GET.
`S3FileReader::read_at_impl` may issue multiple `get_object` attempts after
throttling, and it also issues requests that ultimately fail; this callback
runs only once after a successful logical read. Query, workload, and global
spill GET counters therefore under-report actual request traffic. Please
account attempts at the S3 request boundary (or propagate the per-call attempt
delta, including failures) and test retry-then-success plus exhausted failure.
##########
cloud/src/recycler/recycler.cpp:
##########
@@ -7828,6 +7835,50 @@ int InstanceRecycler::recycle_expired_stage_objects() {
return ret;
}
+int InstanceRecycler::recycle_expired_spill_objects() {
+ LOG_INFO("begin to recycle expired spill objects").tag("instance_id",
instance_id_);
+
+ int64_t start_time =
duration_cast<seconds>(steady_clock::now().time_since_epoch()).count();
+ RecyclerMetricsContext metrics_context(instance_id_,
"recycle_expired_spill_objects");
+
+ DORIS_CLOUD_DEFER {
+ int64_t cost =
+
duration_cast<seconds>(steady_clock::now().time_since_epoch()).count() -
start_time;
+ metrics_context.finish_report();
+ LOG_INFO("recycle expired spill objects, cost={}s",
cost).tag("instance_id", instance_id_);
+ };
+
+ int64_t expiration_time =
Review Comment:
This age-only sweep has no active boot/query fence. A valid query can have
immutable spill parts older than the seven-day default because `query_timeout`
has no upper bound, so the recycler can delete data the query still needs. The
mutable TTL is also unvalidated: `0` selects objects up to now and a negative
value moves the cutoff into the future, selecting new live objects. Please
validate a positive TTL and add a liveness fence or enforce the maximum-query
invariant; the current mock test ignores expiration and cannot cover either
case.
##########
be/src/exec/spill/spill_file.cpp:
##########
@@ -43,23 +43,26 @@ SpillFile::~SpillFile() {
}
void SpillFile::gc() {
- bool exists = false;
- auto status = io::global_local_filesystem()->exists(_spill_dir, &exists);
- if (status.ok() && exists) {
- // Delete spill directory directly instead of moving it to a GC
directory.
- // This simplifies cleanup and avoids retaining spill data under a GC
path.
- status = io::global_local_filesystem()->delete_directory(_spill_dir);
+ if (_dir_created) {
+ // Delete the spill directory (or object key prefix) directly instead
of moving it to a
+ // GC directory. No existence check: for object storage a "directory"
never exists as an
+ // object, while deleting a missing local directory or an empty prefix
is a no-op.
+ auto fs = _data_dir->fs();
+ Status status = fs != nullptr ? fs->delete_directory(_spill_dir)
+ : Status::InternalError("spill store {}
is not ready",
+
_data_dir->path());
DBUG_EXECUTE_IF("fault_inject::spill_file::gc", {
status = Status::Error<INTERNAL_ERROR>("fault_inject spill_file gc
failed");
});
if (!status.ok()) {
LOG_EVERY_T(WARNING, 1) << fmt::format("failed to delete spill
data, dir {}, error: {}",
_spill_dir,
status.to_string());
}
+ _dir_created = false;
}
// Decrease spill data usage even if per-file cleanup failed. QueryContext
teardown deletes the
// whole query spill directory and retains failures for later retries.
- _data_dir->update_spill_data_usage(-_total_written_bytes);
+ _data_dir->release(_total_written_bytes);
Review Comment:
When remote deletion fails, the objects are still kept in object storage,
but this unconditionally releases their bytes from
`spill_s3_storage_limit_bytes`. The pending query-directory retry stores only a
path, so a prolonged deletion outage lets each failed query release and reuse
the quota while retained objects accumulate without bound. Please transfer the
reservation to pending cleanup and release it only after deletion succeeds,
deduplicating per-file bytes, and extend the retry test to assert capacity
remains charged.
##########
cloud/src/meta-service/meta_service.cpp:
##########
@@ -3530,6 +3530,143 @@ void
MetaServiceImpl::get_tablet_stats(::google::protobuf::RpcController* contro
}
}
+void MetaServiceImpl::report_spill_stats(::google::protobuf::RpcController*
controller,
+ const ReportSpillStatsRequest*
request,
+ ReportSpillStatsResponse* response,
+ ::google::protobuf::Closure* done) {
+ RPC_PREPROCESS(report_spill_stats, put);
+ instance_id = get_instance_id(resource_mgr_, request->cloud_unique_id());
+ if (instance_id.empty()) {
+ code = MetaServiceCode::INVALID_ARGUMENT;
+ msg = "empty instance_id";
+ LOG(INFO) << msg << ", cloud_unique_id=" << request->cloud_unique_id();
+ return;
+ }
+ RPC_RATE_LIMIT(report_spill_stats)
+ if (!request->has_stats() || request->stats().cloud_unique_id().empty()) {
+ code = MetaServiceCode::INVALID_ARGUMENT;
+ msg = "spill stats without cloud_unique_id";
+ return;
+ }
+ const auto& spill_stats = request->stats();
+ if (spill_stats.boot_id() <= 0 || spill_stats.remote_write_bytes() < 0 ||
+ spill_stats.remote_put_requests() < 0) {
+ code = MetaServiceCode::INVALID_ARGUMENT;
+ msg = fmt::format("invalid spill stats, boot_id={} write_bytes={}
put_requests={}",
+ spill_stats.boot_id(),
spill_stats.remote_write_bytes(),
+ spill_stats.remote_put_requests());
+ return;
+ }
+
+ TxnErrorCode err = txn_kv_->create_txn(&txn);
+ if (err != TxnErrorCode::TXN_OK) {
+ code = cast_as<ErrCategory::CREATE>(err);
+ msg = fmt::format("failed to create txn, err={}", err);
+ return;
+ }
+ // One record per BE. The report carries totals since boot, so a report of
the same boot
+ // replaces the previous one (a retry cannot double count). The first
report of a new boot
+ // folds the previous process' totals into prior_boots_*, which keeps the
record count
+ // bounded by the number of BEs instead of the number of BE restarts.
+ std::string key = stats_spill_key({instance_id,
spill_stats.cloud_unique_id()});
+ std::string existing_val;
+ err = txn->get(key, &existing_val);
+ if (err != TxnErrorCode::TXN_OK && err != TxnErrorCode::TXN_KEY_NOT_FOUND)
{
+ code = cast_as<ErrCategory::READ>(err);
+ msg = fmt::format("failed to get spill stats, err={}", err);
+ return;
+ }
+ SpillStatsPB value = spill_stats;
+ if (err == TxnErrorCode::TXN_OK) {
+ SpillStatsPB existing;
+ if (!existing.ParseFromString(existing_val)) {
+ code = MetaServiceCode::PROTOBUF_PARSE_ERR;
+ msg = fmt::format("malformed spill stats, key={}", hex(key));
+ return;
+ }
+ int64_t prior_bytes = existing.prior_boots_write_bytes();
+ int64_t prior_requests = existing.prior_boots_put_requests();
+ if (existing.boot_id() != spill_stats.boot_id()) {
+ prior_bytes += existing.remote_write_bytes();
+ prior_requests += existing.remote_put_requests();
+ } else {
+ // Totals of one process only grow; a late-delivered duplicate
must not roll back.
+ value.set_remote_write_bytes(
+ std::max(existing.remote_write_bytes(),
spill_stats.remote_write_bytes()));
+ value.set_remote_put_requests(
+ std::max(existing.remote_put_requests(),
spill_stats.remote_put_requests()));
+ }
+ value.set_prior_boots_write_bytes(prior_bytes);
+ value.set_prior_boots_put_requests(prior_requests);
+ }
+
value.set_update_time_ms(std::chrono::duration_cast<std::chrono::milliseconds>(
+
std::chrono::system_clock::now().time_since_epoch())
+ .count());
+ txn->put(key, value.SerializeAsString());
+ err = txn->commit();
+ if (err != TxnErrorCode::TXN_OK) {
+ code = cast_as<ErrCategory::COMMIT>(err);
+ msg = fmt::format("failed to commit spill stats, err={}
cloud_unique_id={} boot_id={}", err,
+ spill_stats.cloud_unique_id(),
spill_stats.boot_id());
+ return;
+ }
+}
+
+void MetaServiceImpl::get_spill_stats(::google::protobuf::RpcController*
controller,
+ const GetSpillStatsRequest* request,
+ GetSpillStatsResponse* response,
+ ::google::protobuf::Closure* done) {
+ RPC_PREPROCESS(get_spill_stats, get);
+ instance_id = get_instance_id(resource_mgr_, request->cloud_unique_id());
+ if (instance_id.empty()) {
+ code = MetaServiceCode::INVALID_ARGUMENT;
+ msg = "empty instance_id";
+ LOG(INFO) << msg << ", cloud_unique_id=" << request->cloud_unique_id();
+ return;
+ }
+ RPC_RATE_LIMIT(get_spill_stats)
+
+ TxnErrorCode err = txn_kv_->create_txn(&txn);
+ if (err != TxnErrorCode::TXN_OK) {
+ code = cast_as<ErrCategory::CREATE>(err);
+ msg = fmt::format("failed to create txn, err={}", err);
+ return;
+ }
+ std::string key0 = stats_spill_key_prefix(instance_id);
Review Comment:
This range is not bounded by the number of live BEs. Every add/replacement
generates a new identity and persistent stats key, ordinary node/cluster
removal never removes or folds it, and only whole-instance recycling clears the
range. A long-lived elastic instance therefore scans all identities ever
created in one transaction and returns every PB, eventually hitting
read/response limits although FE needs only the scalar lifetime total. Please
compact retired-node totals into a bounded instance aggregate with fenced
finalization and add churn coverage for bounded key/read cardinality.
##########
be/src/common/config.cpp:
##########
@@ -1610,6 +1610,13 @@ DEFINE_String(spill_storage_limit, "20%");
// 20%
DEFINE_mInt32(spill_gc_interval_ms, "2000"); // 2s
DEFINE_mInt32(spill_gc_work_time_ms, "2000"); // 2s
DEFINE_mInt64(spill_file_part_size_bytes, "1073741824"); // 1GB
+DEFINE_String(spill_storage_type, "local");
+DEFINE_Validator(spill_storage_type, [](const std::string& config) -> bool {
+ return config == "local" || config == "s3";
+});
+DEFINE_String(spill_s3_storage_vault, "");
+DEFINE_mInt64(spill_s3_storage_limit_bytes, "0");
Review Comment:
The documented unlimited sentinel is `0`, but this mutable config accepts
negative values too. `SpillDataDir::_reach_limit_unlocked` checks only `> 0`,
so setting `-1` at startup or dynamically silently disables the billable
object-storage cap and exposes a negative limit metric. Please add a validator
requiring `spill_s3_storage_limit_bytes >= 0` and cover rejection of a negative
update.
##########
be/src/exec/spill/spill_file_manager.cpp:
##########
@@ -330,6 +571,61 @@ SpillDataDir::SpillDataDir(std::string path, int64_t
capacity_bytes,
INT_GAUGE_METRIC_REGISTER(spill_data_dir_metric_entity,
spill_disk_has_spill_gc_data);
}
+Status SpillDataDir::ensure_ready() {
+ if (!_is_remote || ready()) {
+ return Status::OK();
+ }
+ std::lock_guard<std::mutex> lock(_init_mutex);
+ if (ready()) {
+ return Status::OK();
+ }
+ if (!config::is_cloud_mode()) {
+ return Status::InternalError("spill to s3 is only supported in cloud
mode");
+ }
+ if (config::cloud_unique_id.empty()) {
+ return Status::InternalError(
+ "spill to s3 is not ready: cloud_unique_id is empty, waiting
for FE heartbeat");
+ }
+ // Resolve from what is already known locally; never trigger a
meta-service sync here.
+ // The vault refresh thread and the heartbeat fill these in, and callers
retry.
+ auto& engine = ExecEnv::GetInstance()->storage_engine().to_cloud();
+ std::string vault_id = _vault_id.empty() ? engine.default_vault_id() :
_vault_id;
+ io::RemoteFileSystemSPtr fs =
+ vault_id.empty() ? engine.latest_fs() :
doris::get_filesystem(vault_id);
+ if (fs == nullptr) {
+ return Status::InternalError(
+ "spill to s3 is not ready: storage vault '{}' not found (empty
means the default "
+ "vault of the instance; set spill_s3_storage_vault to the
vault ID in be.conf if "
+ "the instance has no default vault)",
+ vault_id);
+ }
+ if (fs->type() != io::FileSystemType::S3) {
+ return Status::NotSupported("spill to s3 only supports S3 storage
vaults, vault '{}' is {}",
+ vault_id, fs->type());
+ }
+ init_remote_fs(fs, config::cloud_unique_id);
+ return Status::OK();
+}
+
+void SpillDataDir::init_remote_fs(io::FileSystemSPtr fs, const std::string&
cloud_unique_id) {
+ DCHECK(_is_remote);
+ _fs = std::move(fs);
+ _remote_be_root = fmt::format("{}/{}", SPILL_DIR_PREFIX, cloud_unique_id);
Review Comment:
`cloud_unique_id` is not unique per BE: `CloudSystemInfoService.addBackends`
generates one value before looping over all requested hosts, ResourceManager
supports multiple nodes with that value, and FE sends it to each BE. Two live
BEs can therefore both use `spill/C/{boot_id}`; when either starts cleanup it
classifies the other's active boot as old and deletes its spill objects. Their
reports also collide in one `(instance_id, C)` record and are repeatedly folded
as restarts. Please include a persisted per-node identity such as heartbeat
`backend_id` in both the object root and stats key, and test two simultaneously
live BEs, including existing duplicate cloud IDs.
##########
cloud/src/meta-service/meta_service.cpp:
##########
@@ -3530,6 +3530,143 @@ void
MetaServiceImpl::get_tablet_stats(::google::protobuf::RpcController* contro
}
}
+void MetaServiceImpl::report_spill_stats(::google::protobuf::RpcController*
controller,
+ const ReportSpillStatsRequest*
request,
+ ReportSpillStatsResponse* response,
+ ::google::protobuf::Closure* done) {
+ RPC_PREPROCESS(report_spill_stats, put);
+ instance_id = get_instance_id(resource_mgr_, request->cloud_unique_id());
+ if (instance_id.empty()) {
+ code = MetaServiceCode::INVALID_ARGUMENT;
+ msg = "empty instance_id";
+ LOG(INFO) << msg << ", cloud_unique_id=" << request->cloud_unique_id();
+ return;
+ }
+ RPC_RATE_LIMIT(report_spill_stats)
+ if (!request->has_stats() || request->stats().cloud_unique_id().empty()) {
+ code = MetaServiceCode::INVALID_ARGUMENT;
+ msg = "spill stats without cloud_unique_id";
+ return;
+ }
+ const auto& spill_stats = request->stats();
+ if (spill_stats.boot_id() <= 0 || spill_stats.remote_write_bytes() < 0 ||
+ spill_stats.remote_put_requests() < 0) {
+ code = MetaServiceCode::INVALID_ARGUMENT;
+ msg = fmt::format("invalid spill stats, boot_id={} write_bytes={}
put_requests={}",
+ spill_stats.boot_id(),
spill_stats.remote_write_bytes(),
+ spill_stats.remote_put_requests());
+ return;
+ }
+
+ TxnErrorCode err = txn_kv_->create_txn(&txn);
+ if (err != TxnErrorCode::TXN_OK) {
+ code = cast_as<ErrCategory::CREATE>(err);
+ msg = fmt::format("failed to create txn, err={}", err);
+ return;
+ }
+ // One record per BE. The report carries totals since boot, so a report of
the same boot
+ // replaces the previous one (a retry cannot double count). The first
report of a new boot
+ // folds the previous process' totals into prior_boots_*, which keeps the
record count
+ // bounded by the number of BEs instead of the number of BE restarts.
+ std::string key = stats_spill_key({instance_id,
spill_stats.cloud_unique_id()});
+ std::string existing_val;
+ err = txn->get(key, &existing_val);
+ if (err != TxnErrorCode::TXN_OK && err != TxnErrorCode::TXN_KEY_NOT_FOUND)
{
+ code = cast_as<ErrCategory::READ>(err);
+ msg = fmt::format("failed to get spill stats, err={}", err);
+ return;
+ }
+ SpillStatsPB value = spill_stats;
+ if (err == TxnErrorCode::TXN_OK) {
+ SpillStatsPB existing;
+ if (!existing.ParseFromString(existing_val)) {
+ code = MetaServiceCode::PROTOBUF_PARSE_ERR;
+ msg = fmt::format("malformed spill stats, key={}", hex(key));
+ return;
+ }
+ int64_t prior_bytes = existing.prior_boots_write_bytes();
+ int64_t prior_requests = existing.prior_boots_put_requests();
+ if (existing.boot_id() != spill_stats.boot_id()) {
Review Comment:
`!=` does not establish that this is a newer boot. Starting with boot
1000/current 300, boot 1001/current 80 stores total 380; a late
boot-1000/current-300 report then stores 680, and the next boot-1001 report
raises it to 760. Transaction conflicts serialize overlap but do not reject
sequential stale requests. Please use independently idempotent per-generation
records, or a durable monotonic fencing token that ignores older generations;
`UnixMillis()` alone is unsafe because wall time can move backward or collide.
Add an old-after-new-after-current delivery test.
##########
be/src/exec/spill/spill_file_manager.cpp:
##########
@@ -308,6 +438,99 @@ void SpillFileManager::gc(int32_t max_work_time_ms) {
}
}
+void SpillFileManager::_remote_gc(SpillDataDir* store) {
+ if (!store->ready()) {
+ // Retry about once a minute at the default 2s GC interval.
ensure_ready() only reads
+ // what the vault refresh thread and the FE heartbeat already brought
in.
+ if (_remote_not_ready_rounds++ % 30 != 0) {
+ return;
+ }
+ auto st = store->ensure_ready();
+ if (!st.ok()) {
+ LOG(WARNING) << "remote spill store is not ready yet: " << st;
+ return;
+ }
+ }
+ _report_remote_spill_stats(store);
+ if (!remote_startup_cleanup_pending()) {
+ return;
+ }
+ auto st = _remote_startup_cleanup(store);
+ if (st.ok()) {
+ _remote_startup_cleanup_pending.store(false,
std::memory_order_release);
+ } else {
+ LOG_EVERY_T(WARNING, 60) << "failed to clean up spill objects of
previous boots, will "
+ "retry: "
+ << st;
+ }
+}
+
+void SpillFileManager::flush_remote_spill_stats() {
+ for (auto& [path, store] : _spill_store_map) {
+ if (store->is_remote()) {
+ _report_remote_spill_stats(store.get(), /*final_report=*/true);
+ }
+ }
+}
+
+void SpillFileManager::_report_remote_spill_stats(SpillDataDir* store, bool
final_report) {
+ std::lock_guard<std::mutex> lock(_remote_report_mutex);
+ // About once a minute at the default 2s GC interval; a final report skips
the cadence.
Review Comment:
These counters are process-local and reported only about once per minute;
the final flush executes only on graceful shutdown. If the BE crashes or the
host is lost after successful uploads but before the next report, the restarted
BE has no durable checkpoint and meta-service can preserve only the older
value, so the missing suffix is permanent despite the proto's
no-loss-across-restarts and billing-input contract. Please make the
acknowledged total recoverable across abrupt restarts, or explicitly label it
best-effort and remove the no-loss guarantee; add a crash-before-next-report
test.
##########
be/src/exec/spill/spill_file_writer.cpp:
##########
@@ -90,44 +145,179 @@ 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 = _file_writer->append(_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 (_resource_ctx) {
-
_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);
- }
- _data_dir->update_spill_data_usage(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);
+ 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);
+ }
}
- RETURN_IF_ERROR(_file_writer->close());
- _file_writer.reset();
+ // Issue a non-blocking close so that the upload of this part overlaps
with the next
+ // one. The part is confirmed later by _reap_closing_parts(), which also
reconciles
+ // the upload budget, so it is queued even when something above failed.
+ ClosingPart part;
+ part.path = _current_part_path;
+ part.part_index = _current_part_index;
+ part.part_bytes = _part_written_bytes;
+ part.ledger = std::move(_part_ledger);
+ part.stats = std::move(_part_stats);
+ part.close_status = status.ok() ? _file_writer->close(/*non_block=*/true)
: status;
+ part.writer = std::move(_file_writer);
+ _closing_parts.emplace_back(std::move(part));
// Advance to next part
++_current_part_index;
- if (spill_file) {
- spill_file->increment_part_count();
- }
_part_written_blocks = 0;
_part_written_bytes = 0;
_part_max_sub_block_size = 0;
_part_meta.clear();
+ return status;
+}
+
+Status SpillFileWriter::_reap_closing_parts(bool block,
+ const std::shared_ptr<SpillFile>&
spill_file) {
+ Status first_error;
+ while (!_closing_parts.empty()) {
+ auto& part = _closing_parts.front();
+ Status st = part.close_status;
+ if (st.ok()) {
+ st = part.writer->try_finish_close();
+ if (st.is<ErrorCode::NEED_SEND_AGAIN>()) {
+ if (!block) {
+ break;
+ }
+ st = part.writer->close();
+ } else if (st.is<ErrorCode::NOT_IMPLEMENTED_ERROR>()) {
+ // Writers without an async close protocol (local files)
finished the work
+ // in close(true); close() only flips the state.
+ st = part.writer->state() == io::FileWriter::State::CLOSED ?
Status::OK()
+ :
part.writer->close();
+ }
+ }
+ st = _finish_part(part, spill_file, st);
+ _closing_parts.erase(_closing_parts.begin());
+ if (!st.ok() && first_error.ok()) {
+ first_error = st;
+ if (!block) {
+ break;
+ }
+ }
+ }
+ return first_error;
+}
+
+Status SpillFileWriter::_finish_part(ClosingPart& part,
+ const std::shared_ptr<SpillFile>&
spill_file,
+ Status close_status) {
+ MultipartUploadId upload = _multipart_upload_id(part.writer.get());
+ if (!close_status.ok() && part.writer != nullptr &&
+ part.writer->state() != io::FileWriter::State::CLOSED) {
+ // The part never reached a final state (footer failed, or close(true)
could not be
+ // issued). Destroying the writer waits for every in-flight upload, so
the ledger below
+ // is complete afterwards. close() is not used for this: on a
cancelled query it would
+ // be refused by the upload gate right away and drain nothing.
+ part.writer.reset();
+ }
+
+ // Budget: everything the gate took for this part but the upload callback
never gave
+ // back (buffers that failed before their upload started) is released
here. The writer is
+ // in its final state at this point, so every callback that will ever fire
has fired.
+ if (_budget != nullptr && part.ledger != nullptr) {
+ int64_t remaining = part.ledger->acquired.load() -
part.ledger->released.load();
+ DCHECK_GE(remaining, 0) << "upload callback released more than
acquired, part="
+ << part.path;
+ if (remaining > 0) {
+ _budget->release(remaining);
+ }
+ if (_remote_upload_wait_timer != nullptr) {
+ COUNTER_UPDATE(_remote_upload_wait_timer,
part.ledger->wait_ns.load());
+ }
+ }
+
+ if (part.stats != nullptr) {
+ int64_t requests = part.stats->total_requests();
+ int64_t data_requests = part.stats->put_object_requests +
part.stats->upload_part_requests;
+ int64_t uploaded_bytes = part.stats->uploaded_bytes;
+ if (_remote_write_requests != nullptr) {
+ COUNTER_UPDATE(_remote_write_requests, requests);
+ COUNTER_UPDATE(_remote_upload_part_requests, data_requests);
+ COUNTER_UPDATE(_remote_upload_bytes, uploaded_bytes);
+ COUNTER_UPDATE(_remote_upload_timer,
part.stats->request_time_ns.load());
+ }
+ if (_resource_ctx) {
+
_resource_ctx->io_context()->update_spill_remote_write_requests(requests);
+ }
+
ExecEnv::GetInstance()->spill_file_mgr()->update_spill_remote_write(uploaded_bytes,
+
data_requests);
+ }
+
+ if (!close_status.ok()) {
+ LOG(WARNING) << "failed to close spill part " << part.path << ": " <<
close_status;
+ _abort_multipart_upload(upload);
+ return close_status;
+ }
+ if (spill_file) {
+ spill_file->add_part(part.part_bytes);
+ }
return Status::OK();
}
+SpillFileWriter::MultipartUploadId
SpillFileWriter::_multipart_upload_id(io::FileWriter* writer) {
+ auto* s3_writer = dynamic_cast<io::S3FileWriter*>(writer);
+ if (s3_writer == nullptr || s3_writer->upload_id().empty()) {
+ return {};
+ }
+ return {.path = s3_writer->path().native(),
+ .bucket = s3_writer->bucket(),
+ .key = s3_writer->key(),
+ .upload_id = s3_writer->upload_id()};
+}
+
+void SpillFileWriter::_abort_multipart_upload(const MultipartUploadId& upload)
{
+ if (upload.upload_id.empty()) {
+ return;
+ }
+ auto s3_fs = std::dynamic_pointer_cast<io::S3FileSystem>(_data_dir->fs());
+ if (s3_fs == nullptr) {
+ return;
+ }
+ auto client = s3_fs->client_holder()->get();
+ if (client == nullptr) {
+ return;
+ }
+ auto resp = client->abort_multipart_upload(
Review Comment:
This one-shot best-effort abort is the only owner of the upload ID. If it
fails, or the BE exits before reaching it, the uploaded multipart parts are
invisible to `list()`/`delete_directory()`, while the new recycler deletes only
visible objects by prefix; Doris cannot reclaim that billed residue. Please
persist/retry upload IDs, teach the recycler to list and abort expired
multipart uploads, or require and verify a provider lifecycle rule; test abort
failure plus restart residue.
##########
be/src/exec/spill/spill_file_manager.cpp:
##########
@@ -308,6 +438,99 @@ void SpillFileManager::gc(int32_t max_work_time_ms) {
}
}
+void SpillFileManager::_remote_gc(SpillDataDir* store) {
+ if (!store->ready()) {
+ // Retry about once a minute at the default 2s GC interval.
ensure_ready() only reads
+ // what the vault refresh thread and the FE heartbeat already brought
in.
+ if (_remote_not_ready_rounds++ % 30 != 0) {
+ return;
+ }
+ auto st = store->ensure_ready();
+ if (!st.ok()) {
+ LOG(WARNING) << "remote spill store is not ready yet: " << st;
+ return;
+ }
+ }
+ _report_remote_spill_stats(store);
+ if (!remote_startup_cleanup_pending()) {
+ return;
+ }
+ auto st = _remote_startup_cleanup(store);
+ if (st.ok()) {
+ _remote_startup_cleanup_pending.store(false,
std::memory_order_release);
+ } else {
+ LOG_EVERY_T(WARNING, 60) << "failed to clean up spill objects of
previous boots, will "
+ "retry: "
+ << st;
+ }
+}
+
+void SpillFileManager::flush_remote_spill_stats() {
+ for (auto& [path, store] : _spill_store_map) {
+ if (store->is_remote()) {
+ _report_remote_spill_stats(store.get(), /*final_report=*/true);
+ }
+ }
+}
+
+void SpillFileManager::_report_remote_spill_stats(SpillDataDir* store, bool
final_report) {
+ std::lock_guard<std::mutex> lock(_remote_report_mutex);
+ // About once a minute at the default 2s GC interval; a final report skips
the cadence.
+ if (!final_report && _remote_report_rounds++ % 30 != 0) {
+ return;
+ }
+ if (!store->ready() || !config::is_cloud_mode()) {
+ return;
+ }
+ int64_t write_bytes = remote_write_bytes_since_boot();
+ int64_t put_requests = remote_put_requests_since_boot();
+ if (write_bytes == _reported_remote_write_bytes &&
+ put_requests == _reported_remote_put_requests) {
+ return;
+ }
+ auto st =
ExecEnv::GetInstance()->storage_engine().to_cloud().meta_mgr().report_spill_stats(
+ store->boot_id(), write_bytes, put_requests);
+ if (!st.ok()) {
+ LOG_EVERY_T(WARNING, 60) << "failed to report spill stats to
meta-service"
+ << (final_report ? "" : ", will retry") << ":
" << st;
+ return;
+ }
+ _reported_remote_write_bytes = write_bytes;
+ _reported_remote_put_requests = put_requests;
+}
+
+Status SpillFileManager::_remote_startup_cleanup(SpillDataDir* store) {
+ auto fs = store->fs();
+ const auto& be_root = store->get_remote_be_root();
+ const auto current_boot_id = std::to_string(store->boot_id());
+
+ MonotonicStopWatch watch;
+ watch.start();
+ std::vector<io::FileInfo> files;
+ bool exists = false;
+ RETURN_IF_ERROR(fs->list(be_root, true, &files, &exists));
Review Comment:
This eager recursive `list` materializes every object below the BE root
although cleanup needs only distinct first-level boot IDs.
`S3FileSystem::list_impl` first drains the paginated iterator into a complete
`ObjectMeta` vector and then builds a complete `FileInfo` vector, including
current-boot parts. After failed cleanup/restarts this can be arbitrarily
large; it ignores the GC time budget and blocks the sole spill-GC thread, which
`stop()` joins. Please use streaming/delimiter-based generation discovery and
bound deletion across GC rounds, with a multi-page residue test.
##########
be/src/exec/spill/spill_remote_upload_budget.cpp:
##########
@@ -0,0 +1,88 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+#include "exec/spill/spill_remote_upload_budget.h"
+
+#include <glog/logging.h>
+
+#include <chrono>
+
+#include "util/stopwatch.hpp"
+
+namespace doris {
+
+Status SpillRemoteUploadBudget::acquire(int64_t bytes, const
std::function<bool()>& is_cancelled,
+ int64_t* wait_ns) {
+ MonotonicStopWatch watch;
+ watch.start();
+ std::unique_lock<std::mutex> lock(_mutex);
+ while (_inflight_bytes > 0 && _inflight_bytes + bytes > _limit_bytes) {
+ if (is_cancelled && is_cancelled()) {
+ return Status::Cancelled("query cancelled while waiting for spill
upload budget");
+ }
+ _cv.wait_for(lock, std::chrono::milliseconds(100));
+ }
+ _inflight_bytes += bytes;
+ _total_acquired_bytes += bytes;
+ if (wait_ns != nullptr) {
+ *wait_ns = static_cast<int64_t>(watch.elapsed_time());
+ }
+ return Status::OK();
+}
+
+void SpillRemoteUploadBudget::release(int64_t bytes) {
+ {
+ std::lock_guard<std::mutex> lock(_mutex);
+ _inflight_bytes -= bytes;
+ _total_released_bytes += bytes;
+ DCHECK_GE(_inflight_bytes, 0) << "spill upload budget released more
than acquired";
Review Comment:
This is the defensive-continue pattern prohibited by the repository's
assert-correctness rule. In a release build `DCHECK_GE` disappears, so a double
release is silently clamped to zero while `total_released_bytes` remains
greater than `total_acquired_bytes`; the limiter then wakes waiters after
losing its core invariant. Please replace the debug check plus clamp with an
always-on `DORIS_CHECK_GE` (or established equivalent) and add an over-release
death/invariant test.
##########
be/src/cloud/cloud_meta_mgr.cpp:
##########
@@ -1893,7 +1900,30 @@ Status CloudMetaMgr::finish_restore_job(const int64_t
tablet_id, bool is_complet
});
}
-Status CloudMetaMgr::get_storage_vault_info(StorageVaultInfos* vault_infos,
bool* is_vault_mode) {
+Status CloudMetaMgr::report_spill_stats(int64_t boot_id, int64_t
remote_write_bytes,
+ int64_t remote_put_requests) {
+ ReportSpillStatsRequest req;
+ ReportSpillStatsResponse resp;
+ req.set_cloud_unique_id(config::cloud_unique_id);
+ auto* stats = req.mutable_stats();
+ stats->set_cloud_unique_id(config::cloud_unique_id);
Review Comment:
The reporter identity is reread from mutable `config::cloud_unique_id` on
every call, but boot ID and counters are cumulative for the unchanged process.
Dropping and re-adding the same live endpoint gives it a new FE-generated ID:
after reporting 100 under C1, its next cumulative report of 120 creates C2
while C1 remains, so `get_spill_stats` returns 220. Please keep an immutable
stats identity for the process generation while routing with the current outer
ID, or make generations independently idempotent in meta-service; add a
same-boot C1-to-C2 test.
##########
be/src/io/fs/s3_file_writer.cpp:
##########
@@ -93,13 +100,75 @@ S3FileWriter::~S3FileWriter() {
s3_file_being_written << -1;
}
+void S3FileWriter::_record_request(std::atomic<int64_t> RemoteWriteStats::*
counter,
+ int64_t elapsed_ns, bool ok, int64_t
uploaded_bytes) {
+ if (_remote_write_stats == nullptr) {
+ return;
+ }
+ ((*_remote_write_stats).*counter)++;
Review Comment:
This increments once per aggregate SDK call, not per object-store request
attempt. The AWS client uses `S3CustomRetryStrategy(max_s3_client_retry)`, so a
throttled `PutObject`/`UploadPart` that succeeds on its third attempt sends
three billable requests but contributes one here; an exhausted operation is
likewise under-counted. `_finish_part()` persists this value as the warehouse
PUT-request total. Please attribute initial and retry attempts at the SDK
request boundary, including terminal failures, and cover retry success and
exhaustion for both operations.
##########
be/src/exec/spill/spill_file_manager.cpp:
##########
@@ -123,6 +213,24 @@ void SpillFileManager::_spill_gc_thread_callback() {
for (auto& [path, dir] : _spill_store_map) {
static_cast<void>(dir->update_capacity());
}
+ // Both configs are mutable; observe changes without a restart.
+ int64_t budget_limit = config::spill_s3_max_inflight_upload_bytes;
Review Comment:
Startup treats `limit < 2 * s3_write_buffer_size` as fatal, but this runtime
path installs exactly that invalid value after only a warning; both operands
are mutable and the update can be persisted. `acquire()` then admits one
oversized buffer whenever inflight reaches zero, so the live setting can exceed
its documented bound, and the persisted configuration prevents the next
S3-spill restart. Please enforce the same cross-config invariant for dynamic
updates, including buffer-size changes, or make startup accept the documented
degraded mode; test live-update/restart parity.
--
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]
