public LogicalType getLogicalType(String tableName, String fieldName) {
        ScopeChild scopeChild = aliasMap.get(tableName);
        if (scopeChild == null) {
            throw new RuntimeException("can't find ");
        }

        BaseRowTypeInfo rowTypeInfo = scopeChild.getBaseRowTypeInfo();
        int index = rowTypeInfo.getFieldIndex(fieldName);
        if (index == -1) {
            throw new RuntimeException("can't find field: " + fieldName);
        }

        return rowTypeInfo.getLogicalTypes()[index];
    }
Reason:
Date: 2018/7/20
Company: www.dtstack.com

@author xuchao
core/src/main/java/com/dtstack/flink/sql/side/JoinScope.java