Method processData

src/Interpreters/AsynchronousInsertQueue.cpp:975–1284

}

void AsynchronousInsertQueue::processData(
    InsertQuery key, InsertDataPtr data, ContextPtr global_context, ThreadGroupPtr current_query_thread_group, QueueShardFlushTimeHistory & queue_shard_flush_time_history)
try
{
    if (!data)
        return;

    SCOPE_EXIT(CurrentMetrics::sub(CurrentMetrics::PendingAsyncInsert, data->entries.size()));

    DB::setThreadName(ThreadName::ASYNC_INSERT_QUEUE);

    const auto log = getLogger("AsynchronousInsertQueue");
    const auto & insert_query = assert_cast<const ASTInsertQuery &>(*key.query);

    bool internal = true;
    bool async_insert = true;

    /// Disabled query spans. Could be activated by initializing this to a SpanHolder
    std::shared_ptr<OpenTelemetry::SpanHolder> query_span{nullptr};

    /// 'resetParser' doesn't work for parallel parsing.
    key.settings->set("input_format_parallel_parsing", false);
    /// It maybe insert into distributed table.
    /// We want the remote part to decide if the insert will be async or not.
    key.settings->setDefaultValue("async_insert");

    auto insert_context = Context::createCopy(global_context);
    insert_context->makeQueryContext();

    /// Access rights must be checked for the user who executed the initial INSERT query.
    if (key.user_id)
    {
        insert_context->setUser(*key.user_id);
        insert_context->setCurrentRoles(key.current_roles);
    }

    /// Context::setUser only restores the access-control identity, not the ClientInfo user
    /// names. Restore them from the originating query so currentUser()/user()/
    /// authenticatedUser() and the materialized views triggered by the flush observe the
    /// inserting user instead of an empty string.
    insert_context->setCurrentUserName(key.current_user);
    insert_context->setInitialUserName(key.initial_user);
    insert_context->setAuthenticatedUserName(key.authenticated_user);

    insert_context->setSettings(*key.settings);

    /// Set initial_query_id, because it's used in InterpreterInsertQuery for table lock.
    insert_context->setCurrentQueryId(""); // "" means generate a new query id

    auto insert_query_id = insert_context->getCurrentQueryId();
    auto query_start_time = std::chrono::system_clock::now();

    Stopwatch start_watch{CLOCK_MONOTONIC};
    insert_context->setQueryKind(ClientInfo::QueryKind::INITIAL_QUERY);
    insert_context->setInitialQueryStartTime(query_start_time);
    insert_context->setCurrentQueryId(insert_query_id);
    insert_context->setInitialQueryId(insert_query_id);