Merge branch 'feat_1.10_4.0.x_impalaKudu' into '1.10_test_4.0.x'

Feat 1.10 4.0.x impala kudu

See merge request dt-insight-engine/flinkStreamSQL!151

Author: simenliuxing

impala/impala-sink/src/main/java/com/dtstack/flink/sql/sink/impala/ImpalaOutputFormat.java

@@ -289,7 +289,7 @@ private void executeUpdateBatch() throws SQLException {
             LOG.debug("impala jdbc execute batch error ", e);
             connection.rollback();
             connection.commit();
-            cleanBatchWhenError();
+            updateStatement.clearBatch();
             executeUpdate(connection);
         }
     }
@@ -317,10 +317,6 @@ public void executeUpdate(Connection connection) {
         rows.clear();
     }
 
-    private void cleanBatchWhenError() throws SQLException {
-        updateStatement.clearBatch();
-    }
-
     private void putRowIntoMap(Map<String, ArrayList<String>> rowDataMap, Tuple2<String, String> rowData) {
         Set<String> keySet = rowDataMap.keySet();
         ArrayList<String> tempRowArray;
@@ -361,6 +357,7 @@ private List<String> rebuildFieldNameListAndTypeList(List<String> fieldNames, Li
      * Quote a specific type of value, like string, timestamp
      * before: 1, cast(tiezhu as string), cast(2001-01-09 01:05:01 as timestamp), cast(123 as int)
      * after: 1, cast('tiezhu' as string), cast('2001-01-09 01:05:01' as timestamp), cast(123 as int)
+     * if cast value is null, then cast(null as type)
      *
      * @param valueCondition original value condition
      * @return quoted condition
@@ -374,13 +371,16 @@ private String valueConditionAddQuotation(String valueCondition) {
                     while (matcher.find()) {
                         String value = matcher.group(1);
                         String type = matcher.group(2);
+
                         if (Arrays.asList(NEED_QUOTE_TYPE).contains(type)) {
-                            valueConditionCopy[0] = valueConditionCopy[0].replace(value, "'" + value + "'");
+                            if (!"null".equals(value)) {
+                                valueConditionCopy[0] = valueConditionCopy[0].replace(value, "'" + value + "'");
+                            }
                         }
                     }
                 }
         );
-        return valueConditionCopy[0];
+        return "(" + valueConditionCopy[0] + ")";
     }
 
     @Override
@@ -514,7 +514,6 @@ private synchronized void executeBatchSql(Statement statement,
                     .replace(PARTITION_CONSTANT, "")
                     .replace(TABLE_FIELDS_CONDITION, tableFieldsCondition);
             String substring = executeSql.substring(0, executeSql.length() - 2);
-            LOG.info("current execute sql: {}", substring);
             statement.execute(substring);
             return;
         }
@@ -530,7 +529,6 @@ private synchronized void executeBatchSql(Statement statement,
                 executeSql = executeSql.replace(PARTITION_CONDITION, partitionCondition.toString())
                         .replace(TABLE_FIELDS_CONDITION, tableFieldsCondition)
                         .replace(VALUES_CONDITION, String.join(", ", valuesConditionList));
-                LOG.info("current execute sql: {}", executeSql);
                 statement.execute(executeSql);
                 partitionCondition.delete(0, partitionCondition.length());
             } catch (SQLException sqlException) {
@@ -580,24 +578,16 @@ private String buildTableFieldsCondition(List<String> fieldNames, String partiti
      * @return condition like '(?, ?, cast(? as string))' and '?' will be replaced with row data
      */
     private String buildValuesCondition(List<String> fieldTypes, Row row) {
-        String valuesCondition = "(" + fieldTypes.stream().map(
+        String valuesCondition = fieldTypes.stream().map(
                 f -> {
-                    switch (f.toUpperCase()) {
-                        case STRING_TYPE:
-                            return "cast(? as string)";
-                        case TIMESTAMP_TYPE:
-                            return "cast(? as timestamp)";
-                        default:
-                            return "?";
+                    if (Arrays.asList(NEED_QUOTE_TYPE).contains(f.toLowerCase())) {
+                        return String.format("cast(? as %s)", f.toLowerCase());
                     }
-                }).collect(Collectors.joining(", ")) + ")";
+                    return "?";
+                }).collect(Collectors.joining(", "));
         for (int i = 0; i < row.getArity(); i++) {
             valuesCondition = valuesCondition.replaceFirst("\\?", Objects.isNull(row.getField(i)) ? "null" : row.getField(i).toString());
         }
-        Matcher matcher = TYPE_PATTERN.matcher(valuesCondition);
-        while (matcher.find()) {
-            valuesCondition = valuesCondition.replace(matcher.group(1), "'" + matcher.group(1) + "'");
-        }
         return valuesCondition;
     }