ClickHouse — A Columnar Analytics Database from the Inside

See Duplicates Before Merging and Handle Them with FINAL, argMax and CLEANUP

Goal

You put change records (updates and deletions) into a ReplacingMergeTree and confirm in numbers that old versions are visible before merging, how to hide them with FINAL and argMax, what merges and CLEANUP actually erase, and how the sorting key and resending change the result.

Why it matters

Production data moved to an analytics database keeps changing. ClickHouse handles updates as "new-version INSERT + merge later", and you cannot know when a merge will happen. So the same query is right at some times and wrong at others. In this lab you fix "before merging" with SYSTEM STOP MERGES to see that wrong answer with your own eyes, and learn two ways to produce the right answer at query time. The grader computes the expected values directly from the original generating expression, and confirms whether a merge actually happened from the records in system.part_log.

Steps

Create the database rmt and the table rmt.users — columns user_id UInt64, email String, plan LowCardinality(String), score UInt32, ver UInt32, is_deleted UInt8 (in this order), engine ReplacingMergeTree(ver, is_deleted), ORDER BY user_id.
Stop this table's merges with SYSTEM STOP MERGES rmt.users, then run /opt/lab/fixtures/replacing/changes.sql once (three INSERTs = three parts).
Create /root/ch/replacing/q_raw.sql, which counts all rows without FINAL, and /root/ch/replacing/q_final.sql, which counts with FROM rmt.users FINAL, and write the two results to /root/ch/replacing/counts.json as raw and final.
Create /root/ch/replacing/q_plan.sql, which gives the current number of users per plan without using FINAL — the result has two columns, plan and the number of users.
Create rmt.merged with the same columns and engine as rmt.users, move the rows of rmt.users into it with three INSERTs in the order ver = 1, ver = 2, ver = 3, and run OPTIMIZE TABLE rmt.merged FINAL; then write rows_after_merge (count()), deleted_rows_kept (the number of rows with is_deleted = 1) and final_count (the count with FINAL) to /root/ch/replacing/merged.json.
Create rmt.cleaned with the same columns and engine and the table setting allow_experimental_replacing_merge_with_cleanup = 1, insert three times the same way, and run OPTIMIZE TABLE rmt.cleaned FINAL CLEANUP; then write rows_after_cleanup and deleted_rows_left to /root/ch/replacing/cleanup.json.
Create rmt.bad with the same columns and engine but the sorting key ORDER BY (user_id, plan), insert three times the same way and run OPTIMIZE TABLE rmt.bad FINAL, then write good_final (the count of rmt.merged with FINAL), bad_final (the count of rmt.bad with FINAL) and bad_users (the number of distinct user_id in rmt.bad FINAL) to /root/ch/replacing/key.json.
Imitate a late resend — reinsert the rows of rmt.users with ver = 1 AND user_id <= 1000 into rmt.merged and rmt.cleaned each, and write the FINAL counts of the two tables and their difference to /root/ch/replacing/replay.json as merged_final, cleaned_final and resurrected.

Notes

The server is already running when the Pod starts. Just typing clickhouse-client connects. If it has stopped, run ch-up (if the server is restarted, STOP MERGES is released — start again from step 2).
SYSTEM STOP MERGES applies only to that table. The new tables in steps 5–7 have merges on, so OPTIMIZE works. If you issue OPTIMIZE on a stopped table, it is rejected with "Cancelled merging parts".
Merge records are left in system.part_log (event_type = 'MergeParts', with the merged part names in merged_from). If what you just did is not visible, SYSTEM FLUSH LOGS.
Common mistake: moving in steps 5–7 with a single INSERT ... SELECT * FROM rmt.users — duplicates inside one INSERT shrink at the moment of insertion (optimize_on_insert), leaving nothing to merge. Insert three times, by version.
To get numbers in JSON without quotes, --output_format_json_quote_64bit_integers 0.
Official docs: ReplacingMergeTree · Working with the ReplacingMergeTree engine · OPTIMIZE · system.part_log

A table with versions and deletion markers

Create the database rmt and the table rmt.users. The columns are, in order, user_id UInt64, email String, plan LowCardinality(String), score UInt32, ver UInt32, is_deleted UInt8, the engine is ReplacingMergeTree(ver, is_deleted), and the sorting key is ORDER BY user_id.

The first argument of the engine is the version column that decides which row wins, and the second is the column that tells whether the winning row is a deletion. The deletion-marker column cannot be used without the version column. The sorting key is the definition of "the same row", so put only identifiers that do not change.

Stop merges and insert three batches of change records

Stop this table's merges with SYSTEM STOP MERGES rmt.users, then run /opt/lab/fixtures/replacing/changes.sql once. The three INSERTs, initial load, update and deletion, must remain as three parts.

Merges happen at any time in the background, so if you leave it alone, the "before merging" state can vanish within a few seconds. You must stop them before inserting. If you already inserted, stop them, then TRUNCATE and insert again.

Counting without FINAL and counting with FINAL

Create /root/ch/replacing/q_raw.sql, which counts all rows of rmt.users without FINAL, and /root/ch/replacing/q_final.sql, which counts with FROM rmt.users FINAL, and write the results of the two queries to /root/ch/replacing/counts.json as raw and final.

FINAL applies the merge rule while querying (for the same sorting key, only the row with the larger ver, and if that row is a deletion marker, leave it out). The difference between the two numbers is the old-version rows and deletion rows added together. Do not attach a WHERE to q_final.sql.

The same answer without FINAL — argMax

Create /root/ch/replacing/q_plan.sql, which gives the current number of users per plan (plan) without using FINAL. The result has two columns, plan and the number of users, and deleted users must be left out.

If you just count with GROUP BY plan, you count the old versions and deletion rows too. First you have to reduce it to one row per user — group by user_id and use argMax(value, ver) to pick the plan and is_deleted of the row with the largest version, then leave out the deleted users on the outside and group by plan again.

Even after merging, the deletion rows remain

Create rmt.merged with the same columns and engine as rmt.users, move the rows of rmt.users into it with three INSERTs in the order ver = 1, ver = 2, ver = 3, and run OPTIMIZE TABLE rmt.merged FINAL; then write rows_after_merge (count()), deleted_rows_kept (the number of rows with is_deleted = 1) and final_count (the count with FINAL) to /root/ch/replacing/merged.json.

CREATE TABLE ... AS another_table copies the columns and engine but not the STOP MERGES state. See whether the row count after merging equals the number of users, and why the value counted without FINAL is still wrong. Whether a merge actually happened is left in system.part_log.

A CLEANUP merge erases even the deletion rows

Create rmt.cleaned with the same columns and engine as rmt.users and the table setting allow_experimental_replacing_merge_with_cleanup = 1, move the rows three times as in step 5, run OPTIMIZE TABLE rmt.cleaned FINAL CLEANUP, and write rows_after_cleanup (count()) and deleted_rows_left (the number of rows with is_deleted = 1) to /root/ch/replacing/cleanup.json.

You can attach SETTINGS after CREATE TABLE new_table AS another_table. If you issue CLEANUP without the setting, the server rejects it — erasing the deletion rows makes it impossible to block an old version that arrives later, so it is a feature you must turn on deliberately.

The sorting key decides "the same row"

Create rmt.bad with the same columns and engine but ORDER BY (user_id, plan), move the rows three times the same way and run OPTIMIZE TABLE rmt.bad FINAL, then write good_final (the count of rmt.merged with FINAL), bad_final (the count of rmt.bad with FINAL) and bad_users (the number of distinct user_id in rmt.bad FINAL) to /root/ch/replacing/key.json.

If plan is in the sorting key, the old row and the new row of a user whose plan changed have different keys and cannot become one group. A deletion row also covers only the group of the plan just before the deletion. See whether bad_final is larger than the number of users and whether bad_users is larger than the current number of users.

If old rows come again after CLEANUP

Reinsert the rows of rmt.users with ver = 1 AND user_id <= 1000 into rmt.merged and rmt.cleaned each, then write the FINAL counts of the two tables and their difference to /root/ch/replacing/replay.json as merged_final, cleaned_final and resurrected (cleaned_final − merged_final).

This is a situation where the pipeline resends an old batch. Think about what beats the old rows in the table where the deletion markers remain, and whom the old rows compete against in the table where the deletion rows were erased. The people who came back to life are the users among 1–1000 who had been deleted.