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

時間  2019-11-16
標籤 數據庫 中間件 mycat 源碼 分析 join 欄目 SQL 简体版
原文   原文鏈接

摘要: 原創出處 www.iocoder.cn/MyCAT/two-t… 「芋道源碼」歡迎轉載，保留摘要，謝謝！java

本文主要基於 MyCAT 1.6.5 正式版 sql

🙂🙂🙂關注微信公衆號：【芋道源碼】有福利： 微信

  1. RocketMQ / MyCAT / Sharding-JDBC 全部源碼分析文章列表
  2. RocketMQ / MyCAT / Sharding-JDBC 中文註釋源碼 GitHub 地址
  3. 您對於源碼的疑問每條留言都將獲得認真回覆。甚至不知道如何讀源碼也能夠請教噢。
  4. 新的源碼解析文章實時收到通知。每週更新一篇左右。
  5. 認真的源碼交流微信羣。

1. 概述

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

本文主要分享：併發

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

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

OK，Let's Go。異步

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.username from t_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 等於號左邊的表。例如：select id, uid from t_order。
  • 當 parent 不爲空時，即on/where 等於號右邊的表。例如：select id, username from t_user where id in (1, 2, 3)。

3.2 ShareJoin.processSQL(...)

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

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

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

3.3 BatchSQLJob

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

  1. executeNativeSQLSequnceJob ：順序（非併發）在每一個數據節點執行SQL任務
  2. 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.username from t_order o join t_user u on o.uid = u.id; 時，
sql = getChildSQL() 的返回結果爲 select id, username from t_user where id in (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;
}複製代碼

4. 彩蛋

以下是本文涉及到的核心類，有興趣的同窗能夠翻一翻。

ShareJoin 另外不支持的功能：

  1. 只支持 inner join，不支持 left join、right join 等等鏈接。
  2. 不支持 order by。
  3. 不支持 group by 以及 相關聚合函數。
  4. 即便 join 左表的字段未聲明爲返回 fields 也會返回。

恩，MyCAT 弱XA 源碼繼續走起！

相關文章
相關標籤/搜索
每日一句
    每一个你不满意的现在，都有一个你没有努力的曾经。
本站公眾號
   歡迎關注本站公眾號,獲取更多信息
聯繫我們 最近搜索 最新文章 沪ICP备13005482号-10 MyBatis教程 SQL 教程 MySQL教程 Java 教程 Thymeleaf 教程 Hibernate教程 Spring教程 Redis教程