asdf2014 commented on code in PR #65788:
URL: https://github.com/apache/doris/pull/65788#discussion_r3610800222
##########
be/src/runtime/query_cache/query_cache.cpp:
##########
@@ -282,8 +295,187 @@ 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) {
+ // Index-aligned so the fork-joined 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 behind a shared_ptr because
+ // the fan-out below is waited on with a fast-fail deadline: on a timeout
+ // this frame returns while the 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 (non-timeout) path, where every task is done.
+ auto per_range_reason =
std::make_shared<std::vector<std::string>>(scan_ranges.size());
+ std::vector<std::function<Status()>> tasks;
+ tasks.reserve(scan_ranges.size());
+ for (size_t i = 0; i < scan_ranges.size(); ++i) {
+ int64_t tablet_id =
scan_ranges[i].scan_range.palo_scan_range.tablet_id;
+ tasks.emplace_back([tablet_id, current_version, per_range_reason, i]()
-> Status {
+ auto tablet_res = ExecEnv::get_tablet(tablet_id);
+ if (!tablet_res) {
+ // _capture_tablet_delta reports "tablet not found" from its
own
+ // get_tablet; nothing to sync. Should that later get_tablet
+ // succeed instead (a transient failure here), it takes the
+ // cache-miss load, which itself syncs rowsets AND the delete
+ // bitmap up to the visible version -- that load-bearing
+ // invariant (plus the endpoint check downstream) is what keeps
+ // a tablet that skipped this pre-sync safe to capture.
+ return Status::OK();
+ }
+ BaseTabletSPtr tablet = std::move(tablet_res.value());
+ const bool append_only =
+ tablet->keys_type() == KeysType::DUP_KEYS ||
+ (tablet->keys_type() == KeysType::UNIQUE_KEYS &&
+ tablet->enable_unique_key_merge_on_write());
+ if (!append_only) {
+ // Skip the RPC entirely: _capture_tablet_delta rejects this
+ // tablet at its keys-type check before it would consume a
sync.
+ return Status::OK();
+ }
+ auto cloud_tablet =
std::dynamic_pointer_cast<CloudTablet>(std::move(tablet));
+ if (cloud_tablet == nullptr) {
+ // is_cloud_mode() flipped on a live local deployment
+ // (cloud_unique_id is a mutable config): degrade to a full
+ // recompute rather than dereference the failed cast
downstream.
+ (*per_range_reason)[i] = "tablet is not a cloud tablet";
+ return Status::OK();
+ }
+ SyncOptions options;
+ options.query_version = current_version;
+ options.merge_schema = true;
+ // The history-rewrite check reads the delete bitmap this sync
merges,
+ // so pin the dependency explicitly rather than inherit the struct
+ // default (which a refactor could flip with no compile error): a
+ // merge-on-write delta cannot be classified on a bitmap the sync
did
+ // not bring up to the queried version.
+ options.sync_delete_bitmap = true;
Review Comment:
The classification never reads a weaker bitmap than the MOW read path
itself. Two mechanisms make it version-exact at the point of the
history-rewrite check:
- The decision pre-sync runs `sync_rowsets` with `options.sync_delete_bitmap
= true` (query_cache.cpp, in `_presync_cloud_delta_tablets`), so the local
CloudTablet view is brought to `query_version` for BOTH its rowset list and its
delete bitmap before the decision consumes it. `_capture_tablet_delta` then
reads that same bitmap. There is no separate, weaker read: the classification
inherits exactly the completeness of the MOW read path, no more and no less.
- If the pre-sync could not vouch for the view (cast failure on a
misconfigured deployment, an infrastructure sync failure, or the fast-fail
budget expiring), `_capture_tablet_delta` sees a recorded reason and falls the
whole decision back to a full recompute. So an incomplete-sync outcome is a
fallback, not a silent misclassification.
On `_max_version >= query_version` not proving bitmap completeness: agreed
in the general case, which is exactly why the MOW decision is conservative
rather than relying on that inequality. `_capture_tablet_delta` inspects the
delete bitmap for any marker that rewrites a row predating the cached version
and falls back when it finds one (covered by `mow_history_rewrite_falls_back`,
and for the cloud sync-completeness angle by
`mow_decision_sync_brings_bitmap_detects_rewrite`, which guards that the bitmap
the decision sync carried, not the loader's, feeds the check).
On full-compaction materialization omitting the bitmap: compaction reads
through its output version and relocates later delete markers onto the
compacted output, so a marker is not lost, it moves. A delta that spans a
compaction boundary is not capturable as a clean [cached+1, query] range and
takes the fallback; rowset pins protect the inputs from concurrent GC during
the window. So this parallel sync path does not produce an append-only-looking
rewrite either.
(If a reviewer still wants belt-and-suspenders here, the cheapest hardening
is to make `_capture_tablet_delta` fall back whenever the tablet carries any
post-cached full-compaction output rather than reasoning about marker
relocation. Happy to add it, but I believe the current path is correct.)
##########
be/src/runtime/query_cache/query_cache.cpp:
##########
@@ -282,8 +295,187 @@ 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) {
+ // Index-aligned so the fork-joined 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 behind a shared_ptr because
+ // the fan-out below is waited on with a fast-fail deadline: on a timeout
+ // this frame returns while the 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 (non-timeout) path, where every task is done.
+ auto per_range_reason =
std::make_shared<std::vector<std::string>>(scan_ranges.size());
+ std::vector<std::function<Status()>> tasks;
+ tasks.reserve(scan_ranges.size());
+ for (size_t i = 0; i < scan_ranges.size(); ++i) {
+ int64_t tablet_id =
scan_ranges[i].scan_range.palo_scan_range.tablet_id;
+ tasks.emplace_back([tablet_id, current_version, per_range_reason, i]()
-> Status {
+ auto tablet_res = ExecEnv::get_tablet(tablet_id);
+ if (!tablet_res) {
+ // _capture_tablet_delta reports "tablet not found" from its
own
+ // get_tablet; nothing to sync. Should that later get_tablet
+ // succeed instead (a transient failure here), it takes the
+ // cache-miss load, which itself syncs rowsets AND the delete
+ // bitmap up to the visible version -- that load-bearing
+ // invariant (plus the endpoint check downstream) is what keeps
+ // a tablet that skipped this pre-sync safe to capture.
+ return Status::OK();
+ }
+ BaseTabletSPtr tablet = std::move(tablet_res.value());
+ const bool append_only =
+ tablet->keys_type() == KeysType::DUP_KEYS ||
+ (tablet->keys_type() == KeysType::UNIQUE_KEYS &&
+ tablet->enable_unique_key_merge_on_write());
+ if (!append_only) {
+ // Skip the RPC entirely: _capture_tablet_delta rejects this
+ // tablet at its keys-type check before it would consume a
sync.
+ return Status::OK();
+ }
+ auto cloud_tablet =
std::dynamic_pointer_cast<CloudTablet>(std::move(tablet));
+ if (cloud_tablet == nullptr) {
+ // is_cloud_mode() flipped on a live local deployment
+ // (cloud_unique_id is a mutable config): degrade to a full
+ // recompute rather than dereference the failed cast
downstream.
+ (*per_range_reason)[i] = "tablet is not a cloud tablet";
+ return Status::OK();
+ }
+ SyncOptions options;
+ options.query_version = current_version;
+ options.merge_schema = true;
+ // The history-rewrite check reads the delete bitmap this sync
merges,
+ // so pin the dependency explicitly rather than inherit the struct
+ // default (which a refactor could flip with no compile error): a
+ // merge-on-write delta cannot be classified on a bitmap the sync
did
+ // not bring up to the queried version.
+ options.sync_delete_bitmap = true;
+ Status st = cloud_tablet->sync_rowsets(options);
+ if (!st.ok()) {
+ // Unlike the other fallback reasons (expected data shapes), a
+ // failed sync is an infrastructure error: log the status so
the
+ // profile reason correlates to a cause. Throttled with
+ // LOG_EVERY_N because a meta-service brownout fails this for
+ // every stale tablet of every query at once; the underlying
RPC
+ // errors are already logged per retry inside retry_rpc, so a
+ // sampled line here is enough to correlate without a storm,
and
+ // the exact reason still reaches the user via the query
profile.
+ LOG_EVERY_N(WARNING, 100)
+ << "query cache incremental merge falls back, cloud
rowset sync failed"
+ << ", tablet_id=" << tablet_id << ", status=" <<
st.to_string();
+ (*per_range_reason)[i] = "cloud rowset sync failed";
+ }
+ return Status::OK();
+ });
+ }
+ // Fan the syncs out asynchronously and wait on the result with a fast-fail
+ // budget instead of blocking unconditionally. This decision runs in
+ // operator init on a bounded query-admission pool (the BE light_work_pool,
+ // contractually "must be light, not locked"), so a meta-service brownout
+ // that stalls these RPCs must not hold that thread for the full RPC retry
+ // budget (tens of seconds): sustained, it would exhaust the pool and
reject
+ // query admission cluster-wide. If the fan-out does not finish within
+ // query_cache_decision_sync_timeout_ms the decision abandons the wait and
+ // falls the whole instance back to a full recompute; the still-running
+ // tasks are correctness-harmless and bounded: on a slow-but-healthy sync
+ // each merely advances the tablet view early (work the scan node's own
+ // async sync would otherwise do), and on a failing sync each is at worst
one
+ // extra bounded sync attempt that changes no result; either way they stay
+ // cheap while pending (small per-task captures; the brpc I/O yields the
+ // pthread rather than pinning it) and self-limiting once the meta service
+ // recovers, and keep their slot storage alive through the shared_ptr they
+ // captured. A healthy sync is
+ // milliseconds, far under the budget, so this only trips under real
+ // meta-service degradation and leaves the steady-state incremental path
+ // unchanged.
+ //
+ // Parallelism reuses init_scanner_sync_rowsets_parallelism on purpose:
this
+ // is the same per-tablet rowset-sync fan-out the scan node runs, and one
+ // shared knob keeps the two fan-outs' budgets aligned rather than letting
a
+ // cache-only config silently drift apart from it. Every task swallows its
+ // own error into its slot and returns OK, which is load-bearing:
+ // bthread_fork_join stops DISPATCHING the tasks queued after the first one
+ // that returns non-OK (see cloud_meta_mgr.cpp), so a task that propagated
+ // its sync error would leave the tablets queued behind it un-synced;
+ // recording the failure into the slot keeps every tablet's sync attempted.
+ // The blocking time (bounded by the budget) is folded into
+ // query_cache_decision_sync_time_ms.
+ auto sync_fanout_start = std::chrono::steady_clock::now();
+ std::future<Status> fanout_done;
+ // Clamp the shared parallelism knob to >= 1: bthread_fork_join waits for a
+ // free slot before dispatching its first task (count 0 >= concurrency 0),
+ // so a misconfigured 0/negative would park the driver bthread forever with
+ // nothing to notify it. On the scan node that surfaces as a hung query; on
+ // this fast-fail path the caller's wait_for still expires and falls back,
so
+ // the deadlock would instead leak the driver bthread and its task closures
+ // silently, once per incremental candidate. A local floor avoids that.
+ Status launch_st = cloud::bthread_fork_join(
+ std::move(tasks), std::max(1,
config::init_scanner_sync_rowsets_parallelism),
+ &fanout_done);
+ bool timed_out = false;
+ if (!launch_st.ok()) {
+ // Could not even spawn the fan-out driver bthread: fall back rather
than
+ // block, the same as a timeout (its future would never be fulfilled).
+ timed_out = true;
+ } else if (fanout_done.wait_for(std::chrono::milliseconds(
Review Comment:
Replaced the raw detached bthreads with a dedicated engine-owned pool and
single-owner slot accounting:
- Ownership/lifetime: `_presync_cloud_delta_tablets` submits to
`CloudStorageEngine::query_cache_delta_sync_pool()`, a `ThreadPool` the engine
constructs up front and drains FIRST in `stop()` (before the compaction and
delete-bitmap pools it depends on, and before `~CloudStorageEngine` destroys
`_meta_mgr`/`_tablet_mgr`). A task that outlived a timed-out query therefore
always runs against a live engine, and none can survive the engine. The raw
pointer capture is safe for the same reason.
- Bound: the pool has a fixed width and a bounded queue
(`config::query_cache_delta_sync_thread` /
`query_cache_delta_sync_max_pending_tasks`), so a brownout can neither spawn an
unbounded concurrent fanout nor grow the backlog without bound; a submit past
the queue cap fails fast and that tablet falls back.
- Admission latency: the pool is sized fixed (min == max) and its workers
are pre-started AND verified present at engine construction (a `CHECK` on the
worker count, because `ThreadPool::init` swallows per-thread start failures, so
a successful build alone does not prove the workers exist). So `submit_func`
provably never creates a thread inside the operator-init loop (thread creation
can take hundreds of ms per the ThreadPool docs); it is pure enqueue and never
blocks, keeping the whole submit loop far under the deadline. The cost is a
fixed set of mostly-idle workers on every cloud BE even when this opt-in
feature is unused, which is the deliberate trade for predictable, non-blocking
admission.
- Single deadline: the fan-out plus wait is bounded by one absolute deadline
anchored at the fan-out start, so submission cannot re-grant the full timeout
to the wait; the total budget stays `query_cache_decision_sync_timeout_ms`.
- Cancellation: a shared `abandoned` atomic is set when the caller's bounded
wait expires; a not-yet-started task then skips its sync RPC instead of
spending the full retry budget, so the bounded queue drains quickly under a
brownout.
- Non-positive timeout: it now launches nothing at all. The `<= 0` case
returns up front with every scanned tablet falling back, before the engine is
even touched, so there is no ownerless work.
- No admission-thread loads: the capture loop consumes each tablet's
recorded fallback reason BEFORE its own `get_tablet`, and a worker-side
tablet-load failure publishes a reason instead of leaving the slot empty. So
neither a refused submit nor a failed worker load can push the synchronous
cache-miss meta-service load onto the admission thread outside the deadline;
the admission path performs zero remote loads on every fallback route.
- Slot accounting: each slot is claimed via an atomic exchange BEFORE any
sync RPC, so exactly one of the task and the inline submit-failure path owns
both the RPC and the latch count-down; the loser touches neither. A task that a
later worker picks up after its slot was already claimed inline (the narrow
enqueue-then-fail case where `do_submit` queued it then returned an error)
loses the claim and bails before its RPC, so it cannot fire a meta-service call
for an already-settled query, and the caller past the settled latch is the sole
reader of the result.
On the bthread memory-tracker attach: these tasks intentionally do not
attach a query/ResourceContext. This is a detached engine-layer sync with no
live query context to charge to (the query may already have timed out and
returned). It matches the sibling FE warmup path
`CloudBackendService::sync_load_for_tablets`, which submits the same
`CloudTablet::sync_rowsets` to an engine-owned `ThreadPool` and likewise does
not attach a tracker; the allocations fall to the worker thread's default
accounting and remain process-tracked, not orphaned. I documented this on the
call site. If the project would rather these attach to a background tracker,
point me at the preferred one and I will wire it, but the current behavior is
consistent with the sibling path.
##########
fe/fe-core/src/main/java/org/apache/doris/planner/normalize/QueryCacheNormalizer.java:
##########
@@ -179,6 +179,33 @@ private boolean computeAllowIncremental(CachePoint
cachePoint, SessionVariable s
if (!sessionVariable.getEnableQueryCacheIncremental()) {
return false;
}
+ // Freshness tolerance and prefer-cached-rowset (both cloud-only in
+ // effect) trade exactness for speed and locality: the scan may read a
+ // warmed-up layout that stops below the queried version (freshness
+ // halts at the warmed boundary) or reaches beyond it (neither walk
+ // clips an edge spanning it). An incremental merge would defeat both
knobs,
+ // because the delta capture always targets the exact queried version
+ // (it must, to keep the merged entry correct): under freshness
+ // tolerance it would force the wait for un-warmed data the query
+ // explicitly chose to skip, and under prefer-cached-rowset it would
+ // ignore the layout preference the user set. So exclude incremental
+ // for such queries and let them take their cheap path. Correctness
+ // against their version-inexact reads does NOT rest on this per-query
+ // gate (an entry filled by such a read could still be reused by a
+ // different, knob-free query sharing the cache key): on cloud, the
+ // only mode where these reads occur, the BE suppresses their cache
+ // write-back, so no entry whose content mismatches its version stamp
+ // exists in the first place. These knobs are inert on local storage,
+ // so gating them in any mode only forgoes incremental for a query
+ // that opted into a cloud trade-off; the gate stays mode-agnostic on
+ // purpose, because a mode-conditioned gate cannot be exercised by the
+ // local FE unit-test harness (planning under a flipped cloud flag
casts
+ // SystemInfoService to CloudSystemInfoService and fails), so it would
+ // ship an untestable branch for no correctness gain.
+ if (sessionVariable.getQueryFreshnessToleranceMs() > 0
Review Comment:
FE now resolves the table type before the knob gate and reports it to BE:
- `QueryCacheNormalizer.computeAllowIncremental` was reordered so the table
type is known before the knob is consulted: freshness tolerance blocks all
incremental, and prefer-cached-rowset blocks only non-MOW (cloud MOW ignores
prefer, so it must not be blocked by it).
- A new optional thrift field `TQueryCacheParam.is_merge_on_write` carries
the selected index's MOW state (via `computeIsMergeOnWrite`), independent of
`allow_incremental` because the write-back gate applies to MISS decisions too.
- BE `cache_source_operator` keys write-back on `is_cloud_mode() &&
(freshness || (prefer && !is_merge_on_write))`, so cloud MOW under prefer keeps
its version-exact write-back instead of being suppressed by a knob that does
not apply to it.
Local scans ignoring both cloud knobs: that is the pre-existing
shared-nothing behavior and is unchanged; the gate above is cloud-only
(`is_cloud_mode()`), so it does not touch the local path.
##########
be/src/runtime/query_cache/query_cache.cpp:
##########
@@ -259,9 +262,19 @@ bool QueryCacheRuntime::_try_prepare_incremental(const
std::vector<TScanRangePar
return false;
}
+ // Cloud only: fan the per-tablet meta-service syncs out in parallel before
+ // the serial capture below, so an instance holding many stale tablets does
+ // not stall the shared prepare thread issuing those RPCs one at a time.
+ // Empty (and untouched) in shared-nothing mode, where no sync is needed.
+ std::unordered_map<int64_t, std::string> presync_reasons;
+ if (config::is_cloud_mode()) {
Review Comment:
The follow-up adds the BE-side default the review asked for, through the
optional thrift field:
- old-FE to new-BE: `is_merge_on_write` is absent, so it reads its default
`false`, which drives the conservative branch (write-back suppressed for what
would be a MOW exact fill). No bypass.
- new-FE to old-BE: the old BE ignores the appended optional field and keeps
its prior conservative behavior.
- same-version: aligned.
So the mixed-version matrix defaults to suppression on the unknown side
rather than to an unguarded fill. There is no storage-format or symbol change,
and the field is optional field 9 appended to `TQueryCacheParam`, so it is
skip-safe both directions.
##########
regression-test/suites/query_p0/cache/query_cache_incremental.groovy:
##########
@@ -72,20 +75,26 @@ suite("query_cache_incremental") {
return total
}
- // Besides result consistency, prove on local storage that the stale entry
- // was really merged incrementally (Mode::INCREMENTAL), not recomputed:
- // only the incremental path increases query_cache_stale_hit_total, and no
+ // Besides result consistency, prove that the stale entry was really
+ // merged incrementally (Mode::INCREMENTAL), not recomputed: only the
+ // incremental path increases query_cache_stale_hit_total, and no
// concurrent suite touches it because the session switch defaults to off.
- // Cloud mode always falls back by design, so it only checks consistency.
// Residual caveat (here and in checkIncrementalFallback): a
// memory-pressure prune evicting the entry inside the tiny fill-to-assert
// window surfaces as a plain MISS and would fail the assertion; accepted
- // as rare, and a rerun re-establishes the counters from fresh deltas.
+ // as rare, and a rerun re-establishes the counters from fresh deltas. The
+ // one channel that is NOT left to chance is compaction: every table here
+ // sets disable_auto_compaction, so no background compaction can merge the
+ // delta away mid-window and turn the expected incremental hit into a
+ // fallback (this matters most on cloud, where compaction is driven
+ // externally rather than by the querying BE).
+ // On cloud two more channels surface the same way and fail only
Review Comment:
- The negative cases now exist as deterministic BE unit tests: the
incomplete-bitmap/history-rewrite case (`mow_history_rewrite_falls_back`,
`mow_decision_sync_brings_bitmap_detects_rewrite`), the timeout/abandon case
(`decision_sync_timeout_falls_back`), the engine-stopped and non-cloud-engine
fallbacks, the pool-rejects-submit fallback
(`presync_falls_back_when_pool_rejects_submit`), the rejection-plus-cache-miss
composition proving the admission path performs zero meta loads
(`pool_rejection_skips_admission_tablet_load`), and the worker-load-failure
case proving exactly one worker-side load and no admission-thread retry
(`presync_tablet_load_failure_falls_back`).
- The routing-dependent metric oracle: the stale-hit assertion in the
regression is gated to the shared-nothing path, because an aggregated BE-local
counter cannot prove a specific cloud query used a stale entry when
placement/routing is topology-dependent. On cloud the suite proves correctness
by result consistency, and the cloud INCREMENTAL path is proven directly by the
deterministic BE tests in `QueryCacheCloudIncrementalTest`.
- Both clang-format checks are green in the follow-up (the changed-file
violations the local dry run reproduced are fixed).
--
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]
