diff --git a/docs/query-acceleration/query-cache.md b/docs/query-acceleration/query-cache.md index 4711f89b609c2..b1b2563ec2c17 100644 --- a/docs/query-acceleration/query-cache.md +++ b/docs/query-acceleration/query-cache.md @@ -84,7 +84,7 @@ Intermediate nodes such as `FilterNode` and `ProjectNode` are allowed between th | Trigger Condition | Description | | ------------------ | ------------------------------------------------------------------------------------------------------------------------------------------ | -| Data change | INSERT, DELETE, UPDATE, or compaction increments the tablet version number; subsequent queries compare versions, and a mismatch is a miss. | +| Data change | INSERT, DELETE, UPDATE, or compaction increments the tablet version number; subsequent queries compare versions, and a mismatch is a miss. With [incremental merge](#incremental-merge-experimental) enabled, an entry with an older version may instead be reused by scanning only the delta. | | Schema change | ALTER TABLE changes the table structure, which changes the execution plan and the digest. | | LRU eviction | When cache memory exceeds the limit, entries are evicted according to LRU-K (K=2); a new entry must be accessed at least twice to be admitted. | | Expiration cleanup | Entries older than 24 hours are automatically removed by a periodic cleanup task. | @@ -179,6 +179,54 @@ The Query Cache relies on three properties unique to internal OLAP tables: For caching needs on external tables, use [SQL Cache](./sql-cache-manual.md) instead. +## Incremental Merge (Experimental) + + + + +By default, a cache entry whose version falls behind (for example, the current partition of a table that receives hourly loads) is discarded, and the whole instance is recomputed from a full scan. With **incremental merge** enabled, BE reuses the stale entry instead: it scans only the delta rowsets in `(cached_version, current_version]`, emits their partial aggregation state **together with** the cached partial blocks (the upstream merge aggregation combines both), and writes the merged entry back under the new version. For an hourly-load pattern this reduces the hot partition's scan cost from "the whole partition, every hour" to "one hour of new data". + +![Full recompute vs incremental merge](/images/next/query-cache/incremental-merge-architecture.svg) + +```sql +SET enable_query_cache = true; +SET enable_query_cache_incremental = true; -- shown as experimental_enable_query_cache_incremental in SHOW VARIABLES +``` + +### Prerequisites + +Incremental merge only takes effect when **all** of the following hold; in every other case the query silently falls back to a full recompute, so results are always correct: + +1. The scanned index (base table or the selected rollup) is **append-only**: either the **DUP_KEYS** model, or a **merge-on-write UNIQUE_KEYS** table whose delta window did not rewrite any pre-existing key. For merge-on-write tables BE checks the delete bitmap of `(cached_version, current_version]` per tablet: a window that only appended brand-new keys (the common "hourly load" pattern on a primary-key table) is merged incrementally, while an upsert, a partial update or a delete sign that hit an older key falls back to one full recompute, which re-bases the entry so the next pure-append load is incremental again. Merge-on-read UNIQUE tables resolve duplicates while reading and AGG tables merge rows inside the storage layer, so "cached snapshot + delta = new snapshot" would not hold there. Aggregated materialized views are rejected for the same reason. +2. The cache point is a **non-finalize aggregation directly above the scan** (the common two-phase aggregation shape, e.g. `GROUP BY` on a non-distribution column). One-phase aggregations and nested aggregation cache points are rejected. +3. The cluster runs in the **shared-nothing (local storage)** mode. +4. The delta version path is still capturable (not merged away by compaction) and contains **no DELETE predicate**, because a delete inside the delta logically removes rows that are already folded into the cached blocks, which cannot be undone by merging. + +For merge-on-write tables, the delete bitmap window check works as follows: + +![Merge-on-write delete bitmap window check](/images/next/query-cache/incremental-merge-mow-bitmap.svg) + +### Compaction of Merged Entries + +Every incremental merge appends the delta blocks to the entry, so the entry gets more fragmented over time. When an entry has accumulated `query_cache_max_incremental_merge_count` merges (BE configuration, default `8`, changeable at runtime), the next query recomputes it from a full scan, which naturally compacts the entry. + +The dataflow inside one instance, including the write-back and both entropy controls: + +![Incremental execution dataflow inside one instance](/images/next/query-cache/incremental-merge-dataflow.svg) + +### Observability + +- In the query profile of the CACHE_SOURCE operator: `HitCacheStale = 1` indicates an incremental merge; `IncrementalDeltaVersions` shows the delta range (for example `(100, 114514]`); when a stale entry could not be reused, `IncrementalFallbackReason` explains why (`delta versions not capturable`, `delta contains delete predicates`, `delta rewrites history rows`, `keys type not append-only`, and so on). +- BE metrics: `doris_be_query_cache_stale_hit_total`, `doris_be_query_cache_incremental_fallback_total`, and `doris_be_query_cache_write_back_total`. + +How BE decides between HIT, INCREMENTAL and MISS, and where each fallback reason comes from: + +![Decision flow and the eight fallback reasons](/images/next/query-cache/incremental-merge-decision-flow.svg) + +Putting it all together, from the FE authorization to the storage-level guards: + +![Incremental merge overview across FE, BE and storage](/images/next/query-cache/incremental-merge-overview.svg) + ## Configuration Parameters @@ -192,12 +240,14 @@ For caching needs on external tables, use [SQL Cache](./sql-cache-manual.md) ins | `query_cache_force_refresh` | When set to `true`, the cached result is ignored and the query is re-executed; the new result is still written to the cache. | `false` | | `query_cache_entry_max_bytes` | The maximum size in bytes of a single cache entry; fragment results that exceed this limit are not cached. | `5242880` (5 MB) | | `query_cache_entry_max_rows` | The maximum number of rows of a single cache entry; fragment results that exceed this limit are not cached. | `500000` | +| `enable_query_cache_incremental` | (Experimental) Allows BE to reuse a stale cache entry by scanning only the delta rowsets since the cached version and merging them with the cached partial aggregation blocks. See [Incremental Merge](#incremental-merge-experimental). | `false` | ### BE Configuration (be.conf) | Parameter | Description | Default | | ------------------ | -------------------------------------------------------- | ------- | | `query_cache_size` | The total memory capacity of the Query Cache on each BE (MB). | `512` | +| `query_cache_max_incremental_merge_count` | The maximum number of incremental merges accumulated on one cache entry before a full recompute is forced to compact it. `0` disables incremental merge at runtime. | `8` | :::note The `query_cache_max_size_mb` and `query_cache_elasticity_size_mb` settings in `be.conf` control the legacy SQL Result Cache. They are **not** the pipeline-level Query Cache described in this article. Do not confuse them. diff --git a/i18n/zh-CN/docusaurus-plugin-content-docs/current/query-acceleration/query-cache.md b/i18n/zh-CN/docusaurus-plugin-content-docs/current/query-acceleration/query-cache.md index 92c021c1ff678..9b9355202f9da 100644 --- a/i18n/zh-CN/docusaurus-plugin-content-docs/current/query-acceleration/query-cache.md +++ b/i18n/zh-CN/docusaurus-plugin-content-docs/current/query-acceleration/query-cache.md @@ -84,7 +84,7 @@ SELECT region, SUM(revenue) FROM orders WHERE dt = '2024-01-01' GROUP BY region; | 触发条件 | 说明 | | ------------ | ----------------------------------------------------------------------------------------------------- | -| 数据变更 | INSERT、DELETE、UPDATE 或 Compaction 使 Tablet 版本号递增;后续查询比对版本,不一致即未命中 | +| 数据变更 | INSERT、DELETE、UPDATE 或 Compaction 使 Tablet 版本号递增;后续查询比对版本,不一致即未命中。开启[增量合并](#增量合并实验性)后,版本落后的条目可改为只扫增量而被复用 | | Schema 变更 | ALTER TABLE 改变表结构,从而改变执行计划与摘要 | | LRU 淘汰 | 缓存内存超限时,按 LRU-K(K=2)淘汰;新条目须至少被访问两次才能被准入 | | 过期清理 | 超过 24 小时的条目由周期性清理任务自动移除 | @@ -179,6 +179,54 @@ Query Cache 依赖内部 OLAP 表的三个特有属性: 外部表的缓存需求请改用 [SQL Cache](./sql-cache-manual.md)。 +## 增量合并(实验性) + + + + +默认情况下,版本落后的缓存条目(例如按小时导入的表的当前分区)会被直接作废,对应的 Pipeline 实例回到全量扫描重算。开启**增量合并**后,BE 会复用这条过期条目:只扫描 `(缓存版本, 当前版本]` 区间的增量 rowset,把增量的聚合中间状态与缓存的中间状态**并排**交给上游合并聚合(两者本就是同构的可合并输入),并以新版本回写合并后的条目。对小时级导入场景,热分区的扫描成本从“每小时扫整个分区”降为“只扫最近一小时的新数据”。 + +![全量重算与增量合并对比](/images/next/query-cache/incremental-merge-architecture.svg) + +```sql +SET enable_query_cache = true; +SET enable_query_cache_incremental = true; -- SHOW VARIABLES 中显示为 experimental_enable_query_cache_incremental +``` + +### 生效前提 + +增量合并仅在**同时**满足以下条件时生效;任一不满足都会静默回退为全量重算,查询结果始终正确: + +1. 被扫描的索引(基表或选中的 rollup)为**追加写**模型:**DUP_KEYS** 表,或增量窗口内未改写既有主键的**写时合并(merge-on-write)UNIQUE_KEYS** 表。对写时合并表,BE 会按 tablet 检查 `(cached_version, current_version]` 窗口内的 delete bitmap:窗口内只新增了全新主键(主键表上常见的“每小时追加导入”模式)即可增量合并;一旦有 upsert、部分列更新或 delete sign 命中了更早的主键,则回退一次全量重算,重算会以新版本重建条目,下一次纯追加导入即恢复增量。读时合并(merge-on-read)UNIQUE 表在读取时跨 rowset 归并去重、AGG 表在存储层合并行,“缓存快照 + 增量 = 新快照”不成立;聚合物化视图同理被拒绝。 +2. 缓存点是**直接位于扫描之上、且不做 finalize 的聚合**(常见的两阶段聚合形态,例如按非分桶列 `GROUP BY`)。单阶段聚合与嵌套聚合缓存点会被拒绝。 +3. 集群为**存算一体(本地存储)**部署。 +4. 增量版本路径仍可捕获(未被 compaction 合并跨界),且增量中**不含 DELETE 谓词**(增量中的删除会逻辑上移除已折入缓存块的行,无法通过合并撤销)。 + +写时合并表的 delete bitmap 窗口检查方式如下: + +![写时合并表的 delete bitmap 窗口检查](/images/next/query-cache/incremental-merge-mow-bitmap.svg) + +### 增量部分的合并 + +每次增量合并都会把增量块追加进条目,条目随之逐渐碎片化。当条目累计 `query_cache_max_incremental_merge_count` 次合并(BE 配置,默认 `8`,运行时可调)后,下一次查询会强制全量重算,条目自然压实。 + +单个实例内的增量执行数据流(含回写与两道熵控制): + +![单实例内的增量执行数据流](/images/next/query-cache/incremental-merge-dataflow.svg) + +### 可观测性 + +- CACHE_SOURCE 算子的查询 profile:`HitCacheStale = 1` 表示走了增量合并;`IncrementalDeltaVersions` 展示增量区间(如 `(100, 114514]`);过期条目未能复用时,`IncrementalFallbackReason` 给出原因(`delta versions not capturable`、`delta contains delete predicates`、`delta rewrites history rows`、`keys type not append-only` 等)。 +- BE 指标:`doris_be_query_cache_stale_hit_total`、`doris_be_query_cache_incremental_fallback_total`、`doris_be_query_cache_write_back_total`。 + +BE 如何在 HIT、INCREMENTAL、MISS 三态间决策,以及每种回退原因的来源: + +![决策流程与八种回退原因](/images/next/query-cache/incremental-merge-decision-flow.svg) + +从 FE 授权到存储层防线的全景视图: + +![增量合并全景:FE、BE 与存储层](/images/next/query-cache/incremental-merge-overview.svg) + ## 配置参数 @@ -192,12 +240,14 @@ Query Cache 依赖内部 OLAP 表的三个特有属性: | `query_cache_force_refresh` | 设为 `true` 时忽略缓存结果并重新执行查询;新结果仍会写入缓存 | `false` | | `query_cache_entry_max_bytes` | 单个缓存条目的最大字节数;超过该限制的 Fragment 结果不会被缓存 | `5242880`(5 MB) | | `query_cache_entry_max_rows` | 单个缓存条目的最大行数;超过该限制的 Fragment 结果不会被缓存 | `500000` | +| `enable_query_cache_incremental` | (实验性)允许 BE 以增量合并方式复用过期缓存条目:只扫描缓存版本之后的增量 rowset,并与缓存的聚合中间结果一起交给上游合并。详见[增量合并](#增量合并实验性) | `false` | ### BE 配置(be.conf) | 参数 | 说明 | 默认值 | | ------------------ | ------------------------------------------ | ------ | | `query_cache_size` | 每个 BE 上 Query Cache 的总内存容量(MB) | `512` | +| `query_cache_max_incremental_merge_count` | 单个缓存条目上累计增量合并的最大次数,达到后强制一次全量重算以压实条目;设为 `0` 可在运行时禁用增量合并 | `8` | :::note `be.conf` 中的 `query_cache_max_size_mb` 和 `query_cache_elasticity_size_mb` 控制的是旧版 SQL Result Cache,**不是** 本文描述的流水线级别 Query Cache,请勿混淆。 diff --git a/static/images/next/query-cache/incremental-merge-architecture.svg b/static/images/next/query-cache/incremental-merge-architecture.svg new file mode 100644 index 0000000000000..6e7849ef5d5a2 --- /dev/null +++ b/static/images/next/query-cache/incremental-merge-architecture.svg @@ -0,0 +1,125 @@ + + Query Cache Incremental Merge + Comparison diagram. Left: default behavior discards the stale cache entry and rescans the whole partition every hour. Right: incremental merge reuses the stale entry, scans only one hour of delta rowsets, sends cached partial blocks and delta partial blocks side by side to the upstream merge aggregate, and writes the combined blocks back as the v101 entry. + + + + + + + + + + + + + + + + + + + + + + + + + Full Recompute (default) + Incremental Merge (experimental) + + + + VS + + + + hourly load bumps the version to v101 + + + + + + + + + Cache Entry v100 + stale, discarded + + + + + + Scan ALL rowsets [0, v101] + + + + + + + + + + + + + whole partition, every hour + + + + + + + Partial Aggregate + + + + to upstream Merge Agg + + + + same hourly load, now at version v101 + + + + + + + + + Cache Entry v100 + reused + pinned + + + + + + Scan DELTA only + (v100, v101] + + + + one hour of new data + + + + + + + Partial Aggregate + + + + + cached partial blocks + + + + to upstream Merge Agg + + + + + + write back v101 (delta_count+1) + \ No newline at end of file diff --git a/static/images/next/query-cache/incremental-merge-dataflow.svg b/static/images/next/query-cache/incremental-merge-dataflow.svg new file mode 100644 index 0000000000000..10efafb3e70de --- /dev/null +++ b/static/images/next/query-cache/incremental-merge-dataflow.svg @@ -0,0 +1,82 @@ + + Query Cache Incremental Merge: Dataflow inside one Instance + In incremental mode the scan reads only the delta rowsets and feeds the unchanged partial aggregation; the cache source emits the pinned cached blocks first, streams the delta partial blocks, and on eos writes the concatenated blocks back under the new version with delta_count plus one. Entry growth is bounded by the compaction threshold and the entry size limits. + + + + + + + + + + + + + + + + + + + Incremental Execution inside one Instance + the decision pinned entry v100 and pre-captured the delta read source; the plan itself is unchanged + + + + cache entry v100 (pinned) + partial blocks + slot orders + + delta_count, bytes, rows + + write back on eos: + insert(key, v101, cached clones + delta + blocks, slot orders, delta_count + 1) + same key: the v100 entry gets replaced + + + OlapScan + delta only: + (v100, v101] + single scanner, parallel + builder disabled + + + Partial + Aggregate + delta rows only + + + CacheSink + data queue + + + + CacheSource + 1. emit cached blocks first, + reordered to this query's slot + order, cloned for the write back + 2. stream the delta blocks + 3. on eos: write back v101 + profile: HitCacheStale = 1, + IncrementalDeltaVersions (100, 101] + + cached partial blocks + + + to upstream Merge Agg + + + + entropy control + + every merge appends blocks, so after + query_cache_max_incremental_merge_count + merges (BE config, default 8) the next query + recomputes in full and re-bases the entry + + entry too big? + over entry_max: + still scan only the + delta, but skip the + write back + \ No newline at end of file diff --git a/static/images/next/query-cache/incremental-merge-decision-flow.svg b/static/images/next/query-cache/incremental-merge-decision-flow.svg new file mode 100644 index 0000000000000..a82c0410130a7 --- /dev/null +++ b/static/images/next/query-cache/incremental-merge-decision-flow.svg @@ -0,0 +1,98 @@ + + Query Cache Incremental Merge: Decision Flow + Both operators of an instance call get_or_make_decision on the shared QueryCacheRuntime. The chain checks the cache key, binlog scans, force refresh, the entry lookup and the version. A stale entry is pinned and validated for incremental merge; any failing check falls back to a full recompute with one of eight reasons visible in the profile. + + + + + + + + + + + + + + + + + + + One Decision per Instance + scan and cache-source operators both call get_or_make_decision(): the first caller decides, both see the same object + + + 1. build the cache key + digest + tablet set + partition range + + + 2. row-binlog scan? + a different data stream, never cache + + + 3. force refresh? + user asked to recompute and refill + + + 4. lookup_any_version(key) + exact or stale entry, else fill later + + + 5. cached version == current? + equal versions can serve directly + + + + 6. try incremental on the stale entry + pin the entry, then verify: + - not cloud mode + - cached version < current version + - delta_count below the compaction threshold + - every tablet: append-only keys (DUP, or MoW + with a clean delete-bitmap window), delta + (cached, current] capturable, no delete + predicates in the delta + + MISS without write back + invalid cache key (shared decision, one warning) + or a binlog scan: neither hits nor fills the cache + + MISS: recompute and write back + force refresh, or no entry found yet + + + MISS with a fallback reason + + "cloud mode" + "cached entry is newer" + "delta count reached compaction threshold" + "tablet not found" + "keys type not append-only" + "delta versions not capturable" + "delta contains delete predicates" + "delta rewrites history rows" + + HIT (entry pinned) + serve the cached partial blocks, then eos + + + INCREMENTAL (entry pinned) + delta read source pre-captured at decision time; + cached blocks + delta blocks go upstream together + + + fails + + yes + + yes + + not found + + yes, equal + + any check fails + + all pass + \ No newline at end of file diff --git a/static/images/next/query-cache/incremental-merge-mow-bitmap.svg b/static/images/next/query-cache/incremental-merge-mow-bitmap.svg new file mode 100644 index 0000000000000..7190e43fee0e8 --- /dev/null +++ b/static/images/next/query-cache/incremental-merge-mow-bitmap.svg @@ -0,0 +1,71 @@ + + Query Cache Incremental Merge: Merge-on-Write Delete Bitmap Window + A merge-on-write UNIQUE table qualifies for incremental merge per tablet by inspecting the delete bitmap of the delta window. Entries stamped inside the window that target baseline rowsets mean history was rewritten and force one full recompute; entries targeting the delta itself or stamped outside the window are harmless. + + + + + + + + + + + + + + + + + + + Merge-on-Write: Delete Bitmap Window Check + a stale entry at v100 wants to merge the delta (v100, v101]: may the window be merged on top of the cached blocks? + + baseline rowsets: already folded into the cached entry + + [0-60] + + [61-90] + + [91-100] + delta rowset: (100, 101] + + + v101 + + cached v100 | current v101 + delete bitmap entries, key = (RowsetId, SegmentId, Version): + + stamped v101, targets a baseline rowset + a backfill upsert replaced an old row + + + history rewritten: fall back once, + "delta rewrites history rows" + + stamped v101, targets the delta rowset + the same new key written twice, a losing + sequence-column row, a local delete sign + + + harmless: the delta scan applies the + bitmap itself while reading + + stamped v40, v103 or pending (v0) + before cached, after current, unpublished + + outside the window (v100, v101]: + ignored by the check + + window clean (pure appends): + incremental merge, exactly as safe as on + a duplicate-key table + + + + window rewrote history: + + one full recompute re-bases the entry at v101; + the next pure-append hour is incremental again + \ No newline at end of file diff --git a/static/images/next/query-cache/incremental-merge-overview.svg b/static/images/next/query-cache/incremental-merge-overview.svg new file mode 100644 index 0000000000000..a9808e57cfe2b --- /dev/null +++ b/static/images/next/query-cache/incremental-merge-overview.svg @@ -0,0 +1,99 @@ + + Query Cache Incremental Merge: Overview + FE authorizes incremental merge with four static conditions and ships one optional thrift flag. On BE a per-fragment QueryCacheRuntime makes one idempotent decision per instance, pinning the entry and pre-capturing the delta; the unchanged plan then runs in one of three modes and writes the merged entry back. Storage-level facts (version graph, delete predicates, the merge-on-write delete bitmap window) guard correctness, and profile fields, metrics and two switches make the feature observable and controllable. + + + + + + + + + + + + + + + + + + + Query Cache Incremental Merge: the Whole Picture + + FE + + QueryCacheNormalizer.computeAllowIncremental + 1. enable_query_cache_incremental = true (experimental session switch) + 2. the cache point aggregation does not finalize (merged upstream) + 3. the aggregation sits directly above the olap scan (no nesting) + 4. selected index append-only: DUP_KEYS, or merge-on-write UNIQUE + + plan digest (unchanged) + the switch is not part of the digest: + sessions with it on and off share the + same entries and refresh one another; + reuse mode never changes results + + thrift TQueryCacheParam field 8: optional bool allow_incremental + + BE + + + QueryCacheRuntime (fragment) + get_or_make_decision(): idempotent, + one decision per instance, entry pinned, + delta read source pre-captured + + HIT + + INCREMENTAL + + MISS + 8 fallback reasons in the profile + + unchanged plan, three modes + OlapScan (all / delta / none) + Partial Aggregate + CacheSink and data queue + CacheSource emits cached and + delta blocks side by side + upstream Merge Agg combines + partial states as it always did + + + Query Cache + (LRU, per BE) + key: digest + tablets + range + value: partial blocks, version, + slot orders, delta_count, + bytes and rows + write back replaces same key + + + + Storage + + version graph + capture exactly (cached, current]; + compacted away: VERSION_ALREADY_ + MERGED, fall back + + delete predicates + a DELETE inside the delta removes + rows already folded into the entry: + fall back + + MoW delete bitmap + window entries targeting baseline + rowsets mean rewritten history: + fall back once, then self-heal + + + + + Observe and control + profile: HitCache, HitCacheStale, IncrementalDeltaVersions, IncrementalFallbackReason, InsertCache + metrics: stale_hit_total, incremental_fallback_total, write_back_total (prefix doris_be_query_cache_) + controls: enable_query_cache_incremental (session), query_cache_max_incremental_merge_count (BE config, default 8, 0 disables) + \ No newline at end of file