fix kudu and async quest

DTStack · May 15, 2020 · 7abe4df · 7abe4df
1 parent bbd8cf9
commit 7abe4df
Show file tree

Hide file tree

Showing 2 changed files with 21 additions and 2 deletions.
diff --git a/...side/kudu-async-side/src/main/java/com/dtstack/flink/sql/side/kudu/KuduAsyncSideInfo.java b/...side/kudu-async-side/src/main/java/com/dtstack/flink/sql/side/kudu/KuduAsyncSideInfo.java
@@ -4,6 +4,9 @@
 import com.dtstack.flink.sql.side.JoinInfo;
 import com.dtstack.flink.sql.side.BaseSideInfo;
 import com.dtstack.flink.sql.side.AbstractSideTableInfo;
+import com.dtstack.flink.sql.side.kudu.table.KuduSideTableInfo;
+import com.dtstack.flink.sql.util.ParseUtils;
+import com.google.common.collect.Lists;
 import org.apache.calcite.sql.SqlBasicCall;
 import org.apache.calcite.sql.SqlIdentifier;
 import org.apache.calcite.sql.SqlKind;
@@ -21,6 +24,22 @@ public KuduAsyncSideInfo(RowTypeInfo rowTypeInfo, JoinInfo joinInfo, List<FieldI
 
     @Override
     public void buildEqualInfo(JoinInfo joinInfo, AbstractSideTableInfo sideTableInfo) {
+        KuduSideTableInfo kuduSideTableInfo = (KuduSideTableInfo) sideTableInfo;
+
+        String sideTableName = joinInfo.getSideTableName();
+
+        SqlNode conditionNode = joinInfo.getCondition();
+
+        List<SqlNode> sqlNodeList = Lists.newArrayList();
+        ParseUtils.parseAnd(conditionNode, sqlNodeList);
+
+        for (SqlNode sqlNode : sqlNodeList) {
+            dealOneEqualCon(sqlNode, sideTableName);
+        }
+
+        sqlCondition = "select ${selectField} from ${tableName} ";
+        sqlCondition = sqlCondition.replace("${tableName}", kuduSideTableInfo.getTableName()).replace("${selectField}", sideSelectFields);
+        System.out.println("---------side_exe_sql-----\n" + sqlCondition);
     }
 
     @Override

diff --git a/rdb/rdb-side/src/main/java/com/dtstack/flink/sql/side/rdb/async/RdbAsyncReqRow.java b/rdb/rdb-side/src/main/java/com/dtstack/flink/sql/side/rdb/async/RdbAsyncReqRow.java
@@ -155,13 +155,13 @@ private void connectWithRetry(Map<String, Object> inputParams, CRow input, Resul
                 }
             });
             try {
-                latch.wait();
+                latch.await();
             } catch (InterruptedException e) {
                 LOG.error("", e);
             }
             if(!finishFlag.get()){
                 try {
-                    Thread.sleep(100);
+                    Thread.sleep(3000);
                 } catch (Exception e){
                     LOG.error("", e);
                 }