[Fix-499] [connector] Flink oracle connector can't cast CLOB to String

dlink-connectors/dlink-connector-jdbc-1.11/src/main/java/org/apache/flink/connector/jdbc/internal/converter/OracleRowConverter.java

@@ -18,6 +18,8 @@
 
 package org.apache.flink.connector.jdbc.internal.converter;
 
+import oracle.sql.CLOB;
+
 import org.apache.flink.table.data.DecimalData;
 import org.apache.flink.table.data.StringData;
 import org.apache.flink.table.data.TimestampData;
@@ -101,7 +103,14 @@ public class OracleRowConverter extends AbstractJdbcRowConverter {
                 };
             case CHAR:
             case VARCHAR:
-                return val -> StringData.fromString((String) val);
+                return (val) -> {
+                    if(val instanceof CLOB){
+                        CLOB clob = (CLOB) val;
+                        return  StringData.fromString(clob == null ? null : clob.stringValue());
+                    }else {
+                        return StringData.fromString((String) val);
+                    }
+                };
             case BINARY:
             case VARBINARY:
                 return val -> (byte[]) val;

dlink-connectors/dlink-connector-jdbc-1.12/src/main/java/org/apache/flink/connector/jdbc/internal/converter/OracleRowConverter.java

@@ -18,6 +18,8 @@
 
 package org.apache.flink.connector.jdbc.internal.converter;
 
+import oracle.sql.CLOB;
+
 import org.apache.flink.table.data.DecimalData;
 import org.apache.flink.table.data.StringData;
 import org.apache.flink.table.data.TimestampData;
@@ -101,7 +103,14 @@ public class OracleRowConverter extends AbstractJdbcRowConverter {
                 };
             case CHAR:
             case VARCHAR:
-                return val -> StringData.fromString((String) val);
+                return (val) -> {
+                    if(val instanceof CLOB){
+                        CLOB clob = (CLOB) val;
+                        return  StringData.fromString(clob == null ? null : clob.stringValue());
+                    }else {
+                        return StringData.fromString((String) val);
+                    }
+                };
             case BINARY:
             case VARBINARY:
                 return val -> (byte[]) val;

dlink-connectors/dlink-connector-jdbc-1.13/src/main/java/org/apache/flink/connector/jdbc/internal/converter/OracleRowConverter.java

 package org.apache.flink.connector.jdbc.internal.converter;
 
+import oracle.sql.CLOB;
+
 import org.apache.flink.table.data.DecimalData;
 import org.apache.flink.table.data.StringData;
 import org.apache.flink.table.data.TimestampData;
@@ -83,7 +85,14 @@ public class OracleRowConverter extends AbstractJdbcRowConverter {
                 };
             case CHAR:
             case VARCHAR:
-                return val -> StringData.fromString((String) val);
+                return (val) -> {
+                    if(val instanceof CLOB){
+                        CLOB clob = (CLOB) val;
+                        return  StringData.fromString(clob == null ? null : clob.stringValue());
+                    }else {
+                        return StringData.fromString((String) val);
+                    }
+                };
             case BINARY:
             case VARBINARY:
                 return val -> (byte[]) val;

dlink-connectors/dlink-connector-jdbc-1.14/src/main/java/org/apache/flink/connector/jdbc/internal/converter/OracleRowConverter.java

 package org.apache.flink.connector.jdbc.internal.converter;
 
+import oracle.sql.CLOB;
+
 import org.apache.flink.table.data.DecimalData;
 import org.apache.flink.table.data.StringData;
 import org.apache.flink.table.data.TimestampData;
@@ -83,7 +85,14 @@ public class OracleRowConverter extends AbstractJdbcRowConverter {
                 };
             case CHAR:
             case VARCHAR:
-                return val -> StringData.fromString((String) val);
+                return (val) -> {
+                    if(val instanceof CLOB){
+                        CLOB clob = (CLOB) val;
+                        return  StringData.fromString(clob == null ? null : clob.stringValue());
+                    }else {
+                        return StringData.fromString((String) val);
+                    }
+                };
             case BINARY:
             case VARBINARY:
                 return val -> val;