asdf2014 commented on code in PR #65482:
URL: https://github.com/apache/doris/pull/65482#discussion_r3604667950
##########
be/src/runtime/query_cache/query_cache.cpp:
##########
@@ -70,4 +100,334 @@ 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;
+ }
+
+ {
+ std::lock_guard<std::mutex> lock(_lock);
+ auto it = _decisions.find(cache_key);
+ if (it != _decisions.end()) {
+ return it->second;
+ }
+ }
+
+ // Build the candidate outside the lock: a stale-entry decision walks
+ // tablet metadata under the header lock and, on merge-on-write tables,
+ // scans the delete bitmap, so its cost grows with the tablets per
+ // instance and the bitmap size. This runtime is fragment-wide, so a
+ // single lock-scoped build would serialize even unrelated instances'
+ // keys behind one slow decision. Racing callers of the same instance
+ // build duplicate candidates; the loser adopts the winner's decision
+ // below and its own candidate releases with the shared_ptr (the entry
+ // pin and any captured delta read sources go with it), which only costs
+ // a redundant metadata pass. The metrics are settled once, by the
+ // winner, after publication.
+ 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());
+ TEST_SYNC_POINT("QueryCacheRuntime::get_or_make_decision.before_publish");
+
+ std::lock_guard<std::mutex> lock(_lock);
+ auto [it, inserted] = _decisions.emplace(std::move(cache_key), decision);
+ if (inserted) {
+ if (decision->mode == QueryCacheInstanceDecision::Mode::INCREMENTAL) {
+
DorisMetrics::instance()->query_cache_stale_hit_total->increment(1);
+ } else if (decision->mode == QueryCacheInstanceDecision::Mode::MISS &&
+ !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);
+ }
+ }
+ return it->second;
+}
+
+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;
+ return;
+ }
+
+ // Stale entry not reusable: drop the pin and fall back to a full scan.
+ // The stale-hit/fallback metrics are settled by the caller once the
+ // winning candidate is published, so a losing racer cannot double-count.
+ decision->_delta_read_sources.clear();
+ decision->handle = QueryCacheHandle();
+ decision->mode = QueryCacheInstanceDecision::Mode::MISS;
+}
+
+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 ||
+ decision->handle.get_cache_total_rows() > _param.entry_max_rows) {
+ decision->write_back_feasible = false;
+ }
+
+ decision->cached_version = cached_version;
+ decision->cached_delta_count = cached_delta_count;
+ return true;
+}
+
+bool QueryCacheRuntime::_capture_tablet_delta(int64_t tablet_id, int64_t
cached_version,
+ QueryCacheInstanceDecision*
decision) {
+ auto tablet_res = ExecEnv::get_tablet(tablet_id);
+ if (!tablet_res) {
+ decision->incremental_fallback_reason = "tablet not found";
+ return false;
+ }
+ BaseTabletSPtr tablet = std::move(tablet_res.value());
+ // Defensive re-check of what FE promised with allow_incremental:
+ // "cached snapshot + delta rowsets == new snapshot" holds unconditionally
+ // for append-only DUP_KEYS data, and for merge-on-write UNIQUE data as
+ // long as the delta did not rewrite any row that predates the cached
+ // version (verified below once the delta rowsets are known). It never
+ // holds for merge-on-read UNIQUE data (duplicates are resolved by merging
+ // across rowsets at read time, so a delta-only scan cannot stand alone)
+ // nor for AGG tables (rows merge in the storage layer likewise).
+ const bool merge_on_write = tablet->keys_type() == KeysType::UNIQUE_KEYS &&
+ tablet->enable_unique_key_merge_on_write();
+ if (tablet->keys_type() != KeysType::DUP_KEYS && !merge_on_write) {
+ decision->incremental_fallback_reason = "keys type not append-only";
+ return false;
+ }
+
+ // quiet: on a version-graph miss (the delta merged away by compaction, an
+ // expected per-query situation) this option makes the capture API neither
+ // log nor error; it returns whatever PREFIX of the path it walked before
+ // the miss (empty when the window start itself is gone), so emptiness AND
+ // window coverage are both checked below as the failure signals.
+ auto source_res = tablet->capture_read_source({cached_version + 1,
decision->current_version},
+ {.quiet = true});
+ if (!source_res) {
+ // Non-pathfinding failures (e.g. a rowset object missing for a found
+ // path) still surface as errors even under quiet.
+ decision->incremental_fallback_reason = "delta versions not
capturable";
+ return false;
+ }
+ auto read_source =
std::make_unique<TabletReadSource>(std::move(source_res.value()));
+ if (read_source->rs_splits.empty()) {
+ // A non-empty version window always lies on a consistent path with at
+ // least one rowset; an empty capture therefore means the delta
+ // versions are gone, never "no new data". Treating it as INCREMENTAL
+ // would silently drop the delta rows from the result.
+ decision->incremental_fallback_reason = "delta versions not
capturable";
+ return false;
+ }
+ // A quiet capture broken mid-window (e.g. a replica whose local view ends
+ // short of current_version) yields a partial prefix, and treating it as
+ // INCREMENTAL would silently drop the tail rows and poison the entry on
+ // write-back. The path is contiguous by construction, so checking both
+ // endpoints pins the whole (cached_version, current_version] window.
+ if (read_source->rs_splits.front().rs_reader->rowset()->start_version() !=
cached_version + 1 ||
+ read_source->rs_splits.back().rs_reader->rowset()->end_version() !=
+ decision->current_version) {
+ decision->incremental_fallback_reason = "delta versions not
capturable";
+ return false;
+ }
+ read_source->fill_delete_predicates();
+ if (!read_source->delete_predicates.empty()) {
+ // A delete in the delta logically removes rows that are already folded
+ // into the cached partial result; that cannot be undone by merging, so
+ // fall back to a full recompute.
+ decision->incremental_fallback_reason = "delta contains delete
predicates";
+ return false;
+ }
+ if (merge_on_write &&
Review Comment:
Fixed in 789c8b4158f. The race is real and the mechanism is exactly as you
describe: the delta capture pins only the delta rowsets, an overwritten
cached-side row has no RowIdConversion target so its marker is not relocated
(calc_compaction_output_rowset_delete_bitmap skips a source row whose id does
not convert), and the stale sweep plus unused-rowset GC can then drop that
marker before _delta_rewrites_history reads it, leaving the cached aggregate to
merge with the rows that replaced it.
The classification now pins the cached side before it runs: under the tablet
header lock it captures the rowsets of [0, cached_version] and holds them
across _delta_rewrites_history. What makes the pin load-bearing is
start_delete_unused_rowset(): it collects a retired rowset only once nothing
else references it, and only a collected rowset reaches
remove_rowset_delete_bitmap(), which drops every version of that rowset's
bitmap, so a reference kept here keeps the evidence too. The capture is quiet
and returns whatever prefix it walked, so the pinned path is checked to cover
the whole range; a cached side whose own history was compacted away and swept
comes back as a short prefix and falls back to a full recompute rather than
classifying against evidence it cannot vouch for.
For the sync-point coverage you asked for: two tests run against a real
storage engine. mow_cached_side_pinned_across_classification installs a
callback at a sync point inside the classification and asserts that the full
[0, cached_version] range is pinned at that exact moment;
mow_cached_side_not_pinnable_falls_back gives the cached side a hole so the
quiet capture returns a short prefix and asserts the coverage check falls back.
Both were confirmed to fail on the pre-fix code by reverting the product change
and rerunning: the first because the pin is not held at the sync point, the
second because the incremental path is taken instead of the fallback.
Orchestrating the actual compaction+sweep+GC interleaving inside a unit test is
impractical, so the tests pin the two invariants the fix rests on (full-range
coverage held into classification, and fallback on an incomplete pin) rather
than the timing itself.
--
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]
