CAS: remove a part with one ref drop

docs/en/antalya/cas/architecture/part-lifecycle.md

@@ -128,6 +128,13 @@ Unchanged columns are adopted by hash through a tokenless evidence dependency wi
 `GET`; changed columns are fresh uploads. A repoint therefore costs zero bytes moved for the
 carry-forward portion of the file set — only the changed content re-uploads.
 
+## Part removal {#part-removal}
+
+Removing an outdated part from a `CAS` disk is one atomic ref-log record that drops the part's ref.
+`MergeTree` does not rename the part to `delete_tmp_<name>` first, as it does on other disks, and it
+does not unlink the part's files one by one. A removal costs one ref-log transaction and no manifest
+write, whatever the number of files or projections. A crash leaves either the part or no part.
+
 ## How each MergeTree operation maps {#operation-mapping}
 
 | Operation | CAS mechanics |
@@ -136,7 +143,8 @@ carry-forward portion of the file set — only the changed content re-uploads.
 | Merge | Identical for the output part. `<proj>.tmp_proj → <proj>.proj` is an entry-prefix re-key inside the staged manifest, not a rename |
 | Mutation | `createHardLink` per unchanged file: a source staged in *this* transaction copies the entry and its pending-blob record; a **committed** source records a tokenless evidence dependency with no `HEAD` and no `GET`. A mutation is a manifest rewrite where zero bytes move for the carry-forward |
 | `ALTER` / metadata rewrites | Standalone writes into a committed part, i.e. a repoint |
-| `DROP PART` | `removeDirectory` drops the ref and clears any per-file removal marks — one ref-drop, zero repoints |
+| Outdated-part removal | One ref drop, see [Part removal](#part-removal) |
+| `DROP PART` | The part becomes outdated and is removed as in [Part removal](#part-removal): one ref drop |
 | `DROP TABLE` / `DETACHED` / `UNFREEZE` | A namespace or prefixed-ref drop. Blobs are never deleted here — removal is pointer-unlink plus deferred `GC` |
 | `RENAME TABLE` | Republishes every ref and verbatim file into the new namespace, then drops the old one. Not atomic across namespaces, but idempotent and re-drivable — true atomicity would need a move journal and is out of scope |
 | `FREEZE` / `BACKUP` / `RESTORE` / cross-disk `MOVE` | Each wraps the whole clone in one disk transaction, because a CAS part is one atomic unit |

docs/en/antalya/cas/architecture/replication.md

@@ -141,7 +141,7 @@ but it remains under the root that created it: `<server_root_id>/shadow/<backup>
 check therefore attributes it to exactly that server root, under the same strict prefix rule as live
 content, and that root can confirm its exact refs.
 
-`DETACH`, `ATTACH`, `delete_tmp_` cleanup, and merge-result renames all reduce to the same two
+`DETACH`, `ATTACH`, and merge-result renames all reduce to the same two
 moves: re-key any *staged* source into the destination, then `republishRef(src → dst)` for any
 *committed* source. `republishRef` resolves the source ref freshly and reads its manifest through
 the manifest cache, publishes an

src/Storages/MergeTree/DataPartStorageOnDiskBase.cpp

@@ -928,6 +928,17 @@ void DataPartStorageOnDiskBase::remove(
     bool has_delete_prefix = part_dir_without_slash.filename().string().starts_with("delete_tmp_");
     std::optional<CanRemoveDescription> can_remove_description;
     auto disk = volume->getDisk();
+
+    /// On a content-addressed disk a part is one ref, and dropping it is one atomic ref-log record.
+    /// The `delete_tmp_` rename and the unlink batch below would each publish a new manifest first:
+    /// two manifest writes and five extra ref-log records per removed part, more with projections.
+    if (disk->isContentAddressed() && getParentDirectory() != MergeTreeData::DETACHED_DIR_NAME)
+    {
+        auto description = can_remove_callback();
+        disk->removeSharedRecursive(fs::path(from) / "", !description.can_remove_anything, description.files_not_to_remove);
+        return;
+    }
+
     fs::path to = fs::path(root_path) / part_dir_without_slash;
 
     if (!has_delete_prefix)

tests/queries/0_stateless/05054_cas_part_removal_one_ref_drop.reference

@@ -0,0 +1,8 @@
+plain	outdated_parts_present	1
+plain	outdated_parts_left	0
+plain	delete_tmp_events	0	ref_repoint	0	extra_manifest_publish	0	ref_drop_per_part	1	active_part_ref_drops	0
+plain	rows	400	1000
+projection	outdated_parts_present	1
+projection	outdated_parts_left	0
+projection	delete_tmp_events	0	ref_repoint	0	extra_manifest_publish	0	ref_drop_per_part	1	active_part_ref_drops	0
+projection	rows	400	1000

tests/queries/0_stateless/05054_cas_part_removal_one_ref_drop.sh

@@ -0,0 +1,97 @@
+#!/usr/bin/env bash
+# Tags: no-fasttest
+# ^ cas is an object-storage metadata type; keep it off the minimal fasttest image.
+
+# Removing an outdated part from a cas disk must be one ref drop: no `delete_tmp_` ref, no repoint of a
+# renamed ref, no manifest publish under a ref other than the parts' own. The oracle is `system.cas_log`,
+# selected by what the events name (the table's namespace and the part ref names the test knows), not by
+# time: audit events are timestamped at delivery and can arrive after the removal returns. The test
+# waits for the terminal records, one `ref_drop` per outdated part, before counting. Background GC is off
+# on the disk. The namespace filter drops the disk's mount and watermark events and needs a table with a
+# UUID (an `Atomic` or `Replicated` database).
+
+CUR_DIR=$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)
+# shellcheck source=../shell_config.sh
+. "$CUR_DIR"/../shell_config.sh
+
+DISK_NAME="${CLICKHOUSE_DATABASE}_05054_cas"
+DISK="disk(type = object_storage, object_storage_type = local, metadata_type = cas,
+    cas_server_root_id = '${CLICKHOUSE_DATABASE}_05054',
+    cas_gc_enabled = 0,
+    name = '${DISK_NAME}',
+    path = '${CLICKHOUSE_DATABASE}_05054_cas_pool/')"
+
+run_case()
+{
+    local name=$1
+    local projection=$2
+
+    ${CLICKHOUSE_CLIENT} -q "DROP TABLE IF EXISTS t_$name"
+    ${CLICKHOUSE_CLIENT} -q "CREATE TABLE t_$name (a UInt64, b UInt64 $projection) ENGINE = MergeTree ORDER BY a
+        SETTINGS disk = $DISK, old_parts_lifetime = 0, cleanup_delay_period = 1, max_cleanup_delay_period = 1, cleanup_delay_period_random_add = 0"
+    ${CLICKHOUSE_CLIENT} -q "SYSTEM STOP CLEANUP t_$name"
+    for i in 1 2 3 4
+    do
+        ${CLICKHOUSE_CLIENT} -q "INSERT INTO t_$name SELECT number, $i FROM numbers(100)"
+    done
+    ${CLICKHOUSE_CLIENT} -q "OPTIMIZE TABLE t_$name FINAL"
+
+    ${CLICKHOUSE_CLIENT} -q "SELECT '$name', 'outdated_parts_present', count() > 0 FROM system.parts
+        WHERE database = currentDatabase() AND table = 't_$name' AND NOT active"
+    local outdated_names all_names active_name table_uuid
+    outdated_names=$(${CLICKHOUSE_CLIENT} -q "SELECT groupArray(name) FROM system.parts
+        WHERE database = currentDatabase() AND table = 't_$name' AND NOT active")
+    all_names=$(${CLICKHOUSE_CLIENT} -q "SELECT groupArray(name) FROM system.parts
+        WHERE database = currentDatabase() AND table = 't_$name'")
+    active_name=$(${CLICKHOUSE_CLIENT} -q "SELECT name FROM system.parts
+        WHERE database = currentDatabase() AND table = 't_$name' AND active")
+    table_uuid=$(${CLICKHOUSE_CLIENT} -q "SELECT uuid FROM system.tables
+        WHERE database = currentDatabase() AND name = 't_$name'")
+    local outdated
+    outdated=$(${CLICKHOUSE_CLIENT} -q "SELECT length($outdated_names)")
+
+    ${CLICKHOUSE_CLIENT} -q "SYSTEM START CLEANUP t_$name"
+    local left
+    for _ in {1..120}
+    do
+        left=$(${CLICKHOUSE_CLIENT} -q "SELECT count() FROM system.parts
+            WHERE database = currentDatabase() AND table = 't_$name' AND NOT active")
+        [ "$left" = "0" ] && break
+        sleep 0.5
+    done
+    ${CLICKHOUSE_CLIENT} -q "SELECT '$name', 'outdated_parts_left', count() FROM system.parts
+        WHERE database = currentDatabase() AND table = 't_$name' AND NOT active"
+
+    # Audit events are delivered asynchronously: wait until the terminal record of every removal is there.
+    local drops
+    for _ in {1..120}
+    do
+        ${CLICKHOUSE_CLIENT} -q "SYSTEM FLUSH LOGS cas_log"
+        drops=$(${CLICKHOUSE_CLIENT} -q "SELECT countIf(event_type = 'ref_drop' AND has($outdated_names, ref_name))
+            FROM system.cas_log WHERE disk_name = '${DISK_NAME}' AND position(namespace, '$table_uuid') > 0")
+        [ "$drops" -ge "$outdated" ] && break
+        sleep 0.5
+    done
+
+    ${CLICKHOUSE_CLIENT} -q "
+        SELECT
+            '$name',
+            'delete_tmp_events', countIf(position(ref_name, 'delete_tmp_') > 0),
+            'ref_repoint', countIf(event_type = 'ref_repoint'),
+            'extra_manifest_publish', countIf(event_type = 'build_publish' AND NOT has($all_names, ref_name)),
+            'ref_drop_per_part', countIf(event_type = 'ref_drop' AND has($outdated_names, ref_name)) / $outdated,
+            'active_part_ref_drops', countIf(event_type = 'ref_drop' AND ref_name = '$active_name')
+        FROM system.cas_log
+        WHERE disk_name = '${DISK_NAME}' AND position(namespace, '$table_uuid') > 0"
+
+    ${CLICKHOUSE_CLIENT} -q "SELECT '$name', 'rows', count(), sum(b) FROM t_$name"
+
+    ${CLICKHOUSE_CLIENT} -q "DROP TABLE t_$name"
+}
+
+run_case plain ""
+run_case projection ", PROJECTION p (SELECT a, sum(b) GROUP BY a)"
+
+# FORGET logs an operator WARNING; the harness runs the client at --send_logs_level=warning, which would
+# stream that expected warning to stderr and be flagged as a failure. Suppress it for the FORGET call only.
+${CLICKHOUSE_CLIENT} --send_logs_level=fatal -q "SYSTEM CAS FORGET '${DISK_NAME}'"