diff --git a/src/Common/ProfileEvents.cpp b/src/Common/ProfileEvents.cpp index e306e270f8d61..cd52f29a09537 100644 --- a/src/Common/ProfileEvents.cpp +++ b/src/Common/ProfileEvents.cpp @@ -500,6 +500,10 @@ The server successfully detected this situation and will download merged part fr M(DNSError, "Total count of errors in DNS resolution", ValueType::Number) \ M(PartsLockHoldMicroseconds, "Total time spent holding data parts lock in MergeTree tables", ValueType::Microseconds) \ M(PartsLockWaitMicroseconds, "Total time spent waiting for data parts lock in MergeTree tables", ValueType::Microseconds) \ + M(PartsLocks, "Number of times data parts lock has been acquired for MergeTree tables", ValueType::Number) \ + M(SharedPartsLockHoldMicroseconds, "Total time spent holding shared data parts lock in MergeTree tables", ValueType::Microseconds) \ + M(SharedPartsLockWaitMicroseconds, "Total time spent waiting for shared data parts lock in MergeTree tables", ValueType::Microseconds) \ + M(SharedPartsLocks, "Number of times shared data parts lock has been acquired for MergeTree tables", ValueType::Number) \ \ M(RealTimeMicroseconds, "Total (wall clock) time spent in processing (queries and other tasks) threads (note that this is a sum).", ValueType::Microseconds) \ M(UserTimeMicroseconds, "Total time spent in processing (queries and other tasks) threads executing CPU instructions in user mode. This includes time CPU pipeline was stalled due to main memory access, cache misses, branch mispredictions, hyper-threading, etc.", ValueType::Microseconds) \ diff --git a/src/Processors/QueryPlan/ReadFromMergeTree.cpp b/src/Processors/QueryPlan/ReadFromMergeTree.cpp index 757a37046efd5..e0dfb89a19953 100644 --- a/src/Processors/QueryPlan/ReadFromMergeTree.cpp +++ b/src/Processors/QueryPlan/ReadFromMergeTree.cpp @@ -310,7 +310,7 @@ void ReadFromMergeTree::AnalysisResult::checkLimits(const Settings & settings, c } ReadFromMergeTree::ReadFromMergeTree( - RangesInDataParts parts_, + RangesInDataPartsPtr parts_, MergeTreeData::MutationsSnapshotPtr mutations_, Names all_column_names_, const MergeTreeData & data_, @@ -395,7 +395,8 @@ std::unique_ptr ReadFromMergeTree::createLocalParallelReplica { const bool enable_parallel_reading = true; return std::make_unique( - getParts(), + /// Optimized version of getParts() to avoid extra copy + analyzed_result_ptr ? std::make_shared(analyzed_result_ptr->parts_with_ranges) : prepared_parts, mutations_snapshot, all_column_names, data, @@ -1918,7 +1919,7 @@ void ReadFromMergeTree::applyFilters(ActionDAGNodes added_filter_nodes) } ReadFromMergeTree::AnalysisResultPtr ReadFromMergeTree::selectRangesToRead( - RangesInDataParts parts, + const RangesInDataParts & parts, MergeTreeData::MutationsSnapshotPtr mutations_snapshot, const std::optional & vector_search_parameters, const StorageMetadataPtr & metadata_snapshot, @@ -1933,6 +1934,7 @@ ReadFromMergeTree::AnalysisResultPtr ReadFromMergeTree::selectRangesToRead( bool find_exact_ranges) { AnalysisResult result; + RangesInDataParts res_parts; const auto & settings = context_->getSettingsRef(); size_t total_parts = parts.size(); @@ -1993,7 +1995,7 @@ ReadFromMergeTree::AnalysisResultPtr ReadFromMergeTree::selectRangesToRead( bool add_index_stat_row_for_pk_expand = false; { - MergeTreeDataSelectExecutor::filterPartsByPartition( + res_parts = MergeTreeDataSelectExecutor::filterPartsByPartition( parts, indexes->partition_pruner, indexes->minmax_idx_condition, @@ -2008,7 +2010,7 @@ ReadFromMergeTree::AnalysisResultPtr ReadFromMergeTree::selectRangesToRead( result.sampling = MergeTreeDataSelectExecutor::getSampling( query_info_, metadata_snapshot->getColumns().getAllPhysical(), - parts, + res_parts, indexes->key_condition, data, metadata_snapshot, @@ -2018,13 +2020,13 @@ ReadFromMergeTree::AnalysisResultPtr ReadFromMergeTree::selectRangesToRead( if (result.sampling.read_nothing) return std::make_shared(std::move(result)); - for (const auto & part : parts) + for (const auto & part : res_parts) total_marks_pk += part.data_part->index_granularity->getMarksCountWithoutFinal(); - parts_before_pk = parts.size(); + parts_before_pk = res_parts.size(); auto reader_settings = MergeTreeReaderSettings::create(context_, query_info_); result.parts_with_ranges = MergeTreeDataSelectExecutor::filterPartsByPrimaryKeyAndSkipIndexes( - std::move(parts), + std::move(res_parts), metadata_snapshot, mutations_snapshot, context_, diff --git a/src/Processors/QueryPlan/ReadFromMergeTree.h b/src/Processors/QueryPlan/ReadFromMergeTree.h index 73f523772b47d..b782f2762e82a 100644 --- a/src/Processors/QueryPlan/ReadFromMergeTree.h +++ b/src/Processors/QueryPlan/ReadFromMergeTree.h @@ -139,7 +139,7 @@ class ReadFromMergeTree final : public SourceStepWithFilter using AnalysisResultPtr = std::shared_ptr; ReadFromMergeTree( - RangesInDataParts parts_, + RangesInDataPartsPtr parts_, MergeTreeData::MutationsSnapshotPtr mutations_snapshot_, Names all_column_names_, const MergeTreeData & data_, @@ -206,7 +206,7 @@ class ReadFromMergeTree final : public SourceStepWithFilter }; static AnalysisResultPtr selectRangesToRead( - RangesInDataParts parts, + const RangesInDataParts & parts, MergeTreeData::MutationsSnapshotPtr mutations_snapshot, const std::optional & vector_search_parameters, const StorageMetadataPtr & metadata_snapshot, @@ -246,7 +246,7 @@ class ReadFromMergeTree final : public SourceStepWithFilter AnalysisResultPtr getAnalyzedResult() const { return analyzed_result_ptr; } void setAnalyzedResult(AnalysisResultPtr analyzed_result_ptr_) { analyzed_result_ptr = std::move(analyzed_result_ptr_); } - const RangesInDataParts & getParts() const { return analyzed_result_ptr ? analyzed_result_ptr->parts_with_ranges : prepared_parts; } + const RangesInDataParts & getParts() const { return analyzed_result_ptr ? analyzed_result_ptr->parts_with_ranges : *prepared_parts; } MergeTreeData::MutationsSnapshotPtr getMutationsSnapshot() const { return mutations_snapshot; } const MergeTreeData & getMergeTreeData() const { return data; } @@ -269,7 +269,7 @@ class ReadFromMergeTree final : public SourceStepWithFilter private: MergeTreeReaderSettings reader_settings; - RangesInDataParts prepared_parts; + RangesInDataPartsPtr prepared_parts; MergeTreeData::MutationsSnapshotPtr mutations_snapshot; Names all_column_names; diff --git a/src/Storages/MergeTree/Compaction/PartsCollectors/MergeTreePartsCollector.cpp b/src/Storages/MergeTree/Compaction/PartsCollectors/MergeTreePartsCollector.cpp index 92f5f1c6a7813..d8b77e7612a5b 100644 --- a/src/Storages/MergeTree/Compaction/PartsCollectors/MergeTreePartsCollector.cpp +++ b/src/Storages/MergeTree/Compaction/PartsCollectors/MergeTreePartsCollector.cpp @@ -31,7 +31,7 @@ MergeTreeDataPartsVector collectInitial(const MergeTreeData & data, const MergeT MergeTreeDataPartsVector outdated_parts; { - auto lock = data.lockParts(); + auto lock = data.readLockParts(); active_parts = data.getDataPartsVectorForInternalUsage({MergeTreeData::DataPartState::Active}, affordable_kinds, lock); outdated_parts = data.getDataPartsVectorForInternalUsage({MergeTreeData::DataPartState::Outdated}, affordable_kinds, lock); } diff --git a/src/Storages/MergeTree/DataPartsExchange.cpp b/src/Storages/MergeTree/DataPartsExchange.cpp index f61bff5d3cdff..c21608dc6c3bf 100644 --- a/src/Storages/MergeTree/DataPartsExchange.cpp +++ b/src/Storages/MergeTree/DataPartsExchange.cpp @@ -389,7 +389,7 @@ MergeTreeData::DataPartPtr Service::findPart(const String & name) static const UInt32 wait_timeout_ms = 1000; auto pred = [&] () { - auto lock = data.lockParts(); + auto lock = data.readLockParts(); return part->getState() != MergeTreeDataPartState::PreActive; }; diff --git a/src/Storages/MergeTree/MergeTreeData.cpp b/src/Storages/MergeTree/MergeTreeData.cpp index 4acba33f3bb51..1755db7b1a7d7 100644 --- a/src/Storages/MergeTree/MergeTreeData.cpp +++ b/src/Storages/MergeTree/MergeTreeData.cpp @@ -162,6 +162,10 @@ namespace ProfileEvents extern const Event RejectedLightweightUpdates; extern const Event PartsLockWaitMicroseconds; extern const Event PartsLockHoldMicroseconds; + extern const Event PartsLocks; + extern const Event SharedPartsLockWaitMicroseconds; + extern const Event SharedPartsLockHoldMicroseconds; + extern const Event SharedPartsLocks; extern const Event LoadedDataParts; extern const Event LoadedDataPartsMicroseconds; extern const Event RestorePartsSkippedFiles; @@ -488,23 +492,44 @@ void MergeTreeData::initializeDirectoriesAndFormatVersion(const std::string & re throw Exception(ErrorCodes::METADATA_MISMATCH, "MergeTree data format version on disk doesn't support custom partitioning"); } } - - -DataPartsLock::DataPartsLock(std::mutex & data_parts_mutex_) +DataPartsLock::DataPartsLock(SharedMutex & data_parts_mutex_, const MergeTreeData * data_) : wait_watch(Stopwatch(CLOCK_MONOTONIC)) , lock(data_parts_mutex_) , lock_watch(Stopwatch(CLOCK_MONOTONIC)) + , data(data_) { ProfileEvents::increment(ProfileEvents::PartsLockWaitMicroseconds, wait_watch->elapsedMicroseconds()); + ProfileEvents::increment(ProfileEvents::PartsLocks); } DataPartsLock::~DataPartsLock() { + if (data) + { + data->shared_parts_list.reset(); + data->shared_ranges_in_parts.reset(); + } if (lock_watch.has_value()) ProfileEvents::increment(ProfileEvents::PartsLockHoldMicroseconds, lock_watch->elapsedMicroseconds()); } +DataPartsSharedLock::DataPartsSharedLock(DB::SharedMutex & data_parts_mutex_) + : wait_watch(Stopwatch(CLOCK_MONOTONIC)) + , lock(data_parts_mutex_) + , lock_watch(Stopwatch(CLOCK_MONOTONIC)) +{ + ProfileEvents::increment(ProfileEvents::SharedPartsLockWaitMicroseconds, wait_watch->elapsedMicroseconds()); + ProfileEvents::increment(ProfileEvents::SharedPartsLocks); +} + + +DataPartsSharedLock::~DataPartsSharedLock() +{ + if (lock_watch.has_value()) + ProfileEvents::increment(ProfileEvents::SharedPartsLockHoldMicroseconds, lock_watch->elapsedMicroseconds()); +} + static Int64 extractVersion(const auto & versions, const String & partition_id, Int64 default_value) { if (!versions) @@ -811,7 +836,7 @@ ConditionSelectivityEstimator MergeTreeData::getConditionSelectivityEstimatorByP if (!local_context->getSettingsRef()[Setting::allow_statistics_optimize]) return {}; - const auto & parts = assert_cast(*storage_snapshot->data).parts; + const auto & parts = *assert_cast(*storage_snapshot->data).parts; if (parts.empty() || !filter_dag) return {}; @@ -1683,7 +1708,7 @@ String MergeTreeData::MergingParams::getModeName() const Int64 MergeTreeData::getMaxBlockNumber() const { - auto lock = lockParts(); + auto lock = readLockParts(); Int64 max_block_num = 0; for (const DataPartPtr & part : data_parts_by_info) @@ -1859,7 +1884,7 @@ MergeTreeData::LoadPartResult MergeTreeData::loadDataPart( const String & part_name, const DiskPtr & part_disk_ptr, MergeTreeDataPartState to_state, - std::mutex & part_loading_mutex) + DB::SharedMutex & part_loading_mutex) { LOG_TRACE(log, "Loading {} part {} from disk {}", magic_enum::enum_name(to_state), part_name, part_disk_ptr->getName()); @@ -2037,7 +2062,7 @@ MergeTreeData::LoadPartResult MergeTreeData::loadDataPartWithRetries( const String & part_name, const DiskPtr & part_disk_ptr, MergeTreeDataPartState to_state, - std::mutex & part_loading_mutex, + DB::SharedMutex & part_loading_mutex, size_t initial_backoff_ms, size_t max_backoff_ms, size_t max_tries) @@ -2083,7 +2108,7 @@ std::vector MergeTreeData::loadDataPartsFromDisk( std::shuffle(parts_to_load.begin(), parts_to_load.end(), thread_local_rng); std::mutex part_select_mutex; - std::mutex part_loading_mutex; + DB::SharedMutex part_loading_mutex; std::vector loaded_parts; @@ -2402,7 +2427,7 @@ void MergeTreeData::loadDataParts(bool skip_sanity_checks, std::optionalindex_granularity_info.mark_type.adaptive; have_non_adaptive_parts |= !is_adaptive; @@ -2695,9 +2723,9 @@ try auto blocker_for_runner_thread = CannotAllocateThreadFaultInjector::blockFaultInjections(); auto res = loadDataPartWithRetries( - my_part->info, my_part->name, my_part->disk, - DataPartState::Outdated, data_parts_mutex, loading_parts_initial_backoff_ms, - loading_parts_max_backoff_ms, loading_parts_max_tries); + my_part->info, my_part->name, my_part->disk, + DataPartState::Outdated, data_parts_mutex, loading_parts_initial_backoff_ms, + loading_parts_max_backoff_ms, loading_parts_max_tries); ++num_loaded_parts; if (res.is_broken) @@ -4679,7 +4707,7 @@ MergeTreeData::PartsTemporaryRename::~PartsTemporaryRename() MergeTreeData::PartHierarchy MergeTreeData::getPartHierarchy( const MergeTreePartInfo & part_info, DataPartState state, - DataPartsLock & /* data_parts_lock */) const + const DataPartsAnyLock & /* data_parts_lock */) const { PartHierarchy result; @@ -4764,7 +4792,7 @@ MergeTreeData::DataPartsVector MergeTreeData::getActivePartsToReplace( const MergeTreePartInfo & new_part_info, const String & new_part_name, DataPartPtr & out_covering_part, - DataPartsLock & data_parts_lock) const + const DataPartsAnyLock & data_parts_lock) const { PartHierarchy hierarchy = getPartHierarchy(new_part_info, DataPartState::Active, data_parts_lock); @@ -4781,7 +4809,7 @@ MergeTreeData::DataPartsVector MergeTreeData::getActivePartsToReplace( return std::move(hierarchy.covered_parts); } -void MergeTreeData::checkPartPartition(MutableDataPartPtr & part, DataPartsLock & lock) const +void MergeTreeData::checkPartPartition(MutableDataPartPtr & part, const DataPartsAnyLock & lock) const { if (DataPartPtr existing_part_in_partition = getAnyPartInPartition(part->info.getPartitionId(), lock)) { @@ -4791,7 +4819,7 @@ void MergeTreeData::checkPartPartition(MutableDataPartPtr & part, DataPartsLock } } -void MergeTreeData::checkPartDuplicate(MutableDataPartPtr & part, Transaction & transaction, DataPartsLock & /*lock*/) const +void MergeTreeData::checkPartDuplicate(MutableDataPartPtr & part, Transaction & transaction, const DataPartsAnyLock & /*lock*/) const { auto it_duplicate = data_parts_by_info.find(part->info); @@ -4969,9 +4997,18 @@ MergeTreeData::DataPartsVector MergeTreeData::renameTempPartAndReplace( Transaction & out_transaction, bool rename_in_transaction) { - auto part_lock = lockParts(); + auto lock = lockParts(); + return renameTempPartAndReplaceUnlocked(part, lock, out_transaction, rename_in_transaction); +} + +MergeTreeData::DataPartsVector MergeTreeData::renameTempPartAndReplaceUnlocked( + MutableDataPartPtr & part, + DataPartsLock & lock, + Transaction & out_transaction, + bool rename_in_transaction) +{ DataPartsVector covered_parts; - renameTempPartAndReplaceImpl(part, out_transaction, part_lock, &covered_parts, rename_in_transaction); + renameTempPartAndReplaceImpl(part, out_transaction, lock, &covered_parts, rename_in_transaction); return covered_parts; } @@ -5025,10 +5062,14 @@ void MergeTreeData::removePartsFromWorkingSet(MergeTreeTransaction * txn, const resetObjectColumnsFromActiveParts(acquired_lock); } -void MergeTreeData::removePartsFromWorkingSetImmediatelyAndSetTemporaryState(const DataPartsVector & remove, DataPartsLock * acquired_lock) +void MergeTreeData::removePartsFromWorkingSetImmediatelyAndSetTemporaryState(const DataPartsVector & remove) { - auto lock = (acquired_lock) ? DataPartsLock() : lockParts(); + auto lock = lockParts(); + removePartsFromWorkingSetImmediatelyAndSetTemporaryState(remove, lock); +} +void MergeTreeData::removePartsFromWorkingSetImmediatelyAndSetTemporaryState(const DataPartsVector & remove, DataPartsLock & /*lock*/) +{ for (const auto & part : remove) { auto it_part = data_parts_by_info.find(part->info); @@ -5045,9 +5086,9 @@ void MergeTreeData::removePartsFromWorkingSetImmediatelyAndSetTemporaryState(con } void MergeTreeData::removePartsFromWorkingSet( - MergeTreeTransaction * txn, const DataPartsVector & remove, bool clear_without_timeout, DataPartsLock * acquired_lock) + MergeTreeTransaction * txn, const DataPartsVector & remove, bool clear_without_timeout) { - auto lock = (acquired_lock) ? DataPartsLock() : lockParts(); + auto lock = lockParts(); for (const auto & part : remove) { @@ -5067,14 +5108,14 @@ void MergeTreeData::removePartsInRangeFromWorkingSet(MergeTreeTransaction * txn, } DataPartsVector MergeTreeData::grabActivePartsToRemoveForDropRange( - MergeTreeTransaction * txn, const MergeTreePartInfo & drop_range, DataPartsLock & lock) + MergeTreeTransaction * txn, const MergeTreePartInfo & drop_range, const DataPartsAnyLock & lock) { DataPartsVector parts_to_remove; if (drop_range.min_block > drop_range.max_block) throw Exception(ErrorCodes::LOGICAL_ERROR, "Invalid drop range: {}", drop_range.getPartNameForLogs()); - auto partition_range = getVisibleDataPartsVectorInPartition(txn, drop_range.getPartitionId(), &lock); + auto partition_range = getVisibleDataPartsVectorInPartition(txn, drop_range.getPartitionId(), lock); for (const DataPartPtr & part : partition_range) { @@ -5157,7 +5198,7 @@ MergeTreeData::PartsToRemoveFromZooKeeper MergeTreeData::removePartsInRangeFromW /// Let's also reset timeout for inactive parts /// and add these parts to list of parts to remove from ZooKeeper auto inactive_parts_to_remove_immediately = getDataPartsVectorInPartitionForInternalUsage( - {DataPartState::Outdated, DataPartState::Deleting}, drop_range.getPartitionId(), &lock); + {DataPartState::Outdated, DataPartState::Deleting}, drop_range.getPartitionId(), lock); /// FIXME refactor removePartsFromWorkingSet(...), do not remove parts twice removePartsFromWorkingSet(txn, parts_to_remove, clear_without_timeout, lock); @@ -5199,7 +5240,7 @@ MergeTreeData::PartsToRemoveFromZooKeeper MergeTreeData::removePartsInRangeFromW renameTempPartAndAdd(new_data_part, transaction, lock, /*rename_in_transaction=*/ false); /// All covered parts must be already removed /// It will add the empty part to the set of Outdated parts without making it Active (exactly what we need) - transaction.rollback(&lock); + transaction.rollback(lock); new_data_part->remove_time.store(0, std::memory_order_relaxed); /// Such parts are always local, they don't participate in replication, they don't have shared blobs. /// So we don't have locks for shared data in zk for them, and can just remove blobs (this avoids leaving garbage in S3) @@ -5222,9 +5263,9 @@ MergeTreeData::PartsToRemoveFromZooKeeper MergeTreeData::removePartsInRangeFromW return parts_to_remove_from_zookeeper; } -void MergeTreeData::restoreAndActivatePart(const DataPartPtr & part, DataPartsLock * acquired_lock) +void MergeTreeData::restoreAndActivatePart(const DataPartPtr & part) { - auto lock = (acquired_lock) ? DataPartsLock() : lockParts(); + auto lock = lockParts(); if (part->getState() == DataPartState::Active) return; @@ -5240,10 +5281,10 @@ void MergeTreeData::outdateUnexpectedPartAndCloneToDetached(const DataPartPtr & LOG_INFO(log, "Cloning part {} to unexpected_{} and making it obsolete.", part_to_detach->getDataPartStorage().getPartDirectory(), part_to_detach->name); part_to_detach->makeCloneInDetached("unexpected", getInMemoryMetadataPtr(), /*disk_transaction*/ {}); - DataPartsLock lock = lockParts(); + auto lock = lockParts(); part_to_detach->is_unexpected_local_part = true; if (part_to_detach->getState() == DataPartState::Active) - removePartsFromWorkingSet(NO_TRANSACTION_RAW, {part_to_detach}, true, &lock); + removePartsFromWorkingSet(NO_TRANSACTION_RAW, {part_to_detach}, true, lock); } void MergeTreeData::forcefullyMovePartToDetachedAndRemoveFromMemory(const MergeTreeData::DataPartPtr & part_to_detach, const String & prefix) @@ -5373,7 +5414,7 @@ size_t MergeTreeData::getNumberOfOutdatedPartsWithExpiredRemovalTime() const auto time_now = time(nullptr); - auto parts_lock = lockParts(); + auto parts_lock = readLockParts(); auto outdated_parts_range = getDataPartsStateRange(DataPartState::Outdated); for (const auto & part : outdated_parts_range) { @@ -5387,7 +5428,7 @@ size_t MergeTreeData::getNumberOfOutdatedPartsWithExpiredRemovalTime() const std::pair MergeTreeData::getMaxPartsCountAndSizeForPartitionWithState(DataPartState state) const { - auto lock = lockParts(); + auto lock = readLockParts(); size_t cur_parts_count = 0; size_t cur_parts_size = 0; @@ -5433,7 +5474,7 @@ size_t MergeTreeData::getMaxOutdatedPartsCountForPartition() const std::optional MergeTreeData::getMinPartDataVersion() const { - auto lock = lockParts(); + auto lock = readLockParts(); std::optional result; for (const auto & part : getDataPartsStateRange(DataPartState::Active)) @@ -5624,7 +5665,7 @@ void MergeTreeData::throwLightweightUpdateIfNeeded(UInt64 added_uncompressed_byt } MergeTreeData::DataPartPtr MergeTreeData::getActiveContainingPart( - const MergeTreePartInfo & part_info, MergeTreeData::DataPartState state, DataPartsLock & /*lock*/) const + const MergeTreePartInfo & part_info, MergeTreeData::DataPartState state, const DataPartsAnyLock & /*lock*/) const { auto current_state_parts_range = getDataPartsStateRange(state); @@ -5696,7 +5737,7 @@ void MergeTreeData::swapActivePart(MergeTreeData::DataPartPtr part_copy, DataPar MergeTreeData::DataPartPtr MergeTreeData::getActiveContainingPart(const MergeTreePartInfo & part_info) const { - auto lock = lockParts(); + auto lock = readLockParts(); return getActiveContainingPart(part_info, DataPartState::Active, lock); } @@ -5706,7 +5747,7 @@ MergeTreeData::DataPartPtr MergeTreeData::getActiveContainingPart(const String & return getActiveContainingPart(part_info); } -MergeTreeData::DataPartPtr MergeTreeData::getActiveContainingPart(const String & part_name, DataPartsLock & lock) const +MergeTreeData::DataPartPtr MergeTreeData::getActiveContainingPart(const String & part_name, const DataPartsAnyLock & lock) const { auto part_info = MergeTreePartInfo::fromPartName(part_name, format_version); return getActiveContainingPart(part_info, DataPartState::Active, lock); @@ -5714,16 +5755,18 @@ MergeTreeData::DataPartPtr MergeTreeData::getActiveContainingPart(const String & MergeTreeData::DataPartsVector MergeTreeData::getVisibleDataPartsVectorInPartition(ContextPtr local_context, const String & partition_id) const { - return getVisibleDataPartsVectorInPartition(local_context->getCurrentTransaction().get(), partition_id); + auto lock = readLockParts(); + return getVisibleDataPartsVectorInPartition(local_context->getCurrentTransaction().get(), partition_id, lock); } MergeTreeData::DataPartsVector MergeTreeData::getVisibleDataPartsVectorInPartition( - ContextPtr local_context, const String & partition_id, DataPartsLock & lock) const + ContextPtr local_context, const String & partition_id, const DataPartsAnyLock & lock) const { - return getVisibleDataPartsVectorInPartition(local_context->getCurrentTransaction().get(), partition_id, &lock); + return getVisibleDataPartsVectorInPartition(local_context->getCurrentTransaction().get(), partition_id, lock); } -MergeTreeData::DataPartsVector MergeTreeData::getVisibleDataPartsVectorInPartition(MergeTreeTransaction * txn, const String & partition_id, DataPartsLock * acquired_lock) const +MergeTreeData::DataPartsVector MergeTreeData::getVisibleDataPartsVectorInPartition( + MergeTreeTransaction * txn, const String & partition_id, const DataPartsAnyLock & lock) const { if (txn) { @@ -5731,7 +5774,6 @@ MergeTreeData::DataPartsVector MergeTreeData::getVisibleDataPartsVectorInPartiti DataPartStateAndPartitionID outdated_parts{MergeTreeDataPartState::Outdated, partition_id}; DataPartsVector res; { - auto lock = (acquired_lock) ? DataPartsLock() : lockParts(); res.insert(res.end(), data_parts_by_state_and_info.lower_bound(active_parts), data_parts_by_state_and_info.upper_bound(active_parts)); res.insert(res.end(), data_parts_by_state_and_info.lower_bound(outdated_parts), data_parts_by_state_and_info.upper_bound(outdated_parts)); } @@ -5739,12 +5781,11 @@ MergeTreeData::DataPartsVector MergeTreeData::getVisibleDataPartsVectorInPartiti return res; } - return getDataPartsVectorInPartitionForInternalUsage(MergeTreeDataPartState::Active, partition_id, acquired_lock); + return getDataPartsVectorInPartitionForInternalUsage(MergeTreeDataPartState::Active, partition_id, lock); } -MergeTreeData::DataPartsVector MergeTreeData::getDataPartsVectorInPartitionForInternalUsage(const DataPartStates & affordable_states, const String & partition_id, DataPartsLock * acquired_lock) const +MergeTreeData::DataPartsVector MergeTreeData::getDataPartsVectorInPartitionForInternalUsage(const DataPartStates & affordable_states, const String & partition_id, const DataPartsAnyLock & /* lock */) const { - auto lock = (acquired_lock) ? DataPartsLock() : lockParts(); DataPartsVector res; for (const auto & state : affordable_states) { @@ -5754,11 +5795,11 @@ MergeTreeData::DataPartsVector MergeTreeData::getDataPartsVectorInPartitionForIn return res; } -MergeTreeData::DataPartsVector MergeTreeData::getDataPartsVectorInPartitionForInternalUsage(const MergeTreeData::DataPartState & state, const String & partition_id, DataPartsLock * acquired_lock) const +MergeTreeData::DataPartsVector MergeTreeData::getDataPartsVectorInPartitionForInternalUsage( + const MergeTreeData::DataPartState & state, const String & partition_id, const DataPartsAnyLock & /* lock */) const { DataPartStateAndPartitionID state_with_partition{state, partition_id}; - auto lock = (acquired_lock) ? DataPartsLock() : lockParts(); return DataPartsVector( data_parts_by_state_and_info.lower_bound(state_with_partition), data_parts_by_state_and_info.upper_bound(state_with_partition)); @@ -5769,7 +5810,7 @@ MergeTreeData::DataPartsVector MergeTreeData::getVisibleDataPartsVectorInPartiti auto txn = local_context->getCurrentTransaction(); DataPartsVector res; { - auto lock = lockParts(); + auto lock = readLockParts(); for (const auto & partition_id : partition_ids) { DataPartStateAndPartitionID active_parts{MergeTreeDataPartState::Active, partition_id}; @@ -5800,22 +5841,22 @@ MergeTreeData::DataPartsVector MergeTreeData::getVisibleDataPartsVectorInPartiti MergeTreeData::DataPartPtr MergeTreeData::getPartIfExists(const MergeTreePartInfo & part_info, const MergeTreeData::DataPartStates & valid_states) const { - auto lock = lockParts(); + auto lock = readLockParts(); return getPartIfExistsUnlocked(part_info, valid_states, lock); } MergeTreeData::DataPartPtr MergeTreeData::getPartIfExists(const String & part_name, const MergeTreeData::DataPartStates & valid_states) const { - auto lock = lockParts(); + auto lock = readLockParts(); return getPartIfExistsUnlocked(part_name, valid_states, lock); } -MergeTreeData::DataPartPtr MergeTreeData::getPartIfExistsUnlocked(const String & part_name, const DataPartStates & valid_states, DataPartsLock & acquired_lock) const +MergeTreeData::DataPartPtr MergeTreeData::getPartIfExistsUnlocked(const String & part_name, const DataPartStates & valid_states, const DataPartsAnyLock & acquired_lock) const { return getPartIfExistsUnlocked(MergeTreePartInfo::fromPartName(part_name, format_version), valid_states, acquired_lock); } -MergeTreeData::DataPartPtr MergeTreeData::getPartIfExistsUnlocked(const MergeTreePartInfo & part_info, const DataPartStates & valid_states, DataPartsLock & /* acquired_lock */) const +MergeTreeData::DataPartPtr MergeTreeData::getPartIfExistsUnlocked(const MergeTreePartInfo & part_info, const DataPartStates & valid_states, const DataPartsAnyLock & /* acquired_lock */) const { auto it = data_parts_by_info.find(part_info); if (it == data_parts_by_info.end()) @@ -5841,7 +5882,7 @@ void MergeTreeData::loadPartAndFixMetadataImpl(MergeTreeData::MutableDataPartPtr part->removeVersionMetadata(); } -void MergeTreeData::calculateColumnAndSecondaryIndexSizesImpl() const +void MergeTreeData::calculateColumnAndSecondaryIndexSizesImpl(DataPartsLock & /*parts_lock*/) const { std::unique_lock lock(columns_and_secondary_indices_sizes_mutex); @@ -5856,7 +5897,7 @@ void MergeTreeData::calculateColumnAndSecondaryIndexSizesImpl() const are_columns_and_secondary_indices_sizes_calculated = true; } -void MergeTreeData::calculateColumnAndSecondaryIndexSizesLazily(DataPartsLock && parts_lock) const +void MergeTreeData::calculateColumnAndSecondaryIndexSizesLazily(DataPartsSharedLock & /*parts_lock*/, std::unique_lock & /*sizes_lock*/) const { if (are_columns_and_secondary_indices_sizes_calculated) return; @@ -5873,7 +5914,6 @@ void MergeTreeData::calculateColumnAndSecondaryIndexSizesLazily(DataPartsLock && if (hasColumnsWithDynamicSubcolumns(getInMemoryMetadataPtr()->getSampleBlock())) { DataParts data_parts(committed_parts_range.begin(), committed_parts_range.end()); - parts_lock.lock.unlock(); for (const auto & part : data_parts) addPartContributionToColumnAndSecondaryIndexSizesUnlocked(part); } @@ -6005,7 +6045,7 @@ void MergeTreeData::checkAlterPartitionIsPossible( String partition_id = getPartitionIDFromQuery(command.partition, local_context); if (command.type == PartitionCommand::FORGET_PARTITION) { - DataPartsLock lock = lockParts(); + DataPartsSharedLock lock = readLockParts(); auto parts_in_partition = getDataPartsPartitionRange(partition_id); if (!parts_in_partition.empty()) throw Exception(ErrorCodes::CANNOT_FORGET_PARTITION, "Partition {} is not empty", partition_id); @@ -7018,7 +7058,14 @@ MergeTreeData::MutableDataPartPtr MergeTreeData::loadPartRestoredFromBackup(cons UNREACHABLE(); } -String MergeTreeData::getPartitionIDFromQuery(const ASTPtr & ast, ContextPtr local_context, DataPartsLock * acquired_lock) const +String MergeTreeData::getPartitionIDFromQuery(const ASTPtr & ast, ContextPtr local_context) const +{ + auto lock = readLockParts(); + return getPartitionIDFromQuery(ast, local_context, lock); +} + + +String MergeTreeData::getPartitionIDFromQuery(const ASTPtr & ast, ContextPtr local_context, const DataPartsAnyLock & lock) const { const auto & partition_ast = ast->as(); @@ -7194,16 +7241,13 @@ String MergeTreeData::getPartitionIDFromQuery(const ASTPtr & ast, ContextPtr loc MergeTreePartition partition(std::move(partition_row)); String partition_id = partition.getID(*this); + DataPartPtr existing_part_in_partition = getAnyPartInPartition(partition_id, lock); + if (existing_part_in_partition && existing_part_in_partition->partition.value != partition.value) { - auto data_parts_lock = (acquired_lock) ? DataPartsLock() : lockParts(); - DataPartPtr existing_part_in_partition = getAnyPartInPartition(partition_id, data_parts_lock); - if (existing_part_in_partition && existing_part_in_partition->partition.value != partition.value) - { - auto partition_str = partition.serializeToString(existing_part_in_partition->getMetadataSnapshot()); - throw Exception(ErrorCodes::LOGICAL_ERROR, "Parsed partition value: {} " - "doesn't match partition value for an existing part with the same partition ID: {}", - partition_str, existing_part_in_partition->name); - } + auto partition_str = partition.serializeToString(existing_part_in_partition->getMetadataSnapshot()); + throw Exception(ErrorCodes::LOGICAL_ERROR, "Parsed partition value: {} " + "doesn't match partition value for an existing part with the same partition ID: {}", + partition_str, existing_part_in_partition->name); } return partition_id; @@ -7214,6 +7258,47 @@ DataPartsVector MergeTreeData::getVisibleDataPartsVector(ContextPtr local_contex return getVisibleDataPartsVector(local_context->getCurrentTransaction()); } +std::tuple +MergeTreeData::getPossiblySharedVisibleDataPartsRanges(ContextPtr local_context) const +{ + auto shared_lock_holder = std::make_optional(readLockParts()); + std::optional lock_holder; + + if (const auto * txn = local_context->getCurrentTransaction().get()) + { + /// In case of transactions we cannot use shared version + auto parts = getDataPartsVectorForInternalUsage({DataPartState::Active, DataPartState::Outdated}, *shared_lock_holder); + + /// At this point it is safe to release the lock + shared_lock_holder.reset(); + + filterVisibleDataParts(parts, txn->getSnapshot(), txn->tid); + + return std::make_tuple( + std::make_shared(parts), + std::make_shared(std::move(parts)) + ); + } + else + { + if (!shared_parts_list) + { + /// Convert read to write lock, and re-check the shared_parts_list + shared_lock_holder.reset(); + lock_holder.emplace(data_parts_mutex, /*data_=*/ nullptr); + + if (!shared_parts_list) + { + /// Collect parts under the write lock, since the set of parts may have changed while the lock was released + auto parts = getDataPartsVectorForInternalUsage({DataPartState::Active}, *lock_holder); + shared_ranges_in_parts = std::make_shared(parts); + shared_parts_list = std::make_shared(std::move(parts)); + } + } + return std::make_tuple(shared_ranges_in_parts, shared_parts_list); + } +} + DataPartsVector MergeTreeData::getVisibleDataPartsVectorUnlocked(ContextPtr local_context, const DataPartsLock & lock) const { DataPartsVector res; @@ -7298,7 +7383,7 @@ std::set MergeTreeData::getPartitionIdsAffectedByCommands( std::unordered_set MergeTreeData::getAllPartitionIds() const { - auto lock = lockParts(); + auto lock = readLockParts(); std::unordered_set res; std::string_view prev_id; for (const auto & part : getDataPartsStateRange(DataPartState::Active)) @@ -7312,7 +7397,7 @@ std::unordered_set MergeTreeData::getAllPartitionIds() const return res; } -MergeTreeData::DataPartsVector MergeTreeData::getDataPartsVectorForInternalUsage(const DataPartStates & affordable_states, const DataPartsKinds & affordable_kinds, const DataPartsLock & /*lock*/, DataPartStateVector * out_states) const +MergeTreeData::DataPartsVector MergeTreeData::getDataPartsVectorForInternalUsage(const DataPartStates & affordable_states, const DataPartsKinds & affordable_kinds, const DataPartsAnyLock & /*lock*/, DataPartStateVector * out_states) const { DataPartsVector res; DataPartsVector buf; @@ -7340,18 +7425,18 @@ MergeTreeData::DataPartsVector MergeTreeData::getDataPartsVectorForInternalUsage MergeTreeData::DataPartsVector MergeTreeData::getDataPartsVectorForInternalUsage(const DataPartStates & affordable_states, const DataPartsKinds & affordable_kinds, DataPartStateVector * out_states) const { - auto lock = lockParts(); + auto lock = readLockParts(); return getDataPartsVectorForInternalUsage(affordable_states, affordable_kinds, lock, out_states); } -DataPartsVector MergeTreeData::getDataPartsVectorForInternalUsage(const DataPartStates & affordable_states, const DataPartsLock & lock, DataPartStateVector * out_states) const +DataPartsVector MergeTreeData::getDataPartsVectorForInternalUsage(const DataPartStates & affordable_states, const DataPartsAnyLock & lock, DataPartStateVector * out_states) const { return getDataPartsVectorForInternalUsage(affordable_states, {DataPartKind::Regular}, lock, out_states); } MergeTreeData::DataPartsVector MergeTreeData::getDataPartsVectorForInternalUsage(const DataPartStates & affordable_states, DataPartStateVector * out_states) const { - auto lock = lockParts(); + auto lock = readLockParts(); return getDataPartsVectorForInternalUsage(affordable_states, {DataPartKind::Regular}, lock, out_states); } @@ -7362,17 +7447,17 @@ DataPartsVector MergeTreeData::getPatchPartsVectorForInternalUsage(const DataPar DataPartsVector MergeTreeData::getPatchPartsVectorForInternalUsage(const DataPartStates & affordable_states, DataPartStateVector * out_states) const { - auto lock = lockParts(); + auto lock = readLockParts(); return getDataPartsVectorForInternalUsage(affordable_states, {DataPartKind::Patch}, lock, out_states); } DataPartsVector MergeTreeData::getPatchPartsVectorForInternalUsage() const { - auto lock = lockParts(); + auto lock = readLockParts(); return getDataPartsVectorForInternalUsage({DataPartState::Active}, {DataPartKind::Patch}, lock); } -DataPartsVector MergeTreeData::getPatchPartsVectorForPartition(const String & partition_id, const DataPartsLock & /*lock*/) const +DataPartsVector MergeTreeData::getPatchPartsVectorForPartition(const String & partition_id, const DataPartsAnyLock & /*lock*/) const { DataPartsVector res; auto range = getDataPartsStateRange(DataPartState::Active, DataPartKind::Patch); @@ -7388,13 +7473,13 @@ DataPartsVector MergeTreeData::getPatchPartsVectorForPartition(const String & pa DataPartsVector MergeTreeData::getPatchPartsVectorForPartition(const String & partition_id) const { - auto lock = lockParts(); + auto lock = readLockParts(); return getPatchPartsVectorForPartition(partition_id, lock); } MergeTreeData::ProjectionPartsVector MergeTreeData::getProjectionPartsVectorForInternalUsage(const DataPartStates & affordable_states, DataPartStateVector * out_states) const { - auto lock = lockParts(); + auto lock = readLockParts(); ProjectionPartsVector res; for (auto state : affordable_states) { @@ -7420,7 +7505,7 @@ MergeTreeData::ProjectionPartsVector MergeTreeData::getProjectionPartsVectorForI MergeTreeData::DataPartsVector MergeTreeData::getAllDataPartsVector(MergeTreeData::DataPartStateVector * out_states) const { DataPartsVector res; - auto lock = lockParts(); + auto lock = readLockParts(); res.assign(data_parts_by_info.begin(), data_parts_by_info.end()); if (out_states != nullptr) { @@ -7434,14 +7519,14 @@ MergeTreeData::DataPartsVector MergeTreeData::getAllDataPartsVector(MergeTreeDat size_t MergeTreeData::getAllPartsCount() const { - auto lock = lockParts(); + auto lock = readLockParts(); return data_parts_by_info.size(); } size_t MergeTreeData::getTotalMarksCount() const { size_t total_marks = 0; - auto lock = lockParts(); + auto lock = readLockParts(); for (const auto & part : data_parts_by_info) { total_marks += part->getMarksCount(); @@ -7451,7 +7536,7 @@ size_t MergeTreeData::getTotalMarksCount() const bool MergeTreeData::supportsLightweightDelete() const { - auto lock = lockParts(); + auto lock = readLockParts(); for (const auto & part : data_parts_by_info) { if (part->getState() == MergeTreeDataPartState::Outdated @@ -7466,7 +7551,7 @@ bool MergeTreeData::supportsLightweightDelete() const bool MergeTreeData::hasProjection() const { - auto lock = lockParts(); + auto lock = readLockParts(); for (const auto & part : data_parts_by_info) { if (part->getState() == MergeTreeDataPartState::Outdated @@ -7487,7 +7572,7 @@ bool MergeTreeData::areAsynchronousInsertsEnabled() const MergeTreeData::ProjectionPartsVector MergeTreeData::getAllProjectionPartsVector(MergeTreeData::DataPartStateVector * out_states) const { ProjectionPartsVector res; - auto lock = lockParts(); + auto lock = readLockParts(); for (const auto & part : data_parts_by_info) { res.data_parts.push_back(part); @@ -7960,7 +8045,7 @@ MergeTreeData::DataParts MergeTreeData::getDataParts(const DataPartStates & affo { DataParts res; { - auto lock = lockParts(); + auto lock = readLockParts(); for (auto state : affordable_states) { for (auto kind : affordable_kinds) @@ -7984,7 +8069,7 @@ MergeTreeData::DataPartsVector MergeTreeData::getDataPartsVectorForInternalUsage } MergeTreeData::DataPartPtr MergeTreeData::getAnyPartInPartition( - const String & partition_id, DataPartsLock & /*data_parts_lock*/) const + const String & partition_id, const DataPartsAnyLock & /*data_parts_lock*/) const { auto it = data_parts_by_state_and_info.lower_bound(DataPartStateAndPartitionID{DataPartState::Active, partition_id}); @@ -8035,7 +8120,28 @@ void MergeTreeData::Transaction::addPart(MutableDataPartPtr & part, bool need_re precommitted_parts_need_rename.insert(part); } -void MergeTreeData::Transaction::rollback(DataPartsLock * lock) +void MergeTreeData::Transaction::rollback() +{ + if (!isEmpty()) + { + auto lock = data.lockParts(); + rollback(lock); + } +} + +MergeTreeData::Transaction::~Transaction() +{ + try + { + rollback(); + } + catch (...) + { + tryLogCurrentException("~MergeTreeData::Transaction"); + } +} + +void MergeTreeData::Transaction::rollback(DataPartsLock & lock) { if (!isEmpty()) { @@ -8076,9 +8182,6 @@ void MergeTreeData::Transaction::rollback(DataPartsLock * lock) } LOG_DEBUG(data.log, "Undoing transaction {}. {}", getTID(), buf.str()); - /// It would be much better with TSA... - auto our_lock = (lock) ? DataPartsLock() : data.lockParts(); - if (data.data_parts_indexes.empty()) { /// Table was dropped concurrently and all parts (including PreActive parts) were cleared, so there's nothing to rollback @@ -8096,11 +8199,11 @@ void MergeTreeData::Transaction::rollback(DataPartsLock * lock) { data.removePartsFromWorkingSetImmediatelyAndSetTemporaryState( detached_precommitted_parts, - &our_lock); + lock); data.removePartsFromWorkingSet(txn, DataPartsVector(non_detached_precommitted_parts.begin(), non_detached_precommitted_parts.end()), - /* clear_without_timeout = */ true, &our_lock); + /* clear_without_timeout = */ true, lock); } } @@ -8124,7 +8227,13 @@ void MergeTreeData::Transaction::renameParts() precommitted_parts_need_rename.clear(); } -MergeTreeData::DataPartsVector MergeTreeData::Transaction::commit(DataPartsLock * acquired_parts_lock) +MergeTreeData::DataPartsVector MergeTreeData::Transaction::commit() +{ + auto lock = data.lockParts(); + return commit(lock); +} + +MergeTreeData::DataPartsVector MergeTreeData::Transaction::commit(DataPartsLock & acquired_parts_lock) { DataPartsVector total_covered_parts; @@ -8134,8 +8243,6 @@ MergeTreeData::DataPartsVector MergeTreeData::Transaction::commit(DataPartsLock throw Exception(ErrorCodes::LOGICAL_ERROR, "Parts had not been renamed"); auto settings = data.getSettings(); - auto parts_lock = acquired_parts_lock ? DataPartsLock() : data.lockParts(); - auto * owing_parts_lock = acquired_parts_lock ? acquired_parts_lock : &parts_lock; for (const auto & part : precommitted_parts) if (part->getDataPartStorage().hasActiveTransaction()) @@ -8146,12 +8253,12 @@ MergeTreeData::DataPartsVector MergeTreeData::Transaction::commit(DataPartsLock for (const auto & part : precommitted_parts) { DataPartPtr covering_part; - DataPartsVector covered_active_parts = data.getActivePartsToReplace(part->info, part->name, covering_part, *owing_parts_lock); + DataPartsVector covered_active_parts = data.getActivePartsToReplace(part->info, part->name, covering_part, acquired_parts_lock); /// outdated parts should be also collected here /// the visible outdated parts should be tried to be removed /// more likely the conflict happens at the removing visible outdated parts, what is right actually - DataPartsVector covered_outdated_parts = data.getCoveredOutdatedParts(part, *owing_parts_lock); + DataPartsVector covered_outdated_parts = data.getCoveredOutdatedParts(part, acquired_parts_lock); LOG_TEST(data.log, "Got {} oudated parts covered by {} (TID {} CSN {}): {}", covered_outdated_parts.size(), part->getNameWithState(), txn->tid, txn->getSnapshot(), fmt::join(getPartsNames(covered_outdated_parts), ", ")); @@ -8180,7 +8287,7 @@ MergeTreeData::DataPartsVector MergeTreeData::Transaction::commit(DataPartsLock for (const auto & part : precommitted_parts) { DataPartPtr covering_part; - DataPartsVector covered_parts = data.getActivePartsToReplace(part->info, part->name, covering_part, *owing_parts_lock); + DataPartsVector covered_parts = data.getActivePartsToReplace(part->info, part->name, covering_part, acquired_parts_lock); if (covering_part) { /// It's totally fine for zero-level parts, because of possible race condition between ReplicatedMergeTreeSink and @@ -8224,15 +8331,15 @@ MergeTreeData::DataPartsVector MergeTreeData::Transaction::commit(DataPartsLock } } - data.updateSerializationHints(precommitted_parts, total_covered_parts, parts_lock); + data.updateSerializationHints(precommitted_parts, total_covered_parts, acquired_parts_lock); if (reduce_parts == 0) { for (const auto & part : precommitted_parts) - data.updateObjectColumns(part, parts_lock); + data.updateObjectColumns(part, acquired_parts_lock); } else - data.resetObjectColumnsFromActiveParts(parts_lock); + data.resetObjectColumnsFromActiveParts(acquired_parts_lock); ssize_t diff_bytes = add_bytes - reduce_bytes; ssize_t diff_rows = add_rows - reduce_rows; @@ -8571,7 +8678,7 @@ UInt64 MergeTreeData::estimateNumberOfRowsToRead( MergeTreeDataSelectExecutor reader(*this); auto result_ptr = reader.estimateNumMarksToRead( - snapshot_data.parts, + *snapshot_data.parts, snapshot_data.mutations_snapshot, storage_snapshot->metadata->getColumns().getAll().getNames(), storage_snapshot->metadata, @@ -9884,7 +9991,7 @@ ReservationPtr MergeTreeData::balancedReservation( } { - auto lock_parts = lockParts(); + auto lock_parts = readLockParts(); if (covered_parts.empty()) { // It's a part fetch. Calculate `covered_parts` here. @@ -10112,7 +10219,7 @@ void MergeTreeData::updateSerializationHints(const AddedParts & added_parts, con SerializationInfoByName MergeTreeData::getSerializationHints() const { - auto lock = lockParts(); + auto lock = readLockParts(); SerializationInfoByName res; for (const auto & [name, info] : serialization_hints) res.emplace(name, info->clone()); @@ -10185,21 +10292,21 @@ StorageSnapshotPtr MergeTreeData::createStorageSnapshot(const StorageMetadataPtr { if (without_data) { - auto lock = lockParts(); + auto lock = readLockParts(); return std::make_shared(*this, metadata_snapshot, object_columns, std::make_unique()); } auto snapshot_data = std::make_unique(); ColumnsDescription object_columns_copy; - DataPartsVector parts; + auto [query_ranges, query_parts] = getPossiblySharedVisibleDataPartsRanges(query_context); + snapshot_data->parts = query_ranges; + { - auto lock = lockParts(); - parts = getVisibleDataPartsVectorUnlocked(query_context, lock); + /// Copy object columns after taking parts, so they cover at least all columns of the selected parts. + auto lock = readLockParts(); object_columns_copy = object_columns; } - /// Avoid holding the lock while constructing RangesInDataParts - snapshot_data->parts = RangesInDataParts(parts); bool apply_mutations_on_fly = query_context->getSettingsRef()[Setting::apply_mutations_on_fly]; bool apply_patch_parts = query_context->getSettingsRef()[Setting::apply_patch_parts]; @@ -10207,8 +10314,8 @@ StorageSnapshotPtr MergeTreeData::createStorageSnapshot(const StorageMetadataPtr IMutationsSnapshot::Params params { .metadata_version = metadata_snapshot->getMetadataVersion(), - .min_part_metadata_version = getMinMetadataVersion(parts), - .min_part_data_versions = getMinDataVersionForEachPartition(parts), + .min_part_metadata_version = getMinMetadataVersion(*query_parts), + .min_part_data_versions = getMinDataVersionForEachPartition(*query_parts), .max_mutation_versions = query_context->getPartitionIdToMaxBlock(getStorageID().uuid), .need_data_mutations = apply_mutations_on_fly, .need_alter_mutations = apply_mutations_on_fly || apply_patch_parts, @@ -10439,7 +10546,7 @@ size_t MergeTreeData::unloadPrimaryKeysAndClearCachesOfOutdatedParts() DataPartsVector parts_in_use; { - auto parts_lock = lockParts(); + auto parts_lock = readLockParts(); auto parts_range = getDataPartsStateRange(DataPartState::Outdated); for (const auto & part : parts_range) diff --git a/src/Storages/MergeTree/MergeTreeData.h b/src/Storages/MergeTree/MergeTreeData.h index 42b526fcde701..0106ef67ce61c 100644 --- a/src/Storages/MergeTree/MergeTreeData.h +++ b/src/Storages/MergeTree/MergeTreeData.h @@ -97,13 +97,46 @@ namespace ErrorCodes struct DataPartsLock { + explicit DataPartsLock(SharedMutex & data_parts_mutex_, const MergeTreeData * data_); + ~DataPartsLock(); + + DataPartsLock(const DataPartsLock &) = delete; + DataPartsLock & operator=(const DataPartsLock &) = delete; + DataPartsLock(DataPartsLock &&) = default; + DataPartsLock & operator=(DataPartsLock &&) = default; + +private: std::optional wait_watch; - std::unique_lock lock; + std::unique_lock lock; std::optional lock_watch; - DataPartsLock() = default; - explicit DataPartsLock(std::mutex & data_parts_mutex_); + const MergeTreeData * data; +}; - ~DataPartsLock(); +struct DataPartsSharedLock +{ + explicit DataPartsSharedLock(DB::SharedMutex & data_parts_mutex_); + ~DataPartsSharedLock(); + + DataPartsSharedLock(const DataPartsSharedLock &) = delete; + DataPartsSharedLock & operator=(const DataPartsSharedLock &) = delete; + DataPartsSharedLock(DataPartsSharedLock &&) = default; + DataPartsSharedLock & operator=(DataPartsSharedLock &&) = default; + +private: + std::optional wait_watch; + std::shared_lock lock; + std::optional lock_watch; +}; + +// Some functions can accept either type of lock at the caller's preference. +class DataPartsAnyLock final +{ +public: + DataPartsAnyLock(const DataPartsAnyLock &) = delete; + DataPartsAnyLock(DataPartsAnyLock &&) = delete; + + DataPartsAnyLock(const DataPartsLock &) noexcept {} // NOLINT(google-explicit-constructor) + DataPartsAnyLock(const DataPartsSharedLock &) noexcept {} // NOLINT(google-explicit-constructor) }; /// Data structure for *MergeTree engines. @@ -287,8 +320,10 @@ class MergeTreeData : public IStorage, public WithMutableContext using DataParts = std::set; using MutableDataParts = std::set; using DataPartsVector = std::vector; + using DataPartsVectorPtr = std::shared_ptr; - DataPartsLock lockParts() const { return DataPartsLock(data_parts_mutex); } + DataPartsLock lockParts() { return DataPartsLock(data_parts_mutex, this); } + DataPartsSharedLock readLockParts() const { return DataPartsSharedLock(data_parts_mutex); } using OperationDataPartsLock = std::unique_lock; OperationDataPartsLock lockOperationsWithParts() const { return OperationDataPartsLock(operation_with_data_parts_mutex); } @@ -317,7 +352,8 @@ class MergeTreeData : public IStorage, public WithMutableContext public: Transaction(MergeTreeData & data_, MergeTreeTransaction * txn_); - DataPartsVector commit(DataPartsLock * acquired_parts_lock = nullptr); + DataPartsVector commit(); + DataPartsVector commit(DataPartsLock & lock); /// Rename should be done explicitly, before calling commit(), to /// guarantee that no lock held during rename (since rename is IO @@ -326,7 +362,8 @@ class MergeTreeData : public IStorage, public WithMutableContext void addPart(MutableDataPartPtr & part, bool need_rename); - void rollback(DataPartsLock * lock = nullptr); + void rollback(); + void rollback(DataPartsLock & lock); /// Immediately remove parts from table's data_parts set and change part /// state to temporary. Useful for new parts which not present in table. @@ -335,17 +372,7 @@ class MergeTreeData : public IStorage, public WithMutableContext size_t size() const { return precommitted_parts.size(); } bool isEmpty() const { return precommitted_parts.empty(); } - ~Transaction() - { - try - { - rollback(); - } - catch (...) - { - tryLogCurrentException("~MergeTreeData::Transaction"); - } - } + ~Transaction(); void clear(); TransactionID getTID() const; @@ -584,7 +611,10 @@ class MergeTreeData : public IStorage, public WithMutableContext /// and mutations required to be applied at the moment of the start of query. struct SnapshotData : public StorageSnapshot::Data { - RangesInDataParts parts; + // shared_ptr because lifetime is as long as some query still reading it + // const because we are sharing across multiple queries, we cannot have things mutating this. + RangesInDataPartsPtr parts; + MutationsSnapshotPtr mutations_snapshot; }; @@ -630,9 +660,9 @@ class MergeTreeData : public IStorage, public WithMutableContext /// Returns sorted list of the parts with specified states /// out_states will contain snapshot of each part state - DataPartsVector getDataPartsVectorForInternalUsage(const DataPartStates & affordable_states, const DataPartsKinds & affordable_kinds, const DataPartsLock & lock, DataPartStateVector * out_states = nullptr) const; + DataPartsVector getDataPartsVectorForInternalUsage(const DataPartStates & affordable_states, const DataPartsKinds & affordable_kinds, const DataPartsAnyLock & lock, DataPartStateVector * out_states = nullptr) const; DataPartsVector getDataPartsVectorForInternalUsage(const DataPartStates & affordable_states, const DataPartsKinds & affordable_kinds, DataPartStateVector * out_states = nullptr) const; - DataPartsVector getDataPartsVectorForInternalUsage(const DataPartStates & affordable_states, const DataPartsLock & lock, DataPartStateVector * out_states = nullptr) const; + DataPartsVector getDataPartsVectorForInternalUsage(const DataPartStates & affordable_states, const DataPartsAnyLock & lock, DataPartStateVector * out_states = nullptr) const; DataPartsVector getDataPartsVectorForInternalUsage(const DataPartStates & affordable_states, DataPartStateVector * out_states = nullptr) const; /// Returns parts in Active state. @@ -645,14 +675,14 @@ class MergeTreeData : public IStorage, public WithMutableContext /// Returns patch parts in Active state DataPartsVector getPatchPartsVectorForInternalUsage() const; /// Returns patch parts in Active state that relate to partition_id. - DataPartsVector getPatchPartsVectorForPartition(const String & partition_id, const DataPartsLock & lock) const; + DataPartsVector getPatchPartsVectorForPartition(const String & partition_id, const DataPartsAnyLock & lock) const; DataPartsVector getPatchPartsVectorForPartition(const String & partition_id) const; /// Returns absolutely all parts (and snapshot of their states) DataPartsVector getAllDataPartsVector(DataPartStateVector * out_states = nullptr) const; - DataPartsVector getDataPartsVectorInPartitionForInternalUsage(const DataPartState & state, const String & partition_id, DataPartsLock * acquired_lock = nullptr) const; - DataPartsVector getDataPartsVectorInPartitionForInternalUsage(const DataPartStates & affordable_states, const String & partition_id, DataPartsLock * acquired_lock = nullptr) const; + DataPartsVector getDataPartsVectorInPartitionForInternalUsage(const DataPartState & state, const String & partition_id, const DataPartsAnyLock & acquired_lock) const; + DataPartsVector getDataPartsVectorInPartitionForInternalUsage(const DataPartStates & affordable_states, const String & partition_id, const DataPartsAnyLock & acquired_lock) const; /// Returns the number of data mutations suitable for applying on the fly. virtual MutationCounters getMutationCounters() const = 0; @@ -669,39 +699,35 @@ class MergeTreeData : public IStorage, public WithMutableContext /// Returns parts that visible with current snapshot DataPartsVector getVisibleDataPartsVector(ContextPtr local_context) const; + /// If using a shared lock (it guarantees no mutation has happened) and there is no transactions, we can return a shared copy of parts ranges + std::tuple getPossiblySharedVisibleDataPartsRanges(ContextPtr local_context) const; + /// Whereas if a unique lock is used, mutations could have happened, meaning shared part list *may* have been invalidated. DataPartsVector getVisibleDataPartsVectorUnlocked(ContextPtr local_context, const DataPartsLock & lock) const; DataPartsVector getVisibleDataPartsVector(const MergeTreeTransactionPtr & txn) const; DataPartsVector getVisibleDataPartsVector(CSN snapshot_version, TransactionID current_tid) const; /// Returns all parts in specified partition - DataPartsVector getVisibleDataPartsVectorInPartition(MergeTreeTransaction * txn, const String & partition_id, DataPartsLock * acquired_lock = nullptr) const; - DataPartsVector getVisibleDataPartsVectorInPartition(ContextPtr local_context, const String & partition_id, DataPartsLock & lock) const; + DataPartsVector getVisibleDataPartsVectorInPartition(MergeTreeTransaction * txn, const String & partition_id, const DataPartsAnyLock & acquired_lock) const; + DataPartsVector getVisibleDataPartsVectorInPartition(ContextPtr local_context, const String & partition_id, const DataPartsAnyLock & lock) const; DataPartsVector getVisibleDataPartsVectorInPartition(ContextPtr local_context, const String & partition_id) const; DataPartsVector getVisibleDataPartsVectorInPartitions(ContextPtr local_context, const std::unordered_set & partition_ids) const; /// Return the number of marks in all parts size_t getTotalMarksCount() const; - /// Adjust the set of data parts that should be used for SELECT queries. - /// Skips very new parts if they're not in cache yet (ignore_cold_parts_seconds), and replaces - /// recently merged parts with the pre-merge source parts if the merged part is not in cache yet - /// (prefer_warmed_unmerged_parts_seconds). This improves cache hit rate and latency when cache - /// warmer is enabled. - void adjustDataPartsVectorBasedOnCacheWarmness(DataPartsVector & parts, const ContextPtr & local_context, const DataPartsLock & lock) const; - /// Returns a part in Active state with the given name or a part containing it. If there is no such part, returns nullptr. DataPartPtr getActiveContainingPart(const String & part_name) const; - DataPartPtr getActiveContainingPart(const String & part_name, DataPartsLock & lock) const; + DataPartPtr getActiveContainingPart(const String & part_name, const DataPartsAnyLock & lock) const; DataPartPtr getActiveContainingPart(const MergeTreePartInfo & part_info) const; - DataPartPtr getActiveContainingPart(const MergeTreePartInfo & part_info, DataPartState state, DataPartsLock & lock) const; + DataPartPtr getActiveContainingPart(const MergeTreePartInfo & part_info, DataPartState state, const DataPartsAnyLock & lock) const; /// Swap part with it's identical copy (possible with another path on another disk). /// If original part is not active or doesn't exist exception will be thrown. void swapActivePart(MergeTreeData::DataPartPtr part_copy, DataPartsLock &); /// Returns the part with the given name and state or nullptr if no such part. - DataPartPtr getPartIfExistsUnlocked(const String & part_name, const DataPartStates & valid_states, DataPartsLock & acquired_lock) const; - DataPartPtr getPartIfExistsUnlocked(const MergeTreePartInfo & part_info, const DataPartStates & valid_states, DataPartsLock & acquired_lock) const; + DataPartPtr getPartIfExistsUnlocked(const String & part_name, const DataPartStates & valid_states, const DataPartsAnyLock & acquired_lock) const; + DataPartPtr getPartIfExistsUnlocked(const MergeTreePartInfo & part_info, const DataPartStates & valid_states, const DataPartsAnyLock & acquired_lock) const; DataPartPtr getPartIfExists(const String & part_name, const DataPartStates & valid_states) const; DataPartPtr getPartIfExists(const MergeTreePartInfo & part_info, const DataPartStates & valid_states) const; @@ -768,6 +794,11 @@ class MergeTreeData : public IStorage, public WithMutableContext MutableDataPartPtr & part, Transaction & out_transaction, bool rename_in_transaction); + DataPartsVector renameTempPartAndReplaceUnlocked( + MutableDataPartPtr & part, + DataPartsLock & lock, + Transaction & out_transaction, + bool rename_in_transaction); /// Unlocked version of previous one. Useful when added multiple parts with a single lock. bool renameTempPartAndReplaceUnlocked( @@ -779,13 +810,14 @@ class MergeTreeData : public IStorage, public WithMutableContext /// Remove parts from working set immediately (without wait for background /// process). Transfer part state to temporary. Have very limited usage only /// for new parts which aren't already present in table. - void removePartsFromWorkingSetImmediatelyAndSetTemporaryState(const DataPartsVector & remove, DataPartsLock * acquired_lock = nullptr); + void removePartsFromWorkingSetImmediatelyAndSetTemporaryState(const DataPartsVector & remove); + void removePartsFromWorkingSetImmediatelyAndSetTemporaryState(const DataPartsVector & remove, DataPartsLock & lock); /// Removes parts from the working set parts. /// Parts in add must already be in data_parts with PreActive, Active, or Outdated states. /// If clear_without_timeout is true, the parts will be deleted at once, or during the next call to /// clearOldParts (ignoring old_parts_lifetime). - void removePartsFromWorkingSet(MergeTreeTransaction * txn, const DataPartsVector & remove, bool clear_without_timeout, DataPartsLock * acquired_lock = nullptr); + void removePartsFromWorkingSet(MergeTreeTransaction * txn, const DataPartsVector & remove, bool clear_without_timeout); void removePartsFromWorkingSet(MergeTreeTransaction * txn, const DataPartsVector & remove, bool clear_without_timeout, DataPartsLock & acquired_lock); /// Removes all parts covered by drop_range from the working set parts. @@ -793,7 +825,7 @@ class MergeTreeData : public IStorage, public WithMutableContext void removePartsInRangeFromWorkingSet(MergeTreeTransaction * txn, const MergeTreePartInfo & drop_range, DataPartsLock & lock); DataPartsVector grabActivePartsToRemoveForDropRange( - MergeTreeTransaction * txn, const MergeTreePartInfo & drop_range, DataPartsLock & lock); + MergeTreeTransaction * txn, const MergeTreePartInfo & drop_range, const DataPartsAnyLock & lock); /// This wrapper is required to restrict access to parts in Deleting state class PartToRemoveFromZooKeeper @@ -825,7 +857,7 @@ class MergeTreeData : public IStorage, public WithMutableContext MergeTreeTransaction * txn, const MergeTreePartInfo & drop_range, DataPartsLock & lock, bool create_empty_part = true); /// Restores Outdated part and adds it to working set - void restoreAndActivatePart(const DataPartPtr & part, DataPartsLock * acquired_lock = nullptr); + void restoreAndActivatePart(const DataPartPtr & part); /// Renames the part to detached/_ and removes it from data_parts, //// so it will not be deleted in clearOldParts. @@ -1023,9 +1055,9 @@ class MergeTreeData : public IStorage, public WithMutableContext size_t getColumnCompressedSize(const std::string & name) const { /// Always keep locks order parts_lock -> sizes_lock - auto parts_lock = lockParts(); + auto parts_lock = readLockParts(); std::unique_lock sizes_lock(columns_and_secondary_indices_sizes_mutex); - calculateColumnAndSecondaryIndexSizesLazily(std::move(parts_lock)); + calculateColumnAndSecondaryIndexSizesLazily(parts_lock, sizes_lock); const auto it = column_sizes.find(name); return it == std::end(column_sizes) ? 0 : it->second.data_compressed; } @@ -1033,9 +1065,9 @@ class MergeTreeData : public IStorage, public WithMutableContext ColumnSizeByName getColumnSizes() const override { /// Always keep locks order parts_lock -> sizes_lock - auto parts_lock = lockParts(); + auto parts_lock = readLockParts(); std::unique_lock sizes_lock(columns_and_secondary_indices_sizes_mutex); - calculateColumnAndSecondaryIndexSizesLazily(std::move(parts_lock)); + calculateColumnAndSecondaryIndexSizesLazily(parts_lock, sizes_lock); return column_sizes; } @@ -1046,14 +1078,15 @@ class MergeTreeData : public IStorage, public WithMutableContext IndexSizeByName getSecondaryIndexSizes() const override { /// Always keep locks order parts_lock -> sizes_lock - auto parts_lock = lockParts(); + auto parts_lock = readLockParts(); std::unique_lock sizes_lock(columns_and_secondary_indices_sizes_mutex); - calculateColumnAndSecondaryIndexSizesLazily(std::move(parts_lock)); + calculateColumnAndSecondaryIndexSizesLazily(parts_lock, sizes_lock); return secondary_index_sizes; } /// For ATTACH/DETACH/DROP/FORGET PARTITION. - String getPartitionIDFromQuery(const ASTPtr & ast, ContextPtr context, DataPartsLock * acquired_lock = nullptr) const; + String getPartitionIDFromQuery(const ASTPtr & ast, ContextPtr context) const; + String getPartitionIDFromQuery(const ASTPtr & ast, ContextPtr context, const DataPartsAnyLock & lock) const; std::unordered_set getPartitionIDsFromQuery(const ASTs & asts, ContextPtr context) const; std::set getPartitionIdsAffectedByCommands(const MutationCommands & commands, ContextPtr query_context) const; @@ -1369,6 +1402,7 @@ class MergeTreeData : public IStorage, public WithMutableContext friend class IPartMetadataManager; friend class IMergedBlockOutputStream; // for access to log friend class ExportPartTask; + friend struct DataPartsLock; // for access to shared_parts_list/shared_ranges_in_parts bool require_part_metadata; @@ -1448,12 +1482,20 @@ class MergeTreeData : public IStorage, public WithMutableContext >; /// Current set of data parts. - mutable std::mutex data_parts_mutex; + /// On updates shared_parts_list/shared_ranges_in_parts should be reset (will be updated in getPossiblySharedVisibleDataPartsRanges()) + mutable DB::SharedMutex data_parts_mutex; DataPartsIndexes data_parts_indexes; DataPartsIndexes::index::type & data_parts_by_info; DataPartsIndexes::index::type & data_parts_by_state_and_info; + /// A readonly shared copy of the `data_parts_by_state_and_info`, so we don't need to copy it for every query + /// Pointer may only be updated if you have a DataPartsLock + /// Pointer may only be shared if you have a DataPartsSharedLock + mutable DataPartsVectorPtr shared_parts_list; + /// Same as above, but this time caching a copy of RangesInDataParts + mutable RangesInDataPartsPtr shared_ranges_in_parts; + /// Mutex for critical sections which alter set of parts /// It is like truncate, drop/detach partition mutable std::mutex operation_with_data_parts_mutex; @@ -1581,12 +1623,12 @@ class MergeTreeData : public IStorage, public WithMutableContext void checkStoragePolicy(const StoragePolicyPtr & new_storage_policy) const; - /// Calculates column and secondary indexes sizes in compressed form for the current state of data_parts. Call with data_parts mutex locked. - void calculateColumnAndSecondaryIndexSizesImpl() const; + /// Calculates column and secondary indexes sizes in compressed form for the current state of data_parts. Call with data_parts mutex under lock. + void calculateColumnAndSecondaryIndexSizesImpl(DataPartsLock & parts_lock) const; /// Similar to above but should be called before accessing column and secondary indexes sizes for possible lazy calculation. - /// Call it with data_parts and columns_and_secondary_indices_sizes_mutex mutexes locked. - void calculateColumnAndSecondaryIndexSizesLazily(DataPartsLock && parts_lock) const; + /// Call it with data_parts mutex under a shared lock, and columns_and_secondary_indices_sizes_mutex mutex under an unique lock. + void calculateColumnAndSecondaryIndexSizesLazily(DataPartsSharedLock & parts_lock, std::unique_lock & sizes_lock) const; /// Adds or subtracts the contribution of the part to compressed column and secondary indexes sizes. void addPartContributionToColumnAndSecondaryIndexSizes(const DataPartPtr & part) const; @@ -1594,7 +1636,7 @@ class MergeTreeData : public IStorage, public WithMutableContext void removePartContributionToColumnAndSecondaryIndexSizes(const DataPartPtr & part) const; /// If there is no part in the partition with ID `partition_id`, returns empty ptr. Should be called under the lock. - DataPartPtr getAnyPartInPartition(const String & partition_id, DataPartsLock & data_parts_lock) const; + DataPartPtr getAnyPartInPartition(const String & partition_id, const DataPartsAnyLock & data_parts_lock) const; /// Return parts in the Active set that are covered by the new_part_info or the part that covers it. /// Will check that the new part doesn't already exist and that it doesn't intersect existing part. @@ -1602,7 +1644,7 @@ class MergeTreeData : public IStorage, public WithMutableContext const MergeTreePartInfo & new_part_info, const String & new_part_name, DataPartPtr & out_covering_part, - DataPartsLock & data_parts_lock) const; + const DataPartsAnyLock & data_parts_lock) const; DataPartsVector getCoveredOutdatedParts( const DataPartPtr & part, @@ -1619,7 +1661,7 @@ class MergeTreeData : public IStorage, public WithMutableContext PartHierarchy getPartHierarchy( const MergeTreePartInfo & part_info, DataPartState state, - DataPartsLock & /* data_parts_lock */) const; + const DataPartsAnyLock & /* data_parts_lock */) const; /// Checks whether the column is in the primary key, possibly wrapped in a chain of functions with single argument. bool isPrimaryOrMinMaxKeyColumnPossiblyWrappedInFunctions(const ASTPtr & node, const StorageMetadataPtr & metadata_snapshot) const; @@ -1823,8 +1865,8 @@ class MergeTreeData : public IStorage, public WithMutableContext private: /// Checking that candidate part doesn't break invariants: correct partition - void checkPartPartition(MutableDataPartPtr & part, DataPartsLock & lock) const; - void checkPartDuplicate(MutableDataPartPtr & part, Transaction & transaction, DataPartsLock & lock) const; + void checkPartPartition(MutableDataPartPtr & part, const DataPartsAnyLock & lock) const; + void checkPartDuplicate(MutableDataPartPtr & part, Transaction & transaction, const DataPartsAnyLock & lock) const; void checkPartDynamicColumns(MutableDataPartPtr & part, DataPartsLock & lock) const; /// Preparing itself to be committed in memory: fill some fields inside part, add it to data_parts_indexes @@ -1905,14 +1947,14 @@ class MergeTreeData : public IStorage, public WithMutableContext const String & part_name, const DiskPtr & part_disk_ptr, MergeTreeDataPartState to_state, - std::mutex & part_loading_mutex); + DB::SharedMutex & part_loading_mutex); LoadPartResult loadDataPartWithRetries( const MergeTreePartInfo & part_info, const String & part_name, const DiskPtr & part_disk_ptr, MergeTreeDataPartState to_state, - std::mutex & part_loading_mutex, + DB::SharedMutex & part_loading_mutex, size_t backoff_ms, size_t max_backoff_ms, size_t max_tries); diff --git a/src/Storages/MergeTree/MergeTreeDataSelectExecutor.cpp b/src/Storages/MergeTree/MergeTreeDataSelectExecutor.cpp index c6e8b7bac16e7..723d6d2029a03 100644 --- a/src/Storages/MergeTree/MergeTreeDataSelectExecutor.cpp +++ b/src/Storages/MergeTree/MergeTreeDataSelectExecutor.cpp @@ -575,8 +575,8 @@ std::optional> MergeTreeDataSelectExecutor::filterPar return VirtualColumnUtils::extractSingleValueFromBlock(virtual_columns_block, "_part"); } -void MergeTreeDataSelectExecutor::filterPartsByPartition( - RangesInDataParts & parts, +RangesInDataParts MergeTreeDataSelectExecutor::filterPartsByPartition( + const RangesInDataParts & parts, const std::optional & partition_pruner, const std::optional & minmax_idx_condition, const std::optional> & part_values, @@ -587,6 +587,7 @@ void MergeTreeDataSelectExecutor::filterPartsByPartition( LoggerPtr log, ReadFromMergeTree::IndexStats & index_stats) { + RangesInDataParts res; const Settings & settings = context->getSettingsRef(); DataTypes minmax_columns_types; @@ -610,7 +611,7 @@ void MergeTreeDataSelectExecutor::filterPartsByPartition( PartFilterCounters part_filter_counters; if (query_context->getSettingsRef()[Setting::allow_experimental_query_deduplication]) - selectPartsToReadWithUUIDFilter( + res = selectPartsToReadWithUUIDFilter( parts, part_values, data.getPinnedPartUUIDs(), @@ -622,7 +623,7 @@ void MergeTreeDataSelectExecutor::filterPartsByPartition( part_filter_counters, log); else - selectPartsToRead( + res = selectPartsToRead( parts, part_values, minmax_idx_condition, @@ -659,6 +660,8 @@ void MergeTreeDataSelectExecutor::filterPartsByPartition( .num_parts_after = part_filter_counters.num_parts_after_partition_pruner, .num_granules_after = part_filter_counters.num_granules_after_partition_pruner}); } + + return res; } RangesInDataParts MergeTreeDataSelectExecutor::filterPartsByPrimaryKeyAndSkipIndexes( @@ -1248,7 +1251,7 @@ ReadFromMergeTree::AnalysisResultPtr MergeTreeDataSelectExecutor::estimateNumMar std::optional indexes; return ReadFromMergeTree::selectRangesToRead( - std::move(parts), + parts, mutations_snapshot, std::nullopt, metadata_snapshot, @@ -1264,7 +1267,7 @@ ReadFromMergeTree::AnalysisResultPtr MergeTreeDataSelectExecutor::estimateNumMar } QueryPlanStepPtr MergeTreeDataSelectExecutor::readFromParts( - RangesInDataParts parts, + RangesInDataPartsPtr parts, MergeTreeData::MutationsSnapshotPtr mutations_snapshot, const Names & column_names_to_return, const StorageSnapshotPtr & storage_snapshot, @@ -1283,7 +1286,10 @@ QueryPlanStepPtr MergeTreeDataSelectExecutor::readFromParts( if (merge_tree_select_result_ptr->selected_marks == 0) return {}; } - else if (parts.empty()) + /// If merge_tree_enable_remove_parts_from_snapshot_optimization is true it nukes our list of parts + else if (!parts) + return {}; + else if (parts->empty()) return {}; return std::make_unique( @@ -2008,8 +2014,8 @@ MarkRanges MergeTreeDataSelectExecutor::filterMarksUsingMergedIndex( return res; } -void MergeTreeDataSelectExecutor::selectPartsToRead( - RangesInDataParts & parts, +RangesInDataParts MergeTreeDataSelectExecutor::selectPartsToRead( + const RangesInDataParts & parts, const std::optional> & part_values, const std::optional & minmax_idx_condition, const DataTypes & minmax_columns_types, @@ -2018,10 +2024,9 @@ void MergeTreeDataSelectExecutor::selectPartsToRead( PartFilterCounters & counters, QueryStatusPtr query_status) { - RangesInDataParts prev_parts; - std::swap(prev_parts, parts); + RangesInDataParts res_parts; - for (const auto & prev_part : prev_parts) + for (const auto & prev_part : parts) { const auto & part_or_projection = prev_part.data_part; @@ -2068,12 +2073,13 @@ void MergeTreeDataSelectExecutor::selectPartsToRead( counters.num_parts_after_partition_pruner += 1; counters.num_granules_after_partition_pruner += num_granules; - parts.push_back(prev_part); + res_parts.push_back(prev_part); } + return res_parts; } -void MergeTreeDataSelectExecutor::selectPartsToReadWithUUIDFilter( - RangesInDataParts & parts, +RangesInDataParts MergeTreeDataSelectExecutor::selectPartsToReadWithUUIDFilter( + const RangesInDataParts & parts, const std::optional> & part_values, MergeTreeData::PinnedPartUUIDsPtr pinned_part_uuids, const std::optional & minmax_idx_condition, @@ -2086,15 +2092,12 @@ void MergeTreeDataSelectExecutor::selectPartsToReadWithUUIDFilter( { /// process_parts prepare parts that have to be read for the query, /// returns false if duplicated parts' UUID have been met - auto select_parts = [&](RangesInDataParts & selected_parts) -> bool + auto select_parts = [&](const RangesInDataParts & in_parts, RangesInDataParts & selected_parts) -> bool { auto ignored_part_uuids = query_context->getIgnoredPartUUIDs(); std::unordered_set temp_part_uuids; - RangesInDataParts prev_parts; - std::swap(prev_parts, selected_parts); - - for (const auto & prev_part : prev_parts) + for (const auto & prev_part : in_parts) { const auto & part_or_projection = prev_part.data_part; const auto * part = part_or_projection->isProjectionPart() ? part_or_projection->getParentPart() : part_or_projection.get(); @@ -2164,7 +2167,8 @@ void MergeTreeDataSelectExecutor::selectPartsToReadWithUUIDFilter( }; /// Process parts that have to be read for a query. - auto needs_retry = !select_parts(parts); + RangesInDataParts filtered_parts = {}; + auto needs_retry = !select_parts(parts, filtered_parts); /// If any duplicated part UUIDs met during the first step, try to ignore them in second pass. /// This may happen when `prefer_localhost_replica` is set and "distributed" stage runs in the same process with "remote" stage. @@ -2174,10 +2178,15 @@ void MergeTreeDataSelectExecutor::selectPartsToReadWithUUIDFilter( counters = PartFilterCounters(); + auto initial_filtered_parts = filtered_parts; + filtered_parts = {}; + /// Second attempt didn't help, throw an exception - if (!select_parts(parts)) + if (!select_parts(initial_filtered_parts, filtered_parts)) throw Exception(ErrorCodes::DUPLICATED_PART_UUIDS, "Found duplicate UUIDs while processing query."); } + + return filtered_parts; } } diff --git a/src/Storages/MergeTree/MergeTreeDataSelectExecutor.h b/src/Storages/MergeTree/MergeTreeDataSelectExecutor.h index 7d2b5ab231e36..6808fcfb785c9 100644 --- a/src/Storages/MergeTree/MergeTreeDataSelectExecutor.h +++ b/src/Storages/MergeTree/MergeTreeDataSelectExecutor.h @@ -38,7 +38,7 @@ class MergeTreeDataSelectExecutor /// The same as read, but with specified set of parts. QueryPlanStepPtr readFromParts( - RangesInDataParts parts, + RangesInDataPartsPtr parts, MergeTreeData::MutationsSnapshotPtr mutations_snapshot, const Names & column_names, const StorageSnapshotPtr & storage_snapshot, @@ -123,8 +123,8 @@ class MergeTreeDataSelectExecutor /// Select the parts in which there can be data that satisfy `minmax_idx_condition` and that match the condition on `_part`, /// as well as `max_block_number_to_read`. - static void selectPartsToRead( - RangesInDataParts & parts, + static RangesInDataParts selectPartsToRead( + const RangesInDataParts & parts, const std::optional> & part_values, const std::optional & minmax_idx_condition, const DataTypes & minmax_columns_types, @@ -134,8 +134,8 @@ class MergeTreeDataSelectExecutor QueryStatusPtr query_status); /// Same as previous but also skip parts uuids if any to the query context, or skip parts which uuids marked as excluded. - static void selectPartsToReadWithUUIDFilter( - RangesInDataParts & parts, + static RangesInDataParts selectPartsToReadWithUUIDFilter( + const RangesInDataParts & parts, const std::optional> & part_values, MergeTreeData::PinnedPartUUIDsPtr pinned_part_uuids, const std::optional & minmax_idx_condition, @@ -178,8 +178,8 @@ class MergeTreeDataSelectExecutor ContextPtr context); /// Filter parts using minmax index and partition key. - static void filterPartsByPartition( - RangesInDataParts & parts, + static RangesInDataParts filterPartsByPartition( + const RangesInDataParts & parts, const std::optional & partition_pruner, const std::optional & minmax_idx_condition, const std::optional> & part_values, diff --git a/src/Storages/MergeTree/MergeTreeSink.cpp b/src/Storages/MergeTree/MergeTreeSink.cpp index 433956214983d..e1e3481224eb1 100644 --- a/src/Storages/MergeTree/MergeTreeSink.cpp +++ b/src/Storages/MergeTree/MergeTreeSink.cpp @@ -256,7 +256,7 @@ bool MergeTreeSink::commitPart(MergeTreeMutableDataPartPtr & part, const String /// /// Hence, for now rename_in_transaction is false. added = storage.renameTempPartAndAdd(part, transaction, lock, /*rename_in_transaction=*/ false); - transaction.commit(&lock); + transaction.commit(lock); } return added; diff --git a/src/Storages/MergeTree/RangesInDataPart.h b/src/Storages/MergeTree/RangesInDataPart.h index e44f9ffc4fc65..34212a347d793 100644 --- a/src/Storages/MergeTree/RangesInDataPart.h +++ b/src/Storages/MergeTree/RangesInDataPart.h @@ -94,5 +94,6 @@ struct RangesInDataParts : public std::vector size_t getMarksCountAllParts() const; size_t getRowsCountAllParts() const; }; +using RangesInDataPartsPtr = std::shared_ptr; } diff --git a/src/Storages/MergeTree/StorageFromMergeTreeDataPart.cpp b/src/Storages/MergeTree/StorageFromMergeTreeDataPart.cpp index cef5e64c39b15..8e0854b39c81e 100644 --- a/src/Storages/MergeTree/StorageFromMergeTreeDataPart.cpp +++ b/src/Storages/MergeTree/StorageFromMergeTreeDataPart.cpp @@ -34,7 +34,7 @@ void StorageFromMergeTreeDataPart::read( size_t num_streams) { query_plan.addStep(MergeTreeDataSelectExecutor(storage).readFromParts( - parts, + std::make_shared(parts), mutations_snapshot, column_names, storage_snapshot, diff --git a/src/Storages/MergeTree/StorageFromMergeTreeProjection.cpp b/src/Storages/MergeTree/StorageFromMergeTreeProjection.cpp index 74ddfa19ef9a6..7ddcc36268421 100644 --- a/src/Storages/MergeTree/StorageFromMergeTreeProjection.cpp +++ b/src/Storages/MergeTree/StorageFromMergeTreeProjection.cpp @@ -38,7 +38,7 @@ void StorageFromMergeTreeProjection::read( const auto & parts = snapshot_data.parts; RangesInDataParts projection_parts; - for (const auto & part : parts) + for (const auto & part : *parts) { const auto & created_projections = part.data_part->getProjectionParts(); auto it = created_projections.find(projection->name); @@ -51,7 +51,7 @@ void StorageFromMergeTreeProjection::read( auto step = MergeTreeDataSelectExecutor(merge_tree) .readFromParts( - std::move(projection_parts), + std::make_shared(projection_parts), snapshot_data.mutations_snapshot->cloneEmpty(), column_names, storage_snapshot, diff --git a/src/Storages/StorageMergeTree.cpp b/src/Storages/StorageMergeTree.cpp index 816a4b8143b73..5f81226f4c0f4 100644 --- a/src/Storages/StorageMergeTree.cpp +++ b/src/Storages/StorageMergeTree.cpp @@ -1753,7 +1753,7 @@ MergeMutateSelectedEntryPtr StorageMergeTree::selectPartsToMutate( UInt32 StorageMergeTree::getMaxLevelInBetween(const PartProperties & left, const PartProperties & right) const { - auto parts_lock = lockParts(); + auto parts_lock = readLockParts(); auto begin = data_parts_by_info.find(left.info); if (begin == data_parts_by_info.end()) @@ -2128,7 +2128,11 @@ ActionLock StorageMergeTree::stopMergesAndWaitForPartition(String partition_id) /// This protects against "revival" of data for a removed partition after completion of merge. auto merge_blocker = merger_mutator.merges_blocker.cancelForPartition(partition_id); - const DataPartsVector parts_to_wait = getDataPartsVectorInPartitionForInternalUsage(MergeTreeDataPartState::Active, partition_id); + DataPartsVector parts_to_wait; + { + auto parts_lock = readLockParts(); + parts_to_wait = getDataPartsVectorInPartitionForInternalUsage(MergeTreeDataPartState::Active, partition_id, parts_lock); + } LOG_TRACE(log, "StorageMergeTree::stopMergesAndWaitForPartition parts to wait: {} ({} items)", fmt::join(getNameWithState(parts_to_wait), ", "), parts_to_wait.size()); @@ -2189,7 +2193,7 @@ MergeTreeDataPartPtr StorageMergeTree::outdatePart(MergeTreeTransaction * txn, c if (!part) throw Exception(ErrorCodes::NO_SUCH_DATA_PART, "Part {} not found, won't try to drop it.", part_name); - removePartsFromWorkingSet(txn, {part}, clear_without_timeout, &parts_lock); + removePartsFromWorkingSet(txn, {part}, clear_without_timeout, parts_lock); return part; } @@ -2207,7 +2211,7 @@ MergeTreeDataPartPtr StorageMergeTree::outdatePart(MergeTreeTransaction * txn, c if (currently_merging_mutating_parts.contains(part)) return nullptr; - removePartsFromWorkingSet(txn, {part}, clear_without_timeout, &parts_lock); + removePartsFromWorkingSet(txn, {part}, clear_without_timeout, parts_lock); return part; } @@ -2284,17 +2288,20 @@ void StorageMergeTree::renameAndCommitEmptyParts(MutableDataPartsVector & new_pa { DataPartsVector covered_parts; - for (auto & part: new_parts) { - DataPartsVector covered_parts_by_one_part = renameTempPartAndReplace(part, transaction, /*rename_in_transaction=*/ true); + auto part_lock = lockParts(); + for (auto & part : new_parts) + { + DataPartsVector covered_parts_by_one_part = renameTempPartAndReplaceUnlocked(part, part_lock, transaction, /*rename_in_transaction=*/ true); - if (covered_parts_by_one_part.size() > 1) - throw Exception(ErrorCodes::LOGICAL_ERROR, - "Part {} expected to cover not more then 1 part. " - "{} covered parts have been found. This is a bug.", - part->name, covered_parts_by_one_part.size()); + if (covered_parts_by_one_part.size() > 1) + throw Exception(ErrorCodes::LOGICAL_ERROR, + "Part {} expected to cover not more then 1 part. " + "{} covered parts have been found. This is a bug.", + part->name, covered_parts_by_one_part.size()); - std::move(covered_parts_by_one_part.begin(), covered_parts_by_one_part.end(), std::back_inserter(covered_parts)); + std::move(covered_parts_by_one_part.begin(), covered_parts_by_one_part.end(), std::back_inserter(covered_parts)); + } } LOG_INFO(log, "Remove {} parts by covering them with empty {} parts. With txn {}.", covered_parts.size(), new_parts.size(), transaction.getTID()); @@ -2450,7 +2457,7 @@ void StorageMergeTree::dropPartition(const ASTPtr & partition, bool detach, Cont parts_to_remove = getVisibleDataPartsVectorUnlocked(query_context, data_parts_lock); else { - String partition_id = getPartitionIDFromQuery(partition, query_context, &data_parts_lock); + String partition_id = getPartitionIDFromQuery(partition, query_context, data_parts_lock); parts_to_remove = getVisibleDataPartsVectorInPartition(query_context, partition_id, data_parts_lock); } removePartsFromWorkingSet(txn.get(), parts_to_remove, true, data_parts_lock); @@ -2565,7 +2572,7 @@ PartitionCommandsResultInfo StorageMergeTree::attachPartition( auto lock = lockParts(); auto block_holder = fillNewPartNameAndResetLevel(loaded_parts[i], lock); renameTempPartAndAdd(loaded_parts[i], transaction, lock, /*rename_in_transaction=*/ false); - transaction.commit(&lock); + transaction.commit(lock); } renamed_parts.old_and_new_names[i].old_name.clear(); @@ -2710,7 +2717,7 @@ void StorageMergeTree::replacePartitionFrom(const StoragePtr & source_table, con renameTempPartAndReplaceUnlocked(part, transaction, data_parts_lock, /*rename_in_transaction=*/ false); } /// Populate transaction - transaction.commit(&data_parts_lock); + transaction.commit(data_parts_lock); /// If it is REPLACE (not ATTACH), remove all parts which max_block_number less then min_block_number of the first new block if (replace) @@ -2834,7 +2841,7 @@ void StorageMergeTree::movePartitionToTable(const StoragePtr & dest_table, const } removePartsFromWorkingSet(local_context->getCurrentTransaction().get(), src_parts, true, src_data_parts_lock); - transaction.commit(&src_data_parts_lock); + transaction.commit(src_data_parts_lock); } clearOldPartsFromFilesystem(); @@ -2996,7 +3003,7 @@ void StorageMergeTree::attachRestoredParts(MutableDataPartsVector && parts, cons auto lock = lockParts(); auto block_holder = fillNewPartName(part, lock); renameTempPartAndAdd(part, transaction, lock, /*rename_in_transaction=*/ false); - transaction.commit(&lock); + transaction.commit(lock); } } } diff --git a/src/Storages/StorageReplicatedMergeTree.cpp b/src/Storages/StorageReplicatedMergeTree.cpp index b0bb39ed976dc..eede10d3d1c49 100644 --- a/src/Storages/StorageReplicatedMergeTree.cpp +++ b/src/Storages/StorageReplicatedMergeTree.cpp @@ -3253,11 +3253,14 @@ bool StorageReplicatedMergeTree::executeReplaceRange(LogEntry & entry) Transaction transaction(*this, NO_TRANSACTION_RAW); Coordination::Requests ops; - for (PartDescriptionPtr & part_desc : final_parts) { - renameTempPartAndReplace(part_desc->res_part, transaction, /*rename_in_transaction=*/ true); - getCommitPartOps(ops, part_desc->res_part); - lockSharedData(*part_desc->res_part, /*replace_existing_lock=*/ true, part_desc->hardlinked_files); + auto lock = lockParts(); + for (PartDescriptionPtr & part_desc : final_parts) + { + renameTempPartAndReplaceUnlocked(part_desc->res_part, lock, transaction, /*rename_in_transaction=*/ true); + getCommitPartOps(ops, part_desc->res_part); + lockSharedData(*part_desc->res_part, /*replace_existing_lock=*/ true, part_desc->hardlinked_files); + } } transaction.renameParts(); @@ -3268,7 +3271,7 @@ bool StorageReplicatedMergeTree::executeReplaceRange(LogEntry & entry) { auto data_parts_lock = lockParts(); - transaction.commit(&data_parts_lock); + transaction.commit(data_parts_lock); if (replace) { parts_to_remove = removePartsInRangeFromWorkingSetAndGetPartsToRemoveFromZooKeeper(NO_TRANSACTION_RAW, drop_range, data_parts_lock); @@ -4751,9 +4754,9 @@ void StorageReplicatedMergeTree::removePartAndEnqueueFetch(const String & part_n { if (!broken_part) return; - DataPartsLock lock = lockParts(); + auto lock = lockParts(); if (broken_part->getState() == DataPartState::Active) - removePartsFromWorkingSet(NO_TRANSACTION_RAW, {broken_part}, true, &lock); + removePartsFromWorkingSet(NO_TRANSACTION_RAW, {broken_part}, true, lock); broken_part.reset(); cleanup_thread.wakeup(); }; @@ -4763,8 +4766,14 @@ void StorageReplicatedMergeTree::removePartAndEnqueueFetch(const String & part_n /// It's quite dangerous, so clone covered parts to detached. auto broken_part_info = MergeTreePartInfo::fromPartName(part_name, format_version); - auto partition_range = getDataPartsVectorInPartitionForInternalUsage({MergeTreeDataPartState::Active, MergeTreeDataPartState::Outdated}, - broken_part_info.getPartitionId()); + DataPartsVector partition_range; + { + auto parts_lock = readLockParts(); + partition_range = getDataPartsVectorInPartitionForInternalUsage({MergeTreeDataPartState::Active, MergeTreeDataPartState::Outdated}, + broken_part_info.getPartitionId(), + parts_lock); + } + Strings detached_parts; for (const auto & part : partition_range) { @@ -6246,7 +6255,7 @@ void StorageReplicatedMergeTree::foreachActiveParts(Func && func, bool select_se if (select_sequential_consistency) max_added_blocks = getMaxAddedBlocks(); - auto lock = lockParts(); + auto lock = readLockParts(); /// TODO Transactions: should we count visible parts only? for (const auto & part : getDataPartsStateRange(DataPartState::Active, MergeTreePartInfo::Kind::Regular)) { @@ -6424,7 +6433,7 @@ bool StorageReplicatedMergeTree::optimize( } else { - auto parts_lock = lockParts(); + auto parts_lock = readLockParts(); if (!getAnyPartInPartition(partition_id, parts_lock)) handle_noop("Cannot select parts for optimization: there are no parts in partition {}", partition_id); partition_ids_hint.insert(partition_id); @@ -8234,8 +8243,8 @@ void StorageReplicatedMergeTree::exportPartitionToTable(const PartitionCommand & DataPartsVector parts; { - auto data_parts_lock = lockParts(); - parts = getDataPartsVectorInPartitionForInternalUsage(MergeTreeDataPartState::Active, partition_id, &data_parts_lock); + auto data_parts_lock = readLockParts(); + parts = getDataPartsVectorInPartitionForInternalUsage(MergeTreeDataPartState::Active, partition_id, data_parts_lock); } if (parts.empty()) @@ -9119,7 +9128,7 @@ std::unique_ptr StorageReplicatedMergeTree::rep { /// NOTE: Some covered parts may be missing in src_all_parts if corresponding log entries are not executed yet. - auto parts_lock = src_data.lockParts(); + auto parts_lock = src_data.readLockParts(); src_all_parts = src_data.getVisibleDataPartsVectorInPartition(query_context, partition_id, parts_lock); src_patch_parts = src_data.getPatchPartsVectorForPartition(partition_id, parts_lock); } @@ -9338,10 +9347,10 @@ std::unique_ptr StorageReplicatedMergeTree::rep { auto data_parts_lock = lockParts(); - transaction.commit(&data_parts_lock); + transaction.commit(data_parts_lock); if (replace) { - parts_holder = getDataPartsVectorInPartitionForInternalUsage(MergeTreeDataPartState::Active, drop_range.getPartitionId(), &data_parts_lock); + parts_holder = getDataPartsVectorInPartitionForInternalUsage(MergeTreeDataPartState::Active, drop_range.getPartitionId(), data_parts_lock); /// We ignore the list of parts returned from the function below. We will remove them from zk when executing REPLACE_RANGE removePartsInRangeFromWorkingSetAndGetPartsToRemoveFromZooKeeper(NO_TRANSACTION_RAW, drop_range, data_parts_lock); } @@ -9443,7 +9452,7 @@ void StorageReplicatedMergeTree::movePartitionToTable(const StoragePtr & dest_ta { /// NOTE: Some covered parts may be missing in src_all_parts if corresponding log entries are not executed yet. - auto parts_lock = src_data.lockParts(); + auto parts_lock = src_data.readLockParts(); src_all_parts = src_data.getActivePartsToReplace(drop_range, drop_range_fake_part_name, covering_part, parts_lock); src_patch_parts = src_data.getPatchPartsVectorForPartition(partition_id, parts_lock); } @@ -9600,11 +9609,11 @@ void StorageReplicatedMergeTree::movePartitionToTable(const StoragePtr & dest_ta continue; zkutil::KeeperMultiException::check(code, ops, op_results); - parts_holder = getDataPartsVectorInPartitionForInternalUsage(MergeTreeDataPartState::Active, drop_range.getPartitionId(), &src_data_parts_lock); + parts_holder = getDataPartsVectorInPartitionForInternalUsage(MergeTreeDataPartState::Active, drop_range.getPartitionId(), src_data_parts_lock); /// We ignore the list of parts returned from the function below because we cannot remove them from zk /// because we have not created the DROP_RANGE yet. Yes, MOVE PARTITION is trash. removePartsInRangeFromWorkingSetAndGetPartsToRemoveFromZooKeeper(NO_TRANSACTION_RAW, drop_range, src_data_parts_lock); - transaction.commit(&src_data_parts_lock); + transaction.commit(src_data_parts_lock); } PartLog::addNewParts(getContext(), PartLog::createPartLogEntries(dst_parts, watch.elapsed(), profile_events_scope.getSnapshot())); @@ -10170,7 +10179,7 @@ bool StorageReplicatedMergeTree::addOpsToDropAllPartsInPartition( DataPartsVector patches_in_partition; { - auto parts_lock = lockParts(); + auto parts_lock = readLockParts(); parts_to_detach = grabActivePartsToRemoveForDropRange(NO_TRANSACTION_RAW, drop_range_info, parts_lock); patches_in_partition = getPatchPartsVectorForPartition(partition_id, parts_lock); } @@ -11323,7 +11332,7 @@ bool StorageReplicatedMergeTree::createEmptyPartInsteadOfLost(zkutil::ZooKeeperP MergeTreePartition partition; { - DataPartsLock lock = lockParts(); + auto lock = readLockParts(); auto parts_in_partition = getDataPartsPartitionRange(new_part_info.getPartitionId()); if (!parts_in_partition.empty()) diff --git a/tests/integration/test_cancel_freeze/test.py b/tests/integration/test_cancel_freeze/test.py index d8bc960f4fca3..79f5f5f615cbe 100644 --- a/tests/integration/test_cancel_freeze/test.py +++ b/tests/integration/test_cancel_freeze/test.py @@ -24,11 +24,14 @@ def drop_after_test(): yield finally: node.query("DROP TABLE IF EXISTS tbl SYNC") - node.exec_in_container(["bash", "-c", f"rm -r /var/lib/clickhouse/shadow/"]) + node.exec_in_container(["bash", "-c", f"rm -fr /var/lib/clickhouse/shadow/"]) # Test that FREEZE operation can be cancelled with KILL QUERY. def test_cancel_backup(): + if node.is_built_with_sanitizer(): + pytest.skip("Creating 20K parts under sanitizers can be slow.") + # Freezing so much parts should take at least 2 seconds parts = 20_000 node.query( diff --git a/tests/queries/0_stateless/02340_parts_refcnt_mergetree.sh b/tests/queries/0_stateless/02340_parts_refcnt_mergetree.sh index 9c26ffe94287c..703fa2a1bee5a 100755 --- a/tests/queries/0_stateless/02340_parts_refcnt_mergetree.sh +++ b/tests/queries/0_stateless/02340_parts_refcnt_mergetree.sh @@ -53,11 +53,16 @@ function check_refcnt_for_table() done # NOTE: parts that are used in query will be holded in multiple places, and - # this is where magic 6 came from. Also there could be some other + # this is where magic 8 came from. + # Three of these are held reference held in the MergeTreeData itself + # - one which is the canonical reference to that part + # - another in a precomputed vector of active parts + # - another in a precomputed vector of ranges in active parts (each range in turn refers to the part) + # In addition to this, there also there could be some other # background threads (i.e. asynchronous metrics) that uses the part, so we - # simply filter parts not by "refcount > 1" but with some delta - "3", to + # simply filter parts not by "refcount > 1" but with some delta - "5", to # avoid flakiness. - $CLICKHOUSE_CLIENT -q "select table, name, refcount>=6 from system.parts where database = '$CLICKHOUSE_DATABASE' and table = '$table' and refcount >= 3" + $CLICKHOUSE_CLIENT -q "select table, name, refcount>=8 from system.parts where database = '$CLICKHOUSE_DATABASE' and table = '$table' and refcount >= 5" # Kill the query gracefully. kill -INT $PID