[Bug] [Module Name] pgsql is synchronized to doris, and the update operation cannot read before, resulting in a null pointer exception.

[qzztf]

Search before asking

• I had searched in the issues and found no similar issues.

What happened

尝试使用flink sql local模式同步pgsql的数据到doris, 当在pgsql中修改数据时，在AbstractSqlSinkBuilder类以下代码发生空指针异常
rowCollect( columnNameList, columnTypeList, out, RowKind.UPDATE_BEFORE, (Map) value.get("before"))
将该行代码注释后可以正常同步。

What you expected to happen

如果pgsql中的表没有设置REPLICA IDENTITY为FULL时，将读取不到before的数据，因此这里可能需要兼容一下。

How to reproduce

EXECUTE CDCSOURCE demo_doris WITH ( 'connector' = 'postgres-cdc', 'hostname' = '192.168.0.37', 'port' = '5432', 'database-name' = 'db', 'schema-name' = 'public', 'username' = '111111', 'password' = '111111', 'checkpoint' = '10000', 'scan.startup.mode' = 'initial', 'parallelism' = '1', 'table-name' = 'public\.i_.*', 'debezium.decimal.handling.mode'='string', 'source.decoding.plugin.name' = 'pgoutput', 'sink.connector' = 'doris', 'sink.fenodes' = '192.168.0.223:8030', 'sink.username' = '1111111', 'sink.password' = '1111111', 'sink.doris.batch.size' = '1000', 'sink.sink.max-retries' = '1', 'sink.sink.db' = 'ods_pg', 'sink.sink.properties.format' ='json', 'sink.sink.properties.read_json_by_line' ='true', 'sink.table.identifier' = '#{schemaName}.#{tableName}', 'sink.sink.label-prefix' = '#{schemaName}_#{tableName}_5' );
这里我的flink sql，pgsql需要做一些设置支持逻辑复制。

Anything else

No response

Version

1.1.0

Are you willing to submit PR?

• Yes I am willing to submit a PR!

Code of Conduct

• I agree to follow this project's Code of Conduct

[aiwenmo]

#3935
thx. I have committed into #3933

[aiwenmo]

Please use dev branch to preview CDCSOURCE's sink.