forked from ClickHouse/ClickHouse
-
Notifications
You must be signed in to change notification settings - Fork 3
Expand file tree
/
Copy pathDistributedPlanExecutor.h
More file actions
235 lines (185 loc) · 10.2 KB
/
Copy pathDistributedPlanExecutor.h
File metadata and controls
235 lines (185 loc) · 10.2 KB
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
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
#pragma once
#include "config.h"
#include <atomic>
#include <exception>
#include <mutex>
#include <Processors/Chunk.h>
#include <Disks/DiskObjectStorage/ObjectStorages/IObjectStorage_fwd.h>
#include <Interpreters/Context_fwd.h>
#include <Processors/QueryPlan/QueryPlan.h>
#include <IO/Progress.h>
#if CLICKHOUSE_CLOUD
#include <Server/StatelessWorker/StatelessWorkerAllocation_fwd.h>
#endif
#include <Common/DequeWithMemoryTracking.h>
#include <Common/SettingsChanges.h>
#include <Common/UnorderedMapWithMemoryTracking.h>
#include <Common/UnorderedSetWithMemoryTracking.h>
#include <Common/VectorWithMemoryTracking.h>
#include <Core/ProtocolDefines.h>
namespace DB
{
/// Node count Cascades should plan for, matching the executor's worker source:
/// `distributed_plan_workers_num` for local/Cloud-discovery execution, else the static worker
/// cluster size. Returns 0 when no source is available, so the caller can reject distributed planning.
size_t getCascadesPlanningNodeCount(ContextPtr context);
/// Network endpoint of a worker, resolved on the initiator from the cluster config (and
/// server-level defaults). Both ports may differ per node so several workers can share a host.
struct WorkerAddress
{
String host;
UInt16 stateless_worker_port = 0; /// interserver HTTP port the initiator dispatches tasks to
UInt16 streaming_exchange_port = 0; /// port this node accepts streaming-exchange peer connections on
};
/// Producer endpoint of an exchange stream, shipped to consumers so they dial the producer's
/// actual streaming-exchange port.
struct StreamSourceAddress
{
String host;
UInt16 port = 0;
};
class TaskToHostMap : public boost::noncopyable
{
public:
TaskToHostMap(const DistributedQueryPlan & distributed_query_plan_, ContextPtr context_);
/// Out-of-line so the `worker_allocation` deleter is instantiated where `StatelessWorkerAllocation` is complete.
~TaskToHostMap();
const VectorWithMemoryTracking<WorkerAddress> & getWorkerAddresses() const { return worker_addresses; }
const UnorderedMapWithMemoryTracking<String, WorkerAddress> & getTaskHosts() const { return task_hosts; }
const UnorderedMapWithMemoryTracking<String, StreamSourceAddress> & getExchangeStreamSourceHosts() const { return exchange_stream_source_hosts; }
private:
void fillWorkerAddresses(ContextPtr context);
void assignHostsForTasks(const DistributedQueryPlan & distributed_query_plan);
VectorWithMemoryTracking<WorkerAddress> worker_addresses;
UnorderedMapWithMemoryTracking<String, WorkerAddress> task_hosts;
UnorderedMapWithMemoryTracking<String, StreamSourceAddress> exchange_stream_source_hosts;
#if CLICKHOUSE_CLOUD
StatelessWorkerAllocationPtr worker_allocation; /// Keeps leased workers alive for the query lifetime
#endif
};
using TaskToHostMapPtr = std::shared_ptr<const TaskToHostMap>;
struct DistributedQueryPlan;
class QueryStatus;
using QueryStatusPtr = std::shared_ptr<QueryStatus>;
/// Cancellation state shared by the distributed plan executor, the threads tracking its tasks and
/// the pipeline source driving them. Carries the first failure alongside the flag, so a waiter
/// reports why the query stopped instead of a bare `Query was cancelled`.
class DistributedQueryCancellation
{
public:
/// The pipeline cancelled the source: stop, with no failure to report. Deliberately lock-free,
/// so a cancelling thread never waits on a waiter that holds `mutex`.
void cancel() { cancelled = true; }
/// Store the in-flight exception as the query's first failure and cancel.
void recordCurrentException();
bool isCancelled() const { return cancelled; }
/// The first recorded failure; null if the query was cancelled without one (or not at all).
std::exception_ptr getFailure() const;
/// Rethrow the first recorded failure, if there is one.
void rethrowIfFailed() const;
/// Rethrow the first recorded failure. Without one, throw `QUERY_WAS_CANCELLED` if cancelled.
void throwIfCancelled() const;
private:
void rethrowIfFailedLocked() const TSA_REQUIRES(mutex);
std::atomic<bool> cancelled = false;
mutable std::mutex mutex;
std::exception_ptr first_exception TSA_GUARDED_BY(mutex);
};
using DistributedQueryCancellationPtr = std::shared_ptr<DistributedQueryCancellation>;
/// Implements distributed query plan execution logic by executing stages according to dependencies between them.
class DistributedQueryPlanExecutor
{
public:
virtual ~DistributedQueryPlanExecutor() = default;
void start();
bool execute(); /// Returns true if the execution is finished, false if it is still in progress and should be called again later.
virtual void cleanup() = 0;
private:
void startStageWithDependencies(const String & stage_name, UnorderedSetWithMemoryTracking<String> & executed_stages);
protected:
DistributedQueryPlanExecutor(const UUID & unique_query_id_, const DistributedQueryPlan & distributed_query_plan_, ContextPtr context_, DistributedQueryCancellationPtr cancellation_);
virtual void startStage(const String & stage_name, const DistributedQueryStage & stage) = 0;
virtual bool waitForStage(const String & stage_name, std::optional<UInt64> timeout_ms) = 0;
void checkCancelled() const;
const UUID unique_query_id;
const DistributedQueryPlan & distributed_query_plan;
ContextPtr context;
QueryStatusPtr query_status;
DistributedQueryCancellationPtr cancellation;
DequeWithMemoryTracking<String> running_stages;
LoggerPtr logger;
};
std::unique_ptr<DistributedQueryPlanExecutor> createDistributedQueryExecutor(
const UUID & unique_query_id,
const DistributedQueryPlan & distributed_query_plan,
TaskToHostMapPtr task_to_host_map,
ContextPtr context,
DistributedQueryCancellationPtr cancellation);
/// Wake every in-memory exchange waiter of the query; the waiters rethrow `failure` (or a
/// generic cancellation error when it is null) instead of treating the stream as complete.
/// Idempotent and lock-free with respect to the executor lifecycle, so cancellation paths can call
/// it without waiting for the executor mutex.
void cancelDistributedQueryInMemoryExchanges(const UUID & unique_query_id, std::exception_ptr failure);
/// Contains info about hosts assigned to exchange buckets
struct ExchangeStreamSources
{
/// Exchange stream id -> producer endpoint (host + that producer's streaming-exchange port)
UnorderedMapWithMemoryTracking<String, StreamSourceAddress> stream_hosts;
};
/// Minimal serialization version for a task. A version-1 task carries no per-stream ports: the
/// receiving worker dials every producer on its own fallback exchange port. So v1 is safe only
/// when every producer's port equals the destination worker's exchange port; else v2.
UInt64 chooseTaskSerializationVersion(const ExchangeStreamSources & exchange_stream_sources, UInt64 destination_exchange_port);
/// Contains all info to send a task to remote worker
struct DistributedQueryTaskDescription
{
String initial_query_id;
DistributedQueryTask task;
String serialized_query_plan;
ExchangeDescriptions exchanges;
ExchangeStreamSources exchange_stream_sources;
/// The initiator's changed settings, applied on the worker so query limits and execution-affecting
/// settings (e.g. max_memory_usage) are honored remotely.
SettingsChanges settings_changes;
/// Wire-format version to emit, lowered to v1 for legacy-port-only tasks (rolling-upgrade safe).
UInt64 serialization_version = DBMS_DISTRIBUTED_TASK_SERIALIZATION_VERSION;
};
/// Executes a task locally. `distributed_query_id` is the node-independent identifier of the whole
/// distributed query (the same value on every node); it keys the in-memory and streaming exchanges,
/// while `object_storage_path` locates this node's persisted temporary files.
/// `execute_locally` is true only when the whole query runs in-process on the initiator, false on a
/// worker. It selects the exchange transport, so the caller must pass it (see `createExchangeLookup`).
void doExecuteTask(const DistributedQueryTaskDescription & task, ObjectStoragePtr object_storage,
const String & object_storage_path, const String & distributed_query_id, ContextMutablePtr context,
bool execute_locally, std::function<bool()> is_cancelled = nullptr, ProgressCallback progress_callback = nullptr);
/// Returns object storage and path for temporary files
std::pair<ObjectStoragePtr, String> getObjectStorageForTemporaryFiles(const String & unique_temp_file_path, ContextPtr context);
struct ITemporaryFileLookup;
using TemporaryFileLookupPtr = std::shared_ptr<ITemporaryFileLookup>;
/// ITemporaryFileLookup that is used in buildQueryPipeline() to create readers and writers for temporary files by temporary file logical names
TemporaryFileLookupPtr createTemporaryFilesLookup(ObjectStoragePtr object_storage_, const String & object_storage_path_,
const Strings & input_temporary_files_, const Strings & output_temporary_files_);
struct IExchangeLookup;
using ExchangeLookupPtr = std::shared_ptr<IExchangeLookup>;
struct ExchangeDescription;
/// `execute_locally` must be the value the plan was built with, not a fresh read of
/// `distributed_plan_execute_locally`: an ambient read can disagree with the plan and pick a
/// transport the plan holds no hosts for.
ExchangeLookupPtr createExchangeLookup(
const String & query_id,
const ExchangeDescriptions & exchanges_,
const ExchangeStreamSources & exchange_stream_sources,
TemporaryFileLookupPtr temporary_files_,
ContextPtr context,
bool execute_locally);
class IProcessor;
/// Transform that drops zero-row chunks emitted as scheduling ticks by in-memory exchange
/// sources, so they do not escape the exchange path (e.g. to the client as empty `Data` packets).
std::shared_ptr<IProcessor> makeSkipZeroRowChunksTransform(SharedHeader header);
class ICustomResourceHolder;
/// Helper to clean temporary files after query execution
std::shared_ptr<ICustomResourceHolder> makeTemporaryFilesCleaner(ObjectStoragePtr object_storage_, const String & object_storage_path_,
const Strings & temporary_files_);
/// Helper to drop the query's in-memory exchanges once the query pipeline is destroyed.
std::shared_ptr<ICustomResourceHolder> makeInMemoryExchangesCleaner(const String & query_id);
}