Method processAsyncInsertQuery

src/Server/TCPHandler.cpp:1330–1371

AsynchronousInsertQueue::PushResult TCPHandler::processAsyncInsertQuery(QueryState & state, AsynchronousInsertQueue & insert_queue)
{
    using PushResult = AsynchronousInsertQueue::PushResult;

    startInsertQuery(state);
    Squashing squashing(std::make_shared<const Block>(state.input_header), 0, state.query_context->getSettingsRef()[Setting::async_insert_max_data_size]);

    while (receivePacketsExpectDataConcurrentWithExecutor(state))
    {
        squashing.setHeader(state.block_for_insert.cloneEmpty());

        squashing.add({state.block_for_insert.getColumns(), state.block_for_insert.rows()});
        auto result_chunk = Squashing::squash(squashing.generate(/*flush_if_enough_size*/ true), squashing.getHeader());

        {
            std::lock_guard lock(*callback_mutex);
            sendLogs(state);
            sendInsertProfileEvents(state);
        }

        if (result_chunk)
        {
            auto result = squashing.getHeader()->cloneWithColumns(result_chunk.detachColumns());
            return PushResult
            {
                .status = PushResult::TOO_MUCH_DATA,
                .insert_block = std::move(result),
            };
        }
    }

    Chunk result_chunk = Squashing::squash(
        squashing.flush(),
        squashing.getHeader());
    if (!result_chunk)
    {
        return insert_queue.pushQueryWithBlock(state.parsed_query, squashing.getHeader()->cloneWithoutColumns(), state.query_context);
    }

    auto result = squashing.getHeader()->cloneWithColumns(result_chunk.detachColumns());
    return insert_queue.pushQueryWithBlock(state.parsed_query, std::move(result), state.query_context);
}


void TCPHandler::processInsertQuery(QueryState & state)