解决kudu join条件解析问题

Author: xiuzhu9527

kudu/kudu-side/kudu-async-side/src/main/java/com/dtstack/flink/sql/side/kudu/KuduAsyncSideInfo.java

@@ -5,6 +5,7 @@
 import com.dtstack.flink.sql.side.SideInfo;
 import com.dtstack.flink.sql.side.SideTableInfo;
 import com.dtstack.flink.sql.side.kudu.table.KuduSideTableInfo;
+import com.dtstack.flink.sql.util.ParseUtils;
 import org.apache.calcite.sql.SqlBasicCall;
 import org.apache.calcite.sql.SqlIdentifier;
 import org.apache.calcite.sql.SqlKind;
@@ -30,11 +31,7 @@ public void buildEqualInfo(JoinInfo joinInfo, SideTableInfo sideTableInfo) {
         SqlNode conditionNode = joinInfo.getCondition();
 
         List<SqlNode> sqlNodeList = Lists.newArrayList();
-        if (conditionNode.getKind() == SqlKind.AND) {
-            sqlNodeList.addAll(Lists.newArrayList(((SqlBasicCall) conditionNode).getOperands()));
-        } else {
-            sqlNodeList.add(conditionNode);
-        }
+        ParseUtils.parseAnd(conditionNode, sqlNodeList);
 
         for (SqlNode sqlNode : sqlNodeList) {
             dealOneEqualCon(sqlNode, sideTableName);

kudu/kudu-side/kudu-side-core/src/main/java/com/dtstack/flink/sql/side/kudu/table/KuduSideParser.java

@@ -93,7 +93,6 @@ public Class dbTypeConvertToJavaType(String fieldType) {
             case "int64":
                 return Long.class;
             case "varchar":
-            case "binary":
             case "string":
                 return String.class;
             case "float":
@@ -106,6 +105,8 @@ public Class dbTypeConvertToJavaType(String fieldType) {
                 return Timestamp.class;
             case "decimal":
                 return BigDecimal.class;
+            case "binary":
+                return byte[].class;
         }
 
         throw new RuntimeException("不支持 " + fieldType + " 类型");

kudu/kudu-sink/src/main/java/com/dtstack/flink/sql/sink/kudu/KuduOutputFormat.java

@@ -24,6 +24,8 @@
 import org.apache.flink.configuration.Configuration;
 import org.apache.flink.types.Row;
 import org.apache.kudu.client.*;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 import java.io.IOException;
 import java.math.BigDecimal;
@@ -32,6 +34,10 @@
 
 public class KuduOutputFormat extends MetricOutputFormat {
 
+    private static final long serialVersionUID = 1L;
+
+    private static final Logger LOG = LoggerFactory.getLogger(KuduOutputFormat.class);
+
     public enum WriteMode {INSERT, UPDATE, UPSERT}
 
     private String kuduMasters;
@@ -104,9 +110,17 @@ public void writeRecord(Tuple2 record) throws IOException {
 
         Operation operation = toOperation(writeMode, row);
         AsyncKuduSession session = client.newSession();
-        session.apply(operation);
-        session.close();
-        outRecords.inc();
+
+        try {
+            session.apply(operation);
+            session.close();
+            outRecords.inc();
+        } catch (KuduException e) {
+            outDirtyRecords.inc();
+            LOG.error("record insert failed ..", row.toString().substring(0, 100));
+            LOG.error("", e);
+        }
+
     }
 
     @Override

kudu/kudu-sink/src/main/java/com/dtstack/flink/sql/sink/kudu/table/KuduSinkParser.java

@@ -72,7 +72,6 @@ public Class dbTypeConvertToJavaType(String fieldType) {
             case "int64":
                 return Long.class;
             case "varchar":
-            case "binary":
             case "string":
                 return String.class;
             case "float":
@@ -85,6 +84,8 @@ public Class dbTypeConvertToJavaType(String fieldType) {
                 return Timestamp.class;
             case "decimal":
                 return BigDecimal.class;
+            case "binary":
+                return byte[].class;
         }
 
         throw new RuntimeException("不支持 " + fieldType + " 类型");