Greenplum是在開源PostgreSQL的基礎上，采用MPP架構的關系型分布式資料庫。Greenplum被業界認為是最快最具成本效益的資料庫，具有強大的大規模資料分析任務處理能力。
Greenplum采用Shared-Nothing架構，整個叢集由多個資料節點（Segment sever）和控制節點（Master Server）組成，其中的每個資料節點上可以運作多個資料庫。
簡單來說，Shared-Nothing是一個分布式的架構，每個節點相對獨立。在典型的Shared-Nothing中，每一個節點上所有的資源（CPU、記憶體、磁盤）都是獨立的，每個節點都隻有全部資料的一部分，也隻能使用本節點的資源。
由于采用分布式架構，Greenplum 能夠将查詢并行化，以充分發揮叢集的優勢。Segment内部按照規則将資料組織在一起，有助于提高資料查詢性能，利于資料倉庫的維護工作。如下圖所示，Greenplum資料庫是由Master Server、Segment Server和Interconnect三部分組成，Master Server和Segment Server的互聯通過Interconnect實作。
圖1：Greenplum資料庫架構示意
同時，為了最大限度地實作并行化處理，當節點間需要移動資料時，查詢計劃将被分割，而不同Segment間的資料移動就由Interconnect子產品來執行。
在上次的直播中，我們為大家介紹了Greenplum-Interconnect子產品技術特性和實作流程分析，以下内容根據直播文字整理而成。
Interconnect概要介紹
Interconnect是Greenplum資料庫中負責不同節點進行内部資料傳輸的元件。Greenplum資料庫有一種特有的執行算子Motion，負責查詢處理在執行器節點之間交換資料，底層網絡通信協定通過Interconnect實作。
Greenplum資料庫架構中有一些重要的概念，包括查詢排程器（Query Dispatcher，簡稱QD）、查詢執行器（Query Executor，簡稱QE）、執行算子Motion等。
圖2：Master-Segment查詢執行排程架構示意
QD：是指Master節點上負責處理使用者查詢請求的程序。
QE：是指Segment上負責執行 QD 分發來的查詢任務的程序。
通常，QD和QE之間有兩種類型的網絡連接配接：
- libpq是基于TCP的控制流協定。QD通過libpq與各個QE間傳輸控制資訊，包括發送查詢計劃、收集錯誤資訊、處理取消操作等。libpq是PostgreSQL的标準協定，Greenplum對該協定進行了增強，譬如新增了‘M’消息類型 (QD 使用該消息發送查詢計劃給QE）等。
- Interconnect資料流協定：QD和QE、QE和QE之間的表元組資料傳輸通過Interconnect實作，Greenplum有三種Interconnect實作方式，一種基于TCP協定，一種基于UDP協定，還有一種是Proxy協定。預設方式為 UDP Interconnect連接配接方式。
Motion：PostgreSQL生成的查詢計劃隻能在單節點上執行，Greenplum需要将查詢計劃并行化，以充分發揮叢集的優勢。為此，Greenplum引入Motion算子實作查詢計劃的并行化。Motion算子實作資料在不同節點間的傳輸，在Gang之間通過Interconnect進行資料重分布。
同時，Motion為其他算子隐藏了MPP架構和單機的不同，使得其他大多數算子都可以在叢集或者單機上執行。每個Motion 算子都有發送方和接收方。
此外，Greenplum還對某些算子進行了分布式優化，譬如聚集。Motion算子對資料的重分布有gather、broadcast和redistribute三種操作，底層傳輸協定通過Interconnect實作。Interconnect是一個network abstraction layer，負責各節點之間的資料傳輸。
Greenplum是采用Shared-Nothing架構來存儲資料的，按照某個字段哈希計算後打散到不同Segment節點上。當用到連接配接字段之類的操作時，由于這一字段的某一個值可能在不同Segment上面，是以需要在不同節點上對這一字段所有的值重新哈希，然後Segment間通過UDP的方式把這些資料互相發送到對應位置，聚集到各自哈希出的Segment上去形成一個臨時的資料塊以便後續的聚合操作。
Slice：為了在查詢執行期間實作最大的并行度，Greenplum将查詢計劃的工作劃分為slices。Slice是計劃中可以獨立進行處理的部分。查詢計劃會為motion生成slice，motion的每一側都有一個slice。正是由于motion算子将查詢計劃分割為一個個slice，上一層slice對應的程序會讀取下一層各個slice程序廣播或重分布操作，然後進行計算。
Gang：屬于同一個slice但是運作在不同的segment上的程序，稱為Gang。如上圖2所示，圖中有兩個QE節點，一個QD節點，QD節點被劃分為三個slice。按照相同的slice在不同QE上面運作稱一個元件的Gang，是以上圖共有三個Gang。
Interconnect初始化流程
在做好基礎準備工作之後，會有一系列處理函數，将某個節點或所有節點的資料收集上來。在資料傳輸的過程中，會有buffer管理的機制，在一定的時機，将buffer内的資料刷出，這種機制可以有效地降低存儲和網絡的開銷。以下是初始化一些重要的資料結構說明。
1. Interconnect初始化核心結構
| Go typedef enum GpVars_Interconnect_Type { Interconnect_TYPE_TCP = 0, Interconnect_TYPE_UDPIFC, Interconnect_TYPE_PROXY, } GpVars_Interconnect_Type; typedef struct ChunkTransportState { /* array of per-motion-node chunk transport state */ int size;//來自宏定義CTS_INITIAL_SIZE ChunkTransportStateEntry *states;//上一個成員變量定義的size個數 ChunkTransportStateEntry /* keeps track of if we've "activated" connections via SetupInterconnect(). */ bool activated; bool aggressiveRetry; /* whether we've logged when network timeout happens */ bool networkTimeoutIsLogged;//預設false，在ic_udp中才用到 bool teardownActive; List *incompleteConns; /* slice table stuff. */ struct SliceTable *sliceTable; int sliceId;//目前執行slice的索引号 /* Estate pointer for this statement */ struct EState *estate; /* Function pointers to our send/receive functions */ bool (*SendChunk)(struct ChunkTransportState *transportStates, ChunkTransportStateEntry *pEntry, MotionConn *conn, TupleChunkListItem tcItem, int16 motionId); TupleChunkListItem (*RecvTupleChunkFrom)(struct ChunkTransportState *transportStates, int16 motNodeID, int16 srcRoute); TupleChunkListItem (*RecvTupleChunkFromAny)(struct ChunkTransportState *transportStates, int16 motNodeID, int16 *srcRoute); void (*doSendStopMessage)(struct ChunkTransportState *transportStates, int16 motNodeID); void (*SendEos)(struct ChunkTransportState *transportStates, int motNodeID, TupleChunkListItem tcItem); /* ic_proxy backend context */ struct ICProxyBackendContext *proxyContext; } ChunkTransportState; |
2. Interconnect初始化邏輯接口
初始化的流程會調用setup in Interconnect，然後根據資料類型選擇連接配接協定。預設會選擇UDP，使用者也可以配置成TCP。在TCP的流程裡面，會通過GUC宏來判斷走純TCP協定還是走proxy協定。
| Go void SetupInterconnect(EState *estate) { Interconnect_handle_t *h; h = allocate_Interconnect_handle(); Assert(InterconnectContext != NULL); oldContext = MemoryContextSwitchTo(InterconnectContext); if (Gp_Interconnect_type == Interconnect_TYPE_UDPIFC) SetupUDPIFCInterconnect(estate); #here udp初始化流程 else if (Gp_Interconnect_type == Interconnect_TYPE_TCP || Gp_Interconnect_type == Interconnect_TYPE_PROXY) SetupTCPInterconnect(estate);#here tcp & proxy else elog(ERROR, "unsupported expected Interconnect type"); MemoryContextSwitchTo(oldContext); h->Interconnect_context = estate->Interconnect_context; } SetupUDPIFCInterconnect_Internal初始化一些列相關結構,包括Interconnect_context初始化、以及transportStates->states成員createChunkTransportState的初始化，以及rx_buffer_queue相關成員的初始化。 /* rx_buffer_queue */ //緩沖區相關初始化重要參數 conn->pkt_q_capacity = Gp_Interconnect_queue_depth; conn->pkt_q_size = 0; conn->pkt_q_head = 0; conn->pkt_q_tail = 0; conn->pkt_q = (uint8 **) palloc0(conn->pkt_q_capacity * sizeof(uint8 *)); /* update the max buffer count of our rx buffer pool. */ rx_buffer_pool.maxCount += conn->pkt_q_capacity; |
3. Interconnect 初始化回調接口
當初始化的時候，接口回調函數都是統一的。當真正初始化執行時，會給上對應的函數支撐。
通過回調來對應處理函數，在PG裡面是一種常見方式。比如，對于TCP的流程對應RecvTupleChunkFromTCP。
對于UDP的流程，對應TupleChunkFromUDP。相應函數的尾綴規律與TCP或是UDP對應。
| Go TCP & proxy : Interconnect_context->RecvTupleChunkFrom = RecvTupleChunkFromTCP; Interconnect_context->RecvTupleChunkFromAny = RecvTupleChunkFromAnyTCP; Interconnect_context->SendEos = SendEosTCP; Interconnect_context->SendChunk = SendChunkTCP; Interconnect_context->doSendStopMessage = doSendStopMessageTCP; UDP: Interconnect_context->RecvTupleChunkFrom = RecvTupleChunkFromUDPIFC; Interconnect_context->RecvTupleChunkFromAny = RecvTupleChunkFromAnyUDPIFC; Interconnect_context->SendEos = SendEosUDPIFC; Interconnect_context->SendChunk = SendChunkUDPIFC; Interconnect_context->doSendStopMessage = doSendStopMessageUDPIFC; |
Ic_udp流程分析
1. Ic_udp流程分析之緩沖區核心結構
| Go MotionConn： 核心成員變量分析： /* send side queue for packets to be sent */ ICBufferList sndQueue; //buff來自conn->curBuff，間接來自snd_buffer_pool ICBuffer *curBuff; //snd_buffer_pool在motionconn初始化的時候，分别擷取buffer，放在curBuff uint8 *pBuff; //pBuff初始化後指向其curBuff->pkt /* 依賴aSlice->primaryProcesses擷取程序proc結構進行初始化構造，程序id、IP、端口等資訊 */ struct icpkthdr conn_info; //全局&ic_control_info.connHtab struct CdbProcess *cdbProc;//來自aSlice->primaryProcesses uint8 **pkt_q; /*pkt_q是數組充當環形緩沖區，其中容量求模計算下标操作，Rx線程接收的資料包pkt放置在conn->pkt_q[pos] = (uint8 *) pkt中。而IcBuffer中的pkt指派給motioncon中的pBuff，而pBuff又會在調用prepareRxConnForRead時，被指派pkt_q對應指針指向的資料區,conn->pBuff = conn->pkt_q[conn->pkt_q_head];進而形成資料鍊路關系。 */ motion： ICBufferList sndQueue、 ICBuffer *curBuff、 ICBufferList unackQueue、 uint8 *pBuff、 uint8 **pkt_q; ICBuffer pkt static SendBufferPool snd_buffer_pool; 第一層：snd_buffer_pool在motionconn初始化的時候，分别擷取buffer，放在curBuff，并初始化pBuff。 第二層：了解sndQueue邏輯 中轉站，buff來自conn->curBuff，間接來自snd_buffer_pool 第三層：了解data buffer和 pkt_q 啟用資料緩沖區：pkt_q是數組充當環形緩沖區，其中容量求模計算下标操作，Rx線程接收的資料包pkt放置在conn->pkt_q[pos] = (uint8 *) pkt中。 而IcBuffer中的pkt指派給motioncon中的pBuff，而pBuff又會在調用prepareRxConnForRead時，被指派pkt_q對應指針指向的資料區， conn->pBuff = conn->pkt_q[conn->pkt_q_head];進而形成資料鍊路關系。 |
2. Ic_udp流程分析之緩沖區流程分析
| Go Ic_udp流程分析緩沖區初始化： SetupUDPIFCInterconnect_Internal調用initSndBufferPool(&snd_buffer_pool)進行初始化。 Ic_udp流程分析緩沖擷取： 調用接口getSndBuffer擷取緩沖區buffer，在初始化流程SetupUDPIFCInterconnect_Internal->startOutgoingUDPConnections,為每個con擷取一個buffer，并且填充MotionConn中的curBuff static ICBuffer * getSndBuffer(MotionConn *conn) Ic_udp流程分析緩沖釋放： 通過調用icBufferListReturn接口，釋放buffer進去snd_buffer_pool.freeList static void icBufferListReturn(ICBufferList *list, bool inExpirationQueue) { icBufferListAppend(&snd_buffer_pool.freeList, buf);# here 0 } 清理：cleanSndBufferPool(&snd_buffer_pool);上面釋放回去後接着清理buff。 handleAckedPacket邏輯對于unackQueue也會出發釋放。 |
Ic_Proxy流程分析
1. 簡要介紹
TCP流程buf設計較為簡單，在這裡不做詳細贅述。Proxy代理服務是基于TCP改造而來，主要用來應對在大規模叢集裡面網絡連接配接數巨大的情況。
Ic_Proxy隻需要一個網絡連接配接在每兩個網端之間，相比較于IC-Tcp 模式，它消耗的連接配接總量和端口更少。同時，與 IC-Udp模式相比，在高延遲網絡具有更好的表現。
TCP是一種點對點的有連接配接傳輸協定，一個有N個QE節點的Motion的連接配接數是N^2，一個有k個Motion的查詢将産生k*N^2個連接配接。
舉例來講，如果一個包含500個Segment的叢集，運作一個包含10個Motion的查詢，那麼這個查詢就需要建立10*500^2 = 2,500,000個TCP連接配接。即使不考慮最大連接配接數限制，建立如此多的TCP連接配接也是非常低效的。
Ic_Proxy是用LIBUV開發的，預設情況下禁用IC代理，我們可以使用./configure --enable-ic-proxy。
安裝完成後，我們還需要設定ic代理網絡，它完成了通過設定GUC。例如，如果叢集具有一個主節點、一個備用主節點、一個主分段和一個鏡像分段， 我們可以像下面這樣設定它：
| Go gp_Interconnect_proxy_addresses gpconfig --skipvalidation -c gp_Interconnect_proxy_addresses -v "'1:-1:localhost:2000,2:0:localhost:2002,3:0:localhost:2003,4:-1:localhost:2001'" |
它包含所有主伺服器、備用伺服器、以及主伺服器和鏡像伺服器的資訊段，文法如下：
dbid:segid:hostname:port[,dbid:segid:ip:port]。這裡要注意，将值指定為單引号字元串很重要，否則将被解析為格式無效的中間體。
2. Ic_proxy邏輯連接配接
| Go 在 Ic-Tcp 模式下，QE 之間存在 TCP 連接配接（包括 QD）,以一個收集動作舉例： ┌ ┐ │ │ <===== [ QE1 ] │ QD │ │ │ <===== [ QE2 ] └ ┘ 在 Ic-Udp 模式下，沒有 TCP 連接配接，但仍有邏輯連接配接：如果兩個QE互相通信，則存在邏輯連接配接： ┌ ┐ │ │ <----- [ QE1 ] │ QD │ │ │ <----- [ QE2 ] └ ┘ 在 Ic_Proxy 模式下，我們仍然使用邏輯連接配接的概念： ┌ ┐ ┌ ┐ │ │ │ │ <====> [ proxy ] <~~~~> [ QE1 ] │ QD │ <~~~~> │ proxy │ │ │ │ │ <====> [ proxy ] <~~~~> [ QE2 ] └ ┘ └ ┘ 在 N：1 集合運動中，有 N 個邏輯連接配接; 在N：N重新配置設定/廣播運動中存在邏輯連接配接數N*N |
3. Ic_Proxy邏輯連接配接辨別符
為了識别邏輯連接配接，我們需要知道誰是發送者，誰是接收者。在 Ic_Proxy 中，我們不區分邏輯的方向連接配接，我們使用名稱本地和遠端作為端點。終點至少由segindex和PID辨別，是以邏輯連接配接可以通過以下方式辨別：seg1，p1->seg2，p2
然而，這還不足以區分不同查詢中的子計劃。我們還必須将發送方和接收方切片索引放入考慮：slice[a->b] seg1，p1->seg2，p2
此外，考慮到後端程序可用于不同的查詢會話及其生命周期不是嚴格同步的，我們還必須将指令 ID 放入辨別符中：cmd1,slice[a->b] seg1，p1->seg2，p2
出于調試目的，我們還将會話ID放在辨別符中。在考慮鏡像或備用時，我們必須意識到與 SEG1 主節點的連接配接和與 SEG1 鏡像的連接配接不同，是以我們還需要将 dbid 放入辨別符中：cmd1,slice[a->b] seg1,dbid3,p1->seg2,dbid5,p2
Ic_Proxy資料轉發流程介紹
資料轉發是Ic_Proxy流程最複雜的部分，按照不同的流程，會産生三種轉發類型：
第一種是Loopback，即循環本地；
第二種是proxy client，通過代理去client；
第三種是proxy to proxy，從一個代理發到另一個代理。
然後，按照上述的三種類型再調用對應的route，把資料轉發出去，這樣就形成了一個完整的資料轉發流程。
圖3：Ic_proxy資料包轉發處理流程圖
圖4：Ic_Proxy流程時序圖
今天我們為大家帶來Greenplum-Interconnect子產品的解析，希望能夠幫助大家更好地了解子產品的技術特性和實作處理流程。