CAS: packed parts

ci/defs/job_configs.py

@@ -233,8 +233,6 @@ class JobConfigs:
         runs_on=RunnerLabels.FUNC_TESTER_AMD,
         command="python3 ./ci/jobs/ci_tests_job.py",
         timeout=1200,
-        # NOTE (strtgbb): temp non-blocking — CH server exits 137 during CI Tests setup; dig deeper separately
-        allow_failure=True,
         run_in_docker=f"altinityinfra/integration-tests-runner+root+--privileged+--dns-search='.'+--security-opt seccomp=unconfined+--cap-add=SYS_PTRACE+{docker_sock_mount}+--volume=clickhouse_integration_tests_volume:/var/lib/docker+--cgroupns=host",
         digest_config=Job.CacheDigestConfig(
             include_paths=[

ci/jobs/scripts/clickhouse_proc.py

@@ -1641,8 +1641,6 @@ def set_random_timezone():
             res = ch.start_seaweedfs(param)
         elif command == "start_azurite":
             res = ch.start_azurite()
-        elif command == "start_rustfs":
-            res = ch.start_rustfs()
         else:
             raise ValueError(f"Unknown command: {command}")
     except Exception:

ci/jobs/scripts/stress/stress.py

@@ -244,12 +244,6 @@ def get_options(i: int, upgrade_check: bool, encrypted_storage: bool) -> str:
     options = []
     client_options = []
 
-    # Honour the CAS-as-default MergeTree policy so clickhouse-test skips
-    # `no-cas-storage` / `no-object-storage` / `no-s3-storage` tests. Without
-    # this flag those tests still run and fail against a CAS disk.
-    if os.environ.get("USE_CAS_S3_STORAGE_FOR_MERGE_TREE") == "1":
-        options.append("--cas-s3-storage")
-
     if upgrade_check:
         # Disable settings randomization for upgrade checks to prevent test failures caused by missing settings in old version
         options.append("--no-random-settings")

ci/jobs/stress_job.py

@@ -1,7 +1,6 @@
 import csv
 import logging
 import os
-import shutil
 import socket
 import sys
 from pathlib import Path
@@ -155,12 +154,7 @@ def get_additional_envs(info, check_name: str) -> List[str]:
             "AZURE_STORAGE_ACCOUNT_URL=$AZURE_STORAGE_ACCOUNT_URL",
         ])
 
-    # "cas s3" must win over the plain "s3" substring. Otherwise
-    # `USE_S3_STORAGE_FOR_MERGE_TREE=1` is also set and install.sh's
-    # if/elif chain installs the plain-S3 default policy instead of CAS.
-    if "cas s3" in check_name:
-        result.append("USE_CAS_S3_STORAGE_FOR_MERGE_TREE=1")
-    elif "s3" in check_name:
+    if "s3" in check_name:
         result.append("USE_S3_STORAGE_FOR_MERGE_TREE=1")
 
     result.append(
@@ -328,14 +322,6 @@ def run_stress_test(upgrade_check: bool = False) -> None:
 
     Utils.fix_ownership_after_docker(temp_path, docker_image)
 
-    # ci/tmp is not uploaded. Copying here rather than from stress_runner.sh runs
-    # whatever the container's exit code, so a job that aborts early still publishes
-    # rustfs.log -- the only record of what the pool side did behind a CAS disk.
-    for helper_log in ("rustfs.log", "minio.log", "azurite.log", "kafka.log"):
-        src = temp_path / helper_log
-        if src.is_file():
-            shutil.copy(src, result_path / helper_log)
-
     core_files = ClickHouseService.collect_cores(cores_path)
 
     is_oom = False

ci/tests/test_stress_job.py

@@ -36,7 +36,6 @@
 
 from ci.jobs.stress_job import (
     ensure_ripgrep_installed,
-    get_additional_envs,
     process_results,
     read_test_results,
     sanitize_test_result_line,
@@ -283,35 +282,5 @@ def test_dpkg_progress_in_info_does_not_split_row(tmp_path):
     assert malformed == []
 
 
-@pytest.mark.parametrize(
-    "check_name",
-    [
-        "Stress test (amd_debug, cas s3 storage)",
-        "Stress test (amd_asan_ubsan, cas s3 storage)",
-        "Stress test (amd_tsan, cas s3 storage)",
-        "Stress test (amd_msan, cas s3 storage)",
-    ],
-)
-def test_cas_s3_env_is_exclusive_of_plain_s3(monkeypatch, check_name):
-    monkeypatch.setattr("ci.jobs.ci_utils.is_extended_run", lambda: False)
-    envs = get_additional_envs(None, check_name)
-    assert "USE_CAS_S3_STORAGE_FOR_MERGE_TREE=1" in envs
-    assert "USE_S3_STORAGE_FOR_MERGE_TREE=1" not in envs
-
-
-def test_plain_s3_env_is_unchanged(monkeypatch):
-    monkeypatch.setattr("ci.jobs.ci_utils.is_extended_run", lambda: False)
-    envs = get_additional_envs(None, "Stress test (arm_asan_ubsan, s3)")
-    assert "USE_S3_STORAGE_FOR_MERGE_TREE=1" in envs
-    assert "USE_CAS_S3_STORAGE_FOR_MERGE_TREE=1" not in envs
-
-
-def test_default_stress_env_has_neither_s3_flag(monkeypatch):
-    monkeypatch.setattr("ci.jobs.ci_utils.is_extended_run", lambda: False)
-    envs = get_additional_envs(None, "Stress test (amd_asan_ubsan)")
-    assert "USE_S3_STORAGE_FOR_MERGE_TREE=1" not in envs
-    assert "USE_CAS_S3_STORAGE_FOR_MERGE_TREE=1" not in envs
-
-
 if __name__ == "__main__":
     sys.exit(pytest.main([__file__, "-v"]))

docs/en/antalya/cas/configuration.md

@@ -95,6 +95,7 @@ entirely before release. Treat this table as a snapshot of the current build, no
 | `cas_gc_interval_sec` | `60` | Seconds between background GC rounds (≥ 1) |
 | `cas_blob_hash` | `cityhash128` | Pool blob content-hash function (`cityhash128` \| `xxh3-128` \| `sha256`). Recorded in the pool at creation; a mismatching config is refused at mount |
 | `cas_blob_hash_allow_new` | `false` | Explicit opt-in to admit a new hash algorithm into an existing pool. One-way: once admitted, the pool carries both algorithms permanently |
+| `cas_allow_packed_parts` | `false` | Allow creation of new `Packed` MergeTree parts on this disk. The decision uses the actual destination disk, not every disk in its storage policy. A `type=cache` disk inherits this capability from the wrapped `CAS` disk; set the key in the `metadata_type=cas` disk block. Existing `Packed` parts remain readable when this setting is `false` |
 | `skip_access_check` | `false` | Skip the boot-time capability probe (start now, fix later). Only the preflight probe is skipped — the conditional-write correctness check still runs on every writable mount. **Not available on a writable generation-token (GCS) disk**, which refuses to mount with it: there, the probe battery is the only proof that a token-exact delete carries its generation precondition. Mount such a disk read-only if you need to defer the check |
 | `cas_mount_lease_ttl_ms` | `30000` | Milliseconds for which a mount lease remains valid after a successful claim or renewal (≥ 1). Lower values shorten stale-mount recovery but reduce tolerance for object-storage and scheduling delays |
 | `cas_mount_renew_period_ms` | `10000` | Milliseconds between background mount-lease renewals (≥ 1). It must leave enough time for two attempt envelopes (a renewal write and its settlement read) and the lease safety margin before the TTL expires: `period + 2 × envelope + margin < TTL` |
@@ -112,6 +113,47 @@ entirely before release. Treat this table as a snapshot of the current build, no
 | `cas_unsafe_remount_no_delay` | `0` | Reclaim a mount slot that carries this server's own uuid at once after a hard restart, without observing the slot's token for the lease TTL. Unsafe whenever two processes can hold the same `server_uuid` (a copied uuid file, a stalled predecessor). After such a reclaim the predecessor can still start conditional writes until its own cutoff (`confirmed deadline − cas_lease_safety_margin_ms − 2 × envelope`) or until its next renewal meets the token guard, and a request it already sent may still materialize later. That is not a data hazard: ref-log keys carry `(writer_epoch, sequence)` and creates are conditional, so two writers can never commit different bodies to one key, and recovery's epoch seal settles any straggler (recovery fails closed after 64 successive seal-create attempts displaced by newly materializing old-epoch transactions). The exposure is availability, not data. Intended for test stands and deployments that guarantee one process per uuid |
 | `cas_staging_backend` | `local` | Blob staging backend (`local` \| `s3`); `s3` is opt-in and requires native same-store copy on writable mount |
 
+### Enabling `Packed` parts {#enabling-packed-parts}
+
+Set `cas_allow_packed_parts` in the block of the `CAS` disk with `metadata_type=cas`. A
+`type=cache` disk above it inherits the capability from the wrapped `CAS` disk and cannot override
+it. The setting delegates the existing MergeTree part-storage choice to that disk. When it is
+`true`, the usual `min_bytes_for_full_part_storage` and
+`min_rows_for_full_part_storage` thresholds, together with `min_level_for_full_part_storage`, may
+select `Packed`; when it is `false`, newly written parts on that destination use `Full`. `INSERT`
+and merge first select a format from these thresholds and then apply the capability of the disk that
+actually receives the new part.
+
+A mutation inherits the complete format of its source part. After the destination disk is reserved,
+the result may change from `Packed -> Full` when that disk does not allow `Packed`. A mutation does
+not change `Full -> Packed`, either when thresholds change or when Packed capability is enabled; a
+later merge may select `Packed` from the new thresholds and destination capability.
+
+The setting controls creation, not readability. Existing `Packed` parts can still be opened,
+attached, fetched, and read after the setting is changed to `false`. The setting also does not add
+support for an `encrypted` disk layered over `CAS`; use object-store encryption at rest instead.
+
+For a disk declared in server configuration, `cas_allow_packed_parts` is read when the disk object
+is created. Changing it requires restarting the corresponding server; `SYSTEM RELOAD CONFIG` does
+not change the capability of an existing disk object. For an inline `disk(...)`, the value is stored
+in table metadata, so a restart does not change it and replacing the settings of an existing custom
+disk in place is not supported.
+
+For a rolling upgrade:
+
+1. Keep the key absent from every server that still runs an older binary. An older binary rejects
+   an unknown `cas_*` key even when its value is `false`; updated servers may keep the key absent or
+   use its default `false`.
+2. After every server that may open these parts can read and write the new archive layout, set
+   `cas_allow_packed_parts = true` only in the underlying `metadata_type=cas` disk blocks where
+   `Packed` creation is wanted.
+3. Restart the corresponding servers so the new disk objects receive the capability.
+
+Downgrade to a binary without CAS Packed support is not supported in the current MVP. Setting
+`cas_allow_packed_parts = false` does not rewrite existing `Packed` parts and is not a safe downgrade
+procedure. Such an older binary also rejects the unknown `cas_allow_packed_parts` key even when its
+value is `false`.
+
 All servers sharing a pool must run the same `cas_mount_lease_ttl_ms` and `cas_mount_renew_period_ms`.
 Startup reclaim and GC's fence-out both judge liveness by the mount slot's write token holding stable
 on the observer's own `CLOCK_BOOTTIME`, using the observer's own threshold — nothing about a writer's

docs/en/antalya/cas/roadmap.md

@@ -22,8 +22,12 @@ binary framing, no protobuf) so any object can be read with ordinary line-orient
 
 **Write path.** Conditional writes for mutable metadata/control objects; mandatory blob `HEAD`
 followed by adoption or unconditional, multipart-capable publication; a bounded thread pool
-fanning out multi-blob part materialization in parallel; carry-forward on mutation for `Wide` parts (an
-untouched column is re-referenced, not re-hashed).
+fanning out multi-blob part materialization in parallel; carry-forward on mutation for `Wide` parts
+(an untouched column is re-referenced, not re-hashed); opt-in `Packed` parts through
+`cas_allow_packed_parts`, with each base or projection `data.packed` archive published as one CAS
+blob. `INSERT` and merge select `Packed` from the part-storage thresholds only when the actual
+destination disk allows it. A mutation preserves the source storage type, except that destination
+capability may change `Packed -> Full`; mutation does not provide `Full -> Packed` conversion.
 
 **Read path.** Ref resolution to manifest to ranged blob reads, with a manifest-decode cache and
 a part-folder view cache sitting on that path.
@@ -84,7 +88,8 @@ positioning.
   disk succeeds, but the first `INSERT` fails (`Autocommit writes are not supported for content part
   files on a content-addressed disk`). Layering a `cache` disk in front of `CAS` is the supported
   wrapper shape (see [configuration](/antalya/cas/configuration)); encryption at rest currently has
-  to come from the object store side.
+  to come from the object store side. Enabling `cas_allow_packed_parts` does not change this
+  limitation.
 - **The format and settings surface can still change.** CAS is pre-release: there is no persisted
   production data to keep compatible, so a format change costs a version bump, not a migration.
   Treat every detail on these pages as subject to change until the format is declared stable.

programs/disks/DisksApp.cpp

@@ -235,7 +235,8 @@ bool DisksApp::processQueryText(const String & text)
         {
             int code = err.code();
             error_string = getExceptionMessageForLogging(err, true, false);
-            last_command_exit_code = code;
+            if (command && command->command_name.starts_with("cas-"))
+                last_command_exit_code = code;
             if (code == ErrorCodes::BAD_ARGUMENTS)
             {
                 if (command.get())
@@ -252,12 +253,14 @@ bool DisksApp::processQueryText(const String & text)
         catch (std::exception & err)
         {
             error_string = err.what();
-            last_command_exit_code = ErrorCodes::STD_EXCEPTION;
+            if (command && command->command_name.starts_with("cas-"))
+                last_command_exit_code = ErrorCodes::STD_EXCEPTION;
         }
         catch (...) // Ok: report unknown exception
         {
             error_string = "Unknown exception";
-            last_command_exit_code = ErrorCodes::STD_EXCEPTION;
+            if (command && command->command_name.starts_with("cas-"))
+                last_command_exit_code = ErrorCodes::STD_EXCEPTION;
         }
         if (error_string.has_value())
         {

src/Disks/DiskEncrypted.h

@@ -329,6 +329,7 @@ class DiskEncrypted : public IDisk
     bool isReadOnly() const override { return delegate->isReadOnly(); }
     bool isWriteOnce() const override { return delegate->isWriteOnce(); }
     bool isPlain() const override { return delegate->isPlain(); }
+    bool allowsPackedParts() const override { return delegate->allowsPackedParts(); }
 
     ObjectStoragePtr getObjectStorage() override { return delegate->getObjectStorage(); }
 

src/Disks/DiskObjectStorage/DiskObjectStorage.cpp

@@ -810,6 +810,11 @@ bool DiskObjectStorage::isContentAddressed() const
     return metadata_storage->isContentAddressed();
 }
 
+bool DiskObjectStorage::allowsPackedParts() const
+{
+    return metadata_storage->allowsPackedParts();
+}
+
 bool DiskObjectStorage::supportsAtomicFileWrites() const
 {
     return metadata_storage->supportsAtomicFileWrites();

src/Disks/DiskObjectStorage/DiskObjectStorage.h

@@ -227,6 +227,8 @@ friend class DiskObjectStorageReservation;
 
     bool isContentAddressed() const override;
 
+    bool allowsPackedParts() const override;
+
     bool supportsAtomicFileWrites() const override;
 
     /// Get structure of object storage this disk works with. Examples:

src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/ContentAddressedMetadataStorage.cpp

@@ -83,6 +83,7 @@ namespace ContentAddressedSetting
     extern const ContentAddressedSettingsUInt64 part_folder_cache_bytes;
     extern const ContentAddressedSettingsUInt64 part_folder_cache_max_entries;
     extern const ContentAddressedSettingsUInt64 part_folder_cache_max_entry_bytes;
+    extern const ContentAddressedSettingsBool allow_packed_parts;
     extern const ContentAddressedSettingsUInt64 manifest_decode_cache_bytes;
     extern const ContentAddressedSettingsUInt64 gc_meta_pool_size;
     extern const ContentAddressedSettingsUInt64 gc_io_concurrency;
@@ -307,6 +308,7 @@ ContentAddressedMetadataStorage::ContentAddressedMetadataStorage(
     , cas_part_folder_cache_bytes(settings_[ContentAddressedSetting::part_folder_cache_bytes].value)
     , cas_part_folder_cache_max_entries(settings_[ContentAddressedSetting::part_folder_cache_max_entries].value)
     , cas_part_folder_cache_max_entry_bytes(settings_[ContentAddressedSetting::part_folder_cache_max_entry_bytes].value)
+    , allow_packed_parts(settings_[ContentAddressedSetting::allow_packed_parts].value)
     , manifest_decode_cache_bytes(settings_[ContentAddressedSetting::manifest_decode_cache_bytes].value)
     , gc_meta_pool_size(settings_[ContentAddressedSetting::gc_meta_pool_size].value)
     , gc_io_concurrency(settings_[ContentAddressedSetting::gc_io_concurrency].value)

src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/ContentAddressedMetadataStorage.h

@@ -241,6 +241,7 @@ class ContentAddressedMetadataStorage final : public IMetadataStorage, public IC
     bool supportsStat() const override { return false; }
     bool isReadOnly() const override { return read_only; }
     bool isContentAddressed() const override { return true; }
+    bool allowsPackedParts() const override { return allow_packed_parts; }
 
 
     /// Fail-close gate shared by every mutating entry point (transactions, GC round, GC rebuild,
@@ -639,6 +640,7 @@ class ContentAddressedMetadataStorage final : public IMetadataStorage, public IC
     const uint64_t cas_part_folder_cache_bytes;
     const uint64_t cas_part_folder_cache_max_entries;
     const uint64_t cas_part_folder_cache_max_entry_bytes;
+    const bool allow_packed_parts;
     /// Byte bound for the manifest decode cache in `Cas::Pool`. Zero disables decode caching.
     const uint64_t manifest_decode_cache_bytes;
     /// Bounded pool size for GC's per-hash freshness-metadata writes.

src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/ContentAddressedSettings.cpp

@@ -77,6 +77,7 @@ constexpr std::string_view CAS_KEY_PREFIX = "cas_";
     DECLARE(UInt64, part_folder_cache_bytes, 64ULL << 20, "Part-folder view cache byte budget (0 disables retention)", 0) \
     DECLARE(UInt64, part_folder_cache_max_entries, 10000, "Part-folder view cache entry cap", 0) \
     DECLARE(UInt64, part_folder_cache_max_entry_bytes, 16ULL << 20, "Oversized part-folder views bypass retention above this size", 0) \
+    DECLARE(Bool,   allow_packed_parts, false, "Allow new MergeTree parts on this disk to use the Packed format", 0) \
     DECLARE(UInt64, manifest_decode_cache_bytes, 128ULL << 20, "Manifest DECODE cache byte budget (0 disables)", 0) \
     DECLARE(UInt64, gc_meta_pool_size, 16, "Bounded pool size for GC per-hash freshness-meta writes", 0) \
     DECLARE(UInt64, gc_io_concurrency, 16, "Maximum number of threads in the GC I/O pool. Used for fold and rebuild read-ahead, orphan-manifest sweep planning reads, pending_deletes HEAD plus conditional DELETE, and namespace janitor batch deletes. Per-hash meta writes use gc_meta_pool_size; other GC requests run on the round thread. 1 disables parallel GC I/O", 0) \

src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/ContentAddressedTransaction.cpp

@@ -66,6 +66,12 @@ bool hasSuffix(std::string_view s, std::string_view suffix)
 
 bool partFileMustStayBlob(std::string_view file_name)
 {
+    auto basename = file_name;
+    if (const auto separator = basename.find_last_of('/'); separator != std::string_view::npos)
+        basename.remove_prefix(separator + 1);
+
+    if (basename == "data.packed" || basename == "data.packed.tmp")
+        return true;
     if (file_name == "primary.idx")
         return true;
     for (std::string_view suffix : {".bin", ".mrk", ".mrk2", ".mrk3", ".cmrk", ".cmrk2", ".cmrk3"})

src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/README.md

@@ -191,12 +191,6 @@ not to multipart-capable blob publication.
   (local object storage) and "`cas s3 storage`" run the whole
   stateless suite with `MergeTree` defaulting to a CAS disk. Tests that
   legitimately cannot run there carry the `no-cas-storage` tag.
-- **Stress**: `Stress test (amd_{debug,asan_ubsan,tsan,msan}, cas s3 storage)`
-  run the same `clickhouse-test --stress-tests` loop as the other stress
-  jobs (stateless tests picked at random, several clients in parallel, no
-  result validation) with `cas_s3` as the default `MergeTree` policy
-  and RustFS backing the pool. One job per AMD sanitizer build plus
-  debug (the non-sanitizer stress package).
 - **Soak / chaos**: `utils/ca-soak/` — multi-replica docker-compose
   harnesses (fault proxies, GC sharding variants, AWS S3/GCS backends) and
   adversarial scenarios.