[feature](query-cache) Extend stale-entry incremental merge to cloud mode

be/src/cloud/cloud_meta_mgr.cpp

@@ -668,7 +668,14 @@ Status CloudMetaMgr::sync_tablet_rowsets_unlocked(CloudTablet* tablet,
                                                   SyncRowsetStats* sync_stats) {
     using namespace std::chrono;
 
-    TEST_SYNC_POINT_RETURN_WITH_VALUE("CloudMetaMgr::sync_tablet_rowsets", Status::OK(), tablet);
+    // Forward `options` so a test callback can react to sync_delete_bitmap the
+    // way the real body does below (the delete-bitmap sync is gated on it), i.e.
+    // a mock can drop the bitmap when the flag is false. The pointer is appended
+    // before the macro's own return-value slot, which callbacks read via
+    // args.back(), so existing callbacks that only read args[0]/the ret pair are
+    // unaffected.
+    TEST_SYNC_POINT_RETURN_WITH_VALUE("CloudMetaMgr::sync_tablet_rowsets", Status::OK(), tablet,
+                                      &options);
     DBUG_EXECUTE_IF("CloudMetaMgr::sync_tablet_rowsets.before.inject_error", {
         auto target_tablet_id = dp->param<int64_t>("tablet_id", -1);
         auto target_table_id = dp->param<int64_t>("table_id", -1);

be/src/cloud/cloud_storage_engine.cpp

@@ -103,6 +103,54 @@ CloudStorageEngine::CloudStorageEngine(const EngineOptions& options)
     _cumulative_compaction_policies[CUMULATIVE_TIME_SERIES_POLICY] =
             std::make_shared<CloudTimeSeriesCumulativeCompactionPolicy>();
     _startup_timepoint = std::chrono::system_clock::now();
+    // Build the dedicated, bounded query-cache delta pre-sync pool up front (see
+    // the member declaration): a decision fan-out must never find it null, and the
+    // destructor's stop() drains it before _meta_mgr/_tablet_mgr die. A build
+    // failure here means the process cannot create threads at startup, already
+    // fatal, so CHECK rather than limp on with a null pool.
+    //
+    // Report a misconfigured sizing loudly instead of silently coercing it: an
+    // operator who set an invalid value must see why the BE refused to start rather
+    // than run with a capacity that differs from the config (invalid state fails, it
+    // never silently continues). A max of >= 1 is required (a zero-width pool can run
+    // nothing) and the queue bound must be non-negative.
+    int32_t qc_sync_threads = config::query_cache_delta_sync_thread;
+    int32_t qc_sync_queue = config::query_cache_delta_sync_max_pending_tasks;
+    CHECK_GE(qc_sync_threads, 1) << "query_cache_delta_sync_thread must be >= 1, got "
+                                 << qc_sync_threads;
+    CHECK_GE(qc_sync_queue, 0) << "query_cache_delta_sync_max_pending_tasks must be >= 0, got "
+                               << qc_sync_queue;
+    // Pre-start all workers here (min == max), at engine construction, OFF the query
+    // admission path. The decision fan-out submits from operator init, which runs on
+    // the bounded light_work_pool, and ThreadPool creates workers synchronously
+    // inside submit_func when the pool is short of the demand (creation can take
+    // hundreds of ms). A lazily grown pool (min = 0) would move that creation latency
+    // onto the query's critical path and, worse, could hand the fan-out a submit that
+    // enqueued a task yet returned an error at zero live threads. A fixed pool sized
+    // up front keeps submit_func pure enqueue (it never creates a thread, so it never
+    // blocks), so the fast-fail budget is spent on the sync RPCs rather than on
+    // spawning threads. The cost is a fixed set of mostly-idle workers on every cloud
+    // BE even when this opt-in feature is unused; that steady footprint is the
+    // deliberate trade for predictable, non-blocking admission.
+    Status qc_pool_st = ThreadPoolBuilder("QueryCacheDeltaSyncThreadPool")
+                                .set_min_threads(qc_sync_threads)
+                                .set_max_threads(qc_sync_threads)
+                                .set_max_queue_size(qc_sync_queue)
+                                .build(&_query_cache_delta_sync_pool);
+    CHECK(qc_pool_st.ok()) << "failed to build QueryCacheDeltaSyncThreadPool: " << qc_pool_st;
+    // A successful build does NOT prove the workers exist: ThreadPool::init()
+    // deliberately ignores per-thread creation failures (a worker can be created
+    // later on submit). Verify the fixed worker set actually started and fail
+    // construction if not, so the fan-out's "submit never creates a thread" property
+    // is a guarantee, not a hope -- without this a startup thread shortage would
+    // silently push synchronous thread creation back onto the query admission path,
+    // exactly the latency this fixed pool exists to avoid. num_threads() counts
+    // started plus pending-start workers, so it equals the request precisely when
+    // every worker was spawned.
+    CHECK_EQ(_query_cache_delta_sync_pool->num_threads(), qc_sync_threads)
+            << "QueryCacheDeltaSyncThreadPool started "
+            << _query_cache_delta_sync_pool->num_threads() << " of " << qc_sync_threads
+            << " workers";
 }
 
 CloudStorageEngine::~CloudStorageEngine() {
@@ -274,6 +322,20 @@ void CloudStorageEngine::stop() {
         }
     }
 
+    // Drain the query-cache delta pre-sync fan-out FIRST, before the pools its
+    // sync_rowsets callbacks depend on: a merge-on-write delete-bitmap sync
+    // submits file-read work to _sync_delete_bitmap_thread_pool /
+    // _calc_tablet_delete_bitmap_task_thread_pool and blocks on it. Shutting a
+    // dependency pool down first would discard its queued callbacks and leave a
+    // query-cache worker blocked forever, hanging this drain and BE shutdown.
+    // Draining this pool first lets its in-flight workers finish against the
+    // still-live dependency pools. stop() also runs before any member is
+    // destroyed (~CloudStorageEngine calls it first), so those tasks see live
+    // _meta_mgr/_tablet_mgr as well.
+    if (_query_cache_delta_sync_pool) {
+        _query_cache_delta_sync_pool->shutdown();
+    }
+
     if (_base_compaction_thread_pool) {
         _base_compaction_thread_pool->shutdown();
     }

be/src/cloud/cloud_storage_engine.h

@@ -176,6 +176,11 @@ class CloudStorageEngine final : public BaseStorageEngine {
         return *_sync_load_for_tablets_thread_pool;
     }
 
+    // Dedicated bounded pool for the query-cache incremental decision's per-tablet
+    // rowset pre-sync. Always non-null (built in the constructor), so callers need
+    // no null check. See the member declaration and cloud/config.h.
+    ThreadPool& query_cache_delta_sync_pool() const { return *_query_cache_delta_sync_pool; }
+
     ThreadPool& warmup_cache_async_thread_pool() const { return *_warmup_cache_async_thread_pool; }
 
     Status register_compaction_stop_token(CloudTabletSPtr tablet, int64_t initiator);
@@ -234,6 +239,16 @@ class CloudStorageEngine final : public BaseStorageEngine {
     std::shared_ptr<CloudWarmUpManager> _cloud_warm_up_manager;
     std::unique_ptr<TabletHotspot> _tablet_hotspot;
     std::unique_ptr<ThreadPool> _sync_load_for_tablets_thread_pool;
+    // Dedicated bounded pool for the query-cache incremental decision's per-tablet
+    // rowset pre-sync (QueryCacheRuntime::_presync_cloud_delta_tablets). Built in
+    // the constructor -- not open() -- so a decision fan-out never dereferences a
+    // null pool and unit fixtures that construct the engine without open() have
+    // it; drained in stop() (which the destructor calls first) so every in-flight
+    // task joins before _meta_mgr/_tablet_mgr are torn down. Declared after those
+    // managers so reverse-order member destruction is a second guarantee of the
+    // same ordering. Kept separate from the FE-warmup SyncLoadForTabletsThreadPool
+    // so a meta-service brownout cannot couple the two paths.
+    std::unique_ptr<ThreadPool> _query_cache_delta_sync_pool;
     std::unique_ptr<ThreadPool> _warmup_cache_async_thread_pool;
     std::unique_ptr<CloudSnapshotMgr> _cloud_snapshot_mgr;
 

be/src/cloud/config.cpp

@@ -38,6 +38,10 @@ DEFINE_Int64(tablet_cache_shards, "16");
 DEFINE_mInt32(tablet_sync_interval_s, "1800");
 DEFINE_mInt32(init_scanner_sync_rowsets_parallelism, "10");
 DEFINE_mInt32(sync_rowsets_slow_threshold_ms, "1000");
+DEFINE_mInt32(query_cache_decision_sync_timeout_ms, "2000");
+DEFINE_Int32(query_cache_delta_sync_thread, "16");
+DEFINE_Int32(query_cache_delta_sync_max_pending_tasks, "2048");
+DEFINE_mInt32(query_cache_max_concurrent_decision_sync, "32");
 
 DEFINE_mInt64(min_compaction_failure_interval_ms, "5000");
 DEFINE_mInt64(base_compaction_freeze_interval_s, "1800");

be/src/cloud/config.h

@@ -70,6 +70,54 @@ DECLARE_mInt32(tablet_sync_interval_s);
 // parallelism for scanner init where may issue RPCs to sync rowset meta from MS
 DECLARE_mInt32(init_scanner_sync_rowsets_parallelism);
 DECLARE_mInt32(sync_rowsets_slow_threshold_ms);
+// Fast-fail budget (ms) for the query-cache incremental-merge decision's
+// pre-sync rowset fan-out. That decision runs in operator init on a bounded
+// query-admission thread pool (the BE light_work_pool, contractually "must be
+// light, not locked"), so a meta-service brownout that stalls the sync 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. When the
+// fan-out does not finish within this budget the decision abandons the wait and
+// falls back to one full recompute (the scan node's own async sync still brings
+// the view up for the actual scan). A healthy sync is milliseconds, far under
+// this, so it only trips under real meta-service degradation and leaves the
+// steady-state incremental path unchanged. A value <= 0 disables cloud
+// incremental merge outright: the decision skips the pre-sync fan-out entirely
+// (launching nothing) and falls every scanned tablet back to a full recompute,
+// a fail-safe rather than a useful setting. Cloud only.
+DECLARE_mInt32(query_cache_decision_sync_timeout_ms);
+
+// The dedicated, bounded thread pool that runs the query-cache incremental
+// decision's per-tablet rowset pre-sync (CloudStorageEngine owns it, created at
+// construction and drained in stop()). It is deliberately NOT the shared
+// SyncLoadForTabletsThreadPool: that pool is on the FE-driven warmup path, and
+// giving the decision fan-out its own pool keeps a meta-service brownout from
+// coupling the two. `query_cache_delta_sync_thread` bounds the concurrent
+// syncs; `query_cache_delta_sync_max_pending_tasks` caps the queue so a brownout
+// (every sync stalls in retry_rpc while new stale queries keep enqueuing) cannot
+// grow the backlog without bound -- a submit that would exceed the cap fails
+// fast and that tablet falls back to a full recompute. Cloud only.
+DECLARE_Int32(query_cache_delta_sync_thread);
+DECLARE_Int32(query_cache_delta_sync_max_pending_tasks);
+
+// Upper bound on how many query-cache incremental decisions may block in the
+// decision-sync wait at the same time. That wait runs on brpc's light work pool
+// (query admission; "must be light, not locked"), and single-flight coalesces only
+// IDENTICAL cache keys, so under a meta-service brownout a wave of DISTINCT stale
+// keys could otherwise park every light-pool worker for the full decision-sync
+// timeout and starve unrelated fragment admission (the paired scan and cache-source
+// operators can each park one on a first-call race, so the pressure is per operator,
+// not just per query). The effective bound is the smaller of this value and half the
+// ACTUAL light-pool width (the configured brpc_light_work_pool_threads, or
+// max(128, 4*cores) when left at -1), so even a shrunk pool keeps spare workers for
+// that admission -- this value only tightens below that floor. A query arriving over
+// the bound skips incremental merge this round and recomputes in full (always correct,
+// the right posture for an optimization under load). A non-positive value does not
+// disable the guard; it falls back to the width-derived half. The fully non-blocking
+// alternative is to drive the decision
+// off a pipeline dependency, mirroring OlapScanLocalState::_cloud_tablet_dependency;
+// that is a larger cross-operator change left as a follow-up. Mutable so it can be
+// retuned online. Cloud only.
+DECLARE_mInt32(query_cache_max_concurrent_decision_sync);
 
 // Cloud compaction config
 DECLARE_mInt64(min_compaction_failure_interval_ms);

be/src/common/metrics/doris_metrics.cpp

@@ -57,6 +57,10 @@ DEFINE_COUNTER_METRIC_PROTOTYPE_3ARG(query_cache_incremental_fallback_total, Met
 DEFINE_COUNTER_METRIC_PROTOTYPE_3ARG(query_cache_write_back_total, MetricUnit::REQUESTS,
                                      "Query cache entries handed to the cache to be written back "
                                      "(full or merged); admission may still turn one down.");
+DEFINE_COUNTER_METRIC_PROTOTYPE_3ARG(query_cache_decision_sync_time_ms, MetricUnit::MILLISECONDS,
+                                     "Cumulative wall time the cloud incremental-merge decision "
+                                     "spent blocking on its pre-sync rowset fan-out.");
+DEFINE_GAUGE_CORE_METRIC_PROTOTYPE_2ARG(query_cache_presync_inflight, MetricUnit::NOUNIT);
 DEFINE_COUNTER_METRIC_PROTOTYPE_5ARG(push_requests_success_total, MetricUnit::REQUESTS, "",
                                      push_requests_total, Labels({{"status", "SUCCESS"}}));
 DEFINE_COUNTER_METRIC_PROTOTYPE_5ARG(push_requests_fail_total, MetricUnit::REQUESTS, "",
@@ -303,6 +307,8 @@ DorisMetrics::DorisMetrics() : _metric_registry(_s_registry_name) {
     INT_COUNTER_METRIC_REGISTER(_server_metric_entity, query_cache_stale_hit_total);
     INT_COUNTER_METRIC_REGISTER(_server_metric_entity, query_cache_incremental_fallback_total);
     INT_COUNTER_METRIC_REGISTER(_server_metric_entity, query_cache_write_back_total);
+    INT_COUNTER_METRIC_REGISTER(_server_metric_entity, query_cache_decision_sync_time_ms);
+    INT_GAUGE_METRIC_REGISTER(_server_metric_entity, query_cache_presync_inflight);
 
     INT_COUNTER_METRIC_REGISTER(_server_metric_entity, push_requests_success_total);
     INT_COUNTER_METRIC_REGISTER(_server_metric_entity, push_requests_fail_total);

be/src/common/metrics/doris_metrics.h

@@ -64,6 +64,16 @@ class DorisMetrics {
     IntCounter* query_cache_stale_hit_total = nullptr;
     IntCounter* query_cache_incremental_fallback_total = nullptr;
     IntCounter* query_cache_write_back_total = nullptr;
+    // Cumulative wall time the cloud incremental-merge decision spent blocking
+    // on its pre-sync rowset fan-out. That sync runs during operator init, not
+    // inside a profiled scan node, so without this counter its (potentially
+    // meta-service-bound) latency is invisible.
+    IntCounter* query_cache_decision_sync_time_ms = nullptr;
+    // Live count of cloud pre-sync single-flight entries currently registered
+    // (owned or draining). Steady state is near zero; a persistently rising value
+    // would flag a registry leak or a stuck fan-out. Also makes the coalescing
+    // mechanism observable (how many concurrent identical fan-outs are in flight).
+    IntGauge* query_cache_presync_inflight = nullptr;
 
     IntCounter* push_requests_success_total = nullptr;
     IntCounter* push_requests_fail_total = nullptr;

be/src/exec/operator/cache_source_operator.cpp

@@ -20,13 +20,14 @@
 #include <functional>
 #include <utility>
 
+#include "cloud/config.h"
 #include "common/status.h"
 #include "core/block/block.h"
 #include "exec/operator/operator.h"
 #include "exec/pipeline/dependency.h"
+#include "runtime/runtime_state.h"
 
 namespace doris {
-class RuntimeState;
 
 Status CacheSourceLocalState::init(RuntimeState* state, LocalStateInfo& info) {
     RETURN_IF_ERROR(Base::init(state, info));
@@ -95,8 +96,38 @@ Status CacheSourceLocalState::init(RuntimeState* state, LocalStateInfo& info) {
     // MISS and INCREMENTAL rebuild the entry (from scratch / by merge), unless
     // the decision already knows the merged entry could never fit the entry
     // limits (write_back_feasible == false).
-    _need_insert_cache =
-            _cache_decision->key_valid && !hit_cache && _cache_decision->write_back_feasible;
+    // A query reading through a warmed-up-layout knob (both cloud only) must
+    // not write back either, because its scan is not guaranteed to represent
+    // exactly the queried version the entry would be stamped with. Both
+    // knobs walk the version graph by warmed-up preference and, unlike the
+    // plain capture, never clip an edge whose end lies past the requested
+    // window: the read may OVERSHOOT the queried version (a warmed
+    // compaction rowset spanning it drags the path beyond), and under
+    // freshness tolerance it may also STOP BELOW it (the walk halts at the
+    // warmed boundary). Neither knob carries an affectQueryResult
+    // annotation, so such an entry shares its cache key with plain runs of
+    // the same statement, which would then serve (exact hit) or extend
+    // (incremental merge) a base whose content does not match its version
+    // stamp: missing rows where the read stopped short, double-counted rows
+    // where it overshot. Suppressing the write-back keeps these reads pure
+    // consumers: they may still serve an exact HIT filled by a plain run
+    // (precise data, no worse than asked).
+    //
+    // Carve-out for the prefer knob on cloud merge-on-write UNIQUE tables:
+    // CloudTablet::capture_consistent_versions_unlocked honors
+    // enable_prefer_cached_rowset only for non-MOW tables (it guards on
+    // !enable_unique_key_merge_on_write()), so a MOW query that set prefer but
+    // not freshness still reads the exact queried version. That fill is
+    // version-exact and cacheable; suppressing it would keep a cloud MOW table
+    // with prefer set from ever populating the cache. Freshness tolerance has
+    // no such MOW guard (it is honored for MOW too), so it still forces
+    // suppression regardless of table type.
+    const bool inexact_version_fill =
+            config::is_cloud_mode() &&
+            (state->enable_query_freshness_tolerance() ||
+             (state->enable_prefer_cached_rowset() && !cache_param.is_merge_on_write));
+    _need_insert_cache = _cache_decision->key_valid && !hit_cache &&
+                         _cache_decision->write_back_feasible && !inexact_version_fill;
     _insert_delta_count = _is_incremental ? _cache_decision->cached_delta_count + 1 : 0;
 
     if (hit_cache || _is_incremental) {