Support reading json and jsonb types in PostgreSQL dialect

Author: grzegorz8

flink-connector-jdbc-postgres/src/main/java/org/apache/flink/connector/jdbc/postgres/database/catalog/PostgresTypeMapper.java

@@ -84,6 +84,8 @@ public class PostgresTypeMapper implements JdbcCatalogTypeMapper {
     private static final String PG_CHARACTER_ARRAY = "_character";
     private static final String PG_CHARACTER_VARYING = "varchar";
     private static final String PG_CHARACTER_VARYING_ARRAY = "_varchar";
+    private static final String PG_JSON = "json";
+    private static final String PG_JSONB = "jsonb";
 
     @Override
     public DataType mapping(ObjectPath tablePath, ResultSetMetaData metadata, int colIndex)
@@ -157,6 +159,8 @@ protected DataType getMapping(String pgType, int precision, int scale) {
             case PG_CHARACTER_VARYING_ARRAY:
                 return DataTypes.ARRAY(DataTypes.VARCHAR(precision));
             case PG_TEXT:
+            case PG_JSON:
+            case PG_JSONB:
                 return DataTypes.STRING();
             case PG_TEXT_ARRAY:
                 return DataTypes.ARRAY(DataTypes.STRING());

flink-connector-jdbc-postgres/src/main/java/org/apache/flink/connector/jdbc/postgres/database/dialect/PostgresDialectConverter.java

@@ -21,11 +21,13 @@
 import org.apache.flink.annotation.Internal;
 import org.apache.flink.connector.jdbc.core.database.dialect.AbstractDialectConverter;
 import org.apache.flink.table.data.GenericArrayData;
+import org.apache.flink.table.data.StringData;
 import org.apache.flink.table.types.logical.ArrayType;
 import org.apache.flink.table.types.logical.LogicalType;
 import org.apache.flink.table.types.logical.LogicalTypeRoot;
 import org.apache.flink.table.types.logical.RowType;
 import org.apache.flink.table.types.logical.utils.LogicalTypeUtils;
+import org.postgresql.util.PGobject;
 
 import java.lang.reflect.Array;
 
@@ -91,7 +93,19 @@ private JdbcDeserializationConverter createPostgresArrayConverter(ArrayType arra
     // Have its own method so that Postgres can support primitives that super class doesn't support
     // in the future
     private JdbcDeserializationConverter createPrimitiveConverter(LogicalType type) {
-        return super.createInternalConverter(type);
+        switch (type.getTypeRoot()) {
+            case VARCHAR:
+                return val -> {
+                    if (val instanceof PGobject) {
+                        PGobject obj = (PGobject) val;
+                        return StringData.fromString(obj.getValue());
+                    } else {
+                        return StringData.fromString((String) val);
+                    }
+                };
+            default:
+                return super.createInternalConverter(type);
+        }
     }
 
     @Override