Merge branch 'v1.4.0'

ysq

Author: sishu@dtstack.com

core/src/main/java/com/dtstack/flink/sql/Main.java

@@ -116,9 +116,9 @@ public static void main(String[] args) throws Exception {
         String deployMode = cl.getOptionValue("mode");
         String confProp = cl.getOptionValue("confProp");
 
-        Preconditions.checkNotNull(sql, "it requires input parameters sql");
-        Preconditions.checkNotNull(name, "it requires input parameters name");
-        Preconditions.checkNotNull(localSqlPluginPath, "it requires input parameters localSqlPluginPath");
+        Preconditions.checkNotNull(sql, "parameters of sql is required");
+        Preconditions.checkNotNull(name, "parameters of name is required");
+        Preconditions.checkNotNull(localSqlPluginPath, "parameters of localSqlPluginPath is required");
 
         sql = URLDecoder.decode(sql, Charsets.UTF_8.name());
         SqlParser.setLocalSqlPluginRoot(localSqlPluginPath);
@@ -133,7 +133,7 @@ public static void main(String[] args) throws Exception {
         DtClassLoader dtClassLoader = new DtClassLoader(new URL[]{}, threadClassLoader);
         Thread.currentThread().setContextClassLoader(dtClassLoader);
 
-        URLClassLoader parentClassloader = null;
+        URLClassLoader parentClassloader;
         if(!LOCAL_MODE.equals(deployMode)){
             parentClassloader = (URLClassLoader) threadClassLoader.getParent();
         }else{

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

@@ -44,9 +44,8 @@
 import java.util.Map;
 
 /**
- * 所有的继承接口命名规则：类型 + "AsyncReqRow" 比如==》MysqlAsyncReqRow
- * 当前只支持Left join / inner join(join)
- * FIXME 不支持right join
+ * All interfaces inherit naming rules: type + "AsyncReqRow" such as == "MysqlAsyncReqRow
+ * only support Left join / inner join(join),not support right join
  * Date: 2018/7/9
  * Company: www.dtstack.com
  * @author xuchao
@@ -72,14 +71,13 @@ public abstract class AsyncReqRow extends RichAsyncFunction<Row, Row> {
 
     protected JoinType joinType;
 
-    //key:返回值位置，返回值在输入数据中的索引位置
+    //key:Returns the value of the position, returns the index values ​​in the input data
     protected Map<Integer, Integer> inFieldIndex = Maps.newHashMap();
 
     protected Map<Integer, Integer> sideFieldIndex = Maps.newHashMap();
 
     protected SideTableInfo sideTableInfo;
 
-    //TODO 需要指定类型
     protected AbsSideCache sideCache;
 
     public AsyncReqRow(RowTypeInfo rowTypeInfo, JoinInfo joinInfo, List<FieldInfo> outFieldInfoList,
@@ -202,7 +200,7 @@ public void dealOneEqualCon(SqlNode sqlNode, String sideTableName){
 
     protected void dealMissKey(Row input, ResultFuture<Row> resultFuture){
         if(joinType == JoinType.LEFT){
-            //保留left 表数据
+            //Reserved left table data
             Row row = fillData(input, null);
             resultFuture.complete(Collections.singleton(row));
         }else{

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

@@ -24,7 +24,7 @@
 import com.dtstack.flink.sql.side.cache.CacheObj;
 
 /**
- * 仅仅用来标记未命中的维表数据
+ * Only the data marked to dimension table miss
  * Date: 2018/8/28
  * Company: www.dtstack.com
  * @author xuchao

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

@@ -33,7 +33,7 @@
 import java.util.List;
 
 /**
- *FIXME 需要考虑是直接返回所有的字段然后在外层再包裹原先的查询？
+ * Need to consider is the direct return to the fields and then all wrapped in the outer layer of the original query?
  * Date: 2018/7/20
  * Company: www.dtstack.com
  * @author xuchao
@@ -42,7 +42,7 @@
 public class ParserJoinField {
 
     /**
-     * 需要解析出selectlist和where中的字段信息
+     * Need to parse the fields of information and where selectlist
      * @return
      */
     public static List<FieldInfo> getRowTypeInfo(SqlNode sqlNode, JoinScope scope, boolean getAll){

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

@@ -43,7 +43,7 @@
 import static org.apache.calcite.sql.SqlKind.*;
 
 /**
- * 解析sql，获取维表的执行信息
+ * Parsing sql, obtain execution information dimension table
  * Date: 2018/7/24
  * Company: www.dtstack.com
  * @author xuchao
@@ -169,27 +169,27 @@ private JoinInfo dealJoinNode(SqlJoin joinNode, Set<String> sideTableSet, Queue<
 
 
     private void dealSelectResultWithJoinInfo(JoinInfo joinInfo, SqlSelect sqlNode, Queue<Object> queueInfo){
-        //SideJoinInfo重命名
+        //SideJoinInfo rename
         if(joinInfo.checkIsSide()){
             joinInfo.setSelectFields(sqlNode.getSelectList());
             joinInfo.setSelectNode(sqlNode);
             if(joinInfo.isRightIsSideTable()){
-                //判断left是不是一个简单表
+                //Analyzing left is not a simple table
                 if(joinInfo.getLeftNode().toString().contains("SELECT")){
                     queueInfo.offer(joinInfo.getLeftNode());
                 }
 
                 queueInfo.offer(joinInfo);
             }else{
-                //判断right是不是一个简单表
+                //Determining right is not a simple table
                 if(joinInfo.getRightNode().getKind() == SELECT){
                     queueInfo.offer(joinInfo.getLeftNode());
                 }
 
                 queueInfo.offer(joinInfo);
             }
 
-            //更新from 节点
+            //Update from node
             SqlOperator operator = new SqlAsOperator();
             SqlParserPos sqlParserPos = new SqlParserPos(0, 0);
             String joinLeftTableName = joinInfo.getLeftTableName();

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

@@ -79,7 +79,7 @@ public void exec(String sql, Map<String, SideTableInfo> sideTableMap, StreamTabl
         Queue<Object> exeQueue = sideSQLParser.getExeQueue(sql, sideTableMap.keySet());
         Object pollObj = null;
 
-        //TODO 需要清理
+        //need clean
         boolean preIsSideJoin = false;
         List<FieldReplaceInfo> replaceInfoList = Lists.newArrayList();
 
@@ -150,7 +150,7 @@ public void exec(String sql, Map<String, SideTableInfo> sideTableMap, StreamTabl
                 RowTypeInfo typeInfo = new RowTypeInfo(targetTable.getSchema().getTypes(), targetTable.getSchema().getColumnNames());
                 DataStream adaptStream = tableEnv.toAppendStream(targetTable, org.apache.flink.types.Row.class);
 
-                //join side table 之前先 keyby ===>减少 维表在各个async 的缓存大小
+                //join side table before keyby ===> Reducing the size of each dimension table cache of async
                 if(sideTableInfo.isPartitionedJoin()){
                     List<String> leftJoinColList = getConditionFields(joinInfo.getCondition(), joinInfo.getLeftTableAlias());
                     String[] leftJoinColArr = new String[leftJoinColList.size()];
@@ -159,7 +159,7 @@ public void exec(String sql, Map<String, SideTableInfo> sideTableMap, StreamTabl
                 }
 
                 AsyncReqRow asyncDbReq = loadAsyncReq(sideTableInfo.getType(), localSqlPluginPath, typeInfo, joinInfo, sideJoinFieldInfo, sideTableInfo);
-                //TODO 并行度应该设置为多少？超时时间设置？ capacity设置？
+                //TODO How much should be set for the degree of parallelism? Timeout? capacity settings?
                 DataStream dsOut = AsyncDataStream.orderedWait(adaptStream, asyncDbReq, 10000, TimeUnit.MILLISECONDS, 10)
                         .setParallelism(sideTableInfo.getParallelism());
 
@@ -526,7 +526,7 @@ private SqlNode replaceSelectFieldName(SqlNode selectNode, HashBasedTable<String
     }
 
     /**
-     * 判断维表的join 条件十分包含全部的等值条件(即是维表定义中的主键)
+     * Analyzing conditions are very join the dimension tables include all equivalent conditions (i.e., dimension table is the primary key definition
      * @return
      */
     private boolean checkJoinCondition(SqlNode conditionNode, String sideTableAlias,  List<String> primaryKeys){

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

@@ -49,7 +49,7 @@ public abstract class SideTableInfo extends TableInfo implements Serializable {
 
     private int cacheSize = 10000;
 
-    private long cacheTimeout = 60 * 1000;//默认1分钟
+    private long cacheTimeout = 60 * 1000;//
 
     private boolean partitionedJoin = false;
 

core/src/main/java/com/dtstack/flink/sql/sink/StreamSinkFactory.java

@@ -27,7 +27,7 @@
 import org.apache.flink.table.sinks.TableSink;
 
 /**
- * 根据指定的sink type 加载jar,并初始化对象
+ * Loads jar and initializes the object according to the specified sink type
  * Date: 2017/3/10
  * Company: www.dtstack.com
  * @author xuchao

core/src/main/java/com/dtstack/flink/sql/source/IStreamSourceGener.java

@@ -33,7 +33,6 @@
 public interface IStreamSourceGener<T> {
 
     /**
-     * 获取输入源
      * @param sourceTableInfo
      * @param env
      * @param tableEnv

core/src/main/java/com/dtstack/flink/sql/source/StreamSourceFactory.java

@@ -30,7 +30,7 @@
 import org.apache.flink.table.api.java.StreamTableEnvironment;
 
 /**
- * 创建streamTableSource
+ * streamTableSource
  * Date: 2017/3/10
  * Company: www.dtstack.com
  * @author xuchao
@@ -58,7 +58,7 @@ public static AbsSourceParser getSqlParser(String resultType, String sqlRootDir)
     }
 
     /**
-     * 根据指定的类型构造数据源
+     * The configuration of the type specified data source
      * @param sourceTableInfo
      * @return
      */