author：彭程

介绍

dble进行SQL解析之后，会进行判断对复杂SQL进一步处理，本文主要介绍复杂SQL的处理过程。
8b8d156d-c2f6-3ff1-bcad-8582399506d6.png
dble对复杂SQL的处理主要包括以下几步：

  1. 构造访问器（Visitor）；
  2. 对查询计划进行优化
  3. 执行多结果集

代码入口

在MySQLShardingSQLHandler#routeEndExecuteSQL方法中调用druid能得到RouteResultset(rrs)，之后会调用NonBlockingSession#execute进一步进行处理

  1.     public void routeEndExecuteSQL(String sql, int type, SchemaConfig schemaConfig) {
  2.         TraceManager.TraceObject traceObject = TraceManager.serviceTrace(service, "route&execute");
  3.         try {
  4.             if (service.getSession2().isKilled()) { //检查会话状态
  5.                 service.writeErrMessage(ErrorCode.ER_QUERY_INTERRUPTED, "The query is interrupted.");
  6.                 return;
  7.             }
  9.             RouteResultset rrs;
  10.             try {
  11.                 rrs = RouteService.getInstance().route(schemaConfig, type, sql, service); //这里调用了druid
  12.                 if (rrs == null) {
  13.                     return;
  14.                 }
  15.                 if (rrs.getSqlType() == ServerParse.DDL && rrs.getSchema() != null) {
  16.                     if (ProxyMeta.getInstance().getTmManager().getCatalogs().get(rrs.getSchema()).getView(rrs.getTable()) != null) {
  17.                         ProxyMeta.getInstance().getTmManager().removeMetaLock(rrs.getSchema(), rrs.getTable());
  18.                         String msg = "Table '" + rrs.getTable() + "' already exists as a view";
  19.                         LOGGER.info(msg);
  20.                         throw new SQLNonTransientException(msg);
  21.                     }
  22.                 }
  23.             } catch (Exception e) {
  24.                 service.executeException(e, sql);
  25.                 return;
  26.             }
  27.             service.getSession2().endRoute(rrs);
  28.             service.getSession2().execute(rrs); //对rrs进一步处理
  29.         } finally {
  30.             TraceManager.finishSpan(traceObject);
  31.         }
  32.     }

NonBlockingSession#execute方法会根据rrs#nodes的状态对SQL类型进行判断，如果nodes为null则是进行复杂SQL处理，调用NonBlockingSession#executeMultiSelect方法

  1.  public void execute(RouteResultset rrs) {
  2.         TraceManager.TraceObject traceObject = TraceManager.serviceTrace(shardingService, "execute-sql-for-sharding");
  3.         TraceManager.log(ImmutableMap.of("route-result-set", rrs), traceObject);
  4.         try {
  5.             if (killed) {
  6.                 shardingService.writeErrMessage(ErrorCode.ER_QUERY_INTERRUPTED, "The query is interrupted.");
  7.                 return;
  8.             }
  10.             if (LOGGER.isDebugEnabled()) {
  11.                 StringBuilder s = new StringBuilder();
  12.                 LOGGER.debug(s.append(shardingService).append(rrs).toString() + " rrs ");
  13.             }
  15.             if (PauseShardingNodeManager.getInstance().getIsPausing().get() &&
  16.                     !PauseShardingNodeManager.getInstance().checkTarget(target) &&
  17.                     PauseShardingNodeManager.getInstance().checkRRS(rrs)) {
  18.                 if (PauseShardingNodeManager.getInstance().waitForResume(rrs, shardingService, CONTINUE_TYPE_SINGLE)) {
  19.                     return;
  20.                 }
  21.             }
  23.             // complex query 复杂查询
  24.             RouteResultsetNode[] nodes = rrs.getNodes();
  25.             if (nodes == null || nodes.length == 0 || nodes[0].getName() == null || nodes[0].getName().equals("")) {
  26.                 if (rrs.isNeedOptimizer()) {
  27.                     try {
  28.                         this.complexRrs = rrs;
  29.                         executeMultiSelect(rrs); //复杂SQL处理
  30.                     } catch (MySQLOutPutException e) {
  31.                         LOGGER.warn("execute complex sql cause error", e);
  32.                         shardingService.writeErrMessage(e.getSqlState(), e.getMessage(), e.getErrorCode());
  33.                     }
  34.                 } else {
  35.                     shardingService.writeErrMessage(ErrorCode.ER_NO_DB_ERROR,
  36.                             "No shardingNode found ,please check tables defined in schema:" + shardingService.getSchema());
  37.                 }
  38.                 return;
  39.             }
  40.             setRouteResultToTrace(nodes);
  41.             if (rrs.getDdlHandler() != null) {
  42.                 executeDDL(rrs); //执行DDL语句
  43.             } else {
  44.                 // dml or simple select 简单查询
  45.                 executeOther(rrs);
  46.             }
  47.         } finally {
  48.             TraceManager.finishSpan(shardingService, traceObject);
  49.         }
  50.     }

NonBlockingSession#executeMultiSelect方法实现了复杂SQL处理的主要逻辑

  1.     public void executeMultiSelect(RouteResultset rrs) {
  2.         TraceManager.TraceObject traceObject = TraceManager.serviceTrace(shardingService, "try-complex-query");
  3.         try {
  4.             SQLSelectStatement ast = (SQLSelectStatement) rrs.getSqlStatement();
  5.             MySQLPlanNodeVisitor visitor = new MySQLPlanNodeVisitor(shardingService.getSchema(), shardingService.getCharset().getResultsIndex(), ProxyMeta.getInstance().getTmManager(), false, shardingService.getUsrVariables());
  6.             visitor.visit(ast); //构造访问器
  7.             PlanNode node = visitor.getTableNode();
  8.             if (node.isCorrelatedSubQuery()) {
  9.                 throw new MySQLOutPutException(ErrorCode.ER_UNKNOWN_ERROR, "", "Correlated Sub Queries is not supported ");
  10.             }
  11.             node.setSql(rrs.getStatement());
  12.             node.setUpFields();
  13.             PlanUtil.checkTablesPrivilege(shardingService, node, ast);
  14.             node = MyOptimizer.optimize(node); //对查询计划进行优化
  16.             if (PauseShardingNodeManager.getInstance().getIsPausing().get() &&
  17.                     !PauseShardingNodeManager.getInstance().checkTarget(target) &&
  18.                     PauseShardingNodeManager.getInstance().checkReferredTableNodes(node.getReferedTableNodes())) {
  19.                 if (PauseShardingNodeManager.getInstance().waitForResume(rrs, this.shardingService, CONTINUE_TYPE_MULTIPLE)) {
  20.                     return;
  21.                 }
  22.             }
  23.             setPreExecuteEnd(TraceResult.SqlTraceType.COMPLEX_QUERY);
  24.             if (PlanUtil.containsSubQuery(node)) {
  25.                 setSubQuery();
  26.                 final PlanNode finalNode = node;
  27.                 //sub Query build will be blocked, so use ComplexQueryExecutor
  28.                 DbleServer.getInstance().getComplexQueryExecutor().execute(() -> {
  29.                     executeMultiResultSet(rrs, finalNode);
  30.                 });
  31.             } else {
  32.                 if (!visitor.isContainSchema()) {
  33.                     node.setAst(ast);
  34.                 }
  35.                 executeMultiResultSet(rrs, node); //执行多结果集
  36.             }
  37.         } finally {
  38.             TraceManager.finishSpan(shardingService, traceObject);
  39.         }
  40.     }