github-actions[bot] commented on code in PR #65482:
URL: https://github.com/apache/doris/pull/65482#discussion_r3597506691
##########
be/src/exec/operator/cache_source_operator.cpp:
##########
@@ -125,7 +187,64 @@ 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++);
+ // Reorder the cached block to this query's slot order BEFORE merging
+ // (a zero-copy view: inserting reuses the COW column pointers).
+ // Merging first and permuting afterwards is an order trap on the
+ // second cached block whenever the reused output block keeps its
+ // schema between pulls: a cached-order block then merges positionally
+ // into query-order columns -- a type error for heterogeneous slots,
+ // silently misplaced data for same-typed ones. Today the trap stays
+ // latent by accident: this operator is built without a plan node, so
+ // its own _row_descriptor reports zero materialized slots and the
+ // clear_column_data() above wipes the whole block every pull. The
+ // pipeline driver's clears are schema-keeping though, so giving this
+ // operator a real row descriptor (or dropping the redundant-looking
+ // wipe above) would arm it; reordering the source keeps the merge
+ // order-aligned under both block shapes.
+ Block reordered_cache_block;
+ const Block* cache_block_to_merge = hit_cache_block.get();
+ if (!local_state._hit_cache_column_orders.empty()) {
+ for (auto loc : local_state._hit_cache_column_orders) {
+
reordered_cache_block.insert(hit_cache_block->get_by_position(loc));
+ }
+ cache_block_to_merge = &reordered_cache_block;
+ }
+ if (need_clone_empty) {
+ *block = cache_block_to_merge->clone_empty();
+ }
+ {
+ ScopedMutableBlock scoped_mutable_block(block);
+ auto& mutable_block = scoped_mutable_block.mutable_block();
+ RETURN_IF_ERROR(mutable_block.merge(*cache_block_to_merge));
+ scoped_mutable_block.restore();
+ }
+ if (local_state._is_incremental && local_state._need_insert_cache) {
+ // The emitted block is already reordered to this query's slot
+ // orders; keep a copy so the written-back entry holds
+ // "cached + delta" under one consistent slot order.
+ RETURN_IF_ERROR(local_state._append_block_for_write_back(*block));
Review Comment:
[P1] Do not publish after a failed final delta merge
On the final delta block, `*eos` has already been set before
`mutable_block.merge(*output_block)` runs. If that merge returns non-OK (for
example, a caught allocation failure), `RETURN_IF_ERROR` unwinds through the
unconditional `insert_cache` Defer. Because incremental mode has already copied
the stale cached blocks here, the Defer publishes that incomplete prefix at the
current version/new delta count, and a later exact hit can silently omit the
unmerged delta. Please commit the replacement only on the explicit success path
(or gate the Defer on completed merge/accounting), and add a test proving a
final-block merge failure leaves no current-version entry.
##########
be/src/exec/operator/cache_source_operator.cpp:
##########
@@ -107,6 +143,32 @@ Status CacheSourceLocalState::open(RuntimeState* state) {
return Status::OK();
}
+bool CacheSourceLocalState::_account_write_back(int64_t rows, int64_t bytes) {
+ _current_query_cache_rows += rows;
+ _current_query_cache_bytes += bytes;
+ const auto& cache_param =
_parent->cast<CacheSourceOperatorX>()._cache_param;
+ if (cache_param.entry_max_bytes < _current_query_cache_bytes ||
+ cache_param.entry_max_rows < _current_query_cache_rows) {
+ // over the max bytes/rows: pass the data through, no need to do cache
+ _local_cache_blocks.clear();
+ _need_insert_cache = false;
+ return false;
+ }
+ return true;
+}
+
+Status CacheSourceLocalState::_append_block_for_write_back(Block& block) {
+ if (!_account_write_back(block.rows(), block.allocated_bytes())) {
+ return Status::OK();
+ }
+ auto& cloned = _local_cache_blocks.emplace_back(Block::create_unique());
Review Comment:
[P2] Avoid the extra full cached-payload copy on replacement
This deep-copies each pinned cached block into `_local_cache_blocks`, and
`QueryCache::insert()` deep-copies that accumulated entry again while the
decision still pins the old `CacheValue`. At insertion, incremental reuse can
therefore overlap roughly old + local + new payloads for each active instance.
The per-entry limit bounds each materialization, but it does not reserve their
combined peak; the replacement is cloned before LRU capacity enforcement, and
same-key insertion removes the still-pinned old entry from LRU usage while its
tracked memory remains live. Near-limit entries across many active instances
can therefore push physical memory toward roughly 3x their payload and create
avoidable query or process pressure. Please retain zero-copy references/views
to the pinned cached blocks until one cache-owned materialization (or use an
ownership/transfer path with correct tracker accounting), and cover peak
allocation/copy count in a multi-instance near-limit test.
--
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]
