[fix]Fix a mismatch between the type of jdbc query result and the dat…

…atype
apache · Nov 12, 2024 · 3a2d2cf · 3a2d2cf
1 parent c23abbb
commit 3a2d2cf
Showing 1 changed file with 5 additions and 2 deletions.
diff --git a/flink-doris-connector/src/main/java/org/apache/doris/flink/lookup/Worker.java b/flink-doris-connector/src/main/java/org/apache/doris/flink/lookup/Worker.java
@@ -24,6 +24,7 @@
 import org.apache.doris.flink.connection.JdbcConnectionProvider;
 import org.apache.doris.flink.connection.SimpleJdbcConnectionProvider;
 import org.apache.doris.flink.exception.DorisRuntimeException;
+import org.apache.flink.table.types.DataType;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -200,8 +201,10 @@ private Map<RecordKey, List<Record>> executeQuery(
                     try (ResultSet rs = ps.executeQuery()) {
                         while (rs.next()) {
                             Record record = new Record(schema);
-                            for (int index = 0; index < schema.getFieldTypes().length; index++) {
-                                record.setObject(index, rs.getObject(index + 1));
+                            DataType[] fieldTypes = schema.getFieldTypes();
+                            for (int index = 0; index < fieldTypes.length; index++) {
+                                Class<?> conversionClass = fieldTypes[index].getConversionClass();
+                                record.setObject(index, rs.getObject(index + 1, conversionClass));
                             }
                             List<Record> records =
                                     resultRecordMap.computeIfAbsent(