Merge branch '1.5_v3.6.0_oralce_table_column_case_bug' into '1.5_v3.6.0'

修改oracle 拼接sql，对表和字段 添加 \"\"

(1)修改oracle 拼接sql，对表和字段 添加 \"\"
(2)修改json解析的时候遇到未定义的key的时候不再抛异常，返回null

See merge request !7

Author: toutian

kafka09/kafka09-source/src/main/java/com/dtstack/flink/sql/source/kafka/CustomerJsonDeserialization.java

@@ -150,6 +150,10 @@ public void setFailOnMissingField(boolean failOnMissingField) {
     private JsonNode getIgnoreCase(String key) {
         String nodeMappingKey = rowAndFieldMapping.getOrDefault(key, key);
         JsonNode node = nodeAndJsonNodeMapping.get(nodeMappingKey);
+        if(node == null){
+            return null;
+        }
+
         JsonNodeType nodeType = node.getNodeType();
 
         if (nodeType == JsonNodeType.ARRAY){

kafka10/kafka10-source/src/main/java/com/dtstack/flink/sql/source/kafka/CustomerJsonDeserialization.java

@@ -145,6 +145,11 @@ public Row deserialize(byte[] message) throws IOException {
     public JsonNode getIgnoreCase(String key) {
         String nodeMappingKey = rowAndFieldMapping.getOrDefault(key, key);
         JsonNode node = nodeAndJsonNodeMapping.get(nodeMappingKey);
+
+        if(node == null){
+            return null;
+        }
+
         JsonNodeType nodeType = node.getNodeType();
 
         if (nodeType==JsonNodeType.ARRAY){

kafka11/kafka11-source/src/main/java/com/dtstack/flink/sql/source/kafka/CustomerJsonDeserialization.java

@@ -148,6 +148,10 @@ public Row deserialize(byte[] message) throws IOException {
     public JsonNode getIgnoreCase(String key) {
         String nodeMappingKey = rowAndFieldMapping.getOrDefault(key, key);
         JsonNode node = nodeAndJsonNodeMapping.get(nodeMappingKey);
+        if(node == null){
+            return null;
+        }
+
         JsonNodeType nodeType = node.getNodeType();
 
         if (nodeType==JsonNodeType.ARRAY){

oracle/oracle-sink/src/main/java/com/dtstack/flink/sql/sink/oracle/OracleSink.java

@@ -22,6 +22,7 @@
 import com.dtstack.flink.sql.sink.rdb.format.ExtendOutputFormat;
 import com.dtstack.flink.sql.sink.rdb.format.RetractJDBCOutputFormat;
 import org.apache.commons.lang3.StringUtils;
+import org.apache.flink.shaded.guava18.com.google.common.collect.Lists;
 
 import java.util.ArrayList;
 import java.util.Iterator;
@@ -54,8 +55,14 @@ public void buildSql(String tableName, List<String> fields) {
     }
 
     private void buildInsertSql(String tableName, List<String> fields) {
+
+        tableName = quoteTable(tableName);
         String sqlTmp = "insert into " + tableName + " (${fields}) values (${placeholder})";
-        String fieldsStr = StringUtils.join(fields, ",");
+
+        List<String> adaptFields = Lists.newArrayList();
+        fields.forEach(field -> adaptFields.add(quoteColumn(field)));
+
+        String fieldsStr = StringUtils.join(adaptFields, ",");
         String placeholder = "";
 
         for (String fieldName : fields) {
@@ -68,6 +75,7 @@ private void buildInsertSql(String tableName, List<String> fields) {
 
     @Override
     public String buildUpdateSql(String tableName, List<String> fieldNames, Map<String, List<String>> realIndexes, List<String> fullField) {
+        tableName = quoteTable(tableName);
         return "MERGE INTO " + tableName + " T1 USING "
                 + "(" + makeValues(fieldNames) + ") T2 ON ("
                 + updateKeySql(realIndexes) + ") WHEN MATCHED THEN UPDATE SET "