Flink 1.18连接PostGIS：UUID与Geography类型映射问题求助

场景与表结构 #

需连接部署PostGIS的PostgreSQL数据库，读取包含UUID和Geography类型的表，表结构如下：

CREATE TABLE area_of_interest (
    id uuid NOT NULL,
    geometry public.geography(multipolygon, 4326) NULL,
    CONSTRAINT "id_pk" PRIMARY KEY (id)
);

尝试过的无效方案 #

曾尝试多种表注册方式均失败，代码示例及对应问题如下：

TableDescriptor
    .forConnector("jdbc")
    .option(...)
    ...
    .schema(
        Schema.newBuilder()
            .column("id", DataTypes.STRING().notNull()) // 无效：ClassCastException，UUID无法转String
            // .column("id", DataTypes.RAW(UUID.class).notNull())  // 无效：PostgreSQL dialect不支持RAW类型
            .column("geometry", DataTypes.STRING().notNull()) // 无效：ClassCastException，PGobject无法转String
            // .column("geometry", DataTypes.RAW(Geometry.class))  // 无效：PostgreSQL dialect不支持RAW类型
            // .columnByExpression("geometry", "cast(geometry as varchar)") // 无效：Unknown identifier 'geometry'
            // .columnByExpression("geometry", "ST_asEWKT(geometry)") // 无效：Unknown identifier 'geometry'
            .primaryKey("id")
            .build())
    .build()

对应错误信息 #

• UUID映射为String时的错误：

java.lang.ClassCastException: class java.util.UUID cannot be cast to class java.lang.String (java.util.UUID and java.lang.String are in module java.base of loader 'bootstrap')
    at org.apache.flink.connector.jdbc.converter.AbstractJdbcRowConverter.lambda$createInternalConverter$224afae6$10(AbstractJdbcRowConverter.java:176) ~[flink-connector-jdbc-3.1.2-1.18.jar:3.1.2-1.18]
    at org.apache.flink.connector.jdbc.converter.AbstractJdbcRowConverter.lambda$wrapIntoNullableInternalConverter$ea5b8348$1(AbstractJdbcRowConverter.java:127) ~[flink-connector-jdbc-3.1.2-1.18.jar:3.1.2-1.18]
    at org.apache.flink.connector.jdbc.converter.AbstractJdbcRowConverter.toInternal(AbstractJdbcRowConverter.java:78) ~[flink-connector-jdbc-3.1.2-1.18.jar:3.1.2-1.18]
    at org.apache.flink.connector.jdbc.table.JdbcRowDataInputFormat.nextRecord(JdbcRowDataInputFormat.java:257) ~[flink-connector-jdbc-3.1.2-1.18.jar:3.1.2-1.18]
    at org.apache.flink.connector.jdbc.table.JdbcRowDataInputFormat.nextRecord(JdbcRowDataInputFormat.java:56) ~[flink-connector-jdbc-3.1.2-1.18.jar:3.1.2-1.18]

• Geography映射为String时的错误：

java.lang.ClassCastException: class org.postgresql.util.PGobject cannot be cast to class java.lang.String (org.postgresql.util.PGobject is in unnamed module of loader 'app'; java.lang.String is in module java.base of loader 'bootstrap')
    at org.apache.flink.connector.jdbc.converter.AbstractJdbcRowConverter.lambda$createInternalConverter$224afae6$10(AbstractJdbcRowConverter.java:176) ~[flink-connector-jdbc-3.1.2-1.18.jar:3.1.2-1.18]
    at org.apache.flink.connector.jdbc.converter.AbstractJdbcRowConverter.lambda$wrapIntoNullableInternalConverter$ea5b8348$1(AbstractJdbcRowConverter.java:127) ~[flink-connector-jdbc-3.1.2-1.18.jar:3.1.2-1.18]
    at org.apache.flink.connector.jdbc.converter.AbstractJdbcRowConverter.toInternal(AbstractJdbcRowConverter.java:78) ~[flink-connector-jdbc-3.1.2-1.18.jar:3.1.2-1.18]
    at org.apache.flink.connector.jdbc.table.JdbcRowDataInputFormat.nextRecord(JdbcRowDataInputFormat.java:257) ~[flink-connector-jdbc-3.1.2-1.18.jar:3.1.2-1.18]
    at org.apache.flink.connector.jdbc.table.JdbcRowDataInputFormat.nextRecord(JdbcRowDataInputFormat.java:56) ~[flink-connector-jdbc-3.1.2-1.18.jar:3.1.2-1.18]
    at org.apache.flink.streaming.api.functions.source.InputFormatSourceFunction.run(InputFormatSourceFunction.java:97) ~[flink-streaming-java-1.18.1.jar:1.18.1]

可行解决方案 #

方案1：查询阶段转换字段类型（简单直接） #

利用PostgreSQL/PostGIS的函数在读取时将特殊类型转换为Flink支持的字符串类型，通过scan.query指定查询语句：

TableDescriptor.forConnector("jdbc")
    .option("url", "jdbc:postgresql://your-host:5432/your-db")
    .option("username", "your-username")
    .option("password", "your-password")
    .option("scan.query", "SELECT id::varchar, ST_AsText(geometry) AS geometry FROM area_of_interest")
    .schema(Schema.newBuilder()
        .column("id", DataTypes.STRING().notNull())
        .column("geometry", DataTypes.STRING())
        .primaryKey("id")
        .build())
    .build();

• UUID处理：用id::varchar将UUID类型直接转换为字符串，Flink侧映射为DataTypes.STRING()即可正常读取。
• Geography处理：用ST_AsText(geometry)将Geography类型转换为WKT格式字符串，Flink侧同样映射为DataTypes.STRING()；若需带空间参考的格式，可改用ST_AsEWKT(geometry)。

方案2：自定义JDBC转换器（复杂场景适用） #

若需在Flink侧直接处理UUID或Geometry对象，可自定义JdbcRowConverter扩展PostgreSQL的转换器：

1. UUID转换器：继承PostgresRowConverter，重写toInternal方法，将JDBC返回的UUID对象转换为Flink的StringData。
2. Geography转换器：引入postgis-jdbc依赖，将PGobject解析为JTS的Geometry对象，Flink侧使用DataTypes.RAW(Geometry.class)，同时需自定义PostgreSQL dialect支持该类型。

注：方案2实现复杂度较高，推荐优先使用方案1快速解决问题。

内容的提问来源于stack exchange，提问作者Glauber Dantas