当前位置：首页 > article >正文

Starrocks 命令 Alter table DISTRIBUTED 重分布数据的实现

article 2025/3/25 21:52:03

背景

在前文Starrocks 写入报错 primary key memory usage exceeds the limit中，可以通过ALTER TABLE xxxx DISTRIBUTED BY HASH(xx) BUCKETS 50;来改变数据的分布状态,具体的执行过程是怎么样的呢？

分析

首先对应的g4文件中为 alterTableStatement ，这里最终的调用是 AlterJobExecutor.visitAlterTableStatement:

if (statement.hasSchemaChangeOp()) {
   Locker locker = new Locker();
   locker.lockTableWithIntensiveDbLock(db, table.getId(), LockType.WRITE);
   try {
       SchemaChangeHandler schemaChangeHandler = GlobalStateMgr.getCurrentState().getSchemaChangeHandler();
       assert table instanceof OlapTable;
       schemaChangeHandler.process(statement.getAlterClauseList(), db, (OlapTable) table);
   } catch (UserException e) {
       throw new AlterJobException(e.getMessage());
   } finally {
       locker.unLockTableWithIntensiveDbLock(db, table, LockType.WRITE);
   }

   isSynchronous = false;

schemaChangeHandler.process会创建OptimizeJobV2 实例去优化对象，数据链路如下：

SchemaChangeHandler.process
      ||
      \/
analyzeAndCreateJob
      ||
      \/
createOptimizeTableJob
      ||
      \/
OptimizeJobV2Builder.build()
      ||
      \/
new OptimizeJobV2()

SchemaChangeHandler.process 会把当前的OptimizeJobV2 job 放入要执行的队列中，之后SchemaChangeHandler 以 alter_scheduler_interval_millisecond （10000ms）的轮询间隔从队列中取出要执行的任务，并调用run方法.run方法如下：

public synchronized void run() {
        if (isTimeout()) {
            cancelImpl("Timeout");
            return;
        }

        // create connectcontext
        createConnectContextIfNeeded();

        try {
            while (true) {
                JobState prevState = jobState;
                switch (prevState) {
                    case PENDING:
                        runPendingJob();
                        break;
                    case WAITING_TXN:
                        runWaitingTxnJob();
                        break;
                    case RUNNING:
                        runRunningJob();
                        break;
                    case FINISHED_REWRITING:
                        runFinishedRewritingJob();
                        break;
                    default:
                        break;
                }
                if (jobState == prevState) {
                    break;
                } // else: handle the new state
            }
        } catch (AlterCancelException e) {
            cancelImpl(e.getMessage());
        }
    }