Method execute

src/Interpreters/InterpreterKillQueryQuery.cpp:219–333

BlockIO InterpreterKillQueryQuery::execute()
{
    const auto & query = query_ptr->as<ASTKillQueryQuery &>();

    if (!query.cluster.empty())
        return executeDDLQueryOnCluster(query_ptr, getContext(), getRequiredAccessForDDLOnCluster());

    BlockIO res_io;
    switch (query.type)
    {
    case ASTKillQueryQuery::Type::Query:
    {
            auto context = getContext();
            auto where_clause = DB::collectWhereORClausePredicate(query.where_expression, getContext());
            String query_id;
            std::for_each(where_clause.begin(), where_clause.end(), [&query_id](const std::map<String, Field> & wheres) {
                auto iter = wheres.find("query_id");
                if (iter != wheres.end())
                    query_id = iter->second.get<String>();
            });
            if (!query_id.empty())
                context->getQueueManager()->cancel(query_id);
            Block processes_block = getSelectResult("query_id, user, query", "system.processes");
            if (!processes_block)
                return res_io;

            ProcessList & process_list = context->getProcessList();
            QueryDescriptors queries_to_stop = extractQueriesExceptMeAndCheckAccess(processes_block, context);

            auto header = processes_block.cloneEmpty();
            header.insert(0, {ColumnString::create(), std::make_shared<DataTypeString>(), "kill_status"});

            if (!query.sync || query.test)
            {
                MutableColumns res_columns = header.cloneEmptyColumns();
                for (const auto & query_desc : queries_to_stop)
                {
                    auto code = (query.test) ? CancellationCode::Unknown
                                             : process_list.sendCancelToQuery(query_desc.query_id, (context->is_tenant_user() ? formatTenantEntityName(query_desc.user) : query_desc.user), true);
                    insertResultRow(query_desc.source_num, code, processes_block, header, res_columns);
                }

                res_io.in = std::make_shared<OneBlockInputStream>(header.cloneWithColumns(std::move(res_columns)));
        }
        else
        {
            res_io.in = std::make_shared<SyncKillQueryInputStream>(
                process_list, std::move(queries_to_stop), std::move(processes_block), header);
        }

        break;
    }
    case ASTKillQueryQuery::Type::Mutation:
    {
        Block mutations_block = getSelectResult("database, table, mutation_id, command", "system.mutations");
        if (!mutations_block)
            return res_io;