Skip to content

Commit

Permalink
重构 sql语句解析逻辑;
Browse files Browse the repository at this point in the history
字段替换和重命名相关的逻辑在解析阶段完成;
目标:经过解析sql之后的所有sql是确定的。
  • Loading branch information
xuchao committed Mar 25, 2020
1 parent 14bc810 commit 2833176
Show file tree
Hide file tree
Showing 6 changed files with 45 additions and 11 deletions.
Original file line number Diff line number Diff line change
Expand Up @@ -57,6 +57,7 @@ public void parseSql(String sql, SqlTree sqlTree) {
.configBuilder()
.setLex(Lex.MYSQL)
.build();

SqlParser sqlParser = SqlParser.create(sql,config);
SqlNode sqlNode = null;
try {
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -731,9 +731,9 @@ private boolean checkAndRemoveWhereCondition(Set<String> fromTableNameSet,
Set<String> conditionRefTableNameSet = Sets.newHashSet();

fieldInfos.forEach(fieldInfo -> {
String[] splitInfo = StringUtils.split(fieldInfo, ",");
String[] splitInfo = StringUtils.split(fieldInfo, ".");
if(splitInfo.length == 2){
conditionRefTableNameSet.add(splitInfo[1]);
conditionRefTableNameSet.add(splitInfo[0]);
}
});

Expand Down
36 changes: 31 additions & 5 deletions core/src/main/java/com/dtstack/flink/sql/side/SideInfo.java
Original file line number Diff line number Diff line change
Expand Up @@ -55,6 +55,8 @@ public abstract class SideInfo implements Serializable{

protected String sideSelectFields = "";

protected Map<Integer, String> sideSelectFieldsType = Maps.newHashMap();

protected JoinType joinType;

//key:Returns the value of the position, value: the ref field index​in the input table
Expand Down Expand Up @@ -84,15 +86,17 @@ public void parseSelectFields(JoinInfo joinInfo){
String sideTableName = joinInfo.getSideTableName();
String nonSideTableName = joinInfo.getNonSideTable();
List<String> fields = Lists.newArrayList();
int sideTableFieldIndex = 0;

int sideIndex = 0;
for( int i=0; i<outFieldInfoList.size(); i++){
FieldInfo fieldInfo = outFieldInfoList.get(i);
if(fieldInfo.getTable().equalsIgnoreCase(sideTableName)){
fields.add(sideTableInfo.getPhysicalFields().getOrDefault(fieldInfo.getFieldName(), fieldInfo.getFieldName()));
sideFieldIndex.put(i, sideIndex);
sideFieldNameIndex.put(i, fieldInfo.getFieldName());
sideIndex++;
String sideFieldName = sideTableInfo.getPhysicalFields().getOrDefault(fieldInfo.getFieldName(), fieldInfo.getFieldName());
fields.add(sideFieldName);
sideSelectFieldsType.put(sideTableFieldIndex, getTargetFieldType(sideFieldName));
sideFieldIndex.put(i, sideTableFieldIndex);
sideFieldNameIndex.put(i, sideFieldName);
sideTableFieldIndex++;
}else if(fieldInfo.getTable().equalsIgnoreCase(nonSideTableName)){
int nonSideIndex = rowTypeInfo.getFieldIndex(fieldInfo.getFieldName());
inFieldIndex.put(i, nonSideIndex);
Expand All @@ -108,6 +112,16 @@ public void parseSelectFields(JoinInfo joinInfo){
sideSelectFields = String.join(",", fields);
}

public String getTargetFieldType(String fieldName){
int fieldIndex = sideTableInfo.getFieldList().indexOf(fieldName);
if(fieldIndex == -1){
throw new RuntimeException(sideTableInfo.getName() + "can't find field: " + fieldName);
}

return sideTableInfo.getFieldTypes()[fieldIndex];
}


public void dealOneEqualCon(SqlNode sqlNode, String sideTableName){
if(!SqlKind.COMPARISON.contains(sqlNode.getKind())){
throw new RuntimeException("not compare operator.");
Expand Down Expand Up @@ -255,4 +269,16 @@ public Map<Integer, String> getSideFieldNameIndex() {
public void setSideFieldNameIndex(Map<Integer, String> sideFieldNameIndex) {
this.sideFieldNameIndex = sideFieldNameIndex;
}

public Map<Integer, String> getSideSelectFieldsType() {
return sideSelectFieldsType;
}

public void setSideSelectFieldsType(Map<Integer, String> sideSelectFieldsType) {
this.sideSelectFieldsType = sideSelectFieldsType;
}

public String getSelectSideFieldType(int index){
return sideSelectFieldsType.get(index);
}
}
Original file line number Diff line number Diff line change
Expand Up @@ -394,7 +394,6 @@ private void joinFun(Object pollObj,
HashBasedTable<String, String, String> mappingTable = ((JoinInfo) pollObj).getTableFieldRef();

//获取两个表的所有字段
//TODO 抽取
List<FieldInfo> sideJoinFieldInfo = ParserJoinField.getRowTypeInfo(joinInfo.getSelectNode(), joinScope, true);
//通过join的查询字段信息过滤出需要的字段信息
sideJoinFieldInfo.removeIf(tmpFieldInfo -> mappingTable.get(tmpFieldInfo.getTable(), tmpFieldInfo.getFieldName()) == null);
Expand Down
10 changes: 9 additions & 1 deletion core/src/main/java/com/dtstack/flink/sql/util/TableUtils.java
Original file line number Diff line number Diff line change
Expand Up @@ -26,6 +26,7 @@
import com.google.common.collect.HashBasedTable;
import com.google.common.collect.HashBiMap;
import com.google.common.collect.Lists;
import com.google.common.collect.Sets;
import com.typesafe.config.ConfigException;
import org.apache.calcite.sql.SqlAsOperator;
import org.apache.calcite.sql.SqlBasicCall;
Expand Down Expand Up @@ -200,7 +201,14 @@ public static SqlBasicCall buildAsNodeByJoinInfo(JoinInfo joinInfo, SqlNode sqlN
String joinLeftTableAlias = joinInfo.getLeftTableAlias();
joinLeftTableName = Strings.isNullOrEmpty(joinLeftTableName) ? joinLeftTableAlias : joinLeftTableName;
String newTableName = buildInternalTableName(joinLeftTableName, SPLIT, joinInfo.getRightTableName());
String newTableAlias = !StringUtils.isEmpty(tableAlias) ? tableAlias : buildInternalTableName(joinInfo.getLeftTableAlias(), SPLIT, joinInfo.getRightTableAlias());
String lefTbAlias = joinInfo.getLeftTableAlias();
if(Strings.isNullOrEmpty(lefTbAlias)){
Set<String> fromTableSet = Sets.newHashSet();
TableUtils.getFromTableInfo(joinInfo.getLeftNode(), fromTableSet);
lefTbAlias = StringUtils.join(fromTableSet, "_");
}

String newTableAlias = !StringUtils.isEmpty(tableAlias) ? tableAlias : buildInternalTableName(lefTbAlias, SPLIT, joinInfo.getRightTableAlias());

if (null == sqlNode0) {
sqlNode0 = new SqlIdentifier(newTableName, null, sqlParserPos);
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -201,7 +201,6 @@ protected List<Row> getRows(Row inputRow, List<JsonArray> cacheContent, List<Jso
public Row fillData(Row input, Object line) {
JsonArray jsonArray = (JsonArray) line;
Row row = new Row(sideInfo.getOutFieldInfoList().size());
String[] fields = sideInfo.getSideTableInfo().getFieldTypes();
for (Map.Entry<Integer, Integer> entry : sideInfo.getInFieldIndex().entrySet()) {
Object obj = input.getField(entry.getValue());
boolean isTimeIndicatorTypeInfo = TimeIndicatorTypeInfo.class.isAssignableFrom(sideInfo.getRowTypeInfo().getTypeAt(entry.getValue()).getClass());
Expand All @@ -216,7 +215,8 @@ public Row fillData(Row input, Object line) {
if (jsonArray == null) {
row.setField(entry.getKey(), null);
} else {
Object object = SwitchUtil.getTarget(jsonArray.getValue(entry.getValue()), fields[entry.getValue()]);
String fieldType = sideInfo.getSelectSideFieldType(entry.getValue());
Object object = SwitchUtil.getTarget(jsonArray.getValue(entry.getValue()), fieldType);
row.setField(entry.getKey(), object);
}
}
Expand Down

0 comments on commit 2833176

Please sign in to comment.