開源分布式數(shù)據(jù)庫PolarDB-X源碼解讀

）： 5 ）： ）： ）： ）： ）： ）：）： ）：CN是一個多模塊的Java項目，模塊之間通過接口暴露服務(wù)，模塊關(guān)系記錄在）：包–）：–––––）：––––碼碼e–碼）：g1.整體了解）：?協(xié)議解析是將協(xié)議數(shù)據(jù)對象分發(fā)到具體執(zhí)行邏輯的過程，入口在物理計劃優(yōu)化五個步驟，優(yōu)化產(chǎn)出物理執(zhí)行計劃，傳入執(zhí)行器。優(yōu)化器使用了執(zhí)行器接收到物理執(zhí)行計劃后，首先根據(jù)計劃類型確定執(zhí)行模式，包括）：2.深入了解3.小結(jié)CNServer層的代碼主要包含在polardbx-server模塊中，main函數(shù)位于TddlLauncher。1.CobarServer對象的創(chuàng)建2.參數(shù)加載->CobarConfig.initCobarConfig()->S獲取。3.從MetaDB讀取元數(shù)據(jù)，并初始化實例級的系統(tǒng)組件TddlLauncher.main()->CobarServer.new()->CobarConTddlLauncher.main()->CobarServer.new()->CobarConConfig.initCobarConfig()->ServerLoader.load()->SeMetaDbDataSource.initMetaDbDataSoMetaDbDataSource.initMetaDbDataSoSchemaChangeManager.geSchemaChangeManager.gepolardbx-gms\src\main\resources\ddl\中保存了系統(tǒng)表的表結(jié)構(gòu)，并且使用polardbx.meta.table.d1.t1polardbx.meta.table.d1.t1并且會注冊對應(yīng)的listener，這樣當inst_config表發(fā)生變化的時候，會回調(diào)），7)StorageHaManager4.創(chuàng)建線程池5.CobarServer.init路徑：TddlLauncher.main()->Cob）2)GmsClusterLoader.loadPolarDbXCluster3)warmup6.網(wǎng)絡(luò)層的初始化processorsprocessors=newNIOProcessor[system.getProcessoprocessors[i]=newNIOProces}）：publicpublicNIOAcceptor(Stringname,intFrontendConnectionFactoryfactory,booleathis.serverChannel=ServerSocketthis.serverChannel.}同時，NIOAcceptor也是一個線程，會處理連接建立的請求。當連接建立后，）：7.結(jié)語）：）： ）：）：）：?[模塊]polardbx-cdc-assemble?[模塊]polardbx-cdc-canal）：?[模塊]polardbx-cdc-daemon）：?[模塊]polardbx-cdc-dumper?[模塊]polardbx-cdc-format?[模塊]polardbx-cdc-meta整形的基礎(chǔ)支撐。此外，該模塊還維護了CDC系統(tǒng)庫表的Sql腳本定義），）：?[模塊]polardbx-cdc-storage?[模塊]polardbx-cdc-task）：?[模塊]polardbx-cdc-transfer?binlog_system_config?binlog_task_config?binlog_node_info?binlog_dumper_info?binlog_task_info）：?binlog_logic_meta_history信息。?binlog_phy_ddl_history?binlog_oss_record?binlog_polarx_command?binlog_schedule_history?binlog_storage_history）：?binlog_env_config_history?binlog_schema_history）：(、(、id、,、balance、,、gmt_created、)VALUES(6,100,now());INSERTINTO、transfer_test、.、accounts、(、id、,、balance、,、gmt_created、)VALUES(7,100,now());createdatabasetransfer_test;CREATETABLE、transfer_test、.、accounts、(、id、int(11)NOTNULL,、balance、int(11)NOTNULL,、gmt_created、datetimenotnull,)ENGINE=InnoDBDEFAULTCHARSET=utf8dbpartitionbyhash(、id、)tbpartitionbyhash(、id、)tbpartitions2;INSERTINTO、transfer_test、.、accounts、(、id、,、balance、,、gmt_created、)VALUES(1,100,now());INSERTINTO、transfer_test、.、accounts、(、id、,、balance、,、gmt_created、)VALUES(2,100,now());INSERTINTO、transfer_test、.、accounts、(、id、,、balance、,、gmt_created、)VALUES(3,100,now());INSERTINTO、transfer_test、.、accounts、(、id、,、balance、,、gmt_created、)VALUES(4,100,now());INSERTINTO、transfer_test、.、accounts、(、id、,、balance、,、gmt_created、)VALUES(5,100,now());INSERTINTO、transfer_test、.、accounts、）：INSERTINTO、transfer_test、.、accounts、(、id、,、balance、,、gmt_created、)VALUES(8,100,now());INSERTINTO、transfer_test、.、accounts、(、id、,、balance、,、gmt_created、)VALUES(9,100,now());INSERTINTO、transfer_test、.、accounts、(、id、,、balance、,、gmt_created、)VALUES(10,100,now());dockerrun-itd--namemysql__3309-p3309:3306-e登錄dokcer實例：dockerexec-itmysql__3309bash編輯/etc/mysql/f，a.增加如下配置來關(guān)閉Gtid(polardbx-cdc全局Binlog暫不支持Gtid)gtid_mode=OFFenforce_gtid_consistency=OFFb.更改serverid，避免與主庫重復server_id=2重啟docker實例：dockerrestartmysql__3309stopslave;resetslave;MASTER_HOST='xxx',MASTER_USER='xxx',MASTER_LOG_FILE='binlog.000001',startslave;）：測試程序之后，可以用下面的SQL，驗證兩邊的數(shù)據(jù)是否完全一致SELECTISNULL(balance))))ASUNSIGNED))ASchecksumFROMaccou）：）：）：）：節(jié)。過，這部分邏輯在NIOAcceptor的構(gòu)造函數(shù)中，每個CN進程只啟動一個）：）：封裝了優(yōu)化執(zhí)行部分的邏輯，其中Planner#plan為優(yōu)化器入口，）：）：四、Parser）：）：劃管理。六、Validator）：些對AST改寫的內(nèi)容，主要用于屏蔽相同語義的不同語法結(jié)構(gòu)。然后），）：八、PlanEnumerator替換原來的子樹，而是將生成的新的執(zhí)行計劃保存在RelSubset中，后續(xù)從）：十、PostPlanner）：ExecutorHelper#execute中根據(jù)優(yōu)化器確定的執(zhí)行模式選擇執(zhí)行鏈路。以下以Cursor鏈路中，首先根據(jù)執(zhí)行計中的算子找到對應(yīng)的handler，代碼位置在）：）：）：）：#一個簡單的PolarDB-X中的分庫分表sbtest、k、int(11)NOTNULLDEFAULT'0',、c、char(120)NOTNULLDEFAULT'',、pad、char(60)NOTNULLDEFAULT'',)dbpartitionbyhash(、id、)tbpartitionbyhash(、id、)tbpartitions2;insertintosbtest(id)values(100);values(100)；PolarDB-X接收到該字符串語句后，開始執(zhí)行該SQL，可見）：）：）：}RelNoderelNode=conv）：RelRootroot=sqlToRelConv}）：}}）：）：privateRelNodesqlRewriteCalcitePlanOptimizeRelNodebestPlan=bestPlan.getCluster}）：callreplacedbyRexCallParam.dataindifferenttargetpartitionsmightbedifRexCallParam.BuildFinalPlanVisitor）：ResultCursorresultCurs）：}}）：publicCursorhandle(Re）：}}會根據(jù)來源是否是Select語句選擇不同的執(zhí)行方式，具體執(zhí)行過程在gsiWriter.getInput(executionContext)).）：其中dbIndex是物理庫名，tableNames是物理表名，param保存了這條）：publicint[]executeUpdatPair<String,Map<IntephyTableModify.getDbIndexAndPara//根據(jù)連接創(chuàng)建prepareStatemeps=prepareStatement(}）： 死鎖檢測功能屬于事務(wù)模塊的功能，死鎖檢測任務(wù)則掛載在事務(wù)管理器}}than1s.}}groupDataSources.stream(information.}}}}}碼是ERR_TRANS_DEADLOCK時，就會將當前事務(wù)回滾掉，并給客戶端發(fā)送//Handledeadlockerro}} 實現(xiàn)的一套精簡的定制化Reactor框架。這部分代碼改進自polardbx-sql中的碼。會話。度私有協(xié)議相關(guān)的定時任務(wù)，這個就是XConnectionManager的工作了，件在XConnection這里。JDBC兼容層提供了包括DataSource、Connection、}}}}}List<PolarxResultse}}}}}}}首先XDataSource會根據(jù)存儲的【IP，端口，用戶名】這三元組查找到在我們這個代碼的場景下，由于數(shù)據(jù)源剛新建，后臺的定時任務(wù)還沒跑過，所以），們直接調(diào)用了XConnection中的execQuery，這個函數(shù)等價于直接創(chuàng)建一個首先execQuery會記錄各種調(diào)用信息進行相關(guān)統(tǒng)計，然后會進入關(guān)鍵的經(jīng)過一些列的變量設(shè)置，lazyDB設(shè)置，我們會構(gòu)造一個用于發(fā)送具體請求的），結(jié)果。并緩存到rows里面，而對應(yīng)上述測試代碼中流式執(zhí)行的情況，結(jié)果。）：）：二、Task組件）：）：）：）：）：）：）：privatevoidconsume(TxnMes{}}}）：）： ），我們將重點關(guān)注DDL在執(zhí)行器中的執(zhí)行流程，在閱讀本文前，可預(yù)先閱讀與）：?一條邏輯DDL語句在解析后進入優(yōu)化器，僅做簡單的類型轉(zhuǎn)化后生成publicCursorhandle(Reneeded.booleanreturnImmediat）：validateJob(logicalD}）：），），createdatabasedb1mode="auto";usedb1;createtablet1(xint,yint);altertablet1addglobalindex、g_i_y、(、y、)COVERING(、x、)partitionbyhash(、y、);）：privateResultCursorexecut}}）：}publicclassCreatePaprotectedvoidexclude}）：}bringUpGsi=GsiTas}//3.}）：publicclassCreateTableApublicvoidexecuteI）：}publicvoidrollbackI}protectedvoidonRo}}?addTableMetaTask?TableSyncTask）：executor/src/main/java/com/alibaba/polardbx/execucom.alibaba.polardbx.executor.ddl.job.task.factory.GsiTaspublicstaticList<DdlTask>{）：}）：privateLogicalInserthandle）：}protectedintexecuteIEnumSet.of(WRITE_ONLY}）：?expireSchemaManager(t1,g_i1,v0)：消除舊版本元信息，將新版本元信息）：com.alibaba.polardbx.executor.gms.GmsTableMetpublicvoidtonewversion(StringOptimizerContext.getCOptimizerContext.getCo）：}MdlTicketticket=contex}}）： ）：2.DDLTask3.Worker和Leader）：）：com.alibaba.polardbx.executor.handler.ddl.Locom.alibaba.polardbx.executor.ddl.newengine..alibaba.polardbx.executor.ddl.newengine.）：com.alibaba.polardbx.executor.ddl.newengine.DdlEngin）：}}}}}}}com.alibaba.polardbx.executor.ddl.newengine.）：別隊列。}}}}}）：}}}}）：}}}3.錯誤處理與恢復）：）： ?PolarDB-X強一致分布式事務(wù)原理）：）：的executeSQL方法負責SQL語句的真正執(zhí)行，也負責創(chuàng)建新的事務(wù)對象。信時，就可以通過ExecutionContext拿到事務(wù)對象，再通過事務(wù)對象的、id、intPRIMARYKEY,INSERTINTOtb1VALUES(0,0),(1,1),(2,2),(3,3);--Transaction1:BEGIN;）：）：2.讀分片0publicResultSetexecuteSQ}}privatevoidbeginTr}）：），）：}}}HeldConnectiongroup）：}}}}}）：protectedvoidbeginNonParticip}}3.寫分片1）：}}于在事務(wù)恢復時確定分支事務(wù)所在的分片。在我們的例子中，xid為：'drds-）：}}我們重點關(guān)注一下事務(wù)的提交流程，上述mit()調(diào)用時，會調(diào)用）：}}}}到我們并沒有獲取committimestamp，因為在一階段提交優(yōu)化里，commit）：}}}）：所不同。具體而言，事務(wù)在之前就獲取了分片0的只讀連接，當執(zhí)行到）：}}}}）：碼。3.事務(wù)恢復種情況，事務(wù)恢復時需要正確提交所有分支事務(wù)。另一種情況是，部分分支事務(wù)）：privatevoidrecoverInstance}}}privatebooleanrollBackOrFor）：}}}}）：的操作。在我們插入ABORTED事務(wù)日志后，原本正在提交事務(wù)的線程在插入）：），1.階段一2.階段二3.階段三4.階段四6.現(xiàn)實中的使用場景7.優(yōu)化思路8.具體的優(yōu)化），接。idin(1,2)和select*fromtwhereidin(2,3)。詢字段對齊的全局的二級索引即可，有關(guān)全局二級索引的介紹，可參考鏈接，）： ）：Map<Integer,List<String>>Map<String,Set<String>>cur}}}}}}）：{tableUkMputeIfAArrayList<>()).add(unGlobalIndexMeta.can{ArrayList<>()).add(un}}}}}）：、id、int(11)NOTNULL,、a、int(11)NOTNULL,、b、int(11)NOTNULL,(1,2,3),(2,3,4),(3,4,5);0:id1:g_i_a0:(1,2,3)1:(2,3,4)2:(3,4,5)SELECT1as、value__index、,1a）：），protectedintdoExecute(LMap<String,List<List<String>>MemoryEstimator.calcS）：}}}}}），）：privateMap<String,List<LinewTreeMap<>(String.CASE_INSENSITIVE_ORDE）：beforePkMapping.stream}}}}}}）：protectedintdoExecute(LexecutorContext.getStorageInfo}）：ArrayList<>(replaceSeqAndBu=}）：Optional.ofNullable(tableDeletePks.get(insertIgnore.getLogicalTableName())).map(List::size).orElse(0);}}handlerParams.optimizedWithReturning=true;if(returnIgnored){returnignoredRows;}else{returnaffectedRows;}}else{handlerParams.optimizedWithReturning=false;}//...}最后以一個例子來展現(xiàn)RETURNING優(yōu)化的執(zhí)行流程與邏輯執(zhí)行的不同。通過、id、int(11)NOTNULL,、a、int(11)NOTNULL,、b、int(11)NOTNULL,INSERTINTOtVALUES(1,3,3);）： ），強。），?核心算法目錄?測試代碼目錄1.算法目標Aconsensusalgorithmensuresthatasingleoneamongtheproposedvaluesischosen2.算法實現(xiàn)Phase1.（prepare）Phase2.（accept）碼實現(xiàn)。3.基類Cbaseinstance_id_tmin_stored_instance_id;//Us同時base.h定義了一個節(jié)點的基類Cbase，用于描述了該基礎(chǔ)節(jié)點的狀態(tài)、當前con