Method sendDataFrom

programs/client/Client.cpp:1979–2030

    void sendDataFrom(ReadBuffer & buf, Block & sample, const ColumnsDescription & columns_description)
    {
        String current_format = insert_format;

        /// Data format can be specified in the INSERT query.
        if (const auto * insert = parsed_query->as<ASTInsertQuery>())
        {
            if (!insert->format.empty())
                current_format = insert->format;
        }

        auto source = FormatFactory::instance().getInput(current_format, buf, sample, context, insert_format_max_block_size);
        Pipe pipe(source);

        if (columns_description.hasDefaults())
        {
            pipe.addSimpleTransform([&](const Block & header) {
                return std::make_shared<AddingDefaultsTransform>(header, columns_description, *source, context);
            });
        }

        QueryPipeline pipeline;
        pipeline.init(std::move(pipe));
        PullingAsyncPipelineExecutor executor(pipeline);

        Block block;
        while (executor.pull(block))
        {
            /// Check if server send Log packet
            receiveLogs();

            /// Check if server send Exception packet
            auto packet_type = connection->checkPacket();
            if (packet_type && *packet_type == Protocol::Server::Exception)
            {
                /*
                 * We're exiting with error, so it makes sense to kill the
                 * input stream without waiting for it to complete.
                 */
                executor.cancel();
                return;
            }

            if (block)
            {
                connection->sendData(block);
                processed_rows += block.rows();
            }
        }

        connection->sendData({});
    }


    /// Flush all buffers.