From b4294bb5dba96c044c13e542a65bd4e39722204e Mon Sep 17 00:00:00 2001 From: Benedict Jin Date: Sat, 11 Jul 2026 20:14:21 +0800 Subject: [PATCH] [feature](query-cache) Support incremental merge for stale query cache entries Hourly batch loads keep bumping the version of the hot partition, so its query cache entries never hit again and every "last N days" aggregation recomputes the whole partition each hour. This change lets BE reuse a stale entry instead of discarding it: scan only the delta rowsets in (cached_version, current_version], emit their partial aggregation state side by side with the cached partial blocks (the upstream merge aggregation combines both), and write the merged entry back under the new version. Correctness rests on two properties: append-only snapshots decompose as S(v2) = S(v1) + delta, and partial aggregation states merge homomorphically. Every precondition guards one of them, and any violation falls back to a full recompute silently: - FE only sets TQueryCacheParam.allow_incremental (new optional field 8) when the experimental session switch enable_query_cache_incremental is on, the cache point is a non-finalize aggregation directly above the olap scan node, and the selected index is append-only: DUP_KEYS, or merge-on-write UNIQUE_KEYS. - BE re-validates per tablet at decision time: local storage mode, version order, the compaction threshold (query_cache_max_incremental_merge_count, BE config, default 8), keys type, delta capturability, no delete predicates in the delta, and for merge-on-write tables a delete-bitmap window check that rejects any load that rewrote pre-existing keys, so an occasional backfill costs exactly one full recompute and re-bases the entry. A new fragment-level QueryCacheRuntime makes one idempotent decision (HIT / INCREMENTAL / MISS) per instance, pins the entry and pre-captures the delta read source. Centralizing the decision also fixes three pre-existing defects: the scan/cache-source double-lookup race that could write back an empty poisoned entry, row-binlog scans polluting the cache, and build_cache_key failures aborting the whole query instead of degrading to uncached execution. Observability: profile fields HitCacheStale, IncrementalDeltaVersions and IncrementalFallbackReason, plus three BE metrics (query_cache_stale_hit_total, query_cache_incremental_fallback_total, query_cache_write_back_total). Benchmark (80M-row duplicate-key table, 200k-row hourly appends, group by a non-distribution column): a stale query drops from ~307ms (full recompute) to ~19ms with incremental merge, on par with an exact hit, and the cost no longer grows with the base data size. --- be/src/common/config.cpp | 6 + be/src/common/config.h | 3 + be/src/common/metrics/doris_metrics.cpp | 11 + be/src/common/metrics/doris_metrics.h | 9 + .../exec/operator/cache_source_operator.cpp | 164 +++-- be/src/exec/operator/cache_source_operator.h | 31 +- be/src/exec/operator/olap_scan_operator.cpp | 82 ++- be/src/exec/operator/olap_scan_operator.h | 15 +- .../pipeline/pipeline_fragment_context.cpp | 24 +- .../exec/pipeline/pipeline_fragment_context.h | 7 + be/src/exec/scan/olap_scanner.cpp | 2 +- be/src/exec/scan/olap_scanner.h | 4 + be/src/runtime/query_cache/query_cache.cpp | 296 +++++++++- be/src/runtime/query_cache/query_cache.h | 192 +++++- .../operator/query_cache_operator_test.cpp | 434 +++++++++++++- be/test/exec/pipeline/query_cache_test.cpp | 558 ++++++++++++++++++ .../apache/doris/planner/AggregationNode.java | 4 + .../normalize/QueryCacheNormalizer.java | 70 +++ .../org/apache/doris/qe/SessionVariable.java | 39 ++ .../planner/QueryCacheNormalizerTest.java | 90 ++- gensrc/thrift/QueryCache.thrift | 29 +- .../cache/query_cache_incremental.groovy | 183 ++++++ 22 files changed, 2161 insertions(+), 92 deletions(-) create mode 100644 regression-test/suites/query_p0/cache/query_cache_incremental.groovy diff --git a/be/src/common/config.cpp b/be/src/common/config.cpp index 9ec587bd100ec5..be67881eab92b5 100644 --- a/be/src/common/config.cpp +++ b/be/src/common/config.cpp @@ -1690,6 +1690,12 @@ DEFINE_mBool(enable_pipeline_task_leakage_detect, "false"); DEFINE_mInt32(check_score_rounds_num, "1000"); DEFINE_Int32(query_cache_size, "512"); +// Max number of incremental merges accumulated on one query cache entry before +// a full recompute is forced. Each incremental merge appends the delta partial +// blocks to the entry, so the entry gets more fragmented (and the upstream merge +// aggregation does more work) as deltas accumulate; a periodic full recompute +// compacts the entry back to a minimal set of blocks. +DEFINE_mInt32(query_cache_max_incremental_merge_count, "8"); // Enable validation to check the correctness of table size. DEFINE_Bool(enable_table_size_correctness_check, "false"); diff --git a/be/src/common/config.h b/be/src/common/config.h index 012f09a735c0cd..49a5307a205c94 100644 --- a/be/src/common/config.h +++ b/be/src/common/config.h @@ -1750,6 +1750,9 @@ DECLARE_mInt32(check_score_rounds_num); // MB DECLARE_Int32(query_cache_size); +// Max incremental merges on one query cache entry before forcing a full +// recompute to compact the entry (see query_cache.h QueryCacheRuntime). +DECLARE_mInt32(query_cache_max_incremental_merge_count); DECLARE_Bool(force_regenerate_rowsetid_on_start_error); // Enable validation to check the correctness of table size. diff --git a/be/src/common/metrics/doris_metrics.cpp b/be/src/common/metrics/doris_metrics.cpp index 85317133e12382..64d542970098eb 100644 --- a/be/src/common/metrics/doris_metrics.cpp +++ b/be/src/common/metrics/doris_metrics.cpp @@ -48,6 +48,14 @@ DEFINE_COUNTER_METRIC_PROTOTYPE_2ARG(query_scan_bytes_from_local, MetricUnit::BY DEFINE_COUNTER_METRIC_PROTOTYPE_2ARG(query_scan_bytes_from_remote, MetricUnit::BYTES); DEFINE_COUNTER_METRIC_PROTOTYPE_2ARG(query_scan_rows, MetricUnit::ROWS); DEFINE_COUNTER_METRIC_PROTOTYPE_2ARG(query_scan_count, MetricUnit::NOUNIT); +DEFINE_COUNTER_METRIC_PROTOTYPE_3ARG(query_cache_stale_hit_total, MetricUnit::REQUESTS, + "Query cache decisions that reused a stale entry by " + "incremental merge."); +DEFINE_COUNTER_METRIC_PROTOTYPE_3ARG(query_cache_incremental_fallback_total, MetricUnit::REQUESTS, + "Query cache decisions that could have merged incrementally " + "but fell back to a full recompute."); +DEFINE_COUNTER_METRIC_PROTOTYPE_3ARG(query_cache_write_back_total, MetricUnit::REQUESTS, + "Query cache entries written back (full or merged)."); DEFINE_COUNTER_METRIC_PROTOTYPE_5ARG(push_requests_success_total, MetricUnit::REQUESTS, "", push_requests_total, Labels({{"status", "SUCCESS"}})); DEFINE_COUNTER_METRIC_PROTOTYPE_5ARG(push_requests_fail_total, MetricUnit::REQUESTS, "", @@ -291,6 +299,9 @@ DorisMetrics::DorisMetrics() : _metric_registry(_s_registry_name) { INT_COUNTER_METRIC_REGISTER(_server_metric_entity, query_scan_bytes_from_local); INT_COUNTER_METRIC_REGISTER(_server_metric_entity, query_scan_bytes_from_remote); INT_COUNTER_METRIC_REGISTER(_server_metric_entity, query_scan_rows); + INT_COUNTER_METRIC_REGISTER(_server_metric_entity, query_cache_stale_hit_total); + INT_COUNTER_METRIC_REGISTER(_server_metric_entity, query_cache_incremental_fallback_total); + INT_COUNTER_METRIC_REGISTER(_server_metric_entity, query_cache_write_back_total); INT_COUNTER_METRIC_REGISTER(_server_metric_entity, push_requests_success_total); INT_COUNTER_METRIC_REGISTER(_server_metric_entity, push_requests_fail_total); diff --git a/be/src/common/metrics/doris_metrics.h b/be/src/common/metrics/doris_metrics.h index 764d5dd956a490..ac6e993698d6ad 100644 --- a/be/src/common/metrics/doris_metrics.h +++ b/be/src/common/metrics/doris_metrics.h @@ -54,6 +54,15 @@ class DorisMetrics { IntCounter* query_scan_bytes_from_remote = nullptr; IntCounter* query_scan_rows = nullptr; + // Query cache incremental merge (see runtime/query_cache/query_cache.h): + // how many instance decisions reused a stale entry incrementally, how many + // could have but fell back to a full recompute, and how many entries were + // written back. Per-query breakdown lives in the profile + // (HitCacheStale / IncrementalFallbackReason / InsertCache). + IntCounter* query_cache_stale_hit_total = nullptr; + IntCounter* query_cache_incremental_fallback_total = nullptr; + IntCounter* query_cache_write_back_total = nullptr; + IntCounter* push_requests_success_total = nullptr; IntCounter* push_requests_fail_total = nullptr; IntCounter* push_request_duration_us = nullptr; diff --git a/be/src/exec/operator/cache_source_operator.cpp b/be/src/exec/operator/cache_source_operator.cpp index ec7d947680a6f8..7182868d950a0e 100644 --- a/be/src/exec/operator/cache_source_operator.cpp +++ b/be/src/exec/operator/cache_source_operator.cpp @@ -35,11 +35,11 @@ Status CacheSourceLocalState::init(RuntimeState* state, LocalStateInfo& info) { ((DataQueueSharedState*)_dependency->shared_state()) ->data_queue.set_source_dependency(_shared_state->source_deps.front()); const auto& scan_ranges = info.scan_ranges; - bool hit_cache = false; - const auto& cache_param = _parent->cast()._cache_param; + auto& parent = _parent->cast(); + const auto& cache_param = parent._cache_param; // 1. init the slot orders - const auto& tuple_descs = _parent->cast().row_desc().tuple_descriptors(); + const auto& tuple_descs = parent.row_desc().tuple_descriptors(); for (auto tuple_desc : tuple_descs) { for (auto slot_desc : tuple_desc->slots()) { if (cache_param.output_slot_mapping.find(slot_desc->id()) != @@ -53,8 +53,6 @@ Status CacheSourceLocalState::init(RuntimeState* state, LocalStateInfo& info) { } } - // 2. build cache key by digest_tablet_id - RETURN_IF_ERROR(QueryCache::build_cache_key(scan_ranges, cache_param, &_cache_key, &_version)); std::vector cache_tablet_ids; cache_tablet_ids.reserve(scan_ranges.size()); for (const auto& scan_range : scan_ranges) { @@ -70,14 +68,38 @@ Status CacheSourceLocalState::init(RuntimeState* state, LocalStateInfo& info) { } custom_profile()->add_info_string("CacheTabletId", tablet_ids_str); - // 3. lookup the cache and find proper slot order - if (!cache_param.force_refresh_query_cache) { - hit_cache = _global_cache->lookup(_cache_key, _version, &_query_cache_handle); + // 2. consume the per-instance cache decision. It is made exactly once for + // this instance and shared with the olap scan operator, so both operators + // always act consistently (e.g. the scan skips scanning if and only if this + // operator emits the cached blocks) -- whatever their init order is, and + // even if the entry gets evicted in between (the decision pins it). + if (parent._query_cache_runtime == nullptr) { + // Should not happen: the fragment context always creates the runtime + // together with this operator. Degrade to an uncached scan. + LOG(WARNING) << "query cache runtime is absent, node id " << cache_param.node_id; + _need_insert_cache = false; + custom_profile()->add_info_string("HitCache", "0"); + return Status::OK(); } - custom_profile()->add_info_string("HitCache", std::to_string(hit_cache)); - if (hit_cache) { - _hit_cache_results = _query_cache_handle.get_cache_result(); - auto hit_cache_slot_orders = _query_cache_handle.get_cache_slot_orders(); + _global_cache = parent._query_cache_runtime->cache(); + _cache_decision = parent._query_cache_runtime->get_or_make_decision(scan_ranges); + _cache_key = _cache_decision->cache_key; + _version = _cache_decision->current_version; + + using Mode = QueryCacheInstanceDecision::Mode; + const bool hit_cache = _cache_decision->mode == Mode::HIT; + _is_incremental = _cache_decision->mode == Mode::INCREMENTAL; + // HIT emits the entry unchanged, so there is nothing to write back; both + // MISS and INCREMENTAL rebuild the entry (from scratch / by merge), unless + // the decision already knows the merged entry could never fit the entry + // limits (write_back_feasible == false). + _need_insert_cache = + _cache_decision->key_valid && !hit_cache && _cache_decision->write_back_feasible; + _insert_delta_count = _is_incremental ? _cache_decision->cached_delta_count + 1 : 0; + + if (hit_cache || _is_incremental) { + _hit_cache_results = _cache_decision->handle.get_cache_result(); + auto hit_cache_slot_orders = _cache_decision->handle.get_cache_slot_orders(); if (_slot_orders != *hit_cache_slot_orders) { for (auto slot_id : _slot_orders) { @@ -96,6 +118,18 @@ Status CacheSourceLocalState::init(RuntimeState* state, LocalStateInfo& info) { } } + custom_profile()->add_info_string("HitCache", std::to_string(hit_cache)); + custom_profile()->add_info_string("HitCacheStale", std::to_string(_is_incremental)); + if (_is_incremental) { + custom_profile()->add_info_string( + "IncrementalDeltaVersions", + fmt::format("({}, {}]", _cache_decision->cached_version, _version)); + } + if (!_cache_decision->incremental_fallback_reason.empty()) { + custom_profile()->add_info_string("IncrementalFallbackReason", + _cache_decision->incremental_fallback_reason); + } + return Status::OK(); } @@ -107,6 +141,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()._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()); + cloned->swap(block.clone_empty()); + ScopedMutableBlock scoped_mutable_block(cloned.get()); + auto& mutable_block = scoped_mutable_block.mutable_block(); + RETURN_IF_ERROR(mutable_block.merge(block)); + return Status::OK(); +} + std::string CacheSourceLocalState::debug_string(int indentation_level) const { fmt::memory_buffer debug_string_buffer; fmt::format_to(debug_string_buffer, "{}", Base::debug_string(indentation_level)); @@ -125,7 +185,49 @@ 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++); + if (need_clone_empty) { + *block = hit_cache_block->clone_empty(); + } + { + ScopedMutableBlock scoped_mutable_block(block); + auto& mutable_block = scoped_mutable_block.mutable_block(); + RETURN_IF_ERROR(mutable_block.merge(*hit_cache_block)); + scoped_mutable_block.restore(); + } + if (!local_state._hit_cache_column_orders.empty()) { + auto datas = block->get_columns_with_type_and_name(); + block->clear(); + for (auto loc : local_state._hit_cache_column_orders) { + block->insert(datas[loc]); + } + } + 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)); + } + } else if (local_state._hit_cache_results != nullptr && !local_state._is_incremental) { + // HIT: all cached blocks are emitted. + *eos = true; + } else { + // MISS, or the delta phase of INCREMENTAL after the cached blocks. Defer insert_cache([&] { if (*eos) { local_state.custom_profile()->add_info_string( @@ -134,7 +236,8 @@ Status CacheSourceOperatorX::get_block_impl(RuntimeState* state, Block* block, b local_state._global_cache->insert(local_state._cache_key, local_state._version, local_state._local_cache_blocks, local_state._slot_orders, - local_state._current_query_cache_bytes); + local_state._current_query_cache_bytes, + local_state._insert_delta_count); local_state._local_cache_blocks.clear(); } } @@ -160,42 +263,13 @@ Status CacheSourceOperatorX::get_block_impl(RuntimeState* state, Block* block, b auto& mutable_block = scoped_mutable_block.mutable_block(); RETURN_IF_ERROR(mutable_block.merge(*output_block)); scoped_mutable_block.restore(); - local_state._current_query_cache_rows += output_block->rows(); - auto mem_consume = output_block->allocated_bytes(); - local_state._current_query_cache_bytes += mem_consume; - - if (_cache_param.entry_max_bytes < local_state._current_query_cache_bytes || - _cache_param.entry_max_rows < local_state._current_query_cache_rows) { - // over the max bytes, pass through the data, no need to do cache - local_state._local_cache_blocks.clear(); - local_state._need_insert_cache = false; - } else { + if (local_state._account_write_back(output_block->rows(), + output_block->allocated_bytes())) { local_state._local_cache_blocks.emplace_back(std::move(output_block)); } } else { *block = std::move(*output_block); } - } else { - if (local_state._hit_cache_pos < local_state._hit_cache_results->size()) { - const auto& hit_cache_block = - local_state._hit_cache_results->at(local_state._hit_cache_pos++); - if (need_clone_empty) { - *block = hit_cache_block->clone_empty(); - } - ScopedMutableBlock scoped_mutable_block(block); - auto& mutable_block = scoped_mutable_block.mutable_block(); - RETURN_IF_ERROR(mutable_block.merge(*hit_cache_block)); - scoped_mutable_block.restore(); - if (!local_state._hit_cache_column_orders.empty()) { - auto datas = block->get_columns_with_type_and_name(); - block->clear(); - for (auto loc : local_state._hit_cache_column_orders) { - block->insert(datas[loc]); - } - } - } else { - *eos = true; - } } local_state.reached_limit(block, eos); diff --git a/be/src/exec/operator/cache_source_operator.h b/be/src/exec/operator/cache_source_operator.h index 3a68bd337fc5b7..1d9da4ae183b18 100644 --- a/be/src/exec/operator/cache_source_operator.h +++ b/be/src/exec/operator/cache_source_operator.h @@ -48,6 +48,16 @@ class CacheSourceLocalState final : public PipelineXLocalState; + // Account `rows`/`bytes` against the entry limits. Once the limits are + // exceeded, drop the pending write-back blocks and stop caching (the data + // itself still passes through to the parent). Returns whether the entry is + // still cacheable. + bool _account_write_back(int64_t rows, int64_t bytes); + // Clone `block` (already reordered to this query's slot orders) into the + // write-back set. Used in INCREMENTAL mode for the cached blocks, so the + // written-back entry holds "cached + delta" under one consistent order. + Status _append_block_for_write_back(Block& block); + QueryCache* _global_cache = QueryCache::instance(); std::string _cache_key {}; @@ -58,7 +68,18 @@ class CacheSourceLocalState final : public PipelineXLocalState _cache_decision; + // Shortcut for _cache_decision->mode == INCREMENTAL: after the cached + // blocks are emitted, the delta partial result is drained from the data + // queue, and both are written back as the merged entry. + bool _is_incremental = false; + // delta_count to write back: 0 for a full recompute, cached + 1 for an + // incremental merge (drives periodic compaction, see QueryCacheRuntime). + int64_t _insert_delta_count = 0; + std::vector* _hit_cache_results = nullptr; std::vector _hit_cache_column_orders; int _hit_cache_pos = 0; @@ -68,8 +89,11 @@ class CacheSourceOperatorX final : public OperatorX { public: using Base = OperatorX; CacheSourceOperatorX(ObjectPool* pool, int plan_node_id, int operator_id, - const TQueryCacheParam& cache_param) - : Base(pool, plan_node_id, operator_id), _cache_param(cache_param) { + const TQueryCacheParam& cache_param, + std::shared_ptr query_cache_runtime) + : Base(pool, plan_node_id, operator_id), + _cache_param(cache_param), + _query_cache_runtime(std::move(query_cache_runtime)) { _op_name = "CACHE_SOURCE_OPERATOR"; }; @@ -90,6 +114,7 @@ class CacheSourceOperatorX final : public OperatorX { private: TQueryCacheParam _cache_param; + std::shared_ptr _query_cache_runtime; bool _has_data(RuntimeState* state) const { auto& local_state = get_local_state(state); return local_state._shared_state->data_queue.remaining_has_data(); diff --git a/be/src/exec/operator/olap_scan_operator.cpp b/be/src/exec/operator/olap_scan_operator.cpp index a5ea092ecc0bf7..7a927d1de63b1c 100644 --- a/be/src/exec/operator/olap_scan_operator.cpp +++ b/be/src/exec/operator/olap_scan_operator.cpp @@ -698,6 +698,13 @@ Status OlapScanLocalState::_init_scanners(std::list* 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) @@ -705,7 +712,10 @@ Status OlapScanLocalState::_init_scanners(std::list* scanners) { // 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 and would lose the delta version range; use plain scanners instead. + if (enable_parallel_scan && !cache_incremental && !p._should_run_serial && p._push_down_agg_type == TPushAggOp::NONE && (_storage_no_merge() || p._olap_scan_node.is_preaggregation) // binlog need to be read in order @@ -817,6 +827,7 @@ Status OlapScanLocalState::_init_scanners(std::list* scanners) { palo_scan_range.__isset.end_tso ? std::make_optional(palo_scan_range.end_tso) : std::nullopt, + cache_start_version, }); RETURN_IF_ERROR(scanner->init(state(), _conjuncts)); scanners->push_back(std::move(scanner)); @@ -981,7 +992,31 @@ Status OlapScanLocalState::prepare(RuntimeState* state) { } } + const bool cache_incremental = + _query_cache_decision != nullptr && + _query_cache_decision->mode == QueryCacheInstanceDecision::Mode::INCREMENTAL; for (size_t i = 0; i < _scan_ranges.size(); i++) { + if (cache_incremental) { + // Scan only the delta rowsets in (cached_version, current_version]. + // The read source was already captured (and verified to contain no + // delete predicate) when the cache decision was made, so a capture + // failure cannot surface here, where the cache source operator has + // already committed to emitting the cached blocks. take() returning + // null would mean the FE invariant "one instance per tablet set, no + // shared scan under query cache" was broken (someone else consumed + // this tablet's read source); failing fast is the only safe answer, + // because a silent full-scan fallback would emit the data twice + // (cached blocks + full scan). + auto delta_source = + _query_cache_decision->take_delta_read_source(_tablets[i].tablet->tablet_id()); + if (delta_source == nullptr) { + return Status::InternalError( + "query cache incremental read source is absent, tablet_id={}", + _tablets[i].tablet->tablet_id()); + } + _read_sources[i] = std::move(*delta_source); + continue; + } _read_sources[i] = DORIS_TRY(_tablets[i].tablet->capture_read_source( {0, _tablets[i].version}, {.skip_missing_versions = _state->skip_missing_version(), @@ -1049,28 +1084,29 @@ TOlapScanNode& OlapScanLocalState::olap_scan_node() const { void OlapScanLocalState::set_scan_ranges(RuntimeState* state, const std::vector& scan_ranges) { - const auto& cache_param = _parent->cast()._cache_param; - bool hit_cache = false; - // read binlog scan should not participate in query cache. - if (olap_scan_node().__isset.read_row_binlog && olap_scan_node().read_row_binlog) { - hit_cache = false; - } else if (!cache_param.digest.empty() && !cache_param.force_refresh_query_cache) { - std::string cache_key; - int64_t version = 0; - auto status = QueryCache::build_cache_key(scan_ranges, cache_param, &cache_key, &version); - if (!status.ok()) { - throw doris::Exception(doris::ErrorCode::INTERNAL_ERROR, status.msg()); + auto& parent = _parent->cast(); + // Consume the per-instance cache decision. It is made exactly once and + // shared with the cache source operator of this fragment, so the two + // operators always agree: the scan skips scanning if and only if the cache + // source emits the cached blocks (HIT), and scans only the delta rowsets if + // and only if the cache source merges them with the cached blocks + // (INCREMENTAL). The decision also pins the cache entry, so it cannot be + // evicted between the two operators' init. + // Note: an invalid cache key (e.g. FE could not align this instance to one + // partition, so tablets carry different versions) now degrades to an + // uncached full scan instead of failing the query. + if (parent._query_cache_runtime != nullptr) { + _query_cache_decision = parent._query_cache_runtime->get_or_make_decision(scan_ranges); + if (_query_cache_decision->mode == QueryCacheInstanceDecision::Mode::HIT) { + // Exact hit: the cache source emits the entry, nothing to scan. + return; } - doris::QueryCacheHandle handle; - hit_cache = QueryCache::instance()->lookup(cache_key, version, &handle); } - if (!hit_cache) { - for (auto& scan_range : scan_ranges) { - DCHECK(scan_range.scan_range.__isset.palo_scan_range); - _scan_ranges.emplace_back(new TPaloScanRange(scan_range.scan_range.palo_scan_range)); - COUNTER_UPDATE(_tablet_counter, 1); - } + for (auto& scan_range : scan_ranges) { + DCHECK(scan_range.scan_range.__isset.palo_scan_range); + _scan_ranges.emplace_back(new TPaloScanRange(scan_range.scan_range.palo_scan_range)); + COUNTER_UPDATE(_tablet_counter, 1); } } @@ -1263,10 +1299,12 @@ Status OlapScanLocalState::_build_key_ranges_and_filters() { OlapScanOperatorX::OlapScanOperatorX(ObjectPool* pool, const TPlanNode& tnode, int operator_id, const DescriptorTbl& descs, int parallel_tasks, - const TQueryCacheParam& param) + const TQueryCacheParam& param, + std::shared_ptr query_cache_runtime) : ScanOperatorX(pool, tnode, operator_id, descs, parallel_tasks), _olap_scan_node(tnode.olap_scan_node), - _cache_param(param) { + _cache_param(param), + _query_cache_runtime(std::move(query_cache_runtime)) { _output_tuple_id = tnode.olap_scan_node.tuple_id; if (_olap_scan_node.__isset.sort_info && _olap_scan_node.__isset.sort_limit) { DORIS_CHECK(_limit < 0); diff --git a/be/src/exec/operator/olap_scan_operator.h b/be/src/exec/operator/olap_scan_operator.h index 4e8bab717c6bfa..1d2d2015039d6a 100644 --- a/be/src/exec/operator/olap_scan_operator.h +++ b/be/src/exec/operator/olap_scan_operator.h @@ -33,6 +33,8 @@ namespace doris { class OlapScanner; +class QueryCacheRuntime; +struct QueryCacheInstanceDecision; } // namespace doris namespace doris { @@ -343,6 +345,12 @@ class OlapScanLocalState final : public ScanLocalState { std::vector _tablets; std::vector _read_sources; + // The per-instance query cache decision shared with the cache source + // operator of the same fragment. Null when the query cache is disabled. + // HIT: leave _scan_ranges empty so nothing is scanned; INCREMENTAL: scan + // only the pre-captured delta read sources in (cached, current] version. + std::shared_ptr _query_cache_decision; + std::map _slot_id_to_virtual_column_expr; // ---- Runtime-filter partition pruning ---- @@ -358,7 +366,8 @@ class OlapScanOperatorX final : public ScanOperatorX { public: OlapScanOperatorX(ObjectPool* pool, const TPlanNode& tnode, int operator_id, const DescriptorTbl& descs, int parallel_tasks, - const TQueryCacheParam& cache_param); + const TQueryCacheParam& cache_param, + std::shared_ptr query_cache_runtime = nullptr); Status prepare(RuntimeState* state) override; @@ -374,6 +383,10 @@ class OlapScanOperatorX final : public ScanOperatorX { friend class OlapScanLocalState; TOlapScanNode _olap_scan_node; TQueryCacheParam _cache_param; + // Shared with the cache source operator of the same fragment so both + // consume the same per-instance cache decision (see QueryCacheRuntime). + // Null when the query cache is disabled. + std::shared_ptr _query_cache_runtime; TabletSchemaSPtr _tablet_schema; }; diff --git a/be/src/exec/pipeline/pipeline_fragment_context.cpp b/be/src/exec/pipeline/pipeline_fragment_context.cpp index 5664042dd24e15..45ad8366180258 100644 --- a/be/src/exec/pipeline/pipeline_fragment_context.cpp +++ b/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 query_cache_runtime; + if (enable_query_cache) { + if (_query_cache_runtime == nullptr) { + _query_cache_runtime = + std::make_shared(_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( 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)); RETURN_IF_ERROR(cur_pipe->add_operator(op, _parallel_instances)); fe_with_old_version = !tnode.__isset.is_serial_operator; break; @@ -1583,8 +1598,13 @@ Status PipelineFragmentContext::_create_operator(ObjectPool* pool, const TPlanNo auto create_query_cache_operator = [&](PipelinePtr& new_pipe) { auto cache_node_id = _params.local_params[0].per_node_scan_ranges.begin()->first; auto cache_source_id = next_operator_id(); + if (_query_cache_runtime == nullptr) { + _query_cache_runtime = + std::make_shared(_params.fragment.query_cache_param); + } op = std::make_shared(pool, cache_node_id, cache_source_id, - _params.fragment.query_cache_param); + _params.fragment.query_cache_param, + _query_cache_runtime); RETURN_IF_ERROR(cur_pipe->add_operator(op, _parallel_instances)); const auto downstream_pipeline_id = cur_pipe->id(); diff --git a/be/src/exec/pipeline/pipeline_fragment_context.h b/be/src/exec/pipeline/pipeline_fragment_context.h index fe1d2a6c065ba0..1a63426738bef8 100644 --- a/be/src/exec/pipeline/pipeline_fragment_context.h +++ b/be/src/exec/pipeline/pipeline_fragment_context.h @@ -48,6 +48,7 @@ class ExecEnv; class RuntimeFilterMergeControllerEntity; class TDataSink; class TPipelineFragmentParams; +class QueryCacheRuntime; class Dependency; struct LocalExchangeSharedState; @@ -379,6 +380,12 @@ class PipelineFragmentContext : public TaskExecutionContext { TPipelineFragmentParams _params; int32_t _parallel_instances = 0; + // Query cache context of this fragment, shared by the olap scan operator + // and the cache source operator so both consume the same per-instance + // cache decision (HIT / INCREMENTAL / MISS). Created lazily when the + // fragment carries a query_cache_param. See QueryCacheRuntime. + std::shared_ptr _query_cache_runtime; + std::atomic _need_notify_close = false; // Holds the brpc ClosureGuard for async wait-close during recursive CTE rerun. // When the PFC finishes closing and is destroyed, the shared_ptr destructor fires diff --git a/be/src/exec/scan/olap_scanner.cpp b/be/src/exec/scan/olap_scanner.cpp index a9ef277c373002..894e7253c16266 100644 --- a/be/src/exec/scan/olap_scanner.cpp +++ b/be/src/exec/scan/olap_scanner.cpp @@ -80,7 +80,7 @@ OlapScanner::OlapScanner(ScanLocalStateBase* parent, OlapScanner::Params&& param .reader_type = params.read_row_binlog ? ReaderType::READER_BINLOG : ReaderType::READER_QUERY, .aggregation = params.aggregation, - .version = {0, params.version}, + .version = {params.start_version, params.version}, .start_key {}, .end_key {}, .predicates {}, diff --git a/be/src/exec/scan/olap_scanner.h b/be/src/exec/scan/olap_scanner.h index eb8f8dee4795bf..268165704ff770 100644 --- a/be/src/exec/scan/olap_scanner.h +++ b/be/src/exec/scan/olap_scanner.h @@ -74,6 +74,10 @@ class OlapScanner : public Scanner { TBinlogScanType::type binlog_scan_type = TBinlogScanType::NONE; std::optional start_tso; std::optional end_tso; + // Non-zero only for query cache incremental merge: read the delta + // rowsets in [start_version, version] instead of the full snapshot + // [0, version]. The read_source is pre-captured accordingly. + int64_t start_version = 0; }; OlapScanner(ScanLocalStateBase* parent, Params&& params); diff --git a/be/src/runtime/query_cache/query_cache.cpp b/be/src/runtime/query_cache/query_cache.cpp index 28610d44808686..98112f3854f48b 100644 --- a/be/src/runtime/query_cache/query_cache.cpp +++ b/be/src/runtime/query_cache/query_cache.cpp @@ -17,7 +17,14 @@ #include "runtime/query_cache/query_cache.h" +#include "cloud/config.h" #include "common/logging.h" +#include "common/metrics/doris_metrics.h" +#include "storage/olap_common.h" +#include "storage/rowset/rowset.h" +#include "storage/rowset/rowset_reader.h" +#include "storage/tablet/base_tablet.h" +#include "storage/tablet/tablet_meta.h" namespace doris { @@ -39,8 +46,27 @@ int64_t QueryCacheHandle::get_cache_version() { return ((QueryCache::CacheValue*)(result_ptr))->version; } +int64_t QueryCacheHandle::get_cache_delta_count() { + DCHECK(_handle); + auto result_ptr = reinterpret_cast(_handle)->value; + return ((QueryCache::CacheValue*)(result_ptr))->delta_count; +} + +int64_t QueryCacheHandle::get_cache_total_bytes() { + DCHECK(_handle); + auto result_ptr = reinterpret_cast(_handle)->value; + return ((QueryCache::CacheValue*)(result_ptr))->total_bytes; +} + +int64_t QueryCacheHandle::get_cache_total_rows() { + DCHECK(_handle); + auto result_ptr = reinterpret_cast(_handle)->value; + return ((QueryCache::CacheValue*)(result_ptr))->total_rows; +} + void QueryCache::insert(const CacheKey& key, int64_t version, CacheResult& res, - const std::vector& slot_orders, int64_t cache_size) { + const std::vector& slot_orders, int64_t cache_size, + int64_t delta_count) { SCOPED_SWITCH_THREAD_MEM_TRACKER_LIMITER(ExecEnv::GetInstance()->query_cache_mem_tracker()); CacheResult cache_result; for (auto& block_data : res) { @@ -50,11 +76,12 @@ void QueryCache::insert(const CacheKey& key, int64_t version, CacheResult& res, auto st = mutable_block.merge(*block_data); DORIS_CHECK(st.ok()); } - auto cache_value_ptr = - std::make_unique(version, std::move(cache_result), slot_orders); + auto cache_value_ptr = std::make_unique( + version, std::move(cache_result), slot_orders, delta_count, cache_size); QueryCacheHandle(this, LRUCachePolicy::insert(key, (void*)cache_value_ptr.release(), cache_size, cache_size, CachePriority::NORMAL)); + DorisMetrics::instance()->query_cache_write_back_total->increment(1); } bool QueryCache::lookup(const CacheKey& key, int64_t version, doris::QueryCacheHandle* handle) { @@ -70,4 +97,267 @@ 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 QueryCacheInstanceDecision::take_delta_read_source( + int64_t tablet_id) { + std::lock_guard 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 QueryCacheRuntime::get_or_make_decision( + const std::vector& 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 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(); + } + return _invalid_decision; + } + + // The lock also serializes decision making of different instances of this + // fragment. That is acceptable in the common case: a decision only touches + // tablet metadata (no data IO), so it finishes in microseconds to + // milliseconds. On very wide fragments (hundreds of tablets per instance) + // combined with tablet header lock contention (e.g. compaction meta + // commits) this serialization can stretch the fragment init tail; if that + // ever shows up in profiles, move the capture out of the lock and let the + // losing racer drop its duplicate decision. + std::lock_guard lock(_lock); + auto it = _decisions.find(cache_key); + if (it != _decisions.end()) { + return it->second; + } + + auto decision = std::make_shared(); + decision->key_valid = true; + decision->cache_key = cache_key; + decision->current_version = version; + _make_decision(scan_ranges, decision.get()); + _decisions.emplace(std::move(cache_key), decision); + return decision; +} + +void QueryCacheRuntime::_make_decision(const std::vector& 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; + DorisMetrics::instance()->query_cache_stale_hit_total->increment(1); + return; + } + + // Stale entry not reusable: drop the pin and fall back to a full scan. + decision->_delta_read_sources.clear(); + decision->handle = QueryCacheHandle(); + decision->mode = QueryCacheInstanceDecision::Mode::MISS; + if (!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); + } +} + +bool QueryCacheRuntime::_try_prepare_incremental(const std::vector& 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 an EMPTY read source instead, so emptiness is + // checked below as the failure signal. + 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(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; + } + 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 && + _delta_rewrites_history(*tablet, *read_source, cached_version, decision->current_version)) { + // A load in the delta window marked rows of the cached snapshot as + // deleted through the delete bitmap (an upsert, a partial update or a + // delete sign hit a pre-existing key). Those rows are already folded + // into the cached partial result and cannot be subtracted, so only a + // full recompute is safe. The full run re-bases the entry at the new + // version, so an occasional backfill costs exactly one full scan and + // the next pure-append hour is incremental again. + decision->incremental_fallback_reason = "delta rewrites history rows"; + return false; + } + decision->_delta_read_sources[tablet_id] = std::move(read_source); + return true; +} + +bool QueryCacheRuntime::_delta_rewrites_history(BaseTablet& tablet, + const TabletReadSource& delta_source, + int64_t cached_version, int64_t current_version) { + RowsetIdUnorderedSet delta_rowsets; + for (const auto& split : delta_source.rs_splits) { + delta_rowsets.insert(split.rs_reader->rowset()->rowset_id()); + } + // The tablet delete bitmap is shared with concurrent loads, but entries + // stamped with a version <= current_version are immutable once that + // version is visible (merge-on-write publishes the bitmap update before + // the version becomes visible), so this scan is race free for the window + // we care about. Pending entries use DeleteBitmap::TEMP_VERSION_COMMON + // (= 0) and stay below the window as well. + const DeleteBitmap& delete_bitmap = tablet.tablet_meta()->delete_bitmap(); + std::shared_lock rdlock(delete_bitmap.lock); + for (const auto& [bitmap_key, bitmap] : delete_bitmap.delete_bitmap) { + const auto version = static_cast(std::get<2>(bitmap_key)); + if (version <= cached_version || version > current_version || bitmap.isEmpty()) { + continue; + } + if (!delta_rowsets.contains(std::get<0>(bitmap_key))) { + return true; + } + } + return false; +} + } // namespace doris diff --git a/be/src/runtime/query_cache/query_cache.h b/be/src/runtime/query_cache/query_cache.h index bfa90b011e34f8..f727159eac2915 100644 --- a/be/src/runtime/query_cache/query_cache.h +++ b/be/src/runtime/query_cache/query_cache.h @@ -23,9 +23,13 @@ #include #include +#include #include +#include #include #include +#include +#include #include "common/config.h" #include "common/status.h" @@ -41,6 +45,9 @@ namespace doris { +class BaseTablet; +struct TabletReadSource; + using CacheResult = std::vector; // A handle for mid-result from query lru cache. // The handle will automatically release the cache entry when it is destroyed. @@ -73,12 +80,22 @@ class QueryCacheHandle { return *this; } + bool valid() const { return _handle != nullptr; } + std::vector* get_cache_slot_orders(); CacheResult* get_cache_result(); int64_t get_cache_version(); + // How many incremental merges have been accumulated on this entry since the + // last full recompute. See QueryCacheRuntime for the compaction policy. + int64_t get_cache_delta_count(); + + int64_t get_cache_total_bytes(); + + int64_t get_cache_total_rows(); + private: LRUCachePolicy* _cache = nullptr; Cache::Handle* _handle = nullptr; @@ -95,9 +112,28 @@ class QueryCache : public LRUCachePolicy { int64_t version; CacheResult result; std::vector slot_orders; - - CacheValue(int64_t v, CacheResult&& r, const std::vector& so) - : LRUCacheValueBase(), version(v), result(std::move(r)), slot_orders(so) {} + // Number of incremental merges accumulated on this entry since the last + // full recompute. 0 means the entry was produced by a full scan. + int64_t delta_count; + // Size of this entry, used to decide upfront whether an incremental + // merge could ever be written back under the entry_max_bytes/rows + // limits (a merged entry can only be larger than the cached one). + int64_t total_bytes; + int64_t total_rows; + + CacheValue(int64_t v, CacheResult&& r, const std::vector& so, int64_t dc = 0, + int64_t bytes = 0) + : LRUCacheValueBase(), + version(v), + result(std::move(r)), + slot_orders(so), + delta_count(dc), + total_bytes(bytes) { + total_rows = 0; + for (const auto& block : result) { + total_rows += block->rows(); + } + } }; // Create global instance of this class @@ -213,7 +249,155 @@ class QueryCache : public LRUCachePolicy { bool lookup(const CacheKey& key, int64_t version, QueryCacheHandle* handle); + // Look up the entry by key regardless of its version. The caller decides + // whether the entry is an exact hit (cached version == expected version) or + // a stale entry usable for incremental merge. Returns false if the key is + // not in the cache at all. + bool lookup_any_version(const CacheKey& key, QueryCacheHandle* handle); + void insert(const CacheKey& key, int64_t version, CacheResult& result, - const std::vector& solt_orders, int64_t cache_size); + const std::vector& solt_orders, int64_t cache_size, int64_t delta_count = 0); +}; + +// The per-fragment-instance decision of how the query cache participates in the +// execution, made exactly once (see QueryCacheRuntime) and consumed by both the +// olap scan operator and the cache source operator, so the two operators can +// never disagree (e.g. scan skips scanning because the entry looked fresh while +// cache source misses because the entry got evicted in between -- which would +// silently produce an empty result and poison the cache with it). +struct QueryCacheInstanceDecision { + enum class Mode { + // Run the full scan and (if the key is valid) write the result back. + MISS, + // The cached entry matches the current version: emit cached blocks, + // skip scanning entirely, do not write back. + HIT, + // A stale entry is reusable: scan only the delta rowsets in + // (cached_version, current_version], emit the cached blocks and the + // delta partial result side by side (the upstream merge aggregation + // combines them), then write the merged entry back. + INCREMENTAL, + }; + + ~QueryCacheInstanceDecision(); + + // Take the pre-captured delta read source of one tablet. Returns nullptr if + // absent (already taken or never captured). Only meaningful in INCREMENTAL + // mode; each tablet's read source can be consumed exactly once. + std::unique_ptr take_delta_read_source(int64_t tablet_id); + + Mode mode = Mode::MISS; + // False when build_cache_key failed (e.g. tablets in this instance carry + // different versions because FE could not align instances to partitions). + // In that case the query degrades to an uncached scan: no lookup, no write + // back, but the query itself still succeeds. + bool key_valid = false; + // False when the merged entry could never satisfy entry_max_bytes/rows + // because the reused cached entry alone already exceeds them: the query + // still scans only the delta (INCREMENTAL), but skips cloning blocks for a + // write back that would be discarded anyway. + bool write_back_feasible = true; + // Why a stale entry was not reused incrementally (empty when it was, or + // when incremental merge is not enabled for this query). For the query + // profile only. + std::string incremental_fallback_reason; + std::string cache_key; + // The version this query is reading (from the scan ranges). + int64_t current_version = 0; + // Only set in INCREMENTAL mode: the version of the reused stale entry. + int64_t cached_version = 0; + // Only set in HIT/INCREMENTAL mode: delta merges accumulated on the entry. + int64_t cached_delta_count = 0; + // Pins the cache entry in HIT/INCREMENTAL mode so it cannot be evicted (and + // its blocks cannot be freed) while this query is using it. Note the pin + // lives until the fragment is torn down; when the merged entry replaces + // this one under the same key, both stay in memory for that window and the + // LRU usage accounting only sees the new one (the mem tracker still sees + // both) -- bounded by (in-flight incremental queries) x entry size. + QueryCacheHandle handle; + +private: + friend class QueryCacheRuntime; + std::mutex _take_lock; + // INCREMENTAL mode: read sources of (cached_version, current_version] + // captured at decision time, keyed by tablet id. Captured eagerly so that a + // capture failure (e.g. the delta versions were merged away by compaction) + // downgrades the decision to MISS *before* any operator acts on it; if the + // scan discovered the failure only at prepare time, the cache source might + // already have decided to emit the stale blocks. + std::unordered_map> _delta_read_sources; }; + +// Fragment-level query cache context shared by the olap scan operator and the +// cache source operator of the same fragment. Both operators obtain the cache +// decision of their instance through get_or_make_decision(); the first caller +// makes the decision and the other one observes the same object, whatever the +// operator local-state init order is. +class QueryCacheRuntime { +public: + // `cache` is injectable for tests; production callers pass nullptr and the + // global instance is used. + explicit QueryCacheRuntime(const TQueryCacheParam& param, QueryCache* cache = nullptr) + : _param(param), _cache(cache != nullptr ? cache : QueryCache::instance()) {} + + QueryCache* cache() const { return _cache; } + + // Row-binlog scans read a different data stream and must not serve or fill + // the query cache. Called while building the operator tree (single + // threaded, before any local state init), so no locking is needed. + void disable_for_binlog_scan() { _binlog_scan = true; } + + // Idempotent: the first call for a given instance (identified by the cache + // key derived from its scan ranges) makes the decision, later calls return + // the same decision object. Never returns nullptr. + std::shared_ptr get_or_make_decision( + const std::vector& scan_ranges); + +#ifdef BE_TEST + // Tests inject a hand-crafted decision (e.g. INCREMENTAL) for an instance, + // since a real storage engine is unavailable to capture delta read sources. + void inject_decision_for_test(const std::string& cache_key, + std::shared_ptr decision) { + std::lock_guard lock(_lock); + _decisions[cache_key] = std::move(decision); + } +#endif + +private: + void _make_decision(const std::vector& scan_ranges, + QueryCacheInstanceDecision* decision); + + // Try to turn a stale entry into an INCREMENTAL decision. Returns true on + // success; on any failure the caller keeps the decision as MISS (full + // recompute), which is always safe. + bool _try_prepare_incremental(const std::vector& scan_ranges, + QueryCacheInstanceDecision* decision); + + // Validate one tablet for incremental merge and capture its delta read + // source of (cached_version, current_version]. On any failure records the + // fallback reason in the decision and returns false. + bool _capture_tablet_delta(int64_t tablet_id, int64_t cached_version, + QueryCacheInstanceDecision* decision); + + // Merge-on-write only: true if any delete-bitmap entry stamped with a + // version inside (cached_version, current_version] targets a rowset + // OUTSIDE the captured delta set, i.e. the delta window rewrote rows that + // are already folded into the cached partial result (an upsert, a partial + // update or a delete sign hit a key that predates the cached version). + // Entries targeting the delta rowsets themselves are harmless: the delta + // scan reads those rowsets with the delete bitmap applied. + static bool _delta_rewrites_history(BaseTablet& tablet, const TabletReadSource& delta_source, + int64_t cached_version, int64_t current_version); + + TQueryCacheParam _param; + QueryCache* _cache = nullptr; + bool _binlog_scan = false; + + std::mutex _lock; + std::map> _decisions; + // Shared by every instance whose cache key cannot be built (see + // get_or_make_decision): one immutable MISS decision, one log line. + std::shared_ptr _invalid_decision; +}; + } // namespace doris diff --git a/be/test/exec/operator/query_cache_operator_test.cpp b/be/test/exec/operator/query_cache_operator_test.cpp index 91c73b99077247..306f624d84ba0c 100644 --- a/be/test/exec/operator/query_cache_operator_test.cpp +++ b/be/test/exec/operator/query_cache_operator_test.cpp @@ -65,11 +65,14 @@ struct QueryCacheOperatorTest : public ::testing::Test { scan_ranges.push_back(scan_range); } void TearDown() override { - // Must clear state before query_cache_uptr is destroyed, because - // local states inside `state` hold QueryCacheHandle which references - // the QueryCache. C++ destroys members in reverse declaration order, - // so state (declared before query_cache_uptr) would outlive the cache. + // Must release every QueryCacheHandle holder before query_cache_uptr is + // destroyed. C++ destroys members in reverse declaration order, so + // `state` (local states) and `source`/`sink` (whose QueryCacheRuntime + // owns the decisions that pin cache entries) would otherwise outlive + // the cache and release their handles into a destroyed QueryCache. state.reset(); + source.reset(); + sink.reset(); } void create_local_state() { shared_state = sink->create_shared_state(); @@ -134,10 +137,6 @@ struct QueryCacheOperatorTest : public ::testing::Test { TEST_F(QueryCacheOperatorTest, test_no_hit_cache1) { sink = std::make_unique(); - source = std::make_unique(); - EXPECT_TRUE(source->set_child(child_op)); - child_op->_mock_row_desc.reset( - new MockRowDescriptor {{std::make_shared()}, &pool}); TQueryCacheParam cache_param; cache_param.node_id = 0; cache_param.digest = "test_digest"; @@ -147,7 +146,14 @@ TEST_F(QueryCacheOperatorTest, test_no_hit_cache1) { cache_param.entry_max_bytes = 1024 * 1024; cache_param.entry_max_rows = 1000; - source->_cache_param = cache_param; + // Exercise the production constructor once; the other tests use the + // BE_TEST default constructor and assign the members directly. + source = std::make_unique( + &pool, /*plan_node_id=*/0, /*operator_id=*/0, cache_param, + std::make_shared(cache_param, query_cache)); + EXPECT_TRUE(source->set_child(child_op)); + child_op->_mock_row_desc.reset( + new MockRowDescriptor {{std::make_shared()}, &pool}); create_local_state(); std::cout << query_cache->get_element_count() << std::endl; @@ -189,6 +195,7 @@ TEST_F(QueryCacheOperatorTest, test_no_hit_cache2) { cache_param.entry_max_rows = 3; source->_cache_param = cache_param; + source->_query_cache_runtime = std::make_shared(cache_param, query_cache); create_local_state(); std::cout << query_cache->get_element_count() << std::endl; @@ -241,6 +248,7 @@ TEST_F(QueryCacheOperatorTest, test_hit_cache) { } source->_cache_param = cache_param; + source->_query_cache_runtime = std::make_shared(cache_param, query_cache); create_local_state(); std::cout << query_cache->get_element_count() << std::endl; @@ -276,4 +284,412 @@ TEST_F(QueryCacheOperatorTest, test_hit_cache) { } } +TEST_F(QueryCacheOperatorTest, test_stale_full_recompute) { + sink = std::make_unique(); + source = std::make_unique(); + EXPECT_TRUE(source->set_child(child_op)); + child_op->_mock_row_desc.reset( + new MockRowDescriptor {{std::make_shared()}, &pool}); + TQueryCacheParam cache_param; + cache_param.node_id = 0; + cache_param.digest = "test_digest"; + cache_param.output_slot_mapping[0] = 0; + cache_param.tablet_to_range.insert({42, "test"}); + cache_param.force_refresh_query_cache = false; + cache_param.entry_max_bytes = 1024 * 1024; + cache_param.entry_max_rows = 1000; + + std::string cache_key; + { + // A stale entry (version 100 < scan version 114514) without + // allow_incremental: plain miss, full recompute, write back overwrites. + int64_t version = 0; + EXPECT_TRUE( + QueryCache::build_cache_key(scan_ranges, cache_param, &cache_key, &version).ok()); + CacheResult result; + result.push_back(std::make_unique()); + *result.back() = ColumnHelper::create_block({100, 200}); + query_cache->insert(cache_key, 100, result, {0}, 1); + } + + source->_cache_param = cache_param; + source->_query_cache_runtime = std::make_shared(cache_param, query_cache); + create_local_state(); + + EXPECT_TRUE(source_local_state->_need_insert_cache); + EXPECT_FALSE(source_local_state->_is_incremental); + + { + auto block = ColumnHelper::create_block({1, 2, 3, 4, 5}); + auto st = sink->sink(state.get(), &block, true); + EXPECT_TRUE(st.ok()) << st.msg(); + } + { + Block block; + bool eos = false; + auto st = source->get_block(state.get(), &block, &eos); + EXPECT_TRUE(st.ok()) << st.msg(); + EXPECT_TRUE(eos); + EXPECT_EQ(block.rows(), 5); + EXPECT_TRUE(ColumnHelper::block_equal( + block, ColumnHelper::create_block({1, 2, 3, 4, 5}))); + } + + // The stale entry is overwritten by the recomputed one. + doris::QueryCacheHandle handle; + EXPECT_TRUE(query_cache->lookup(cache_key, 114514, &handle)); + EXPECT_EQ(handle.get_cache_delta_count(), 0); +} + +TEST_F(QueryCacheOperatorTest, test_incremental_merge) { + sink = std::make_unique(); + source = std::make_unique(); + EXPECT_TRUE(source->set_child(child_op)); + child_op->_mock_row_desc.reset( + new MockRowDescriptor {{std::make_shared()}, &pool}); + TQueryCacheParam cache_param; + cache_param.node_id = 0; + cache_param.digest = "test_digest"; + cache_param.output_slot_mapping[0] = 0; + cache_param.tablet_to_range.insert({42, "test"}); + cache_param.force_refresh_query_cache = false; + cache_param.entry_max_bytes = 1024 * 1024; + cache_param.entry_max_rows = 1000; + cache_param.__set_allow_incremental(true); + + std::string cache_key; + int64_t version = 0; + EXPECT_TRUE(QueryCache::build_cache_key(scan_ranges, cache_param, &cache_key, &version).ok()); + { + // The cached partial blocks of version 100, already merged once before. + CacheResult result; + result.push_back(std::make_unique()); + *result.back() = ColumnHelper::create_block({1, 2, 3, 4, 5}); + query_cache->insert(cache_key, 100, result, {0}, 1, /*delta_count=*/1); + } + + source->_cache_param = cache_param; + source->_query_cache_runtime = std::make_shared(cache_param, query_cache); + // Hand-craft the INCREMENTAL decision (capturing a real delta read source + // needs a storage engine, which unit tests do not have). The scan side + // would scan only (100, 114514] and feed the delta through the sink. + { + auto decision = std::make_shared(); + decision->mode = QueryCacheInstanceDecision::Mode::INCREMENTAL; + decision->key_valid = true; + decision->cache_key = cache_key; + decision->current_version = version; + decision->cached_version = 100; + decision->cached_delta_count = 1; + EXPECT_TRUE(query_cache->lookup_any_version(cache_key, &decision->handle)); + source->_query_cache_runtime->inject_decision_for_test(cache_key, decision); + } + create_local_state(); + + EXPECT_TRUE(source_local_state->_need_insert_cache); + EXPECT_TRUE(source_local_state->_is_incremental); + const std::string* stale = + source_local_state->custom_profile()->get_info_string("HitCacheStale"); + ASSERT_NE(stale, nullptr); + EXPECT_EQ(*stale, "1"); + const std::string* delta_versions = + source_local_state->custom_profile()->get_info_string("IncrementalDeltaVersions"); + ASSERT_NE(delta_versions, nullptr); + EXPECT_EQ(*delta_versions, "(100, 114514]"); + + { + // The delta partial result produced by scanning only (100, 114514]. + auto block = ColumnHelper::create_block({6, 7}); + auto st = sink->sink(state.get(), &block, true); + EXPECT_TRUE(st.ok()) << st.msg(); + } + { + // First the cached blocks are emitted ... + Block block; + bool eos = false; + auto st = source->get_block(state.get(), &block, &eos); + EXPECT_TRUE(st.ok()) << st.msg(); + EXPECT_FALSE(eos); + EXPECT_EQ(block.rows(), 5); + EXPECT_TRUE(ColumnHelper::block_equal( + block, ColumnHelper::create_block({1, 2, 3, 4, 5}))); + } + { + // ... then the delta from the data queue; both are partial aggregation + // states merged by the upstream aggregation. + Block block; + bool eos = false; + auto st = source->get_block(state.get(), &block, &eos); + EXPECT_TRUE(st.ok()) << st.msg(); + EXPECT_TRUE(eos); + EXPECT_EQ(block.rows(), 2); + EXPECT_TRUE(ColumnHelper::block_equal(block, + ColumnHelper::create_block({6, 7}))); + } + + // The merged entry is written back under the new version with an increased + // delta count, holding both the cached and the delta blocks. + doris::QueryCacheHandle handle; + EXPECT_TRUE(query_cache->lookup(cache_key, 114514, &handle)); + EXPECT_EQ(handle.get_cache_delta_count(), 2); + int64_t total_rows = 0; + for (const auto& cached_block : *handle.get_cache_result()) { + total_rows += cached_block->rows(); + } + EXPECT_EQ(total_rows, 7); +} + +TEST_F(QueryCacheOperatorTest, test_incremental_over_entry_limit) { + sink = std::make_unique(); + source = std::make_unique(); + EXPECT_TRUE(source->set_child(child_op)); + child_op->_mock_row_desc.reset( + new MockRowDescriptor {{std::make_shared()}, &pool}); + TQueryCacheParam cache_param; + cache_param.node_id = 0; + cache_param.digest = "test_digest"; + cache_param.output_slot_mapping[0] = 0; + cache_param.tablet_to_range.insert({42, "test"}); + cache_param.force_refresh_query_cache = false; + cache_param.entry_max_bytes = 1024 * 1024; + // The cached part alone (5 rows) already exceeds the limit, so the merged + // entry must not be written back; the stale entry stays untouched. + cache_param.entry_max_rows = 3; + cache_param.__set_allow_incremental(true); + + std::string cache_key; + int64_t version = 0; + EXPECT_TRUE(QueryCache::build_cache_key(scan_ranges, cache_param, &cache_key, &version).ok()); + { + CacheResult result; + result.push_back(std::make_unique()); + *result.back() = ColumnHelper::create_block({1, 2, 3, 4, 5}); + query_cache->insert(cache_key, 100, result, {0}, 1, /*delta_count=*/0); + } + + source->_cache_param = cache_param; + source->_query_cache_runtime = std::make_shared(cache_param, query_cache); + { + auto decision = std::make_shared(); + decision->mode = QueryCacheInstanceDecision::Mode::INCREMENTAL; + decision->key_valid = true; + decision->cache_key = cache_key; + decision->current_version = version; + decision->cached_version = 100; + decision->cached_delta_count = 0; + EXPECT_TRUE(query_cache->lookup_any_version(cache_key, &decision->handle)); + source->_query_cache_runtime->inject_decision_for_test(cache_key, decision); + } + create_local_state(); + + EXPECT_TRUE(source_local_state->_need_insert_cache); + + { + auto block = ColumnHelper::create_block({6, 7}); + auto st = sink->sink(state.get(), &block, true); + EXPECT_TRUE(st.ok()) << st.msg(); + } + { + // Emitting the cached blocks overruns entry_max_rows: caching stops but + // the data still flows. + Block block; + bool eos = false; + auto st = source->get_block(state.get(), &block, &eos); + EXPECT_TRUE(st.ok()) << st.msg(); + EXPECT_FALSE(eos); + EXPECT_EQ(block.rows(), 5); + EXPECT_FALSE(source_local_state->_need_insert_cache); + } + { + // The delta passes through unchanged. + Block block; + bool eos = false; + auto st = source->get_block(state.get(), &block, &eos); + EXPECT_TRUE(st.ok()) << st.msg(); + EXPECT_TRUE(eos); + EXPECT_EQ(block.rows(), 2); + EXPECT_TRUE(ColumnHelper::block_equal(block, + ColumnHelper::create_block({6, 7}))); + } + + // No write back happened: the stale entry still carries version 100. + doris::QueryCacheHandle handle; + EXPECT_FALSE(query_cache->lookup(cache_key, 114514, &handle)); + doris::QueryCacheHandle stale_handle; + EXPECT_TRUE(query_cache->lookup_any_version(cache_key, &stale_handle)); + EXPECT_EQ(stale_handle.get_cache_version(), 100); +} + +TEST_F(QueryCacheOperatorTest, test_incremental_write_back_infeasible) { + sink = std::make_unique(); + source = std::make_unique(); + EXPECT_TRUE(source->set_child(child_op)); + child_op->_mock_row_desc.reset( + new MockRowDescriptor {{std::make_shared()}, &pool}); + TQueryCacheParam cache_param; + cache_param.node_id = 0; + cache_param.digest = "test_digest"; + cache_param.output_slot_mapping[0] = 0; + cache_param.tablet_to_range.insert({42, "test"}); + cache_param.force_refresh_query_cache = false; + cache_param.entry_max_bytes = 1024 * 1024; + cache_param.entry_max_rows = 1000; + cache_param.__set_allow_incremental(true); + + std::string cache_key; + int64_t version = 0; + EXPECT_TRUE(QueryCache::build_cache_key(scan_ranges, cache_param, &cache_key, &version).ok()); + { + CacheResult result; + result.push_back(std::make_unique()); + *result.back() = ColumnHelper::create_block({1, 2, 3, 4, 5}); + query_cache->insert(cache_key, 100, result, {0}, 1); + } + + source->_cache_param = cache_param; + source->_query_cache_runtime = std::make_shared(cache_param, query_cache); + // The decision already knows the merged entry could never fit the limits: + // scan only the delta, but never clone blocks for a doomed write back. + { + auto decision = std::make_shared(); + decision->mode = QueryCacheInstanceDecision::Mode::INCREMENTAL; + decision->key_valid = true; + decision->write_back_feasible = false; + decision->cache_key = cache_key; + decision->current_version = version; + decision->cached_version = 100; + decision->cached_delta_count = 0; + EXPECT_TRUE(query_cache->lookup_any_version(cache_key, &decision->handle)); + source->_query_cache_runtime->inject_decision_for_test(cache_key, decision); + } + create_local_state(); + + EXPECT_FALSE(source_local_state->_need_insert_cache); + EXPECT_TRUE(source_local_state->_is_incremental); + + { + auto block = ColumnHelper::create_block({6, 7}); + auto st = sink->sink(state.get(), &block, true); + EXPECT_TRUE(st.ok()) << st.msg(); + } + { + // Cached blocks flow through without being cloned for a write back. + Block block; + bool eos = false; + auto st = source->get_block(state.get(), &block, &eos); + EXPECT_TRUE(st.ok()) << st.msg(); + EXPECT_FALSE(eos); + EXPECT_EQ(block.rows(), 5); + EXPECT_TRUE(source_local_state->_local_cache_blocks.empty()); + } + { + Block block; + bool eos = false; + auto st = source->get_block(state.get(), &block, &eos); + EXPECT_TRUE(st.ok()) << st.msg(); + EXPECT_TRUE(eos); + EXPECT_EQ(block.rows(), 2); + EXPECT_TRUE(source_local_state->_local_cache_blocks.empty()); + } + + // No write back: the stale entry still carries version 100. + doris::QueryCacheHandle handle; + EXPECT_FALSE(query_cache->lookup(cache_key, 114514, &handle)); + doris::QueryCacheHandle stale_handle; + EXPECT_TRUE(query_cache->lookup_any_version(cache_key, &stale_handle)); + EXPECT_EQ(stale_handle.get_cache_version(), 100); +} + +TEST_F(QueryCacheOperatorTest, test_incremental_fallback_reason_in_profile) { + sink = std::make_unique(); + source = std::make_unique(); + EXPECT_TRUE(source->set_child(child_op)); + child_op->_mock_row_desc.reset( + new MockRowDescriptor {{std::make_shared()}, &pool}); + // Use a tablet id that exists nowhere so fetching the tablet fails even if + // another suite left a storage engine registered in this test process. + constexpr int64_t kMissingTabletId = 424242424242; + scan_ranges.clear(); + TScanRangeParams scan_range; + TPaloScanRange palo_scan_range; + palo_scan_range.__set_tablet_id(kMissingTabletId); + palo_scan_range.__set_version("114514"); + scan_range.scan_range.__set_palo_scan_range(palo_scan_range); + scan_ranges.push_back(scan_range); + + TQueryCacheParam cache_param; + cache_param.node_id = 0; + cache_param.digest = "test_digest"; + cache_param.output_slot_mapping[0] = 0; + cache_param.tablet_to_range.insert({kMissingTabletId, "test"}); + cache_param.force_refresh_query_cache = false; + cache_param.entry_max_bytes = 1024 * 1024; + cache_param.entry_max_rows = 1000; + cache_param.__set_allow_incremental(true); + + std::string cache_key; + int64_t version = 0; + EXPECT_TRUE(QueryCache::build_cache_key(scan_ranges, cache_param, &cache_key, &version).ok()); + { + // A stale entry with incremental allowed, but the tablet does not exist + // in this test process: the decision falls back to a full recompute and + // reports why in the query profile. + CacheResult result; + result.push_back(std::make_unique()); + *result.back() = ColumnHelper::create_block({100, 200}); + query_cache->insert(cache_key, 100, result, {0}, 1); + } + + source->_cache_param = cache_param; + source->_query_cache_runtime = std::make_shared(cache_param, query_cache); + create_local_state(); + + EXPECT_FALSE(source_local_state->_is_incremental); + EXPECT_TRUE(source_local_state->_need_insert_cache); + const std::string* reason = + source_local_state->custom_profile()->get_info_string("IncrementalFallbackReason"); + ASSERT_NE(reason, nullptr); + EXPECT_EQ(*reason, "tablet not found"); +} + +TEST_F(QueryCacheOperatorTest, test_missing_runtime_degrades_to_uncached) { + sink = std::make_unique(); + source = std::make_unique(); + EXPECT_TRUE(source->set_child(child_op)); + child_op->_mock_row_desc.reset( + new MockRowDescriptor {{std::make_shared()}, &pool}); + TQueryCacheParam cache_param; + cache_param.node_id = 0; + cache_param.digest = "test_digest"; + cache_param.output_slot_mapping[0] = 0; + cache_param.tablet_to_range.insert({42, "test"}); + cache_param.force_refresh_query_cache = false; + cache_param.entry_max_bytes = 1024 * 1024; + cache_param.entry_max_rows = 1000; + + // No runtime injected: init degrades to an uncached scan, data passes + // through and nothing is written back. + source->_cache_param = cache_param; + create_local_state(); + + EXPECT_FALSE(source_local_state->_need_insert_cache); + + { + auto block = ColumnHelper::create_block({1, 2, 3}); + auto st = sink->sink(state.get(), &block, true); + EXPECT_TRUE(st.ok()) << st.msg(); + } + { + Block block; + bool eos = false; + auto st = source->get_block(state.get(), &block, &eos); + EXPECT_TRUE(st.ok()) << st.msg(); + EXPECT_TRUE(eos); + EXPECT_EQ(block.rows(), 3); + EXPECT_TRUE(ColumnHelper::block_equal( + block, ColumnHelper::create_block({1, 2, 3}))); + } + EXPECT_EQ(query_cache->get_element_count(), 0); +} + } // namespace doris diff --git a/be/test/exec/pipeline/query_cache_test.cpp b/be/test/exec/pipeline/query_cache_test.cpp index ec0f8c6c697686..c6c0e08e81aad7 100644 --- a/be/test/exec/pipeline/query_cache_test.cpp +++ b/be/test/exec/pipeline/query_cache_test.cpp @@ -17,13 +17,33 @@ #include "runtime/query_cache/query_cache.h" +#include +#include +#include #include +#include #include +#include #include +#include "cloud/config.h" +#include "common/config.h" +#include "common/metrics/doris_metrics.h" #include "core/data_type/data_type_number.h" +#include "io/fs/local_file_system.h" +#include "json2pb/json_to_pb.h" +#include "storage/data_dir.h" +#include "storage/options.h" +#include "storage/rowset/rowset_meta.h" +#include "storage/storage_engine.h" +#include "storage/tablet/base_tablet.h" +#include "storage/tablet/tablet.h" +#include "storage/tablet/tablet_manager.h" +#include "storage/tablet/tablet_meta.h" #include "testutil/column_helper.h" +#include "util/debug_points.h" +#include "util/uid_util.h" namespace doris { class QueryCacheTest : public testing::Test { @@ -287,4 +307,542 @@ TEST_F(QueryCacheTest, insert_and_lookup) { // ./run-be-ut.sh --run --filter=DataQueueTest.* +namespace { + +std::vector make_scan_ranges(int64_t tablet_id, const std::string& version) { + std::vector scan_ranges; + TScanRangeParams scan_range; + TPaloScanRange palo_scan_range; + palo_scan_range.__set_tablet_id(tablet_id); + palo_scan_range.__set_version(version); + scan_range.scan_range.__set_palo_scan_range(palo_scan_range); + scan_ranges.push_back(scan_range); + return scan_ranges; +} + +TQueryCacheParam make_cache_param(int64_t tablet_id) { + TQueryCacheParam cache_param; + cache_param.__set_digest("runtime_test_digest"); + cache_param.tablet_to_range.insert({tablet_id, "range"}); + // FE always sets the entry limits; keep them roomy so decision tests do + // not trip the write-back feasibility check unintentionally. + cache_param.__set_entry_max_bytes(1024 * 1024); + cache_param.__set_entry_max_rows(100000); + return cache_param; +} + +void insert_entry(QueryCache* cache, const std::string& cache_key, int64_t version, + int64_t delta_count) { + CacheResult result; + result.push_back(std::make_unique()); + *result.back() = ColumnHelper::create_block({1, 2, 3}); + cache->insert(cache_key, version, result, {0}, 1, delta_count); +} + +} // namespace + +TEST_F(QueryCacheTest, runtime_decision_miss_and_idempotent) { + std::unique_ptr cache(QueryCache::create_global_cache(1024 * 1024)); + auto scan_ranges = make_scan_ranges(42, "100"); + QueryCacheRuntime runtime(make_cache_param(42), cache.get()); + + auto decision = runtime.get_or_make_decision(scan_ranges); + EXPECT_EQ(decision->mode, QueryCacheInstanceDecision::Mode::MISS); + EXPECT_TRUE(decision->key_valid); + EXPECT_EQ(decision->current_version, 100); + EXPECT_FALSE(decision->handle.valid()); + + // Idempotent: the second caller (e.g. the other operator) observes the + // same decision object. + auto decision2 = runtime.get_or_make_decision(scan_ranges); + EXPECT_EQ(decision.get(), decision2.get()); +} + +TEST_F(QueryCacheTest, runtime_decision_invalid_key) { + std::unique_ptr cache(QueryCache::create_global_cache(1024 * 1024)); + auto scan_ranges = make_scan_ranges(42, "100"); + // Tablet 42 is missing from tablet_to_range, so build_cache_key fails and + // the query degrades to an uncached scan instead of failing. + TQueryCacheParam cache_param; + cache_param.__set_digest("runtime_test_digest"); + cache_param.tablet_to_range.insert({43, "range"}); + QueryCacheRuntime runtime(cache_param, cache.get()); + + auto decision = runtime.get_or_make_decision(scan_ranges); + EXPECT_EQ(decision->mode, QueryCacheInstanceDecision::Mode::MISS); + EXPECT_FALSE(decision->key_valid); + // Every caller shares one immutable invalid decision (and one log line). + EXPECT_EQ(decision.get(), runtime.get_or_make_decision(scan_ranges).get()); +} + +TEST_F(QueryCacheTest, runtime_decision_hit) { + std::unique_ptr cache(QueryCache::create_global_cache(1024 * 1024)); + auto scan_ranges = make_scan_ranges(42, "100"); + auto cache_param = make_cache_param(42); + + std::string cache_key; + int64_t version = 0; + EXPECT_TRUE(QueryCache::build_cache_key(scan_ranges, cache_param, &cache_key, &version).ok()); + insert_entry(cache.get(), cache_key, 100, 0); + + QueryCacheRuntime runtime(cache_param, cache.get()); + auto decision = runtime.get_or_make_decision(scan_ranges); + EXPECT_EQ(decision->mode, QueryCacheInstanceDecision::Mode::HIT); + EXPECT_TRUE(decision->handle.valid()); + EXPECT_EQ(decision->handle.get_cache_version(), 100); +} + +TEST_F(QueryCacheTest, runtime_decision_force_refresh) { + std::unique_ptr cache(QueryCache::create_global_cache(1024 * 1024)); + auto scan_ranges = make_scan_ranges(42, "100"); + auto cache_param = make_cache_param(42); + + std::string cache_key; + int64_t version = 0; + EXPECT_TRUE(QueryCache::build_cache_key(scan_ranges, cache_param, &cache_key, &version).ok()); + insert_entry(cache.get(), cache_key, 100, 0); + + cache_param.__set_force_refresh_query_cache(true); + QueryCacheRuntime runtime(cache_param, cache.get()); + auto decision = runtime.get_or_make_decision(scan_ranges); + // Recompute and write back even though a fresh entry exists. + EXPECT_EQ(decision->mode, QueryCacheInstanceDecision::Mode::MISS); + EXPECT_TRUE(decision->key_valid); + EXPECT_FALSE(decision->handle.valid()); +} + +TEST_F(QueryCacheTest, runtime_decision_binlog_scan) { + std::unique_ptr cache(QueryCache::create_global_cache(1024 * 1024)); + auto scan_ranges = make_scan_ranges(42, "100"); + auto cache_param = make_cache_param(42); + + std::string cache_key; + int64_t version = 0; + EXPECT_TRUE(QueryCache::build_cache_key(scan_ranges, cache_param, &cache_key, &version).ok()); + insert_entry(cache.get(), cache_key, 100, 0); + + QueryCacheRuntime runtime(cache_param, cache.get()); + runtime.disable_for_binlog_scan(); + auto decision = runtime.get_or_make_decision(scan_ranges); + // A binlog scan must neither serve the cached entry nor write back. + EXPECT_EQ(decision->mode, QueryCacheInstanceDecision::Mode::MISS); + EXPECT_FALSE(decision->key_valid); +} + +TEST_F(QueryCacheTest, runtime_decision_stale_without_incremental) { + std::unique_ptr cache(QueryCache::create_global_cache(1024 * 1024)); + auto scan_ranges = make_scan_ranges(42, "100"); + auto cache_param = make_cache_param(42); + + std::string cache_key; + int64_t version = 0; + EXPECT_TRUE(QueryCache::build_cache_key(scan_ranges, cache_param, &cache_key, &version).ok()); + insert_entry(cache.get(), cache_key, 50, 0); + + // allow_incremental unset: a stale entry is a plain miss (full recompute). + QueryCacheRuntime runtime(cache_param, cache.get()); + auto decision = runtime.get_or_make_decision(scan_ranges); + EXPECT_EQ(decision->mode, QueryCacheInstanceDecision::Mode::MISS); + EXPECT_TRUE(decision->key_valid); + EXPECT_FALSE(decision->handle.valid()); +} + +TEST_F(QueryCacheTest, runtime_decision_stale_incremental_fallbacks) { + std::unique_ptr cache(QueryCache::create_global_cache(1024 * 1024)); + auto scan_ranges = make_scan_ranges(42, "100"); + auto cache_param = make_cache_param(42); + cache_param.__set_allow_incremental(true); + + std::string cache_key; + int64_t version = 0; + EXPECT_TRUE(QueryCache::build_cache_key(scan_ranges, cache_param, &cache_key, &version).ok()); + + { + // The entry is newer than the version this replica is asked to read: + // never usable, fall back to a full scan of the requested version. + insert_entry(cache.get(), cache_key, 200, 0); + QueryCacheRuntime runtime(cache_param, cache.get()); + auto decision = runtime.get_or_make_decision(scan_ranges); + EXPECT_EQ(decision->mode, QueryCacheInstanceDecision::Mode::MISS); + EXPECT_TRUE(decision->key_valid); + EXPECT_EQ(decision->incremental_fallback_reason, "cached entry is newer"); + } + { + // Too many accumulated incremental merges: force a full recompute to + // compact the entry. (Checked before any tablet access.) + insert_entry(cache.get(), cache_key, 50, config::query_cache_max_incremental_merge_count); + QueryCacheRuntime runtime(cache_param, cache.get()); + auto decision = runtime.get_or_make_decision(scan_ranges); + EXPECT_EQ(decision->mode, QueryCacheInstanceDecision::Mode::MISS); + EXPECT_TRUE(decision->key_valid); + EXPECT_EQ(decision->incremental_fallback_reason, + "delta count reached compaction threshold"); + } + { + // Use a tablet id that exists nowhere, so fetching the tablet fails + // regardless of whether some other suite left a storage engine behind + // in this test process, and the decision safely falls back to MISS. + auto missing_scan_ranges = make_scan_ranges(424242424242, "100"); + auto missing_cache_param = make_cache_param(424242424242); + missing_cache_param.__set_allow_incremental(true); + std::string missing_cache_key; + int64_t missing_version = 0; + EXPECT_TRUE(QueryCache::build_cache_key(missing_scan_ranges, missing_cache_param, + &missing_cache_key, &missing_version) + .ok()); + insert_entry(cache.get(), missing_cache_key, 50, 0); + QueryCacheRuntime runtime(missing_cache_param, cache.get()); + auto decision = runtime.get_or_make_decision(missing_scan_ranges); + EXPECT_EQ(decision->mode, QueryCacheInstanceDecision::Mode::MISS); + EXPECT_TRUE(decision->key_valid); + EXPECT_FALSE(decision->handle.valid()); + EXPECT_EQ(decision->incremental_fallback_reason, "tablet not found"); + } +} + +TEST_F(QueryCacheTest, lookup_any_version_and_delta_count) { + std::unique_ptr cache(QueryCache::create_global_cache(1024 * 1024)); + auto scan_ranges = make_scan_ranges(42, "100"); + auto cache_param = make_cache_param(42); + + std::string cache_key; + int64_t version = 0; + EXPECT_TRUE(QueryCache::build_cache_key(scan_ranges, cache_param, &cache_key, &version).ok()); + + QueryCacheHandle handle; + EXPECT_FALSE(cache->lookup_any_version(cache_key, &handle)); + + int64_t write_backs_before = DorisMetrics::instance()->query_cache_write_back_total->value(); + insert_entry(cache.get(), cache_key, 50, 3); + EXPECT_EQ(DorisMetrics::instance()->query_cache_write_back_total->value(), + write_backs_before + 1); + QueryCacheHandle handle2; + EXPECT_TRUE(cache->lookup_any_version(cache_key, &handle2)); + EXPECT_EQ(handle2.get_cache_version(), 50); + EXPECT_EQ(handle2.get_cache_delta_count(), 3); + // insert_entry stores one 3-row block with cache_size 1. + EXPECT_EQ(handle2.get_cache_total_rows(), 3); + EXPECT_EQ(handle2.get_cache_total_bytes(), 1); + + // Exact-version lookup still rejects the stale entry. + QueryCacheHandle handle3; + EXPECT_FALSE(cache->lookup(cache_key, 100, &handle3)); +} + +TEST_F(QueryCacheTest, runtime_default_global_cache) { + // The single-argument constructor falls back to the global instance + // (whatever it is in this test environment). + QueryCacheRuntime runtime(make_cache_param(42)); + EXPECT_EQ(runtime.cache(), QueryCache::instance()); +} + +TEST_F(QueryCacheTest, take_delta_read_source) { + auto decision = std::make_shared(); + decision->_delta_read_sources[42] = std::make_unique(); + + EXPECT_EQ(decision->take_delta_read_source(41), nullptr); + auto source = decision->take_delta_read_source(42); + EXPECT_NE(source, nullptr); + // Each read source can be consumed exactly once. + EXPECT_EQ(decision->take_delta_read_source(42), nullptr); +} + +TEST_F(QueryCacheTest, runtime_decision_stale_incremental_cloud_mode) { + std::unique_ptr cache(QueryCache::create_global_cache(1024 * 1024)); + auto scan_ranges = make_scan_ranges(42, "100"); + auto cache_param = make_cache_param(42); + cache_param.__set_allow_incremental(true); + + std::string cache_key; + int64_t version = 0; + EXPECT_TRUE(QueryCache::build_cache_key(scan_ranges, cache_param, &cache_key, &version).ok()); + insert_entry(cache.get(), cache_key, 50, 0); + + // Incremental merge only supports local storage for now. + std::string saved_deploy_mode = config::deploy_mode; + config::deploy_mode = "cloud"; + QueryCacheRuntime runtime(cache_param, cache.get()); + auto decision = runtime.get_or_make_decision(scan_ranges); + config::deploy_mode = saved_deploy_mode; + + EXPECT_EQ(decision->mode, QueryCacheInstanceDecision::Mode::MISS); + EXPECT_TRUE(decision->key_valid); + EXPECT_EQ(decision->incremental_fallback_reason, "cloud mode"); +} + +// Exercises the per-tablet part of the incremental decision against a real +// (metadata-only) tablet registered in a real storage engine: capturing the +// delta read source never touches segment files, so no data is needed. +class QueryCacheIncrementalTest : public testing::Test { +protected: + static constexpr int64_t kTabletId = 15673; + static constexpr const char* kTestDir = "/ut_dir/query_cache_incremental_test"; + + void SetUp() override { + char buffer[1024]; + EXPECT_NE(getcwd(buffer, sizeof(buffer)), nullptr); + _absolute_dir = std::string(buffer) + kTestDir; + EXPECT_TRUE(io::global_local_filesystem()->delete_directory(_absolute_dir).ok()); + EXPECT_TRUE(io::global_local_filesystem()->create_directory(_absolute_dir).ok()); + + auto engine = std::make_unique(EngineOptions {}); + _engine = engine.get(); + ExecEnv::GetInstance()->set_storage_engine(std::move(engine)); + _data_dir = std::make_unique(*_engine, _absolute_dir); + static_cast(_data_dir->init()); + _cache.reset(QueryCache::create_global_cache(1024 * 1024)); + } + + void TearDown() override { + _cache.reset(); + _data_dir.reset(); + ExecEnv::GetInstance()->set_storage_engine(nullptr); + _engine = nullptr; + EXPECT_TRUE(io::global_local_filesystem()->delete_directory(_absolute_dir).ok()); + } + + void init_rs_meta(RowsetMetaSharedPtr& rs_meta, const TabletMetaSharedPtr& tablet_meta, + int64_t start, int64_t end, bool with_delete_predicate) { + static const std::string json_rowset_meta = R"({ + "rowset_id": 540081, + "tablet_id": 15673, + "txn_id": 4042, + "tablet_schema_hash": 567997577, + "rowset_type": "BETA_ROWSET", + "rowset_state": "VISIBLE", + "start_version": 2, + "end_version": 2, + "num_rows": 3929, + "total_disk_size": 84699, + "data_disk_size": 84464, + "index_disk_size": 235, + "empty": false, + "load_id": { + "hi": -5350970832824939812, + "lo": -6717994719194512122 + }, + "creation_time": 1553765670 + })"; + RowsetMetaPB rowset_meta_pb; + json2pb::JsonToProtoMessage(json_rowset_meta, &rowset_meta_pb); + rowset_meta_pb.set_start_version(start); + rowset_meta_pb.set_end_version(end); + // One distinct rowset id per rowset (the json template repeats one id): + // the merge-on-write history-rewrite check tells baseline and delta + // rowsets apart by rowset id, see rowset_id_for(). + rowset_meta_pb.set_rowset_id(540081 + start); + rowset_meta_pb.set_creation_time(10000); + if (with_delete_predicate) { + DeletePredicatePB* delete_predicate = rowset_meta_pb.mutable_delete_predicate(); + delete_predicate->set_version(static_cast(start)); + delete_predicate->add_sub_predicates("k1='1'"); + } + rs_meta->init_from_pb(rowset_meta_pb); + rs_meta->set_tablet_schema(tablet_meta->tablet_schema()); + } + + TabletSharedPtr create_tablet(TKeysType::type keys_type, bool enable_merge_on_write, + const std::vector>& versions, + int64_t delete_predicate_start_version = -1) { + TTabletSchema schema; + schema.keys_type = keys_type; + TabletMetaSharedPtr tablet_meta(new TabletMeta( + 1, 2, kTabletId, 15674, 4, 5, schema, 6, {{7, 8}}, UniqueId(9, 10), + TTabletType::TABLET_TYPE_DISK, TCompressionType::LZ4F, 0, enable_merge_on_write)); + for (auto [start, end] : versions) { + RowsetMetaSharedPtr rs_meta(new RowsetMeta()); + init_rs_meta(rs_meta, tablet_meta, start, end, start == delete_predicate_start_version); + static_cast(tablet_meta->add_rs_meta(rs_meta)); + } + auto tablet = std::make_shared(*_engine, std::move(tablet_meta), _data_dir.get()); + static_cast(tablet->init()); + auto& tablet_map = _engine->tablet_manager()->_get_tablet_map(kTabletId); + tablet_map[kTabletId] = tablet; + return tablet; + } + + // The rowset id a fixture rowset starting at `start_version` ends up with, + // matching init_rs_meta() above (numeric ids use the v1 format). + static RowsetId rowset_id_for(int64_t start_version) { + RowsetId id; + id.init(540081 + start_version); + return id; + } + + // Common scenario: a stale entry of version 50 while the query reads + // version 100 with allow_incremental set. + std::shared_ptr make_stale_decision() { + auto scan_ranges = make_scan_ranges(kTabletId, "100"); + auto cache_param = make_cache_param(kTabletId); + cache_param.__set_allow_incremental(true); + std::string cache_key; + int64_t version = 0; + EXPECT_TRUE( + QueryCache::build_cache_key(scan_ranges, cache_param, &cache_key, &version).ok()); + insert_entry(_cache.get(), cache_key, 50, 0); + QueryCacheRuntime runtime(cache_param, _cache.get()); + return runtime.get_or_make_decision(scan_ranges); + } + + std::string _absolute_dir; + StorageEngine* _engine = nullptr; + std::unique_ptr _data_dir; + std::unique_ptr _cache; +}; + +TEST_F(QueryCacheIncrementalTest, incremental_success) { + create_tablet(TKeysType::DUP_KEYS, false, {{0, 50}, {51, 80}, {81, 100}}); + int64_t stale_hits_before = DorisMetrics::instance()->query_cache_stale_hit_total->value(); + auto decision = make_stale_decision(); + EXPECT_EQ(DorisMetrics::instance()->query_cache_stale_hit_total->value(), + stale_hits_before + 1); + + EXPECT_EQ(decision->mode, QueryCacheInstanceDecision::Mode::INCREMENTAL); + EXPECT_TRUE(decision->key_valid); + EXPECT_TRUE(decision->handle.valid()); + EXPECT_TRUE(decision->write_back_feasible); + EXPECT_TRUE(decision->incremental_fallback_reason.empty()); + EXPECT_EQ(decision->cached_version, 50); + EXPECT_EQ(decision->cached_delta_count, 0); + EXPECT_EQ(decision->current_version, 100); + + // The pre-captured read source covers exactly the delta (50, 100]. + auto source = decision->take_delta_read_source(kTabletId); + ASSERT_NE(source, nullptr); + EXPECT_EQ(source->rs_splits.size(), 2); + EXPECT_TRUE(source->delete_predicates.empty()); + // Consumable exactly once. + EXPECT_EQ(decision->take_delta_read_source(kTabletId), nullptr); +} + +TEST_F(QueryCacheIncrementalTest, incremental_write_back_infeasible) { + create_tablet(TKeysType::DUP_KEYS, false, {{0, 50}, {51, 100}}); + // The cached entry alone (3 rows) already exceeds entry_max_rows, so the + // merged entry could never be written back: still scan only the delta, but + // announce upfront that cloning blocks for a write back is pointless. + auto scan_ranges = make_scan_ranges(kTabletId, "100"); + auto cache_param = make_cache_param(kTabletId); + cache_param.__set_allow_incremental(true); + cache_param.__set_entry_max_rows(2); + std::string cache_key; + int64_t version = 0; + EXPECT_TRUE(QueryCache::build_cache_key(scan_ranges, cache_param, &cache_key, &version).ok()); + insert_entry(_cache.get(), cache_key, 50, 0); + + QueryCacheRuntime runtime(cache_param, _cache.get()); + auto decision = runtime.get_or_make_decision(scan_ranges); + EXPECT_EQ(decision->mode, QueryCacheInstanceDecision::Mode::INCREMENTAL); + EXPECT_FALSE(decision->write_back_feasible); +} + +TEST_F(QueryCacheIncrementalTest, fallback_on_agg_keys) { + create_tablet(TKeysType::AGG_KEYS, false, {{0, 50}, {51, 100}}); + int64_t fallbacks_before = + DorisMetrics::instance()->query_cache_incremental_fallback_total->value(); + auto decision = make_stale_decision(); + EXPECT_EQ(DorisMetrics::instance()->query_cache_incremental_fallback_total->value(), + fallbacks_before + 1); + // Storage-layer aggregation breaks "cached + delta == new snapshot". + EXPECT_EQ(decision->mode, QueryCacheInstanceDecision::Mode::MISS); + EXPECT_TRUE(decision->key_valid); + EXPECT_FALSE(decision->handle.valid()); + EXPECT_EQ(decision->incremental_fallback_reason, "keys type not append-only"); +} + +TEST_F(QueryCacheIncrementalTest, fallback_on_merge_on_read) { + // Merge-on-read UNIQUE resolves duplicates by merging across rowsets at + // read time, so a delta-only scan cannot stand alone: always fall back. + create_tablet(TKeysType::UNIQUE_KEYS, false, {{0, 50}, {51, 100}}); + auto decision = make_stale_decision(); + EXPECT_EQ(decision->mode, QueryCacheInstanceDecision::Mode::MISS); + EXPECT_EQ(decision->incremental_fallback_reason, "keys type not append-only"); +} + +TEST_F(QueryCacheIncrementalTest, mow_pure_append_incremental) { + // Merge-on-write UNIQUE with no delete-bitmap entry in the delta window: + // the hourly-append pattern. Incremental merge is as safe as on DUP. + create_tablet(TKeysType::UNIQUE_KEYS, true, {{0, 50}, {51, 80}, {81, 100}}); + auto decision = make_stale_decision(); + EXPECT_EQ(decision->mode, QueryCacheInstanceDecision::Mode::INCREMENTAL); + EXPECT_TRUE(decision->incremental_fallback_reason.empty()); + auto source = decision->take_delta_read_source(kTabletId); + ASSERT_NE(source, nullptr); + EXPECT_EQ(source->rs_splits.size(), 2); + // capture_read_source hands the delete bitmap to the delta scan, so rows + // replaced within the delta window itself are filtered by the reader. + EXPECT_NE(source->delete_bitmap, nullptr); +} + +TEST_F(QueryCacheIncrementalTest, mow_history_rewrite_falls_back) { + // A load inside the delta window marked a row of a baseline rowset as + // deleted (an upsert / backfill hit a pre-existing key): rows already + // folded into the cached entry cannot be subtracted, so fall back. + auto tablet = create_tablet(TKeysType::UNIQUE_KEYS, true, {{0, 50}, {51, 100}}); + tablet->tablet_meta()->delete_bitmap().add({rowset_id_for(0), 0, 60}, 7); + int64_t fallbacks_before = + DorisMetrics::instance()->query_cache_incremental_fallback_total->value(); + auto decision = make_stale_decision(); + EXPECT_EQ(DorisMetrics::instance()->query_cache_incremental_fallback_total->value(), + fallbacks_before + 1); + EXPECT_EQ(decision->mode, QueryCacheInstanceDecision::Mode::MISS); + EXPECT_FALSE(decision->handle.valid()); + EXPECT_EQ(decision->incremental_fallback_reason, "delta rewrites history rows"); +} + +TEST_F(QueryCacheIncrementalTest, mow_irrelevant_bitmap_entries_ignored) { + // Delete-bitmap entries that cannot affect the cached snapshot must not + // spoil the incremental path: entries targeting the delta rowsets (a key + // written twice within the window, a lost sequence-column race or a delete + // sign on a window-local key), entries outside the version window (older + // dedup history, a concurrent load newer than the read version, pending + // entries at TEMP_VERSION_COMMON = 0) and empty bitmaps. + auto tablet = create_tablet(TKeysType::UNIQUE_KEYS, true, {{0, 50}, {51, 100}}); + auto& delete_bitmap = tablet->tablet_meta()->delete_bitmap(); + delete_bitmap.add({rowset_id_for(51), 0, 90}, 3); // targets the delta itself + delete_bitmap.add({rowset_id_for(0), 0, 40}, 1); // version <= cached + delete_bitmap.add({rowset_id_for(0), 0, 150}, 2); // version > current + delete_bitmap.add({rowset_id_for(0), 0, 0}, 4); // pending (TEMP_VERSION_COMMON) + delete_bitmap.set({rowset_id_for(0), 0, 60}, roaring::Roaring()); // empty bitmap + auto decision = make_stale_decision(); + EXPECT_EQ(decision->mode, QueryCacheInstanceDecision::Mode::INCREMENTAL); + EXPECT_TRUE(decision->incremental_fallback_reason.empty()); +} + +TEST_F(QueryCacheIncrementalTest, fallback_on_version_gap) { + // The whole history lives in one compacted rowset [0, 100]: no version + // path can serve (50, 100] alone, so capture fails and we fall back. + create_tablet(TKeysType::DUP_KEYS, false, {{0, 100}}); + auto decision = make_stale_decision(); + EXPECT_EQ(decision->mode, QueryCacheInstanceDecision::Mode::MISS); + EXPECT_TRUE(decision->key_valid); + EXPECT_EQ(decision->incremental_fallback_reason, "delta versions not capturable"); +} + +TEST_F(QueryCacheIncrementalTest, fallback_on_capture_error) { + // Non-pathfinding capture failures surface as real errors even under the + // quiet option (a pathfinding miss instead returns an empty read source, + // see fallback_on_version_gap). The storage debug point simulates such a + // failure, e.g. a rowset object missing for a found path. + create_tablet(TKeysType::DUP_KEYS, false, {{0, 50}, {51, 100}}); + config::enable_debug_points = true; + DebugPoints::instance()->add_with_params("Tablet::capture_consistent_versions.inject_failure", + {{"tablet_id", "-2"}}); + auto decision = make_stale_decision(); + DebugPoints::instance()->remove("Tablet::capture_consistent_versions.inject_failure"); + config::enable_debug_points = false; + EXPECT_EQ(decision->mode, QueryCacheInstanceDecision::Mode::MISS); + EXPECT_TRUE(decision->key_valid); + EXPECT_EQ(decision->incremental_fallback_reason, "delta versions not capturable"); +} + +TEST_F(QueryCacheIncrementalTest, fallback_on_delete_predicate) { + // A delete predicate inside the delta logically removes rows that are + // already folded into the cached blocks; that cannot be merged away. + create_tablet(TKeysType::DUP_KEYS, false, {{0, 50}, {51, 60}, {61, 100}}, + /*delete_predicate_start_version=*/61); + auto decision = make_stale_decision(); + EXPECT_EQ(decision->mode, QueryCacheInstanceDecision::Mode::MISS); + EXPECT_EQ(decision->incremental_fallback_reason, "delta contains delete predicates"); +} + } // namespace doris diff --git a/fe/fe-core/src/main/java/org/apache/doris/planner/AggregationNode.java b/fe/fe-core/src/main/java/org/apache/doris/planner/AggregationNode.java index e5820c4e426755..786e026660af39 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/planner/AggregationNode.java +++ b/fe/fe-core/src/main/java/org/apache/doris/planner/AggregationNode.java @@ -89,6 +89,10 @@ public void unsetNeedsFinalize() { updateplanNodeName(); } + public boolean isNeedsFinalize() { + return needsFinalize; + } + // Used by new optimizer public void setUseStreamingPreagg(boolean useStreamingPreagg) { this.useStreamingPreagg = useStreamingPreagg; diff --git a/fe/fe-core/src/main/java/org/apache/doris/planner/normalize/QueryCacheNormalizer.java b/fe/fe-core/src/main/java/org/apache/doris/planner/normalize/QueryCacheNormalizer.java index b74ebd0e7c3aca..5212c5a5fd6533 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/planner/normalize/QueryCacheNormalizer.java +++ b/fe/fe-core/src/main/java/org/apache/doris/planner/normalize/QueryCacheNormalizer.java @@ -21,6 +21,7 @@ package org.apache.doris.planner.normalize; import org.apache.doris.analysis.DescriptorTable; +import org.apache.doris.catalog.KeysType; import org.apache.doris.catalog.OlapTable; import org.apache.doris.catalog.Tablet; import org.apache.doris.common.Pair; @@ -98,6 +99,7 @@ private Optional setQueryCacheParam( queryCacheParam.setForceRefreshQueryCache(sessionVariable.isQueryCacheForceRefresh()); queryCacheParam.setEntryMaxBytes(sessionVariable.getQueryCacheEntryMaxBytes()); queryCacheParam.setEntryMaxRows(sessionVariable.getQueryCacheEntryMaxRows()); + queryCacheParam.setAllowIncremental(computeAllowIncremental(cachePoint, sessionVariable)); queryCacheParam.setOutputSlotMapping( cachePoint.cacheRoot.getOutputTupleIds() @@ -145,6 +147,74 @@ private Optional doComputeCachePoint(PlanNode planRoot) { return Optional.empty(); } + /** + * Decide whether BE may serve a stale cache entry by scanning only the delta + * rowsets since the cached version and emitting them together with the cached + * partial aggregation blocks (the upstream aggregation merges both). + * + *

This is only correct when all of the following hold: + *

    + *
  • The cache point aggregation does not finalize: its output is a partial + * state that an upstream aggregation always merges, so emitting the cached + * blocks and the delta blocks side by side yields the correct result. A + * finalized output has no downstream merge, and the two emissions would + * produce duplicated group keys.
    • + *
  • The cache point aggregates the raw detail rows directly (its child is + * the olap scan node): with another aggregation in between, that inner + * aggregation would see only the delta rows during an incremental run, and + * its finalized output over the delta is not a mergeable complement of its + * output over the cached snapshot.
    • + *
  • The scanned index is append-only, guaranteeing "cached snapshot + + * delta rowsets == new snapshot": either DUP_KEYS, or merge-on-write + * UNIQUE_KEYS, for which BE additionally verifies per tablet (through the + * delete bitmap of the delta window) that no pre-existing key was rewritten + * and falls back otherwise. Merge-on-read UNIQUE resolves duplicates by + * merging across rowsets at read time, so a delta-only scan cannot stand + * alone there; AGG tables merge rows inside the storage layer likewise. + * DELETE predicates are handled on BE: a delta containing delete predicates + * falls back to a full recompute.
    • + *
+ */ + private boolean computeAllowIncremental(CachePoint cachePoint, SessionVariable sessionVariable) { + if (!sessionVariable.getEnableQueryCacheIncremental()) { + return false; + } + // The cache point is always an aggregation node (see doComputeCachePoint). + if (((AggregationNode) cachePoint.cacheRoot).isNeedsFinalize()) { + return false; + } + // The cached partial state and the delta partial state merge correctly + // only when the cache point aggregates the raw detail rows directly. + // With a nested cache point (partial agg over a finalized/deduplicating + // agg over scan), the inner agg sees only the delta rows during an + // incremental run, so its finalized output over the delta is NOT a + // mergeable complement of its output over the cached snapshot (e.g. + // "group by cnt" buckets computed from partial counts are simply wrong). + if (!(cachePoint.cacheRoot.getChild(0) instanceof OlapScanNode)) { + return false; + } + OlapScanNode scanNode = (OlapScanNode) cachePoint.cacheRoot.getChild(0); + OlapTable olapTable = scanNode.getOlapTable(); + long selectIndexId = scanNode.getSelectedIndexId() == -1 + ? olapTable.getBaseIndexId() + : scanNode.getSelectedIndexId(); + // Note: judged on the selected index, not the base table: a DUP table + // may serve the query from an aggregated materialized view, whose data + // is no longer append-only. + KeysType keysType = olapTable.getKeysTypeByIndexId(selectIndexId); + if (keysType == KeysType.DUP_KEYS) { + return true; + } + // A merge-on-write UNIQUE index is append-only as long as a load does + // not touch pre-existing keys, which covers the common "hourly append + // plus occasional backfill" pattern: BE verifies per tablet through + // the delete bitmap of the delta window and falls back to a full + // recompute for the rare load that rewrites history. A merge-on-read + // UNIQUE index resolves duplicates by merging across rowsets at read + // time, so a delta-only scan cannot stand alone there. + return keysType == KeysType.UNIQUE_KEYS && olapTable.getEnableUniqueKeyMergeOnWrite(); + } + private List normalizePlanTree(ConnectContext context, CachePoint cachePoint) { List normalizedPlans = new ArrayList<>(); doNormalizePlanTree(context, cachePoint.digestRoot, normalizedPlans); diff --git a/fe/fe-core/src/main/java/org/apache/doris/qe/SessionVariable.java b/fe/fe-core/src/main/java/org/apache/doris/qe/SessionVariable.java index edae724ae165f0..51d4b758d4ed20 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/qe/SessionVariable.java +++ b/fe/fe-core/src/main/java/org/apache/doris/qe/SessionVariable.java @@ -202,6 +202,7 @@ public class SessionVariable implements Serializable, Writable { public static final String ENABLE_SQL_CACHE = "enable_sql_cache"; public static final String ENABLE_HIVE_SQL_CACHE = "enable_hive_sql_cache"; public static final String ENABLE_QUERY_CACHE = "enable_query_cache"; + public static final String ENABLE_QUERY_CACHE_INCREMENTAL = "enable_query_cache_incremental"; public static final String QUERY_CACHE_FORCE_REFRESH = "query_cache_force_refresh"; public static final String QUERY_CACHE_ENTRY_MAX_BYTES = "query_cache_entry_max_bytes"; public static final String QUERY_CACHE_ENTRY_MAX_ROWS = "query_cache_entry_max_rows"; @@ -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, + description = {"是否允许 BE 以增量合并的方式复用过期的 Query Cache 条目：只扫描缓存版本之后的" + + "增量 rowset，并与缓存的聚合中间结果一起交给上游合并。仅对聚合直压扫描且不做 finalize " + + "的缓存点生效，且选中索引须为追加写：DUP_KEYS 表，或增量窗口内未改写既有主键的写时合并" + + "（merge-on-write）UNIQUE_KEYS 表（BE 按 tablet 检查 delete bitmap）；增量不可捕获（如" + + "已被 compaction 合并）、含 DELETE 谓词或改写了历史行时自动回退全量重算。需与 " + + "enable_query_cache 同时开启。", + "Whether BE may reuse a stale query cache entry by incremental merge: scan only" + + " the delta rowsets since the cached version and emit them together with the" + + " cached partial aggregation blocks for the upstream merge. Only takes effect" + + " when the cache point is a non-finalize aggregation directly over the scan" + + " 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); falls back to a full recompute whenever the" + + " delta cannot be captured (e.g. compacted away), contains delete predicates" + + " or rewrites history rows. Requires enable_query_cache."}) + public boolean enableQueryCacheIncremental = false; + @VarAttrDef.VarAttr(name = QUERY_CACHE_FORCE_REFRESH) private boolean queryCacheForceRefresh = false; @@ -4657,6 +4688,14 @@ public void setEnableQueryCache(boolean enableQueryCache) { this.enableQueryCache = enableQueryCache; } + public boolean getEnableQueryCacheIncremental() { + return enableQueryCacheIncremental; + } + + public void setEnableQueryCacheIncremental(boolean enableQueryCacheIncremental) { + this.enableQueryCacheIncremental = enableQueryCacheIncremental; + } + public boolean isQueryCacheForceRefresh() { return queryCacheForceRefresh; } diff --git a/fe/fe-core/src/test/java/org/apache/doris/planner/QueryCacheNormalizerTest.java b/fe/fe-core/src/test/java/org/apache/doris/planner/QueryCacheNormalizerTest.java index e8bf8b628c3171..d7cb0e7d94c434 100644 --- a/fe/fe-core/src/test/java/org/apache/doris/planner/QueryCacheNormalizerTest.java +++ b/fe/fe-core/src/test/java/org/apache/doris/planner/QueryCacheNormalizerTest.java @@ -117,7 +117,29 @@ protected void runBeforeAll() throws Exception { + "distributed by hash(k1) buckets 3\n" + "properties('replication_num' = '1')"; - createTables(nonPart, part1, part2, multiLeveParts, variantTable); + String uniqueMowTable = "create table db1.uniq_mow(" + + " k1 int,\n" + + " v1 int)\n" + + "UNIQUE KEY(k1)\n" + + "distributed by hash(k1) buckets 3\n" + + "properties('replication_num' = '1', 'enable_unique_key_merge_on_write' = 'true')"; + + String uniqueMorTable = "create table db1.uniq_mor(" + + " k1 int,\n" + + " v1 int)\n" + + "UNIQUE KEY(k1)\n" + + "distributed by hash(k1) buckets 3\n" + + "properties('replication_num' = '1', 'enable_unique_key_merge_on_write' = 'false')"; + + String aggTable = "create table db1.agg_tbl(" + + " k1 int,\n" + + " v1 int sum)\n" + + "AGGREGATE KEY(k1)\n" + + "distributed by hash(k1) buckets 3\n" + + "properties('replication_num' = '1')"; + + createTables(nonPart, part1, part2, multiLeveParts, variantTable, uniqueMowTable, + uniqueMorTable, aggTable); connectContext.getSessionVariable().setDisableNereidsRules("PRUNE_EMPTY_PARTITION"); connectContext.getSessionVariable().setEnableQueryCache(true); @@ -382,6 +404,72 @@ public void testVariantSubColumnDigest() throws Exception { Assertions.assertEquals(digest1, digest3); } + @Test + public void testAllowIncremental() throws Exception { + // Switch off (the default): never allow incremental merge. + Assertions.assertFalse(connectContext.getSessionVariable().getEnableQueryCacheIncremental()); + TQueryCacheParam withoutSwitch = getQueryCacheParam( + "select k2, sum(v1) as v from db1.part1 group by k2"); + Assertions.assertFalse(withoutSwitch.allow_incremental); + + connectContext.getSessionVariable().setEnableQueryCacheIncremental(true); + try { + // DUP_KEYS with a two-phase aggregation (group by a non-distribution + // column): the cache point is the non-finalize first phase, whose + // output is merged again upstream, so incremental merge is safe. + TQueryCacheParam dupTwoPhase = getQueryCacheParam( + "select k2, sum(v1) as v from db1.part1 group by k2"); + Assertions.assertTrue(dupTwoPhase.allow_incremental); + + // One-phase aggregation finalizes inside the cached fragment: its + // output has no downstream merge, so emitting cached and delta + // blocks side by side would duplicate group keys. + TQueryCacheParam onePhase = phaseAgg(1, () -> getQueryCacheParam( + "select k1, sum(v1) as v from db1.non_part group by k1")); + Assertions.assertFalse(onePhase.allow_incremental); + + // UNIQUE merge-on-write is append-only as long as loads do not + // touch pre-existing keys; FE grants the capability and BE checks + // the delete bitmap of the delta window per tablet. Group by a + // non-distribution column so the aggregation stays two-phase and + // the decision reaches the keys-type check. + TQueryCacheParam uniqueMow = getQueryCacheParam( + "select v1, count(*) as v from db1.uniq_mow group by v1"); + Assertions.assertTrue(uniqueMow.allow_incremental); + + // UNIQUE merge-on-read resolves duplicates by merging across + // rowsets at read time: a delta-only scan cannot stand alone. + TQueryCacheParam uniqueMor = getQueryCacheParam( + "select v1, count(*) as v from db1.uniq_mor group by v1"); + Assertions.assertFalse(uniqueMor.allow_incremental); + + // AGG_KEYS merges rows inside the storage layer. + TQueryCacheParam aggKeys = getQueryCacheParam( + "select v1, count(*) as v from db1.agg_tbl group by v1"); + Assertions.assertFalse(aggKeys.allow_incremental); + + // Agg over Agg inside one fragment (distinct aggregation grouped by + // the distribution column): the cache point is not directly on the + // scan (and finalizes here as well), so incremental is not allowed. + TQueryCacheParam distinctAgg = getQueryCacheParam( + "select k1, count(distinct k2) from db1.non_part group by k1"); + Assertions.assertFalse(distinctAgg.allow_incremental); + + // Nested cache point: a non-finalize partial agg over a finalized + // colocate agg over scan. The inner agg would see only the delta + // rows during an incremental run, so its finalized output is not a + // mergeable complement of the cached snapshot -- must be rejected + // even though the cache point itself does not finalize. + TQueryCacheParam nestedAgg = getQueryCacheParam( + "select cnt, count(*) from" + + " (select k1, count(*) cnt from db1.non_part group by k1) x" + + " group by cnt"); + Assertions.assertFalse(nestedAgg.allow_incremental); + } finally { + connectContext.getSessionVariable().setEnableQueryCacheIncremental(false); + } + } + private String getDigest(String sql) throws Exception { return Hex.encodeHexString(getQueryCacheParam(sql).digest); } diff --git a/gensrc/thrift/QueryCache.thrift b/gensrc/thrift/QueryCache.thrift index 048b27f4572dd1..67c31cd4bc6120 100644 --- a/gensrc/thrift/QueryCache.thrift +++ b/gensrc/thrift/QueryCache.thrift @@ -50,4 +50,31 @@ struct TQueryCacheParam { 6: optional i64 entry_max_bytes 7: optional i64 entry_max_rows -} \ No newline at end of file + + // Whether BE is allowed to serve a stale cache entry by incremental merge: + // when the cached version is behind the current version, BE may scan only the + // delta rowsets in (cached_version, current_version], produce the partial + // aggregation of the delta, emit it together with the cached partial blocks + // (the upstream merge aggregation combines both), and write the merged entry + // back with the new version. + // + // FE only sets this to true when all of the following hold, otherwise the + // "cached + delta" union would not equal the new snapshot or could not be + // merged safely: + // - the scanned index is append-only: DUP_KEYS, or merge-on-write + // UNIQUE_KEYS (BE verifies per tablet, through the delete bitmap of the + // delta window, that no pre-existing key was rewritten); merge-on-read + // UNIQUE resolves duplicates while reading and AGG tables merge rows in + // the storage layer, so those always fall back + // - the cache point aggregation does not finalize (its output is a partial + // state that is always merged again by an upstream aggregation, so the + // cached blocks and the delta blocks can be emitted side by side) + // - the cache point aggregates the raw detail rows directly (its child is + // the olap scan node); with a nested aggregation the inner finalized agg + // would see only the delta rows, whose output is not a mergeable + // complement of the cached snapshot + // BE additionally falls back to a full recompute when the delta version path + // cannot be captured (e.g. merged away by compaction), when the delta + // contains delete predicates, or when the delta rewrites history rows. + 8: optional bool allow_incremental +} diff --git a/regression-test/suites/query_p0/cache/query_cache_incremental.groovy b/regression-test/suites/query_p0/cache/query_cache_incremental.groovy new file mode 100644 index 00000000000000..9a0b32de639fb1 --- /dev/null +++ b/regression-test/suites/query_p0/cache/query_cache_incremental.groovy @@ -0,0 +1,183 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +// Correctness of the query cache incremental merge: when a partition keeps +// receiving hourly loads, a stale cache entry is reused by scanning only the +// delta rowsets since the cached version and merging them with the cached +// partial aggregation blocks. Every query below is checked against the same +// query with the cache disabled, so the suite passes in any environment +// (including those where incremental merge falls back to a full recompute, +// e.g. cloud mode) while exercising the incremental path on local storage. +suite("query_cache_incremental") { + def tableName = "test_query_cache_incremental" + def uniqueTableName = "test_query_cache_incremental_mow" + def querySql = """ + SELECT + url, + SUM(cost) AS total_cost, + COUNT(*) AS cnt + FROM ${tableName} + WHERE dt >= '2026-01-01' + AND dt < '2026-01-15' + GROUP BY url + """ + + def normalize = { rows -> + return rows.collect { row -> row.collect { col -> String.valueOf(col) }.join("|") }.sort() + } + + // Compare the cached query result against the uncached one, twice: the + // first cached run may fill or incrementally merge the entry, the second + // one should serve it. + def checkConsistency = { String sqlText -> + sql "set enable_query_cache=false" + def expected = normalize(sql(sqlText)) + sql "set enable_query_cache=true" + assertEquals(expected, normalize(sql(sqlText))) + assertEquals(expected, normalize(sql(sqlText))) + } + + sql "set enable_nereids_distribute_planner=true" + sql "set enable_query_cache=true" + sql "set enable_query_cache_incremental=true" + sql "set parallel_pipeline_task_num=3" + sql "set enable_sql_cache=false" + + sql "DROP TABLE IF EXISTS ${tableName}" + sql """ + CREATE TABLE ${tableName} ( + dt DATE, + user_id INT, + url STRING, + cost BIGINT + ) + ENGINE=OLAP + DUPLICATE KEY(dt, user_id) + PARTITION BY RANGE(dt) + ( + PARTITION p20260101 VALUES LESS THAN ("2026-01-05"), + PARTITION p20260105 VALUES LESS THAN ("2026-01-10"), + PARTITION p20260110 VALUES LESS THAN ("2026-01-15") + ) + DISTRIBUTED BY HASH(user_id) BUCKETS 3 + PROPERTIES + ( + "replication_num" = "1" + ) + """ + + sql """ + INSERT INTO ${tableName} VALUES + ('2026-01-01',1,'/a',10), + ('2026-01-01',2,'/b',20), + ('2026-01-02',3,'/c',30), + ('2026-01-06',1,'/a',15), + ('2026-01-07',3,'/c',35), + ('2026-01-11',1,'/a',50), + ('2026-01-12',3,'/c',70) + """ + + // The query cache participates in this plan. + explain { + sql(querySql) + contains("DIGEST") + } + + // Fill the cache, then serve from it. + checkConsistency(querySql) + + // Simulate the hourly load pattern: only the latest partition receives new + // data, so the entries of the older partitions stay valid while the hot + // partition entry is stale and gets incrementally merged. Crossing + // query_cache_max_incremental_merge_count (BE config, default 8) also + // exercises the forced full recompute that compacts the entry. + for (int i = 1; i <= 10; i++) { + sql """ + INSERT INTO ${tableName} VALUES + ('2026-01-13',${i},'/a',${i}), + ('2026-01-13',${100 + i},'/inc',${10 * i}) + """ + checkConsistency(querySql) + } + + // A delete in the delta cannot be merged into the cached partial result; + // the query must fall back to a full recompute and stay correct. + sql "DELETE FROM ${tableName} PARTITION p20260110 WHERE user_id = 1" + checkConsistency(querySql) + + sql """ + INSERT INTO ${tableName} VALUES + ('2026-01-14',999,'/after-delete',1) + """ + checkConsistency(querySql) + + sql "DROP TABLE IF EXISTS ${tableName}" + + // A UNIQUE merge-on-write table takes the incremental path as long as the + // loads only append new keys (the delete bitmap of the delta window stays + // empty); a load that rewrites a pre-existing key falls back to one full + // recompute and re-bases the entry, after which pure appends are + // incremental again. Results must stay correct in every phase. + sql "DROP TABLE IF EXISTS ${uniqueTableName}" + sql """ + CREATE TABLE ${uniqueTableName} ( + dt DATE, + user_id INT, + cost BIGINT + ) + ENGINE=OLAP + UNIQUE KEY(dt, user_id) + PARTITION BY RANGE(dt) + ( + PARTITION p20260101 VALUES LESS THAN ("2026-01-05"), + PARTITION p20260105 VALUES LESS THAN ("2026-01-10") + ) + DISTRIBUTED BY HASH(user_id) BUCKETS 3 + PROPERTIES + ( + "replication_num" = "1", + "enable_unique_key_merge_on_write" = "true" + ) + """ + def uniqueQuerySql = """ + SELECT dt, SUM(cost) AS total_cost + FROM ${uniqueTableName} + GROUP BY dt + """ + sql """ + INSERT INTO ${uniqueTableName} VALUES + ('2026-01-01',1,10), + ('2026-01-02',2,20), + ('2026-01-06',3,30) + """ + checkConsistency(uniqueQuerySql) + // The hourly-append pattern on a primary-key table: every load only adds + // brand-new keys, so the stale entry merges incrementally. + for (int i = 1; i <= 3; i++) { + sql "INSERT INTO ${uniqueTableName} VALUES ('2026-01-06',${100 + i},${i})" + checkConsistency(uniqueQuerySql) + } + // A backfill rewrites history through the delete bitmap: user_id 1 gets a + // new cost, which forces one full recompute that re-bases the entry. + sql "INSERT INTO ${uniqueTableName} VALUES ('2026-01-01',1,100)" + checkConsistency(uniqueQuerySql) + // After the re-base, pure appends take the incremental path again. + sql "INSERT INTO ${uniqueTableName} VALUES ('2026-01-06',200,7)" + checkConsistency(uniqueQuerySql) + + sql "DROP TABLE IF EXISTS ${uniqueTableName}" +}