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

be/src/common/config.cpp

@@ -1704,6 +1704,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

@@ -1759,6 +1759,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,15 @@ 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 handed to the cache to be written back "
+                                     "(full or merged); admission may still turn one down.");
 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 +300,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,17 @@ 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
+    // handed to the cache to be written back (the cache may still turn one
+    // down: its LRU-K admission only keeps a key that comes back while the
+    // shard is full). 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,40 @@ 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) {
+        // The fragment context always creates the runtime together with this
+        // operator. Degrading to a pass-through here would silently drop data
+        // if the paired scan operator still made a HIT decision (it skips
+        // scanning while nothing emits the entry), so a broken setup must
+        // fail loudly, mirroring the scan side.
+        return Status::InternalError(
+                "query cache runtime is absent at the cache source, node_id={}",
+                cache_param.node_id);
     }
-    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 +120,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 +143,35 @@ 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(const Block& block) {
+    if (!_account_write_back(block.rows(), block.allocated_bytes())) {
+        return Status::OK();
+    }
+    // Zero-copy snapshot: inserting shares the COW column pointers instead
+    // of cloning the payload. The cached columns stay alive (and immutable)
+    // through the decision's entry pin until QueryCache::insert() deep-copies
+    // the accumulated set once.
+    auto& kept = _local_cache_blocks.emplace_back(Block::create_unique());
+    for (const auto& column : block.get_columns_with_type_and_name()) {
+        kept->insert(column);
+    }
+    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));
@@ -124,25 +189,100 @@ 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) {
-        Defer insert_cache([&] {
+    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++);
+        // Reorder the cached block to this query's slot order BEFORE merging
+        // (a zero-copy view: inserting reuses the COW column pointers).
+        // Merging first and permuting afterwards is an order trap on the
+        // second cached block whenever the reused output block keeps its
+        // schema between pulls: a cached-order block then merges positionally
+        // into query-order columns -- a type error for heterogeneous slots,
+        // silently misplaced data for same-typed ones. Today the trap stays
+        // latent by accident: this operator is built without a plan node, so
+        // its own _row_descriptor reports zero materialized slots and the
+        // clear_column_data() above wipes the whole block every pull. The
+        // pipeline driver's clears are schema-keeping though, so giving this
+        // operator a real row descriptor (or dropping the redundant-looking
+        // wipe above) would arm it; reordering the source keeps the merge
+        // order-aligned under both block shapes.
+        Block reordered_cache_block;
+        const Block* cache_block_to_merge = hit_cache_block.get();
+        if (!local_state._hit_cache_column_orders.empty()) {
+            for (auto loc : local_state._hit_cache_column_orders) {
+                reordered_cache_block.insert(hit_cache_block->get_by_position(loc));
+            }
+            cache_block_to_merge = &reordered_cache_block;
+        }
+        if (need_clone_empty) {
+            *block = cache_block_to_merge->clone_empty();
+        }
+        {
+            ScopedMutableBlock scoped_mutable_block(block);
+            auto& mutable_block = scoped_mutable_block.mutable_block();
+            RETURN_IF_ERROR(mutable_block.merge(*cache_block_to_merge));
+            scoped_mutable_block.restore();
+        }
+        if (local_state._is_incremental && local_state._need_insert_cache) {
+            // Snapshot the cached block (already in this query's slot order)
+            // for the write-back, so the new entry holds "cached + delta"
+            // under one consistent slot order. The snapshot shares the COW
+            // columns of the pinned entry, so no payload is copied before
+            // the insert-time materialization.
+            RETURN_IF_ERROR(local_state._append_block_for_write_back(*cache_block_to_merge));
+        }
+    } 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.
+        // The entry is committed only from the explicit success paths below,
+        // never during error unwinding: on the final delta block *eos is set
+        // BEFORE the block is merged, so a deferred/unconditional commit
+        // would publish an entry that is missing the failed block under the
+        // current version, and a later exact hit would silently serve it.
+        auto commit_entry_if_finished = [&]() {
             if (*eos) {
                 local_state.custom_profile()->add_info_string(
                         "InsertCache", std::to_string(local_state._need_insert_cache));
                 if (local_state._need_insert_cache) {
                     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();
+                    // Latch off so the commit is exactly-once by construction:
+                    // a spurious re-poll after eos (no in-tree driver does
+                    // this today) would otherwise re-publish the now-cleared,
+                    // EMPTY block set over the entry just inserted, and later
+                    // exact hits would silently serve an empty result. This
+                    // is the only source operator whose eos path carries a
+                    // global side effect, so it must not rely on the pull
+                    // protocol never poking it again.
+                    local_state._need_insert_cache = false;
                 }
             }
-        });
+        };
 
         auto queue_block = DORIS_TRY(local_state._shared_state->data_queue.get_block_from_queue());
         *eos = queue_block.eos;
 
         if (!queue_block.block) {
+            commit_entry_if_finished();
             return Status::OK();
         }
 
@@ -155,42 +295,14 @@ 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;
-        }
+        commit_entry_if_finished();
     }
 
     local_state.reached_limit(block, eos);