Function executeQueryByProxy

src/Interpreters/executeQueryHelper.cpp:193–263

}

void executeQueryByProxy(ContextMutablePtr context, const HostWithPorts & server, const ASTPtr & ast, BlockIO & res, bool in_interactive_txn, const String & query)
{
    auto session_txn = in_interactive_txn ? context->getSessionContext()->getCurrentTransaction() : nullptr;
    ProxyTransactionPtr proxy_txn;
    if (session_txn && session_txn->isPrimary())
    {
        proxy_txn = context->getCnchTransactionCoordinator().createProxyTransaction(server, session_txn->getPrimaryTransactionID(),
            isReadOnlyTransaction(ast.get()));
        context->setCurrentTransaction(proxy_txn);
        session_txn->as<CnchExplicitTransaction>()->addStatement(query);
    }

    /// Create connection to host
    const auto & query_client_info = context->getClientInfo();
    auto settings = context->getSettingsRef();
    res.remote_execution_conn = std::make_shared<Connection>(
        server.getHost(),
        server.tcp_port,
        context->getCurrentDatabase(), /*default_database_*/
        query_client_info.current_user,
        query_client_info.current_password,
        "", /*cluster_*/
        "", /*cluster_secret*/
        "server", /*client_name_*/
        Protocol::Compression::Enable,
        Protocol::Secure::Disable);
    res.remote_execution_conn->setDefaultDatabase(context->getCurrentDatabase());

    // PipelineExecutor requires block header.
    LOG_DEBUG(&Poco::Logger::get("executeQuery"), "Sending query as ordinary query");
    Block header;
    if (context->getSettingsRef().enable_select_query_forwarding && ast->as<ASTSelectWithUnionQuery>())
    {
        if (settings.enable_optimizer &&  QueryUseOptimizerChecker::check(ast, context))
            header = InterpreterSelectQueryUseOptimizer(ast, context, SelectQueryOptions(QueryProcessingStage::Complete).analyze()).getSampleBlock();
        else
            header = InterpreterSelectWithUnionQuery(ast, context, SelectQueryOptions(QueryProcessingStage::Complete).analyze()).getSampleBlock();
    }

    Pipes remote_pipes;
    auto remote_query_executor = std::make_shared<RemoteQueryExecutor>(*res.remote_execution_conn, query, header, context);
    remote_query_executor->setPoolMode(PoolMode::GET_ONE);
    remote_query_executor->setServerForwarding(true);
    remote_query_executor->setQueryId(query_client_info.initial_query_id);
    remote_pipes.emplace_back(createRemoteSourcePipe(remote_query_executor, true, false, false, true));
    remote_pipes.back().addInterpreterContext(context);

    auto plan = std::make_unique<QueryPlan>();
    auto read_from_remote = std::make_unique<ReadFromPreparedSource>(Pipe::unitePipes(std::move(remote_pipes)));
    read_from_remote->setStepDescription("Read from remote server");
    plan->addStep(std::move(read_from_remote));
    res.pipeline = std::move(
        *plan->buildQueryPipeline(QueryPlanOptimizationSettings::fromContext(context), BuildQueryPipelineSettings::fromContext(context)));
    res.pipeline.addInterpreterContext(context);

    res.finish_callback = [proxy_txn, context, remote_query_executor](IBlockInputStream *, IBlockOutputStream *, QueryPipeline *) {
        /// Get the extended profile info which is mainly for INSERT SELECT/INSERT INFILE
        context->setExtendedProfileInfo(remote_query_executor->getExtendedProfileInfo());