github-actions[bot] commented on code in PR #65482:
URL: https://github.com/apache/doris/pull/65482#discussion_r3596103090
##########
be/src/runtime/query_cache/query_cache.cpp:
##########
@@ -70,4 +97,279 @@ bool QueryCache::lookup(const CacheKey& key, int64_t
version, doris::QueryCacheH
return false;
}
+bool QueryCache::lookup_any_version(const CacheKey& key, QueryCacheHandle*
handle) {
+
SCOPED_SWITCH_THREAD_MEM_TRACKER_LIMITER(ExecEnv::GetInstance()->query_cache_mem_tracker());
+ auto* lru_handle = LRUCachePolicy::lookup(key);
+ if (lru_handle) {
+ *handle = QueryCacheHandle(this, lru_handle);
+ return true;
+ }
+ return false;
+}
+
+QueryCacheInstanceDecision::~QueryCacheInstanceDecision() = default;
+
+std::unique_ptr<TabletReadSource>
QueryCacheInstanceDecision::take_delta_read_source(
+ int64_t tablet_id) {
+ std::lock_guard<std::mutex> lock(_take_lock);
+ auto it = _delta_read_sources.find(tablet_id);
+ if (it == _delta_read_sources.end()) {
+ return nullptr;
+ }
+ auto res = std::move(it->second);
+ _delta_read_sources.erase(it);
+ return res;
+}
+
+std::shared_ptr<QueryCacheInstanceDecision>
QueryCacheRuntime::get_or_make_decision(
+ const std::vector<TScanRangeParams>& scan_ranges) {
+ std::string cache_key;
+ int64_t version = 0;
+ Status st = QueryCache::build_cache_key(scan_ranges, _param, &cache_key,
&version);
+ if (!st.ok()) {
+ // No reliable cache key for this instance (e.g. FE could not align the
+ // instance to a single partition so tablets carry different versions).
+ // Degrade to an uncached scan instead of failing the query: both the
+ // scan operator and the cache source operator observe the shared MISS
+ // decision with key_valid=false, so nothing is looked up and nothing
+ // is written back. This is an expected plan shape, so log only once
+ // per fragment instead of twice per instance.
+ std::lock_guard<std::mutex> lock(_lock);
+ if (_invalid_decision == nullptr) {
+ LOG(WARNING) << "query cache degrades to uncached scan, node_id="
<< _param.node_id
+ << ", reason: " << st.to_string();
+ _invalid_decision = std::make_shared<QueryCacheInstanceDecision>();
+ }
+ return _invalid_decision;
+ }
+
+ // The lock also serializes decision making of different instances of this
+ // fragment. That is acceptable in the common case: a decision only touches
+ // tablet metadata (no data IO), so it finishes in microseconds to
+ // milliseconds. On very wide fragments (hundreds of tablets per instance)
+ // combined with tablet header lock contention (e.g. compaction meta
+ // commits) this serialization can stretch the fragment init tail; if that
+ // ever shows up in profiles, move the capture out of the lock and let the
+ // losing racer drop its duplicate decision.
+ std::lock_guard<std::mutex> lock(_lock);
Review Comment:
[P2] Do not hold the fragment mutex across bitmap scans
`_lock` remains held through `_make_decision()`. For a stale merge-on-write
entry, each instance then captures every tablet in its assigned scan-range set,
and `_delta_rewrites_history()` takes the tablet's delete-bitmap lock in shared
mode while scanning its entire historical map. Because this runtime is
fragment-wide, unrelated instance keys initialize serially; a large bitmap also
blocks load/compaction writers that need the lock exclusively. The cost is not
bounded by the microsecond-to-millisecond assumption above. Please use per-key
single-flight state or build decisions outside the map mutex, and avoid a full
live-bitmap traversal under a writer-excluding lock (for example with a
version-indexed summary or bounded snapshot).
##########
be/src/exec/operator/cache_source_operator.cpp:
##########
@@ -125,7 +187,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));
Review Comment:
[P1] Rebuild each cached block before remapping it
On the second cached block, `block` still has columns: pipeline tasks reuse
it and `clear_column_data()` preserves the schema that the previous pull
already reordered at lines 215-220. This merge therefore applies cached-order
columns positionally to current-order destinations before the permutation. For
normalized-equivalent queries with swapped heterogeneous outputs, such as an
INT group key and BIGINT aggregate, the second block enters
`MutableBlock::merge_impl()`'s unsupported type-mismatch path before it can be
permuted, so stale incremental reuse fails instead of returning the query
result. Please materialize every cached block in cached order before remapping
it (or permute the source before merging), and cover stale reuse with two
cached blocks and reordered heterogeneous slots.
--
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]
