Skip to content

D
dlink
Unverified Commit f7f7810f authored by aiwenmo's avatar aiwenmo Committed by GitHub
Browse files
Options

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

[Fix-499] [connector] Flink oracle connector can't cast CLOB to String
parents 44083ff6 8fcf21fd
Hide whitespace changes
Inline Side-by-side
Showing with 40 additions and 4 deletions
dlink-connectors/dlink-connector-jdbc-1.11/src/main/java/org/apache/flink/connector/jdbc/internal/converter/OracleRowConverter.java
View file @ f7f7810f
......@@ -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
View file @ f7f7810f
......@@ -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
View file @ f7f7810f
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
View file @ f7f7810f
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;
......
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment