Function prepareCnchTransaction

src/Interpreters/executeQuery.cpp:679–748

}

static TransactionCnchPtr prepareCnchTransaction(ContextMutablePtr context, [[maybe_unused]] ASTPtr & ast)
{
    auto server_type = context->getServerType();

    if (server_type != ServerType::cnch_server && server_type != ServerType::cnch_worker)
        return {};
    if (auto txn = context->getCurrentTransaction(); txn)
    {
        LOG_DEBUG(&Poco::Logger::get("executeQuery"), "Cnch query is already in a transaction " + txn->getTransactionRecord().toString());
        return txn;
    }

    if (server_type == ServerType::cnch_server)
    {
        bool read_only = isReadOnlyTransaction(ast.get());
        auto session_txn = isQueryInInteractiveSession(context, ast)
            ? context->getSessionContext()->getCurrentTransaction()->as<CnchExplicitTransaction>()
            : nullptr;
        TxnTimestamp primary_txn_id = session_txn ? session_txn->getTransactionID() : TxnTimestamp{0};
        auto txn = context->getCnchTransactionCoordinator().createTransaction(
            CreateTransactionOption()
                .setContext(context)
                .setReadOnly(read_only)
                .setForceCleanByDM(context->getSettingsRef().force_clean_transaction_by_dm)
                .setAsyncPostCommit(context->getSettingsRef().async_post_commit)
                .setPrimaryTransactionId(primary_txn_id));
        context->setCurrentTransaction(txn);
        if (session_txn && !read_only)
            session_txn->addStatement(queryToString(ast));
        return txn;
    }
    else if (server_type == ServerType::cnch_worker)
    {
        /// TODO: test it
        bool is_initial_query = (context->getClientInfo().query_kind == ClientInfo::QueryKind::INITIAL_QUERY);

        String database;
        String table;
        if (auto * insert = ast->as<ASTInsertQuery>())
        {
            database = insert->table_id.database_name;
            table = insert->table_id.table_name;
        }
        else if (auto * system = ast->as<ASTSystemQuery>(); system && system->type == ASTSystemQuery::Type::DEDUP)
        {
            database = system->database;
            table = system->table;
        }

        if (is_initial_query && !table.empty())
        {
            if (database.empty())
                database = context->getCurrentDatabase();

            auto storage = DatabaseCatalog::instance().getTable(StorageID(database, table), context);
            if (!dynamic_cast<StorageCnchMergeTree *>(storage.get()) && !dynamic_cast<StorageCloudMergeTree *>(storage.get()))
                return {};