Function executeSubPipelineWithOneRow

src/Interpreters/executeSubQuery.cpp:169–218

}

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

    Block block;
    std::exception_ptr exception;
    auto thread = ThreadFromGlobalPool([query_context = std::move(query_context), &query, pre_execute, tolerate_multi_rows, &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);

                while (block.rows() == 0 && executor.pull(block))
                    ;

                if (!tolerate_multi_rows && block.rows() != 1)
                    throw Exception(ErrorCodes::TOO_MANY_ROWS, "Unexcepted block");

                Block tmp_block;
                while (tmp_block.rows() == 0 && executor.pull(tmp_block))
                {
                    if (tmp_block.rows() > 0)
                        throw Exception(ErrorCodes::TOO_MANY_ROWS, "Unexcepted block");
                }
            }
        }
        catch (...)
        {
            exception = constructException(query_context);
        }
    });
    thread.join();

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

    return block;
}

void executeSubPipeline(
    const ASTPtr & query,