Skip to content

Commit 6f8c9aa

Browse files
committed
fix 别名join异常
1 parent 83429b2 commit 6f8c9aa

File tree

2 files changed

+6
-5
lines changed

2 files changed

+6
-5
lines changed

core/src/main/java/com/dtstack/flink/sql/side/SideInfo.java

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -93,7 +93,7 @@ public void parseSelectFields(JoinInfo joinInfo){
9393
if(fieldInfo.getTable().equalsIgnoreCase(sideTableName)){
9494
String sideFieldName = sideTableInfo.getPhysicalFields().getOrDefault(fieldInfo.getFieldName(), fieldInfo.getFieldName());
9595
fields.add(sideFieldName);
96-
sideSelectFieldsType.put(sideTableFieldIndex, getTargetFieldType(sideFieldName));
96+
sideSelectFieldsType.put(sideTableFieldIndex, getTargetFieldType(fieldInfo.getFieldName()));
9797
sideFieldIndex.put(i, sideTableFieldIndex);
9898
sideFieldNameIndex.put(i, sideFieldName);
9999
sideTableFieldIndex++;

rdb/rdb-side/src/main/java/com/dtstack/flink/sql/side/rdb/async/RdbAsyncSideInfo.java

Lines changed: 5 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -18,8 +18,6 @@
1818

1919
package com.dtstack.flink.sql.side.rdb.async;
2020

21-
import org.apache.flink.api.java.typeutils.RowTypeInfo;
22-
2321
import com.dtstack.flink.sql.side.FieldInfo;
2422
import com.dtstack.flink.sql.side.JoinInfo;
2523
import com.dtstack.flink.sql.side.PredicateInfo;
@@ -33,9 +31,11 @@
3331
import org.apache.calcite.sql.SqlKind;
3432
import org.apache.calcite.sql.SqlNode;
3533
import org.apache.commons.lang3.StringUtils;
34+
import org.apache.flink.api.java.typeutils.RowTypeInfo;
3635

3736
import java.util.Arrays;
3837
import java.util.List;
38+
import java.util.Map;
3939
import java.util.stream.Collectors;
4040

4141

@@ -86,6 +86,7 @@ public void dealOneEqualCon(SqlNode sqlNode, String sideTableName) {
8686

8787
SqlIdentifier left = (SqlIdentifier) ((SqlBasicCall) sqlNode).getOperands()[0];
8888
SqlIdentifier right = (SqlIdentifier) ((SqlBasicCall) sqlNode).getOperands()[1];
89+
Map<String, String> physicalFields = sideTableInfo.getPhysicalFields();
8990

9091
String leftTableName = left.getComponent(0).getSimple();
9192
String leftField = left.getComponent(1).getSimple();
@@ -94,7 +95,7 @@ public void dealOneEqualCon(SqlNode sqlNode, String sideTableName) {
9495
String rightField = right.getComponent(1).getSimple();
9596

9697
if (leftTableName.equalsIgnoreCase(sideTableName)) {
97-
equalFieldList.add(leftField);
98+
equalFieldList.add(physicalFields.get(leftField));
9899
int equalFieldIndex = -1;
99100
for (int i = 0; i < rowTypeInfo.getFieldNames().length; i++) {
100101
String fieldName = rowTypeInfo.getFieldNames()[i];
@@ -110,7 +111,7 @@ public void dealOneEqualCon(SqlNode sqlNode, String sideTableName) {
110111

111112
} else if (rightTableName.equalsIgnoreCase(sideTableName)) {
112113

113-
equalFieldList.add(rightField);
114+
equalFieldList.add(physicalFields.get(rightField));
114115
int equalFieldIndex = -1;
115116
for (int i = 0; i < rowTypeInfo.getFieldNames().length; i++) {
116117
String fieldName = rowTypeInfo.getFieldNames()[i];

0 commit comments

Comments
 (0)