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}" +}