blob: 05b3d6565794696f6ddb643a8a5309685ec7482e (
plain) (
blame)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
|
#pragma once
#include <base/types.h>
#include <optional>
#include <mutex>
#include <vector>
#include <atomic>
#include <boost/noncopyable.hpp>
namespace DB
{
class StorageReplicatedMergeTree;
struct ReplicatedMergeTreeLogEntryData;
/// In some use cases merging can be more expensive than fetching
/// (so instead of doing exactly the same merge cluster-wise you can do merge once and fetch ready part)
/// Fetches may be desirable for other operational reasons (backup replica without lot of CPU resources).
///
/// That class allow to take a decisions about preferred strategy for a concrete merge.
///
/// Since that code is used in shouldExecuteLogEntry we need to be able to:
/// 1) make decision fast
/// 2) avoid excessive zookeeper operations
///
/// Because of that we need to cache some important things,
/// like list of active replicas (to limit the number of zookeeper operations)
///
/// That means we need to refresh the state of that object regularly
///
/// NOTE: This class currently supports only single feature (execute_merges_on_single_replica_time_threshold),
/// may be extended to postpone merges in some other scenarios, namely
/// * always_fetch_merged_part
/// * try_fetch_recompressed_part_timeout
/// * (maybe, not for postpone) prefer_fetch_merged_part_time_threshold
///
/// NOTE: execute_merges_on_single_replica_time_threshold feature doesn't provide any strict guarantees.
/// When some replicas are added / removed we may execute some merges on more than one replica,
/// or not execute merge on any of replicas during execute_merges_on_single_replica_time_threshold interval.
/// (so it may be a bad idea to set that threshold to high values).
///
class ReplicatedMergeTreeMergeStrategyPicker: public boost::noncopyable
{
public:
explicit ReplicatedMergeTreeMergeStrategyPicker(StorageReplicatedMergeTree & storage_);
/// triggers refreshing the cached state (list of replicas etc.)
/// used when we get new merge event from the zookeeper queue ( see queueUpdatingTask() etc )
void refreshState();
/// return true if execute_merges_on_single_replica_time_threshold feature is active
/// and we may need to do a fetch (or postpone) instead of merge
bool shouldMergeOnSingleReplica(const ReplicatedMergeTreeLogEntryData & entry) const;
/// returns the replica name
/// and it's not current replica should do the merge
std::optional<String> pickReplicaToExecuteMerge(const ReplicatedMergeTreeLogEntryData & entry);
/// checks (in zookeeper) if the picked replica finished the merge
bool isMergeFinishedByReplica(const String & replica, const ReplicatedMergeTreeLogEntryData & entry);
private:
StorageReplicatedMergeTree & storage;
/// calculate entry hash based on zookeeper path and new part name
/// ATTENTION: it's not a general-purpose hash, it just allows to select replicas consistently
uint64_t getEntryHash(const ReplicatedMergeTreeLogEntryData & entry) const;
std::atomic<time_t> execute_merges_on_single_replica_time_threshold = 0;
std::atomic<time_t> remote_fs_execute_merges_on_single_replica_time_threshold = 0;
std::atomic<time_t> last_refresh_time = 0;
std::mutex mutex;
/// those 2 members accessed under the mutex, only when
/// execute_merges_on_single_replica_time_threshold enabled
int current_replica_index = -1;
std::vector<String> active_replicas;
};
}
|
