Function executeSubPipeline

src/Interpreters/executeSubQuery.cpp:220–264

}

void executeSubPipeline(
    const ASTPtr & query,
    ContextMutablePtr query_context,
    std::function<void(InterpreterSelectQueryUseOptimizer &)> pre_execute,
    std::function<void(Block &)> proc_block)
{
    if (!query->as<ASTSelectWithUnionQuery>())
        throw Exception(
            ErrorCodes::LOGICAL_ERROR,
            "Unrecognized query type '{}' when executing subquery {}",
            query->getID(),
            query->formatForErrorMessage());

    std::exception_ptr exception;
    auto thread = ThreadFromGlobalPool([query_context = std::move(query_context), &query, pre_execute, proc_block, &exception]() {
        try
        {
            CurrentThread::QueryScope query_scope{query_context};
            {
                SelectQueryOptions query_options;
                InterpreterSelectQueryUseOptimizer interpreter{query, query_context, query_options};
                auto block_io = interpreter.execute();
                PullingPipelineExecutor executor(block_io.pipeline);

                pre_execute(interpreter);

                Block block;
                while (executor.pull(block))
                {
                    if (block.rows() == 0)
                        continue;
                    proc_block(block);
                }
            }
        }
        catch (...)
        {
            exception = constructException(query_context);
        }
    });
    thread.join();

    if (exception)
        std::rethrow_exception(exception);
}
}