PolarDB-X 源码解读：DDL 的一生（上）

public Cursor handle(RelNode logicalPlan, ExecutionContext executionContext) {        BaseDdlOperation logicalDdlPlan = (BaseDdlOperation) logicalPlan;
        initDdlContext(logicalDdlPlan, executionContext);
        // Validate the plan first and then return immediately if needed.        boolean returnImmediately = validatePlan(logicalDdlPlan, executionContext);
        .....        setPartitionDbIndexAndPhyTable(logicalDdlPlan);        // Build a specific DDL job by subclass that override buildDdlJob        DdlJob ddlJob = returnImmediately?            new TransientDdlJob():            // @override            buildDdlJob(logicalDdlPlan, executionContext);
        // Validate the DDL job before request.        // @override        validateJob(logicalDdlPlan, ddlJob, executionContext);
        // Handle the client DDL request on the worker side.        handleDdlRequest(ddlJob, executionContext);        .....        return buildResultCursor(logicalDdlPlan, executionContext);    }

## 本文涉及的SQL语句####
create database db1 mode = "auto";
use db1;
create table t1(x int, y int);
## 本文详解的DDL语句
alter table t1 add global index `g_i_y` (`y`) COVERING (`x`) partition by hash(`y`);

private ResultCursor executeQuery(ByteString sql, ExecutionContext executionContext,                                      AtomicBoolean trxPolicyModified) {
        // Get all meta version before optimization        final long[] metaVersions = MdlContext.snapshotMetaVersions();
        // Planner        ExecutionPlan plan = Planner.getInstance().plan(sql, executionContext);
        ...        //for requireMDL transaction        if (requireMdl && enableMdl) {            if (!isClosed()) {                // Acquire meta data lock for each statement modifies table data                acquireTransactionalMdl(sql.toString(), plan, executionContext);            }            ...            //update Plan if metaVersion changed, which indicate meta updated        }        ...
        ResultCursor resultCursor = executor.execute(plan, executionContext);        ...        return resultCursor;    }

public class CreatePartitionGsiJobFactory extends CreateGsiJobFactory{    @Override    protected void excludeResources(Set<String> resources) {        super.excludeResources(resources);        //meta data lock in MetaDB        resources.add(concatWithDot(schemaName, primaryTableName)); //db1.t1        resources.add(concatWithDot(schemaName, indexTableName)); //db1.g_i_y        ...    }    
    @Override    protected ExecutableDdlJob doCreate() {      ...        if (needOnlineSchemaChange) {            bringUpGsi = GsiTaskFactory.addGlobalIndexTasks(                schemaName,                primaryTableName,                indexTableName,                stayAtDeleteOnly,                stayAtWriteOnly,                stayAtBackFill            );        }      ...        List<DdlTask> taskList = new ArrayList<>();        //1. validate        taskList.add(validateTask);        //2. create gsi table        //2.1 insert tablePartition meta for gsi table        taskList.add(createTableAddTablesPartitionInfoMetaTask);        //2.2 create gsi physical table        CreateGsiPhyDdlTask createGsiPhyDdlTask =            new CreateGsiPhyDdlTask(schemaName, primaryTableName, indexTableName, physicalPlanData);        taskList.add(createGsiPhyDdlTask);        //2.3 insert tables meta for gsi table        taskList.add(addTablesMetaTask);        taskList.add(showTableMetaTask);        //3.         //3.1 insert indexes meta for primary table        taskList.add(addIndexMetaTask);        //3.2 gsi status: CREATING -> DELETE_ONLY -> WRITE_ONLY -> WRITE_REORG -> PUBLIC        taskList.addAll(bringUpGsi);        //last tableSyncTask        DdlTask tableSyncTask = new TableSyncTask(schemaName, indexTableName);        taskList.add(tableSyncTask);
        final ExecutableDdlJob4CreatePartitionGsi result = new ExecutableDdlJob4CreatePartitionGsi();        result.addSequentialTasks(taskList);        ....
        return result;    }    ...}

public class CreateTableAddTablesMetaTask extends BaseGmsTask {    @Override    public void executeImpl(Connection metaDbConnection, ExecutionContext executionContext) {        PhyInfoSchemaContext phyInfoSchemaContext = TableMetaChanger.buildPhyInfoSchemaContext(schemaName,            logicalTableName, dbIndex, phyTableName, sequenceBean, tablesExtRecord, partitioned, ifNotExists, sqlKind,            executionContext);        FailPoint.injectRandomExceptionFromHint(executionContext);        FailPoint.injectRandomSuspendFromHint(executionContext);        TableMetaChanger.addTableMeta(metaDbConnection, phyInfoSchemaContext);    }
    @Override    public void rollbackImpl(Connection metaDbConnection, ExecutionContext executionContext) {        TableMetaChanger.removeTableMeta(metaDbConnection, schemaName, logicalTableName, false, executionContext);    }
    @Override    protected void onRollbackSuccess(ExecutionContext executionContext) {        TableMetaChanger.afterRemovingTableMeta(schemaName, logicalTableName);    }}

polardbx-executor/src/main/java/com/alibaba/polardbx/executor/ddl/job/task/basic

com.alibaba.polardbx.executor.ddl.job.task.factory.GsiTaskFactory#addGlobalIndexTasks

·    public static List<DdlTask> addGlobalIndexTasks(String schemaName,                                                    String primaryTableName,                                                    String indexName,                                                    boolean stayAtDeleteOnly,                                                    boolean stayAtWriteOnly,                                                    boolean stayAtBackFill) {        ....        DdlTask writeOnlyTask = new GsiUpdateIndexStatusTask(            schemaName,            primaryTableName,            indexName,            IndexStatus.DELETE_ONLY,            IndexStatus.WRITE_ONLY        ).onExceptionTryRecoveryThenRollback();        ....        taskList.add(deleteOnlyTask);        taskList.add(new TableSyncTask(schemaName, primaryTableName));        ....        taskList.add(writeOnlyTask);        taskList.add(new TableSyncTask(schemaName, primaryTableName));        ...        taskList.add(new LogicalTableBackFillTask(schemaName, primaryTableName, indexName));        ...        taskList.add(writeReOrgTask);        taskList.add(new TableSyncTask(schemaName, primaryTableName));        taskList.add(publicTask);        taskList.add(new TableSyncTask(schemaName, primaryTableName));        return taskList;    }

com.alibaba.polardbx.gms.metadb.table.IndexStatus

insert into t1 values(1,2)

//OptimizeLogicalInsertRule.javaprivate LogicalInsert handlePushdown(LogicalInsert origin, boolean deterministicPushdown, ExecutionContext ec){      ...//other writers        final List<InsertWriter> gsiInsertWriters = new ArrayList<>();    IntStream.range(0, gsiMetas.size()).forEach(i -> {        final TableMeta gsiMeta = gsiMetas.get(i);        final RelOptTable gsiTable = catalog.getTableForMember(ImmutableList.of(schema, gsiMeta.getTableName()));        final List<Integer> gsiValuePermute = gsiColumnMappings.get(i);        final boolean isGsiBroadcast = TableTopologyUtil.isBroadcast(gsiMeta);        final boolean isGsiSingle = TableTopologyUtil.isSingle(gsiMeta);        //different write stragety for corresponding table type.        gsiInsertWriters.add(WriterFactory                             .createInsertOrReplaceWriter(newInsert, gsiTable, sourceRowType, gsiValuePermute, gsiMeta, gsiKeywords,                                                          null, isReplace, isGsiBroadcast, isGsiSingle, isValueSource, ec));        });   ...}

//LogicalInsertWriter.java    protected int executeInsert(LogicalInsert logicalInsert, ExecutionContext executionContext,                                HandlerParams handlerParams) {        ...       final List<InsertWriter> gsiWriters = logicalInsert.getGsiInsertWriters();       gsiWriters.stream()                .map(gsiWriter -> gsiWriter.getInput(executionContext))                .filter(w -> !w.isEmpty())                .forEach(w -> {                    writableGsiCount.incrementAndGet();                    allPhyPlan.addAll(w);                });
//IndexStatus.java...public static final EnumSet<IndexStatus> WRITABLE = EnumSet.of(WRITE_ONLY, WRITE_REORG, PUBLIC, DROP_WRITE_ONLY);
public boolean isWritable() {   return WRITABLE.contains(this);}...

com.alibaba.polardbx.executor.gms.GmsTableMetaManager#tonewversion

   public void tonewversion(String tableName, boolean preemptive, Long initWait, Long interval, TimeUnit timeUnit) {        synchronized (OptimizerContext.getContext(schemaName)) {            GmsTableMetaManager oldSchemaManager =                (GmsTableMetaManager) OptimizerContext.getContext(schemaName).getLatestSchemaManager();            TableMeta currentMeta = oldSchemaManager.getTableWithNull(tableName);
            long version = -1;

            ....//查询当前MetaDB中的元数据版本并将其赋值给vesion
            //1. loadSchema            SchemaManager newSchemaManager =                new GmsTableMetaManager(oldSchemaManager, tableName, rule);            newSchemaManager.init();
            OptimizerContext.getContext(schemaName).setSchemaManager(newSchemaManager);
            //2. mdl(v0).writeLock            final MdlContext context;            if (preemptive) {                context = MdlManager.addContext(schemaName, initWait, interval, timeUnit);            } else {                context = MdlManager.addContext(schemaName, false);            }
            MdlTicket ticket = context.acquireLock(new MdlRequest(1L,                    MdlKey                        .getTableKeyWithLowerTableName(schemaName, currentMeta.getDigest()),                    MdlType.MDL_EXCLUSIVE,                    MdlDuration.MDL_TRANSACTION));
            //3. expireSchemaManager(t1, g_i1, v0)            oldSchemaManager.expire();

            ....//失效使用旧版本元信息的PlanCache.
            context.releaseLock(1L, ticket);        }     }