ReplacingMergeTree

虽然事务型数据库针对事务型更新和删除工作负载进行了优化，但 OLAP 数据库对这类操作提供的保证较少。相反，它们针对批量插入不变数据进行了优化，以提高分析查询的速度。虽然 ClickHouse 通过变动提供更新操作，以及一种删除行的轻量级方法，但其面向列的结构意味着应谨慎调度这些操作，如上所述。这些操作是异步处理的，由单个线程处理，并且需要（在更新情况下）将数据重写到磁盘。因此，它们不应用于大量的小更改。为了处理更新和删除行的流，同时避免上述使用模式，我们可以使用 ClickHouse 表引擎 ReplacingMergeTree。

自动插入行的 upsert

ReplacingMergeTree 表引擎允许对行应用更新操作，而无需使用效率低下的 ALTER 或 DELETE 语句，因为它为用户提供了插入同一行的多个副本并指定其中一个为最新版本的能力。一个后台进程反过来异步删除同一行的旧版本，通过使用不可变插入来有效地模仿更新操作。这依赖于表引擎识别重复行的能力。这是通过使用 ORDER BY 子句来确定唯一性实现的，即，如果两行在 ORDER BY 中指定列的相同值相同，则它们被视为重复。在定义表时指定的 version 列允许在两行被识别为重复时保留行的最新版本，即保留版本值最高的行。我们在下面的示例中说明了此过程。这里，行由 A 列（表的 ORDER BY）唯一标识。我们假设这些行已作为两个批次插入，导致在磁盘上形成两个数据部分。稍后，在异步后台进程期间，这些部分将合并在一起。

ReplacingMergeTree 另外允许指定一个已删除列。这可以包含 0 或 1，其中值为 1 表示行（及其重复项）已被删除，而零则用于其他情况。注意：删除的行在合并时不会被删除。

在此过程中，以下内容在部分合并期间发生

由 A 列的值 1 标识的行具有版本为 2 的更新行和版本为 3 的删除行（以及已删除列值为 1）。因此，保留了标记为已删除的最新行。
由 A 列的值 2 标识的行具有两个更新行。保留了后一行，其价格列的值为 6。
由 A 列的值 3 标识的行具有版本为 1 的行和版本为 2 的删除行。保留了此删除行。

由于此合并过程，我们有四行代表最终状态

请注意，删除的行永远不会被删除。可以使用 OPTIMIZE table FINAL CLEANUP 强制删除它们。这需要实验性设置 allow_experimental_replacing_merge_with_cleanup=1。这应仅在以下条件下执行

您可以确定在发出操作后，不会插入具有旧版本的行（对于那些要使用清理删除的行）。如果插入这些行，它们将被错误地保留，因为删除的行将不再存在。
在发出清理之前，确保所有副本都已同步。这可以通过以下命令实现

SYSTEM SYNC REPLICA table

我们建议在 (1) 得到保证后暂停插入，直到此命令和随后的清理完成。

仅建议针对删除次数很少或中等（少于 10%）的表使用 ReplacingMergeTree 处理删除，除非可以安排使用上述条件进行清理的时期。

提示：用户还可以针对不再受更改影响的选择性分区发出 OPTIMIZE FINAL CLEANUP。

选择主键/去重键

上面，我们重点介绍了 ReplacingMergeTree 情况下还必须满足的一个重要附加约束：ORDER BY 列的值在更改中唯一地标识一行。如果从 Postgres 等事务型数据库迁移，则应将原始 Postgres 主键包含在 Clickhouse ORDER BY 子句中。

ClickHouse 的用户将熟悉选择其表中 ORDER BY 子句中的列以优化查询性能。通常，应根据您的频繁查询并按基数递增的顺序列出这些列。重要的是，ReplacingMergeTree 强制执行一个额外的约束 - 这些列必须是不可变的，即，如果从 Postgres 复制，仅在这些列在底层 Postgres 数据中没有更改时才将列添加到此子句中。虽然其他列可以更改，但这些列必须一致才能进行唯一行标识。对于分析工作负载，Postgres 主键通常没有什么用处，因为用户很少执行点行查找。鉴于我们建议按基数递增的顺序对列进行排序，以及匹配在 ORDER BY 中较早列出的列上的匹配通常更快这一事实，Postgres 主键应附加到 ORDER BY 的末尾（除非它具有分析价值）。在 Postgres 中有多个列形成主键的情况下，应将它们附加到 ORDER BY，尊重基数和查询值的可能性。用户还可以希望使用 MATERIALIZED 列通过值连接生成唯一主键。

考虑来自 Stack Overflow 数据集的帖子表。

CREATE TABLE stackoverflow.posts_updateable
(
       `Version` UInt32,
       `Deleted` UInt8,
    `Id` Int32 CODEC(Delta(4), ZSTD(1)),
    `PostTypeId` Enum8('Question' = 1, 'Answer' = 2, 'Wiki' = 3, 'TagWikiExcerpt' = 4, 'TagWiki' = 5, 'ModeratorNomination' = 6, 'WikiPlaceholder' = 7, 'PrivilegeWiki' = 8),
    `AcceptedAnswerId` UInt32,
    `CreationDate` DateTime64(3, 'UTC'),
    `Score` Int32,
    `ViewCount` UInt32 CODEC(Delta(4), ZSTD(1)),
    `Body` String,
    `OwnerUserId` Int32,
    `OwnerDisplayName` String,
    `LastEditorUserId` Int32,
    `LastEditorDisplayName` String,
    `LastEditDate` DateTime64(3, 'UTC') CODEC(Delta(8), ZSTD(1)),
    `LastActivityDate` DateTime64(3, 'UTC'),
    `Title` String,
    `Tags` String,
    `AnswerCount` UInt16 CODEC(Delta(2), ZSTD(1)),
    `CommentCount` UInt8,
    `FavoriteCount` UInt8,
    `ContentLicense` LowCardinality(String),
    `ParentId` String,
    `CommunityOwnedDate` DateTime64(3, 'UTC'),
    `ClosedDate` DateTime64(3, 'UTC')
)
ENGINE = ReplacingMergeTree(Version, Deleted)
PARTITION BY toYear(CreationDate)
ORDER BY (PostTypeId, toDate(CreationDate), CreationDate, Id)

我们使用 ORDER BY 键 (PostTypeId, toDate(CreationDate), CreationDate, Id)。Id 列是每篇文章唯一的，确保可以对行进行去重。根据需要向模式添加 Version 和 Deleted 列。

查询 ReplacingMergeTree

在合并时，ReplacingMergeTree 使用 ORDER BY 列的值作为唯一标识符来识别重复行，并仅保留最高版本，或者如果最新版本指示删除，则删除所有重复项。但是，这仅提供最终一致性 - 它不保证行将被去重，您不应依赖它。因此，由于查询中考虑了更新和删除行，查询可能会产生不正确的答案。

为了获得正确的答案，用户需要用查询时去重和删除移除来补充后台合并。这可以通过使用 FINAL 运算符来实现。

考虑上面的帖子表。我们可以使用加载此数据集的正常方法，但除了值 0 之外，还要指定一个删除列和版本列。出于示例目的，我们只加载 10000 行。

INSERT INTO stackoverflow.posts_updateable SELECT 0 AS Version, 0 AS Deleted, *
FROM s3('https://datasets-documentation.s3.eu-west-3.amazonaws.com/stackoverflow/parquet/posts/*.parquet') WHERE AnswerCount > 0 LIMIT 10000

0 rows in set. Elapsed: 1.980 sec. Processed 8.19 thousand rows, 3.52 MB (4.14 thousand rows/s., 1.78 MB/s.)

让我们确认行数

SELECT count() FROM stackoverflow.posts_updateable

┌─count()─┐
│   10000 │
└─────────┘

1 row in set. Elapsed: 0.002 sec.

我们现在更新帖子答案统计信息。与其更新这些值，不如插入 5000 行的新副本，并将它们的版本号加一（这意味着表中将存在 150 行）。我们可以用简单的 INSERT INTO SELECT 来模拟这一点

INSERT INTO posts_updateable SELECT
    Version + 1 AS Version,
    Deleted,
    Id,
    PostTypeId,
    AcceptedAnswerId,
    CreationDate,
    Score,
    ViewCount,
    Body,
    OwnerUserId,
    OwnerDisplayName,
    LastEditorUserId,
    LastEditorDisplayName,
    LastEditDate,
    LastActivityDate,
    Title,
    Tags,
    AnswerCount,
    CommentCount,
    FavoriteCount,
    ContentLicense,
    ParentId,
    CommunityOwnedDate,
    ClosedDate
FROM posts_updateable --select 100 random rows
WHERE (Id % toInt32(floor(randUniform(1, 11)))) = 0
LIMIT 5000

0 rows in set. Elapsed: 4.056 sec. Processed 1.42 million rows, 2.20 GB (349.63 thousand rows/s., 543.39 MB/s.)

此外，我们通过重新插入行但将删除列值设置为 1 来删除 1000 个随机帖子。同样，可以简单地用 INSERT INTO SELECT 来模拟这一点。

INSERT INTO posts_updateable SELECT
    Version + 1 AS Version,
    1 AS Deleted,
    Id,
    PostTypeId,
    AcceptedAnswerId,
    CreationDate,
    Score,
    ViewCount,
    Body,
    OwnerUserId,
    OwnerDisplayName,
    LastEditorUserId,
    LastEditorDisplayName,
    LastEditDate,
    LastActivityDate,
    Title,
    Tags,
    AnswerCount + 1 AS AnswerCount,
    CommentCount,
    FavoriteCount,
    ContentLicense,
    ParentId,
    CommunityOwnedDate,
    ClosedDate
FROM posts_updateable --select 100 random rows
WHERE (Id % toInt32(floor(randUniform(1, 11)))) = 0 AND AnswerCount > 0
LIMIT 1000

0 rows in set. Elapsed: 0.166 sec. Processed 135.53 thousand rows, 212.65 MB (816.30 thousand rows/s., 1.28 GB/s.)

上述操作的结果将是 16000 行，即 10000 + 5000 + 1000。这里正确的总数应该是，实际上我们应该只有比原始总数少 1000 行，即 10000 - 1000 = 9000。

SELECT count()
FROM posts_updateable

┌─count()─┐
│   10000 │
└─────────┘
1 row in set. Elapsed: 0.002 sec.

您的结果将因已发生的合并而异。我们可以看到这里的结果不同，因为我们有重复的行。将 FINAL 应用于表会提供正确的结果。

SELECT count()
FROM posts_updateable
FINAL

┌─count()─┐
│   9000 │
└─────────┘

1 row in set. Elapsed: 0.006 sec. Processed 11.81 thousand rows, 212.54 KB (2.14 million rows/s., 38.61 MB/s.)
Peak memory usage: 8.14 MiB.

FINAL 性能

尽管一直在改进，FINAL 运算符在查询中仍会带来性能开销。当查询没有过滤主键列时，这种开销最为明显，因为会导致读取更多数据，并增加去重开销。如果用户使用 WHERE 条件过滤关键列，则加载和传递到去重的數據量将减少。

如果 WHERE 条件未使用关键列，ClickHouse 目前在使用 FINAL 时不使用 PREWHERE 优化。这种优化旨在减少对未过滤列的读取行数。有关模拟这种 PREWHERE 以及如何潜在提高性能的示例，请访问这里。

利用 ReplacingMergeTree 的分区

ClickHouse 中的数据合并发生在分区级别。使用 ReplacingMergeTree 时，我们建议用户按照最佳实践对表格进行分区，前提是用户可以确保 **分区键不会改变**。这样可以确保与同一行相关的更新将被发送到同一 ClickHouse 分区。您可以重复使用与 Postgres 相同的分区键，前提是您遵循此处概述的最佳实践。

假设这是事实，用户可以使用 do_not_merge_across_partitions_select_final=1 设置来提高 FINAL 查询性能。此设置会导致在使用 FINAL 时独立合并和处理分区。

考虑以下帖子表格，其中我们不使用分区

CREATE TABLE stackoverflow.posts_no_part
(
    `Version` UInt32,
    `Deleted` UInt8,
    `Id` Int32 CODEC(Delta(4), ZSTD(1)),
    …
)
ENGINE = ReplacingMergeTree
ORDER BY (PostTypeId, toDate(CreationDate), CreationDate, Id)

INSERT INTO stackoverflow.posts_no_part SELECT 0 AS Version, 0 AS Deleted, *
FROM s3('https://datasets-documentation.s3.eu-west-3.amazonaws.com/stackoverflow/parquet/posts/*.parquet')

0 rows in set. Elapsed: 182.895 sec. Processed 59.82 million rows, 38.07 GB (327.07 thousand rows/s., 208.17 MB/s.)

为了确保 FINAL 需要执行一些工作，我们将更新 100 万行 - 通过插入重复行来增加它们的 AnswerCount。

INSERT INTO posts_no_part SELECT Version + 1 AS Version, Deleted, Id, PostTypeId, AcceptedAnswerId, CreationDate, Score, ViewCount, Body, OwnerUserId, OwnerDisplayName, LastEditorUserId, LastEditorDisplayName, LastEditDate, LastActivityDate, Title, Tags, AnswerCount + 1 AS AnswerCount, CommentCount, FavoriteCount, ContentLicense, ParentId, CommunityOwnedDate, ClosedDate
FROM posts_no_part
LIMIT 1000000

使用 FINAL 计算每年答案的总和

SELECT toYear(CreationDate) AS year, sum(AnswerCount) AS total_answers
FROM posts_no_part
FINAL
GROUP BY year
ORDER BY year ASC

┌─year─┬─total_answers─┐
│ 2008 │        371480 │
…
│ 2024 │        127765 │
└──────┴───────────────┘

17 rows in set. Elapsed: 2.338 sec. Processed 122.94 million rows, 1.84 GB (52.57 million rows/s., 788.58 MB/s.)
Peak memory usage: 2.09 GiB.

对按年份分区的表格重复这些相同步骤，并使用 do_not_merge_across_partitions_select_final=1 重复上述查询。

CREATE TABLE stackoverflow.posts_with_part
(
    `Version` UInt32,
    `Deleted` UInt8,
    `Id` Int32 CODEC(Delta(4), ZSTD(1)),
    ...
)
ENGINE = ReplacingMergeTree
PARTITION BY toYear(CreationDate)
ORDER BY (PostTypeId, toDate(CreationDate), CreationDate, Id)

// populate & update omitted

SELECT toYear(CreationDate) AS year, sum(AnswerCount) AS total_answers
FROM posts_with_part
FINAL
GROUP BY year
ORDER BY year ASC

┌─year─┬─total_answers─┐
│ 2008 │        387832 │
│ 2009 │    1165506 │
│ 2010 │    1755437 │
...
│ 2023 │        787032 │
│ 2024 │        127765 │
└──────┴───────────────┘

17 rows in set. Elapsed: 0.994 sec. Processed 64.65 million rows, 983.64 MB (65.02 million rows/s., 989.23 MB/s.)

如所示，在这种情况下，分区显著提高了查询性能，因为允许在分区级别并行进行去重过程。

自动插入行的 upsert​

选择主键/去重键​

查询 ReplacingMergeTree​

FINAL 性能​

利用 ReplacingMergeTree 的分区​

自动插入行的 upsert

选择主键/去重键

查询 ReplacingMergeTree

FINAL 性能

利用 ReplacingMergeTree 的分区