Add fixes for CrateDB

Author: grzegorz8

flink-connector-jdbc-cratedb/src/main/java/org/apache/flink/connector/jdbc/cratedb/database/catalog/CrateDBTypeMapper.java

@@ -47,6 +47,10 @@ protected DataType getMapping(String pgType, int precision, int scale) {
                 return DataTypes.STRING();
             case PG_STRING_ARRAY:
                 return DataTypes.ARRAY(DataTypes.STRING());
+            case PG_JSON: // CrateDB supports but not supported in the dialect yet.
+                return null;
+            case PG_JSONB: // CrateDB supports JSON type, but not JSONB.
+                return null;
             default:
                 return super.getMapping(pgType, precision, scale);
         }

flink-connector-jdbc-cratedb/src/main/java/org/apache/flink/connector/jdbc/cratedb/database/dialect/CrateDBDialectConverter.java

@@ -20,6 +20,7 @@
 
 import org.apache.flink.annotation.Internal;
 import org.apache.flink.connector.jdbc.postgres.database.dialect.CompatiblePostgresDialectConverter;
+import org.apache.flink.table.data.StringData;
 import org.apache.flink.table.types.logical.RowType;
 
 /**
@@ -39,4 +40,9 @@ public CrateDBDialectConverter(RowType rowType) {
     public String compatibleConverterName() {
         return "CrateDB";
     }
+
+    @Override
+    protected JdbcDeserializationConverter createVarcharConverter() {
+        return val -> StringData.fromString((String) val);
+    }
 }

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

@@ -84,8 +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";
+    protected static final String PG_JSON = "json";
+    protected static final String PG_JSONB = "jsonb";
 
     @Override
     public DataType mapping(ObjectPath tablePath, ResultSetMetaData metadata, int colIndex)

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

@@ -96,19 +96,23 @@ private JdbcDeserializationConverter createPostgresArrayConverter(ArrayType arra
     private JdbcDeserializationConverter createPrimitiveConverter(LogicalType 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);
-                    }
-                };
+                return createVarcharConverter();
             default:
                 return super.createInternalConverter(type);
         }
     }
 
+    protected JdbcDeserializationConverter createVarcharConverter() {
+        return val -> {
+            if (val instanceof PGobject) {
+                PGobject obj = (PGobject) val;
+                return StringData.fromString(obj.getValue());
+            } else {
+                return StringData.fromString((String) val);
+            }
+        };
+    }
+
     @Override
     public String converterName() {
         return "PostgreSQL";