From 20ccae3dedb06c665500bdf7c3b3ef37b1b82808 Mon Sep 17 00:00:00 2001 From: Konstantin Morozov Date: Tue, 13 Oct 2026 16:12:58 +0200 Subject: [PATCH 01/34] Fix count for memory-backed file views --- src/IO/ReadBufferFromFileView.cpp | 26 +++++++++++++++ src/IO/ReadBufferFromFileView.h | 1 + .../gtest_read_buffer_from_file_view.cpp | 32 +++++++++++++++++++ 3 files changed, 59 insertions(+) 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..1f1879a56e438 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(); From eb6faaab70ce432a07db03b3c32cdfb2a34f6db1 Mon Sep 17 00:00:00 2001 From: Konstantin Morozov Date: Tue, 13 Oct 2026 16:35:22 +0200 Subject: [PATCH 02/34] Add opt-in capability for CAS packed parts --- src/Disks/DiskEncrypted.h | 1 + .../DiskObjectStorage/DiskObjectStorage.cpp | 5 +++++ src/Disks/DiskObjectStorage/DiskObjectStorage.h | 2 ++ .../ContentAddressedMetadataStorage.cpp | 2 ++ .../ContentAddressedMetadataStorage.h | 2 ++ .../ContentAddressedSettings.cpp | 1 + .../MetadataStorages/IMetadataStorage.h | 2 ++ src/Disks/IDisk.h | 2 ++ src/Disks/ReadOnlyDiskWrapper.h | 1 + src/Disks/tests/gtest_cas_settings.cpp | 16 ++++++++++++++++ 10 files changed, 34 insertions(+) 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/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_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"}) From 2b217efd2160afec0c9ff984606dd907bf659085 Mon Sep 17 00:00:00 2001 From: Konstantin Morozov Date: Tue, 13 Oct 2026 17:46:33 +0200 Subject: [PATCH 03/34] Gate packed part creation by disk capability --- src/Storages/MergeTree/MergeTreeData.cpp | 12 ++++++ .../MergeTree/MergeTreeDataPartBuilder.cpp | 6 +++ .../05217_cas_packed_parts.reference | 2 + .../0_stateless/05217_cas_packed_parts.sh | 42 +++++++++++++++++++ 4 files changed, 62 insertions(+) create mode 100644 tests/queries/0_stateless/05217_cas_packed_parts.reference create mode 100755 tests/queries/0_stateless/05217_cas_packed_parts.sh diff --git a/src/Storages/MergeTree/MergeTreeData.cpp b/src/Storages/MergeTree/MergeTreeData.cpp index e10ac17fc7c39..1b8c0eabfcfca 100644 --- a/src/Storages/MergeTree/MergeTreeData.cpp +++ b/src/Storages/MergeTree/MergeTreeData.cpp @@ -6287,6 +6287,18 @@ MergeTreeDataPartFormat MergeTreeData::choosePartFormat( if (satisfies((*settings)[MergeTreeSetting::min_bytes_for_full_part_storage], (*settings)[MergeTreeSetting::min_rows_for_full_part_storage], (*settings)[MergeTreeSetting::min_level_for_full_part_storage])) storage_type = PartStorageType::Packed; + if (storage_type == PartStorageType::Packed) + { + for (const auto & disk : getStoragePolicy()->getDisks()) + { + if (!disk->allowsPackedParts()) + { + storage_type = PartStorageType::Full; + break; + } + } + } + /// The trained `pq` method of the `Quantize(...)` codec stores a per-part codebook (one artifact for the whole /// column, trained at serialization suffix). A compact part serializes each granule with a fresh serialization /// state, which would train and write a separate codebook per granule and leave the reader unable to pair each row 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/tests/queries/0_stateless/05217_cas_packed_parts.reference b/tests/queries/0_stateless/05217_cas_packed_parts.reference new file mode 100644 index 0000000000000..c3b49229c34cd --- /dev/null +++ b/tests/queries/0_stateless/05217_cas_packed_parts.reference @@ -0,0 +1,2 @@ +cas_packed_disabled Full +cas_packed_local Packed diff --git a/tests/queries/0_stateless/05217_cas_packed_parts.sh b/tests/queries/0_stateless/05217_cas_packed_parts.sh new file mode 100755 index 0000000000000..a3bd13e7c83e0 --- /dev/null +++ b/tests/queries/0_stateless/05217_cas_packed_parts.sh @@ -0,0 +1,42 @@ +#!/usr/bin/env bash +# Tags: no-fasttest, no-random-merge-tree-settings + +CUR_DIR=$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd) +# shellcheck source=../shell_config.sh +. "$CUR_DIR"/../shell_config.sh + +${CLICKHOUSE_CLIENT} --multiquery < Date: Tue, 13 Oct 2026 17:53:07 +0200 Subject: [PATCH 04/34] Keep packed part archives as CAS blobs --- .../ContentAddressedTransaction.cpp | 6 ++++++ src/Disks/tests/gtest_cas_inline_placement.cpp | 13 +++++++++++++ 2 files changed, 19 insertions(+) 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/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")); +} From f0af879819956cd0d7421c5488f739e89497c0d8 Mon Sep 17 00:00:00 2001 From: Konstantin Morozov Date: Tue, 13 Oct 2026 18:23:11 +0200 Subject: [PATCH 05/34] Share transaction-aware part storage reads --- src/IO/PackedFilesReader.cpp | 23 ++- src/IO/PackedFilesReader.h | 6 + .../MergeTree/DataPartStorageOnDiskBase.cpp | 144 +++++++++++++++++ .../MergeTree/DataPartStorageOnDiskBase.h | 15 ++ .../MergeTree/DataPartStorageOnDiskFull.cpp | 152 +----------------- 5 files changed, 192 insertions(+), 148 deletions(-) diff --git a/src/IO/PackedFilesReader.cpp b/src/IO/PackedFilesReader.cpp index a81f7dce3fd92..76151e0cd7b11 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_)) { @@ -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..a9f15fee8f146 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,6 +50,10 @@ 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); diff --git a/src/Storages/MergeTree/DataPartStorageOnDiskBase.cpp b/src/Storages/MergeTree/DataPartStorageOnDiskBase.cpp index 352c4419d5f44..044f042777b5c 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 { @@ -1325,6 +1329,146 @@ 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); +} + +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; +}; + +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 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 +{ + StoredObjects objects; + if (transaction) + if (auto inflight = transaction->tryGetInFlightStorageObjects(path)) + objects = std::move(*inflight); + if (objects.empty()) + objects = volume->getDisk()->getStorageObjects(path); + return objects; +} + +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)) + inflight_objects = StoredObjects{StoredObject(path, path, *size)}; + + if (!inflight_objects.empty()) + { + auto * current_transaction = transaction.get(); + pipeline.setSource( + [current_transaction, path](const StoredObject &, const ReadSettings & read_settings, bool, bool) + { + return current_transaction->tryReadFileInFlight(path, read_settings, std::nullopt); + }, + 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) + if (auto buffer = transaction->tryReadFileInFlight(path, settings, read_hint)) + return buffer; + 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..c0f38204e089c 100644 --- a/src/Storages/MergeTree/DataPartStorageOnDiskBase.h +++ b/src/Storages/MergeTree/DataPartStorageOnDiskBase.h @@ -183,6 +183,21 @@ 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; + 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..015303175891d 100644 --- a/src/Storages/MergeTree/DataPartStorageOnDiskFull.cpp +++ b/src/Storages/MergeTree/DataPartStorageOnDiskFull.cpp @@ -62,101 +62,22 @@ 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); + 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); + 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); + 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 +87,7 @@ 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); + return getFileSizeInTransactionOrDisk(fs::path(root_path) / part_dir / file_name); } std::optional DataPartStorageOnDiskFull::getPackedFileUncompressedSize(const std::string & file_name) const @@ -191,16 +107,7 @@ 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); + auto objects = getStorageObjectsInTransactionOrDisk(path); std::vector remote_paths; remote_paths.reserve(objects.size()); @@ -226,43 +133,7 @@ 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); + prepareReadInTransactionOrDisk(fs::path(root_path) / part_dir / name, settings, read_hint, pipeline); } std::unique_ptr DataPartStorageOnDiskFull::readFileIfExistsImpl( @@ -270,14 +141,7 @@ 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); + return readFileIfExistsInTransactionOrDisk(fs::path(root_path) / part_dir / name, settings, read_hint); } std::unique_ptr DataPartStorageOnDiskFull::writeFile( From d8f44574b18783ccfa21dfdbc5d5b669936ce522 Mon Sep 17 00:00:00 2001 From: Konstantin Morozov Date: Tue, 13 Oct 2026 18:50:19 +0200 Subject: [PATCH 06/34] Support borrowed packed parts in CAS transactions --- src/IO/PackedFilesReader.cpp | 2 +- src/IO/PackedFilesReader.h | 1 + .../MergeTree/DataPartStorageOnDiskBase.cpp | 23 ++- .../MergeTree/DataPartStorageOnDiskPacked.cpp | 122 +++++++---- .../MergeTree/DataPartStorageOnDiskPacked.h | 12 ++ .../gtest_projection_borrowed_transaction.cpp | 193 ++++++++++++++++++ 6 files changed, 301 insertions(+), 52 deletions(-) diff --git a/src/IO/PackedFilesReader.cpp b/src/IO/PackedFilesReader.cpp index 76151e0cd7b11..57aeff991116d 100644 --- a/src/IO/PackedFilesReader.cpp +++ b/src/IO/PackedFilesReader.cpp @@ -101,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) diff --git a/src/IO/PackedFilesReader.h b/src/IO/PackedFilesReader.h index a9f15fee8f146..0faec5fef25a9 100644 --- a/src/IO/PackedFilesReader.h +++ b/src/IO/PackedFilesReader.h @@ -57,6 +57,7 @@ class PackedFilesReader 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/Storages/MergeTree/DataPartStorageOnDiskBase.cpp b/src/Storages/MergeTree/DataPartStorageOnDiskBase.cpp index 044f042777b5c..8e6043f0ef4b8 100644 --- a/src/Storages/MergeTree/DataPartStorageOnDiskBase.cpp +++ b/src/Storages/MergeTree/DataPartStorageOnDiskBase.cpp @@ -1418,13 +1418,14 @@ size_t DataPartStorageOnDiskBase::getFileSizeInTransactionOrDisk(const std::stri StoredObjects DataPartStorageOnDiskBase::getStorageObjectsInTransactionOrDisk(const std::string & path) const { - StoredObjects objects; if (transaction) + { if (auto inflight = transaction->tryGetInFlightStorageObjects(path)) - objects = std::move(*inflight); - if (objects.empty()) - objects = volume->getDisk()->getStorageObjects(path); - return objects; + return std::move(*inflight); + if (transaction->tryGetInFlightFileSize(path)) + return {}; + } + return volume->getDisk()->getStorageObjects(path); } void DataPartStorageOnDiskBase::prepareReadInTransactionOrDisk( @@ -1443,11 +1444,14 @@ void DataPartStorageOnDiskBase::prepareReadInTransactionOrDisk( if (!inflight_objects.empty()) { - auto * current_transaction = transaction.get(); + auto current_transaction = transaction; pipeline.setSource( [current_transaction, path](const StoredObject &, const ReadSettings & read_settings, bool, bool) { - return current_transaction->tryReadFileInFlight(path, read_settings, std::nullopt); + 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); @@ -1464,8 +1468,13 @@ std::unique_ptr DataPartStorageOnDiskBase::readFileIfExi 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 (in_flight_size) + throw Exception(ErrorCodes::LOGICAL_ERROR, "In-flight file {} cannot be read", path); + } return volume->getDisk()->readFileIfExists(path, settings, read_hint); } diff --git a/src/Storages/MergeTree/DataPartStorageOnDiskPacked.cpp b/src/Storages/MergeTree/DataPartStorageOnDiskPacked.cpp index 6ce082f9f90ab..48e3a087ff2cf 100644 --- a/src/Storages/MergeTree/DataPartStorageOnDiskPacked.cpp +++ b/src/Storages/MergeTree/DataPartStorageOnDiskPacked.cpp @@ -78,7 +78,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 +89,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 +97,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 +133,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 +172,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 +212,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 +230,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 +250,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 +291,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 +335,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 +345,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 +358,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); @@ -516,9 +509,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(); @@ -541,15 +536,15 @@ 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; + transaction->commit(); if (!reader) @@ -559,21 +554,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)) @@ -594,7 +597,7 @@ void DataPartStorageOnDiskPacked::undoTransaction() return; if (has_shared_transaction) - throw Exception(ErrorCodes::LOGICAL_ERROR, "Cannot commit shared transaction"); + return; transaction->undo(); transaction.reset(); @@ -610,6 +613,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)) @@ -654,8 +685,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 @@ -688,13 +722,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; } } @@ -741,7 +775,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); @@ -920,7 +954,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(); @@ -1060,7 +1094,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(); @@ -1091,7 +1125,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/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"); +} From 6d9733612c0b7d5026cb00800063ea85a0d00232 Mon Sep 17 00:00:00 2001 From: Konstantin Morozov Date: Tue, 13 Oct 2026 22:25:42 +0200 Subject: [PATCH 07/34] Expand CAS packed parts lifecycle coverage --- ...orage_policy_for_merge_tree_by_default.xml | 1 + ...orage_policy_for_merge_tree_by_default.xml | 1 + .../05217_cas_packed_parts.reference | 24 +++ .../0_stateless/05217_cas_packed_parts.sh | 204 ++++++++++++++++-- 4 files changed, 213 insertions(+), 17 deletions(-) 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..391fa24e1cef2 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,6 +24,7 @@ clickhouse cas_s3_scratch/ + 1 diff --git a/tests/config/config.d/cas_storage_policy_for_merge_tree_by_default.xml b/tests/config/config.d/cas_storage_policy_for_merge_tree_by_default.xml index 271a492103e31..ccf77b30037fa 100644 --- a/tests/config/config.d/cas_storage_policy_for_merge_tree_by_default.xml +++ b/tests/config/config.d/cas_storage_policy_for_merge_tree_by_default.xml @@ -12,6 +12,7 @@ cas_pool/ cas_scratch/ + 1 diff --git a/tests/queries/0_stateless/05217_cas_packed_parts.reference b/tests/queries/0_stateless/05217_cas_packed_parts.reference index c3b49229c34cd..50744644b01f0 100644 --- a/tests/queries/0_stateless/05217_cas_packed_parts.reference +++ b/tests/queries/0_stateless/05217_cas_packed_parts.reference @@ -1,2 +1,26 @@ cas_packed_disabled Full cas_packed_local Packed +initial_parts 4 Packed +aggregate 40000 799980000 19980000 +minmax_query 80 +merged_parts 1 Packed +projection_active 1 +projection_query 0 40 780000 +projection_query 1 40 780040 +projection_query 2 40 780080 +1 +unfreeze_parts 1 +partition_reload 40000 799980000 +table_reload 40000 799980000 +copy_rows 40000 799980000 +copy_storage Packed +copy_projection_active 1 +copy_after_source_drop 40000 799980000 +copy_after_reload 40000 799980000 +has_blob_put 1 +default_gc_drained 1 +default_fsck_unreachable 0 +default_fsck_dangling 0 +disabled_gc_drained 1 +disabled_fsck_unreachable 0 +disabled_fsck_dangling 0 diff --git a/tests/queries/0_stateless/05217_cas_packed_parts.sh b/tests/queries/0_stateless/05217_cas_packed_parts.sh index a3bd13e7c83e0..9ecedf9bf057a 100755 --- a/tests/queries/0_stateless/05217_cas_packed_parts.sh +++ b/tests/queries/0_stateless/05217_cas_packed_parts.sh @@ -1,42 +1,212 @@ #!/usr/bin/env bash -# Tags: no-fasttest, no-random-merge-tree-settings +# Tags: no-fasttest, no-random-merge-tree-settings, no-parallel + +set -euo pipefail CUR_DIR=$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd) # shellcheck source=../shell_config.sh . "$CUR_DIR"/../shell_config.sh +DISABLED_DISK_NAME="${CLICKHOUSE_DATABASE}_05217_disabled_${CLICKHOUSE_TEST_UNIQUE_NAME}_${RANDOM}" +DISABLED_POOL="${CLICKHOUSE_USER_FILES_UNIQUE}_05217_disabled_${RANDOM}" +UNFREEZE_STRUCTURE='command_type String, partition_id String, part_name String, backup_name String, backup_path String, part_backup_path String' + +rm -rf "${DISABLED_POOL:?}" +mkdir -p "${DISABLED_POOL}" + +drain_gc() +{ + local disk_name=$1 + local output_name=$2 + local pending=1 + + for _ in $(seq 1 60); do + pending=$(${CLICKHOUSE_CLIENT} --query "SYSTEM CAS GC RUN '${disk_name}'" --format TSVWithNames \ + | awk -F'\t' 'NR==1 { for (i = 1; i <= NF; i++) col[$i] = i; next } { print $col["pending_condemned"] }') + [ "${pending}" = "0" ] && break + sleep 0.5 + done + + if [ "${pending}" != "0" ]; then + echo "GC did not drain for ${disk_name}: pending=${pending}" >&2 + exit 1 + fi + + echo "${output_name}_gc_drained" 1 +} + +check_fsck() +{ + local disk_name=$1 + local output_name=$2 + + ${CLICKHOUSE_CLIENT} --query "SYSTEM CAS FSCK '${disk_name}'" --format TSVWithNames \ + | awk -F'\t' -v name="${output_name}" 'NR==1 { for (i = 1; i <= NF; i++) col[$i] = i; next } + { print name "_fsck_unreachable", $col["unreachable"]; print name "_fsck_dangling", $col["dangling"] }' +} + ${CLICKHOUSE_CLIENT} --multiquery < 0 + FROM system.cas_log + WHERE disk_name = '${MAIN_DISK}' + AND event_type = 'blob_put' + AND query_id = '${CLICKHOUSE_DATABASE}_05217_blob_put'" +else + printf 'has_blob_put\t1\n' +fi + +${CLICKHOUSE_CLIENT} --query "DROP TABLE cas_packed_copy SYNC" +${CLICKHOUSE_CLIENT} --query "DROP TABLE cas_packed_disabled SYNC" + +if [ "${MAIN_DISK_IS_CAS}" = "1" ]; then + drain_gc "${MAIN_DISK}" default + check_fsck "${MAIN_DISK}" default +else + echo "default_gc_drained" 1 + echo "default_fsck_unreachable" 0 + echo "default_fsck_dangling" 0 +fi + +drain_gc "${DISABLED_DISK_NAME}" disabled +check_fsck "${DISABLED_DISK_NAME}" disabled + +${CLICKHOUSE_CLIENT} --allow_repeated_settings --send_logs_level=fatal \ + --query "SYSTEM CAS FORGET '${DISABLED_DISK_NAME}'" + +rm -rf "${DISABLED_POOL:?}" From b412e19bc6538e86a61a7bb1c1641e3808234620 Mon Sep 17 00:00:00 2001 From: Konstantin Morozov Date: Thu, 15 Oct 2026 13:58:50 +0200 Subject: [PATCH 08/34] Add integration coverage for packed CAS parts --- .../test_cas_s3/configs/storage_conf.xml | 1 + tests/integration/test_cas_s3/test.py | 65 +++++++++++++++++++ 2 files changed, 66 insertions(+) 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