Update (2022-08-01): everything described below is outdated and irrelevant.

Requirements:

0. we want to delete data from MergeTree-family tables efficiently.
1. The cost of those deletes should be close to the cost of a simple SELECT, which matches the same conditions.

   SELECT count() WHERE <list of conditions>
   

   and

   DELETE ... WHERE <same list of conditions>
   

   should also take a similar amount of time.
2. we can have many DELETEs (same limitations on the number of DELETEs as for the number of INSERTs are ok).
3. DELETEs should be synchronous (on a single node), and eventual (go together with replication queue) in a replicated system
4. DELETEs should not reprocess significant amounts of data - we don't want to rewrite gigabytes of data to remove a single row.
   That means we need to store a mask of deleted rows instead of deleting them immediately. Real data deletion should happen eventually during normal background operations (merges etc.).
5. Once we calculate the mask of deleted rows (calculation can be expensive), we should use that mask in a very efficient/close to zero cost during further selects.

      SELECT count() FROM table;
      DELETE FROM table WHERE <conditions expensive to calculate>; <-- that may take a lot of time, it's ok
      SELECT count() FROM table; <-- that should be fast (i.e., we don't want to reapply the same conditions with every select)
   

6. data parts should be immutable during DELETEs - that is 'must-have' for replication.
7. standard SQL syntax

Concept

Generally, that means that we need

1. when DELETE fired - calculate a mask of deleted rows for every data part (for complicated conditions calculation can be costly)
2. instead of deleting anything, do an insert in some special place the following information

   data_part_name, mask_of_deleted_rows + may be some meta (like max block number)
   

3. selects / merges should check that 'special' place when selecting data, applying it as a row mask (applying the mask should be very cheap)
4. we need to run some upkeeping procedures that will
   • consolidate those masks in the background: if frequent DELETEs happen, we still want
     to keep a minimum number of masks - ideally one per part, and merge several masks into a single mask efficiently
   • remove the masks for parts that were already merged and deleted - to keep that 'special place' compact.

CRATE TABLE _deleted_rows_mask.my_table (
   part_name String,
   mask AggregateFunction(groupBitmap, UInt32) 
) engine=AggregatingMergeTree ORDER BY part_name;

INSERT INTO _deleted_rows_mask.my_table SELECT _part, groupBitmapState( _part_offset) FROM my_table WHERE ... 

SELECT * FROM my_table
WHERE (_part, _part_offset) NOT IN
(
   SELECT part_name, arrayJoin( bitmapToArray(  groupBitmapMerge( mask ) ) )
   FROM _deleted_rows_mask.my_table
   GROUP BY part_name
);

Implementation details (updated)

How mutations work

PARTITION_MINBLOCK_MAXBLOCK_LEVEL + mutation_XXX = PARTITION_MINBLOCK_MAXBLOCK_LEVEL_XXX

for example:
20100101_1_1_0 + mutation_2.txt = 20100101_1_1_0_2

Disadvantages:
Even a single row update require 'touching' all the parts (renaming them)
Creates a lot of inactive parts
Makes backups problematic.
ALTER DELETE is always IO (read/write) and CPU (unpack/pack) heavy

Lightweight delete:

get the number in every partition (Replicated) / for a table (nonReplicated) exactly the same as mutation.

Share numbering with mutations: so after mutation nr. 2, the lightweight mutation will get a number nr. 3.

Works very similar to mutation, and share some invariants, but the result of lightweight mutation is stored 'aside'.

20100101_1_1_0_2 + lightweight_mutation_3.txt = 20100101_1_1_0_2 + deleted_rows_masks_3.bin ( internally it store a mask for part 20100101_1_1_0_2 )

Obviously, masks need to be stored persistently. (fsyncs?)

So we will have one file deleted_rows_masks_3.bin (mutable, appendable) inside the table data folder for every 'lightweight_mutation'.

Lightweight_mutation mask for lightweight_mutations N+1 includes a mask for lightweight_mutations N (and all previous).

This way we single lightweight_mutation will write a single file only, and we can keep the last N files.

The last lightweight_mutation number is always stored in RAM together with other part metainformation.

The last mask may also be stored in RAM to make usage of it efficient (questionable - masks for large old parts can be big, so maybe smth like LRU cache + or some smart memory structures may be needed for that).

We also need a nice way to read deleted_rows_masks_3.bin for a particular part, so we may also need smth like deleted_rows_masks_3.idx

So the how that 'lightweight_mutation' work

1. DELETE FROM TABLE ... fired
2. we get the block number(s) for delete (just like with ALTER DELETE), let's say 100
3. work very similar to mutation (part by part), but instead of writing new parts - it
   produces deleted_rows_masks_100.bin
   • to achieve speed similar to SELECT - it may be needed to use max_threads here.
4. if we get part 'younger' than 100 (for example from another replica, from a merge scheduled before DELETE) we should
   calculate mask for it when the part is added to a working set.
5. merge on 2 parts ( 20100101_1_1_0 + 20100101_2_2_0 ) which have mutation id 3 and masks in
   deleted_rows_masks_3.bin will produce a part with masks applied, and number in the part name, i.e.
   20100101_1_1_0 + 20100101_2_2_0 + deleted_rows_masks_3.bin = 20100101_1_2_1_3 (and no mask).

Existing invariant for mutations:
'every part which has block numbers and mutation_nr younger than last mutation id considered unmutated' get changed to
'every part which has block numbers and max( mutation_nr || deleted_rows_masks_last_mutation ) younger than last mutation id considered unmutated'

In replicated:

1. every replica pays the same price for lightweight_mutation and applies it on its own set of parts.
   state of deleted_rows_masks_X.bin can differ on different replicas (and will differ, because set of parts on every replica can differ), but eventually, they will get in sync. (after all known parts with block numbers younger than mutation will be processed).

2. extend FETCH part to send also latest mask together with part. ( Alternatively: introduce deleted_rows_masks exchange, so if some replica has missing masks for some reason it can request from other replica masks for any of parts existing on that replica.)

Steps

A. Plain MergeTree implementation

1. Prepare simple in-memory storage for masks + API around (mask getter per part, mask setter per mutation_id)
2. parser for SQL deletes, write lightweight mutation to files (similar to mutationXXX.sql)
3. Create a mutation mode that uses lightweight_mutationXXX.sql and creates masks
4. Make MergeTreeReader aware of deleted rows bitmask on SELECT.
   <-- MVP (deletes w/o persistancy)
5. Add serialization for masks to file, pick the best format supporting appending, checksum validation, and indexing for a mask. (reuse the writers from Log family?), support fsync.
6. Cleanup of obsolete mask files
   <-- + persistancy
7. Review / adjust various places where part checks/manipulation happen (server start, merges, normal mutations, alters, etc).
   <-- cover complex scenarios
8. Consider in-memory optimization for mask storage (LRU cache with fixed size?)
9. Consider multithreaded execution of lightweight mutations
   <-- performance optimizations

B. ReplicatedMergeTree implementation

1. Extend replication queue with a new type of task.
2. Extend part sending to send also mask (preserve backward compatibility for older ClickHouse)
3. Review / adjust various places where part checks/manipulation happen (server start, merges, normal mutations, alters etc).
4. Extend part-exchange to be able to recover masks from other replicas.

(
   part_name String,
   part_state Enum8('active' = 0, 'eol' = 1),  /* once there is scheduled merge for a part we need to store it's state separately */
   delete_ids AggregateFunction(groupArray, UInt32) /* ids of all deletes / may be just max will be enough */
   mask AggregateFunction(groupBitmap, UInt32),
   deleted_timestamp SimpleAggregateFunction(max, DateTime),
) engine=AggregatingMergeTree
ORDER BY (part_name, part_state)
TTL deleted_timestamp + 1 HOUR

UPDATEs?

Executing a 'proper' update requires access to the previous state of the row, which may be not available (or can be outdated) due to eventual consistency on any concrete replica.

The quite easy thing is a special kind of UPDATE similar to UPSERT (internally DELETE as above + INSERT, w/o accessing the previous state of the row).

Alternatives:

• wait for n-1 replication state (finishing all merges/fetches etc assigned before the update) - it can be very long delays.
• apply on the fly till n-1 replication state will be reached. That may end up with quite strange behavior, when different rows are affected by same update depending on replication state. Questionable: i guess users expect smth similar to ACID rdbms