改写引擎的职责定位是进行SQL的修改，因为ShardingSphere的核心目标就是屏蔽分库分表对用户的影响（当然后来还增加影子表、加解密等功能），使开发者可以按照像原来传统单库单表一样编写SQL。表拆分后，表名往往会带有编号或者日期等标识，但应用中的SQL中表名并不会带有这些标识，一般称之为逻辑表（和未拆分前表名完全相同），因此改写引擎需要用路由引擎计算得到的真正物理表名替换SQL中的逻辑表名，这样SQL才能正确执行。
除了sharding功能中表名替换，目前在ShardingSphere中需要很多种情况会进行SQL改写，具体有：
- 数据分片功能中表名改写；
- 数据分片功能中聚合函数distinct；
- 数据分片功能中avg聚合函数需添加count、sum；
- 数据分片功能中索引重命名；
- 数据分片功能中分页时offset、rowcount改写；
- 配置分布式自增键时自增列、值添加；
- 加解密功能下对列、值得添加修改；
- 影子表功能下对列与值的修改。
代码调用分析
回到BasePrepareEngine类中，可以看到在使用改写功能前注册了改写装饰器。
org.apache.shardingsphere.underlying.pluggble.prepare.BasePrepareEngine
private Collection<ExecutionUnit> executeRewrite(final String sql, final List<Object> parameters, final RouteContext routeContext) {
registerRewriteDecorator();
//创建SQL改写上下文，主要是生成对应的Token以及参数
SQLRewriteContext sqlRewriteContext = rewriter.createSQLRewriteContext(sql, parameters, routeContext.getSqlStatementContext(), routeContext);
return routeContext.getRouteResult().getRouteUnits().isEmpty() ? rewrite(sqlRewriteContext) : rewrite(routeContext, sqlRewriteContext);
}
private void registerRewriteDecorator() {
for (Class<? extends SQLRewriteContextDecorator> each : OrderedRegistry.getRegisteredClasses(SQLRewriteContextDecorator.class)) {
SQLRewriteContextDecorator rewriteContextDecorator = createRewriteDecorator(each);
Class<?> ruleClass = (Class<?>) rewriteContextDecorator.getType();
// FIXME rule.getClass().getSuperclass() == ruleClass for orchestration, should decouple extend between orchestration rule and sharding rule
rules.stream().filter(rule -> rule.getClass() == ruleClass || rule.getClass().getSuperclass() == ruleClass).collect(Collectors.toList())
.forEach(rule -> rewriter.registerDecorator(rule, rewriteContextDecorator));
}
}
之后便通过SQL改写入口类SQLRewriteEntry创建SQL改写上下文对象
/**
* SQL rewrite entry.
*/
@RequiredArgsConstructor
public final class SQLRewriteEntry {
private final SchemaMetaData schemaMetaData;
private final ConfigurationProperties properties;
private final Map<BaseRule, SQLRewriteContextDecorator> decorators = new LinkedHashMap<>();
/**
* Register route decorator.
*
* @param rule rule
* @param decorator SQL rewrite context decorator
*/
public void registerDecorator(final BaseRule rule, final SQLRewriteContextDecorator decorator) {
decorators.put(rule, decorator);
}
/**
* Create SQL rewrite context.
*
* @param sql SQL
* @param parameters parameters
* @param sqlStatementContext SQL statement context
* @param routeContext route context
* @return SQL rewrite context
*/
public SQLRewriteContext createSQLRewriteContext(final String sql, final List<Object> parameters, final SQLStatementContext sqlStatementContext, final RouteContext routeContext) {
SQLRewriteContext result = new SQLRewriteContext(schemaMetaData, sqlStatementContext, sql, parameters);// 创建一个初始SQL改写上下文
decorate(decorators, result, routeContext);// 进行装饰器处理，其实就是根据Statement上下文，生成一系列的Token生成器
result.generateSQLTokens();// 运行各Token生成器，解构出对应的Token
return result;
}
@SuppressWarnings("unchecked")
private void decorate(final Map<BaseRule, SQLRewriteContextDecorator> decorators, final SQLRewriteContext sqlRewriteContext, final RouteContext routeContext) {
for (Entry<BaseRule, SQLRewriteContextDecorator> entry : decorators.entrySet()) {
BaseRule rule = entry.getKey();
SQLRewriteContextDecorator decorator = entry.getValue();
if (decorator instanceof RouteContextAware) {
((RouteContextAware) decorator).setRouteContext(routeContext);
}
decorator.decorate(rule, properties, sqlRewriteContext);
}
}
}
以最用到的sharding功能实现的SQL改写上下文装饰器，看下其实现逻辑：
org.apache.shardingsphere.sharding.rewrite.context.ShardingSQLRewriteContextDecorator
/**
* SQL rewrite context decorator for sharding.
*/
@Setter
public final class ShardingSQLRewriteContextDecorator implements SQLRewriteContextDecorator<ShardingRule>, RouteContextAware {
private RouteContext routeContext;
@SuppressWarnings("unchecked")
@Override
public void decorate(final ShardingRule shardingRule, final ConfigurationProperties properties, final SQLRewriteContext sqlRewriteContext) {
// 获取参数改写器（参数化SQL才需要），然后依次对SQL改写上下文中的参数构造器parameterBuilder进行改写操作，分片功能下主要是自增键以及分页参数
for (ParameterRewriter each : new ShardingParameterRewriterBuilder(shardingRule, routeContext).getParameterRewriters(sqlRewriteContext.getSchemaMetaData())) {
if (!sqlRewriteContext.getParameters().isEmpty() && each.isNeedRewrite(sqlRewriteContext.getSqlStatementContext())) {
each.rewrite(sqlRewriteContext.getParameterBuilder(), sqlRewriteContext.getSqlStatementContext(), sqlRewriteContext.getParameters());
}
}
//添加分片功能下对应的Token生成器
sqlRewriteContext.addSQLTokenGenerators(new ShardingTokenGenerateBuilder(shardingRule, routeContext).getSQLTokenGenerators());
}
…
}
可以看到首先会通过ShardingParameterRewriterBuilder创建了数据分片功能对应的参数改写器，包括了insert自增分布式主键参数和分页参数两个重写器。
org.apache.shardingsphere.sharding.rewrite.parameter.ShardingParameterRewriterBuilder
/**
* Parameter rewriter builder for sharding.
*/
@RequiredArgsConstructor
public final class ShardingParameterRewriterBuilder implements ParameterRewriterBuilder {
private final ShardingRule shardingRule;
private final RouteContext routeContext;
@Override
public Collection<ParameterRewriter> getParameterRewriters(final SchemaMetaData schemaMetaData) {
Collection<ParameterRewriter> result = getParameterRewriters();
for (ParameterRewriter each : result) {
setUpParameterRewriters(each, schemaMetaData);
}
return result;
}
private static Collection<ParameterRewriter> getParameterRewriters() {
Collection<ParameterRewriter> result = new LinkedList<>();
result.add(new ShardingGeneratedKeyInsertValueParameterRewriter());
result.add(new ShardingPaginationParameterRewriter());
return result;
}
…
}
然后通过ShardingTokenGenerateBuilder生成数据分片Token生成器
org.apache.shardingsphere.sharding.rewrite.token.pojo.ShardingTokenGenerateBuilder
/**
* SQL token generator builder for sharding.
*/
@RequiredArgsConstructor
public final class ShardingTokenGenerateBuilder implements SQLTokenGeneratorBuilder {
private final ShardingRule shardingRule;
private final RouteContext routeContext;
@Override
public Collection<SQLTokenGenerator> getSQLTokenGenerators() {
Collection<SQLTokenGenerator> result = buildSQLTokenGenerators();
for (SQLTokenGenerator each : result) {
if (each instanceof ShardingRuleAware) {
((ShardingRuleAware) each).setShardingRule(shardingRule);
}
if (each instanceof RouteContextAware) {
((RouteContextAware) each).setRouteContext(routeContext);
}
}
return result;
}
private Collection<SQLTokenGenerator> buildSQLTokenGenerators() {
Collection<SQLTokenGenerator> result = new LinkedList<>();
addSQLTokenGenerator(result, new TableTokenGenerator());// 表名token处理，用于真实表名替换
addSQLTokenGenerator(result, new DistinctProjectionPrefixTokenGenerator());// select distinct关键字处理
addSQLTokenGenerator(result, new ProjectionsTokenGenerator());// select列名处理，主要是衍生列avg处理
addSQLTokenGenerator(result, new OrderByTokenGenerator());// Order by Token处理
addSQLTokenGenerator(result, new AggregationDistinctTokenGenerator());// 聚合函数的distinct关键字处理
addSQLTokenGenerator(result, new IndexTokenGenerator());// 索引重命名
addSQLTokenGenerator(result, new OffsetTokenGenerator());// offset 改写
addSQLTokenGenerator(result, new RowCountTokenGenerator());// rowCount改写
addSQLTokenGenerator(result, new GeneratedKeyInsertColumnTokenGenerator());// 分布式主键列添加，在insert sql列最后添加
addSQLTokenGenerator(result, new GeneratedKeyForUseDefaultInsertColumnsTokenGenerator());// insert SQL使用默认列名时需要完成补齐真实列名，包括自增列
addSQLTokenGenerator(result, new GeneratedKeyAssignmentTokenGenerator());// SET自增键生成
addSQLTokenGenerator(result, new ShardingInsertValuesTokenGenerator());// insert SQL 的values Token解析，为后续添加自增值做准备
addSQLTokenGenerator(result, new GeneratedKeyInsertValuesTokenGenerator());//为insert values添加自增列值
return result;
}
private void addSQLTokenGenerator(final Collection<SQLTokenGenerator> sqlTokenGenerators, final SQLTokenGenerator toBeAddedSQLTokenGenerator) {
if (toBeAddedSQLTokenGenerator instanceof IgnoreForSingleRoute && routeContext.getRouteResult().isSingleRouting()) {
return;
}
sqlTokenGenerators.add(toBeAddedSQLTokenGenerator);
}
}
可以看到ShardingTokenGenerateBuilder类针对数据分片需要改写SQL的各种情况分别添加了对应的Token生成器，看下最主要的表名Token生成器类TableTokenGenerator
org.apache.shardingsphere.sharding.rewrite.token.generator.impl.TableTokenGenerator
/**
* Table token generator.
*/
@Setter
public final class TableTokenGenerator implements CollectionSQLTokenGenerator, ShardingRuleAware {
private ShardingRule shardingRule;
@Override
public boolean isGenerateSQLToken(final SQLStatementContext sqlStatementContext) {
return true;
}
@Override
public Collection<TableToken> generateSQLTokens(final SQLStatementContext sqlStatementContext) {
return sqlStatementContext instanceof TableAvailable ? generateSQLTokens((TableAvailable) sqlStatementContext) : Collections.emptyList();
}
private Collection<TableToken> generateSQLTokens(final TableAvailable sqlStatementContext) {
Collection<TableToken> result = new LinkedList<>();
for (SimpleTableSegment each : sqlStatementContext.getAllTables()) {
if (shardingRule.findTableRule(each.getTableName().getIdentifier().getValue()).isPresent()) {// 分片功能下，添加TableToken
result.add(new TableToken(each.getStartIndex(), each.getStopIndex(), each.getTableName().getIdentifier(), (SQLStatementContext) sqlStatementContext, shardingRule));
}
}
return result;
}
}
可以看到generateSQLTokens方法中，在判断时数据分片规则中配置的表后，创建TableToken对象添加到集合中返回。
org.apache.shardingsphere.sharding.rewrite.token.pojo.TableToken
/**
* Table token.
*/
public final class TableToken extends SQLToken implements Substitutable, RouteUnitAware {
…
@Override
public String toString(final RouteUnit routeUnit) {
String actualTableName = getLogicAndActualTables(routeUnit).get(identifier.getValue().toLowerCase());
actualTableName = null == actualTableName ? identifier.getValue().toLowerCase() : actualTableName;
//替换成真实物理表名
return Joiner.on("").join(identifier.getQuoteCharacter().getStartDelimiter(), actualTableName, identifier.getQuoteCharacter().getEndDelimiter());
}
…
}
TableToken的toString方法即根据RouteUnit对象生成该Token在SQL改写时需要替换成的字符串。
再次回到BasePrepareEngine类
org.apache.shardingsphere.underlying.pluggble.prepare.BasePrepareEngine# executeRewrite方法，在创建完SQL改写上下文后，调用了rewrite方法生成执行单元集合
private Collection<ExecutionUnit> executeRewrite(final String sql, final List<Object> parameters, final RouteContext routeContext) {
registerRewriteDecorator();
//创建SQL重写上下文，主要是生成对应的Token以及参数
SQLRewriteContext sqlRewriteContext = rewriter.createSQLRewriteContext(sql, parameters, routeContext.getSqlStatementContext(), routeContext);
return routeContext.getRouteResult().getRouteUnits().isEmpty() ? rewrite(sqlRewriteContext) : rewrite(routeContext, sqlRewriteContext);
}
可以看到根据路由单元数量，分别对应两个rewrite私有方法
// 此方法负责将SQL改写上下文转化为执行单元ExecutionUnit集合
private Collection<ExecutionUnit> rewrite(final SQLRewriteContext sqlRewriteContext) {
SQLRewriteResult sqlRewriteResult = new SQLRewriteEngine().rewrite(sqlRewriteContext);// 将SQL改写上下文转化为SQL改写结果，主要是获取改写后的SQL与参数
String dataSourceName = metaData.getDataSources().getAllInstanceDataSourceNames().iterator().next();
return Collections.singletonList(new ExecutionUnit(dataSourceName, new SQLUnit(sqlRewriteResult.getSql(), sqlRewriteResult.getParameters())));
}
private Collection<ExecutionUnit> rewrite(final RouteContext routeContext, final SQLRewriteContext sqlRewriteContext) {
Collection<ExecutionUnit> result = new LinkedHashSet<>();
for (Entry<RouteUnit, SQLRewriteResult> entry : new SQLRouteRewriteEngine().rewrite(sqlRewriteContext, routeContext.getRouteResult()).entrySet()) {
result.add(new ExecutionUnit(entry.getKey().getDataSourceMapper().getActualName(), new SQLUnit(entry.getValue().getSql(), entry.getValue().getParameters())));
}
return result;
}
第一个rewrite方法中，创建SQLRewriteEngine实例，然后执行其rewrite方法生成SQLRewriteResult对象，通过此对象获取改写后的SQL以及参数，之后创建SQLUnit对象、创建ExecutionUnit对象，添加到集合中返回。接下来看下SQLRewriteEngine类：
org.apache.shardingsphere.underlying.rewrite.engine.SQLRewriteEngine
/**
* SQL rewrite engine.
*/
public final class SQLRewriteEngine {
/**
* Rewrite SQL and parameters.
*
* @param sqlRewriteContext SQL rewrite context
* @return SQL rewrite result
*/
public SQLRewriteResult rewrite(final SQLRewriteContext sqlRewriteContext) {
// 将SQL改写上下文中Token转化成SQL， 获取改写后参数，然后构造成SQL改写结果返回
return new SQLRewriteResult(new DefaultSQLBuilder(sqlRewriteContext).toSQL(), sqlRewriteContext.getParameterBuilder().getParameters());
}
}
可以看到该类通过DefaultSQLBuilder类对象的toSQL()生成了改写后SQL，然后创建了SQLRewriteResult实例，那么接下来继续看下DefaultSQLBuilder类
org.apache.shardingsphere.underlying.rewrite.sql.impl.DefaultSQLBuilder
/**
* Default SQL builder.
*/
public final class DefaultSQLBuilder extends AbstractSQLBuilder {
public DefaultSQLBuilder(final SQLRewriteContext context) {
super(context);
}
@Override
protected String getSQLTokenText(final SQLToken sqlToken) {
return sqlToken.toString();// 返回Token对应的文本字符
}
}
DefaultSQLBuilder类只是实现了getSQLTokenText方法，调用Token.toString方法返回。
看下其父类AbstractSQLBuilder
org.apache.shardingsphere.underlying.rewrite.sql.impl.AbstractSQLBuilder
public abstract class AbstractSQLBuilder implements SQLBuilder {
private final SQLRewriteContext context;
@Override
public final String toSQL() {
if (context.getSqlTokens().isEmpty()) {
return context.getSql();
}
Collections.sort(context.getSqlTokens());// 按照Token的起始位置排序
StringBuilder result = new StringBuilder();
result.append(context.getSql().substring(0, context.getSqlTokens().get(0).getStartIndex()));// 添加第一个Token之前的原始SQL
for (SQLToken each : context.getSqlTokens()) {
result.append(getSQLTokenText(each));// 添加Token对应的SQL片段
result.append(getConjunctionText(each));// 添加Token之间的连接字符
}
return result.toString();
}
…
}
可以看到在AbstractSQLBuilder的toSQL方法中，对Token进行排序，然后通过拼接原始SQL和替换的Token以及连接符，最后形成完整的改写SQL。
回头再看下BasePrepareEngine类中的第二个rewrite方法
private Collection<ExecutionUnit> rewrite(final RouteContext routeContext, final SQLRewriteContext sqlRewriteContext) {
Collection<ExecutionUnit> result = new LinkedHashSet<>();
for (Entry<RouteUnit, SQLRewriteResult> entry : new SQLRouteRewriteEngine().rewrite(sqlRewriteContext, routeContext.getRouteResult()).entrySet()) {
result.add(new ExecutionUnit(entry.getKey().getDataSourceMapper().getActualName(), new SQLUnit(entry.getValue().getSql(), entry.getValue().getParameters())));
}
return result;
}
可以看到通过执行了SQLRouteRewriteEngine对象的rewrite方法返回了一个Map<RouteUnit, SQLRewriteResult>对象，然后遍历构建了ExecutionUnit，然后添加到集合中进行返回。那么我们看下SQLRouteRewriteEngine类：
org.apache.shardingsphere.underlying.rewrite.engine.SQLRouteRewriteEngine
/**
* SQL rewrite engine with route.
*/
public final class SQLRouteRewriteEngine {
/**
* Rewrite SQL and parameters.
*
* @param sqlRewriteContext SQL rewrite context
* @param routeResult route result
* @return SQL map of route unit and rewrite result
*/
public Map<RouteUnit, SQLRewriteResult> rewrite(final SQLRewriteContext sqlRewriteContext, final RouteResult routeResult) {
Map<RouteUnit, SQLRewriteResult> result = new LinkedHashMap<>(routeResult.getRouteUnits().size(), 1);
for (RouteUnit each : routeResult.getRouteUnits()) {
result.put(each, new SQLRewriteResult(new RouteSQLBuilder(sqlRewriteContext, each).toSQL(), getParameters(sqlRewriteContext.getParameterBuilder(), routeResult, each)));
}
return result;
}
…
}
可以看到，与SQLRewriteEngine类相似，只不过改为遍历路由结果中包含的RouteUnit，然后分别创建对应的SQLRewriteResult实例，这里构建改写SQL的类用了一个RouteSQLBuilder类
org.apache.shardingsphere.underlying.rewrite.sql.impl.RouteSQLBuilder
/**
* SQL builder with route.
*/
public final class RouteSQLBuilder extends AbstractSQLBuilder {
private final RouteUnit routeUnit;
public RouteSQLBuilder(final SQLRewriteContext context, final RouteUnit routeUnit) {
super(context);
this.routeUnit = routeUnit;
}
@Override
protected String getSQLTokenText(final SQLToken sqlToken) {
if (sqlToken instanceof RouteUnitAware) {
return ((RouteUnitAware) sqlToken).toString(routeUnit);
}
return sqlToken.toString();
}
}
…
}
这个类与DefaultSQLBuilder类似，也继承自AbstractSQLBuilder类，只不过getSQLTokenText方法会判断是否是RouteUnitAware类型的Token，如果是则调用RouteUnit参数的toSQL方法生成SQL。
总结
最后总结下改写引擎的执行流程：
改写引擎的输入即为路由上下文RouteContext，输出为SQL改写上下文SQLRewriteResult。
- 改写功能入口SQLRewriteEntry类中创建一个初始化SQLRewriteResult对象。
- 顺序执行已注册的SQL改写上下文装饰器，添加对应的一系列Token生成器对象。
- 调用SQLRewriteContext. generateSQLTokens方法，运行各Token生成器，解析出对应的Token。
- 由SQL改写引擎SQLRewriteEngine或者SQLRouteRewriteEngine类调用DefaultSQLBuilder或RouteSQLBuilder类将解析出的Token拼接成改写后的SQL，基于此SQL和参数，创建SQLRewriteResult实例返回。
- 基于SQLRewriteResult，创建ExecutionUnit，然后再封装成ExecutionContext。
改写引擎执行流程
