RdbAsyncReqRow convertDataType

Author: todd5167

rdb/rdb-side/src/main/java/com/dtstack/flink/sql/side/rdb/async/RdbAsyncReqRow.java

@@ -39,7 +39,9 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import java.math.BigDecimal;
 import java.sql.Timestamp;
+import java.time.Instant;
 import java.util.Date;
 import java.util.List;
 import java.util.Map;
@@ -87,7 +89,7 @@ public void asyncInvoke(Row input, ResultFuture<Row> resultFuture) throws Except
                 dealMissKey(inputRow, resultFuture);
                 return;
             }
-            inputParams.add(convertDateType(equalObj));
+            inputParams.add(convertDataType(equalObj));
         }
 
         String key = buildCacheKey(inputParams);
@@ -151,15 +153,43 @@ public void asyncInvoke(Row input, ResultFuture<Row> resultFuture) throws Except
         });
     }
 
-    private Object convertDateType(Object val) {
-        if (val instanceof Timestamp) {
+
+    private Object convertDataType(Object val) {
+        if (val == null) {
+            // OK
+        } else if (val instanceof Number && !(val instanceof BigDecimal)) {
+            // OK
+        } else if (val instanceof Boolean) {
+            // OK
+        } else if (val instanceof String) {
+            // OK
+        } else if (val instanceof Character) {
+            // OK
+        } else if (val instanceof CharSequence) {
+
+        } else if (val instanceof JsonObject) {
+
+        } else if (val instanceof JsonArray) {
+
+        } else if (val instanceof Map) {
+
+        } else if (val instanceof List) {
+
+        } else if (val instanceof byte[]) {
+
+        } else if (val instanceof Instant) {
+
+        } else if (val instanceof Timestamp) {
             val = DateUtil.getStringFromTimestamp((Timestamp) val);
-        } else if (val instanceof Date) {
+        } else if (val instanceof java.util.Date) {
             val = DateUtil.getStringFromDate((java.sql.Date) val);
+        } else {
+            val = val.toString();
         }
         return val;
     }
 
+
     protected List<Row> getRows(Row inputRow, List<JsonArray> cacheContent, List<JsonArray> results) {
         List<Row> rowList = Lists.newArrayList();
         for (JsonArray line : results) {