Repository: ClickHouse/ClickHouse
File: src/Interpreters/executeQuery.cpp (C++)

Code before:
    bool pulling_pipeline = pipeline.pulling();

    std::unique_ptr<WriteBuffer> compressed_buffer;
    try
    {
        if (pipeline.pushing())
        {
            auto pipe = getSourceFromASTInsertQuery(ast, true, pipeline.getHeader(), context, nullptr);
            pipeline.complete(std::move(pipe));
        }
        else if (pipeline.pulling())
        {
            const ASTQueryWithOutput * ast_query_with_output = dynamic_cast<const ASTQueryWithOutput *>(ast.get());
            format_name = ast_query_with_output && ast_query_with_output->format_ast != nullptr
                ? getIdentifierName(ast_query_with_output->format_ast)
                : context->getDefaultFormat();

            WriteBuffer * out_buf = &ostr;
            if (ast_query_with_output && ast_query_with_output->out_file)
            {
                if (!allow_into_outfile)
                    throw Exception(ErrorCodes::INTO_OUTFILE_NOT_ALLOWED, "INTO OUTFILE is not allowed");

                const auto & out_file = typeid_cast<const ASTLiteral &>(*ast_query_with_output->out_file).value.safeGet<std::string>();

                if (context->getSettingsRef()[Setting::into_outfile_create_parent_directories])
                {
                    std::filesystem::path file_path(out_file);
                    std::filesystem::path parent_dir = file_path.parent_path();

                    if (!parent_dir.empty() && !std::filesystem::exists(parent_dir))
                    {
                        try
                        {
                            std::filesystem::create_directories(parent_dir);
                        }
                        catch (const std::filesystem::filesystem_error & e)
                        {
                            throw Exception(ErrorCodes::CANNOT_CREATE_DIRECTORY,
                                "Cannot create parent directories for INTO OUTFILE '{}': {}",
                                parent_dir.string(), e.what());
                        }
                    }
                }

                std::string compression_method;
                if (ast_query_with_output->compression)
                {
                    const auto & compression_method_node = ast_query_with_output->compression->as<ASTLiteral &>();
                    compression_method = compression_method_node.value.safeGet<std::string>();
                }

Review comment (refactor):
This code is executed on the server side and I don't think we should create new directories on server side. Actually I doubth there are places where `INTO OUTFILE` is executed on server side at all, so let's remove this.

Code after:
    try
    {
        if (pipeline.pushing())
        {
            auto pipe = getSourceFromASTInsertQuery(ast, true, pipeline.getHeader(), context, nullptr);
            pipeline.complete(std::move(pipe));
        }
        else if (pipeline.pulling())
        {
            const ASTQueryWithOutput * ast_query_with_output = dynamic_cast<const ASTQueryWithOutput *>(ast.get());
            format_name = ast_query_with_output && ast_query_with_output->format_ast != nullptr
                ? getIdentifierName(ast_query_with_output->format_ast)
                : context->getDefaultFormat();

            WriteBuffer * out_buf = &ostr;
            if (ast_query_with_output && ast_query_with_output->out_file)
            {
                if (!allow_into_outfile)
                    throw Exception(ErrorCodes::INTO_OUTFILE_NOT_ALLOWED, "INTO OUTFILE is not allowed");

                const auto & out_file = typeid_cast<const ASTLiteral &>(*ast_query_with_output->out_file).value.safeGet<std::string>();

                std::string compression_method;
                if (ast_query_with_output->compression)
                {
                    const auto & compression_method_node = ast_query_with_output->compression->as<ASTLiteral &>();
                    compression_method = compression_method_node.value.safeGet<std::string>();
                }
                const auto & settings = context->getSettingsRef();
                compressed_buffer = wrapWriteBufferWithCompressionMethod(
                    std::make_unique<WriteBufferFromFile>(out_file, DBMS_DEFAULT_BUFFER_SIZE, O_WRONLY | O_EXCL | O_CREAT),
                    chooseCompressionMethod(out_file, compression_method),
                    /* compression level = */ static_cast<int>(settings[Setting::output_format_compression_level]),
                    /* zstd_window_log = */ static_cast<int>(settings[Setting::output_format_compression_zstd_window_log]));
            }

            output_format = FormatFactory::instance().getOutputFormatParallelIfPossible(
                format_name,
                compressed_buffer ? *compressed_buffer : *out_buf,
                materializeBlock(pipeline.getHeader()),
                context,
                output_format_settings);

            output_format->setAutoFlush();

            /// Save previous progress callback if any. TODO Do it more conveniently.
            auto previous_progress_callback = context->getProgressCallback();

            /// NOTE Progress callback takes shared ownership of 'out'.
            pipeline.setProgressCallback([output_format, previous_progress_callback] (const Progress & progress)
            {