阿里云+開(kāi)源分布式數(shù)據(jù)庫(kù)PolarDB-X源碼解讀

(待分享)?2021年10月，在云棲大會(huì)上，阿里云正式對(duì)外開(kāi)源了云原生分布式數(shù)據(jù)庫(kù)PolarDB-X源碼解讀(一)：CN代碼結(jié)構(gòu) 5PolarDB-X源碼解讀(二)：CN啟動(dòng)流程 13PolarDB-X源碼解讀(三)：CDC代碼結(jié)構(gòu) 22PolarDB-X源碼解讀(四)：SQL的一生 37PolarDB-X源碼解讀(五)：DML之Insert流程 52PolarDB-X源碼解讀(六)：分布式死鎖檢測(cè) 68PolarDB-X源碼解讀(七)：私有協(xié)議連接的一生(CN篇) 74PolarDB-X源碼解讀(八)：GlobalBinlog的一生 97PolarDB-X源碼解讀(九)：DDL的一生(上) 107PolarDB-X源碼解讀(十)：DDL的一生(下) 123PolarDB-X源碼解讀(十一)：事務(wù)的一生 133PolarDB-X源碼解讀(十二)：談?wù)刬n常量查詢的設(shè)計(jì)與優(yōu)化 151PolarDB-X源碼解讀(十三)：DML之INSERTIGNORE流程 168PolarDB-X源碼解讀(番外)：如何實(shí)現(xiàn)一個(gè)Paxos 180PolarDB-X源碼解讀(一)：CN代碼結(jié)構(gòu)特性。PolarDB-X源碼解讀(一)：CN代碼結(jié)構(gòu)6CN是一個(gè)多模塊的Java項(xiàng)目，模塊之間通過(guò)接口暴露服務(wù)，模塊關(guān)系記錄在eePolarDB-X源碼解讀(一)：CN代碼結(jié)構(gòu)7塊。包–PolarDB-X源碼解讀(一)：CN代碼結(jié)構(gòu)8––– –接口(包括登陸鑒權(quán)、文本協(xié)議、–PolarDB-X源碼解讀(一)：CN代碼結(jié)構(gòu)9––––r碼碼e–碼g1.整體了解1)協(xié)議層ver?協(xié)議解析是將協(xié)議數(shù)據(jù)對(duì)象分發(fā)到具體執(zhí)行邏輯的過(guò)程，入口在2)優(yōu)化器物理計(jì)劃優(yōu)化五個(gè)步驟，優(yōu)化產(chǎn)出物理執(zhí)行計(jì)劃，傳入執(zhí)行器。優(yōu)化器使用了3)執(zhí)行器執(zhí)行器接收到物理執(zhí)行計(jì)劃后，首先根據(jù)計(jì)劃類型確定執(zhí)行模式，包括t2.深入了解3.小結(jié)PolarDB-X源碼解讀(二)：CN啟動(dòng)流程本文主要講解PolarDB-X的CN節(jié)點(diǎn)(polardbx-sql)的啟動(dòng)過(guò)程包括參數(shù)加載、CNServer層的代碼主要包含在polardbx-server模塊中，main函數(shù)位于TddlLauncher。1.CobarServer對(duì)象的創(chuàng)建2.參數(shù)加載rConfiginitCobarConfigServerLoaderloadconfSystemgetPropertyserverconfclasspathserverproperties);Java運(yùn)行參數(shù)中?環(huán)境變量中tem3.從MetaDB讀取元數(shù)據(jù)，并初始化實(shí)例級(jí)的系統(tǒng)組件ginitCobarConfigServerLoaderloadServerLoaderinitts始化元數(shù)據(jù)庫(kù)的連接池etaDbDataSourceinitMetaDbDataSourceaDbDataSourcergetMetaDbNamegetMetaDbPropgetMetaDbUsermgetMetaDbPasswdMetaDbDataSource是一個(gè)單例，實(shí)現(xiàn)了JDBC的程序中可以使用 2)對(duì)系統(tǒng)表進(jìn)行創(chuàng)建或者升級(jí)ManagergetInstancehandlepolardbx-gms\src\main\resources\ddl\中保存了系統(tǒng)表的表結(jié)構(gòu)，并且使用agergetInstance4)MetaDbConfigManagerpolardbxpolardbx.meta.table.d1.t15)MetaDbInstConfigManager并且會(huì)注冊(cè)對(duì)應(yīng)的listener，這樣當(dāng)inst_config表發(fā)生變化的時(shí)候，會(huì)回調(diào)6)ConnPoolConfigManagerig7)StorageHaManager8)初始化系統(tǒng)庫(kù)4.創(chuàng)建線程池ernew5.CobarServer.initlLaunchermainCobarServerinit1)邏輯庫(kù)(TDataSource)的初始化p andlerinitPolarDbXTopology?獲取每個(gè)DN的信息，包括版本，特性的支持程度等 foManagerinit(有哪些表、每個(gè)表有哪些列哪些索引等) etaManagerinittransactionManagerinit))?啟動(dòng)DDL任務(wù)引擎(ddlEngineInit())cManagerinit2)GmsClusterLoader.loadPolarDbXCluster3)warmup6.網(wǎng)絡(luò)層的初始化1)NIOProcessorprocessorsprocessorsnewNIOProcessor[system.getProcessors()];for(inti=0;i<processors.length;i++){processors[i]=newNIOProcessor(i,"Processor"+utorprocessorsistartup;}2)NIOAcceptorpublicpublicNIOAcceptorStringnameintport,onnectionFactoryfactorybooleanonlinethrowsIOException{lectorSelectoropenrverChannelServerSocketChannelopenChannelsocketbindnewAddressport}NIOAcceptor接建立的請(qǐng)求。當(dāng)連接建立后，3)MPPServer的啟動(dòng)7.結(jié)語(yǔ)PolarDBXCDC22PolarDB-X源碼解讀(三)：CDC代碼結(jié)構(gòu) 隊(duì)列。PolarDBXCDC23能。PolarDBXCDC24?Dumper負(fù)責(zé)從Task拉取經(jīng)過(guò)處理后的binlog數(shù)據(jù)，將數(shù)據(jù)持久化到邏輯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整形的基礎(chǔ)支撐。此外，該模塊還維護(hù)了CDC系統(tǒng)庫(kù)表的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 測(cè)試程序之后，可以用下面的SQL，驗(yàn)證兩邊的數(shù)據(jù)是否完全一致SELECTISNULLbalanceASUNSIGNED))ASchecksumFROMaccounts。CerrCPolarDB-X源碼解讀(四)：SQL的一生節(jié)。過(guò)，這部分邏輯在NIOAcceptor的構(gòu)造函數(shù)中，每個(gè)CN進(jìn)程只啟動(dòng)一個(gè)wSQL封裝了優(yōu)化執(zhí)行部分的邏輯，其中Planner#plan為優(yōu)化器入口， SQL在優(yōu)化器中需要經(jīng)過(guò)Parser-->PlanManagement-->Validator-->SQLtPlannererSELECTFROMtWHEREid>1;SELECTkeywordidentifierFROMkeyword,t1(identifier),WHERE(keword),id(identifier),>(gt),1(literal_int),;TeSELECT*FROMt1WHEREid>1會(huì)被轉(zhuǎn)換為SELECT*FROMt1WHEREid>?和一e推工作(更多內(nèi)容參考PolarDB-X優(yōu)化器核心技術(shù)~計(jì)算下推)。替換原來(lái)的子樹(shù)，而是將生成的新的執(zhí)行計(jì)劃保存在RelSubset中，后續(xù)從eSELECTFROMrJOINtONrname=WHEREr.id=0ANDt.id=1;=WHEREr.id=?ANDt.id=?;由于沒(méi)有分區(qū)鍵上的相等ExecutorHelper#execute中根據(jù)優(yōu)化器確定的執(zhí)行模式選擇執(zhí)行鏈路。以下以Cursor鏈路中，首先根據(jù)執(zhí)行計(jì)中的算子找到對(duì)應(yīng)的handler，代碼位置在子轉(zhuǎn)換為一個(gè)Cursor接口的具體實(shí)現(xiàn)，并且嵌套的調(diào)用Local鏈路的執(zhí)行入口在SqlQueryLocalExecution#start，首先在LBOPolarDB-X源碼解讀(五)：DML之Insert流程 CREATETABLE`sbtest`(FAULTEFAULTIMARYKEYidashidtbpartitionbyhashidtbpartitionstosbtestidvaluesvalues(100)；PolarDB-X接收到該字符串語(yǔ)句后，開(kāi)始執(zhí)行該SQL，可見(jiàn)xecutionPlanplanPlannergetInstanceplansqlontext征。tdatedNodeconvertervalidateastatorNamespacetargetNamespacegetNamespaceinsertvalidateNamespacetargetNamespaceunknownTypesourceinsertgetSourceif(sourceinstanceofSqlSelect){lectsqlSelectSqlSelectsourcevalidateSelectsqlSelecttargetRowType}else{datorScopescopescopesgetsourcevalidateQuerysourcescopetargetRowType);}RelNoderelNodeconvertertoRelvalidatedNodeplannerContext);ConvertersqlToRelConverternewTddlSqlToRelConverter(...);RelRootrootsqlToRelConverterconvertQuery(validatedNode,false,RelNoderelNodesuperconvertInsertcallif(relNodeinstanceofTableModify){}elVisitortoDrdsRelVisitornewVisitorvalidatedNodeplannerContextRelNodedrdsRelNoderelNodeaccepttoDrdsRelVisitorif((otherinstanceofLogicalTableModify)){if(operation==TableModify.Operation.INSERT||...){calInsertlogicalInsertnewLogicalInsertmodify}}該類，LogicalInsert的內(nèi)容如下(還有部分變量不在截圖中)：i等等。然后便是經(jīng)過(guò)優(yōu)化器階段，優(yōu)化器執(zhí)行過(guò)程代碼在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//根據(jù)連接創(chuàng)建prepareStatementpsprepareStatementsql,connection,executionContext,ethodsetParameterspssqlAndParamparamfectRowPreparedStatementpsexecuteUpdate}PolarDB-X源碼解讀(六)：分布式死鎖檢測(cè)68PolarDB-X源碼解讀(六)：分布式死鎖檢測(cè)死鎖檢測(cè)功能屬于事務(wù)模塊的功能，死鎖檢測(cè)任務(wù)則掛載在事務(wù)管理器鎖檢測(cè)(默認(rèn)是1秒)。if(!hasLeadership()){}PolarDB-X源碼解讀(六)：分布式死鎖檢測(cè)69nsactioninformationpSetlookupSetfetchTransInfonallongbeforeTimeMillisSystemcurrentTimeMillisLfinallongbeforeTxidIdGeneratorassembleIdbeforeTimeMillis,for(ITransactiontran:transactions){if(!tran.isDistributed()){}kdetectiononlyfortransactionsthattakelongerthan1s.if(tran.getId()>=beforeTxid){}//Getinformationfromthistran.}接的id，事務(wù)涉及的分片，分片上分支事務(wù)(MySQL上的事務(wù))的連接id等。以PolarDB-X源碼解讀(六)：分布式死鎖檢測(cè)70lgroupdatasourcesandgroupbyDNsIDhostportfinalMapStringListTGroupDataSourceinstIdGroupList=cUtilsgetInstIdGroupListallSchemasTrxLookupSet全局的事務(wù)等待關(guān)系圖(代碼中的GraphTrxLookupSetTransactiongraphnewDiGraphfor(List<TGroupDataSource>groupDataSources:instId2GroupList.values()){if(CollectionUtils.isNotEmpty(groupDataSources)){lldatasourcesareinthesameDNanydatasourceisok.oupDataSourcegroupDataSourceataSourcesget//GetallgroupnamesinthisDN.etStringgroupNamesllectorstoSetkwaitinformationforthisDNtethelookupsetandthegraphwiththeinformation.groupDataSourcegroupNameslookupSet}}fetchLockWaits方法主要就是查詢DN上PolarDB-X源碼解讀(六)：分布式死鎖檢測(cè)71graph.detect().ifPresent((cycle)->{erparseGlobalDeadlockcycleendConnIdcyclegethHYPERLINK"/?target=https%3A///ApsaraDB/galaxysql/