Sharding-JDBC 源码之 SQL 改写

以你之姓@ 2022-12-31 01:21 78阅读 0赞

### Sharding-JDBC 系列 ###

1.  第一篇 [Sharding-JDBC 源码之启动流程分析][Sharding-JDBC]
2.  第二篇 [Sharding-JDBC 源码之 SQL 解析][Sharding-JDBC _ SQL]
3.  第三篇 [Sharding-JDBC 源码之 SQL 路由][Sharding-JDBC _ SQL 1]
4.  第四篇 [Sharding-JDBC 源码之 SQL 改写][Sharding-JDBC _ SQL 2]（本文）
5.  第五篇 [Sharding-JDBC 源码之 SQL 执行][Sharding-JDBC _ SQL 3]
6.  第六篇 [Sharding-JDBC 源码之结果集归并][Sharding-JDBC 1]

--------------------

> 在完成 SQL 路由后，我们就会拿到实际要查询的数据源名称及实际表节点。这时候就需要对原逻辑 SQL 进行改写，将逻辑表名替换为真实表名，这一步 Sharding-JDBC 是由改写引擎 `SQLRewriteEngine` 来完成的。

### 目录 ###

*   *   *  Sharding-JDBC 系列
         *  改写类型
         *   *  正确性改写
             *   *  标识符改写
                 *  补列
             *  优化改写
         *  源码分析
         *   *  改写引擎获取 SQLBuilder 对象
             *  执行改写引擎中的 generateSQL ，替换真实表名
         *  总结

在看具体源码之前，先来了解下 SQL 改写有哪些类型

### 改写类型 ###

#### 正确性改写 ####

##### 标识符改写 #####

需要改写的标识符包括表名称、索引名称以及Schema名称。如逻辑 SQL 为：

select goods_name from t_goods where goods_id = 1;

当 goods\_id = 1时，将会路由到 分片表 1，那么改写后的 SQL 为：

select goods_name from t_goods_1 where goods_id = 1;

##### 补列 #####

补列通常由三种情况导致：

1.  需要在结果归并时获取数据，但该列数据并未从查询的 SQL 中返回，这种场景主要针对**GROUP BY**和**ORDER BY**。结果归并时需要根据 **GROUP BY**和**ORDER BY**的字段进行分组和排序，但如果原始SQL的选择项中若并未包含分组项或排序项，则需要对原始SQL进行改写。如逻辑 SQL 为：

select goods_name from t_goods order by goods_id;

由于原 SQL 中不包含归并结果时需要的 goods\_id 列，因此需要改写补列，改写后的 SQL 为：

select goods_name, goods_id AS ORDER_BY_DERIVED_0 from t_goods order by goods_id;

补列只会补充缺失的列，不会全部补充。

1.  在使用 **AVG** 聚集函数时，在分布式的场景中，使用avg1 + avg2 + avg3 / 3计算平均值并不正确，需要改写为 (sum1 + sum2 + sum3) / (count1 + count2 + count3)。 这就需要将包含AVG的SQL改写为**SUM**和**COUNT**，并在结果归并时重新计算平均值。如以下SQL：

SELECT AVG(price) FROM t_order WHERE user_id=1;

需要改写为：

SELECT COUNT(price) AS AVG_DERIVED_COUNT_0, SUM(price) AS AVG_DERIVED_SUM_0 FROM t_order WHERE user_id=1;

然后才能够通过结果归并正确的计算平均值。

1.  在执行**INSERT**的SQL语句时，如果使用了Sharding-JDBC分布式自增主键的生成策略，则需要通过补列，让使用方无需改动现有代码，即可将分布式自增主键透明的替换数据库现有的自增主键，如原逻辑 SQL 为：

INSERT INTO t_goods (`goods_name`) VALUES ('番茄鸡蛋盖浇饭');

配置自增主键后，SQL 改写为：

INSERT INTO t_goods (`goods_name`, goods_id) VALUES ('番茄鸡蛋盖浇饭', xxxxx);

改写后的SQL将在INSERT FIELD和INSERT VALUE的最后部分增加主键列名称以及自动生成的自增主键值。上述SQL中的xxxxx表示自动生成的自增主键值。

*  分页修正(略)
 *  批量拆分(略)

#### 优化改写 ####

*  单节点优化(略)
 *  流式归并优化(略)

由于篇幅原因，其他改写类型本文暂不再介绍，官网介绍的很详细，请移步官网查看：[**内部剖析-改写引擎**][-]  
难得的中文官网，大家不要错过。配张官网改写引擎结构图：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3NtaWxlX2Zyb21fMjAxNQ_size_16_color_FFFFFF_t_70_pic_center]

### 源码分析 ###

从源码上，改写的代码主要在 rewrite 包中：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3NtaWxlX2Zyb21fMjAxNQ_size_16_color_FFFFFF_t_70_pic_center 1]

上一章中，拿到 `routingResult` 后,会初始化 SQL 改写引擎 `SQLRewriteEngine`，

public SQLRouteResult route(final String logicSQL, final List<Object> parameters, final SQLStatement sqlStatement) { 
            // ...... 省略
            // 获取路由结果
            RoutingResult routingResult = route(parameters, sqlStatement, shardingConditions);
            // 1. 初始化改写引擎
            SQLRewriteEngine rewriteEngine = new SQLRewriteEngine(shardingRule, logicSQL, databaseType, sqlStatement, shardingConditions, parameters);
            // 2. 根据 tableUnits.size 判断是否单路由
            boolean isSingleRouting = routingResult.isSingleRouting();
            if (sqlStatement instanceof SelectStatement && null != ((SelectStatement) sqlStatement).getLimit()) { 
                processLimit(parameters, (SelectStatement) sqlStatement, isSingleRouting);
            }
            
            // 3. 执行改写，生成 SQLBuilder 对象
            SQLBuilder sqlBuilder = rewriteEngine.rewrite(!isSingleRouting);
            for (TableUnit each : routingResult.getTableUnits().getTableUnits()) { 
                // 4. 执行改写引擎中的 generateSQL 方法，替换逻辑表名等变量 token
                result.getExecutionUnits().add(new SQLExecutionUnit(each.getDataSourceName(), rewriteEngine.generateSQL(each, sqlBuilder)));
            }
            if (showSQL) { 
                SQLLogger.logSQL(logicSQL, sqlStatement, result.getExecutionUnits());
            }
            return result;
        }

1.  获取路由结果后先初始化 SQL 改写引擎；
2.  根据路由结果中的 `tableUnits` 列表的长度判断是否是单路由，本例中只有一张表，所以是单路由；
3.  执行改写引擎的 `rewrite` 方法，生成 `SQLBuilder` 对象；
4.  执行改写引擎中的 `generateSQL` 方法，替换逻辑表名等变量 **token**为实际的节点名称。

#### 改写引擎获取 SQLBuilder 对象 ####

先来看下 `rewrite` 是如何生成 `SQLBuilder` 对象的

public SQLBuilder rewrite(final boolean isRewriteLimit) { 
            SQLBuilder result = new SQLBuilder(parameters);
            // 1. SQL 解析获取到的 token 列表
            if (sqlTokens.isEmpty()) { 
                result.appendLiterals(originalSQL);
                return result;
            }
            int count = 0;
            sortByBeginPosition();
            // 判断 token 类型
            for (SQLToken each : sqlTokens) { 
                // 2. 初次循环将常量添加到 SQLBuilder 对象中
                if (0 == count) { 
                    result.appendLiterals(originalSQL.substring(0, each.getBeginPosition()));
                }
                // 3. 表类型 token
                if (each instanceof TableToken) { 
                    appendTablePlaceholder(result, (TableToken) each, count, sqlTokens);
                } else if (each instanceof SchemaToken) { 
                    appendSchemaPlaceholder(result, (SchemaToken) each, count, sqlTokens);
                } else if (each instanceof IndexToken) { 
                    appendIndexPlaceholder(result, (IndexToken) each, count, sqlTokens);
                } else if (each instanceof ItemsToken) { 
                    appendItemsToken(result, (ItemsToken) each, count, sqlTokens);
                } else if (each instanceof InsertValuesToken) { 
                    appendInsertValuesToken(result, (InsertValuesToken) each, count, sqlTokens);
                } else if (each instanceof RowCountToken) { 
                    appendLimitRowCount(result, (RowCountToken) each, count, sqlTokens, isRewriteLimit);
                } else if (each instanceof OffsetToken) { 
                    appendLimitOffsetToken(result, (OffsetToken) each, count, sqlTokens, isRewriteLimit);
                } else if (each instanceof OrderByToken) { 
                    appendOrderByToken(result, count, sqlTokens);
                } else if (each instanceof InsertColumnToken) { 
                    appendSymbolToken(result, (InsertColumnToken) each, count, sqlTokens);
                }
                count++;
            }
            return result;
        }

1.  首先判断在 SQL 解析阶段获取到的变量 **token** 列表是否为空，为空则直接返回原逻辑 SQL,不继续进行改写操作；
2.  **token** 列表不为空，则循环遍历 **token** 类型。初次循环，将常量部分添加到 `result` 对象中，这里 `substring` 方法中的 `beginPosition` 是 SQL 解析阶段获得的逻辑表起始索引， 也就是将逻辑 SQL 从开始的 select 到 t\_goods （不包含 t\_goods）前的部分添加到 result 中；
3.  本例中的 **token** 类型是表类型，因此会执行  
    `appendTablePlaceholder` 方法，添加表占位符。

private void appendTablePlaceholder(final SQLBuilder sqlBuilder, final TableToken tableToken, final int count, final List<SQLToken> sqlTokens) { 
            // 1. 将逻辑表名存入 sqlBuilder 对象中
            sqlBuilder.appendPlaceholder(new TablePlaceholder(tableToken.getTableName().toLowerCase()));
            // 2. 计算逻辑表后的起始索引
            int beginPosition = tableToken.getBeginPosition() + tableToken.getSkippedSchemaNameLength() + tableToken.getOriginalLiterals().length();
            appendRest(sqlBuilder, count, sqlTokens, beginPosition);
        }
        
        private void appendRest(final SQLBuilder sqlBuilder, final int count, final List<SQLToken> sqlTokens, final int beginPosition) { 
            // 3. 获取逻辑 SQL 最后一位字符的索引
            int endPosition = sqlTokens.size() - 1 == count ? originalSQL.length() : sqlTokens.get(count + 1).getBeginPosition();
            // 4. 截取逻辑表后的条件
            sqlBuilder.appendLiterals(originalSQL.substring(beginPosition, endPosition));
        }

1.  将逻辑表存入 `sqlBuilder` 对象；
2.  计算逻辑表后的起始索引，`tableToken.getOriginalLiterals()` 也就是逻辑表的长度，计算后的 `beginPosition` 索引从**空字符**开始；
3.  获取逻辑SQL最后一位字符的索引；
4.  截取逻辑表的条件，如逻辑SQL是：

select goods_name from t_goods where goods_id=?

截取后的条件是：  
"  where goods\_id=?"。（注意： **where 前面还有一个空字符**）

到这里，`SQLBuilder` 对象就已经生成了，看下该对象的具体结构：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3NtaWxlX2Zyb21fMjAxNQ_size_16_color_FFFFFF_t_70_pic_center 2]

#### 执行改写引擎中的 generateSQL ，替换真实表名 ####

public SQLRouteResult route(final String logicSQL, final List<Object> parameters, final SQLStatement sqlStatement) { 
            // ......省略
            
            // 获取 sqlBuilder 对象
            SQLBuilder sqlBuilder = rewriteEngine.rewrite(!isSingleRouting);
            for (TableUnit each : routingResult.getTableUnits().getTableUnits()) { 
                result.getExecutionUnits().add(new SQLExecutionUnit(each.getDataSourceName(), rewriteEngine.generateSQL(each, sqlBuilder)));
            }
            if (showSQL) { 
                SQLLogger.logSQL(logicSQL, sqlStatement, result.getExecutionUnits());
            }
            return result;
        }
        
        public SQLUnit generateSQL(final TableUnit tableUnit, final SQLBuilder sqlBuilder) { 
            // 实际调用的是 sqlBuilder 中的 toSQL 方法
            // 1. getTableTokens 获取逻辑表与真实表之前的映射
            return sqlBuilder.toSQL(tableUnit, getTableTokens(tableUnit), shardingRule);
        }
        
        public SQLUnit toSQL(final TableUnit tableUnit, final Map<String, String> logicAndActualTableMap, final ShardingRule shardingRule) { 
            List<Object> insertParameters = new LinkedList<>();
            StringBuilder result = new StringBuilder();
            for (Object each : segments) { 
                // 2. 非 ShardingPlaceholder 类型，即为常量字符串，直接添加
                if (!(each instanceof ShardingPlaceholder)) { 
                    result.append(each);
                    continue;
                }
                String logicTableName = ((ShardingPlaceholder) each).getLogicTableName();
                // 3. 从 logicAndActualTableMap 中获取真实表名
                String actualTableName = logicAndActualTableMap.get(logicTableName);
                if (each instanceof TablePlaceholder) { 
                    result.append(null == actualTableName ? logicTableName : actualTableName);
                } else if (each instanceof SchemaPlaceholder) { 
                    SchemaPlaceholder schemaPlaceholder = (SchemaPlaceholder) each;
                    Optional<TableRule> tableRule = shardingRule.tryFindTableRuleByActualTable(actualTableName);
                    if (!tableRule.isPresent() && Strings.isNullOrEmpty(shardingRule.getShardingDataSourceNames().getDefaultDataSourceName())) { 
                        throw new ShardingException("Cannot found schema name '%s' in sharding rule.", schemaPlaceholder.getLogicSchemaName());
                    }
                    Preconditions.checkState(tableRule.isPresent());
                    result.append(tableRule.get().getActualDatasourceNames().iterator().next());
                } else if (each instanceof IndexPlaceholder) { 
                    IndexPlaceholder indexPlaceholder = (IndexPlaceholder) each;
                    result.append(indexPlaceholder.getLogicIndexName());
                    if (!Strings.isNullOrEmpty(actualTableName)) { 
                        result.append("_");
                        result.append(actualTableName);
                    }
                } else if (each instanceof InsertValuesPlaceholder) { 
                    InsertValuesPlaceholder insertValuesPlaceholder = (InsertValuesPlaceholder) each;
                    List<String> expressions = new LinkedList<>();
                    for (ShardingCondition shardingCondition : insertValuesPlaceholder.getShardingConditions().getShardingConditions()) { 
                        processInsertShardingCondition(tableUnit, (InsertShardingCondition) shardingCondition, expressions, insertParameters);
                    }
                    int count = 0;
                    for (String s : expressions) { 
                        if (0 != count) { 
                            result.append(", ");
                        }
                        result.append(s);
                        count++;
                    }
                } else { 
                    result.append(each);
                }
            }
            // 4. 将替换结果封装为 SQLUnit 对象返回
            if (insertParameters.isEmpty()) { 
                return new SQLUnit(result.toString(), new ArrayList<>(Collections.singleton(parameters)));
            } else { 
                return new SQLUnit(result.toString(), new ArrayList<>(Collections.singleton(insertParameters)));
            }
        }

1.  `getTableTokens` 方法获取逻辑表与真实表之前的映射，key 为**逻辑表名**，value 为**真实表名**；
2.  非 `ShardingPlaceholder` 类型，即为常量字符串，直接添加到 `result` 中，`ShardingPlaceholder` 是接口类型，是 `TablePlaceholder`、`SchemaPlaceholder` 等占位符类型的父类型；
3.  从 `logicAndActualTableMap` 中获取真实表名，并添加到 `result` 中，**到这一步便已完成逻辑表名的替换**；
4.  将替换结果封装为 `SQLUnit` 对象返回，`SQLUnit` 对象又封装在 `SQLExecutionUnit` 对象中，并添加在 `SQLRouteResult` 对象的 `executionUnits` 列表中。

这时候，我们再来看下 SQLRouteResult 的结构：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3NtaWxlX2Zyb21fMjAxNQ_size_16_color_FFFFFF_t_70_pic_center 3]

### 总结 ###

1.  首先介绍了 SQL 改写的常见类型，并做了简单介绍，详细介绍可见官网：[***内部剖析-改写引擎***][-]；
2.  从源码看，SQL 改写主要是在 `SqlBuilder` 中将 SQL 进行截取并设置不同的类型，如**常量字符串类型**、**表占位符类型**等；
3.  根据不同的类型拼接实际可执行的字符串 SQL，遇到 `ShardingPlaceholder` 类型的占位符类型时，则进行相应的替换，如将逻辑表替换为真实表；
4.  将改写后的 SQL 封装到 `SQLExecutionUnit` 对象中，并添加到 `SQLRouteResult` 的 `executionUnits` 列表中，为接下来的执行做准备；
5.  下一篇章，将会学习获取到路由结果后，SQL 是如何执行的。

[Sharding-JDBC]: https://blog.csdn.net/smile_from_2015/article/details/111463029
[Sharding-JDBC _ SQL]: https://blog.csdn.net/smile_from_2015/article/details/111657901
[Sharding-JDBC _ SQL 1]: https://blog.csdn.net/smile_from_2015/article/details/111770894
[Sharding-JDBC _ SQL 2]: https://blog.csdn.net/smile_from_2015/article/details/111800577
[Sharding-JDBC _ SQL 3]: https://blog.csdn.net/smile_from_2015/article/details/111828140
[Sharding-JDBC 1]: https://blog.csdn.net/smile_from_2015/article/details/112101676
[-]: https://shardingsphere.apache.org/document/legacy/4.x/document/cn/features/sharding/principle/rewrite/
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3NtaWxlX2Zyb21fMjAxNQ_size_16_color_FFFFFF_t_70_pic_center]: /images/20221120/782608b18484495fa8dbbfdad7a3ea1e.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3NtaWxlX2Zyb21fMjAxNQ_size_16_color_FFFFFF_t_70_pic_center 1]: https://img-blog.csdnimg.cn/20201227144148912.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3NtaWxlX2Zyb21fMjAxNQ==,size_16,color_FFFFFF,t_70#pic_center
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3NtaWxlX2Zyb21fMjAxNQ_size_16_color_FFFFFF_t_70_pic_center 2]: https://img-blog.csdnimg.cn/2020122714474525.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3NtaWxlX2Zyb21fMjAxNQ==,size_16,color_FFFFFF,t_70#pic_center
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3NtaWxlX2Zyb21fMjAxNQ_size_16_color_FFFFFF_t_70_pic_center 3]: https://img-blog.csdnimg.cn/20201227144954608.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3NtaWxlX2Zyb21fMjAxNQ==,size_16,color_FFFFFF,t_70#pic_center