阿里云+开源分布式数据库PolarDB-X源码解读

(待分享)•2021年10月，在云栖大会上，阿里云正式对外开源了云原生分布式数据库PolarDB-X源码解读(一)：CN代码结构 5PolarDB-X源码解读(二)：CN启动流程 13PolarDB-X源码解读(三)：CDC代码结构 22PolarDB-X源码解读(四)：SQL的一生 37PolarDB-X源码解读(五)：DML之Insert流程 52PolarDB-X源码解读(六)：分布式死锁检测 68PolarDB-X源码解读(七)：私有协议连接的一生(CN篇) 74PolarDB-X源码解读(八)：GlobalBinlog的一生 97PolarDB-X源码解读(九)：DDL的一生(上) 107PolarDB-X源码解读(十)：DDL的一生(下) 123PolarDB-X源码解读(十一)：事务的一生 133PolarDB-X源码解读(十二)：谈谈in常量查询的设计与优化 151PolarDB-X源码解读(十三)：DML之INSERTIGNORE流程 168PolarDB-X源码解读(番外)：如何实现一个Paxos 180PolarDB-X源码解读(一)：CN代码结构特性。PolarDB-X源码解读(一)：CN代码结构6CN是一个多模块的Java项目，模块之间通过接口暴露服务，模块关系记录在eePolarDB-X源码解读(一)：CN代码结构7块。包–PolarDB-X源码解读(一)：CN代码结构8––– –接口(包括登陆鉴权、文本协议、–PolarDB-X源码解读(一)：CN代码结构9––––r码码e–码g1.整体了解1)协议层ver•协议解析是将协议数据对象分发到具体执行逻辑的过程，入口在2)优化器物理计划优化五个步骤，优化产出物理执行计划，传入执行器。优化器使用了3)执行器执行器接收到物理执行计划后，首先根据计划类型确定执行模式，包括t2.深入了解3.小结PolarDB-X源码解读(二)：CN启动流程本文主要讲解PolarDB-X的CN节点(polardbx-sql)的启动过程包括参数加载、CNServer层的代码主要包含在polardbx-server模块中，main函数位于TddlLauncher。1.CobarServer对象的创建2.参数加载rConfiginitCobarConfigServerLoaderloadconfSystemgetPropertyserverconfclasspathserverproperties);Java运行参数中•环境变量中tem3.从MetaDB读取元数据，并初始化实例级的系统组件ginitCobarConfigServerLoaderloadServerLoaderinitts始化元数据库的连接池etaDbDataSourceinitMetaDbDataSourceaDbDataSourcergetMetaDbNamegetMetaDbPropgetMetaDbUsermgetMetaDbPasswdMetaDbDataSource是一个单例，实现了JDBC的程序中可以使用 2)对系统表进行创建或者升级ManagergetInstancehandlepolardbx-gms\src\main\resources\ddl\中保存了系统表的表结构，并且使用agergetInstance4)MetaDbConfigManagerpolardbxpolardbx.meta.table.d1.t15)MetaDbInstConfigManager并且会注册对应的listener，这样当inst_config表发生变化的时候，会回调6)ConnPoolConfigManagerig7)StorageHaManager8)初始化系统库4.创建线程池ernew5.CobarServer.initlLaunchermainCobarServerinit1)逻辑库(TDataSource)的初始化p andlerinitPolarDbXTopology•获取每个DN的信息，包括版本，特性的支持程度等 foManagerinit(有哪些表、每个表有哪些列哪些索引等) etaManagerinittransactionManagerinit))•启动DDL任务引擎(ddlEngineInit())cManagerinit2)GmsClusterLoader.loadPolarDbXCluster3)warmup6.网络层的初始化1)NIOProcessorprocessorsprocessorsnewNIOProcessor[system.getProcessors()];for(inti=0;i<processors.length;i++){processors[i]=newNIOProcessor(i,"Processor"+utorprocessorsistartup;}2)NIOAcceptorpublicpublicNIOAcceptorStringnameintport,onnectionFactoryfactorybooleanonlinethrowsIOException{lectorSelectoropenrverChannelServerSocketChannelopenChannelsocketbindnewAddressport}NIOAcceptor接建立的请求。当连接建立后，3)MPPServer的启动7.结语PolarDBXCDC22PolarDB-X源码解读(三)：CDC代码结构 队列。PolarDBXCDC23能。PolarDBXCDC24•Dumper负责从Task拉取经过处理后的binlog数据，将数据持久化到逻辑ePolarDBXCDC25•[模块]codestye•[模块]docker•[模块]polardbx-cdc-assemble•[模块]polardbx-cdc-canal包(文件)名称PolarDBXCDC26•[模块]polardbx-cdc-common包(文件)名称t•[模块]polardbx-cdc-daemon包(文件)名称PolarDBXCDC27•[模块]polardbx-cdc-dumper包(文件)名称•[模块]polardbx-cdc-format•[模块]polardbx-cdc-meta整形的基础支撑。此外，该模块还维护了CDC系统库表的Sql脚本定义 y•[模块]polardbx-cdc-monitor•[模块]polardbx-cdc-protocolPolarDBXCDC28包(文件)名称•[模块]polardbx-cdc-storage•[模块]polardbx-cdc-task包(文件)名称PolarDBXCDC29•[模块]polardbx-cdc-transfer•binlog_system_config•binlog_task_config•binlog_node_info•binlog_dumper_info•binlog_task_infoC•binlog_logic_meta_history信息。•binlog_phy_ddl_history•binlog_oss_record•binlog_polarx_command化等。•binlog_schedule_history•binlog_storage_historyC•binlog_env_config_history•binlog_schema_historyeDlardbxinstanceidmem_sizeetaDburletaDbusernametaDbPasswdarxurlarxusernamearxpasswordyCbasetransfertestCREATETABLE`transfer_test`.`accounts`(IMARYKEYid)ENGINE=InnoDBDEFAULTCHARSET=utf8dbpartitionbyhash(`id`)tbpartitionbyhashidtbpartitions2;ransfertestaccountsransfertestaccountsransfertestaccountsfertestaccountsransfertestaccountsransfertestaccountsransfertestaccountsransfertestaccountsransfertestaccountsransfertestaccountsdockerrun-itd--namemysql_3309-p3309:3306-eMYSQLROOTPASSWORD=rootmysqlycnfidconsistencyOFFserverid2GEMASTERTOMASTERHOST='xxx',MASTERUSER'xxx',MASTERPASSWORD='xxx',MASTER_PORT=xxx,MASTER_LOG_FILE='binlog.000001',MASTER_LOG_POS=4,MASTERCONNECTRETRY=100;C 测试程序之后，可以用下面的SQL，验证两边的数据是否完全一致SELECTISNULLbalanceASUNSIGNED))ASchecksumFROMaccounts。CerrCPolarDB-X源码解读(四)：SQL的一生节。过，这部分逻辑在NIOAcceptor的构造函数中，每个CN进程只启动一个wSQL封装了优化执行部分的逻辑，其中Planner#plan为优化器入口， SQL在优化器中需要经过Parser-->PlanManagement-->Validator-->SQLtPlannererSELECTFROMtWHEREid>1;SELECTkeywordidentifierFROMkeyword,t1(identifier),WHERE(keword),id(identifier),>(gt),1(literal_int),;TeSELECT*FROMt1WHEREid>1会被转换为SELECT*FROMt1WHEREid>?和一e推工作(更多内容参考PolarDB-X优化器核心技术~计算下推)。替换原来的子树，而是将生成的新的执行计划保存在RelSubset中，后续从eSELECTFROMrJOINtONrname=WHEREr.id=0ANDt.id=1;=WHEREr.id=?ANDt.id=?;由于没有分区键上的相等ExecutorHelper#execute中根据优化器确定的执行模式选择执行链路。以下以Cursor链路中，首先根据执行计中的算子找到对应的handler，代码位置在子转换为一个Cursor接口的具体实现，并且嵌套的调用Local链路的执行入口在SqlQueryLocalExecution#start，首先在LBOPolarDB-X源码解读(五)：DML之Insert流程 CREATETABLE`sbtest`(FAULTEFAULTIMARYKEYidashidtbpartitionbyhashidtbpartitionstosbtestidvaluesvalues(100)；PolarDB-X接收到该字符串语句后，开始执行该SQL，可见xecutionPlanplanPlannergetInstanceplansqlontext征。tdatedNodeconvertervalidateastatorNamespacetargetNamespacegetNamespaceinsertvalidateNamespacetargetNamespaceunknownTypesourceinsertgetSourceif(sourceinstanceofSqlSelect){lectsqlSelectSqlSelectsourcevalidateSelectsqlSelecttargetRowType}else{datorScopescopescopesgetsourcevalidateQuerysourcescopetargetRowType);}RelNoderelNodeconvertertoRelvalidatedNodeplannerContext);ConvertersqlToRelConverternewTddlSqlToRelConverter(...);RelRootrootsqlToRelConverterconvertQuery(validatedNode,false,RelNoderelNodesuperconvertInsertcallif(relNodeinstanceofTableModify){}elVisitortoDrdsRelVisitornewVisitorvalidatedNodeplannerContextRelNodedrdsRelNoderelNodeaccepttoDrdsRelVisitorif((otherinstanceofLogicalTableModify)){if(operation==TableModify.Operation.INSERT||...){calInsertlogicalInsertnewLogicalInsertmodify}}该类，LogicalInsert的内容如下(还有部分变量不在截图中)：i等等。然后便是经过优化器阶段，优化器执行过程代码在privateRelNodesqlRewriteAndPlanEnumerate(RelNodeinput,PlannerContextplannerContext){citePlanOptimizerTracegetOptimizerTracergetaddSnapshotSutplannerContextlNodelogicalOutputoptimizeBySqlWriterinputannerContextitePlanOptimizerTracegetOptimizerTracergetannerContextlNodebestPlanoptimizeByPlanEnumeratorlogicalOutputannerContextuldcleartheplannertoreleasememorybestPlangetCluster).getPlanner().clear();bestPlangetClusterinvalidateMetadataQuery();}publicenumExecutionStrategy{*Foreachrow,existsonlyonetargetpartition.instatementwithfunctioncallnotpushableuencecallreplacedbyRexCallParam*Typicalforsingletableandpartitionedtablewithoutgsi.*Foreachrow,mightexistsmorethanonetargetpartition.instatementwithnondeterministicfunctioncallreplacedbyRexCallParam.*Typicalforbroadcasttable.CPUSHDOWNchrowmightexistsmorethanonetargetpartitionanddataindifferenttargetpartitionsmightbedifferent.executewithallfunctioncallreplacedbyRexCallParam.*Typicalfortablewithgsiortablearedoingscaleout.量)：euildFinalPlanVisitorvisitornewBuildFinalPlanVisitorexecutionPlangetAstplannerContextxecutionPlanecutionPlancopyexecutionPlangetPlanacceptvisitordvaluesesultCursorresultCursorexecutorexecuteplanexecutionContextMyPhyTableModifyCursormodifyCursor=(MyPhyTableModifyCursor)CursorFactoryrepoCursorexecutionContextlogicalPlanowsmodifyCursorbatchUpdatenpublicint[]batchUpdate(){try{erexecuteUpdatethisplan}catch(SQLExceptione){ralUtilnestedExceptione}}publicCursorhandleRelNodelogicalPlan,ExecutionContextontexticalInsertlogicalInsertLogicalInsertlogicalPlanif(!logicalInsert.isSourceSelect()){owsdoExecutelogicalInsertexecutionContextrParams}else{sselectForInsertlogicalInsertnContexthandlerParams}}ertWriterprimaryWriteretPrimaryInsertWriterstRelNodeinputsprimaryWritergetInputexecutionContextistInsertWritergsiWritersetGsiInsertWritersgsiWriter.getInput(executionContext))...;ffectRowsexecutePhysicalPlanallPhyPlanionContextschemaNameisBroadcastpublicintexecuteUpdate(BaseQueryOperationphyTableModify)throwsSQLException{PairStringMapIntegerParameterContextdbIndexAndParam=phyTableModify.getDbIndexAndParam(executionContext.getParams()==ullnullexecutionContextgetParamstiongetPhyConnectiontransactionrwgroupNamelbuildSqlsqlAndParamsqlexecutionContext//根据连接创建prepareStatementpsprepareStatementsql,connection,executionContext,ethodsetParameterspssqlAndParamparamfectRowPreparedStatementpsexecuteUpdate}PolarDB-X源码解读(六)：分布式死锁检测68PolarDB-X源码解读(六)：分布式死锁检测死锁检测功能属于事务模块的功能，死锁检测任务则挂载在事务管理器锁检测(默认是1秒)。if(!hasLeadership()){}PolarDB-X源码解读(六)：分布式死锁检测69nsactioninformationpSetlookupSetfetchTransInfonallongbeforeTimeMillisSystemcurrentTimeMillisLfinallongbeforeTxidIdGeneratorassembleIdbeforeTimeMillis,for(ITransactiontran:transactions){if(!tran.isDistributed()){}kdetectiononlyfortransactionsthattakelongerthan1s.if(tran.getId()>=beforeTxid){}//Getinformationfromthistran.}接的id，事务涉及的分片，分片上分支事务(MySQL上的事务)的连接id等。以PolarDB-X源码解读(六)：分布式死锁检测70lgroupdatasourcesandgroupbyDNsIDhostportfinalMapStringListTGroupDataSourceinstIdGroupList=cUtilsgetInstIdGroupListallSchemasTrxLookupSet全局的事务等待关系图(代码中的GraphTrxLookupSetTransactiongraphnewDiGraphfor(List<TGroupDataSource>groupDataSources:instId2GroupList.values()){if(CollectionUtils.isNotEmpty(groupDataSources)){lldatasourcesareinthesameDNanydatasourceisok.oupDataSourcegroupDataSourceataSourcesget//GetallgroupnamesinthisDN.etStringgroupNamesllectorstoSetkwaitinformationforthisDNtethelookupsetandthegraphwiththeinformation.groupDataSourcegroupNameslookupSet}}fetchLockWaits方法主要就是查询DN上PolarDB-X源码解读(六)：分布式死锁检测71graph.detect().ifPresent((cycle)->{erparseGlobalDeadlockcycleendConnIdcyclegethHYPERLINK"/?target=https%3A///ApsaraDB/galaxysql/