[cdc-connector][postgres] apply DBZ-5470 to fix lose precision in Pos…

…tgresParallelSource
wuzhenhua01 · Feb 5, 2024 · 170e442 · 170e442
1 parent 167bb40
commit 170e442
Show file tree

Hide file tree

Showing 2 changed files with 3 additions and 7 deletions.
diff --git a/...c/main/java/com/ververica/cdc/connectors/postgres/source/fetch/PostgresScanFetchTask.java b/...c/main/java/com/ververica/cdc/connectors/postgres/source/fetch/PostgresScanFetchTask.java
@@ -309,10 +309,8 @@ private void createDataEventsForTable(
 
                 while (rs.next()) {
                     rows++;
-                    final Object[] row = new Object[columnArray.getGreatestColumnPosition()];
-                    for (int i = 0; i < columnArray.getColumns().length; i++) {
-                        row[columnArray.getColumns()[i].position() - 1] = rs.getObject(i + 1);
-                    }
+                    final Object[] row = jdbcConnection.rowToArray(table, null, rs, columnArray);
+
                     if (logTimer.expired()) {
                         long stop = clock.currentTimeInMillis();
                         LOG.info(

diff --git a/...res-cdc/src/main/java/io/debezium/connector/postgresql/connection/PostgresConnection.java b/...res-cdc/src/main/java/io/debezium/connector/postgresql/connection/PostgresConnection.java
@@ -12,7 +12,6 @@
 import io.debezium.config.Configuration;
 import io.debezium.connector.postgresql.PgOid;
 import io.debezium.connector.postgresql.PostgresConnectorConfig;
-import io.debezium.connector.postgresql.PostgresSchema;
 import io.debezium.connector.postgresql.PostgresType;
 import io.debezium.connector.postgresql.PostgresValueConverter;
 import io.debezium.connector.postgresql.TypeRegistry;
@@ -767,8 +766,7 @@ public <T extends DatabaseSchema<TableId>> Object getColumnValue(
         try {
             final ResultSetMetaData metaData = rs.getMetaData();
             final String columnTypeName = metaData.getColumnTypeName(columnIndex);
-            final PostgresType type =
-                    ((PostgresSchema) schema).getTypeRegistry().get(columnTypeName);
+            final PostgresType type = getTypeRegistry().get(columnTypeName);
 
             LOGGER.trace("Type of incoming data is: {}", type.getOid());
             LOGGER.trace("ColumnTypeName is: {}", columnTypeName);