asdf2014 commented on code in PR #65788:
URL: https://github.com/apache/doris/pull/65788#discussion_r3612002430
##########
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:
Fixed in 53d7f26. The per-tablet sync fan-out is now single-flighted across
fragment runtimes.
QueryCacheRuntime is per PipelineFragmentContext, so before this each
concurrent identical query built its own fan-out against the BE-global
QueryCache::instance(), and the arithmetic in the finding is right: three
768-tablet instances would submit 2304 jobs into the 16-worker/2048-queue pool
and reject the overflow even though any one fan-out fits.
The fix adds a single-flight registry on QueryCache keyed by (cache_key,
current_version). The first arriving runtime (the flight owner) submits the
fan-out; later identical runtimes join as waiters on the SAME completion latch,
each under its own query_cache_decision_sync_timeout_ms deadline, and read the
shared per-tablet result slots (keyed by the owner's tablet order, so two
runtimes whose scan ranges hold the same tablets in a different order still
attribute each reason to the right tablet). So N identical concurrent queries
now issue ONE fan-out, and an unrelated key is never starved by the duplicates.
Deadlines, fallback, and delta-source capture stay per query: a waiter that
times out falls its own query back without disturbing the others.
Lifecycle: the registry entry is reaped only when the fan-out actually
drains (its latch reaches zero), by the last waiter present or, if every waiter
already timed out, by the final task itself. A last leaver that exits while
tasks are still running marks the flight abandoned, so not-yet-started tasks
skip their sync RPC and the pool drains fast, and it keeps the entry as a
tombstone, so a later identical arrival joins the draining flight instead of
becoming a fresh owner and re-submitting a duplicate fan-out during the very
brownout this exists to smooth. Erasing on timeout would recreate wave after
wave of duplicate syncs under a sustained meta-service stall, which is why the
tombstone stays until the drain.
The flight's per-tablet buffers (reasons, tablet ids) are charged to the
stable query-cache MemTracker via allocator-aware DorisVector, with the release
going back to the same limiter from whichever thread drops the last reference,
instead of the transient per-query tracker the flight outlives. Live registry
entries are observable via the new query_cache_presync_inflight gauge.
Tests (query_cache_test.cpp):
presync_single_flight_coalesces_and_unrelated_key_proceeds parks the shared
decision sync on a gate, witnesses a second runtime joining the same flight,
runs an unrelated key to INCREMENTAL while the first flight is parked, then
releases the gate and asserts both coalesced runtimes reach INCREMENTAL off a
single shared fan-out and the registry drains.
presync_single_flight_last_leaver_keeps_tombstone_reaped_on_drain drives two
waiters through deterministic per-role deadlines and asserts the first leaver
neither abandons nor erases, the last leaver abandons but keeps the tombstone,
and the final task reaps it on drain.
presync_single_flight_post_timeout_arrival_joins_not_reowns proves a
post-timeout identical arrival joins the draining flight instead of
re-submitting a duplicate (the shared sync count stays put).
presync_single_flight_tombstone_reaped_on_drain_without_arrival proves the
waiterless tombstone is reaped by the final task with no later arr
ival needed, and also asserts the MemTracker charge returns exactly to the
pre-flight baseline on reap.
presync_single_flight_reversed_order_maps_reasons_by_tablet pins the
owner-order reason mapping, and presync_single_flight_coalesces_mow_tablet
covers merge-on-write, asserting the shared sync carries the delete bitmap.
--
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]
