Method commit

src/Storages/MergeTree/CnchAttachProcessor.cpp:251–276

}

void AttachContext::commit()
{
    if (new_txn != nullptr)
    {
        query_ctx.getCnchTransactionCoordinator().finishTransaction(new_txn);
    }

    /// If we're not in the interactive transaction session, at this point it's safe
    /// to remove the directory lock
    if (!src_directory.empty() && !isQueryInInteractiveSession(query_ctx.shared_from_this()) && !query_ctx.getSettingsRef().force_clean_transaction_by_dm)
        query_ctx.getCnchCatalog()->clearFilesysLocks({src_directory}, std::nullopt);

    if (!meta_files_to_delete.empty())
    {
        size_t total_records = 0;
        for (const auto & [_, meta_name_records] : meta_files_to_delete)
            total_records += meta_name_records.rename_map.size();
        ThreadPool & pool = getWorkerPool(total_records);
        for (const auto & [_, meta_name_records] : meta_files_to_delete)
        {
            for (const auto & [file_path, _] : meta_name_records.rename_map)
                pool.scheduleOrThrowOnError([&disk = meta_name_records.disk, path = file_path]() { disk->removeFileIfExists(path); });
        }
        pool.wait();
    }
}

void AttachContext::rollback()
{