github-actions[bot] commented on code in PR #68032: URL: https://github.com/apache/doris/pull/68032#discussion_r4091727722
########## be/src/exec/spill/remote_spill_data_dir.cpp: ########## @@ -0,0 +1,133 @@ +// 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/remote_spill_data_dir.h" + +#include <glog/logging.h> + +#include <utility> + +#include "cloud/cloud_storage_engine.h" +#include "cloud/config.h" +#include "common/config.h" +#include "common/logging.h" +#include "common/metrics/metrics.h" +#include "io/fs/remote_file_system.h" +#include "runtime/exec_env.h" +#include "service/backend_options.h" +#include "storage/olap_define.h" +#include "storage/storage_policy.h" +#include "util/pretty_printer.h" + +namespace doris { + +RemoteSpillDataDir::RemoteSpillDataDir(std::string vault_id) + : SpillDataDir(fmt::format("s3:{}", vault_id.empty() ? "default" : vault_id), + /*spill_root=*/"", + fmt::format("s3:{}", vault_id.empty() ? "default" : vault_id), + /*capacity_bytes=*/0, TStorageMedium::S3), + _vault_id(std::move(vault_id)) {} + +Status RemoteSpillDataDir::init() { + RETURN_IF_ERROR(update_capacity()); + LOG(INFO) << fmt::format("remote spill store registered, vault_id={}, limit={}", + _vault_id.empty() ? "<default>" : _vault_id, + PrettyPrinter::print_bytes(_spill_data_limit_bytes)); + return Status::OK(); +} + +Status RemoteSpillDataDir::ensure_ready() { + if (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"); + } + const std::string& host = BackendOptions::get_localhost(); + if (host.empty()) { + return Status::InternalError("spill to s3 is not ready: the address of this BE is unknown"); + } + std::string endpoint = fmt::format("{}_{}", host, config::heartbeat_service_port); Review Comment: [P1] Include instance ownership in the remote spill prefix. Storage vaults may be shared by multiple instances, but this root is only `{host}_{heartbeat_port}`. If two instances use overlapping private BE addresses, they write the same `spill/...` root; then either BE's first `_remote_startup_cleanup()` subtracts only its process-local query registry and deletes the other instance's live query directories as residue. Please namespace the root by an immutable instance identity (and a stable BE generation), and add a shared-vault/same-endpoint test. ########## cloud/src/recycler/recycler.cpp: ########## @@ -1000,6 +1003,14 @@ int InstanceRecycler::recycle_deleted_instance_data() { instance_info().snapshot_switch_status() != SnapshotSwitchStatus::SNAPSHOT_SWITCH_DISABLED; if (snapshot_enabled) { + // Only referenced rowsets are recycled selectively below because the vault may be + // shared with other instances. Spill objects (spill/{ip}_{port}/...) do not say which + // instance wrote them, so only the expired ones are removed, as for a live instance. + if (recycle_expired_spill_objects() != 0) { Review Comment: [P1] Do not finish deleted-instance data cleanup while fresh spill remains. This age-filtered pass returns success after skipping groups newer than the TTL, but the caller then advances to metadata cleanup; later recycler initialization skips accessors, and metadata cleanup removes the instance's vault keys. A normally deleted instance can therefore orphan all recent spill indefinitely. The new test hides this by setting `force_immediate_recycle=true`. Please retain a durable cleanup phase/owner until no owned spill remains, or add instance identity to the spill namespace so deletion can safely remove that instance's prefix, and cover the default-TTL case. ########## be/src/exec/operator/spill_counters.h: ########## @@ -33,6 +33,11 @@ struct SpillWriteCounters { RuntimeProfile::Counter* spill_write_block_count = nullptr; RuntimeProfile::Counter* spill_write_block_data_size = nullptr; RuntimeProfile::Counter* spill_write_rows_count = nullptr; + RuntimeProfile::Counter* spill_remote_write_requests = nullptr; Review Comment: [P2] Register these counters for Iceberg spill as well. `SpillIcebergTableSinkLocalState::_init_spill_counters()` manually duplicates the old read/write setup, and `VIcebergSortWriter` uses the same `SpillFileWriter`/`SpillFileReader` under the global S3 spill selection. Because none of the new remote names are registered there, the optional lookups return null and S3 Iceberg spilling silently loses all remote request/upload/time counters from its operator profile. Please reuse these shared initializers (or update the parallel initializer) and cover that path. ########## be/src/runtime/workload_management/resource_context.cpp: ########## @@ -53,6 +53,10 @@ void ResourceContext::to_thrift_query_statistics(TQueryStatistics* statistics) c io_context_->spill_write_bytes_to_local_storage()); statistics->__set_spill_read_bytes_from_local_storage( io_context_->spill_read_bytes_from_local_storage()); + statistics->__set_spill_write_bytes_to_remote_storage( Review Comment: [P2] Propagate the remote pair through the existing live-stat consumers. These fields are returned by each BE, but `ProfileManager.getQueryStatistic()` never aggregates them and `QueryProfileAction` exposes only the local pair; `show proc /current_queries` and `information_schema.backend_active_tasks` likewise retain only local spill columns. Consequently an active query spilling entirely to S3 appears to have zero/no spill in the same observability paths that report local spill. Please update the aggregation and corresponding REST/proc/schema-table mappings and tests. ########## be/src/exec/spill/spill_file_writer.cpp: ########## @@ -175,13 +326,15 @@ Status SpillFileWriter::close() { } _closed = true; + auto spill_file = _spill_file_wptr.lock(); Review Comment: [P1] Keep the `SpillFile` alive through writer close. Repartition output files live in a stack vector, but their writers persist in the local state's `SpillRepartitioner`; any post-`setup_output()` error bypasses `finalize()`, destroys the files first, and runs `gc()`. When those writers are later destroyed, this lock is null, yet `_close_current_part()` force-reserves/appends the footer and cannot attach those bytes to an accounting owner, permanently leaking spill capacity (and S3 close may publish after the earlier file-prefix delete). Please use strong writer-to-file ownership or explicitly close/abort and clear all writers before the caller's file vector can unwind, with an injected repartition-error test. ########## fe/fe-core/src/main/java/org/apache/doris/plugin/AuditEvent.java: ########## @@ -104,6 +104,10 @@ public enum EventType { public long spillWriteBytesToLocalStorage = -1; @AuditField(value = "SpillReadBytesFromLocalStorage", colName = "spill_read_bytes_from_local_storage") public long spillReadBytesFromLocalStorage = -1; + @AuditField(value = "SpillWriteBytesToRemoteStorage", colName = "spill_write_bytes_to_remote_storage") Review Comment: [P2] Initialize the HTTP-plan audit producer too. These fields default to -1, while `TableQueryPlanAction.addToAuditLog()` explicitly sets the analogous local-spill and local/remote scan fields to zero but never calls the new remote setters. HTTP plan generation has no BE query statistics to overwrite them, so its persisted audit rows report -1/-1 for a zero-spill operation. Please set both fields to zero there and cover the serialized HTTP-plan audit output. ########## cloud/src/recycler/recycler.cpp: ########## @@ -7828,6 +7839,144 @@ int InstanceRecycler::recycle_expired_stage_objects() { return ret; } +std::string InstanceRecycler::spill_object_prefix() const { + return "spill/"; +} + +std::optional<int64_t> InstanceRecycler::spill_objects_expiration_time() const { + if (config::force_immediate_recycle) { + return INT64_MAX; + } + if (config::spill_objects_expire_time_second <= 0) { + // A non-positive TTL would select objects of running queries; treat it as "disabled". + return std::nullopt; + } + return duration_cast<seconds>(system_clock::now().time_since_epoch()).count() - + config::spill_objects_expire_time_second; +} + +int InstanceRecycler::list_expired_spill_groups(StorageVaultAccessor& accessor, + int64_t expiration_time, + std::vector<ExpiredSpillGroup>* groups) { + // Objects are written by BE under "{vault prefix}/spill/{ip}_{port}/...", and a live BE + // rewrites "spill/{ip}_{port}/_heartbeat" every hour. A BE directory is expired only when + // nothing in it changed for the whole TTL: objects of a long query of a live BE are kept + // however old they are, since the heartbeat keeps the directory fresh. The keys do not name + // the instance, so in a vault shared with other instances the sweep also removes the + // directories of their dead BEs. + const std::string prefix = spill_object_prefix(); + std::unique_ptr<ListIterator> list_iter; + if (accessor.list_directory(prefix, &list_iter) != 0) { Review Comment: [P1] Avoid a full shared-vault spill scan per instance. Every normal `InstanceRecycler` runs this hourly and this iterator exhausts `list_directory("spill/")`. Because the code explicitly supports physical vaults shared by instances, N instances independently list the same M objects, potentially concurrently, with no time/page budget even when every heartbeat is fresh. This multiplies billable LIST traffic and can monopolize recycler workers, delaying all other cleanup. Please assign one leased sweep owner per physical vault (or namespace/index spill by instance/BE), bound or checkpoint pagination, and add a multi-page shared-vault test. ########## be/src/exec/spill/spill_file_writer.cpp: ########## @@ -90,42 +142,136 @@ 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 (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); + } } - if (_write_file_current_size) { - COUNTER_UPDATE(_write_file_current_size, meta_size); + + // Close synchronously. Spilling already runs on an IO thread that blocks on every + // append, 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()) { + status = writer->close(); Review Comment: [P1] Do not rely on destructor-only close for Iceberg spill. Both `_do_spill()` and `_do_intermediate_merge()` return success without closing their local writer, while the destructor only logs a close failure. On S3 this line performs the final PutObject/CompleteMultipartUpload and can fail after every `write_block()` returned OK; `finish_writing()` is then skipped and the failed tail is absent from `_part_sizes`. The release reader can silently omit those rows, and intermediate merge has already deleted its input files. Please `RETURN_IF_ERROR(writer->close())` before resetting state or deleting inputs, remove the failed output on error, and add final-upload/completion fault tests. ########## fe/fe-core/src/main/java/org/apache/doris/cloud/catalog/RemoteSpillStatsPoller.java: ########## @@ -0,0 +1,159 @@ +// 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. + +package org.apache.doris.cloud.catalog; + +import org.apache.doris.catalog.Env; +import org.apache.doris.common.AnalysisException; +import org.apache.doris.common.Config; +import org.apache.doris.common.Pair; +import org.apache.doris.common.Status; +import org.apache.doris.common.util.MasterDaemon; +import org.apache.doris.proto.InternalService; +import org.apache.doris.rpc.BackendServiceProxy; +import org.apache.doris.system.Backend; +import org.apache.doris.thrift.TStatusCode; + +import com.google.common.annotations.VisibleForTesting; +import org.apache.logging.log4j.LogManager; +import org.apache.logging.log4j.Logger; + +import java.util.ArrayList; +import java.util.List; +import java.util.concurrent.Future; +import java.util.concurrent.TimeUnit; + +/** + * Polls the bytes of query spill held in object storage (spill_storage_type=s3) from the alive + * backends of all clusters, so that SHOW DATA reads them from memory. Runs on every FE on its own + * schedule (cloud_spill_stats_poll_interval_second): the freshness of the value must not depend on + * how long a round of the tablet stats takes. + */ +public class RemoteSpillStatsPoller extends MasterDaemon { + private static final Logger LOG = LogManager.getLogger(RemoteSpillStatsPoller.class); + + private static final int RPC_TIMEOUT_SECOND = 5; + + /** One successful poll: the value and when it was fetched. */ + private static final class RemoteSpillStats { + private final long bytes; + private final long fetchTimeMs; + + private RemoteSpillStats(long bytes, long fetchTimeMs) { + this.bytes = bytes; + this.fetchTimeMs = fetchTimeMs; + } + } + + // Summed over the alive BEs of all clusters. Null until the first successful poll. A BE that is + // gone no longer contributes: its leftover objects are removed by its restart or by the + // meta-service recycler and are not counted meanwhile. + private volatile RemoteSpillStats remoteSpillStats = null; + + public RemoteSpillStatsPoller() { + super("remote spill stats poller", pollIntervalMs()); + } + + private static long pollIntervalMs() { + return Math.max(1, Config.cloud_spill_stats_poll_interval_second) * 1000L; + } + + /** + * A value older than this is not served: the configured max age, but at least three poll + * intervals so that a longer interval cannot make every value stale. + */ + @VisibleForTesting + static long maxAgeSecond() { + return Math.max(Config.cloud_spill_stats_max_age_second, + 3L * Math.max(1, Config.cloud_spill_stats_poll_interval_second)); + } + + @Override + protected void runAfterCatalogReady() { + refresh(); + // The interval is mutable. + setInterval(pollIntervalMs()); + } + + private void refresh() { + List<Backend> backends; + try { + backends = Env.getCurrentSystemInfo().getAllBackendsByAllCluster().values().asList(); + } catch (AnalysisException e) { + LOG.warn("failed to list the backends for the remote spill stats", e); + return; + } + InternalService.PGetBeResourceRequest request = InternalService.PGetBeResourceRequest.newBuilder().build(); + List<Pair<Backend, Future<InternalService.PGetBeResourceResponse>>> futures = new ArrayList<>(); + for (Backend be : backends) { + if (!be.isAlive()) { + continue; + } + futures.add(Pair.of(be, BackendServiceProxy.getInstance() + .getBeResourceAsync(be.getBrpcAddress(), RPC_TIMEOUT_SECOND, request))); + } + // Any failure keeps the previous value: a partial sum would under-report a billing input, + // and getRemoteSpillBytes() refuses a value that stays stale for too long. + long totalBytes = 0; + for (Pair<Backend, Future<InternalService.PGetBeResourceResponse>> beFuture : futures) { + if (beFuture.second == null) { + LOG.warn("failed to send get_be_resource to backend {}", beFuture.first.getId()); + return; + } + try { + InternalService.PGetBeResourceResponse response = + beFuture.second.get(RPC_TIMEOUT_SECOND, TimeUnit.SECONDS); + if (!response.hasStatus() || new Status(response.getStatus()).getErrorCode() != TStatusCode.OK) { + LOG.warn("get_be_resource of backend {} failed: {}", beFuture.first.getId(), + response.hasStatus() ? response.getStatus().getErrorMsgsList() : "no status"); + return; + } + totalBytes += response.getGlobalBeResourceUsage().getRemoteSpillBytes(); + } catch (Exception e) { + LOG.warn("get_be_resource of backend {} failed", beFuture.first.getId(), e); + return; + } + } + remoteSpillStats = new RemoteSpillStats(totalBytes, System.currentTimeMillis()); + } + + @VisibleForTesting + void setRemoteSpillStatsForTest(long bytes, long fetchTimeMs) { + remoteSpillStats = new RemoteSpillStats(bytes, fetchTimeMs); + } + + /** + * Bytes of query spill currently held in object storage, as last polled from the backends. + * This is a billing input, so a value that is missing or older than maxAgeSecond() is reported + * as an error instead of being shown as current. + */ + public long getRemoteSpillBytes() throws AnalysisException { + RemoteSpillStats stats = remoteSpillStats; + if (stats == null) { + throw new AnalysisException("spill stats have not been polled from the backends yet"); + } + long ageSecond = (System.currentTimeMillis() - stats.fetchTimeMs) / 1000; Review Comment: [P2] Use a monotonic clock for this freshness limit. `refresh()` records `System.currentTimeMillis()`, and this subtracts another adjustable wall-clock reading. If NTP or an operator moves the FE clock backward while subsequent polls fail, `ageSecond` becomes negative and the stale billing sample remains valid until wall time catches up plus the configured maximum age. Please track elapsed age with `System.nanoTime()` or an injected monotonic ticker (keeping wall time only for display) and test a rollback with repeated refresh failures. -- 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]
