Starrocks 物化视图的实现以及在刷新期间能否读数据

背景

本司在用Starrocks做一些业务上的分析的时候，用到了物化视图，并且在高QPS的情况下，RT也没有很大的波动，所以在此研究一下Starrock的实现，以及在刷新的时候是不是原子性的
本文基于Starrocks 3.3.5

结论

Starrocks的物化视图的更新是通过Insert Overwrite的方式实现的，在执行该SQL期间，会新建临时分区并进行替换，最后在替换分区的时候，会进行表加锁操作，所以说物化视图不存在读不到数据的情况。

分析

这里我们只关心主流程，其他的细节直接跳过

物化视图的创建

直接到 Starrocks.g4

createMaterializedViewStatement

这里会用 AstBuilder.visitCreateMaterializedViewStatement 生成 CreateMaterializedViewStatement和MaterializedViewAnalyzer.visitCreateMaterializedViewStatement 方法进行物化视图的解析,主要是解析里面的SQL，并生成物理计划,
之后再走到StmtExecutor.handleDdlStmt 方法:

    private void handleDdlStmt() throws DdlException {
        try {
            ShowResultSet resultSet = DDLStmtExecutor.execute(parsedStmt, context);
            if (resultSet == null) {
                context.getState().setOk();
            } else {

最后会走到LoaclMetaStore.createMaterializedView方法：

 @Override
    public void createMaterializedView(CreateMaterializedViewStatement stmt)
            throws DdlException {
        // check mv exists,name must be different from view/mv/table which exists in metadata
        String mvName = stmt.getTableName().getTbl();
        String dbName = stmt.getTableName().getDb();
        。。。
        createTaskForMaterializedView(dbName, materializedView, optHints);

这个createTaskForMaterializedView方法

• 有个Task task = TaskBuilder.buildMvTask(materializedView, dbName);方法：

 public static Task buildMvTask(MaterializedView materializedView, String dbName) {
        Task task = new Task(getMvTaskName(materializedView.getId()));
        ...
        task.setDefinition(materializedView.getTaskDefinition());
        task.setPostRun(getAnalyzeMVStmt(materializedView.getName()));
        task.setExpireTime(0L);
        if (ConnectContext.get() != null) {
            task.setCreateUser(ConnectContext.get().getCurrentUserIdentity().getUser());
            task.setUserIdentity(ConnectContext.get().getCurrentUserIdentity());
        }
        handleSpecialTaskProperties(task);
        return task;
    }

其中 materializedView.getTaskDefinition 代码如下：

    public String getTaskDefinition() {
        return String.format("insert overwrite `%s` %s", getName(), getViewDefineSql());
    }

可以看到这里是insert overwrite的方式.

• 与此同时，createTaskForMaterializedView还调用TaskManager.createTask方法用来周期性的调度任务

物化视图的执行

上面看到物化视图的执行是通过Insert overwrite的方式实现的，所以
直接找到Starrocks.g4:

insertStatement

通过AstBuilder.visitInsertStatement解析，解析为InsertStmt,最后会转到StmtExecutor.handleDMLStmtWithProfile:

handleDMLStmt
 ||
 \/
handleInsertOverwrite
 ||
 \/
InsertOverwriteJobMgr.executeJob
 ||
 \/
InsertOverwriteJobRunner.run()
 ||
 \/
InsertOverwriteJobRunner.doLoad()

InsertOverwriteJobRunner.doLoad() 方法如下：

        Preconditions.checkState(job.getJobState() == InsertOverwriteJobState.OVERWRITE_RUNNING);
        createTempPartitions();
        prepareInsert();
        executeInsert();
        doCommit(false);
        transferTo(InsertOverwriteJobState.OVERWRITE_SUCCESS);

• createTempPartitions(); 这个主要创建临时分区。
• prepareInsert/executeInsert 这里主要是进行临时分区的数据写入
• doCommit 这会进行分区的替换

       Locker locker = new Locker();
       if (!locker.lockDatabaseAndCheckExist(db, tableId, LockType.WRITE)) {
           throw new DmlException("insert overwrite commit failed because locking db:%s failed", dbId);
       }
       ...
       targetTable.replacePartition(sourcePartitionNames.get(0), tmpPartitionNames.get(0));
       ...
       } finally {
           locker.unLockDatabase(db, tableId, LockType.WRITE);
       }
  

  不同于之前的数据写入操作,这里会进行锁表的操作，也就是在这期间读写是有互斥的,并且这里面分区(非分区表其实也是有单个分区的存在)的替换都是内存操作，所以会很快。

所以从实现来说，insert overwrite在执行阶段是互斥的，并且只有在元数据操纵期间才会加锁，数据写入阶段是不会加锁的，所以速度是很快的。