github-actions[bot] commented on code in PR #65788:
URL: https://github.com/apache/doris/pull/65788#discussion_r3610880457
##########
be/src/runtime/query_cache/query_cache.cpp:
##########
@@ -282,8 +297,333 @@ bool QueryCacheRuntime::_try_prepare_incremental(const
std::vector<TScanRangePar
return true;
}
-bool QueryCacheRuntime::_capture_tablet_delta(int64_t tablet_id, int64_t
cached_version,
- QueryCacheInstanceDecision*
decision) {
+std::unordered_map<int64_t, std::string>
QueryCacheRuntime::_presync_cloud_delta_tablets(
+ const std::vector<TScanRangeParams>& scan_ranges, int64_t
current_version) {
+ std::unordered_map<int64_t, std::string> fallback_reasons;
+
+ // (c) A non-positive budget disables cloud incremental merge: return
WITHOUT
+ // launching any work. The previous code launched the fan-out and abandoned
+ // it on an instantly expired wait, spawning tasks whose only effect was to
+ // be orphaned. Fall every scanned tablet back so the whole instance
+ // recomputes in full. Checked before touching the engine so the semantics
+ // (and its unit test) do not depend on engine state.
+ if (config::query_cache_decision_sync_timeout_ms <= 0) {
+ for (const auto& scan_range : scan_ranges) {
+ fallback_reasons[scan_range.scan_range.palo_scan_range.tablet_id] =
+ "cloud incremental sync disabled";
+ }
+ return fallback_reasons;
+ }
+
+ // Reached only under config::is_cloud_mode() (the caller gates on it), but
+ // cloud_unique_id is a mutable config, so is_cloud_mode() can flip true
on a
+ // live LOCAL deployment whose engine stays a local StorageEngine. Degrade
+ // gracefully rather than aborting in the hard CHECK inside to_cloud():
fall
+ // every tablet back, mirroring the per-tablet cloud-cast fallback below.
+ auto* engine =
dynamic_cast<CloudStorageEngine*>(&ExecEnv::GetInstance()->storage_engine());
+ if (engine == nullptr) {
+ for (const auto& scan_range : scan_ranges) {
+ fallback_reasons[scan_range.scan_range.palo_scan_range.tablet_id] =
+ "storage engine is not cloud";
+ }
+ return fallback_reasons;
+ }
+
+ // (a) The BE is tearing down: do not add new sync work. Fall every scanned
+ // tablet back rather than race teardown.
+ if (engine->stopped()) {
+ for (const auto& scan_range : scan_ranges) {
+ fallback_reasons[scan_range.scan_range.palo_scan_range.tablet_id] =
+ "be is stopping, sync skipped";
+ }
+ return fallback_reasons;
+ }
+
+ // Index-aligned so the fan-out tasks write disjoint slots without a lock;
an
+ // empty slot means synced (or skipped as non-append-only), a non-empty
slot
+ // is the fallback reason. Held (with the latch and per-slot count guard)
+ // behind a shared_ptr because the fan-out is waited on with a fast-fail
+ // deadline: on a timeout this frame returns while still-running tasks keep
+ // writing their slots, so the storage must outlive the frame -- the
shared_ptr
+ // each task captured keeps it alive until the last one finishes. Only
read on
+ // the completed path.
+ auto per_range_reason =
std::make_shared<std::vector<std::string>>(scan_ranges.size());
+ auto fanout_done =
std::make_shared<CountDownLatch>(static_cast<int>(scan_ranges.size()));
+ // Single-owner claim per slot. Two parties can target the same slot: the
task,
+ // and -- on the narrow ThreadPool path where do_submit enqueues the task
and
+ // THEN returns an error (thread creation failed with zero live workers)
-- the
+ // inline submit-failure path, which then runs concurrently with the very
task it
+ // failed to schedule. Each party claims the slot with an atomic exchange
BEFORE
+ // touching it: the winner runs the sync (or the inline fallback), writes
+ // per_range_reason[idx] via publish_slot, and counts the latch down; the
loser
+ // returns having touched neither. Claiming before the RPC (an exchange,
not a
+ // check-then-act load) is what stops the enqueue-then-fail task from
firing a
+ // sync for an already-settled query, and it keeps the caller, once the
latch
+ // settles, the sole reader of a slot no task is still writing.
(Value-initialized
+ // to false.)
+ auto slot_counted =
std::make_shared<std::vector<std::atomic<bool>>>(scan_ranges.size());
+ // Set true when the caller abandons the wait (its deadline passed): a
not-yet-
+ // started task then skips its sync RPC instead of spending the full retry
+ // budget, so a meta-service brownout drains the bounded queue fast rather
than
+ // running abandoned work to completion and starving the pool.
+ auto abandoned = std::make_shared<std::atomic<bool>>(false);
+ // Called ONLY by whoever already won the exclusive slot claim (the
slot_counted
+ // exchange below), so it is the sole writer of the slot: publish its
reason (an
+ // empty reason means synced OK / skipped, leaving the default-empty slot)
and
+ // release the latch exactly once. The claim winner counting down is what
lets the
+ // caller move the slot out the instant the latch settles.
+ auto publish_slot = [fanout_done, per_range_reason](size_t idx,
std::string reason) {
+ if (!reason.empty()) {
+ (*per_range_reason)[idx] = std::move(reason);
+ }
+ fanout_done->count_down();
+ };
+
+ // Run the per-tablet syncs on the engine-owned, bounded query-cache delta
+ // pre-sync pool -- a DEDICATED pool, not the shared SyncLoadForTablets
warmup
+ // pool, so a meta-service brownout cannot couple the two paths. Properties
+ // this relies on: (1) fixed width + a bounded queue (cloud/config.h
+ // query_cache_delta_sync_*), so a brownout can neither spawn unbounded
+ // concurrent syncs nor grow the backlog without bound (a submit past the
queue
+ // cap fails fast and that tablet falls back); (2)
CloudStorageEngine::stop()
+ // drains this pool (joins running tasks, discards queued ones) and the
+ // destructor calls stop() before _meta_mgr/_tablet_mgr are destroyed, so a
+ // task that outlived its timed-out query still runs against a live engine
and
+ // none can survive it. The raw detached bthreads this replaced were
joined by
+ // nothing, so one sleeping in retry_rpc could wake after the engine was
freed.
+ //
+ // Fast-fail is preserved: this runs in operator init on the bounded query
+ // admission pool (light_work_pool, "must be light, not locked").
submit_func
+ // never creates a worker -- the pool's fixed worker set (min == max) is
+ // pre-started AND verified present in the engine ctor (a CHECK_EQ on
num_threads,
+ // because ThreadPool::init swallows creation failures), so submit is pure
enqueue:
+ // it returns immediately at capacity/shutdown and never blocks on thread
creation,
+ // keeping thread-creation latency off this critical path. The whole
fan-out plus
+ // wait is then bounded by ONE absolute deadline
(query_cache_decision_sync_
+ // timeout_ms from the start below), which the pure-enqueue submit loop
cannot
+ // exhaust before the wait even begins. A healthy sync is milliseconds,
far under
+ // the budget, so the steady-state path is unchanged; this only trips
under real
+ // meta-service degradation. Every task records its own sync error into
its slot so
+ // no failure aborts the others.
+ //
+ // The tasks do no explicit MemTracker/ResourceContext attach: this is a
+ // detached engine-layer sync with no live query context to charge to (the
+ // query may have already timed out and returned, so attaching to it would
be
+ // both unavailable here -- this is a static engine path -- and wrong).
Their
+ // allocations fall to the worker thread's default accounting, matching the
+ // same-shaped sync on the same kind of pool by CloudStorageEngine's other
+ // sync_rowsets caller, CloudBackendService::sync_load_for_tablets (the FE
+ // warmup path), which likewise does not attach.
+ auto sync_fanout_start = std::chrono::steady_clock::now();
+ auto& pool = engine->query_cache_delta_sync_pool();
+ // Safe to capture the engine by raw pointer: stop() drains this pool
before
+ // the engine (and _meta_mgr/_tablet_mgr) is destroyed, so a running task
+ // always sees a live engine.
+ CloudStorageEngine* engine_ptr = engine;
+ for (size_t i = 0; i < scan_ranges.size(); ++i) {
+ int64_t tablet_id =
scan_ranges[i].scan_range.palo_scan_range.tablet_id;
+ Status submit_st = pool.submit_func([tablet_id, current_version,
abandoned, i, engine_ptr,
Review Comment:
[P1] Coalesce presync across fragment runtimes
`QueryCacheRuntime` only coalesces inside one PFC, but each concurrent query
fragment has a private decision map while all of them read the BE-global
`QueryCache::instance()`. Under default single-primary placement (or a one-BE
compute group), identical query-cache queries reach the same BE and stale key,
so each runtime independently executes this per-tablet submit loop. While
metadata syncs are slow, three default-limit 768-tablet instances submit 2304
jobs into the default 16-worker/2048-pending pool: at least 240 submissions
reject even though any one fan-out fits, forcing avoidable full recomputation
and denying unrelated keys admission. The tablet sync lock acts only after pool
admission. Please single-flight sync completion globally by `(cache_key,
current_version)` while retaining per-query deadlines, fallback, and
delta-source capture, and test distinct runtimes plus an unrelated key.
--
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]
