asdf2014 commented on code in PR #65482:
URL: https://github.com/apache/doris/pull/65482#discussion_r3580274997
##########
be/src/runtime/query_cache/query_cache.cpp:
##########
@@ -70,4 +97,267 @@ 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);
+ auto it = _decisions.find(cache_key);
+ if (it != _decisions.end()) {
+ return it->second;
+ }
+
+ auto decision = std::make_shared<QueryCacheInstanceDecision>();
+ decision->key_valid = true;
+ decision->cache_key = cache_key;
+ decision->current_version = version;
+ _make_decision(scan_ranges, decision.get());
+ _decisions.emplace(std::move(cache_key), decision);
+ return decision;
+}
+
+void QueryCacheRuntime::_make_decision(const std::vector<TScanRangeParams>&
scan_ranges,
+ QueryCacheInstanceDecision* decision) {
+ decision->mode = QueryCacheInstanceDecision::Mode::MISS;
+ if (_binlog_scan) {
+ // A row-binlog scan reads a different data stream under the same plan
+ // digest: it must neither serve cached blocks nor write its output
+ // back (that would poison the cache for normal queries).
+ decision->key_valid = false;
+ return;
+ }
+ if (_param.force_refresh_query_cache) {
+ return;
+ }
+
+ QueryCacheHandle handle;
+ if (!_cache->lookup_any_version(decision->cache_key, &handle)) {
+ return;
+ }
+ // Pin the entry: as long as this decision object lives, the entry cannot
be
+ // evicted, so every operator consuming this decision sees the same blocks.
+ decision->handle = std::move(handle);
+
+ if (decision->handle.get_cache_version() == decision->current_version) {
+ decision->mode = QueryCacheInstanceDecision::Mode::HIT;
+ decision->cached_delta_count =
decision->handle.get_cache_delta_count();
+ return;
+ }
+
+ if (_try_prepare_incremental(scan_ranges, decision)) {
+ decision->mode = QueryCacheInstanceDecision::Mode::INCREMENTAL;
+ DorisMetrics::instance()->query_cache_stale_hit_total->increment(1);
+ return;
+ }
+
+ // Stale entry not reusable: drop the pin and fall back to a full scan.
+ decision->_delta_read_sources.clear();
+ decision->handle = QueryCacheHandle();
+ decision->mode = QueryCacheInstanceDecision::Mode::MISS;
+ if (!decision->incremental_fallback_reason.empty()) {
+ // Only count real fallbacks: an empty reason means incremental merge
+ // was not enabled for this query in the first place.
+
DorisMetrics::instance()->query_cache_incremental_fallback_total->increment(1);
+ }
+}
+
+bool QueryCacheRuntime::_try_prepare_incremental(const
std::vector<TScanRangeParams>& scan_ranges,
+ QueryCacheInstanceDecision*
decision) {
+ if (!(_param.__isset.allow_incremental && _param.allow_incremental)) {
+ // Not a fallback: incremental merge is simply not enabled for this
+ // query, so leave incremental_fallback_reason empty.
+ return false;
+ }
+ // Cloud tablets capture rowsets through a different (partly asynchronous)
+ // path; incremental merge only supports local storage for now.
+ if (config::is_cloud_mode()) {
Review Comment:
Cloud mode is planned. The first version sticks to local storage because the
approach is easiest to validate there (the delta capture and the merge-on-write
bitmap check run against a view that is complete by construction), which keeps
the impact of this PR contained. Once the approach is confirmed workable, I
will submit a separate follow-up PR to complete cloud support.
##########
be/src/runtime/query_cache/query_cache.cpp:
##########
@@ -70,4 +97,267 @@ 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);
+ auto it = _decisions.find(cache_key);
+ if (it != _decisions.end()) {
+ return it->second;
+ }
+
+ auto decision = std::make_shared<QueryCacheInstanceDecision>();
+ decision->key_valid = true;
+ decision->cache_key = cache_key;
+ decision->current_version = version;
+ _make_decision(scan_ranges, decision.get());
+ _decisions.emplace(std::move(cache_key), decision);
+ return decision;
+}
+
+void QueryCacheRuntime::_make_decision(const std::vector<TScanRangeParams>&
scan_ranges,
+ QueryCacheInstanceDecision* decision) {
+ decision->mode = QueryCacheInstanceDecision::Mode::MISS;
+ if (_binlog_scan) {
+ // A row-binlog scan reads a different data stream under the same plan
+ // digest: it must neither serve cached blocks nor write its output
+ // back (that would poison the cache for normal queries).
+ decision->key_valid = false;
+ return;
+ }
+ if (_param.force_refresh_query_cache) {
+ return;
+ }
+
+ QueryCacheHandle handle;
+ if (!_cache->lookup_any_version(decision->cache_key, &handle)) {
+ return;
+ }
+ // Pin the entry: as long as this decision object lives, the entry cannot
be
+ // evicted, so every operator consuming this decision sees the same blocks.
+ decision->handle = std::move(handle);
+
+ if (decision->handle.get_cache_version() == decision->current_version) {
+ decision->mode = QueryCacheInstanceDecision::Mode::HIT;
+ decision->cached_delta_count =
decision->handle.get_cache_delta_count();
+ return;
+ }
+
+ if (_try_prepare_incremental(scan_ranges, decision)) {
+ decision->mode = QueryCacheInstanceDecision::Mode::INCREMENTAL;
+ DorisMetrics::instance()->query_cache_stale_hit_total->increment(1);
+ return;
+ }
+
+ // Stale entry not reusable: drop the pin and fall back to a full scan.
+ decision->_delta_read_sources.clear();
+ decision->handle = QueryCacheHandle();
+ decision->mode = QueryCacheInstanceDecision::Mode::MISS;
+ if (!decision->incremental_fallback_reason.empty()) {
+ // Only count real fallbacks: an empty reason means incremental merge
+ // was not enabled for this query in the first place.
+
DorisMetrics::instance()->query_cache_incremental_fallback_total->increment(1);
+ }
+}
+
+bool QueryCacheRuntime::_try_prepare_incremental(const
std::vector<TScanRangeParams>& scan_ranges,
+ QueryCacheInstanceDecision*
decision) {
+ if (!(_param.__isset.allow_incremental && _param.allow_incremental)) {
+ // Not a fallback: incremental merge is simply not enabled for this
+ // query, so leave incremental_fallback_reason empty.
+ return false;
+ }
+ // Cloud tablets capture rowsets through a different (partly asynchronous)
+ // path; incremental merge only supports local storage for now.
+ if (config::is_cloud_mode()) {
+ decision->incremental_fallback_reason = "cloud mode";
+ return false;
+ }
+
+ int64_t cached_version = decision->handle.get_cache_version();
+ if (cached_version >= decision->current_version) {
+ // The entry is newer than what this replica is asked to read (e.g. the
+ // entry was filled from another replica with a newer visible version).
+ // A full scan of the requested version is the only safe answer.
+ decision->incremental_fallback_reason = "cached entry is newer";
+ return false;
+ }
+ int64_t cached_delta_count = decision->handle.get_cache_delta_count();
+ if (cached_delta_count >= config::query_cache_max_incremental_merge_count)
{
+ // Force a full recompute to compact the entry: every incremental merge
+ // appends the delta blocks to the entry, so both the entry and the
+ // upstream merge get more fragmented as deltas accumulate.
+ decision->incremental_fallback_reason = "delta count reached
compaction threshold";
+ return false;
+ }
+
+ for (const auto& scan_range : scan_ranges) {
+ int64_t tablet_id = scan_range.scan_range.palo_scan_range.tablet_id;
+ if (!_capture_tablet_delta(tablet_id, cached_version, decision)) {
+ return false;
+ }
+ }
+
+ // If the cached entry alone already exceeds the entry limits, the merged
+ // entry (which can only be larger) could never be written back. Still scan
+ // only the delta, but tell the cache source upfront so it does not clone
+ // blocks for a write back that would be discarded anyway. Such an entry
+ // stays stale until compaction merges its delta away (then the capture
+ // above fails and a full recompute takes over).
+ if (decision->handle.get_cache_total_bytes() > _param.entry_max_bytes ||
Review Comment:
Exactly, that is the scenario. `entry_max_bytes` and `entry_max_rows` come
from the session variables and are sent with every query, but they do not
participate in the cache digest, so a query with a smaller limit still hits the
entry written under a larger one. The write-back side already refuses to store
anything above the current limits (the size accounting in
`cache_source_operator.cpp`), so with unchanged limits a cached entry can never
exceed them; the only way this condition turns true is a limit lowered after
the entry was written: `set` in the same session, a different session using a
smaller value for the same digest, or an admin lowering the global value to cap
the cache memory footprint.
Without this branch such a query would still deep-copy the cached blocks
into the write-back accumulation and pay one extra merge copy per delta block,
only for the size accounting to clear it all once the running total crosses the
limits; and since the entry can never be replaced under the new limits, copying
up to the configured limit would repeat on every stale query. With the branch
we skip the accumulation upfront but keep the delta scan benefit. Both
directions are covered in the UTs (`write_back_feasible` true and false).
--
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]
