[feature](query-cache) Support incremental merge for stale query cache entries

be/src/common/config.cpp

@@ -1690,6 +1690,12 @@ DEFINE_mBool(enable_pipeline_task_leakage_detect, "false");
 DEFINE_mInt32(check_score_rounds_num, "1000");
 
 DEFINE_Int32(query_cache_size, "512");
+// Max number of incremental merges accumulated on one query cache entry before
+// a full recompute is forced. Each incremental merge appends the delta partial
+// blocks to the entry, so the entry gets more fragmented (and the upstream merge
+// aggregation does more work) as deltas accumulate; a periodic full recompute
+// compacts the entry back to a minimal set of blocks.
+DEFINE_mInt32(query_cache_max_incremental_merge_count, "8");
 
 // Enable validation to check the correctness of table size.
 DEFINE_Bool(enable_table_size_correctness_check, "false");

be/src/common/config.h

@@ -1750,6 +1750,9 @@ DECLARE_mInt32(check_score_rounds_num);
 
 // MB
 DECLARE_Int32(query_cache_size);
+// Max incremental merges on one query cache entry before forcing a full
+// recompute to compact the entry (see query_cache.h QueryCacheRuntime).
+DECLARE_mInt32(query_cache_max_incremental_merge_count);
 DECLARE_Bool(force_regenerate_rowsetid_on_start_error);
 
 // Enable validation to check the correctness of table size.

be/src/common/metrics/doris_metrics.cpp

@@ -48,6 +48,14 @@ DEFINE_COUNTER_METRIC_PROTOTYPE_2ARG(query_scan_bytes_from_local, MetricUnit::BY
 DEFINE_COUNTER_METRIC_PROTOTYPE_2ARG(query_scan_bytes_from_remote, MetricUnit::BYTES);
 DEFINE_COUNTER_METRIC_PROTOTYPE_2ARG(query_scan_rows, MetricUnit::ROWS);
 DEFINE_COUNTER_METRIC_PROTOTYPE_2ARG(query_scan_count, MetricUnit::NOUNIT);
+DEFINE_COUNTER_METRIC_PROTOTYPE_3ARG(query_cache_stale_hit_total, MetricUnit::REQUESTS,
+                                     "Query cache decisions that reused a stale entry by "
+                                     "incremental merge.");
+DEFINE_COUNTER_METRIC_PROTOTYPE_3ARG(query_cache_incremental_fallback_total, MetricUnit::REQUESTS,
+                                     "Query cache decisions that could have merged incrementally "
+                                     "but fell back to a full recompute.");
+DEFINE_COUNTER_METRIC_PROTOTYPE_3ARG(query_cache_write_back_total, MetricUnit::REQUESTS,
+                                     "Query cache entries written back (full or merged).");
 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, "",
@@ -291,6 +299,9 @@ DorisMetrics::DorisMetrics() : _metric_registry(_s_registry_name) {
     INT_COUNTER_METRIC_REGISTER(_server_metric_entity, query_scan_bytes_from_local);
     INT_COUNTER_METRIC_REGISTER(_server_metric_entity, query_scan_bytes_from_remote);
     INT_COUNTER_METRIC_REGISTER(_server_metric_entity, query_scan_rows);
+    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, push_requests_success_total);
     INT_COUNTER_METRIC_REGISTER(_server_metric_entity, push_requests_fail_total);

be/src/common/metrics/doris_metrics.h

@@ -54,6 +54,15 @@ class DorisMetrics {
     IntCounter* query_scan_bytes_from_remote = nullptr;
     IntCounter* query_scan_rows = nullptr;
 
+    // Query cache incremental merge (see runtime/query_cache/query_cache.h):
+    // how many instance decisions reused a stale entry incrementally, how many
+    // could have but fell back to a full recompute, and how many entries were
+    // written back. Per-query breakdown lives in the profile
+    // (HitCacheStale / IncrementalFallbackReason / InsertCache).
+    IntCounter* query_cache_stale_hit_total = nullptr;
+    IntCounter* query_cache_incremental_fallback_total = nullptr;
+    IntCounter* query_cache_write_back_total = nullptr;
+
     IntCounter* push_requests_success_total = nullptr;
     IntCounter* push_requests_fail_total = nullptr;
     IntCounter* push_request_duration_us = nullptr;

be/src/exec/operator/cache_source_operator.cpp

@@ -35,11 +35,11 @@ Status CacheSourceLocalState::init(RuntimeState* state, LocalStateInfo& info) {
     ((DataQueueSharedState*)_dependency->shared_state())
             ->data_queue.set_source_dependency(_shared_state->source_deps.front());
     const auto& scan_ranges = info.scan_ranges;
-    bool hit_cache = false;
 
-    const auto& cache_param = _parent->cast<CacheSourceOperatorX>()._cache_param;
+    auto& parent = _parent->cast<CacheSourceOperatorX>();
+    const auto& cache_param = parent._cache_param;
     // 1. init the slot orders
-    const auto& tuple_descs = _parent->cast<CacheSourceOperatorX>().row_desc().tuple_descriptors();
+    const auto& tuple_descs = parent.row_desc().tuple_descriptors();
     for (auto tuple_desc : tuple_descs) {
         for (auto slot_desc : tuple_desc->slots()) {
             if (cache_param.output_slot_mapping.find(slot_desc->id()) !=
@@ -53,8 +53,6 @@ Status CacheSourceLocalState::init(RuntimeState* state, LocalStateInfo& info) {
         }
     }
 
-    // 2. build cache key by digest_tablet_id
-    RETURN_IF_ERROR(QueryCache::build_cache_key(scan_ranges, cache_param, &_cache_key, &_version));
     std::vector<int64_t> cache_tablet_ids;
     cache_tablet_ids.reserve(scan_ranges.size());
     for (const auto& scan_range : scan_ranges) {
@@ -70,14 +68,38 @@ Status CacheSourceLocalState::init(RuntimeState* state, LocalStateInfo& info) {
     }
     custom_profile()->add_info_string("CacheTabletId", tablet_ids_str);
 
-    // 3. lookup the cache and find proper slot order
-    if (!cache_param.force_refresh_query_cache) {
-        hit_cache = _global_cache->lookup(_cache_key, _version, &_query_cache_handle);
+    // 2. consume the per-instance cache decision. It is made exactly once for
+    // this instance and shared with the olap scan operator, so both operators
+    // always act consistently (e.g. the scan skips scanning if and only if this
+    // operator emits the cached blocks) -- whatever their init order is, and
+    // even if the entry gets evicted in between (the decision pins it).
+    if (parent._query_cache_runtime == nullptr) {
+        // Should not happen: the fragment context always creates the runtime
+        // together with this operator. Degrade to an uncached scan.
+        LOG(WARNING) << "query cache runtime is absent, node id " << cache_param.node_id;
+        _need_insert_cache = false;
+        custom_profile()->add_info_string("HitCache", "0");
+        return Status::OK();
     }
-    custom_profile()->add_info_string("HitCache", std::to_string(hit_cache));
-    if (hit_cache) {
-        _hit_cache_results = _query_cache_handle.get_cache_result();
-        auto hit_cache_slot_orders = _query_cache_handle.get_cache_slot_orders();
+    _global_cache = parent._query_cache_runtime->cache();
+    _cache_decision = parent._query_cache_runtime->get_or_make_decision(scan_ranges);
+    _cache_key = _cache_decision->cache_key;
+    _version = _cache_decision->current_version;
+
+    using Mode = QueryCacheInstanceDecision::Mode;
+    const bool hit_cache = _cache_decision->mode == Mode::HIT;
+    _is_incremental = _cache_decision->mode == Mode::INCREMENTAL;
+    // HIT emits the entry unchanged, so there is nothing to write back; both
+    // 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;
+    _insert_delta_count = _is_incremental ? _cache_decision->cached_delta_count + 1 : 0;
+
+    if (hit_cache || _is_incremental) {
+        _hit_cache_results = _cache_decision->handle.get_cache_result();
+        auto hit_cache_slot_orders = _cache_decision->handle.get_cache_slot_orders();
 
         if (_slot_orders != *hit_cache_slot_orders) {
             for (auto slot_id : _slot_orders) {
@@ -96,6 +118,18 @@ Status CacheSourceLocalState::init(RuntimeState* state, LocalStateInfo& info) {
         }
     }
 
+    custom_profile()->add_info_string("HitCache", std::to_string(hit_cache));
+    custom_profile()->add_info_string("HitCacheStale", std::to_string(_is_incremental));
+    if (_is_incremental) {
+        custom_profile()->add_info_string(
+                "IncrementalDeltaVersions",
+                fmt::format("({}, {}]", _cache_decision->cached_version, _version));
+    }
+    if (!_cache_decision->incremental_fallback_reason.empty()) {
+        custom_profile()->add_info_string("IncrementalFallbackReason",
+                                          _cache_decision->incremental_fallback_reason);
+    }
+
     return Status::OK();
 }
 
@@ -107,6 +141,32 @@ Status CacheSourceLocalState::open(RuntimeState* state) {
     return Status::OK();
 }
 
+bool CacheSourceLocalState::_account_write_back(int64_t rows, int64_t bytes) {
+    _current_query_cache_rows += rows;
+    _current_query_cache_bytes += bytes;
+    const auto& cache_param = _parent->cast<CacheSourceOperatorX>()._cache_param;
+    if (cache_param.entry_max_bytes < _current_query_cache_bytes ||
+        cache_param.entry_max_rows < _current_query_cache_rows) {
+        // over the max bytes/rows: pass the data through, no need to do cache
+        _local_cache_blocks.clear();
+        _need_insert_cache = false;
+        return false;
+    }
+    return true;
+}
+
+Status CacheSourceLocalState::_append_block_for_write_back(Block& block) {
+    if (!_account_write_back(block.rows(), block.allocated_bytes())) {
+        return Status::OK();
+    }
+    auto& cloned = _local_cache_blocks.emplace_back(Block::create_unique());
+    cloned->swap(block.clone_empty());
+    ScopedMutableBlock scoped_mutable_block(cloned.get());
+    auto& mutable_block = scoped_mutable_block.mutable_block();
+    RETURN_IF_ERROR(mutable_block.merge(block));
+    return Status::OK();
+}
+
 std::string CacheSourceLocalState::debug_string(int indentation_level) const {
     fmt::memory_buffer debug_string_buffer;
     fmt::format_to(debug_string_buffer, "{}", Base::debug_string(indentation_level));
@@ -125,7 +185,49 @@ Status CacheSourceOperatorX::get_block_impl(RuntimeState* state, Block* block, b
     block->clear_column_data(_row_descriptor.num_materialized_slots());
     bool need_clone_empty = block->columns() == 0;
 
-    if (local_state._hit_cache_results == nullptr) {
+    const bool has_cached_block =
+            local_state._hit_cache_results != nullptr &&
+            local_state._hit_cache_pos < local_state._hit_cache_results->size();
+
+    if (has_cached_block) {
+        // Emit one cached block: the whole result on HIT, or the cached part
+        // ahead of the delta on INCREMENTAL. Both the cached blocks and the
+        // delta blocks are partial aggregation states, so the upstream merge
+        // aggregation combines them into the final result.
+        // Note: this operator is only scheduled once the data queue has data or
+        // finished, so on INCREMENTAL the cached blocks are not emitted before
+        // the first delta block arrives. Making the source dependency initially
+        // ready for HIT/INCREMENTAL would overlap emitting the cached part with
+        // the delta scan -- a possible future latency optimization.
+        const auto& hit_cache_block =
+                local_state._hit_cache_results->at(local_state._hit_cache_pos++);
+        if (need_clone_empty) {
+            *block = hit_cache_block->clone_empty();
+        }
+        {
+            ScopedMutableBlock scoped_mutable_block(block);
+            auto& mutable_block = scoped_mutable_block.mutable_block();
+            RETURN_IF_ERROR(mutable_block.merge(*hit_cache_block));
+            scoped_mutable_block.restore();
+        }
+        if (!local_state._hit_cache_column_orders.empty()) {
+            auto datas = block->get_columns_with_type_and_name();
+            block->clear();
+            for (auto loc : local_state._hit_cache_column_orders) {
+                block->insert(datas[loc]);
+            }
+        }
+        if (local_state._is_incremental && local_state._need_insert_cache) {
+            // The emitted block is already reordered to this query's slot
+            // orders; keep a copy so the written-back entry holds
+            // "cached + delta" under one consistent slot order.
+            RETURN_IF_ERROR(local_state._append_block_for_write_back(*block));
+        }
+    } else if (local_state._hit_cache_results != nullptr && !local_state._is_incremental) {
+        // HIT: all cached blocks are emitted.
+        *eos = true;
+    } else {
+        // MISS, or the delta phase of INCREMENTAL after the cached blocks.
         Defer insert_cache([&] {
             if (*eos) {
                 local_state.custom_profile()->add_info_string(
@@ -134,7 +236,8 @@ Status CacheSourceOperatorX::get_block_impl(RuntimeState* state, Block* block, b
                     local_state._global_cache->insert(local_state._cache_key, local_state._version,
                                                       local_state._local_cache_blocks,
                                                       local_state._slot_orders,
-                                                      local_state._current_query_cache_bytes);
+                                                      local_state._current_query_cache_bytes,
+                                                      local_state._insert_delta_count);
                     local_state._local_cache_blocks.clear();
                 }
             }
@@ -160,42 +263,13 @@ Status CacheSourceOperatorX::get_block_impl(RuntimeState* state, Block* block, b
             auto& mutable_block = scoped_mutable_block.mutable_block();
             RETURN_IF_ERROR(mutable_block.merge(*output_block));
             scoped_mutable_block.restore();
-            local_state._current_query_cache_rows += output_block->rows();
-            auto mem_consume = output_block->allocated_bytes();
-            local_state._current_query_cache_bytes += mem_consume;
-
-            if (_cache_param.entry_max_bytes < local_state._current_query_cache_bytes ||
-                _cache_param.entry_max_rows < local_state._current_query_cache_rows) {
-                // over the max bytes, pass through the data, no need to do cache
-                local_state._local_cache_blocks.clear();
-                local_state._need_insert_cache = false;
-            } else {
+            if (local_state._account_write_back(output_block->rows(),
+                                                output_block->allocated_bytes())) {
                 local_state._local_cache_blocks.emplace_back(std::move(output_block));
             }
         } else {
             *block = std::move(*output_block);
         }
-    } else {
-        if (local_state._hit_cache_pos < local_state._hit_cache_results->size()) {
-            const auto& hit_cache_block =
-                    local_state._hit_cache_results->at(local_state._hit_cache_pos++);
-            if (need_clone_empty) {
-                *block = hit_cache_block->clone_empty();
-            }
-            ScopedMutableBlock scoped_mutable_block(block);
-            auto& mutable_block = scoped_mutable_block.mutable_block();
-            RETURN_IF_ERROR(mutable_block.merge(*hit_cache_block));
-            scoped_mutable_block.restore();
-            if (!local_state._hit_cache_column_orders.empty()) {
-                auto datas = block->get_columns_with_type_and_name();
-                block->clear();
-                for (auto loc : local_state._hit_cache_column_orders) {
-                    block->insert(datas[loc]);
-                }
-            }
-        } else {
-            *eos = true;
-        }
     }
 
     local_state.reached_limit(block, eos);

be/src/exec/operator/cache_source_operator.h

@@ -48,6 +48,16 @@ class CacheSourceLocalState final : public PipelineXLocalState<DataQueueSharedSt
     friend class CacheSourceOperatorX;
     friend class OperatorX<CacheSourceLocalState>;
 
+    // Account `rows`/`bytes` against the entry limits. Once the limits are
+    // exceeded, drop the pending write-back blocks and stop caching (the data
+    // itself still passes through to the parent). Returns whether the entry is
+    // still cacheable.
+    bool _account_write_back(int64_t rows, int64_t bytes);
+    // Clone `block` (already reordered to this query's slot orders) into the
+    // write-back set. Used in INCREMENTAL mode for the cached blocks, so the
+    // written-back entry holds "cached + delta" under one consistent order.
+    Status _append_block_for_write_back(Block& block);
+
     QueryCache* _global_cache = QueryCache::instance();
 
     std::string _cache_key {};
@@ -58,7 +68,18 @@ class CacheSourceLocalState final : public PipelineXLocalState<DataQueueSharedSt
     int64_t _current_query_cache_rows = 0;
     bool _need_insert_cache = true;
 
-    QueryCacheHandle _query_cache_handle;
+    // The per-instance cache decision shared with the olap scan operator of the
+    // same fragment (both observe the same object, so they can never disagree).
+    // It also pins the reused cache entry for the lifetime of this query.
+    std::shared_ptr<QueryCacheInstanceDecision> _cache_decision;
+    // Shortcut for _cache_decision->mode == INCREMENTAL: after the cached
+    // blocks are emitted, the delta partial result is drained from the data
+    // queue, and both are written back as the merged entry.
+    bool _is_incremental = false;
+    // delta_count to write back: 0 for a full recompute, cached + 1 for an
+    // incremental merge (drives periodic compaction, see QueryCacheRuntime).
+    int64_t _insert_delta_count = 0;
+
     std::vector<BlockUPtr>* _hit_cache_results = nullptr;
     std::vector<int> _hit_cache_column_orders;
     int _hit_cache_pos = 0;
@@ -68,8 +89,11 @@ class CacheSourceOperatorX final : public OperatorX<CacheSourceLocalState> {
 public:
     using Base = OperatorX<CacheSourceLocalState>;
     CacheSourceOperatorX(ObjectPool* pool, int plan_node_id, int operator_id,
-                         const TQueryCacheParam& cache_param)
-            : Base(pool, plan_node_id, operator_id), _cache_param(cache_param) {
+                         const TQueryCacheParam& cache_param,
+                         std::shared_ptr<QueryCacheRuntime> query_cache_runtime)
+            : Base(pool, plan_node_id, operator_id),
+              _cache_param(cache_param),
+              _query_cache_runtime(std::move(query_cache_runtime)) {
         _op_name = "CACHE_SOURCE_OPERATOR";
     };
 
@@ -90,6 +114,7 @@ class CacheSourceOperatorX final : public OperatorX<CacheSourceLocalState> {
 
 private:
     TQueryCacheParam _cache_param;
+    std::shared_ptr<QueryCacheRuntime> _query_cache_runtime;
     bool _has_data(RuntimeState* state) const {
         auto& local_state = get_local_state(state);
         return local_state._shared_state->data_queue.remaining_has_data();