Clickhouse源码分析：ALTER TABLE xxx MODIFY COLUMN yyy流程分析

代码基于21.3.4.25-lts版本


非复制表

# 非复制表
StorageMergeTree::alter
    setProperties // 内存中修改元数据
    alterTable // 修改元数据文件
    startMutation // 确定blocknum，即partname中的mutation的值，添加mutation_blocknum.txt
        background_executor.triggerTask // 后台线程处理merge或者mutate
            selectPartsToMutate
            mutateSelectedPart
                mutatePartToTemporaryPart
                renameTempPartAndReplace
                updateMutationEntriesErrors // 更新状态，并通知mutate完成
    waitForMutation // 等待数据变更结束

复制表

# 复制表
# 复制表的创建，这里主要创建几个重要的监控线程
StorageReplicatedMergeTree::StorageReplicatedMergeTree
    queue // 表的队列，保存了需要处理的操作
    restarting_thread // 重连zk的线程，保证能够一直连接zk ReplicatedMergeTreeRestartingThread::run
        // run -> tryStartup -> cloneReplicaIfNeeded -> cloneReplica 流程会将副本的数据同步到此节点
    queue_updating_task // 队列更新任务，监控zk上的log，将它们加载到queue中
    mutations_updating_task // mutations更新任务，监控zk上的mutations节点。 mutationsUpdatingTask
    merge_selecting_task // 选择part进行merge或者mutate的任务
    background_executor // 后台处理队列的线程

# 复制表执行alter入口    
StorageReplicatedMergeTree::alter
    makeSetRequest // 更新zk上的元数据和列信息
    new_metadata_version = metadata_version + 1; // 确定元数据的版本
    allocateBlockNumbersInAffectedPartitions // 根据zk的自增特性，得到blocknumber。在alter和mutate中，该值用于表示mutation的值
    makeCreateRequest // 在zk的log节点创建新的entry，内容为ReplicatedMergeTreeLogEntryData，类型为ALTER_METADATA
    makeCreateRequest // 在zk的mutations节点下创建entry，内容为ReplicatedMergeTreeMutationEntry
    if replication_alter_partitions_sync == 2 // 等待所有的副本处理结束
        waitForAllReplicasToProcessLogEntry
    if replication_alter_partitions_sync == 1 // 等待本节点处理结束，默认为1
        waitForReplicaToProcessLogEntry // 根据log_pointer判断是否将对应log送到队列，根据queue下的节点是否消失来判断是否处理结束。注意，这时只是元数据的变更，因为log类型是ALTER_METADATA
            getZooKeeper()->waitForDisappear
    waitMutation // 由于前面的等待，元数据已经变更结束，此时等待数据处理结束
    
# mutations更新任务
StorageReplicatedMergeTree::mutationsUpdatingTask
    ReplicatedMergeTreeQueue::updateMutations // 根据zk的mutations节点，刷新mutations_by_znode和mutations_by_partition

# 选择part进行merge或者mutate的任务
StorageReplicatedMergeTree::mergeSelectingTask
    mutations_by_znode.size() > 0
        createLogEntryToMutatePart // 只有zk的parts节点下有part才会进行mutate
        makeCreateRequest // 在zk的log下创建entry，内容为ReplicatedMergeTreeLogEntryData，类型为MUTATE_PART
        
# 队列更新线程 queueUpdatingTask，也就是log entry的监控线程
ReplicatedMergeTreeQueue::pullLogsToQueue // 监控zk上是否有新的log产生
    makeCreateRequest // 将log的内容放到zk上的/queue节点下
    insertUnlocked // 在queue中新增当前entry

# 队列任务处理进程 scheduling_task
StorageReplicatedMergeTree::getDataProcessingJob
    selectQueueEntry
        queue.selectEntryToProcess // 从queue中选择可以执行的log entry
            ReplicatedMergeTreeQueue::shouldExecuteLogEntry
    processQueueEntry // 处理队列中的任务
        processEntry
            executeLogEntry // 先更新元数据，后更新数据
                // 如果该part已经被覆盖，则返回true，该过期内容会被removeProcessedEntry删除。针对GET_PART,MERGE_PARTS,MUTATE_PART
                if entry.type == LogEntry::ALTER_METADATA // 更新元数据
                    executeMetadataAlter
                        setTableStructure
                            setProperties
                            alterTable // 如DatabaseOrdinary::alterTable
                if entry.type == LogEntry::MUTATE_PART // 更新数据
                    tryExecutePartMutation
                        mutatePartToTemporaryPart
                        renameTempPartAndReplace
            removeProcessedEntry // 删除zk上的队列，即queue节点下的内容。也会删除内存中队列的一些信息