asdf2014 commented on code in PR #65482:
URL: https://github.com/apache/doris/pull/65482#discussion_r3597194218
##########
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:
Fixed in b73ecf77946. The cached block is now reordered to this query's slot
order through a zero-copy column view before the merge, so every merge is
order-aligned no matter what schema the reused output block carries, and the
write-back keeps storing blocks in this query's order together with its
slot_orders, so chained permuted hits stay consistent. One honest note on
reachability: today the second-pull shape is latent, because this operator is
constructed without a plan node and its own row descriptor reports zero
materialized slots, so clear_column_data() wipes the block and need_clone_empty
is true on every pull (on master as well). The driver-side clears are
schema-keeping though, so giving the operator a real row descriptor or dropping
the redundant wipe would have armed it silently. The two new operator tests
inject a real slot count and reuse one block across pulls, exactly the armed
shape: they crash on the pre-fix code and pin the invariant against that
cleanup.
##########
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:
Fixed in b73ecf77946, both parts. The decision is now built outside the
fragment mutex and published through an emplace race: the lock covers only the
map find/emplace plus one atomic counter increment, a losing racer adopts the
winner's decision and releases its duplicate candidate (entry pin and captured
read sources) outside the lock, and the metrics are settled exactly once by the
published winner. Three racer tests pin this through a pre-publish sync point:
one shared decision object, counters moving exactly once across the
INCREMENTAL, plain-MISS and fallback modes, and the winner's captured read
source intact. The history-rewrite scan no longer walks the whole delete bitmap
under its shared lock: it hops between (rowset, segment) groups with
lower_bound/upper_bound, so the writer-excluding hold is bounded by the
in-window entries plus one logarithmic hop per group instead of the full map.
The comparisons stay in the uint64 key domain, which keeps the hop strictly
advancing
for arbitrary map contents; a lone-group out-of-int64-range pin verifies the
int64-cast variant would livelock there, and a later-segment pin verifies the
hop never skips past its own (rowset, segment) group. A version-indexed summary
looked oversized for this one window check; the group hop already removes the
O(map) hold.
--
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]
