再試行時の挿入の重複排除

挿入操作は、タイムアウトなどのエラーによって失敗することがあります。挿入に失敗した場合、データがすでに正常に挿入されていることもあれば、まだ挿入されていないこともあります。このガイドでは、同じデータが複数回挿入されないように、挿入の再試行時の重複排除の仕組みを説明します。

挿入が再試行されると、ClickHouse はそのデータがすでに正常に挿入済みかどうかを判定しようとします。挿入済みのデータが重複としてマークされた場合、ClickHouse はそれを宛先テーブルに挿入しません。ただし、ユーザーには、そのデータが通常どおり挿入された場合と同様に、操作成功のステータスが返されます。

重複排除は、同期挿入、非同期挿入、INSERT ... SELECT クエリに対応しています。deduplicate_insert 設定は、同期挿入と非同期挿入を制御します。INSERT ... SELECT には追加の注意が必要で、専用の設定があります。挿入の重複排除を制御する設定を参照してください。

ユーザーは、挿入操作が成功するまで再試行する必要があります。すべての再試行が失敗した場合、データが挿入されたかどうかは判断できません。materialized view が関係する場合、データがどのテーブルに現れている可能性があるのかも不明です。materialized view がソーステーブルと同期していない可能性もあります。

再試行の過程で、*_deduplication_window を超える数の他の挿入操作が行われると、重複排除が意図どおりに機能しない場合があります。この場合、同じデータが複数回挿入される可能性があります。

ClickHouse が挿入時に重複排除を行うのは、次の両方の条件が満たされる場合のみです。

1. 宛先テーブルで重複排除ログが保持されていること。これはテーブルレベルの設定です。
2. クエリで重複排除が有効になっていること。これはクエリレベルの設定です。

挿入時の重複排除をサポートするのは *MergeTree エンジンのみです。

*ReplicatedMergeTree エンジンでは、重複排除ログはデフォルトで有効になっており、replicated_deduplication_window および replicated_deduplication_window_seconds 設定で制御されます。非レプリケートの *MergeTree エンジンでは、ログは non_replicated_deduplication_window 設定で制御され、デフォルト値は 0 です。したがって、通常の MergeTree テーブルでは、そのウィンドウを正の値に設定するまで重複排除は行われません。

上記の設定は、テーブルの重複排除ログのパラメーターを決定します。重複排除ログには有限個の block_id が保存され、これによって重複排除の動作が決まります (以下を参照) 。

deduplicate_insert には次の 3 つの値を指定できます。

• enable — INSERT クエリの重複排除を有効にします。
• disable — INSERT クエリの重複排除を無効にします。
• backward_compatible_choice — レガシー設定の insert_deduplicate (同期挿入) および async_insert_deduplicate (非同期挿入) に判断を委ねます。

deduplicate_insert = disable で実行したクエリでは、そのブロックの block_id は書き込まれないことに注意してください。このようなデータは、後から deduplicate_insert = enable を指定して挿入を再試行しても重複排除できません。宛先テーブルに重複排除ログがない場合も同様です。何も記録されないため、再試行時に照合できません。

1. INSERT ... SELECT クエリでは、deduplicate_insert_select が使用されます。INSERT … SELECT の重複排除を参照してください。
2. その他のすべての INSERT では、deduplicate_insert が使用されます。
3. insert_deduplicate と async_insert_deduplicate は、deduplicate_insert が backward_compatible_choice の場合にのみ読み取られます。

バージョン 26.2 では、async_insert と deduplicate_blocks_in_dependent_materialized_views のデフォルトも有効に変更されました。compatibility 互換性設定は、これら 3 つすべてを制御します。compatibility を 26.2 より前のバージョンに設定すると、これらの設定には以前のデフォルトが適用されます。deduplicate_insert は backward_compatible_choice となり、insert_deduplicate と async_insert_deduplicate に判断を委ねます。明示的に設定した値は常に優先され、compatibility の影響を受けません。

ClickHouse にデータが挿入されると、行数とバイト数に基づいてデータはブロックに分割されます。

*MergeTree エンジンを使用するテーブルでは、各ブロックに一意の block_id が割り当てられます。これは、そのブロック内のデータのハッシュです。この block_id は、挿入操作の一意なキーとして使用されます。同じ block_id が重複排除ログ内で見つかった場合、そのブロックは重複と見なされ、テーブルには挿入されません。

このアプローチは、挿入操作に異なるデータが含まれる場合には有効です。ただし、同じデータを意図的に複数回挿入する場合は、重複排除の処理を制御するために insert_deduplication_token 設定を使用する必要があります。この設定では、各 insert に対して一意のトークンを指定でき、ClickHouse はそれを使用してデータが重複かどうかを判定します。insert_deduplication_token の優先度が高く、トークンが指定されている場合、ClickHouse はデータのハッシュ値を使用しません。

INSERT ... VALUES クエリでは、挿入されるデータのブロックへの分割は決定論的であり、設定によって決まります。したがって、挿入を再試行する際は、初回の操作と同じ設定値を使用する必要があります。

INSERT ... SELECT クエリでは、SELECT 部分が試行のたびに同じデータを同じ順序で返す必要があります。そうでない場合、ブロックと block_id が異なるため、再試行は重複として認識されません。

ClickHouse はログソースデータが変更されていないことを検証できませんが、クエリ自体が再現可能な結果を生成するかどうかは確認できます。次の両方を満たす場合、SELECT は安定していると見なされます。

• クエリに ORDER BY ALL 句が含まれている。認識されるのはリテラルの ORDER BY ALL のみです。通常の ORDER BY <expressions> は認識されず、2 つ以上の SELECT を UNION したクエリは安定とは見なされません。
• 読み取りパイプラインが単一ストリームで終了する。

空でない insert_deduplication_token は安定性の代替手段として同等です。この場合、データではなくトークンによって insert が識別されます。

設定 deduplicate_insert_select で動作を選択します。

enable_when_possible と enable_even_for_bad_queries は deduplicate_insert の設定も考慮します。これが disable の場合、クエリは重複排除されません。force_enable は deduplicate_insert をオーバーライドします。

選択したテーブルは再試行の間に更新される可能性があることに注意してください。この場合、2 つの方法は逆の動作をします。

• insert_deduplication_token がない場合、block_id はデータから計算されます。結果が変更されると異なる block_id が生成され、重複排除は行われません。そのため、再試行では最初の試行で書き込まれたデータに加えて、新しいデータが挿入されます。
• insert_deduplication_token がある場合、トークンだけで insert が識別されます。異なるデータを挿入することになった場合でも、再試行は重複として認識され、破棄されます。

再試行にどのような意味を持たせたいかに応じて、適切な方法を選択してください。また、大量のデータを挿入すると、ブロック数が重複排除ログウィンドウの上限を超えることがあり、その場合 ClickHouse はブロックを重複排除すべきか判断できなくなります。

非同期挿入 (バージョン 26.2 以降はデフォルトで有効な async_insert) では、同期挿入と同様に、再試行時に重複排除が行われます。deduplicate_insert は両方を制御するため、個別の設定は不要です。

この 2 種類の挿入は同じ重複排除ログを共有し、block_id も同じ方法で計算します。そのため、重複排除を損なうことなくクライアントを同期挿入と非同期挿入の間で切り替えられます。また、一方のモードで送信した再試行も、もう一方のモードで送信した試行の重複として認識されます。重複排除に依存するテーブルでも、ワークロードを同期挿入から非同期挿入へ安全に移行できます。

サーバーは複数の非同期挿入を1つのバッチにまとめ、そのバッチを1つ以上のパーツとして書き込みます。パーティションキーの値ごとに、少なくとも1つのパーツが作成されます。重複排除はバッチ単位ではなく、ユーザークエリ単位で行われます。

• キューに入れられた各クエリは、バッチに1つの重複排除トークンを追加します。
• トークンは、クエリで insert_deduplication_token が指定されている場合はその値、指定されていない場合はそのクエリが追加した行のハッシュです。
• バッチ化はトークンに影響せず、insert_deduplication_token はクエリのバッチへのグループ化方法に影響しません。

これには2つの結果があります。

• バッチ内の1つのクエリが重複している場合、ClickHouse はそのクエリの行のみを削除します。バッチ内の残りの行は通常どおり挿入されます。パーツ内のすべての行が削除される場合にのみ、そのパーツ全体がスキップされます。
• 同じバッチ内の2つのクエリが同じトークンを持つ場合、2つ目のクエリはパーツが書き込まれる前に破棄されます。これはパーティションごとに適用されます。2つのクエリが異なるパーティションに行を書き込む場合、両方とも保持されます。

system.events の DuplicatedAsyncInserts および SelfDuplicatedAsyncInserts イベントは、これら2つのケースをカウントします。

非同期挿入の重複排除は、依存するmaterialized viewと連動して機能します。ルールは単純です。1つのブロックが入力されると、1つのブロックが出力されます。ビューの内部クエリが1つの入力ブロックを1つの出力ブロックに変換する場合、重複排除は機能します。ビューが2つ目のブロックを出力すると、ClickHouseはNOT_IMPLEMENTED例外をスローします。

ビューの出力が1つのブロックに収まらない場合、2つ目のブロックが出力されます。max_block_sizeは、1つのブロックに収まる行数を設定します。カラム変換、フィルタリング、集約で行数が増えることはないため、常に1つのブロックに収まります。JOINでは行数が増える場合があります。結果がmax_block_size以下であれば機能しますが、それを超えると失敗します。

複数のブロックを出力するビューを介して挿入するには、deduplicate_blocks_in_dependent_materialized_views = 0を設定するか、同期挿入を使用します。

テーブルに 1 つ以上の materialized view がある場合、挿入されたデータは、定義された変換を適用したうえで、それらの view の宛先にも挿入されます。変換後のデータも、再試行時には重複排除されます。ClickHouse は、materialized view に対する重複排除を、ターゲットテーブルに挿入されたデータを重複排除するのと同じ方法で実行します。

このプロセスは、ソーステーブルに対する次の設定で制御できます。

• replicated_deduplication_window
• replicated_deduplication_window_seconds
• non_replicated_deduplication_window

materialized view 配下のテーブルでの重複排除は、ユーザープロファイル設定 deduplicate_blocks_in_dependent_materialized_views によっても制御され、バージョン 26.2 以降はデフォルトで有効になっています。両方の設定で許可する必要があります。deduplicate_insert はソーステーブルに挿入されたデータを重複排除し、deduplicate_blocks_in_dependent_materialized_views は依存先テーブル内のデータも追加で重複排除します。完全な重複排除を行う場合は、両方を有効にしてください。

materialized view 配下のテーブルにブロックを挿入する際、ClickHouse はソーステーブルの block_id と追加の識別子を組み合わせた文字列をハッシュ化して block_id を計算します。これにより、materialized view 内で正確な重複排除が保証され、materialized view 配下の宛先テーブルに到達する前にどのような変換が適用されたかにかかわらず、元の挿入に基づいてデータを区別できるようになります。

materialized view 内の変換処理で生成された同一ブロックは、基になっている挿入データが異なるため、重複排除されません。

以下に例を示します。

CREATE TABLE dst
(
    `key` Int64,
    `value` String
)
ENGINE = MergeTree
ORDER BY tuple()
SETTINGS non_replicated_deduplication_window=1000;

CREATE MATERIALIZED VIEW mv_dst
(
    `key` Int64,
    `value` String
)
ENGINE = MergeTree
ORDER BY tuple()
SETTINGS non_replicated_deduplication_window=1000
AS SELECT
    0 AS key,
    value AS value
FROM dst;

SET max_block_size=1;
SET min_insert_block_size_rows=0;
SET min_insert_block_size_bytes=0;

上記の設定により、1行しか含まないブロックが連続するテーブルから選択できるようになります。これらの小さなブロックはまとめられず、テーブルに挿入されるまでそのままの状態が保たれます。

デフォルトで有効になっていますが、materialized view での重複排除を明示的に指定します:

SET deduplicate_blocks_in_dependent_materialized_views=1;

INSERT INTO dst SELECT
    number + 1 AS key,
    IF(key = 0, 'A', 'B') AS value
FROM numbers(2);

SELECT
    *,
    _part
FROM dst
ORDER BY all;

┌─key─┬─value─┬─_part─────┐
│   1 │ B     │ all_0_0_0 │
│   2 │ B     │ all_1_1_0 │
└─────┴───────┴───────────┘

ここでは、2 つのパーツが dst テーブルに挿入されていることがわかります。select からの 2 つのブロック – INSERT 時の 2 つのパーツです。各パーツには異なるデータが含まれています。

SELECT
    *,
    _part
FROM mv_dst
ORDER BY all;

┌─key─┬─value─┬─_part─────┐
│   0 │ B     │ all_0_0_0 │
│   0 │ B     │ all_1_1_0 │
└─────┴───────┴───────────┘

ここでは、mv_dst テーブルに 2 つのパーツが挿入されていることがわかります。これらのパーツには同じデータが含まれていますが、重複排除されていません。

INSERT INTO dst SELECT
    number + 1 AS key,
    IF(key = 0, 'A', 'B') AS value
FROM numbers(2);

SELECT
    *,
    _part
FROM dst
ORDER BY all;

┌─key─┬─value─┬─_part─────┐
│   1 │ B     │ all_0_0_0 │
│   2 │ B     │ all_1_1_0 │
└─────┴───────┴───────────┘

SELECT
    *,
    _part
FROM mv_dst
ORDER by all;

┌─key─┬─value─┬─_part─────┐
│   0 │ B     │ all_0_0_0 │
│   0 │ B     │ all_1_1_0 │
└─────┴───────┴───────────┘

ここでは、insert を再試行すると、すべてのデータが重複排除されることがわかります。重複排除は dst テーブルと mv_dst テーブルの両方で有効です。

CREATE TABLE dst
(
    `key` Int64,
    `value` String
)
ENGINE = MergeTree
ORDER BY tuple()
SETTINGS non_replicated_deduplication_window=1000;

SET max_block_size=1;
SET min_insert_block_size_rows=0;
SET min_insert_block_size_bytes=0;

挿入:

INSERT INTO dst SELECT
    0 AS key,
    'A' AS value
FROM numbers(2);

SELECT
    'from dst',
    *,
    _part
FROM dst
ORDER BY all;

┌─'from dst'─┬─key─┬─value─┬─_part─────┐
│ from dst   │   0 │ A     │ all_0_0_0 │
└────────────┴─────┴───────┴───────────┘

上記の設定では、select– の結果として 2 つのブロックが生成されるため、table dst への挿入用にも 2 つのブロックがあるはずです。しかし実際には、table dst に挿入されたのは 1 つのブロックだけであることがわかります。これは、2 つ目のブロックが重複排除されたためです。このブロックは同じデータを持ち、さらに挿入されたデータからハッシュとして計算される重複排除用の秘密鍵 block_id も同一です。この動作は想定どおりではありません。このようなケースが発生することはまれですが、理論上は起こりえます。このようなケースを正しく処理するには、ユーザーが insert_deduplication_token を指定する必要があります。以下の例でこれを修正してみましょう。

CREATE TABLE dst
(
    `key` Int64,
    `value` String
)
ENGINE = MergeTree
ORDER BY tuple()
SETTINGS non_replicated_deduplication_window=1000;

SET max_block_size=1;
SET min_insert_block_size_rows=0;
SET min_insert_block_size_bytes=0;

データの挿入:

INSERT INTO dst SELECT
    0 AS key,
    'A' AS value
FROM numbers(2)
SETTINGS insert_deduplication_token='some_user_token';

SELECT
    'from dst',
    *,
    _part
FROM dst
ORDER BY all;

┌─'from dst'─┬─key─┬─value─┬─_part─────┐
│ from dst   │   0 │ A     │ all_2_2_0 │
│ from dst   │   0 │ A     │ all_3_3_0 │
└────────────┴─────┴───────┴───────────┘

想定どおり、同一の2つのブロックが挿入されました。

SELECT 'second attempt';

INSERT INTO dst SELECT
    0 AS key,
    'A' AS value
FROM numbers(2)
SETTINGS insert_deduplication_token='some_user_token';

SELECT
    'from dst',
    *,
    _part
FROM dst
ORDER BY all;

┌─'from dst'─┬─key─┬─value─┬─_part─────┐
│ from dst   │   0 │ A     │ all_2_2_0 │
│ from dst   │   0 │ A     │ all_3_3_0 │
└────────────┴─────┴───────┴───────────┘

再試行した挿入は、想定どおり重複排除されます。

SELECT 'third attempt';

INSERT INTO dst SELECT
    1 AS key,
    'b' AS value
FROM numbers(2)
SETTINGS insert_deduplication_token='some_user_token';

SELECT
    'from dst',
    *,
    _part
FROM dst
ORDER BY all;

┌─'from dst'─┬─key─┬─value─┬─_part─────┐
│ from dst   │   0 │ A     │ all_2_2_0 │
│ from dst   │   0 │ A     │ all_3_3_0 │
└────────────┴─────┴───────┴───────────┘

その挿入も、挿入されたデータが異なっていても重複排除の対象になります。insert_deduplication_token のほうが優先される点に注意してください。insert_deduplication_token が指定されている場合、ClickHouse はデータのハッシュ値を使用しません。

CREATE TABLE dst
(
    `key` Int64,
    `value` String
)
ENGINE = MergeTree
ORDER BY tuple()
SETTINGS non_replicated_deduplication_window=1000;

CREATE MATERIALIZED VIEW mv_dst
(
    `key` Int64,
    `value` String
)
ENGINE = MergeTree
ORDER BY tuple()
SETTINGS non_replicated_deduplication_window=1000
AS SELECT
    0 AS key,
    value AS value
FROM dst;

SET deduplicate_blocks_in_dependent_materialized_views=1;

select 'first attempt';

INSERT INTO dst VALUES (1, 'A');

SELECT
    'from dst',
    *,
    _part
FROM dst
ORDER by all;

┌─'from dst'─┬─key─┬─value─┬─_part─────┐
│ from dst   │   1 │ A     │ all_0_0_0 │
└────────────┴─────┴───────┴───────────┘

SELECT
    'from mv_dst',
    *,
    _part
FROM mv_dst
ORDER by all;

┌─'from mv_dst'─┬─key─┬─value─┬─_part─────┐
│ from mv_dst   │   0 │ A     │ all_0_0_0 │
└───────────────┴─────┴───────┴───────────┘

select 'second attempt';

INSERT INTO dst VALUES (2, 'A');

SELECT
    'from dst',
    *,
    _part
FROM dst
ORDER by all;

┌─'from dst'─┬─key─┬─value─┬─_part─────┐
│ from dst   │   1 │ A     │ all_0_0_0 │
│ from dst   │   2 │ A     │ all_1_1_0 │
└────────────┴─────┴───────┴───────────┘

SELECT
    'from mv_dst',
    *,
    _part
FROM mv_dst
ORDER by all;

┌─'from mv_dst'─┬─key─┬─value─┬─_part─────┐
│ from mv_dst   │   0 │ A     │ all_0_0_0 │
│ from mv_dst   │   0 │ A     │ all_1_1_0 │
└───────────────┴─────┴───────┴───────────┘

毎回異なるデータを挿入します。しかし、mv_dst テーブルには毎回同じデータが挿入されます。ソースデータが異なるため、データは重複排除されません。

CREATE TABLE dst
(
    `key` Int64,
    `value` String
)
ENGINE = MergeTree
ORDER BY tuple()
SETTINGS non_replicated_deduplication_window=1000;

CREATE TABLE mv_dst
(
    `key` Int64,
    `value` String
)
ENGINE = MergeTree
ORDER BY tuple()
SETTINGS non_replicated_deduplication_window=1000;

CREATE MATERIALIZED VIEW mv_first
TO mv_dst
AS SELECT
    0 AS key,
    value AS value
FROM dst;

CREATE MATERIALIZED VIEW mv_second
TO mv_dst
AS SELECT
    0 AS key,
    value AS value
FROM dst;

SET deduplicate_blocks_in_dependent_materialized_views=1;

select 'first attempt';

INSERT INTO dst VALUES (1, 'A');

SELECT
    'from dst',
    *,
    _part
FROM dst
ORDER by all;

┌─'from dst'─┬─key─┬─value─┬─_part─────┐
│ from dst   │   1 │ A     │ all_0_0_0 │
└────────────┴─────┴───────┴───────────┘

SELECT
    'from mv_dst',
    *,
    _part
FROM mv_dst
ORDER by all;

┌─'from mv_dst'─┬─key─┬─value─┬─_part─────┐
│ from mv_dst   │   0 │ A     │ all_0_0_0 │
│ from mv_dst   │   0 │ A     │ all_1_1_0 │
└───────────────┴─────┴───────┴───────────┘

同じ内容の2つのブロックがテーブル mv_dst に挿入されました (予想どおり) 。

SELECT 'second attempt';

INSERT INTO dst VALUES (1, 'A');

SELECT
    'from dst',
    *,
    _part
FROM dst
ORDER BY all;

┌─'from dst'─┬─key─┬─value─┬─_part─────┐
│ from dst   │   1 │ A     │ all_0_0_0 │
└────────────┴─────┴───────┴───────────┘

SELECT
    'from mv_dst',
    *,
    _part
FROM mv_dst
ORDER by all;

┌─'from mv_dst'─┬─key─┬─value─┬─_part─────┐
│ from mv_dst   │   0 │ A     │ all_0_0_0 │
│ from mv_dst   │   0 │ A     │ all_1_1_0 │
└───────────────┴─────┴───────┴───────────┘

その再試行操作は、テーブル dst と mv_dst の両方で重複排除されます。