asdf2014 commented on code in PR #65482:
URL: https://github.com/apache/doris/pull/65482#discussion_r3599774490
##########
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:
Fixed in bdd1b13ec05. The commit is no longer tied to scope exit: the
unconditional Defer is gone, and the entry is published only from the two
explicit success exits, the empty pull that observes eos and the exit where the
pulled block was fully consumed. Every error return now leaves the cache
untouched, so the window you describe (eos observed at the top of the pull, the
merge failing afterwards; allocation failure is reachable there as a non-OK
Status through RETURN_IF_CATCH_EXCEPTION) can no longer cache the incomplete
prefix, and the pinned stale entry stays the served truth for later queries.
Two side effects of the rewrite worth noting: the exception-unwind exit is
closed as well (the old Defer ran the insert inside a destructor during
unwinding, which could also std::terminate if the insert itself threw), and the
insert is now latched exactly-once, so even an out-of-contract extra poll after
eos cannot replace the committed entry with an empty set. A new test injects a
de
terministic merge failure on the final delta block (a malformed two-column
block, with the row descriptor pinned so the reused block keeps its schema) and
asserts the query fails, nothing is published at the current version, and the
stale entry still serves its old version untouched; on the previous code it
goes red precisely by publishing the incomplete 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:
Fixed in bdd1b13ec05. The write-back set no longer deep-copies the cached
payload: the cached blocks are snapshotted as zero-copy views sharing the COW
column pointers of the pinned entry (the reordered view is itself zero-copy),
and the single materialization stays inside QueryCache::insert under the cache
mem tracker. A replacement now peaks at the old entry plus the new entry
instead of roughly three payloads, which also shrinks the window where the
displaced-but-pinned old entry sits outside the LRU accounting. Delta blocks
were already moved into the set and are unchanged. The reordered write-back
test pins the mechanism from both sides: before the insert the kept columns are
pointer-identical to the pinned entry's columns in this query's order, and
after the insert the new entry aliases none of them, which proves the single
materialization. Accounting input is equivalent in substance, the shared
columns report the same allocated bytes the old merged copy did, so the entry li
mits and the LRU charge read the same values as before.
--
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]
