Merge branch '1.10_release_4.1.x' into 1.10_release_4.2.x

Author: gituser

core/src/main/java/com/dtstack/flink/sql/format/dtnest/DtNestRowDeserializationSchema.java

@@ -36,7 +36,9 @@
 
 import java.io.IOException;
 import java.lang.reflect.Array;
+import java.sql.Clob;
 import java.sql.Date;
+import java.sql.NClob;
 import java.sql.Time;
 import java.sql.Timestamp;
 import java.util.Iterator;
@@ -148,6 +150,10 @@ private Object convert(JsonNode node, TypeInformation<?> info) {
             } else {
                 return node.asText();
             }
+        } else if (info.getTypeClass().equals(Clob.class)) {
+            return node.asText();
+        } else if (info.getTypeClass().equals(NClob.class)) {
+            return node.asText();
         } else if (info.getTypeClass().equals(Types.SQL_DATE.getTypeClass())) {
             return convertToDate(node.asText());
         } else if (info.getTypeClass().equals(Types.SQL_TIME.getTypeClass())) {

core/src/main/java/com/dtstack/flink/sql/util/ClassUtil.java

@@ -22,7 +22,9 @@
 
 import java.lang.reflect.Array;
 import java.math.BigDecimal;
+import java.sql.Clob;
 import java.sql.Date;
+import java.sql.NClob;
 import java.sql.Time;
 import java.sql.Timestamp;
 import java.util.HashMap;
@@ -101,6 +103,10 @@ public static Class<?> stringConvertClass(String str) {
             case "decimal":
             case "decimalunsigned":
                 return BigDecimal.class;
+            case "clob":
+                return Clob.class;
+            case "nclob":
+                return NClob.class;
             default:
                 break;
         }

rdb/rdb-sink/src/main/java/com/dtstack/flink/sql/sink/rdb/JDBCTypeConvertUtils.java

@@ -22,8 +22,11 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import java.io.StringReader;
 import java.math.BigDecimal;
+import java.sql.Clob;
 import java.sql.Date;
+import java.sql.NClob;
 import java.sql.PreparedStatement;
 import java.sql.SQLException;
 import java.sql.Time;
@@ -155,6 +158,12 @@ public static void setField(PreparedStatement upload, int type, Object field, in
                     case java.sql.Types.LONGVARBINARY:
                         upload.setBytes(index + 1, (byte[]) field);
                         break;
+                    case Types.CLOB:
+                    case Types.NCLOB:
+                        try (StringReader reader = new StringReader(field.toString())) {
+                            upload.setClob(index + 1, reader);
+                        }
+                        break;
                     default:
                         upload.setObject(index + 1, field);
                         LOG.warn("Unmanaged sql type ({}) for column {}. Best effort approach to set its value: {}.",
@@ -163,8 +172,7 @@ public static void setField(PreparedStatement upload, int type, Object field, in
                         // case java.sql.Types.ARRAY:
                         // case java.sql.Types.JAVA_OBJECT:
                         // case java.sql.Types.BLOB:
-                        // case java.sql.Types.CLOB:
-                        // case java.sql.Types.NCLOB:
+//                         case java.sql.Types.NCLOB:
                         // case java.sql.Types.DATALINK:
                         // case java.sql.Types.DISTINCT:
                         // case java.sql.Types.OTHER:
@@ -272,7 +280,11 @@ public static int[] buildSqlTypes(List<Class> fieldTypeArray) {
             } else if (fieldType.equals(Date.class.getName())) {
                 tmpFieldsType[i] = Types.DATE;
             } else if (fieldType.equals(Time.class.getName())) {
-				tmpFieldsType[i] = Types.TIME;
+                tmpFieldsType[i] = Types.TIME;
+            }else if (fieldType.equals(Clob.class.getName())){
+                tmpFieldsType[i] = Types.CLOB;
+            } else if (fieldType.equals(NClob.class.getName())) {
+                tmpFieldsType[i] = Types.NCLOB;
             } else {
                 throw new RuntimeException("no support field type for sql. the input type:" + fieldType);
             }