Function tryOutfile

src/Interpreters/executeQuery.cpp:1866–1943

}

void tryOutfile(BlockIO & streams, ASTPtr ast, ContextMutablePtr context)
{
    const ASTQueryWithOutput * ast_query_with_output = dynamic_cast<const ASTQueryWithOutput *>(ast.get());

    if (ast_query_with_output == nullptr || ast_query_with_output->out_file == nullptr
        || !OutfileTarget::checkOutfileWithTcpOnServer(context))
    {
        return;
    }

    try
    {
        String format_name = ast_query_with_output && (ast_query_with_output->format != nullptr)
            ? getIdentifierName(ast_query_with_output->format)
            : context->getDefaultFormat();

        String compression_method_str;
        UInt64 compression_level = 1;
        OutfileTarget::setOutfileCompression(ast_query_with_output, compression_method_str, compression_level);

        const auto & out_path = typeid_cast<const ASTLiteral &>(*ast_query_with_output->out_file).value.safeGet<std::string>();
        OutfileTargetPtr outfile_target = std::make_shared<OutfileTarget>(context, out_path, format_name, compression_method_str, compression_level);
        std::shared_ptr<WriteBuffer> out_buf = outfile_target->getOutfileBuffer();

        auto & pipeline = streams.pipeline;

        if (streams.in)
        {
            BlockOutputStreamPtr out = FormatFactory::instance().getOutputStreamParallelIfPossible(
                format_name, *out_buf, streams.in->getHeader(), context, {});

            copyData(
                *streams.in, *out, []() { return false; }, [&out](const Block &) { out->flush(); });
        }
        else if (pipeline.initialized())
        {
            if (!pipeline.isCompleted())
            {
                pipeline.addSimpleTransform([](const Block & header) { return std::make_shared<MaterializingTransform>(header); });

                OutputFormatPtr out = FormatFactory::instance().getOutputFormatParallelIfPossible(
                    format_name, *out_buf, pipeline.getHeader(), context, outfile_target->outToMultiFile(), {});
                out->setOutFileTarget(outfile_target);

                out->setAutoFlush();
                /// Save previous progress callback if any.
                auto previous_progress_callback = context->getProgressCallback();

                /// NOTE Progress callback takes shared ownership of 'out'.
                pipeline.setProgressCallback([out, previous_progress_callback](const Progress & progress) {
                    if (previous_progress_callback)
                        previous_progress_callback(progress);
                    out->onProgress(progress);
                });

                pipeline.setOutputFormat(std::move(out));
            }
            else