2017-04-01 09:19:00 +00:00
|
|
|
#include <Interpreters/InterpreterKillQueryQuery.h>
|
|
|
|
#include <Parsers/ASTKillQueryQuery.h>
|
|
|
|
#include <Parsers/queryToString.h>
|
2017-05-23 18:24:43 +00:00
|
|
|
#include <Interpreters/Context.h>
|
2020-11-03 13:47:26 +00:00
|
|
|
#include <Interpreters/executeDDLQueryOnCluster.h>
|
2017-04-01 09:19:00 +00:00
|
|
|
#include <Interpreters/ProcessList.h>
|
|
|
|
#include <Interpreters/executeQuery.h>
|
2019-02-04 13:04:02 +00:00
|
|
|
#include <Interpreters/CancellationCode.h>
|
2020-03-06 17:13:28 +00:00
|
|
|
#include <Interpreters/InterpreterAlterQuery.h>
|
2022-03-10 21:29:58 +00:00
|
|
|
#include <Interpreters/TransactionLog.h>
|
2020-03-06 17:13:28 +00:00
|
|
|
#include <Parsers/ASTAlterQuery.h>
|
|
|
|
#include <Parsers/ParserAlterQuery.h>
|
|
|
|
#include <Parsers/parseQuery.h>
|
2020-03-07 17:37:38 +00:00
|
|
|
#include <Access/ContextAccess.h>
|
2017-04-01 09:19:00 +00:00
|
|
|
#include <Columns/ColumnString.h>
|
2017-07-13 20:58:19 +00:00
|
|
|
#include <Common/typeid_cast.h>
|
2017-04-01 09:19:00 +00:00
|
|
|
#include <DataTypes/DataTypeString.h>
|
|
|
|
#include <Columns/ColumnsNumber.h>
|
|
|
|
#include <DataTypes/DataTypesNumber.h>
|
2021-09-15 19:35:48 +00:00
|
|
|
#include <Processors/Sources/SourceFromSingleChunk.h>
|
2022-05-20 19:49:31 +00:00
|
|
|
#include <Processors/ISource.h>
|
2021-09-15 19:35:48 +00:00
|
|
|
#include <Processors/Executors/PullingPipelineExecutor.h>
|
2022-05-20 19:49:31 +00:00
|
|
|
#include <QueryPipeline/Pipe.h>
|
2019-01-10 18:19:29 +00:00
|
|
|
#include <Storages/IStorage.h>
|
2020-01-24 16:20:36 +00:00
|
|
|
#include <Common/quoteString.h>
|
2016-11-30 17:31:05 +00:00
|
|
|
#include <thread>
|
|
|
|
#include <iostream>
|
2017-03-14 00:14:19 +00:00
|
|
|
#include <cstddef>
|
2016-11-30 17:31:05 +00:00
|
|
|
|
|
|
|
|
|
|
|
namespace DB
|
|
|
|
{
|
|
|
|
|
|
|
|
namespace ErrorCodes
|
|
|
|
{
|
2017-04-01 07:20:54 +00:00
|
|
|
extern const int LOGICAL_ERROR;
|
2020-01-24 16:20:36 +00:00
|
|
|
extern const int ACCESS_DENIED;
|
2021-09-16 16:03:31 +00:00
|
|
|
extern const int NOT_IMPLEMENTED;
|
2016-11-30 17:31:05 +00:00
|
|
|
}
|
|
|
|
|
|
|
|
|
|
|
|
static const char * cancellationCodeToStatus(CancellationCode code)
|
|
|
|
{
|
2017-04-01 07:20:54 +00:00
|
|
|
switch (code)
|
|
|
|
{
|
|
|
|
case CancellationCode::NotFound:
|
|
|
|
return "finished";
|
|
|
|
case CancellationCode::QueryIsNotInitializedYet:
|
|
|
|
return "pending";
|
|
|
|
case CancellationCode::CancelCannotBeSent:
|
2019-02-07 13:53:06 +00:00
|
|
|
return "cant_cancel";
|
2017-04-01 07:20:54 +00:00
|
|
|
case CancellationCode::CancelSent:
|
|
|
|
return "waiting";
|
|
|
|
default:
|
|
|
|
return "unknown_status";
|
2019-01-04 14:18:49 +00:00
|
|
|
}
|
2016-11-30 17:31:05 +00:00
|
|
|
}
|
|
|
|
|
|
|
|
|
|
|
|
struct QueryDescriptor
|
|
|
|
{
|
2017-04-01 07:20:54 +00:00
|
|
|
String query_id;
|
|
|
|
String user;
|
|
|
|
size_t source_num;
|
|
|
|
bool processed = false;
|
2016-11-30 17:31:05 +00:00
|
|
|
|
2020-03-06 17:13:28 +00:00
|
|
|
QueryDescriptor(String query_id_, String user_, size_t source_num_, bool processed_ = false)
|
2017-04-01 07:20:54 +00:00
|
|
|
: query_id(std::move(query_id_)), user(std::move(user_)), source_num(source_num_), processed(processed_) {}
|
2016-11-30 17:31:05 +00:00
|
|
|
};
|
|
|
|
|
|
|
|
using QueryDescriptors = std::vector<QueryDescriptor>;
|
|
|
|
|
|
|
|
|
2019-01-10 18:19:29 +00:00
|
|
|
static void insertResultRow(size_t n, CancellationCode code, const Block & source, const Block & header, MutableColumns & columns)
|
2016-11-30 17:31:05 +00:00
|
|
|
{
|
2018-10-22 08:54:54 +00:00
|
|
|
columns[0]->insert(cancellationCodeToStatus(code));
|
2016-11-30 17:31:05 +00:00
|
|
|
|
2017-12-15 18:23:05 +00:00
|
|
|
for (size_t col_num = 1, size = columns.size(); col_num < size; ++col_num)
|
2019-01-10 18:19:29 +00:00
|
|
|
columns[col_num]->insertFrom(*source.getByName(header.getByPosition(col_num).name).column, n);
|
2016-11-30 17:31:05 +00:00
|
|
|
}
|
|
|
|
|
2021-04-10 23:33:54 +00:00
|
|
|
static QueryDescriptors extractQueriesExceptMeAndCheckAccess(const Block & processes_block, ContextPtr context)
|
2016-11-30 17:31:05 +00:00
|
|
|
{
|
2017-04-01 07:20:54 +00:00
|
|
|
QueryDescriptors res;
|
|
|
|
size_t num_processes = processes_block.rows();
|
|
|
|
res.reserve(num_processes);
|
2016-11-30 17:31:05 +00:00
|
|
|
|
2017-04-01 07:20:54 +00:00
|
|
|
const ColumnString & query_id_col = typeid_cast<const ColumnString &>(*processes_block.getByName("query_id").column);
|
|
|
|
const ColumnString & user_col = typeid_cast<const ColumnString &>(*processes_block.getByName("user").column);
|
2021-04-10 23:33:54 +00:00
|
|
|
const ClientInfo & my_client = context->getProcessListElement()->getClientInfo();
|
2020-03-07 17:37:38 +00:00
|
|
|
|
2020-06-20 22:44:52 +00:00
|
|
|
bool access_denied = false;
|
|
|
|
std::optional<bool> is_kill_query_granted_value;
|
|
|
|
auto is_kill_query_granted = [&]() -> bool
|
2020-03-07 17:37:38 +00:00
|
|
|
{
|
2020-06-20 22:44:52 +00:00
|
|
|
if (!is_kill_query_granted_value)
|
2020-03-07 17:37:38 +00:00
|
|
|
{
|
2021-04-10 23:33:54 +00:00
|
|
|
is_kill_query_granted_value = context->getAccess()->isGranted(AccessType::KILL_QUERY);
|
2020-06-20 22:44:52 +00:00
|
|
|
if (!*is_kill_query_granted_value)
|
|
|
|
access_denied = true;
|
2020-03-07 17:37:38 +00:00
|
|
|
}
|
2020-06-20 22:44:52 +00:00
|
|
|
return *is_kill_query_granted_value;
|
2020-03-07 17:37:38 +00:00
|
|
|
};
|
|
|
|
|
2020-01-24 16:20:36 +00:00
|
|
|
String query_user;
|
2016-11-30 17:31:05 +00:00
|
|
|
|
2017-04-01 07:20:54 +00:00
|
|
|
for (size_t i = 0; i < num_processes; ++i)
|
|
|
|
{
|
2020-01-24 16:20:36 +00:00
|
|
|
if ((my_client.current_query_id == query_id_col.getDataAt(i).toString())
|
|
|
|
&& (my_client.current_user == user_col.getDataAt(i).toString()))
|
2017-04-01 07:20:54 +00:00
|
|
|
continue;
|
2016-11-30 17:31:05 +00:00
|
|
|
|
2020-01-24 16:20:36 +00:00
|
|
|
auto query_id = query_id_col.getDataAt(i).toString();
|
|
|
|
query_user = user_col.getDataAt(i).toString();
|
|
|
|
|
2020-06-20 22:44:52 +00:00
|
|
|
if ((my_client.current_user != query_user) && !is_kill_query_granted())
|
2020-03-07 17:37:38 +00:00
|
|
|
continue;
|
2016-11-30 17:31:05 +00:00
|
|
|
|
2020-03-06 17:13:28 +00:00
|
|
|
res.emplace_back(std::move(query_id), query_user, i, false);
|
2017-04-01 07:20:54 +00:00
|
|
|
}
|
2016-11-30 17:31:05 +00:00
|
|
|
|
2020-03-06 17:13:28 +00:00
|
|
|
if (res.empty() && access_denied)
|
2020-01-24 16:20:36 +00:00
|
|
|
throw Exception("User " + my_client.current_user + " attempts to kill query created by " + query_user, ErrorCodes::ACCESS_DENIED);
|
|
|
|
|
2017-04-01 07:20:54 +00:00
|
|
|
return res;
|
2016-11-30 17:31:05 +00:00
|
|
|
}
|
|
|
|
|
|
|
|
|
2022-05-20 19:49:31 +00:00
|
|
|
class SyncKillQuerySource : public ISource
|
2016-11-30 17:31:05 +00:00
|
|
|
{
|
|
|
|
public:
|
2021-09-15 19:35:48 +00:00
|
|
|
SyncKillQuerySource(ProcessList & process_list_, QueryDescriptors && processes_to_stop_, Block && processes_block_,
|
2017-04-01 07:20:54 +00:00
|
|
|
const Block & res_sample_block_)
|
2022-05-20 19:49:31 +00:00
|
|
|
: ISource(res_sample_block_)
|
2021-09-15 19:35:48 +00:00
|
|
|
, process_list(process_list_)
|
|
|
|
, processes_to_stop(std::move(processes_to_stop_))
|
|
|
|
, processes_block(std::move(processes_block_))
|
2022-03-02 17:22:12 +00:00
|
|
|
, res_sample_block(res_sample_block_)
|
2017-04-01 07:20:54 +00:00
|
|
|
{
|
2018-02-23 10:02:29 +00:00
|
|
|
addTotalRowsApprox(processes_to_stop.size());
|
2017-04-01 07:20:54 +00:00
|
|
|
}
|
|
|
|
|
|
|
|
String getName() const override
|
|
|
|
{
|
|
|
|
return "SynchronousQueryKiller";
|
|
|
|
}
|
|
|
|
|
2021-09-15 19:35:48 +00:00
|
|
|
Chunk generate() override
|
2017-04-01 07:20:54 +00:00
|
|
|
{
|
|
|
|
size_t num_result_queries = processes_to_stop.size();
|
|
|
|
|
|
|
|
if (num_processed_queries >= num_result_queries)
|
2021-09-15 19:35:48 +00:00
|
|
|
return {};
|
2017-04-01 07:20:54 +00:00
|
|
|
|
2017-12-15 18:23:05 +00:00
|
|
|
MutableColumns columns = res_sample_block.cloneEmptyColumns();
|
2017-04-01 07:20:54 +00:00
|
|
|
|
|
|
|
do
|
|
|
|
{
|
|
|
|
for (auto & curr_process : processes_to_stop)
|
|
|
|
{
|
|
|
|
if (curr_process.processed)
|
|
|
|
continue;
|
|
|
|
|
2018-03-05 21:09:39 +00:00
|
|
|
auto code = process_list.sendCancelToQuery(curr_process.query_id, curr_process.user, true);
|
2017-04-01 07:20:54 +00:00
|
|
|
|
2019-02-07 13:53:06 +00:00
|
|
|
if (code != CancellationCode::QueryIsNotInitializedYet && code != CancellationCode::CancelSent)
|
2017-04-01 07:20:54 +00:00
|
|
|
{
|
|
|
|
curr_process.processed = true;
|
2017-12-15 18:23:05 +00:00
|
|
|
insertResultRow(curr_process.source_num, code, processes_block, res_sample_block, columns);
|
2017-04-01 07:20:54 +00:00
|
|
|
++num_processed_queries;
|
|
|
|
}
|
2018-11-01 14:56:37 +00:00
|
|
|
/// Wait if CancelSent
|
2017-04-01 07:20:54 +00:00
|
|
|
}
|
|
|
|
|
|
|
|
/// KILL QUERY could be killed also
|
2018-03-05 21:09:39 +00:00
|
|
|
if (isCancelled())
|
2017-04-01 07:20:54 +00:00
|
|
|
break;
|
|
|
|
|
|
|
|
/// Sleep if there are unprocessed queries
|
|
|
|
if (num_processed_queries < num_result_queries)
|
|
|
|
std::this_thread::sleep_for(std::chrono::milliseconds(100));
|
|
|
|
|
|
|
|
/// Don't produce empty block
|
2017-12-15 18:23:05 +00:00
|
|
|
} while (columns.empty() || columns[0]->empty());
|
2017-04-01 07:20:54 +00:00
|
|
|
|
2021-09-17 12:05:54 +00:00
|
|
|
size_t num_rows = columns.empty() ? 0 : columns.front()->size();
|
2021-09-15 19:35:48 +00:00
|
|
|
return Chunk(std::move(columns), num_rows);
|
2017-04-01 07:20:54 +00:00
|
|
|
}
|
|
|
|
|
|
|
|
ProcessList & process_list;
|
|
|
|
QueryDescriptors processes_to_stop;
|
|
|
|
Block processes_block;
|
|
|
|
Block res_sample_block;
|
|
|
|
size_t num_processed_queries = 0;
|
2016-11-30 17:31:05 +00:00
|
|
|
};
|
|
|
|
|
|
|
|
|
|
|
|
BlockIO InterpreterKillQueryQuery::execute()
|
|
|
|
{
|
2019-03-15 16:14:13 +00:00
|
|
|
const auto & query = query_ptr->as<ASTKillQueryQuery &>();
|
2017-04-01 07:20:54 +00:00
|
|
|
|
2019-03-15 16:14:13 +00:00
|
|
|
if (!query.cluster.empty())
|
2022-04-22 12:15:29 +00:00
|
|
|
{
|
|
|
|
DDLQueryOnClusterParams params;
|
|
|
|
params.access_to_check = getRequiredAccessForDDLOnCluster();
|
|
|
|
return executeDDLQueryOnCluster(query_ptr, getContext(), params);
|
|
|
|
}
|
2018-07-20 05:46:48 +00:00
|
|
|
|
2017-04-01 07:20:54 +00:00
|
|
|
BlockIO res_io;
|
2019-03-15 16:14:13 +00:00
|
|
|
switch (query.type)
|
2019-02-05 13:15:11 +00:00
|
|
|
{
|
|
|
|
case ASTKillQueryQuery::Type::Query:
|
2019-01-10 18:19:29 +00:00
|
|
|
{
|
2019-01-11 13:28:13 +00:00
|
|
|
Block processes_block = getSelectResult("query_id, user, query", "system.processes");
|
2019-01-10 18:19:29 +00:00
|
|
|
if (!processes_block)
|
|
|
|
return res_io;
|
|
|
|
|
2021-04-10 23:33:54 +00:00
|
|
|
ProcessList & process_list = getContext()->getProcessList();
|
|
|
|
QueryDescriptors queries_to_stop = extractQueriesExceptMeAndCheckAccess(processes_block, getContext());
|
2019-01-10 18:19:29 +00:00
|
|
|
|
|
|
|
auto header = processes_block.cloneEmpty();
|
|
|
|
header.insert(0, {ColumnString::create(), std::make_shared<DataTypeString>(), "kill_status"});
|
|
|
|
|
2019-03-15 16:14:13 +00:00
|
|
|
if (!query.sync || query.test)
|
2019-01-10 18:19:29 +00:00
|
|
|
{
|
|
|
|
MutableColumns res_columns = header.cloneEmptyColumns();
|
|
|
|
for (const auto & query_desc : queries_to_stop)
|
|
|
|
{
|
2019-03-15 16:14:13 +00:00
|
|
|
auto code = (query.test) ? CancellationCode::Unknown : process_list.sendCancelToQuery(query_desc.query_id, query_desc.user, true);
|
2019-01-10 18:19:29 +00:00
|
|
|
insertResultRow(query_desc.source_num, code, processes_block, header, res_columns);
|
|
|
|
}
|
2017-04-01 07:20:54 +00:00
|
|
|
|
2021-09-15 19:35:48 +00:00
|
|
|
res_io.pipeline = QueryPipeline(std::make_shared<SourceFromSingleChunk>(header.cloneWithColumns(std::move(res_columns))));
|
2019-01-10 18:19:29 +00:00
|
|
|
}
|
|
|
|
else
|
|
|
|
{
|
2021-09-15 19:35:48 +00:00
|
|
|
res_io.pipeline = QueryPipeline(std::make_shared<SyncKillQuerySource>(
|
|
|
|
process_list, std::move(queries_to_stop), std::move(processes_block), header));
|
2019-01-10 18:19:29 +00:00
|
|
|
}
|
2019-02-05 13:15:11 +00:00
|
|
|
|
|
|
|
break;
|
2019-01-10 18:19:29 +00:00
|
|
|
}
|
2019-02-05 13:15:11 +00:00
|
|
|
case ASTKillQueryQuery::Type::Mutation:
|
2017-04-01 07:20:54 +00:00
|
|
|
{
|
2020-03-06 17:13:28 +00:00
|
|
|
Block mutations_block = getSelectResult("database, table, mutation_id, command", "system.mutations");
|
2019-01-10 18:19:29 +00:00
|
|
|
if (!mutations_block)
|
|
|
|
return res_io;
|
|
|
|
|
|
|
|
const ColumnString & database_col = typeid_cast<const ColumnString &>(*mutations_block.getByName("database").column);
|
|
|
|
const ColumnString & table_col = typeid_cast<const ColumnString &>(*mutations_block.getByName("table").column);
|
|
|
|
const ColumnString & mutation_id_col = typeid_cast<const ColumnString &>(*mutations_block.getByName("mutation_id").column);
|
2020-03-06 17:13:28 +00:00
|
|
|
const ColumnString & command_col = typeid_cast<const ColumnString &>(*mutations_block.getByName("command").column);
|
2019-01-10 18:19:29 +00:00
|
|
|
|
|
|
|
auto header = mutations_block.cloneEmpty();
|
|
|
|
header.insert(0, {ColumnString::create(), std::make_shared<DataTypeString>(), "kill_status"});
|
|
|
|
|
2018-01-07 00:35:44 +00:00
|
|
|
MutableColumns res_columns = header.cloneEmptyColumns();
|
2020-02-17 19:28:25 +00:00
|
|
|
auto table_id = StorageID::createEmpty();
|
2020-03-06 17:13:28 +00:00
|
|
|
AccessRightsElements required_access_rights;
|
2021-04-10 23:33:54 +00:00
|
|
|
auto access = getContext()->getAccess();
|
2020-03-06 17:13:28 +00:00
|
|
|
bool access_denied = false;
|
2017-04-01 07:20:54 +00:00
|
|
|
|
2019-01-10 18:19:29 +00:00
|
|
|
for (size_t i = 0; i < mutations_block.rows(); ++i)
|
2017-04-01 07:20:54 +00:00
|
|
|
{
|
2020-02-17 19:28:25 +00:00
|
|
|
table_id = StorageID{database_col.getDataAt(i).toString(), table_col.getDataAt(i).toString()};
|
2019-01-10 18:19:29 +00:00
|
|
|
auto mutation_id = mutation_id_col.getDataAt(i).toString();
|
2018-11-01 14:56:37 +00:00
|
|
|
|
2019-01-10 18:19:29 +00:00
|
|
|
CancellationCode code = CancellationCode::Unknown;
|
2019-03-15 16:14:13 +00:00
|
|
|
if (!query.test)
|
2019-01-10 18:19:29 +00:00
|
|
|
{
|
2021-04-10 23:33:54 +00:00
|
|
|
auto storage = DatabaseCatalog::instance().tryGetTable(table_id, getContext());
|
2019-01-10 18:19:29 +00:00
|
|
|
if (!storage)
|
|
|
|
code = CancellationCode::NotFound;
|
|
|
|
else
|
2020-01-24 16:20:36 +00:00
|
|
|
{
|
2020-03-06 17:13:28 +00:00
|
|
|
ParserAlterCommand parser;
|
2021-04-10 23:33:54 +00:00
|
|
|
auto command_ast
|
|
|
|
= parseQuery(parser, command_col.getDataAt(i).toString(), 0, getContext()->getSettingsRef().max_parser_depth);
|
|
|
|
required_access_rights = InterpreterAlterQuery::getRequiredAccessForCommand(
|
|
|
|
command_ast->as<const ASTAlterCommand &>(), table_id.database_name, table_id.table_name);
|
2020-06-20 22:44:52 +00:00
|
|
|
if (!access->isGranted(required_access_rights))
|
2020-03-06 17:13:28 +00:00
|
|
|
{
|
|
|
|
access_denied = true;
|
2020-01-24 16:20:36 +00:00
|
|
|
continue;
|
2020-03-06 17:13:28 +00:00
|
|
|
}
|
2019-02-04 13:04:02 +00:00
|
|
|
code = storage->killMutation(mutation_id);
|
2020-01-24 16:20:36 +00:00
|
|
|
}
|
2019-01-10 18:19:29 +00:00
|
|
|
}
|
2018-11-01 14:56:37 +00:00
|
|
|
|
2019-01-10 18:19:29 +00:00
|
|
|
insertResultRow(i, code, mutations_block, header, res_columns);
|
2017-04-01 07:20:54 +00:00
|
|
|
}
|
|
|
|
|
2020-03-06 17:13:28 +00:00
|
|
|
if (res_columns[0]->empty() && access_denied)
|
2020-01-24 16:20:36 +00:00
|
|
|
throw Exception(
|
2020-03-06 17:13:28 +00:00
|
|
|
"Not allowed to kill mutation. To execute this query it's necessary to have the grant " + required_access_rights.toString(),
|
2020-01-24 16:20:36 +00:00
|
|
|
ErrorCodes::ACCESS_DENIED);
|
|
|
|
|
2021-09-15 19:35:48 +00:00
|
|
|
res_io.pipeline = QueryPipeline(Pipe(std::make_shared<SourceFromSingleChunk>(header.cloneWithColumns(std::move(res_columns)))));
|
2019-02-05 13:15:11 +00:00
|
|
|
|
|
|
|
break;
|
|
|
|
}
|
2021-09-16 16:03:31 +00:00
|
|
|
case ASTKillQueryQuery::Type::PartMoveToShard:
|
|
|
|
{
|
|
|
|
if (query.sync)
|
|
|
|
throw Exception(ErrorCodes::NOT_IMPLEMENTED, "SYNC modifier is not supported for this statement.");
|
|
|
|
|
|
|
|
Block moves_block = getSelectResult(
|
|
|
|
"database, table, task_name, task_uuid, part_name, to_shard, state",
|
|
|
|
"system.part_moves_between_shards");
|
|
|
|
|
|
|
|
if (!moves_block)
|
|
|
|
return res_io;
|
|
|
|
|
|
|
|
const ColumnString & database_col = typeid_cast<const ColumnString &>(*moves_block.getByName("database").column);
|
|
|
|
const ColumnString & table_col = typeid_cast<const ColumnString &>(*moves_block.getByName("table").column);
|
|
|
|
const ColumnUUID & task_uuid_col = typeid_cast<const ColumnUUID &>(*moves_block.getByName("task_uuid").column);
|
|
|
|
|
|
|
|
auto header = moves_block.cloneEmpty();
|
|
|
|
header.insert(0, {ColumnString::create(), std::make_shared<DataTypeString>(), "kill_status"});
|
|
|
|
|
|
|
|
MutableColumns res_columns = header.cloneEmptyColumns();
|
|
|
|
auto table_id = StorageID::createEmpty();
|
|
|
|
AccessRightsElements required_access_rights;
|
|
|
|
auto access = getContext()->getAccess();
|
|
|
|
bool access_denied = false;
|
|
|
|
|
|
|
|
for (size_t i = 0; i < moves_block.rows(); ++i)
|
|
|
|
{
|
|
|
|
table_id = StorageID{database_col.getDataAt(i).toString(), table_col.getDataAt(i).toString()};
|
2022-09-10 03:04:40 +00:00
|
|
|
auto task_uuid = task_uuid_col[i].get<UUID>();
|
2021-09-16 16:03:31 +00:00
|
|
|
|
|
|
|
CancellationCode code = CancellationCode::Unknown;
|
|
|
|
|
|
|
|
if (!query.test)
|
|
|
|
{
|
|
|
|
auto storage = DatabaseCatalog::instance().tryGetTable(table_id, getContext());
|
|
|
|
if (!storage)
|
|
|
|
code = CancellationCode::NotFound;
|
|
|
|
else
|
|
|
|
{
|
|
|
|
ASTAlterCommand alter_command{};
|
|
|
|
alter_command.type = ASTAlterCommand::MOVE_PARTITION;
|
|
|
|
alter_command.move_destination_type = DataDestinationType::SHARD;
|
|
|
|
required_access_rights = InterpreterAlterQuery::getRequiredAccessForCommand(
|
|
|
|
alter_command, table_id.database_name, table_id.table_name);
|
|
|
|
if (!access->isGranted(required_access_rights))
|
|
|
|
{
|
|
|
|
access_denied = true;
|
|
|
|
continue;
|
|
|
|
}
|
|
|
|
code = storage->killPartMoveToShard(task_uuid);
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
insertResultRow(i, code, moves_block, header, res_columns);
|
|
|
|
}
|
|
|
|
|
|
|
|
if (res_columns[0]->empty() && access_denied)
|
|
|
|
throw Exception(
|
|
|
|
"Not allowed to kill move partition. To execute this query it's necessary to have the grant " + required_access_rights.toString(),
|
|
|
|
ErrorCodes::ACCESS_DENIED);
|
|
|
|
|
2021-10-06 08:59:41 +00:00
|
|
|
res_io.pipeline = QueryPipeline(Pipe(std::make_shared<SourceFromSingleChunk>(header.cloneWithColumns(std::move(res_columns)))));
|
2021-09-16 16:03:31 +00:00
|
|
|
|
|
|
|
break;
|
|
|
|
}
|
2022-03-10 21:29:58 +00:00
|
|
|
case ASTKillQueryQuery::Type::Transaction:
|
|
|
|
{
|
|
|
|
getContext()->checkAccess(AccessType::KILL_TRANSACTION);
|
|
|
|
|
|
|
|
Block transactions_block = getSelectResult("tid, tid_hash, elapsed, is_readonly, state", "system.transactions");
|
|
|
|
|
|
|
|
if (!transactions_block)
|
|
|
|
return res_io;
|
|
|
|
|
|
|
|
const ColumnUInt64 & tid_hash_col = typeid_cast<const ColumnUInt64 &>(*transactions_block.getByName("tid_hash").column);
|
|
|
|
|
|
|
|
auto header = transactions_block.cloneEmpty();
|
|
|
|
header.insert(0, {ColumnString::create(), std::make_shared<DataTypeString>(), "kill_status"});
|
|
|
|
MutableColumns res_columns = header.cloneEmptyColumns();
|
|
|
|
|
|
|
|
for (size_t i = 0; i < transactions_block.rows(); ++i)
|
|
|
|
{
|
|
|
|
UInt64 tid_hash = tid_hash_col.getUInt(i);
|
|
|
|
|
|
|
|
CancellationCode code = CancellationCode::Unknown;
|
|
|
|
if (!query.test)
|
|
|
|
{
|
|
|
|
auto txn = TransactionLog::instance().tryGetRunningTransaction(tid_hash);
|
|
|
|
if (txn)
|
|
|
|
{
|
|
|
|
txn->onException();
|
|
|
|
if (txn->getState() == MergeTreeTransaction::ROLLED_BACK)
|
|
|
|
code = CancellationCode::CancelSent;
|
|
|
|
else
|
|
|
|
code = CancellationCode::CancelCannotBeSent;
|
|
|
|
}
|
|
|
|
else
|
|
|
|
{
|
|
|
|
code = CancellationCode::NotFound;
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
insertResultRow(i, code, transactions_block, header, res_columns);
|
|
|
|
}
|
|
|
|
|
|
|
|
res_io.pipeline = QueryPipeline(Pipe(std::make_shared<SourceFromSingleChunk>(header.cloneWithColumns(std::move(res_columns)))));
|
|
|
|
break;
|
|
|
|
}
|
2017-04-01 07:20:54 +00:00
|
|
|
}
|
|
|
|
|
|
|
|
return res_io;
|
2016-11-30 17:31:05 +00:00
|
|
|
}
|
|
|
|
|
2019-01-11 13:28:13 +00:00
|
|
|
Block InterpreterKillQueryQuery::getSelectResult(const String & columns, const String & table)
|
2016-11-30 17:31:05 +00:00
|
|
|
{
|
2019-01-11 13:28:13 +00:00
|
|
|
String select_query = "SELECT " + columns + " FROM " + table;
|
2019-03-11 13:22:51 +00:00
|
|
|
auto & where_expression = query_ptr->as<ASTKillQueryQuery>()->where_expression;
|
2018-12-24 08:17:22 +00:00
|
|
|
if (where_expression)
|
2019-01-11 13:28:13 +00:00
|
|
|
select_query += " WHERE " + queryToString(where_expression);
|
2018-12-24 08:17:22 +00:00
|
|
|
|
2021-09-15 19:35:48 +00:00
|
|
|
auto io = executeQuery(select_query, getContext(), true);
|
|
|
|
PullingPipelineExecutor executor(io.pipeline);
|
|
|
|
Block res;
|
2021-09-20 09:05:34 +00:00
|
|
|
while (!res && executor.pull(res));
|
2019-01-10 18:19:29 +00:00
|
|
|
|
2021-09-20 09:05:34 +00:00
|
|
|
Block tmp_block;
|
2021-09-20 13:08:52 +00:00
|
|
|
while (executor.pull(tmp_block));
|
2019-01-10 18:19:29 +00:00
|
|
|
|
2021-09-20 09:05:34 +00:00
|
|
|
if (tmp_block)
|
2019-01-10 18:19:29 +00:00
|
|
|
throw Exception("Expected one block from input stream", ErrorCodes::LOGICAL_ERROR);
|
|
|
|
|
|
|
|
return res;
|
|
|
|
}
|
|
|
|
|
2020-01-24 16:20:36 +00:00
|
|
|
|
|
|
|
AccessRightsElements InterpreterKillQueryQuery::getRequiredAccessForDDLOnCluster() const
|
|
|
|
{
|
|
|
|
const auto & query = query_ptr->as<ASTKillQueryQuery &>();
|
|
|
|
AccessRightsElements required_access;
|
|
|
|
if (query.type == ASTKillQueryQuery::Type::Query)
|
|
|
|
required_access.emplace_back(AccessType::KILL_QUERY);
|
|
|
|
else if (query.type == ASTKillQueryQuery::Type::Mutation)
|
2021-07-31 18:17:06 +00:00
|
|
|
required_access.emplace_back(
|
|
|
|
AccessType::ALTER_UPDATE
|
|
|
|
| AccessType::ALTER_DELETE
|
|
|
|
| AccessType::ALTER_MATERIALIZE_INDEX
|
|
|
|
| AccessType::ALTER_MATERIALIZE_COLUMN
|
|
|
|
| AccessType::ALTER_MATERIALIZE_TTL
|
|
|
|
);
|
2020-01-24 16:20:36 +00:00
|
|
|
return required_access;
|
|
|
|
}
|
|
|
|
|
2016-11-30 17:31:05 +00:00
|
|
|
}
|