asdf2014 commented on code in PR #65482:
URL: https://github.com/apache/doris/pull/65482#discussion_r3586910510
##########
be/src/exec/pipeline/pipeline_fragment_context.cpp:
##########
@@ -1519,9 +1519,24 @@ Status
PipelineFragmentContext::_create_operator(ObjectPool* pool, const TPlanNo
bool fe_with_old_version = false;
switch (tnode.node_type) {
case TPlanNodeType::OLAP_SCAN_NODE: {
+ std::shared_ptr<QueryCacheRuntime> query_cache_runtime;
+ if (enable_query_cache) {
+ if (_query_cache_runtime == nullptr) {
Review Comment:
Fixed in ba09543cf71. The plan tree is built in pre-order and the cache
source sits above the scan, so the runtime created with it must already exist
when the scan node is reached; a missing one means FE sent a malformed plan
shape. Silently creating a runtime here could even drop data: with a HIT
decision the scan skips scanning while no cache source emits the entry. So the
lazy creation is replaced with an `InternalError`. The same reasoning applies
to the cache source side, whose old degrade-to-pass-through branch is now an
error too, and both paths are covered by unit tests
(`QueryCacheFragmentContextTest` on the scan side,
`QueryCacheOperatorTest.test_missing_runtime_fails_init` on the cache source
side).
##########
be/src/exec/pipeline/pipeline_fragment_context.cpp:
##########
@@ -1519,9 +1519,24 @@ Status
PipelineFragmentContext::_create_operator(ObjectPool* pool, const TPlanNo
bool fe_with_old_version = false;
switch (tnode.node_type) {
case TPlanNodeType::OLAP_SCAN_NODE: {
+ std::shared_ptr<QueryCacheRuntime> query_cache_runtime;
+ if (enable_query_cache) {
+ if (_query_cache_runtime == nullptr) {
+ _query_cache_runtime =
+
std::make_shared<QueryCacheRuntime>(_params.fragment.query_cache_param);
+ }
+ if (tnode.olap_scan_node.__isset.read_row_binlog &&
+ tnode.olap_scan_node.read_row_binlog) {
+ // Row-binlog scans read a different data stream: they must
+ // neither serve nor fill the query cache.
+ _query_cache_runtime->disable_for_binlog_scan();
+ }
+ query_cache_runtime = _query_cache_runtime;
+ }
op = std::make_shared<OlapScanOperatorX>(
pool, tnode, next_operator_id(), descs, _num_instances,
- enable_query_cache ? _params.fragment.query_cache_param :
TQueryCacheParam {});
+ enable_query_cache ? _params.fragment.query_cache_param :
TQueryCacheParam {},
+ std::move(query_cache_runtime));
Review Comment:
The scan does need the shared runtime: the runtime is the single decision
point for the instance. Whichever operator initializes first triggers
`get_or_make_decision()`, and both consume the same decision object, so the
pair always agrees: the scan skips scanning if and only if the cache source
emits just the cached blocks (HIT), and scans only the delta if and only if the
cache source merges it (INCREMENTAL). Two independent lookups are exactly the
pre-existing race this PR removes: they could interleave with an eviction and
write back an empty poisoned entry. The local `shared_ptr` copy was indeed
unnecessary; ba09543cf71 passes the member directly.
##########
be/src/exec/operator/olap_scan_operator.cpp:
##########
@@ -698,14 +698,24 @@ 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 only cover the delta
+ // versions (cached_version, current_version], see prepare().
+ const bool cache_incremental =
+ _query_cache_decision != nullptr &&
+ _query_cache_decision->mode ==
QueryCacheInstanceDecision::Mode::INCREMENTAL;
+ const int64_t cache_start_version =
+ cache_incremental ? _query_cache_decision->cached_version + 1 : 0;
// 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
Review Comment:
You are right. Removed in ba09543cf71. I traced every consumer of the
reader's `version.first`: the delete predicates and the merge-on-write bitmap
only use `version.second`, the segment-level single-version checks use the
rowset's own version, and the scanner's self-capture path only runs when
`rs_splits` is empty, which the pre-captured delta read source makes
unreachable. So `Params.start_version` and its plumbing are gone and the reader
keeps the plain `{0, version}` range. The `cache_incremental` flag itself
stays: it still gates `take_delta_read_source()` in `prepare()` and keeps the
parallel scanner builder off for delta scans. The delta is typically a couple
of small rowsets landed since the cached version, and the plain path still
gives every tablet its own scanner, so size-based redistribution has nothing to
split there and only adds the builder's segment loading cost.
##########
be/src/exec/operator/olap_scan_operator.cpp:
##########
@@ -698,14 +698,24 @@ 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 only cover the delta
+ // versions (cached_version, current_version], see prepare().
+ const bool cache_incremental =
+ _query_cache_decision != nullptr &&
+ _query_cache_decision->mode ==
QueryCacheInstanceDecision::Mode::INCREMENTAL;
+ const int64_t cache_start_version =
Review Comment:
Removed in ba09543cf71 together with the plumbing below. The
`cache_incremental` flag itself stays: it still gates
`take_delta_read_source()` and keeps the parallel scanner builder off for delta
scans (details in the sibling thread).
--
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]
