asdf2014 commented on code in PR #65482:
URL: https://github.com/apache/doris/pull/65482#discussion_r3600701053
##########
fe/fe-core/src/main/java/org/apache/doris/qe/SessionVariable.java:
##########
@@ -1544,6 +1545,36 @@ public enum IgnoreSplitType {
@VarAttrDef.VarAttr(name = ENABLE_QUERY_CACHE, fuzzy = true)
public boolean enableQueryCache = false;
+ // Allow BE to reuse a stale query cache entry by scanning only the delta
+ // rowsets since the cached version and merging them with the cached
partial
+ // aggregation blocks. Only takes effect when the cache point aggregation
is
+ // non-finalize (its output is merged again upstream) and the selected
index
+ // is append-only: DUP_KEYS, or merge-on-write UNIQUE_KEYS whose delta did
+ // not rewrite pre-existing keys (BE checks the delete bitmap per tablet);
+ // BE falls back to a full recompute whenever the delta cannot be captured
+ // (compacted away), contains delete predicates or rewrites history rows.
+ // Experimental: shown as `experimental_enable_query_cache_incremental` and
+ // settable with or without the prefix; to be promoted to
EXPERIMENTAL_ONLINE
+ // once it graduates.
+ @VarAttrDef.VarAttr(name = ENABLE_QUERY_CACHE_INCREMENTAL,
+ varType = VariableAnnotation.EXPERIMENTAL, needForward = true,
Review Comment:
Fixed in b25ce1f51a3. You are right, and the call path is exactly as you
describe: query cache normalization runs on whichever FE plans the statement, a
forwarded statement is planned by the master in a fresh ConnectContext, and
getForwardVariables() sends only the annotated variables, so the incremental
flag arrived alone and both planner gates skipped normalization.
`enable_query_cache` now carries needForward as well, which also closes the
same gap the base query cache had on its own before this PR: a session-level
setting, or a SET_VAR hint, never reached the planner of a forwarded statement,
which instead followed the master's global value.
Chasing the one hop further found a second gap that forwarding the switch
alone would have opened: QueryCacheNormalizer builds the cache param on the
planning FE from three more session variables, so with the cache now engaged on
forwarded paths a forced refresh would have been silently dropped and entries
would have been sized by the master's defaults. `query_cache_force_refresh`,
`query_cache_entry_max_bytes` and `query_cache_entry_max_rows` are therefore
forwarded too, so every variable the query cache reads at plan time travels
with the statement. The digest is unaffected, since it is salted from the
variables annotated as affecting the query result and needForward does not join
that set.
Two consequences worth stating plainly rather than leaving for you to find.
First, forwarded statements now also inherit the scan-side semantics
`enable_query_cache` already carries locally: OlapScanNode pins the fixed
replica, and as the skip_missing_version doc notes, that pinning turns the
missing-version rescue off. This lands on more than the follower-cannot-read
window, since every redirected statement is forwarded, including each
non-group-commit INSERT issued on a follower. In every case the forwarded plan
now matches what the same session already gets without forwarding, so this
makes the forwarded path stop being the odd one out rather than introducing a
new mode. Second, an adjacent gap I deliberately did not widen this PR with:
`short_circuit_evaluation` is serialized into the normalized plan and therefore
into the digest, but it is not forwarded, so a session that sets it away from
the default gets a different digest when its statement is forwarded, which
costs a dupl
icate entry and one recompute but never correctness. Forwarding it would
change execution semantics for all forwarded queries, well beyond the cache, so
it belongs in its own change. The same holds for `enable_sql_cache`,
`enable_hive_sql_cache` and `enable_condition_cache`, which are read at plan
time and not forwarded either; I would rather file those as one separate issue
than bundle three distinct features into a query-cache pairing fix.
On coverage: the contract is pinned in SessionVariablesTest, where a
follower-side session sets all five, and a fresh master-side session is
asserted to receive every one of them through
getForwardVariables()/setForwardedSessionVariables() instead of keeping its
defaults. That test fails on the previous code at the first assertion, since
the base switch was absent from the forward map. A true two-FE
follower-to-master case is also feasible here, `test_forward_qeury.groovy` does
exactly that with connectToFollower plus the forward-all-queries debug point,
so if you would rather see the end-to-end version, say so and I will add it; my
reasoning for the unit level is that the defect lives entirely in the
annotation-to-forward-map contract, which the unit test pins per variable,
whereas a docker suite would spend a two-FE cluster to re-observe the same
contract through a plan.
##########
be/src/exec/operator/olap_scan_operator.cpp:
##########
@@ -698,14 +698,23 @@ Status
OlapScanLocalState::_init_scanners(std::list<ScannerSPtr>* scanners) {
bool read_row_binlog =
p._olap_scan_node.__isset.read_row_binlog &&
p._olap_scan_node.read_row_binlog;
bool has_tso_predicate = _scan_ranges[0]->__isset.start_tso ||
_scan_ranges[0]->__isset.end_tso;
+ // Query cache incremental merge: the read sources captured in prepare()
+ // only cover the delta versions (cached_version, current_version], which
+ // is all the scanners need; no version plumbing is required here.
+ const bool cache_incremental =
+ _query_cache_decision != nullptr &&
+ _query_cache_decision->mode ==
QueryCacheInstanceDecision::Mode::INCREMENTAL;
// The flag of preagg's meaning is whether return pre agg data(or partial
agg data)
// PreAgg ON: The storage layer returns partially aggregated data without
additional processing. (Fast data reading)
// for example, if a table is select userid,count(*) from base table.
// And the user send a query like select userid,count(*) from base table
group by userid.
// then the storage layer do not need do aggregation, it could just return
the partial agg data, because the compute layer will do aggregation.
// PreAgg OFF: The storage layer must complete pre-aggregation and return
fully aggregated data. (Slow data reading)
- if (enable_parallel_scan && !p._should_run_serial &&
+ // Incremental merge deltas are small by construction, so the parallel
+ // scanner builder (which redistributes rowsets by size) is pointless for
+ // them; use plain scanners instead.
+ if (enable_parallel_scan && !cache_incremental && !p._should_run_serial &&
Review Comment:
Correction to my previous note on this thread, before it misleads anyone:
the dormant phase I added did not split that delta into many scanners, and my
claim that it did was wrong. `parallel_scan_min_rows_per_scanner` defaults to
2M rows, and BE takes `max(1024, session_value)`, so lowering the variable only
reaches a floor of 1024 rows, which a 210-row delta never crosses: the phase
ran with exactly one scanner and covered none of the splitting it was written
for. Fixed in b25ce1f51a3, which keeps the lowered variable, grows the delta to
6000 rows across 30 rowsets, and states the real mechanism in the comment:
against the 1024-row floor the builder cuts that into four or five scanners, so
the per-scanner reader clone and the delete-predicate/delete-bitmap copy into
each partial read source are actually exercised. Thanks for asking for this
coverage; the request stood up better than my first attempt at it.
--
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]
