[WIP] CAS draft (adopting to CI/CD, not for review / merge)

.gitignore

@@ -198,3 +198,6 @@ tests/casa_del_dolor/_instances*
 ci/praktika/json_test.html
 ci/praktika/json_test.html.gz
 ci/praktika/json.html.gz
+
+# SDD scratch (briefs, reports, review diffs, progress ledger) — never committed
+.superpowers/

ci/defs/job_configs.py

@@ -777,6 +777,22 @@ class JobConfigs:
             runs_on=RunnerLabels.ARM_SMALL,
             requires=[ArtifactNames.CH_ARM_BINARY_GH],
         ),
+        # Content-addressed disk as the default MergeTree storage (CAS M6 drop-in). Mirrors the
+        # "s3 storage" variant but installs only content_addressed_storage_policy_for_merge_tree_by_default.xml.
+        Job.ParamSet(
+            parameter="arm_binary, content_addressed storage, parallel",
+            runs_on=RunnerLabels.ARM_MEDIUM_CPU,
+            requires=[ArtifactNames.CH_ARM_BINARY],
+        ),
+        # Content-addressed disk over S3 (minio) as the default MergeTree storage — the real-S3
+        # (north star) counterpart of the local-object-storage variant above. minio is started
+        # by the stateless job itself; installs only
+        # content_addressed_s3_storage_policy_for_merge_tree_by_default.xml.
+        Job.ParamSet(
+            parameter="arm_binary, content_addressed s3 storage, parallel",
+            runs_on=RunnerLabels.ARM_MEDIUM_CPU,
+            requires=[ArtifactNames.CH_ARM_BINARY],
+        ),
     )
     functional_tests_jobs_coverage = common_ft_job_config.parametrize(
         *[

ci/jobs/functional_tests.py

@@ -142,6 +142,8 @@ def run_tests(
     "old analyzer": "--analyzer",
     "WasmEdge": "--wasm-engine wasmedge",
     "s3 storage": "--s3-storage",
+    "content_addressed storage": "--content-addressed-storage",
+    "content_addressed s3 storage": "--content-addressed-s3-storage",
     "DatabaseReplicated": "--db-replicated",
     "DatabaseOrdinary": "--db-ordinary",
     "wide parts enabled": "--wide-parts",
@@ -155,6 +157,8 @@ def run_tests(
 
 OPTIONS_TO_TEST_RUNNER_ARGUMENTS = {
     "s3 storage": "--s3-storage --no-stateful",
+    "content_addressed storage": "--content-addressed-storage",
+    "content_addressed s3 storage": "--content-addressed-s3-storage",
     "ParallelReplicas": "--no-zookeeper --no-shard --no-parallel-replicas",
     "AsyncInsert": " --no-async-insert",
     "DatabaseReplicated": " --no-stateful --replicated-database",
@@ -241,6 +245,7 @@ def main():
     is_targeted_check = False
     is_bugfix_validation = False
     is_s3_storage = False
+    is_content_addressed_s3 = False
     is_azure_storage = False
     is_database_replicated = False
     is_shared_catalog = False
@@ -294,8 +299,13 @@ def main():
             is_excluded_from_llvm = True
         if "per_test_coverage" in to:
             is_per_test_coverage = True
-        if "s3 storage" in to:
+        if "s3 storage" in to and "content_addressed" not in to:
+            # The content-addressed-over-s3 variant ("content_addressed s3 storage") installs
+            # only its own default policy and must not pull in the s3 stateful-data / encrypted
+            # storage machinery, so it is deliberately excluded from is_s3_storage.
             is_s3_storage = True
+        if "content_addressed s3 storage" in to:
+            is_content_addressed_s3 = True
         if "azure" in to:
             is_azure_storage = True
         if "DatabaseReplicated" in to:
@@ -636,6 +646,14 @@ def main():
 
         def start():
             res = CH.start_minio(test_type="stateless") and CH.start_azurite()
+            if res and is_content_addressed_s3:
+                # The CA-over-S3 pool lives on RustFS (M-W D-W8): the incarnation pool
+                # needs ENFORCED conditional deletes, which MinIO OSS lacks (the
+                # fail-closed capability probe rejects it). start_rustfs wipes its data
+                # dir per run, so no pool state bleeds between runs (the local-CA
+                # analogue is the per-run server-store wipe). MinIO keeps the non-CA
+                # s3 disks.
+                res = CH.start_rustfs()
             res = res and CH.start()
             res = res and CH.wait_ready()
             if res:

ci/jobs/scripts/check_style/check_cpp.sh

@@ -462,6 +462,17 @@ xargs < "$STYLE_TMPDIR/all_excluded" rg -n '\bassert[[:space:]]*\(' |
     echo "Use chassert instead of assert"
 } > "$O.18" 2>&1 &
 
+# 19: CAS wiring must not mutate committed refs through raw Cas::Store / Cas::Build (spec
+# docs/superpowers/specs/2026-07-08-cas-part-folder-cache-design.md): committed part-ref mutations
+# go through CachedPartFolderAccess (dot-syntax on the partAccess() reference). Best-effort textual
+# guard: Core/ (the protocol implementation) and the facade itself are exempt.
+{
+find $ROOT_PATH/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed -maxdepth 1 \( -name '*.h' -or -name '*.cpp' \) 2>/dev/null |
+    grep -v 'CachedPartFolderAccess' |
+    xargs -r rg -n -e '->(dropRef|updateRefPayload|dropNamespace|promote)\(' &&
+    echo "Committed part-ref mutations in CAS wiring must go through CachedPartFolderAccess (docs/superpowers/specs/2026-07-08-cas-part-folder-cache-design.md)"
+} > "$O.19" 2>&1 &
+
 # Wait for all parallel checks to complete, then output results in order
 wait
 cat "$O".* 2>/dev/null

ci/jobs/scripts/clickhouse_proc.py

@@ -163,6 +163,56 @@ def start_minio(self, test_type):
         print("Failed to start minio")
         return False
 
+    def start_rustfs(self):
+        # RustFS backs the content-addressed-over-S3 pool (M-W D-W8): the incarnation pool needs
+        # ENFORCED conditional operations (a wrong-token DELETE must fail with 412), which MinIO
+        # OSS lacks - the CA disk's fail-closed capability probe rejects it. The static binary
+        # lives at ci/tmp/rustfs (extracted from rustfs/rustfs:1.0.0-beta.8); the data dir is
+        # wiped per run so no pool state bleeds between runs. MinIO keeps serving the non-CA s3
+        # disks on its own port.
+        rustfs_bin = f"{temp_dir}/rustfs"
+        if not Path(rustfs_bin).is_file():
+            print(f"rustfs binary not found at {rustfs_bin}")
+            return False
+        data_dir = f"{temp_dir}/rustfs_data"
+        Shell.check(f"rm -rf {data_dir} && mkdir -p {data_dir}", verbose=True)
+        # Disable the background data-scanner and auto-heal manager (B93). On a single-disk
+        # ephemeral test pool they do no useful work (no parity to heal from; data-usage/bitrot
+        # accounting on throwaway data is pointless) but their cost is NOT free: the scanner walks
+        # the whole object tree (CPU grows with object count - observed ~120% sustained after the
+        # pool reached ~177k objects in a full lane) and the heal manager's 10s auto-disk-scan
+        # takes namespace locks, producing periodic multi-minute bursts of 503 ServiceUnavailable
+        # (NamespaceLockQuorumUnavailable) that stalled client I/O and caused ~50 test timeouts.
+        # Client GET/PUT/LIST/DELETE do not depend on either service, so disabling them is safe.
+        # Env names: RUSTFS_SCANNER_ENABLED/RUSTFS_HEAL_ENABLED (the RUSTFS_ENABLE_* forms are
+        # deprecated in 1.0.0-beta.8).
+        command = (
+            f"RUSTFS_SCANNER_ENABLED=false RUSTFS_HEAL_ENABLED=false "
+            f"{rustfs_bin} server --address 0.0.0.0:11121 "
+            f"--access-key clickhouse --secret-key clickhouse {data_dir}"
+        )
+        with open(f"{temp_dir}/rustfs.log", "w") as log_file:
+            self.rustfs_proc = subprocess.Popen(
+                command, stdout=log_file, stderr=subprocess.STDOUT, shell=True
+            )
+        print(f"Started rustfs asynchronously with PID {self.rustfs_proc.pid}")
+
+        if not Shell.check(
+            "curl -s -o /dev/null -w '%{http_code}' http://127.0.0.1:11121/ | grep -qE '403|200'",
+            verbose=False,
+            retries=6,
+        ):
+            print("Failed to start rustfs")
+            return False
+        # The `test` bucket the storage policy expects.
+        res = Shell.check(
+            "/mc alias set carustfs http://localhost:11121 clickhouse clickhouse && /mc mb --ignore-existing carustfs/test",
+            verbose=True,
+        )
+        if not res:
+            print("Failed to create rustfs test bucket")
+        return res
+
     def start_azurite(self):
         # Raise the open files limit before launching azurite-rs.
         # Each concurrent test query opens a TCP connection plus an in-memory

contrib/CMakeLists.txt

@@ -236,6 +236,10 @@ add_contrib(simdcomp-cmake simdcomp)
 
 add_contrib (numactl-cmake numactl)
 
+# CRC32C: a tiny, dependency-free checksum library. Built unconditionally because the content-addressed
+# storage core (CasRunFile) uses it for per-block integrity; google-cloud-cpp also depends on it.
+add_contrib (crc32c-cmake crc32c)
+
 # Google Cloud Cpp
 if(ENABLE_LIBRARIES AND (OS_LINUX OR OS_DARWIN OR OS_FREEBSD) AND (ARCH_AMD64 OR (ARCH_AARCH64 AND NOT NO_ARMV81_OR_HIGHER)))
     set(ENABLE_GOOGLE_CLOUD_CPP_DEFAULT ON)
@@ -245,7 +249,6 @@ endif()
 option(ENABLE_GOOGLE_CLOUD_CPP "Enable Google Cloud Cpp" ${ENABLE_GOOGLE_CLOUD_CPP_DEFAULT})
 if(ENABLE_GOOGLE_CLOUD_CPP)
     add_contrib (nlohmann-json-cmake nlohmann-json)
-    add_contrib (crc32c-cmake crc32c)
     add_contrib (google-cloud-cpp-cmake google-cloud-cpp) # requires grpc, protobuf, absl, nlohmann's json, crc32c
 else()
     message(STATUS "Not using Google Cloud Cpp")

docs/en/operations/system-tables/content_addressed_garbage_collection_log.md

@@ -0,0 +1,67 @@
+---
+description: 'System table containing per-round records of the content-addressed (CA) MergeTree garbage collector.'
+sidebar_label: 'content_addressed_garbage_collection_log'
+sidebar_position: 30
+slug: /operations/system-tables/content_addressed_garbage_collection_log
+title: 'system.content_addressed_garbage_collection_log'
+doc_type: 'reference'
+---
+
+## Description {#description}
+
+The `system.content_addressed_garbage_collection_log` table contains per-round records of the
+content-addressed (CA) MergeTree garbage collector. For every garbage-collection round it stores a
+`Start` row and a `Finish` row (like `system.part_log` stores events per data part), with the counts
+of objects marked and deleted, the round duration, the outcome, and a per-round `ProfileEvents`
+delta.
+
+Rounds are emitted both by the background GC scheduler (`trigger = 'Scheduled'`) and by the
+synchronous [`SYSTEM CONTENT ADDRESSED GARBAGE COLLECTION`](/sql-reference/statements/system#content-addressed-garbage-collection)
+command (`trigger = 'Manual'`).
+
+The table is created only if the `content_addressed_garbage_collection_log` server setting is
+specified (it is enabled by default in the shipped `config.xml`).
+
+## Columns {#columns}
+
+- `hostname` ([LowCardinality(String)](/sql-reference/data-types/lowcardinality)) — Host name of the server executing the round.
+- `event_date` ([Date](/sql-reference/data-types/date)) — Event date.
+- `event_time` ([DateTime](/sql-reference/data-types/datetime)) — Event time.
+- `event_time_microseconds` ([DateTime64(6)](/sql-reference/data-types/datetime64)) — Event time with microseconds precision.
+- `event_type` ([Enum8](/sql-reference/data-types/enum)) — `Start` or `Finish` of a GC round.
+- `disk_name` ([LowCardinality(String)](/sql-reference/data-types/lowcardinality)) — The content-addressed disk (pool) the round ran on.
+- `gc_id` ([String](/sql-reference/data-types/string)) — The GC scheduler instance id (which mounter ran the round).
+- `trigger` ([Enum8](/sql-reference/data-types/enum)) — `Scheduled` (background tick) or `Manual` (`SYSTEM` command).
+- `round` ([UInt64](/sql-reference/data-types/int-uint)) — The GC round number (`0` on a `Start` row).
+- `outcome` ([Enum8](/sql-reference/data-types/enum)) — `Success`, `NotALeader`, or `Error` (meaningful on a `Finish` row). `NotALeader` means another mounter held the GC lease; `Error` means the round threw.
+- `candidates_marked` ([UInt64](/sql-reference/data-types/int-uint)) — Objects retired (marked) this round.
+- `objects_deleted` ([UInt64](/sql-reference/data-types/int-uint)) — Objects physically deleted this round.
+- `objects_absent` ([UInt64](/sql-reference/data-types/int-uint)) — Retire candidates found already absent.
+- `objects_replaced` ([UInt64](/sql-reference/data-types/int-uint)) — `412`-saves: a resurrection won the race against the delete (a health metric).
+- `objects_spared` ([UInt64](/sql-reference/data-types/int-uint)) — Candidates spared because their in-degree was greater than zero at recheck.
+- `children_cascaded` ([UInt64](/sql-reference/data-types/int-uint)) — Child edges freed by the cascade this round.
+- `duration_ms` ([UInt64](/sql-reference/data-types/int-uint)) — The round wall-clock duration (on a `Finish` row).
+- `error` ([String](/sql-reference/data-types/string)) — The exception text when `outcome = 'Error'`.
+- `ProfileEvents` ([Map(LowCardinality(String), UInt64)](/sql-reference/data-types/map)) — The per-round `ProfileEvents` delta (the `Cas*` counters and S3/disk events for this round).
+
+## Example {#example}
+
+```sql
+SELECT
+    event_type,
+    disk_name,
+    trigger,
+    outcome,
+    candidates_marked,
+    objects_deleted,
+    duration_ms
+FROM system.content_addressed_garbage_collection_log
+ORDER BY event_time_microseconds DESC
+LIMIT 2
+FORMAT Vertical;
+```
+
+## See Also {#see-also}
+
+- [`SYSTEM CONTENT ADDRESSED GARBAGE COLLECTION`](/sql-reference/statements/system#content-addressed-garbage-collection) — run one GC round synchronously.
+- [`system.part_log`](/operations/system-tables/part_log) — the analogous per-part event log.

docs/en/sql-reference/statements/system.md

@@ -457,6 +457,18 @@ Wait until all asynchronously loading data parts of a table (outdated data parts
 SYSTEM WAIT LOADING PARTS [ON CLUSTER cluster_name] [db.]merge_tree_family_table_name
 ```
 
+### SYSTEM CONTENT ADDRESSED GARBAGE COLLECTION {#content-addressed-garbage-collection}
+
+Runs one garbage-collection round of the content-addressed (CA) MergeTree garbage collector synchronously and node-local: it reclaims content-addressed objects that are no longer referenced by any part. This is the on-demand counterpart of the background GC scheduler; it is useful for tests and diagnostics.
+
+```sql
+SYSTEM CONTENT ADDRESSED GARBAGE COLLECTION [ON CLUSTER cluster_name] [disk_name]
+```
+
+When `disk_name` is given, the round runs on that content-addressed disk only; targeting a non-content-addressed disk raises an exception. When `disk_name` is omitted, one round runs on every content-addressed disk configured on the node; if none are configured, the command raises an exception.
+
+Each round is recorded in [`system.content_addressed_garbage_collection_log`](/operations/system-tables/content_addressed_garbage_collection_log) as a `Start` and a `Finish` row (with `trigger = 'Manual'`).
+
 ## Managing ReplicatedMergeTree Tables {#managing-replicatedmergetree-tables}
 
 ClickHouse can manage background replication related processes in [ReplicatedMergeTree](/engines/table-engines/mergetree-family/replication) tables.