diff --git a/ci/defs/job_configs.py b/ci/defs/job_configs.py index 4cb789c470d52..eaec79a77d782 100644 --- a/ci/defs/job_configs.py +++ b/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=[ diff --git a/ci/jobs/scripts/clickhouse_proc.py b/ci/jobs/scripts/clickhouse_proc.py index bd82a477c917d..72c457ed3dacc 100644 --- a/ci/jobs/scripts/clickhouse_proc.py +++ b/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: diff --git a/ci/jobs/scripts/stress/stress.py b/ci/jobs/scripts/stress/stress.py index e0e61d5dda1bc..793e71d1d8d70 100755 --- a/ci/jobs/scripts/stress/stress.py +++ b/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") diff --git a/ci/jobs/stress_job.py b/ci/jobs/stress_job.py index 7ff3788846cc7..0f139d47c8e80 100644 --- a/ci/jobs/stress_job.py +++ b/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 diff --git a/ci/tests/test_stress_job.py b/ci/tests/test_stress_job.py index efb797a55f54e..0f0e035404fd8 100644 --- a/ci/tests/test_stress_job.py +++ b/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"])) diff --git a/docs/en/antalya/cas/configuration.md b/docs/en/antalya/cas/configuration.md index bd77ad8203456..939b10ae93112 100644 --- a/docs/en/antalya/cas/configuration.md +++ b/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 diff --git a/docs/en/antalya/cas/roadmap.md b/docs/en/antalya/cas/roadmap.md index 4df03af9ab3ce..695ddc00a380a 100644 --- a/docs/en/antalya/cas/roadmap.md +++ b/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. diff --git a/programs/disks/DisksApp.cpp b/programs/disks/DisksApp.cpp index 39f15a9a6f7e0..ef3f3ebfc9f10 100644 --- a/programs/disks/DisksApp.cpp +++ b/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()) { diff --git a/src/Disks/DiskEncrypted.h b/src/Disks/DiskEncrypted.h index d7e445961a5ca..5f70dbfb66823 100644 --- a/src/Disks/DiskEncrypted.h +++ b/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(); } diff --git a/src/Disks/DiskObjectStorage/DiskObjectStorage.cpp b/src/Disks/DiskObjectStorage/DiskObjectStorage.cpp index 548c8870e613e..172a8338a08e4 100644 --- a/src/Disks/DiskObjectStorage/DiskObjectStorage.cpp +++ b/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(); diff --git a/src/Disks/DiskObjectStorage/DiskObjectStorage.h b/src/Disks/DiskObjectStorage/DiskObjectStorage.h index 9c0c59028d1f1..1bb5980708741 100644 --- a/src/Disks/DiskObjectStorage/DiskObjectStorage.h +++ b/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: diff --git a/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/ContentAddressedMetadataStorage.cpp b/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/ContentAddressedMetadataStorage.cpp index c1346be05cc87..4e3dc902c5bd6 100644 --- a/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/ContentAddressedMetadataStorage.cpp +++ b/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) diff --git a/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/ContentAddressedMetadataStorage.h b/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/ContentAddressedMetadataStorage.h index 5f0aadb2d7d40..a22b6cbf90bea 100644 --- a/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/ContentAddressedMetadataStorage.h +++ b/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. diff --git a/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/ContentAddressedSettings.cpp b/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/ContentAddressedSettings.cpp index f27a40242f463..ea2cef27fe60a 100644 --- a/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/ContentAddressedSettings.cpp +++ b/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) \ diff --git a/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/ContentAddressedTransaction.cpp b/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/ContentAddressedTransaction.cpp index 4f033113e90d3..3408c7f4309fe 100644 --- a/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/ContentAddressedTransaction.cpp +++ b/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"}) diff --git a/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/README.md b/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/README.md index a086b53cf903a..44430e021ddbb 100644 --- a/src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/README.md +++ b/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. diff --git a/src/Disks/DiskObjectStorage/MetadataStorages/IMetadataStorage.h b/src/Disks/DiskObjectStorage/MetadataStorages/IMetadataStorage.h index 27dfec46900af..1939a99fc41bd 100644 --- a/src/Disks/DiskObjectStorage/MetadataStorages/IMetadataStorage.h +++ b/src/Disks/DiskObjectStorage/MetadataStorages/IMetadataStorage.h @@ -357,6 +357,8 @@ class IMetadataStorage : private boost::noncopyable /// disk transaction delegates writes to the metadata transaction's content-addressed buffer. virtual bool isContentAddressed() const { return false; } + virtual bool allowsPackedParts() const { return true; } + /// [TXN-ONE-PIPELINE] True when a transaction from this storage stages every mutation into a /// transaction-private overlay at call time (eager) rather than queuing effects for FIFO replay in /// commit. When true, DiskObjectStorageTransaction routes every mutating method straight to the diff --git a/src/Disks/IDisk.h b/src/Disks/IDisk.h index b79a3171bde72..cfb7bfd661283 100644 --- a/src/Disks/IDisk.h +++ b/src/Disks/IDisk.h @@ -479,6 +479,8 @@ class IDisk : public Space /// True when a file write on this disk publishes atomically (see `IDataPartStorage::supportsAtomicFileWrites`). virtual bool supportsAtomicFileWrites() const { return false; } + virtual bool allowsPackedParts() const { return true; } + virtual bool isWriteOnce() const { return false; } virtual bool supportsHardLinks() const { return true; } diff --git a/src/Disks/ReadOnlyDiskWrapper.h b/src/Disks/ReadOnlyDiskWrapper.h index 9a38e85cde77e..55ae104e2c7b0 100644 --- a/src/Disks/ReadOnlyDiskWrapper.h +++ b/src/Disks/ReadOnlyDiskWrapper.h @@ -94,6 +94,7 @@ class ReadOnlyDiskWrapper : public IDisk /// the delegate's answer through the wrapper, or a wrapped content-addressed disk silently /// drops out of the CAS introspection paths. bool isContentAddressed() const override { return delegate->isContentAddressed(); } + bool allowsPackedParts() const override { return delegate->allowsPackedParts(); } std::unordered_map getSerializedMetadata(const std::vector & file_paths) const override { return delegate->getSerializedMetadata(file_paths); } diff --git a/src/Disks/tests/gtest_ca_transaction.cpp b/src/Disks/tests/gtest_ca_transaction.cpp index 1819648fec46b..fc21603bbd3c5 100644 --- a/src/Disks/tests/gtest_ca_transaction.cpp +++ b/src/Disks/tests/gtest_ca_transaction.cpp @@ -7,6 +7,7 @@ #include #include #include +#include #include namespace ProfileEvents @@ -377,6 +378,41 @@ TEST(CASTransactionInlining, EagerFileInlinedDataBinBlobbed) EXPECT_EQ(storage->getFileSize("a99/a99a99a9-9999-4999-8999-999999999999/all_1_1_0/checksums.txt"), 13u); } +TEST(CASTransactionPackedManifest, ExactArchivesAreBlobs) +{ + auto storage = openTxStorage(); + auto tx = storage->createTransaction(); + const std::string tmp_part = "c01/c01c01c0-0101-4101-8101-010101010101/tmp_insert_all_1_1_0"; + const std::string final_part = "c01/c01c01c0-0101-4101-8101-010101010101/all_1_1_0"; + + writeFileTx(*tx, tmp_part + "/data.packed", "base"); + writeFileTx(*tx, tmp_part + "/p1.proj/data.packed", "projection-1"); + writeFileTx(*tx, tmp_part + "/p2.proj/data.packed", "projection-2"); + writeFileTx(*tx, tmp_part + "/checksums.txt", "base-checksums"); + writeFileTx(*tx, tmp_part + "/p1.proj/checksums.txt", "projection-1-checksums"); + writeFileTx(*tx, tmp_part + "/p2.proj/checksums.txt", "projection-2-checksums"); + tx->moveDirectory(tmp_part, final_part); + tx->commit(DB::NoCommitOptions{}); + + const auto ns = storage->liveNamespace("c01c01c0-0101-4101-8101-010101010101"); + const auto resolved = storage->store()->resolveRef(ns, "all_1_1_0"); + ASSERT_TRUE(resolved.has_value()); + const auto manifest = storage->store()->readManifest(resolved->manifest_id); + + std::vector blob_paths; + for (const auto & entry : manifest.entries) + if (entry.placement == DB::Cas::EntryPlacement::Blob) + blob_paths.push_back(entry.path); + std::sort(blob_paths.begin(), blob_paths.end()); + + const std::vector expected{ + "data.packed", + "p1.proj/data.packed", + "p2.proj/data.packed", + }; + EXPECT_EQ(blob_paths, expected); +} + /// all-tree-part-files Task 4: a standalone /// write of ONE file onto an ALREADY-COMMITTED part must carry every other file of that part forward /// (a repoint, Task 3), never replace the manifest with just the touched file. diff --git a/src/Disks/tests/gtest_cas_inline_placement.cpp b/src/Disks/tests/gtest_cas_inline_placement.cpp index 5b4000636fa0d..327cea75b826e 100644 --- a/src/Disks/tests/gtest_cas_inline_placement.cpp +++ b/src/Disks/tests/gtest_cas_inline_placement.cpp @@ -26,3 +26,16 @@ TEST(CASInlinePlacement, EagerMetadataFilesAreInlineCandidates) EXPECT_FALSE(partFileMustStayBlob("minmax_date.idx")); EXPECT_FALSE(partFileMustStayBlob("default_compression_codec.txt")); } + +TEST(CASInlinePlacement, PackedArchiveBasenameStaysBlob) +{ + EXPECT_TRUE(partFileMustStayBlob("data.packed")); + EXPECT_TRUE(partFileMustStayBlob("data.packed.tmp")); + EXPECT_TRUE(partFileMustStayBlob("projection.proj/data.packed")); + EXPECT_TRUE(partFileMustStayBlob("projection.proj/data.packed.tmp")); + + EXPECT_FALSE(partFileMustStayBlob("not_data.packed")); + EXPECT_FALSE(partFileMustStayBlob("data.packed.backup")); + EXPECT_FALSE(partFileMustStayBlob("data.packed.tmp.backup")); + EXPECT_FALSE(partFileMustStayBlob("projection.proj/not_data.packed")); +} diff --git a/src/Disks/tests/gtest_cas_settings.cpp b/src/Disks/tests/gtest_cas_settings.cpp index a362d96e67cf8..61c800b7491ee 100644 --- a/src/Disks/tests/gtest_cas_settings.cpp +++ b/src/Disks/tests/gtest_cas_settings.cpp @@ -28,6 +28,7 @@ namespace DB::ContentAddressedSetting extern const ContentAddressedSettingsUInt64 gc_bulk_delete_chunk_keys; extern const ContentAddressedSettingsString scratch_path; extern const ContentAddressedSettingsBool unsafe_remount_no_delay; + extern const ContentAddressedSettingsBool allow_packed_parts; } namespace @@ -111,6 +112,21 @@ TEST(CASContentAddressedSettings, DefaultsAndOverridesLand) EXPECT_EQ(s[ContentAddressedSetting::scratch_path].value, "/data/default_scratch"); } +TEST(CASContentAddressedSettings, PackedPartsDefaultOffAndOptIn) +{ + auto default_cfg = makeConfig("srv1"); + ContentAddressedSettings defaults; + defaults.loadFromConfig(*default_cfg, "disk", "/data", "/data/default_scratch", identity_macros); + EXPECT_FALSE(defaults[ContentAddressedSetting::allow_packed_parts].value); + + auto enabled_cfg = makeConfig( + "srv1" + "true"); + ContentAddressedSettings enabled; + enabled.loadFromConfig(*enabled_cfg, "disk", "/data", "/data/default_scratch", identity_macros); + EXPECT_TRUE(enabled[ContentAddressedSetting::allow_packed_parts].value); +} + TEST(CASContentAddressedSettings, RemovedCacheSettingsAreRejected) { for (const std::string & suffix : {"cache_bytes", "head_first_min_bytes"}) diff --git a/src/IO/PackedFilesReader.cpp b/src/IO/PackedFilesReader.cpp index a81f7dce3fd92..57aeff991116d 100644 --- a/src/IO/PackedFilesReader.cpp +++ b/src/IO/PackedFilesReader.cpp @@ -19,6 +19,11 @@ PackedFilesReader::PackedFilesReader( { } +PackedFilesReader::PackedFilesReader(ReadBufferFromFileBase & data_file) + : index(readIndex(data_file)) +{ +} + PackedFilesReader::PackedFilesReader(PackedFilesIO::Index index_) : index(std::move(index_)) { @@ -96,7 +101,7 @@ Names PackedFilesReader::getFileNames() const return res; } -static ReadSettings patchSettings(ReadSettings settings) +ReadSettings PackedFilesReader::patchSettings(ReadSettings settings) { settings.local_fs_settings.direct_io_threshold = 0; if (settings.local_fs_settings.method == LocalFSReadMethod::mmap) @@ -115,14 +120,24 @@ std::unique_ptr PackedFilesReader::readFile( if (it == index.end()) throw Exception(ErrorCodes::FILE_DOESNT_EXIST, "File {} does not exist", file_name); - const auto offset = it->second.offset; - const auto size = it->second.size; - /// ReadBufferFromFileView doesn't support reading with mmap and direct io methods, because /// they require special alignment which cannot be achieved while reading archive file. /// So just disable them. auto in = disk->readFile(data_file_name, patchSettings(settings), read_hint); - return std::make_unique(std::move(in), file_name, offset, offset + size); + return readFile(std::move(in), file_name); +} + +std::unique_ptr PackedFilesReader::readFile( + std::unique_ptr data_file, + const String & file_name) const +{ + auto it = index.find(file_name); + if (it == index.end()) + throw Exception(ErrorCodes::FILE_DOESNT_EXIST, "File {} does not exist", file_name); + + const auto offset = it->second.offset; + const auto size = it->second.size; + return std::make_unique(std::move(data_file), file_name, offset, offset + size); } } diff --git a/src/IO/PackedFilesReader.h b/src/IO/PackedFilesReader.h index b3475849830f7..0faec5fef25a9 100644 --- a/src/IO/PackedFilesReader.h +++ b/src/IO/PackedFilesReader.h @@ -27,6 +27,8 @@ class PackedFilesReader /// only to read the index here; they are not retained. PackedFilesReader(const DiskPtr & disk, const String & data_file_name, const ReadSettings & read_settings); + explicit PackedFilesReader(ReadBufferFromFileBase & data_file); + /// Constructor that initializes the index by the provided one. explicit PackedFilesReader(PackedFilesIO::Index index_); @@ -48,9 +50,14 @@ class PackedFilesReader const ReadSettings & settings, std::optional read_hint) const; + std::unique_ptr readFile( + std::unique_ptr data_file, + const String & file_name) const; + const PackedFilesIO::Index & getIndex() const { return index; } static PackedFilesIO::Index readIndex(ReadBuffer & in); + static ReadSettings patchSettings(ReadSettings settings); private: /// Index of archive: immutable and path-independent. diff --git a/src/IO/ReadBufferFromFileView.cpp b/src/IO/ReadBufferFromFileView.cpp index 22de4731673f1..0b957cda5b490 100644 --- a/src/IO/ReadBufferFromFileView.cpp +++ b/src/IO/ReadBufferFromFileView.cpp @@ -28,11 +28,14 @@ ReadBufferFromFileView::ReadBufferFromFileView( file_offset_of_buffer_end = impl_buffer_end; original_working_buffer = working_buffer; resizeWorkingBuffer(); + restoreViewWorkingBuffer(0); } void ReadBufferFromFileView::prefetch(Priority priority) { + const size_t view_position = getPosition(); executeWithOriginalBuffer([&]{ impl->prefetch(priority); }); + restoreViewWorkingBuffer(view_position); } void ReadBufferFromFileView::setReadUntilPosition(size_t position) @@ -54,6 +57,7 @@ void ReadBufferFromFileView::setReadUntilPosition(size_t position) }); file_offset_of_buffer_end = impl_buffer_end; resizeWorkingBuffer(); + restoreViewWorkingBuffer(getPosition()); } void ReadBufferFromFileView::setReadUntilEnd() @@ -68,6 +72,7 @@ void ReadBufferFromFileView::setReadUntilEnd() }); file_offset_of_buffer_end = impl_buffer_end; resizeWorkingBuffer(); + restoreViewWorkingBuffer(getPosition()); } off_t ReadBufferFromFileView::getPosition() @@ -96,7 +101,10 @@ bool ReadBufferFromFileView::nextImpl() /// the impl continues from ITS position - incrementing would mislabel the new chunk. file_offset_of_buffer_end = impl_buffer_end; resizeWorkingBuffer(); + restoreViewWorkingBuffer(getPosition()); } + else + restoreViewWorkingBuffer(current_position - left_bound); return result; } @@ -130,6 +138,7 @@ off_t ReadBufferFromFileView::seek(off_t off, int whence) file_offset_of_buffer_end = impl_buffer_end; resizeWorkingBuffer(); + restoreViewWorkingBuffer(result - left_bound); return result - left_bound; } @@ -137,6 +146,8 @@ off_t ReadBufferFromFileView::seek(off_t off, int whence) template void ReadBufferFromFileView::executeWithOriginalBuffer(Op && op) { + const size_t view_position = getPosition(); + /// Set original buffer as working buffer. working_buffer.swap(original_working_buffer); @@ -154,6 +165,7 @@ void ReadBufferFromFileView::executeWithOriginalBuffer(Op && op) /// seek bound checks), so restore-on-exception is required for the view to stay consistent. swap(*impl); original_working_buffer = working_buffer; + restoreViewWorkingBuffer(view_position); throw; } swap(*impl); @@ -161,6 +173,20 @@ void ReadBufferFromFileView::executeWithOriginalBuffer(Op && op) original_working_buffer = working_buffer; } +void ReadBufferFromFileView::restoreViewWorkingBuffer(size_t view_position) +{ + if (working_buffer.empty()) + { + bytes = view_position; + return; + } + + const size_t native_offset = pos - working_buffer.begin(); + const size_t view_offset = std::min(native_offset, view_position); + working_buffer = Buffer(pos - view_offset, working_buffer.end()); + bytes = view_position - view_offset; +} + size_t ReadBufferFromFileView::getRightBound() const { return read_until_position ? *read_until_position : right_bound; diff --git a/src/IO/ReadBufferFromFileView.h b/src/IO/ReadBufferFromFileView.h index 615d06fa8d4de..db71fa9b76922 100644 --- a/src/IO/ReadBufferFromFileView.h +++ b/src/IO/ReadBufferFromFileView.h @@ -42,6 +42,7 @@ class ReadBufferFromFileView : public ReadBufferFromFileBase /// Resizes working buffer if it exceeds the right bound. void resizeWorkingBuffer(); + void restoreViewWorkingBuffer(size_t view_position); const std::unique_ptr impl; const String file_name; diff --git a/src/IO/tests/gtest_read_buffer_from_file_view.cpp b/src/IO/tests/gtest_read_buffer_from_file_view.cpp index b4a4333618565..35c489e9bdb85 100644 --- a/src/IO/tests/gtest_read_buffer_from_file_view.cpp +++ b/src/IO/tests/gtest_read_buffer_from_file_view.cpp @@ -2,6 +2,7 @@ #include #include +#include #include @@ -123,6 +124,37 @@ class ReadBufferFromFileViewTest : public ::testing::TestWithParam } +TEST(ReadBufferFromFileViewCount, MemoryBacked) +{ + const auto data = makePayload(1057); + auto inner = std::make_unique("memory", data); + ReadBufferFromFileView view(std::move(inner), "viewed", 1053, 1057); + + EXPECT_EQ(view.count(), 0); + EXPECT_EQ(readExact(view, 4), data.substr(1053, 4)); + EXPECT_EQ(view.count(), 4); + EXPECT_EQ(view.seek(1, SEEK_SET), 1); + EXPECT_EQ(view.getPosition(), 1); + EXPECT_EQ(readExact(view, 3), data.substr(1054, 3)); + EXPECT_EQ(view.count(), 4); +} + +TEST(ReadBufferFromFileViewCount, Nested) +{ + const auto data = makePayload(1088); + auto memory = std::make_unique("memory", data); + auto inner = std::make_unique(std::move(memory), "inner", 1024, 1088); + ReadBufferFromFileView view(std::move(inner), "outer", 29, 33); + + EXPECT_EQ(view.count(), 0); + EXPECT_EQ(readExact(view, 4), data.substr(1053, 4)); + EXPECT_EQ(view.count(), 4); + EXPECT_EQ(view.seek(1, SEEK_SET), 1); + EXPECT_EQ(view.getPosition(), 1); + EXPECT_EQ(readExact(view, 3), data.substr(1054, 3)); + EXPECT_EQ(view.count(), 4); +} + TEST_P(ReadBufferFromFileViewTest, SequentialReadWholeView) { const auto [chunk, mode] = GetParam(); @@ -193,9 +225,27 @@ TEST_P(ReadBufferFromFileViewTest, SetReadUntilTruncateThenExtend) EXPECT_EQ(view->getPosition(), 30); EXPECT_EQ(readExact(*view, 100), payload.substr(30, 100)); + EXPECT_EQ(view->count(), 130); view->setReadUntilEnd(); + EXPECT_EQ(view->count(), 130); EXPECT_EQ(readExact(*view, payload.size() - 130), payload.substr(130)); EXPECT_TRUE(view->eof()); + EXPECT_EQ(view->count(), payload.size()); +} + +TEST_P(ReadBufferFromFileViewTest, PrefetchKeepsCount) +{ + const auto [chunk, mode] = GetParam(); + const auto payload = makePayload(1000); + auto view = makeView(payload, chunk, mode); + + EXPECT_EQ(readExact(*view, 37), payload.substr(0, 37)); + const auto count = view->count(); + EXPECT_EQ(count, 37); + view->prefetch(Priority{0}); + EXPECT_EQ(view->count(), count); + EXPECT_EQ(readExact(*view, 23), payload.substr(37, 23)); + EXPECT_EQ(view->count(), 60); } /// The exact shape of the failing compact-part in-order read: per granule, adjust the right diff --git a/src/Storages/MergeTree/DataPartStorageOnDiskBase.cpp b/src/Storages/MergeTree/DataPartStorageOnDiskBase.cpp index 352c4419d5f44..362597ece842f 100644 --- a/src/Storages/MergeTree/DataPartStorageOnDiskBase.cpp +++ b/src/Storages/MergeTree/DataPartStorageOnDiskBase.cpp @@ -4,6 +4,7 @@ #include #include #include +#include #include #include #include @@ -32,6 +33,9 @@ #include +#include +#include + namespace DB { @@ -47,6 +51,50 @@ namespace ErrorCodes namespace { +class DataPartStorageIteratorOnDisk final : public IDataPartStorageIterator +{ +public: + DataPartStorageIteratorOnDisk(DiskPtr disk_, DirectoryIteratorPtr it_) + : disk(std::move(disk_)), it(std::move(it_)) + { + } + + void next() override { it->next(); } + bool isValid() const override { return it->isValid(); } + bool isFile() const override { return isValid() && disk->existsFile(it->path()); } + std::string name() const override { return it->name(); } + std::string path() const override { return it->path(); } + +private: + DiskPtr disk; + DirectoryIteratorPtr it; +}; + +/// Iterates names prepared by `iterateDirectoryInTransactionOrDisk`: the union of committed and staged +/// immediate children. `isFile` checks only the disk, so it returns false for a staged-only file. +/// `MergeTreeIndexGranularityInfo::getMarksTypeFromFilesystem` uses this view to detect a mark file by +/// its extension while the parent part is still staged. +class DataPartStorageMergedIterator final : public IDataPartStorageIterator +{ +public: + DataPartStorageMergedIterator(DiskPtr disk_, std::string dir_path_, std::vector names_) + : disk(std::move(disk_)), dir_path(std::move(dir_path_)), names(std::move(names_)) + { + } + + void next() override { ++pos; } + bool isValid() const override { return pos < names.size(); } + std::string name() const override { return names[pos]; } + std::string path() const override { return fs::path(dir_path) / names[pos]; } + bool isFile() const override { return isValid() && disk->existsFile(path()); } + +private: + DiskPtr disk; + std::string dir_path; + std::vector names; + size_t pos = 0; +}; + /// fsync `dir_path` and every subdirectory below it (children first), so that the directory /// entries created by a freeze/hardlink clone become durable. Used only for local disks /// (on remote/object disks getDirectorySyncGuard() returns nullptr and this is a no-op). @@ -1325,6 +1373,118 @@ bool DataPartStorageOnDiskBase::isCaseInsensitive() const return getDisk()->isCaseInsensitive(); } +bool DataPartStorageOnDiskBase::existsFileInTransactionOrDisk(const std::string & path) const +{ + if (transaction && transaction->tryGetInFlightFileSize(path).has_value()) + return true; + return volume->getDisk()->existsFile(path); +} + +bool DataPartStorageOnDiskBase::existsDirectoryInTransactionOrDisk(const std::string & path) const +{ + if (transaction && transaction->hasInFlightDirectory(path)) + return true; + return volume->getDisk()->existsDirectory(path); +} + +DataPartStorageIteratorPtr DataPartStorageOnDiskBase::iterateDirectoryInTransactionOrDisk(const std::string & path) const +{ + if (transaction) + { + if (auto staged = transaction->listInFlightDirectory(path); !staged.empty()) + { + std::set names(staged.begin(), staged.end()); + if (volume->getDisk()->existsDirectory(path)) + for (auto it = volume->getDisk()->iterateDirectory(path); it->isValid(); it->next()) + names.insert(it->name()); + return std::make_unique( + volume->getDisk(), path, std::vector(names.begin(), names.end())); + } + } + + return std::make_unique( + volume->getDisk(), + volume->getDisk()->iterateDirectory(path)); +} + +size_t DataPartStorageOnDiskBase::getFileSizeInTransactionOrDisk(const std::string & path) const +{ + if (transaction) + if (auto size = transaction->tryGetInFlightFileSize(path)) + return *size; + return volume->getDisk()->getFileSize(path); +} + +StoredObjects DataPartStorageOnDiskBase::getStorageObjectsInTransactionOrDisk(const std::string & path) const +{ + if (transaction) + { + if (auto inflight = transaction->tryGetInFlightStorageObjects(path)) + return std::move(*inflight); + /// An in-flight file can exist before it has storage objects, for example while an upload is pending. + /// Do not expose a committed object at the same path in that case. + if (transaction->tryGetInFlightFileSize(path)) + return {}; + } + return volume->getDisk()->getStorageObjects(path); +} + +void DataPartStorageOnDiskBase::prepareReadInTransactionOrDisk( + const std::string & path, + const ReadSettings & settings, + std::optional read_hint, + ReadPipeline & pipeline) const +{ + if (transaction) + { + StoredObjects inflight_objects; + if (auto objects = transaction->tryGetInFlightStorageObjects(path)) + inflight_objects = std::move(*objects); + else if (auto size = transaction->tryGetInFlightFileSize(path)) + /// `ReadPipeline::build` treats an empty object list as an empty file and does not invoke + /// the source. This placeholder makes it invoke the source, which ignores the object and + /// reads the staged bytes through the transaction. + inflight_objects = StoredObjects{StoredObject(path, path, *size)}; + + if (!inflight_objects.empty()) + { + /// Keep the transaction alive for as long as the pipeline can invoke the custom source. + auto current_transaction = transaction; + pipeline.setSource( + [current_transaction, path](const StoredObject &, const ReadSettings & read_settings, bool, bool) + { + auto buffer = current_transaction->tryReadFileInFlight(path, read_settings, std::nullopt); + if (!buffer) + throw Exception(ErrorCodes::LOGICAL_ERROR, "In-flight file {} cannot be read", path); + return buffer; + }, + std::move(inflight_objects), + settings); + return; + } + } + + volume->getDisk()->prepareRead(path, settings, read_hint, pipeline); +} + +std::unique_ptr DataPartStorageOnDiskBase::readFileIfExistsInTransactionOrDisk( + const std::string & path, + const ReadSettings & settings, + std::optional read_hint) const +{ + if (transaction) + { + auto in_flight_size = transaction->tryGetInFlightFileSize(path); + if (auto buffer = transaction->tryReadFileInFlight(path, settings, read_hint)) + return buffer; + /// If the transaction knows this file but cannot read it, this is a logical error. + /// Do not fall back to a committed file at the same path. + if (in_flight_size) + throw Exception(ErrorCodes::LOGICAL_ERROR, "In-flight file {} cannot be read", path); + } + return volume->getDisk()->readFileIfExists(path, settings, read_hint); +} + std::shared_ptr DataPartStorageOnDiskBase::getArchiveReaderForFile(const std::string & name) const { /// Prefix gate: only "skp_idx_..." names can be archive members, so unrelated files never load diff --git a/src/Storages/MergeTree/DataPartStorageOnDiskBase.h b/src/Storages/MergeTree/DataPartStorageOnDiskBase.h index dbe0f39159879..1cc3abad7e011 100644 --- a/src/Storages/MergeTree/DataPartStorageOnDiskBase.h +++ b/src/Storages/MergeTree/DataPartStorageOnDiskBase.h @@ -183,6 +183,24 @@ class DataPartStorageOnDiskBase : public IDataPartStorage DataPartStorageOnDiskBase(VolumePtr volume_, std::string root_path_, std::string part_dir_, DiskTransactionPtr transaction_); virtual MutableDataPartStoragePtr create(VolumePtr volume_, std::string root_path_, std::string part_dir_, bool initialize_) const = 0; + /// Read-your-writes helpers for a part that `transaction` is still assembling. File lookups prefer + /// staged state and fall back to the disk. Directory existence and iteration expose both committed + /// and staged state; iteration returns the union of their immediate children. + bool existsFileInTransactionOrDisk(const std::string & path) const; + bool existsDirectoryInTransactionOrDisk(const std::string & path) const; + DataPartStorageIteratorPtr iterateDirectoryInTransactionOrDisk(const std::string & path) const; + size_t getFileSizeInTransactionOrDisk(const std::string & path) const; + StoredObjects getStorageObjectsInTransactionOrDisk(const std::string & path) const; + void prepareReadInTransactionOrDisk( + const std::string & path, + const ReadSettings & settings, + std::optional read_hint, + ReadPipeline & pipeline) const; + std::unique_ptr readFileIfExistsInTransactionOrDisk( + const std::string & path, + const ReadSettings & settings, + std::optional read_hint) const; + /// Lazily load the per-part skp_idx.packed archive (if any), reading it as a standalone disk /// file. Subsequent calls return the cached reader, or nullptr when there is no such file -- /// including on storages that don't keep skp_idx.packed standalone (e.g. packed part storage, diff --git a/src/Storages/MergeTree/DataPartStorageOnDiskFull.cpp b/src/Storages/MergeTree/DataPartStorageOnDiskFull.cpp index 0d840aea1cb80..b041372481eef 100644 --- a/src/Storages/MergeTree/DataPartStorageOnDiskFull.cpp +++ b/src/Storages/MergeTree/DataPartStorageOnDiskFull.cpp @@ -62,101 +62,30 @@ DataPartStoragePtr DataPartStorageOnDiskFull::getProjection(const std::string & bool DataPartStorageOnDiskFull::exists() const { - auto path = fs::path(root_path) / part_dir; - /// CA read-your-writes: a part dir being assembled by this transaction (e.g. a carried-forward - /// projection dir staged into the open whole-part txn) is not on committed metadata yet. Mirrors - /// existsDirectory at directory granularity for the part's OWN directory. - if (transaction && transaction->hasInFlightDirectory(path)) - return true; - return volume->getDisk()->existsDirectory(path); + /// CAS does not report the top-level part directory through the transaction overlay. A temporary + /// part whose transaction was never committed then takes the early return in `removeIfNeeded` + /// instead of removing a path that committed metadata does not have. A projection storage asks for + /// an inner directory of its parent part, so a staged projection directory is visible here. + return existsDirectoryInTransactionOrDisk(fs::path(root_path) / part_dir); } bool DataPartStorageOnDiskFull::existsFileImpl(const std::string & name) const { - auto path = fs::path(root_path) / part_dir / name; - /// B59: a part still being assembled by this transaction can have staged-but-uncommitted files - /// (e.g. projection temp blocks on a content-addressed disk). Consult the held transaction first. - if (transaction && transaction->tryGetInFlightFileSize(path).has_value()) - return true; - return volume->getDisk()->existsFile(path); + /// A part being assembled can have staged files that committed metadata cannot see, for example + /// projection temporary blocks on CAS. The helper consults the held transaction before the disk. + return existsFileInTransactionOrDisk(fs::path(root_path) / part_dir / name); } bool DataPartStorageOnDiskFull::existsDirectory(const std::string & name) const { - auto path = fs::path(root_path) / part_dir / name; - /// CA read-your-writes: a part still being assembled by this transaction can have a staged-but-uncommitted - /// directory (e.g. a carried-forward projection hardlinked into the open whole-part txn) that committed - /// metadata cannot see yet. Mirrors existsFile (B59) at directory granularity. - if (transaction && transaction->hasInFlightDirectory(path)) - return true; - return volume->getDisk()->existsDirectory(path); + /// A carried-forward projection may be hard-linked into the open part transaction before committed + /// metadata exposes its directory. The helper detects staged files below it before falling back to the disk. + return existsDirectoryInTransactionOrDisk(fs::path(root_path) / part_dir / name); } -class DataPartStorageIteratorOnDisk final : public IDataPartStorageIterator -{ -public: - DataPartStorageIteratorOnDisk(DiskPtr disk_, DirectoryIteratorPtr it_) - : disk(std::move(disk_)), it(std::move(it_)) - { - } - - void next() override { it->next(); } - bool isValid() const override { return it->isValid(); } - bool isFile() const override { return isValid() && disk->existsFile(it->path()); } - std::string name() const override { return it->name(); } - std::string path() const override { return it->path(); } - -private: - DiskPtr disk; - DirectoryIteratorPtr it; -}; - -/// CA read-your-writes directory enumeration: a merged view of the committed disk entries PLUS the -/// immediate children this transaction has STAGED under the part dir (deduplicated). Used so -/// loadProjections' withPartFormatFromDisk can iterate a staged-but-uncommitted projection directory and -/// find its mark file. Mirrors existsFile/existsDirectory (B59) at the enumeration level; the committed -/// entries dominate (a name present both on disk and staged appears once). -class DataPartStorageMergedIterator final : public IDataPartStorageIterator -{ -public: - DataPartStorageMergedIterator(DiskPtr disk_, std::string dir_path_, std::vector names_) - : disk(std::move(disk_)), dir_path(std::move(dir_path_)), names(std::move(names_)) - { - } - - void next() override { ++pos; } - bool isValid() const override { return pos < names.size(); } - std::string name() const override { return names[pos]; } - std::string path() const override { return fs::path(dir_path) / names[pos]; } - bool isFile() const override { return isValid() && disk->existsFile(path()); } - -private: - DiskPtr disk; - std::string dir_path; - std::vector names; - size_t pos = 0; -}; - DataPartStorageIteratorPtr DataPartStorageOnDiskFull::iterate() const { - auto dir_path = fs::path(root_path) / part_dir; - if (transaction) - { - if (auto staged = transaction->listInFlightDirectory(dir_path); !staged.empty()) - { - /// Union the committed entries with the staged children (set semantics, committed dominates). - std::set names(staged.begin(), staged.end()); - if (volume->getDisk()->existsDirectory(dir_path)) - for (auto it = volume->getDisk()->iterateDirectory(dir_path); it->isValid(); it->next()) - names.insert(it->name()); - return std::make_unique( - volume->getDisk(), dir_path, std::vector(names.begin(), names.end())); - } - } - - return std::make_unique( - volume->getDisk(), - volume->getDisk()->iterateDirectory(dir_path)); + return iterateDirectoryInTransactionOrDisk(fs::path(root_path) / part_dir); } Poco::Timestamp DataPartStorageOnDiskFull::getFileLastModified(const String & file_name) const @@ -166,12 +95,9 @@ Poco::Timestamp DataPartStorageOnDiskFull::getFileLastModified(const String & fi size_t DataPartStorageOnDiskFull::getFileSizeImpl(const String & file_name) const { - auto path = fs::path(root_path) / part_dir / file_name; - /// B59: see existsFile — the merge stats the staged temp files before reading them back. - if (transaction) - if (auto size = transaction->tryGetInFlightFileSize(path)) - return *size; - return volume->getDisk()->getFileSize(path); + /// A merge gets the size of staged projection temporary files before reading them back. Committed + /// metadata cannot expose their sizes until the parent part commits, so consult the transaction first. + return getFileSizeInTransactionOrDisk(fs::path(root_path) / part_dir / file_name); } std::optional DataPartStorageOnDiskFull::getPackedFileUncompressedSize(const std::string & file_name) const @@ -191,16 +117,12 @@ std::vector DataPartStorageOnDiskFull::getRemotePaths(const std::st { const std::string path = fs::path(root_path) / part_dir / file_name; - /// B59: a file staged by this transaction resolves to its already-uploaded blob object(s) before commit. - /// A mutable per-part file intentionally does NOT resolve here (tryGetInFlightStorageObjects returns - /// nullopt → falls through): it has no blob object and must be read via tryReadFileInFlight. The merge - /// reads projection column blocks (blob-backed) through this path, not mutable files. - StoredObjects objects; - if (transaction) - if (auto inflight = transaction->tryGetInFlightStorageObjects(path)) - objects = std::move(*inflight); - if (objects.empty()) - objects = volume->getDisk()->getStorageObjects(path); + /// A staged blob that is not pending upload, for example one carried forward by `createHardLink` from a + /// committed part, resolves to its pool object. Blobs written by this transaction stay pending until + /// commit publishes them. A pending blob has no storage object yet, so do not fall back to a committed + /// object at the same logical path. An inline entry yields one synthetic object with an empty remote + /// path, the same as a committed inline entry. + auto objects = getStorageObjectsInTransactionOrDisk(path); std::vector remote_paths; remote_paths.reserve(objects.size()); @@ -226,43 +148,12 @@ void DataPartStorageOnDiskFull::prepareReadImpl( std::optional read_hint, ReadPipeline & pipeline) const { - auto path = fs::path(root_path) / part_dir / name; - - /// B59: read-your-writes for a part still being assembled by this transaction. A projection - /// spill-and-merge reads its own temp blocks back before the parent part's single commit; on a - /// content-addressed disk those files are staged in the transaction (blob uploaded, no ref yet), - /// so the committed metadata path can't see them. If the held transaction resolves the file - /// in-flight, serve it via a custom pipeline source that reads through the transaction. Gated on - /// `transaction != nullptr` so committed-part reads (no open transaction) are unchanged. - if (transaction) - { - StoredObjects inflight_objects; - if (auto objs = transaction->tryGetInFlightStorageObjects(path)) - inflight_objects = std::move(*objs); - else if (auto size = transaction->tryGetInFlightFileSize(path)) - /// Mutable per-part file staged inline (no blob object); synthesize a placeholder so the - /// single-object pipeline is satisfied — the custom creator below ignores it and reads the - /// inline bytes through the transaction. - inflight_objects = StoredObjects{StoredObject(path, path, *size)}; - - if (!inflight_objects.empty()) - { - /// Safe to capture the raw transaction pointer: no cache/gather/async stage is added on this - /// branch, so the custom source is consumed synchronously inside build() during this read and - /// the pointer is never retained past it. - auto * tx = transaction.get(); - pipeline.setSource( - [tx, path](const StoredObject &, const ReadSettings & read_settings, bool /*use_external_buffer*/, bool /*restrict_seek*/) - { - return tx->tryReadFileInFlight(path, read_settings, std::nullopt); - }, - std::move(inflight_objects), - settings); - return; - } - } - - volume->getDisk()->prepareRead(path, settings, read_hint, pipeline); + /// A projection spill-and-merge can read its own temporary blocks before the parent part commits. + /// On CAS a staged blob may still be pending upload or may already be in the pool, for example when + /// carried forward from a committed part, while inline content is stored in the staged entry. Committed + /// metadata cannot see that state, so the helper installs a source that reads through the held transaction. + /// Without a transaction, or if it does not know the path, the helper uses the disk pipeline unchanged. + prepareReadInTransactionOrDisk(fs::path(root_path) / part_dir / name, settings, read_hint, pipeline); } std::unique_ptr DataPartStorageOnDiskFull::readFileIfExistsImpl( @@ -270,14 +161,11 @@ std::unique_ptr DataPartStorageOnDiskFull::readFileIfExi const ReadSettings & settings, std::optional read_hint) const { - auto path = fs::path(root_path) / part_dir / name; - /// B59: serve a file staged by this transaction (uploaded blob or inline mutable bytes) before commit. - /// This direct delegate bypasses prepareRead, so the in-flight guard must be repeated here; it is the - /// only path that reaches the inline-mutable case via a returned buffer. - if (transaction) - if (auto rb = transaction->tryReadFileInFlight(path, settings, read_hint)) - return rb; - return volume->getDisk()->readFileIfExists(path, settings, read_hint); + /// `readFileIfExists` does not go through `prepareRead`, so it must apply the same transaction overlay. + /// Staged blobs, including pending uploads, and inline bytes are served through the transaction before + /// commit. If the transaction knows the path but cannot read it, the helper does not expose a + /// committed version. + return readFileIfExistsInTransactionOrDisk(fs::path(root_path) / part_dir / name, settings, read_hint); } std::unique_ptr DataPartStorageOnDiskFull::writeFile( diff --git a/src/Storages/MergeTree/DataPartStorageOnDiskPacked.cpp b/src/Storages/MergeTree/DataPartStorageOnDiskPacked.cpp index d6f8debe5bf2d..488459c56af84 100644 --- a/src/Storages/MergeTree/DataPartStorageOnDiskPacked.cpp +++ b/src/Storages/MergeTree/DataPartStorageOnDiskPacked.cpp @@ -14,13 +14,20 @@ #include #include #include +#include #include namespace DB { +namespace FailPoints +{ + extern const char part_storage_fail_commit_transaction[]; +} + namespace ErrorCodes { + extern const int FAULT_INJECTED; extern const int LOGICAL_ERROR; extern const int NOT_IMPLEMENTED; extern const int NOT_INITIALIZED; @@ -78,7 +85,7 @@ DataPartStoragePtr DataPartStorageOnDiskPacked::getProjection(const std::string bool DataPartStorageOnDiskPacked::exists() const { - return volume->getDisk()->existsFileOrDirectory(fs::path(root_path) / part_dir); + return existsDirectoryInTransactionOrDisk(fs::path(root_path) / part_dir); } bool DataPartStorageOnDiskPacked::existsDirectory(const std::string & file_name) const @@ -89,7 +96,7 @@ bool DataPartStorageOnDiskPacked::existsDirectory(const std::string & file_name) /// exceed the OS filename limit, so probing them would throw ENAMETOOLONG instead of returning false /// and wrongly mark the part broken during checkConsistency. if (isWrittenSeparately(file_name)) - return volume->getDisk()->existsDirectory(fs::path(root_path) / part_dir / file_name); + return existsDirectoryInTransactionOrDisk(fs::path(root_path) / part_dir / file_name); return false; } @@ -97,12 +104,10 @@ class DataPartStorageIteratorOnDiskPacked final : public IDataPartStorageIterato { public: DataPartStorageIteratorOnDiskPacked( - Names files_, DiskPtr disk_, - const String & root_path_, const String & part_dir_) + Names files_, DataPartStorageIteratorPtr dir_it_, const String & part_dir_) : files(std::move(files_)) , files_it(files.begin()) - , disk(std::move(disk_)) - , dir_it(disk->iterateDirectory(fs::path(root_path_) / part_dir_)) + , dir_it(std::move(dir_it_)) , part_dir(part_dir_) { } @@ -135,7 +140,7 @@ class DataPartStorageIteratorOnDiskPacked final : public IDataPartStorageIterato if (files_it == files.end()) { skipDataFile(); - return dir_it->isValid() && disk->existsFile(dir_it->path()); + return dir_it->isFile(); } return true; @@ -174,15 +179,15 @@ class DataPartStorageIteratorOnDiskPacked final : public IDataPartStorageIterato Names files; Names::const_iterator files_it; - DiskPtr disk; - DirectoryIteratorPtr dir_it; + DataPartStorageIteratorPtr dir_it; String part_dir; }; DataPartStorageIteratorPtr DataPartStorageOnDiskPacked::iterate() const { auto files = reader ? reader->getFileNames() : Names{}; - return std::make_unique(std::move(files), volume->getDisk(), root_path, part_dir); + return std::make_unique( + std::move(files), iterateDirectoryInTransactionOrDisk(fs::path(root_path) / part_dir), part_dir); } Poco::Timestamp DataPartStorageOnDiskPacked::getFileLastModified(const String & file_name) const @@ -214,14 +219,13 @@ bool DataPartStorageOnDiskPacked::existsFileImpl(const std::string & file_name) return true; } - auto disk = volume->getDisk(); if (isWrittenSeparately(file_name)) - return disk->existsFile(fs::path(root_path) / part_dir / file_name); + return existsFileInTransactionOrDisk(fs::path(root_path) / part_dir / file_name); if (!reader) return false; - return disk->existsFile(getRelativeDataPath()) && reader->exists(file_name); + return existsFileInTransactionOrDisk(getRelativeDataPath()) && reader->exists(file_name); } size_t DataPartStorageOnDiskPacked::getFileSizeImpl(const String & file_name) const @@ -233,16 +237,15 @@ size_t DataPartStorageOnDiskPacked::getFileSizeImpl(const String & file_name) co return skip_reader->getFileSize(file_name); } - auto disk = volume->getDisk(); if (isWrittenSeparately(file_name)) - return disk->getFileSize(fs::path(root_path) / part_dir / file_name); + return getFileSizeInTransactionOrDisk(fs::path(root_path) / part_dir / file_name); /// The archive itself is a physical file on disk; its size can be read directly /// without going through the packed index (which only knows individual file sizes). if (file_name == DATA_FILE_NAME) - return disk->getFileSize(fs::path(root_path) / part_dir / DATA_FILE_NAME); + return getFileSizeInTransactionOrDisk(fs::path(root_path) / part_dir / DATA_FILE_NAME); - if (!disk->existsFile(getRelativeDataPath())) + if (!existsFileInTransactionOrDisk(getRelativeDataPath())) throw Exception(ErrorCodes::FILE_DOESNT_EXIST, "File {} does not exist, cannot get file size of {}", getRelativeDataPath(), file_name); if (!reader) @@ -254,15 +257,13 @@ size_t DataPartStorageOnDiskPacked::getFileSizeImpl(const String & file_name) co PackedFilesIO::FileOffset DataPartStorageOnDiskPacked::getFileOffsetAndSize(const std::string & file_name) const { - auto disk = volume->getDisk(); - if (file_name == DATA_FILE_NAME) - return {0, disk->getFileSize(fs::path(root_path) / part_dir / DATA_FILE_NAME)}; + return {0, getFileSizeInTransactionOrDisk(fs::path(root_path) / part_dir / DATA_FILE_NAME)}; if (isWrittenSeparately(file_name)) - return {0, disk->getFileSize(fs::path(root_path) / part_dir / file_name)}; + return {0, getFileSizeInTransactionOrDisk(fs::path(root_path) / part_dir / file_name)}; - if (!disk->existsFile(getRelativeDataPath())) + if (!existsFileInTransactionOrDisk(getRelativeDataPath())) throw Exception(ErrorCodes::FILE_DOESNT_EXIST, "File {} does not exist, cannot get file size of {}", getRelativeDataPath(), file_name); if (!reader) @@ -297,7 +298,7 @@ std::vector DataPartStorageOnDiskPacked::getRemotePaths(const std:: if (isWrittenSeparately(file_name)) file_path = fs::path(root_path) / part_dir / file_name; - auto objects = volume->getDisk()->getStorageObjects(file_path); + auto objects = getStorageObjectsInTransactionOrDisk(file_path); std::vector remote_paths; remote_paths.reserve(objects.size()); @@ -341,8 +342,7 @@ void DataPartStorageOnDiskPacked::prepareReadImpl( [this, file_name = name, inner](const StoredObject &, const ReadSettings & s, bool, bool) -> std::unique_ptr { - auto outer_buf = reader->readFile( - volume->getDisk(), getRelativeDataPath(), String(SKIP_INDICES_PACKED_FILENAME), s, std::nullopt); + auto outer_buf = readArchiveFile(String(SKIP_INDICES_PACKED_FILENAME), s, std::nullopt); return std::make_unique(std::move(outer_buf), file_name, inner.offset, inner.offset + inner.size); }; pipeline.setSource(std::move(creator), StoredObjects{StoredObject{}}, settings); @@ -352,7 +352,7 @@ void DataPartStorageOnDiskPacked::prepareReadImpl( if (isWrittenSeparately(name)) { - volume->getDisk()->prepareRead(fs::path(root_path) / part_dir / name, settings, read_hint, pipeline); + prepareReadInTransactionOrDisk(fs::path(root_path) / part_dir / name, settings, read_hint, pipeline); return; } @@ -365,7 +365,7 @@ void DataPartStorageOnDiskPacked::prepareReadImpl( [this, file_name = name, read_hint](const StoredObject &, const ReadSettings & read_settings, bool, bool) -> std::unique_ptr { - return reader->readFile(volume->getDisk(), getRelativeDataPath(), file_name, read_settings, read_hint); + return readArchiveFile(file_name, read_settings, read_hint); }, StoredObjects{StoredObject(name, "", reader->getFileSize(name))}, settings); @@ -523,9 +523,11 @@ void DataPartStorageOnDiskPacked::createProjection(const std::string & name) void DataPartStorageOnDiskPacked::beginTransaction() { + if (has_shared_transaction) + return; + if (transaction || writer) - throw Exception(ErrorCodes::LOGICAL_ERROR, - "Uncommitted{}transaction already exists", has_shared_transaction ? " shared " : " "); + throw Exception(ErrorCodes::LOGICAL_ERROR, "Uncommitted transaction already exists"); transaction = volume->getDisk()->createTransaction(); resetWriterFromTransaction(); @@ -548,15 +550,20 @@ void DataPartStorageOnDiskPacked::commitTransaction() if (!transaction || (!writer && !is_precommitted)) throw Exception(ErrorCodes::LOGICAL_ERROR, "There is no uncommitted transaction"); - if (has_shared_transaction) - throw Exception(ErrorCodes::LOGICAL_ERROR, "Cannot commit shared transaction"); - if (!is_precommitted) { finalizeWriter(); is_precommitted = true; } + if (has_shared_transaction) + return; + + fiu_do_on(FailPoints::part_storage_fail_commit_transaction, + { + throw Exception(ErrorCodes::FAULT_INJECTED, "part_storage_fail_commit_transaction"); + }); + transaction->commit(); if (!reader) @@ -566,21 +573,29 @@ void DataPartStorageOnDiskPacked::commitTransaction() is_precommitted = false; } +void DataPartStorageOnDiskPacked::checkpointTransaction() +{ + if (isContentAddressed()) + return; + + IDataPartStorage::checkpointTransaction(); +} + #if CLICKHOUSE_CLOUD TransactionCommitOutcomeVariant DataPartStorageOnDiskPacked::tryCommitTransaction(const TransactionCommitOptionsVariant & options) { if (!transaction || (!writer && !is_precommitted)) throw Exception(ErrorCodes::LOGICAL_ERROR, "There is no uncommitted transaction"); - if (has_shared_transaction) - throw Exception(ErrorCodes::LOGICAL_ERROR, "Cannot commit shared transaction"); - if (!is_precommitted) { finalizeWriter(); is_precommitted = true; } + if (has_shared_transaction) + return true; + auto result = transaction->tryCommit(options); if (!mayRetryCommit(options, result)) @@ -601,7 +616,7 @@ void DataPartStorageOnDiskPacked::undoTransaction() return; if (has_shared_transaction) - throw Exception(ErrorCodes::LOGICAL_ERROR, "Cannot commit shared transaction"); + return; transaction->undo(); transaction.reset(); @@ -617,6 +632,34 @@ String DataPartStorageOnDiskPacked::getRelativeDataPath() const return fs::path(root_path) / part_dir / DATA_FILE_NAME; } +std::unique_ptr DataPartStorageOnDiskPacked::readArchive( + const String & archive_path, + const ReadSettings & settings, + std::optional read_hint) const +{ + auto buffer = readFileIfExistsInTransactionOrDisk(archive_path, PackedFilesReader::patchSettings(settings), read_hint); + if (!buffer) + throw Exception(ErrorCodes::FILE_DOESNT_EXIST, "File {} does not exist", archive_path); + return buffer; +} + +std::unique_ptr DataPartStorageOnDiskPacked::readArchive( + const ReadSettings & settings, + std::optional read_hint) const +{ + return readArchive(getRelativeDataPath(), settings, read_hint); +} + +std::unique_ptr DataPartStorageOnDiskPacked::readArchiveFile( + const String & file_name, + const ReadSettings & settings, + std::optional read_hint) const +{ + if (!reader) + throw Exception(ErrorCodes::NOT_INITIALIZED, "Cannot read file, because reader is not initialized"); + return reader->readFile(readArchive(settings, read_hint), file_name); +} + bool DataPartStorageOnDiskPacked::isWrittenSeparately(const String & file_name) const { if (files_written_separately.contains(file_name)) @@ -661,8 +704,11 @@ void DataPartStorageOnDiskPacked::resetReader(const ReadSettings & read_settings return; auto data_path = getRelativeDataPath(); - if (volume->getDisk()->existsFile(data_path)) - reader.emplace(volume->getDisk(), data_path, read_settings); + if (existsFileInTransactionOrDisk(data_path)) + { + auto buffer = readArchive(read_settings, std::nullopt); + reader.emplace(*buffer); + } } std::shared_ptr DataPartStorageOnDiskPacked::getSkipIndicesPackedReader() const @@ -695,13 +741,13 @@ std::shared_ptr DataPartStorageOnDiskPacked::getSkipInd try { auto inner_archive_buf = reader->readFile( - volume->getDisk(), data_path, String(SKIP_INDICES_PACKED_FILENAME), getReadSettings(), std::nullopt); + readArchive(data_path, getReadSettings(), std::nullopt), String(SKIP_INDICES_PACKED_FILENAME)); auto inner_index = PackedFilesReader::readIndex(*inner_archive_buf); seedSkipIndicesPackedReader(inner_index); } catch (const Exception &) { - if (volume->getDisk()->existsFile(data_path)) + if (existsFileInTransactionOrDisk(data_path)) throw; } } @@ -748,7 +794,7 @@ void DataPartStorageOnDiskPacked::finalizeWriter() if (writer->isWritten(name)) continue; - auto in = reader->readFile(volume->getDisk(), getRelativeDataPath(), name, {}, {}); + auto in = readArchiveFile(name, {}, {}); auto out = writer->writeFile(name); copyData(*in, *out); @@ -927,7 +973,7 @@ MutableDataPartStoragePtr DataPartStorageOnDiskPacked::freeze( continue; } - auto read_buf = reader->readFile(volume->getDisk(), getRelativeDataPath(), file, read_settings, {}); + auto read_buf = readArchiveFile(file, read_settings, {}); auto write_buf = dest_storage->writeFile(file, DBMS_DEFAULT_BUFFER_SIZE, WriteMode::Rewrite, write_settings); copyData(*read_buf, *write_buf); write_buf->finalize(); @@ -1067,7 +1113,7 @@ MutableDataPartStoragePtr DataPartStorageOnDiskPacked::freezeRemote( continue; } - auto read_buf = reader->readFile(volume->getDisk(), getRelativeDataPath(), file, read_settings, {}); + auto read_buf = readArchiveFile(file, read_settings, {}); auto write_buf = dest_storage->writeFile(file, DBMS_DEFAULT_BUFFER_SIZE, WriteMode::Rewrite, write_settings); copyData(*read_buf, *write_buf); write_buf->finalize(); @@ -1098,7 +1144,7 @@ MutableDataPartStoragePtr DataPartStorageOnDiskPacked::freezeRemote( { /// Transactions doesn't support copy between different metadata storages, so doing it manually auto write_buf = dest_storage->transaction->writeFile(dest_storage->getRelativeDataPath(), DBMS_DEFAULT_BUFFER_SIZE, WriteMode::Rewrite, write_settings); - auto read_buf = src_disk->readFile(getRelativeDataPath(), read_settings); + auto read_buf = readArchive(read_settings, std::nullopt); copyData(*read_buf, *write_buf); write_buf->finalize(); } diff --git a/src/Storages/MergeTree/DataPartStorageOnDiskPacked.h b/src/Storages/MergeTree/DataPartStorageOnDiskPacked.h index 689da32f836d4..395ff00438d5a 100644 --- a/src/Storages/MergeTree/DataPartStorageOnDiskPacked.h +++ b/src/Storages/MergeTree/DataPartStorageOnDiskPacked.h @@ -81,6 +81,7 @@ class DataPartStorageOnDiskPacked final : public DataPartStorageOnDiskBase void beginTransaction() override; void commitTransaction() override; + void checkpointTransaction() override; void setPreferredFileOrder(const Strings & file_names) override { preferred_file_order = file_names; } @@ -130,6 +131,17 @@ class DataPartStorageOnDiskPacked final : public DataPartStorageOnDiskBase String getRelativeDataPath() const; bool isWrittenSeparately(const String & file_name) const; + std::unique_ptr readArchive( + const String & archive_path, + const ReadSettings & settings, + std::optional read_hint) const; + std::unique_ptr readArchive( + const ReadSettings & settings, + std::optional read_hint) const; + std::unique_ptr readArchiveFile( + const String & file_name, + const ReadSettings & settings, + std::optional read_hint) const; /// True if any of the given files lives inside data.packed (rather than being written separately), /// which forces the whole archive to be copied rather than hardlinked during a clone/freeze. diff --git a/src/Storages/MergeTree/FutureMergedMutatedPart.cpp b/src/Storages/MergeTree/FutureMergedMutatedPart.cpp index 7a8f7f77e6288..eccedb413054c 100644 --- a/src/Storages/MergeTree/FutureMergedMutatedPart.cpp +++ b/src/Storages/MergeTree/FutureMergedMutatedPart.cpp @@ -105,6 +105,7 @@ void FutureMergedMutatedPart::updatePath(const MergeTreeData & storage, const IR { path = fs::path(storage.getFullPathOnDisk(reservation->getDisk())) / name; path += "/"; + part_format = MergeTreeData::normalizePartFormatForDisk(part_format, reservation->getDisk()); } } diff --git a/src/Storages/MergeTree/MergeTreeData.cpp b/src/Storages/MergeTree/MergeTreeData.cpp index dc7c3710dc638..09541e0891936 100644 --- a/src/Storages/MergeTree/MergeTreeData.cpp +++ b/src/Storages/MergeTree/MergeTreeData.cpp @@ -6437,6 +6437,13 @@ MergeTreeDataPartFormat MergeTreeData::choosePartFormat( return {part_type, storage_type}; } +MergeTreeDataPartFormat MergeTreeData::normalizePartFormatForDisk(MergeTreeDataPartFormat format, const DiskPtr & disk) +{ + if (format.storage_type == MergeTreeDataPartStorageType::Packed && !disk->allowsPackedParts()) + format.storage_type = MergeTreeDataPartStorageType::Full; + return format; +} + MergeTreeDataPartBuilder MergeTreeData::getDataPartBuilder( const String & name, const VolumePtr & volume, const String & part_dir, const ReadSettings & read_settings_, PartDirIntent intent) const { diff --git a/src/Storages/MergeTree/MergeTreeData.h b/src/Storages/MergeTree/MergeTreeData.h index 26845f11083fa..28e23b7276cde 100644 --- a/src/Storages/MergeTree/MergeTreeData.h +++ b/src/Storages/MergeTree/MergeTreeData.h @@ -349,7 +349,7 @@ class MergeTreeData : public WithMutableContext, public IStorage, public IBackgr MergeTreeDataPartFormat choosePartFormat(size_t bytes_uncompressed, size_t rows_count, UInt32 part_level, ProjectionDescriptionRawPtr projection) const; - MergeTreeDataPartFormat choosePartFormatOnDisk(size_t bytes_uncompressed, size_t rows_count) const; + static MergeTreeDataPartFormat normalizePartFormatForDisk(MergeTreeDataPartFormat format, const DiskPtr & disk); /// return pair . Sometimes we may fail to load existing part (network issues, oom and so on), /// in this case pair of is returned. diff --git a/src/Storages/MergeTree/MergeTreeDataPartBuilder.cpp b/src/Storages/MergeTree/MergeTreeDataPartBuilder.cpp index ef891c1a6d8af..cb7ab95dad104 100644 --- a/src/Storages/MergeTree/MergeTreeDataPartBuilder.cpp +++ b/src/Storages/MergeTree/MergeTreeDataPartBuilder.cpp @@ -162,6 +162,12 @@ MergeTreeDataPartBuilder & MergeTreeDataPartBuilder::withPartType(MergeTreeDataP MergeTreeDataPartBuilder & MergeTreeDataPartBuilder::withPartStorageType(MergeTreeDataPartStorageType storage_type_) { + if (volume + && intent == PartDirIntent::CreateFresh + && storage_type_ == MergeTreeDataPartStorageType::Packed + && !volume->getDisk()->allowsPackedParts()) + storage_type_ = MergeTreeDataPartStorageType::Full; + part_storage = getPartStorageByType(storage_type_, volume, root_path, part_dir, intent == PartDirIntent::OpenExisting, read_settings); return *this; } diff --git a/src/Storages/MergeTree/MergeTreeDataWriter.cpp b/src/Storages/MergeTree/MergeTreeDataWriter.cpp index 30626a63e9fec..e703c47e48a83 100644 --- a/src/Storages/MergeTree/MergeTreeDataWriter.cpp +++ b/src/Storages/MergeTree/MergeTreeDataWriter.cpp @@ -920,6 +920,7 @@ MergeTreeTemporaryPartPtr MergeTreeDataWriter::writeTempPartImpl( /// every subsequent load of the part. if (metadata_snapshot->hasUniqueKey()) part_format.storage_type = MergeTreeDataPartStorageType::Full; + part_format = MergeTreeData::normalizePartFormatForDisk(part_format, data_part_volume->getDisk()); auto new_data_part = data.getDataPartBuilder(part_name, data_part_volume, part_dir, getReadSettings(), PartDirIntent::CreateFresh) .withPartFormat(part_format) diff --git a/src/Storages/MergeTree/tests/gtest_projection_borrowed_transaction.cpp b/src/Storages/MergeTree/tests/gtest_projection_borrowed_transaction.cpp index ae079a61450fa..c63eb9c30d40c 100644 --- a/src/Storages/MergeTree/tests/gtest_projection_borrowed_transaction.cpp +++ b/src/Storages/MergeTree/tests/gtest_projection_borrowed_transaction.cpp @@ -1,12 +1,23 @@ #include #include +#include +#include +#include +#include #include +#include +#include +#include #include +#include + +#include #include #include #include +#include #include using namespace DB; @@ -40,6 +51,39 @@ namespace std::filesystem::remove_all(base_path, ec); } }; + + struct CASPackedStorageFixture + { + std::filesystem::path scratch_path; + ObjectStoragePtr object_storage; + std::shared_ptr metadata_storage; + std::shared_ptr disk; + VolumePtr volume; + const String root_path = "a11/a11a11a1-1111-4111-8111-111111111111"; + const String part_dir = "all_1_1_0"; + + CASPackedStorageFixture() + { + Cas::tests::ensureBlobUploadPoolForTest(); + const auto unique = std::to_string(::getpid()) + "_" + + std::to_string(reinterpret_cast(this)); + scratch_path = std::filesystem::temp_directory_path() / ("packed_cas_gtest_" + unique); + object_storage = Cas::tests::makeLocalObjectStorageForTest(); + auto settings = Cas::tests::makeSettingsForTest("packed", scratch_path); + metadata_storage = std::make_shared( + object_storage, "pool", "srv1", "", nullptr, settings); + metadata_storage->startup(); + + std::unordered_map cluster_registry = {{"main", {true, true, ""}}}; + std::unordered_map object_storage_registry = {{"main", object_storage}}; + auto cluster = std::make_shared("packed_cas", std::move(cluster_registry)); + auto object_storages = std::make_shared(std::move(object_storage_registry)); + Poco::AutoPtr config(new Poco::Util::XMLConfiguration()); + disk = std::make_shared( + "packed_cas", std::move(cluster), metadata_storage, std::move(object_storages), nullptr, *config, ""); + volume = std::make_shared("packed_cas", disk); + } + }; } /// A projection sub-part that BORROWS the parent's whole-part transaction (the CA-disk shape: @@ -84,3 +128,152 @@ TEST(ProjectionBorrowedTransaction, OwnedProjectionStorageStillBeginsAndCommits) EXPECT_NO_THROW(proj->commitTransaction()); EXPECT_FALSE(proj->hasActiveTransaction()); } + +TEST(ProjectionBorrowedTransaction, BorrowedPackedStorageFinalizesWithoutCommittingParent) +{ + ParentStorageFixture fx; + auto parent = std::make_shared(fx.volume, "", "all_1_1_0", ReadSettings{}, false); + + parent->beginTransaction(); + ASSERT_TRUE(parent->hasActiveTransaction()); + + auto parent_out = parent->writeFile("parent.bin", 4096, WriteMode::Rewrite, WriteSettings{}); + writeString("parent payload", *parent_out); + parent_out->finalize(); + + auto proj = parent->getProjectionNoInitialize("p.proj", true); + ASSERT_TRUE(proj->hasActiveTransaction()); + EXPECT_NO_THROW(proj->beginTransaction()); + proj->createDirectories(); + + auto out = proj->writeFile("value.bin", 4096, WriteMode::Rewrite, WriteSettings{}); + writeString("packed payload", *out); + out->finalize(); + + EXPECT_NO_THROW(proj->precommitTransaction()); + EXPECT_NO_THROW(proj->commitTransaction()); + EXPECT_TRUE(parent->hasActiveTransaction()); + + auto in = proj->readFile("value.bin", ReadSettings{}, std::nullopt); + String value; + readStringUntilEOF(value, *in); + EXPECT_EQ(value, "packed payload"); + + EXPECT_NO_THROW(parent->commitTransaction()); + EXPECT_FALSE(parent->hasActiveTransaction()); +} + +TEST(ProjectionBorrowedTransaction, BorrowedPackedStorageCommitFinalizesWriter) +{ + ParentStorageFixture fx; + auto parent = std::make_shared(fx.volume, "", "all_1_1_0", ReadSettings{}, false); + + parent->beginTransaction(); + auto parent_out = parent->writeFile("parent.bin", 4096, WriteMode::Rewrite, WriteSettings{}); + writeString("parent payload", *parent_out); + parent_out->finalize(); + + auto proj = parent->getProjectionNoInitialize("p.proj", true); + proj->beginTransaction(); + proj->createDirectories(); + + auto out = proj->writeFile("value.bin", 4096, WriteMode::Rewrite, WriteSettings{}); + writeString("packed payload", *out); + out->finalize(); + + EXPECT_NO_THROW(proj->commitTransaction()); + EXPECT_TRUE(parent->hasActiveTransaction()); + + auto in = proj->readFile("value.bin", ReadSettings{}, std::nullopt); + String value; + readStringUntilEOF(value, *in); + EXPECT_EQ(value, "packed payload"); + + EXPECT_NO_THROW(parent->commitTransaction()); + EXPECT_FALSE(parent->hasActiveTransaction()); +} + +TEST(ProjectionBorrowedTransaction, OwnedPackedStorageStillCommitsAndResets) +{ + ParentStorageFixture fx; + auto storage = std::make_shared(fx.volume, "", "all_1_1_0", ReadSettings{}, false); + + storage->beginTransaction(); + auto out = storage->writeFile("value.bin", 4096, WriteMode::Rewrite, WriteSettings{}); + writeString("packed payload", *out); + out->finalize(); + + EXPECT_NO_THROW(storage->commitTransaction()); + EXPECT_FALSE(storage->hasActiveTransaction()); + + auto in = storage->readFile("value.bin", ReadSettings{}, std::nullopt); + String value; + readStringUntilEOF(value, *in); + EXPECT_EQ(value, "packed payload"); +} + +TEST(ProjectionBorrowedTransaction, CASPackedStorageReadsStagedReplacement) +{ + CASPackedStorageFixture fx; + + { + auto parent = std::make_shared( + fx.volume, fx.root_path, fx.part_dir, ReadSettings{}, false); + parent->beginTransaction(); + parent->createDirectories(); + + auto parent_out = parent->writeFile("parent.bin", 4096, WriteMode::Rewrite, WriteSettings{}); + writeString("parent payload", *parent_out); + parent_out->finalize(); + + auto proj = parent->getProjectionNoInitialize("p.proj", true); + proj->beginTransaction(); + proj->createDirectories(); + auto out = proj->writeFile("value.bin", 4096, WriteMode::Rewrite, WriteSettings{}); + writeString("old payload", *out); + out->finalize(); + proj->commitTransaction(); + parent->commitTransaction(); + } + + auto parent = std::make_shared( + fx.volume, fx.root_path, fx.part_dir, ReadSettings{}, true); + parent->beginTransaction(); + auto proj = parent->getProjection("p.proj", true); + ASSERT_FALSE(proj->getRemotePaths("value.bin").empty()); + + auto out = proj->writeFile("value.bin", 4096, WriteMode::Rewrite, WriteSettings{}); + writeString("new payload", *out); + out->finalize(); + proj->commitTransaction(); + + EXPECT_TRUE(parent->hasActiveTransaction()); + EXPECT_TRUE(proj->getRemotePaths("value.bin").empty()); + + auto in = proj->readFile("value.bin", ReadSettings{}, std::nullopt); + String value; + readStringUntilEOF(value, *in); + EXPECT_EQ(value, "new payload"); + + parent->checkpointTransaction(); + EXPECT_TRUE(parent->hasActiveTransaction()); + + auto before_commit_parent = std::make_shared( + fx.volume, fx.root_path, fx.part_dir, ReadSettings{}, true); + auto before_commit_proj = before_commit_parent->getProjection("p.proj"); + auto before_commit_in = before_commit_proj->readFile("value.bin", ReadSettings{}, std::nullopt); + value.clear(); + readStringUntilEOF(value, *before_commit_in); + EXPECT_EQ(value, "old payload"); + + parent->commitTransaction(); + EXPECT_FALSE(parent->hasActiveTransaction()); + + auto committed_parent = std::make_shared( + fx.volume, fx.root_path, fx.part_dir, ReadSettings{}, true); + auto committed_proj = committed_parent->getProjection("p.proj"); + auto committed_in = committed_proj->readFile("value.bin", ReadSettings{}, std::nullopt); + value.clear(); + readStringUntilEOF(value, *committed_in); + EXPECT_EQ(value, "new payload"); +} diff --git a/tests/config/config.d/cas_s3_storage_policy_for_merge_tree_by_default.xml b/tests/config/config.d/cas_s3_storage_policy_for_merge_tree_by_default.xml index e8ac8c5dbe90d..3a747e1b8a25c 100644 --- a/tests/config/config.d/cas_s3_storage_policy_for_merge_tree_by_default.xml +++ b/tests/config/config.d/cas_s3_storage_policy_for_merge_tree_by_default.xml @@ -24,12 +24,27 @@ clickhouse cas_s3_scratch/ + 1 1 20 + + object_storage + s3 + cas + 30 + 10000 + stateless-ca-s3-packed-disabled + http://localhost:11121/test/cas_s3_packed_disabled/ + clickhouse + clickhouse + cas_s3_packed_disabled_scratch/ + 1 + 20 + cas_scratch/ + 1 1 20 + + object_storage + local + cas + stateless-ca-local-packed-disabled + cas_packed_disabled_pool/ + cas_packed_disabled_scratch/ + 1 + 20 + @@ -27,6 +38,16 @@ + + + + cas + + + cas_packed_disabled + + + diff --git a/tests/docker_scripts/stress_runner.sh b/tests/docker_scripts/stress_runner.sh index 732ebdc955e0b..0d013452f68a7 100755 --- a/tests/docker_scripts/stress_runner.sh +++ b/tests/docker_scripts/stress_runner.sh @@ -52,11 +52,6 @@ cd /repo && python3 /repo/ci/jobs/scripts/clickhouse_proc.py logs_export_config cd /repo && python3 /repo/ci/jobs/scripts/clickhouse_proc.py start_seaweedfs stateless || { echo "Failed to start seaweedfs"; exit 1; } cd /repo && python3 /repo/ci/jobs/scripts/clickhouse_proc.py start_azurite || { echo "Failed to start azurite"; exit 1; } -if [[ "$USE_CAS_S3_STORAGE_FOR_MERGE_TREE" == "1" ]]; then - # CAS-over-S3 needs RustFS (enforced If-Match deletes). MinIO stays up for - # the non-CAS s3 disks that EXPORT_S3_STORAGE_POLICIES still installs. - cd /repo && python3 /repo/ci/jobs/scripts/clickhouse_proc.py start_rustfs || { echo "Failed to start rustfs"; exit 1; } -fi # Start Redpanda (Kafka-compatible broker) so that Kafka engine tests work and # do not leave behind broken StorageKafka tables whose background threads cause @@ -113,10 +108,7 @@ clickhouse-client --query "SHOW TABLES FROM tpcds" clickhouse-client --query "SHOW TABLES FROM tpch" clickhouse-client --query "SHOW TABLES FROM test" -if [[ "$USE_CAS_S3_STORAGE_FOR_MERGE_TREE" == "1" ]]; then - TEMP_POLICY="cas_s3" - echo "Using cas_s3 storage policy" -elif [[ "$USE_S3_STORAGE_FOR_MERGE_TREE" == "1" ]]; then +if [[ "$USE_S3_STORAGE_FOR_MERGE_TREE" == "1" ]]; then TEMP_POLICY="s3_cache" elif [[ "$USE_AZURE_STORAGE_FOR_MERGE_TREE" == "1" ]]; then TEMP_POLICY="azure_cache" @@ -349,9 +341,6 @@ start_server 10 || { echo "Failed to start server"; exit 1; } check_server_start -# The server may be killed rather than shut down, so don't rely on the shutdown flush. -clickhouse-client --receive_timeout 30 -q "SYSTEM FLUSH LOGS" ||: - stop_server [ -f /var/log/clickhouse-server/clickhouse-server.log ] || echo -e "Server log does not exist\tFAIL" diff --git a/tests/docker_scripts/stress_tests.lib b/tests/docker_scripts/stress_tests.lib index 2b8dd60557233..a4eec918a339f 100644 --- a/tests/docker_scripts/stress_tests.lib +++ b/tests/docker_scripts/stress_tests.lib @@ -415,18 +415,6 @@ function check_logs_for_critical_errors() function collect_query_and_trace_logs() { - # A writable open of a CAS disk claims the server-root and fails closed against the - # real server's owner uuid; read-only skips the claim, which is all a dump needs. - # `--follow-symlinks`: install.sh symlinks these configs. See dump_system_tables. - grep -Rl 'cas' /etc/clickhouse-server/ 2>/dev/null \ - | xargs -r sed -i --follow-symlinks 's|cas|castrue|g' - - # A declared but not read-only CAS disk means this scrape is about to die on ownership. - if grep -Rlq 'cas' /etc/clickhouse-server/ 2>/dev/null \ - && ! grep -Rlq 'castrue' /etc/clickhouse-server/ 2>/dev/null; then - echo "WARNING: a CAS disk is declared but the read-only marker was not inserted -- clickhouse-local will claim server-root ownership and this scrape will fail" - fi - for table in query_log trace_log metric_log aggregated_zookeeper_log do # Don't ignore errors here, it leads to ignore sanitizer reports when running clickhouse-local diff --git a/tests/integration/test_cas_s3/configs/storage_conf.xml b/tests/integration/test_cas_s3/configs/storage_conf.xml index 773441a00944d..65bc0a2cc5237 100644 --- a/tests/integration/test_cas_s3/configs/storage_conf.xml +++ b/tests/integration/test_cas_s3/configs/storage_conf.xml @@ -5,6 +5,7 @@ object_storage s3 cas + 1 itest-content-addressed-s3