[fix] impala varchar data error; storeType Null point Exception; values condition quote bug;

Author: FlechazoW

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

@@ -76,7 +76,7 @@ public class ImpalaOutputFormat extends AbstractDtRichOutputFormat<Tuple2<Boolea
     // cast(value as string) -> cast('value' as string)  cast(value as timestamp) -> cast('value' as timestamp)
     private static final Pattern TYPE_PATTERN = Pattern.compile("cast\\((.*) as (.*)\\)");
     //specific type which values need to be quoted
-    private static final String[] NEED_QUOTE_TYPE = {"string", "timestamp"};
+    private static final String[] NEED_QUOTE_TYPE = {"string", "timestamp", "varchar"};
 
     private static final Integer DEFAULT_CONN_TIME_OUT = 60;
     private static final int RECEIVE_DATA_PRINT_FREQUENCY = 1000;
@@ -85,8 +85,6 @@ public class ImpalaOutputFormat extends AbstractDtRichOutputFormat<Tuple2<Boolea
     private static final String KUDU_TYPE = "kudu";
     private static final String UPDATE_MODE = "update";
     private static final String PARTITION_CONSTANT = "PARTITION";
-    private static final String STRING_TYPE = "STRING";
-    private static final String TIMESTAMP_TYPE = "TIMESTAMP";
     private static final String DRIVER_NAME = "com.cloudera.impala.jdbc41.Driver";
 
     private static final String VALUES_CONDITION = "${valuesCondition}";
@@ -363,8 +361,8 @@ private List<String> rebuildFieldNameListAndTypeList(List<String> fieldNames, Li
      * @return quoted condition
      */
     private String valueConditionAddQuotation(String valueCondition) {
-        final String[] valueConditionCopy = {valueCondition};
         String[] temps = valueCondition.split(",");
+        List<String> replacedItem = new ArrayList<>();
         Arrays.stream(temps).forEach(
                 item -> {
                     Matcher matcher = TYPE_PATTERN.matcher(item);
@@ -374,13 +372,15 @@ private String valueConditionAddQuotation(String valueCondition) {
 
                         if (Arrays.asList(NEED_QUOTE_TYPE).contains(type)) {
                             if (!"null".equals(value)) {
-                                valueConditionCopy[0] = valueConditionCopy[0].replace(value, "'" + value + "'");
+                                item = item.replace(value, "'" + value + "'");
                             }
                         }
                     }
+                    replacedItem.add(item);
                 }
         );
-        return "(" + valueConditionCopy[0] + ")";
+
+        return "(" + String.join(", ", replacedItem) + ")";
     }
 
     @Override
@@ -745,6 +745,8 @@ public ImpalaOutputFormat build() {
                 checkNotNull(format.password, "password is required!");
             }
 
+            checkNotNull(format.storeType, "storeType is required!");
+
             return format;
         }
 

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

@@ -46,7 +46,6 @@
 public class ImpalaSink implements RetractStreamTableSink<Row>, IStreamSinkGener<ImpalaSink> {
 
     private static final String DEFAULT_STORE_TYPE = "kudu";
-    private static final String DEFAULT_PARTITION_MODE = "dynamic";
 
     protected String[] fieldNames;
     TypeInformation<?>[] fieldTypes;

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

@@ -25,6 +25,7 @@
 import java.util.Arrays;
 import java.util.List;
 import java.util.Map;
+import java.util.Objects;
 
 /**
  * Date: 2020/10/14
@@ -76,6 +77,8 @@ public class ImpalaSinkParser extends AbstractTableParser {
 
     private static final String KUDU_TYPE = "kudu";
 
+    private static final String DEFAULT_STORE_TYPE = "kudu";
+
     private static final String STORE_TYPE_KEY = "storeType";
 
     private static final String KRB_DEFAULT_REALM = "HADOOP.COM";
@@ -123,13 +126,13 @@ public AbstractTableInfo getTableInfo(String tableName, String fieldsInfo, Map<S
         }
 
         String storeType = MathUtil.getString(props.get(STORE_TYPE_KEY.toLowerCase()));
-        impalaTableInfo.setStoreType(storeType);
+        impalaTableInfo.setStoreType(Objects.isNull(storeType) ? DEFAULT_STORE_TYPE : storeType);
 
         String enablePartitionStr = (String) props.get(ENABLE_PARTITION_KEY.toLowerCase());
         boolean enablePartition = MathUtil.getBoolean(enablePartitionStr == null ? "false" : enablePartitionStr);
         impalaTableInfo.setEnablePartition(enablePartition);
 
-        if (!storeType.equalsIgnoreCase(KUDU_TYPE) && enablePartition) {
+        if (!impalaTableInfo.getStoreType().equalsIgnoreCase(KUDU_TYPE) && enablePartition) {
             String partitionFields = MathUtil.getString(props.get(PARTITION_FIELDS_KEY.toLowerCase()));
             impalaTableInfo.setPartitionFields(partitionFields);
         }

localTest/pom.xml

@@ -99,6 +99,31 @@
             <version>1.0-SNAPSHOT</version>
         </dependency>
 
+
+        <dependency>
+            <groupId>com.dtstack.flink</groupId>
+            <artifactId>sql.mysql</artifactId>
+            <version>1.0-SNAPSHOT</version>
+        </dependency>
+
+        <dependency>
+            <groupId>com.dtstack.flink</groupId>
+            <artifactId>sql.side.all.mysql</artifactId>
+            <version>1.0-SNAPSHOT</version>
+        </dependency>
+
+        <dependency>
+            <groupId>com.dtstack.flink</groupId>
+            <artifactId>sql.sink.mysql</artifactId>
+            <version>1.0-SNAPSHOT</version>
+        </dependency>
+
+        <dependency>
+            <groupId>com.dtstack.flink</groupId>
+            <artifactId>sql.side.async.mysql</artifactId>
+            <version>1.0-SNAPSHOT</version>
+        </dependency>
+
         <dependency>
             <groupId>com.dtstack.flink</groupId>
             <artifactId>sql.console</artifactId>