github-actions[bot] commented on code in PR #65788:
URL: https://github.com/apache/doris/pull/65788#discussion_r3610138909
##########
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:
[P1] Gate on an actually inexact capture path
These raw session values do not imply that this plan can read an inexact
version. Local storage clamps both options off, and cloud MOW ignores
`enable_prefer_cached_rowset`; nevertheless this disables incremental in both
cases, while the BE also discards every exact MOW fill when the prefer flag is
set. That can turn an otherwise cacheable MOW workload into a full recompute on
every execution. Please derive the gate/write-back decision from the selected
capture path (at least cloud mode plus the applicable table type), rather than
from the unconsumed raw knobs.
##########
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:
[P1] Require a delete-bitmap watermark before MOW incremental merge
`sync_delete_bitmap=true` does not prove that this cached tablet's bitmap is
current. A full-compaction path can first cache the tablet with all rowsets
through version N but `sync_delete_bitmap=false`; `CloudTablet::sync_rowsets()`
then returns on `_max_version >= query_version`, so this call never backfills
the missing bitmap prefix. A rewrite marker targeting the cached side is
absent, `_delta_rewrites_history()` treats the delta as append-only, and the
merged result can be stored as version N with the overwritten row still
counted. Please gate MOW incremental reuse on a separately tracked
bitmap-synced watermark (or retain the cloud/MOW fallback) and cover the
rowsets-current/bitmap-missing sequence in a test.
##########
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:
[P1] Make the cloud path assertion deterministic
Query-cache entries are BE-local, so summing counters cannot compensate for
an unpinned follow-up query. If it is routed to another BE, the first cached
execution is a plain MISS and the second an exact HIT; neither
`query_cache_stale_hit_total` nor `query_cache_incremental_fallback_total`
advances. The comment's proposed rerun makes this a topology-dependent flaky
test, and a permitted presync timeout similarly defeats the stale-hit
assertion. Please pin these executions to one BE/compute node (or keep cloud
coverage result-only and prove the mode through deterministic instrumentation).
##########
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:
[P1] Keep timed-out sync work owned and cancellable
The timeout only abandons this future; it does not cancel or join the
driver/worker bthreads. Under a meta-service brownout every stale query can
therefore add another fanout that retries long after its fragment is gone,
without a query/background `ThreadContext`, and `CloudStorageEngine::stop()`
has no handle to join it before destroying the engine referenced by
`CloudTablet`. `query_cache_decision_sync_timeout_ms <= 0` also launches the
work before immediately detaching it. Please run this on a globally bounded,
engine-owned executor/task group with query cancellation and shutdown joining
(or share the scan sync), and do not launch when the timeout disables the wait.
##########
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:
[P2] Preserve the gate across mixed FE/BE versions
This new cloud path relies entirely on the upgraded FE to clear
`allow_incremental`. During a rolling upgrade, an old FE can still send
`allow_incremental=true` together with freshness/prefer options to this BE; the
BE then installs an exact delta read source, and scan setup skips the normal
warmed/freshness capture options. Please enforce the exclusion at a BE boundary
that sees both query options and the cache parameter (or add a compatible
capability), and test an old-FE-shaped request.
--
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]
