mirror of
https://github.com/ClickHouse/ClickHouse.git
synced 2024-09-20 08:40:50 +00:00
Add requested changes. [#CLICKHOUSE-2120]
This commit is contained in:
parent
891437e531
commit
55390715b1
@ -84,8 +84,9 @@ protected:
|
|||||||
BlockInputStreamPtr query_stream_in;
|
BlockInputStreamPtr query_stream_in;
|
||||||
BlockOutputStreamPtr query_stream_out;
|
BlockOutputStreamPtr query_stream_out;
|
||||||
|
|
||||||
/// Abovemetioned streams haved delayed initialization, use this flag to track initialization
|
/// Abovemetioned streams have delayed initialization, this flag indicates thier initialization
|
||||||
/// Can be set to true by single thread, can be read from multiple
|
/// It is better to use atomic (instead of raw bool) with tryGet/setQueryStreams() thread-safe methods despite that
|
||||||
|
/// now in all contexts ProcessListElement is always used under ProcessList::mutex (and raw bool is also Ok)
|
||||||
std::atomic<bool> query_streams_initialized{false};
|
std::atomic<bool> query_streams_initialized{false};
|
||||||
|
|
||||||
public:
|
public:
|
||||||
@ -146,7 +147,7 @@ public:
|
|||||||
return res;
|
return res;
|
||||||
}
|
}
|
||||||
|
|
||||||
/// Copies pointers to in/out streams
|
/// Copies pointers to in/out streams, it can be called once
|
||||||
void setQueryStreams(const BlockIO & io);
|
void setQueryStreams(const BlockIO & io);
|
||||||
/// Get query in/out pointers
|
/// Get query in/out pointers
|
||||||
bool tryGetQueryStreams(BlockInputStreamPtr & in, BlockOutputStreamPtr & out) const;
|
bool tryGetQueryStreams(BlockInputStreamPtr & in, BlockOutputStreamPtr & out) const;
|
||||||
@ -264,13 +265,13 @@ public:
|
|||||||
{
|
{
|
||||||
NotFound = 0, /// already cancelled
|
NotFound = 0, /// already cancelled
|
||||||
QueryIsNotInitializedYet = 1,
|
QueryIsNotInitializedYet = 1,
|
||||||
CancelCannotBeSended = 2,
|
CancelCannotBeSent = 2,
|
||||||
CancelSended = 3,
|
CancelSent = 3,
|
||||||
Unknown
|
Unknown
|
||||||
};
|
};
|
||||||
|
|
||||||
/// Try call cancel() for input and output streams of query with specified id and user
|
/// Try call cancel() for input and output streams of query with specified id and user
|
||||||
CancellationCode sendCancelToQuery(const String & query_id, const String & intial_user);
|
CancellationCode sendCancelToQuery(const String & current_query_id, const String & current_user);
|
||||||
};
|
};
|
||||||
|
|
||||||
}
|
}
|
||||||
|
@ -7,9 +7,9 @@ namespace DB
|
|||||||
class ASTKillQueryQuery : public ASTQueryWithOutput
|
class ASTKillQueryQuery : public ASTQueryWithOutput
|
||||||
{
|
{
|
||||||
public:
|
public:
|
||||||
ASTPtr where_expression;
|
ASTPtr where_expression; // expression to filter processes from system.processes table
|
||||||
bool sync = false;
|
bool sync = false; // SYNC or ASYNC mode
|
||||||
bool test = false;
|
bool test = false; // does it TEST mode? (doesn't cancel queries just checks and shows them)
|
||||||
|
|
||||||
ASTKillQueryQuery() = default;
|
ASTKillQueryQuery() = default;
|
||||||
|
|
||||||
|
@ -32,9 +32,9 @@ static const char * cancellationCodeToStatus(CancellationCode code)
|
|||||||
return "finished";
|
return "finished";
|
||||||
case CancellationCode::QueryIsNotInitializedYet:
|
case CancellationCode::QueryIsNotInitializedYet:
|
||||||
return "pending";
|
return "pending";
|
||||||
case CancellationCode::CancelCannotBeSended:
|
case CancellationCode::CancelCannotBeSent:
|
||||||
return "error";
|
return "error";
|
||||||
case CancellationCode::CancelSended:
|
case CancellationCode::CancelSent:
|
||||||
return "waiting";
|
return "waiting";
|
||||||
default:
|
default:
|
||||||
return "unknown_status";
|
return "unknown_status";
|
||||||
@ -141,13 +141,13 @@ public:
|
|||||||
|
|
||||||
auto code = process_list.sendCancelToQuery(curr_process.query_id, curr_process.user);
|
auto code = process_list.sendCancelToQuery(curr_process.query_id, curr_process.user);
|
||||||
|
|
||||||
if (code != CancellationCode::QueryIsNotInitializedYet && code != CancellationCode::CancelSended)
|
if (code != CancellationCode::QueryIsNotInitializedYet && code != CancellationCode::CancelSent)
|
||||||
{
|
{
|
||||||
curr_process.processed = true;
|
curr_process.processed = true;
|
||||||
insertResultRow(curr_process.source_num, code, processes_block, res);
|
insertResultRow(curr_process.source_num, code, processes_block, res);
|
||||||
++num_processed_queries;
|
++num_processed_queries;
|
||||||
}
|
}
|
||||||
/// Wait if QueryIsNotInitializedYet or CancelSended
|
/// Wait if QueryIsNotInitializedYet or CancelSent
|
||||||
}
|
}
|
||||||
|
|
||||||
/// KILL QUERY could be killed also
|
/// KILL QUERY could be killed also
|
||||||
|
@ -20,7 +20,7 @@ ProcessList::EntryPtr ProcessList::insert(
|
|||||||
const String & query_, const IAST * ast, const ClientInfo & client_info, const Settings & settings)
|
const String & query_, const IAST * ast, const ClientInfo & client_info, const Settings & settings)
|
||||||
{
|
{
|
||||||
EntryPtr res;
|
EntryPtr res;
|
||||||
bool is_kill_query = ast && dynamic_cast<const ASTKillQueryQuery *>(ast);
|
bool is_kill_query = ast && typeid_cast<const ASTKillQueryQuery *>(ast);
|
||||||
|
|
||||||
{
|
{
|
||||||
std::lock_guard<std::mutex> lock(mutex);
|
std::lock_guard<std::mutex> lock(mutex);
|
||||||
@ -151,7 +151,7 @@ void ProcessListElement::setQueryStreams(const BlockIO & io)
|
|||||||
{
|
{
|
||||||
query_stream_in = io.in;
|
query_stream_in = io.in;
|
||||||
query_stream_out = io.out;
|
query_stream_out = io.out;
|
||||||
query_streams_initialized = true;
|
query_streams_initialized = true; // forces strict memory ordering
|
||||||
}
|
}
|
||||||
|
|
||||||
bool ProcessListElement::tryGetQueryStreams(BlockInputStreamPtr & in, BlockOutputStreamPtr & out) const
|
bool ProcessListElement::tryGetQueryStreams(BlockInputStreamPtr & in, BlockOutputStreamPtr & out) const
|
||||||
@ -229,9 +229,9 @@ ProcessList::CancellationCode ProcessList::sendCancelToQuery(const String & curr
|
|||||||
if (input_stream && (input_stream_casted = dynamic_cast<IProfilingBlockInputStream *>(input_stream.get())))
|
if (input_stream && (input_stream_casted = dynamic_cast<IProfilingBlockInputStream *>(input_stream.get())))
|
||||||
{
|
{
|
||||||
input_stream_casted->cancel();
|
input_stream_casted->cancel();
|
||||||
return CancellationCode::CancelSended;
|
return CancellationCode::CancelSent;
|
||||||
}
|
}
|
||||||
return CancellationCode::CancelCannotBeSended;
|
return CancellationCode::CancelCannotBeSent;
|
||||||
}
|
}
|
||||||
return CancellationCode::QueryIsNotInitializedYet;
|
return CancellationCode::QueryIsNotInitializedYet;
|
||||||
}
|
}
|
||||||
|
Loading…
Reference in New Issue
Block a user