[feat-858][doris] Doris Connector add sql mode.

Author: FlechazoW

chunjun-connectors/chunjun-connector-doris/pom.xml

@@ -42,6 +42,12 @@
 			<artifactId>jackson-databind</artifactId>
 			<version>2.9.10.1</version>
 		</dependency>
+
+		<dependency>
+			<groupId>com.dtstack.chunjun</groupId>
+			<artifactId>chunjun-connector-mysql</artifactId>
+			<version>${project.version}</version>
+		</dependency>
 	</dependencies>
 
 	<build>

chunjun-connectors/chunjun-connector-doris/src/main/java/com/dtstack/chunjun/connector/doris/DorisUtil.java

@@ -0,0 +1,97 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.dtstack.chunjun.connector.doris;
+
+import org.apache.commons.lang3.StringUtils;
+
+import java.util.Collections;
+import java.util.List;
+
+public class DorisUtil {
+
+    private static final String JDBC_QUERY_PORT = "9030";
+
+    private static final String JDBC_TEMPLATE = "jdbc:mysql://%s:%s?useSSL=false";
+
+    /**
+     * split fe and get fe ip. like: http://172.16.21.193:8030
+     *
+     * @param fe fe
+     * @return ip of fe.
+     */
+    private static String splitFe(String fe) {
+        String[] split = fe.split("://");
+        for (String s : split) {
+            String[] items = s.split(":");
+            if (items.length == 2) {
+                return items[0];
+            }
+        }
+        throw new RuntimeException("Get fe ip from fe uri: " + fe + " failed.");
+    }
+
+    public static String getJdbcUrlFromFe(List<String> feNodes, String url) {
+        if (StringUtils.isEmpty(url)) {
+            Collections.shuffle(feNodes);
+            String fe = feNodes.get(0);
+            String feIp = splitFe(fe);
+            return String.format(JDBC_TEMPLATE, feIp, JDBC_QUERY_PORT);
+        }
+        return url;
+    }
+
+    public interface Accept<T> {
+        void accept(T t) throws Exception;
+    }
+
+    public interface Action {
+        void action() throws Exception;
+    }
+
+    public static <T> void doRetry(
+            Accept<T> accept, Action action, T t, int retryTimes, long sleepTimeOut)
+            throws Exception {
+        for (int i = 0; i < retryTimes; i++) {
+            try {
+                accept.accept(t);
+                return;
+            } catch (Exception exception) {
+                boolean retry = needRetry(exception);
+                if (i + 1 == retryTimes || !retry) {
+                    throw exception;
+                }
+
+                try {
+                    Thread.sleep(sleepTimeOut);
+                } catch (InterruptedException interruptedException) {
+                    // ignore
+                }
+                action.action();
+            }
+        }
+    }
+
+    private static boolean needRetry(Exception exception) {
+        if (exception != null) {
+            String errorMessage = exception.getMessage();
+            return StringUtils.isNotEmpty(errorMessage) && errorMessage.contains("err=-235");
+        }
+        return false;
+    }
+}

…tor/doris/converter/DorisRowConvert.java …ris/converter/DorisHttpRowConverter.javachunjun-connectors/chunjun-connector-doris/src/main/java/com/dtstack/chunjun/connector/doris/converter/DorisRowConvert.java renamed to chunjun-connectors/chunjun-connector-doris/src/main/java/com/dtstack/chunjun/connector/doris/converter/DorisHttpRowConverter.java chunjun-connectors/chunjun-connector-doris/src/main/java/com/dtstack/chunjun/connector/doris/converter/DorisRowConvert.java renamed to chunjun-connectors/chunjun-connector-doris/src/main/java/com/dtstack/chunjun/connector/doris/converter/DorisHttpRowConverter.java

@@ -35,14 +35,14 @@
  * @author xuchao
  * @date 2021-11-21
  */
-public class DorisRowConvert
+public class DorisHttpRowConverter
         extends AbstractRowConverter<RowData, RowData, StringJoiner, LogicalType> {
 
     private static final long serialVersionUID = 2L;
 
     private static final String NULL_VALUE = "\\N";
 
-    public DorisRowConvert(RowType rowType) {
+    public DorisHttpRowConverter(RowType rowType) {
         super(rowType);
         for (int i = 0; i < rowType.getFieldCount(); i++) {
             toInternalConverters.add(

chunjun-connectors/chunjun-connector-doris/src/main/java/com/dtstack/chunjun/connector/doris/converter/DorisJdbcRowConverter.java

@@ -0,0 +1,29 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.dtstack.chunjun.connector.doris.converter;
+
+import com.dtstack.chunjun.connector.jdbc.converter.JdbcRowConverter;
+
+import org.apache.flink.table.types.logical.RowType;
+
+public class DorisJdbcRowConverter extends JdbcRowConverter {
+    public DorisJdbcRowConverter(RowType rowType) {
+        super(rowType);
+    }
+}

chunjun-connectors/chunjun-connector-doris/src/main/java/com/dtstack/chunjun/connector/doris/converter/DorisRowConverter.java

@@ -0,0 +1,95 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.dtstack.chunjun.connector.doris.converter;
+
+import com.dtstack.chunjun.throwable.UnsupportedTypeException;
+
+import org.apache.flink.table.api.DataTypes;
+import org.apache.flink.table.types.DataType;
+
+import java.util.Locale;
+
+/**
+ * @author xuchao
+ * @date 2021-11-21
+ */
+public class DorisRowTypeConverter {
+
+    public static DataType apply(String type) {
+        switch (type.toUpperCase(Locale.ENGLISH)) {
+            case "BOOLEAN":
+            case "BIT":
+                return DataTypes.BOOLEAN();
+            case "TINYINT":
+                return DataTypes.TINYINT();
+            case "SMALLINT":
+            case "MEDIUMINT":
+            case "INT":
+            case "INTEGER":
+            case "INT24":
+                return DataTypes.INT();
+            case "BIGINT":
+                return DataTypes.BIGINT();
+            case "REAL":
+            case "FLOAT":
+                return DataTypes.FLOAT();
+            case "DECIMAL":
+            case "NUMERIC":
+            case "DECIMALV2":
+                return DataTypes.DECIMAL(38, 18);
+            case "DOUBLE":
+                return DataTypes.DOUBLE();
+            case "CHAR":
+            case "VARCHAR":
+            case "STRING":
+            case "JSON":
+            case "TINYTEXT":
+            case "TEXT":
+            case "MEDIUMTEXT":
+            case "LONGTEXT":
+            case "ENUM":
+            case "SET":
+                return DataTypes.STRING();
+            case "DATE":
+                return DataTypes.DATE();
+            case "TIME":
+                return DataTypes.TIME();
+            case "YEAR":
+                return DataTypes.INTERVAL(DataTypes.YEAR());
+            case "TIMESTAMP":
+            case "DATETIME":
+                return DataTypes.TIMESTAMP(0);
+            case "TINYBLOB":
+            case "BLOB":
+            case "MEDIUMBLOB":
+            case "LONGBLOB":
+            case "BINARY":
+            case "VARBINARY":
+                // BYTES 底层调用的是VARBINARY最大长度
+            case "GEOMETRY":
+                return DataTypes.BYTES();
+            case "NULL_TYPE":
+            case "NULL":
+                return DataTypes.NULL();
+            case "HLL":
+            default:
+                throw new UnsupportedTypeException(type);
+        }
+    }
+}