[轉]數據庫中間件 MyCAT源碼分析——跨庫兩表Join

時間 2019-12-14

標籤數據庫中間件 mycat 源碼分析 join 欄目 SQL 简体版

原文原文鏈接

1. 概述java
2. 主流程sql
3. ShareJoinsession

3.1 JoinParser併發
3.2 ShareJoin.processSQL(...)app
3.3 BatchSQLJob異步
3.4 ShareDBJoinHandler函數
3.5 ShareRowOutPutDataHandler源碼分析

4. 彩蛋性能

1. 概述

MyCAT 支持跨庫表 Join，目前版本僅支持跨庫兩表 Join。雖然如此，已經可以知足咱們大部分的業務場景。何況，Join 過多的表可能帶來的性能問題也是很麻煩的。ui

本文主要分享：

總體流程、調用順序圖
核心代碼的分析

前置閱讀：《MyCAT 源碼分析 —— 【單庫單表】查詢》。

2. 主流程

當執行跨庫兩表 Join SQL 時，經歷的大致流程以下：

SQL 上，須要添加註解 /*!mycat:catlet=io.mycat.catlets.ShareJoin */${SQL} 。 RouteService#route(...) 解析註解 mycat:catlet 後，路由給 HintCatletHandler 做進一步處理。

HintCatletHandler 獲取註解對應的 Catlet 實現類， io.mycat.catlets.ShareJoin 就是其中一種實現（目前也只有這一種實現），提供了跨庫兩表 Join 的功能。從類命名上看， ShareJoin 很大可能性後續會提供完整的跨庫多表的 Join 功能。

核心代碼以下：

// HintCatletHandler.java
public RouteResultset route(SystemConfig sysConfig, SchemaConfig schema,
int sqlType, String realSQL, String charset, ServerConnection sc,
LayerCachePool cachePool, String hintSQLValue, int hintSqlType, Map hintMap)
throws SQLNonTransientException {
String cateletClass = hintSQLValue;
if (LOGGER.isDebugEnabled()) {
LOGGER.debug("load catelet class:" + hintSQLValue + " to run sql " + realSQL);
}
try {
Catlet catlet = (Catlet) MycatServer.getInstance().getCatletClassLoader().getInstanceofClass(cateletClass);
catlet.route(sysConfig, schema, sqlType, realSQL, charset, sc, cachePool);
catlet.processSQL(realSQL, new EngineCtx(sc.getSession2()));
} catch (Exception e) {
LOGGER.warn("catlet error " + e);
throw new SQLNonTransientException(e);
}
return null;
}

3. ShareJoin

目前支持跨庫兩表 Join。 ShareJoin 將 SQL 拆分紅左表 SQL 和右表 SQL，發送給各數據節點執行，彙總數據結果進行合後返回。

僞代碼以下：

// SELECT u.id, o.id FROM t_order o
// INNER JOIN t_user u ON o.uid = u.id
// 【順序】查詢左表
String leftSQL = "SELECT o.id, u.id FROM t_order o";
List leftList = dn[0].select(leftSQL) + dn[1].select(leftSQL) + ... + dn[n].select(leftsql);
// 【並行】查詢右表
String rightSQL = "SELECT u.id FROM t_user u WHERE u.id IN (${leftList.uid})";
for (dn : dns) { // 此處是並行執行，使用回調邏輯
for (rightRecord : dn.select(rightSQL)) { // 查詢右表
// 合併結果
for (leftRecord : leftList) {
if (leftRecord.uid == rightRecord.id) {
write(leftRecord + leftRecord.uid 拼接結果);
}
}
}
}

實際狀況會更加複雜，咱們接下來一點點往下看。

3.1 JoinParser

JoinParser 負責對 SQL 進行解析。總體流程以下：

舉個例子， /*!mycat:catlet=io.mycat.catlets.ShareJoin */SELECT o.id,u.usernamefromt_order o join t_user u on o.uid=u.id; 解析後， TableFilter 結果以下：

tName ：表名
tAlia ：表自定義命名
where ：過濾條件
order ：排序條件
parenTable ：左鏈接的 Join 的表名。 t_user表在 join屬性的 parenTable 爲 "o"，即 t_order。
joinParentkey ：左鏈接的 Join 字段
joinKey ：join 字段。 t_user表在 join屬性爲 id。
join ：子 tableFilter。即，該錶鏈接的右邊的表。
parent ：和 join屬性相對。

看到此處，你們可能有疑問，爲何要把 SQL 解析成 TableFilter。 JoinParser 根據 TableFilter 生成數據節點執行 SQL。代碼以下：

// TableFilter.java
public String getSQL() {
String sql = "";
// fields
for (Entry<String, String> entry : fieldAliasMap.entrySet()) {
String key = entry.getKey();
String val = entry.getValue();
if (val == null) {
sql = unionsql(sql, getFieldfrom(key), ",");
} else {
sql = unionsql(sql, getFieldfrom(key) + " as " + val, ",");
}
}
// where
if (parent == null) { // on/where 等於號左邊的表
String parentJoinKey = getJoinKey(true);
// fix sharejoin bug：
// (AbstractConnection.java:458) -close connection,reason:program err:java.lang.IndexOutOfBoundsException:
// 緣由是左表的select列沒有包含 join 列，在獲取結果時報上面的錯誤
if (sql != null && parentJoinKey != null &&
!sql.toUpperCase().contains(parentJoinKey.trim().toUpperCase())) {
sql += ", " + parentJoinKey;
}
sql = "select " + sql + " from " + tName;
if (!(where.trim().equals(""))) {
sql += " where " + where.trim();
}
} else { // on/where 等於號右邊邊的表
if (allField) {
sql = "select " + sql + " from " + tName;
} else {
sql = unionField("select " + joinKey, sql, ",");
sql = sql + " from " + tName;
//sql="select "+joinKey+","+sql+" from "+tName;
}
if (!(where.trim().equals(""))) {
sql += " where " + where.trim() + " and (" + joinKey + " in %s )";
} else {
sql += " where " + joinKey + " in %s ";
}
}
// order
if (!(order.trim().equals(""))) {
sql += " order by " + order.trim();
}
// limit
if (parent == null) {
if ((rowCount > 0) && (offset > 0)) {
sql += " limit" + offset + "," + rowCount;
} else {
if (rowCount > 0) {
sql += " limit " + rowCount;
}
}
}
return sql;
}

當 parent 爲空時，即on/where 等於號左邊的表。例如： selectid,uidfromt_order。
當 parent 不爲空時，即on/where 等於號右邊的表。例如： selectid,usernamefromt_userwhereidin(1,2,3)。

3.2 ShareJoin.processSQL(...)

當 SQL 解析完後，生成左邊的表執行的 SQL，發送給對應的數據節點查詢數據。大致流程以下：

當 SQL 爲 /*!mycat:catlet=io.mycat.catlets.ShareJoin */SELECT o.id,u.usernamefromt_order o join t_user u on o.uid=u.id; 時， sql=getSql() 的返回結果爲 selectid,uidfromt_order。

生成左邊的表執行的 SQL 後，順序順序順序發送給對應的數據節點查詢數據。具體順序查詢是怎麼實現的，咱們來看下章 BatchSQLJob。

3.3 BatchSQLJob

EngineCtx 對 BatchSQLJob 封裝，提供上層兩個方法：

executeNativeSQLSequnceJob ：順序（非併發）在每一個數據節點執行SQL任務
executeNativeSQLParallJob ：併發在每一個數據節點執行SQL任務

核心代碼以下：

// EngineCtx.java
public void executeNativeSQLSequnceJob(String[] dataNodes, String sql,
SQLJobHandler jobHandler) {
for (String dataNode : dataNodes) {
SQLJob job = new SQLJob(jobId.incrementAndGet(), sql, dataNode,
jobHandler, this);
bachJob.addJob(job, false);
}
}
public void executeNativeSQLParallJob(String[] dataNodes, String sql,
SQLJobHandler jobHandler) {
for (String dataNode : dataNodes) {
SQLJob job = new SQLJob(jobId.incrementAndGet(), sql, dataNode,
jobHandler, this);
bachJob.addJob(job, true);
}
}

BatchSQLJob 經過執行中任務列表、待執行任務列表來實現順序/併發執行任務。核心代碼以下：

// BatchSQLJob.java
/**
* 執行中任務列表
*/
private ConcurrentHashMap<Integer, SQLJob> runningJobs = new ConcurrentHashMap<Integer, SQLJob>();
/**
* 待執行任務列表
*/
private ConcurrentLinkedQueue<SQLJob> waitingJobs = new ConcurrentLinkedQueue<SQLJob>();
public void addJob(SQLJob newJob, boolean parallExecute) {
if (parallExecute) {
runJob(newJob);
} else {
waitingJobs.offer(newJob);
if (runningJobs.isEmpty()) { // 若無正在執行中的任務，則從等待隊列裏獲取任務進行執行。
SQLJob job = waitingJobs.poll();
if (job != null) {
runJob(job);
}
}
}
}
public boolean jobFinished(SQLJob sqlJob) {
runningJobs.remove(sqlJob.getId());
SQLJob job = waitingJobs.poll();
if (job != null) {
runJob(job);
return false;
} else {
if (noMoreJobInput) {
return runningJobs.isEmpty() && waitingJobs.isEmpty();
} else {
return false;
}
}
}

順序執行時，當 runningJobs 存在執行中的任務時， #addJob(...) 時，不當即執行，添加到 waitingJobs。當 SQLJob 完成時，順序調用下一個任務。
併發執行時， #addJob(...) 時，當即執行。

SQLJob SQL 異步執行任務。其 jobHandler(SQLJobHandler) 屬性，在 SQL 執行有返回結果時，會進行回調，從而實現異步執行。

在 ShareJoin 裏， SQLJobHandler 有兩個實現： ShareDBJoinHandler、 ShareRowOutPutDataHandler。前者，左邊的表執行的 SQL 回調；後者，右邊的表執行的 SQL 回調。

3.4 ShareDBJoinHandler

ShareDBJoinHandler，左邊的表執行的 SQL 回調。流程以下：

#fieldEofResponse(...) ：接收數據節點返回的 fields，放入內存。
#rowResponse(...) ：接收數據節點返回的 row，放入內存。
#rowEofResponse(...) ：接收完一個數據節點返回全部的 row。當全部數據節點都完成 SQL 執行時，提交右邊的表執行的 SQL 任務，並行執行，即圖中#createQryJob(...)。

當 SQL 爲 /*!mycat:catlet=io.mycat.catlets.ShareJoin */SELECT o.id,u.usernamefromt_order o join t_user u on o.uid=u.id; 時， sql=getChildSQL() 的返回結果爲selectid,usernamefromt_userwhereidin(1,2,3)。

核心代碼以下：

// ShareJoin.java
private void createQryJob(int batchSize) {
int count = 0;
Map<String, byte[]> batchRows = new ConcurrentHashMap<String, byte[]>();
String theId = null;
StringBuilder sb = new StringBuilder().append('(');
String svalue = "";
for (Map.Entry<String, String> e : ids.entrySet()) {
theId = e.getKey();
byte[] rowbyte = rows.remove(theId);
if (rowbyte != null) {
batchRows.put(theId, rowbyte);
}
if (!svalue.equals(e.getValue())) {
if (joinKeyType == Fields.FIELD_TYPE_VAR_STRING
|| joinKeyType == Fields.FIELD_TYPE_STRING) { // joinkey 爲varchar
sb.append("'").append(e.getValue()).append("'").append(','); // ('digdeep','yuanfang')
} else { // 默認joinkey爲int/long
sb.append(e.getValue()).append(','); // (1,2,3)
}
}
svalue = e.getValue();
if (count++ > batchSize) {
break;
}
}
if (count == 0) {
return;
}
jointTableIsData = true;
sb.deleteCharAt(sb.length() - 1).append(')');
String sql = String.format(joinParser.getChildSQL(), sb);
getRoute(sql);
ctx.executeNativeSQLParallJob(getDataNodes(), sql, new ShareRowOutPutDataHandler(this, fields, joinindex, joinParser.getJoinRkey(), batchRows, ctx.getSession()));
}

3.5 ShareRowOutPutDataHandler

ShareRowOutPutDataHandler，右邊的表執行的 SQL 回調。流程以下：

#fieldEofResponse(...) ：接收數據節點返回的 fields，返回 header 給 MySQL Client。
#rowResponse(...) ：接收數據節點返回的 row，匹配左表的記錄，返回合併後返回的 row 給 MySQL Client。
#rowEofResponse(...) ：當全部 row 都返回完後，返回 eof 給 MySQL Client。

核心代碼以下：

// ShareRowOutPutDataHandler.java
public boolean onRowData(String dataNode, byte[] rowData) {
RowDataPacket rowDataPkgold = ResultSetUtil.parseRowData(rowData, bfields);
//拷貝一份batchRows
Map<String, byte[]> batchRowsCopy = new ConcurrentHashMap<String, byte[]>();
batchRowsCopy.putAll(arows);
// 獲取Id字段，
String id = ByteUtil.getString(rowDataPkgold.fieldValues.get(joinR));
// 查找ID對應的A表的記錄
byte[] arow = getRow(batchRowsCopy, id, joinL);
while (arow != null) {
RowDataPacket rowDataPkg = ResultSetUtil.parseRowData(arow, afields);//ctx.getAllFields());
for (int i = 1; i < rowDataPkgold.fieldCount; i++) {
// 設置b.name 字段
byte[] bname = rowDataPkgold.fieldValues.get(i);
rowDataPkg.add(bname);
rowDataPkg.addFieldCount(1);
}
// huangyiming add
MiddlerResultHandler middlerResultHandler = session.getMiddlerResultHandler();
if (null == middlerResultHandler) {
ctx.writeRow(rowDataPkg);
} else {
if (middlerResultHandler instanceof MiddlerQueryResultHandler) {
byte[] columnData = rowDataPkg.fieldValues.get(0);
if (columnData != null && columnData.length > 0) {
String rowValue = new String(columnData);
middlerResultHandler.add(rowValue);
}
//}
}
}
arow = getRow(batchRowsCopy, id, joinL);
}
return false;
}