Postgres data type mapping in stream processing #5410

divbok · 2025-02-04T19:36:14Z

Description

Adds the data type mapping support for Postgres stream processing

Issues Resolved

Contributes to #5309

Check List

New functionality includes testing.
New functionality has a documentation issue. Please link to it in this PR.
- New functionality has javadoc added
Commits are signed with a real name per the DCO

By submitting this pull request, I confirm that my contribution is made under the terms of the Apache 2.0 license.
For more information on following Developer Certificate of Origin and signing off your commits, please check here.

san81 · 2025-02-05T19:57:19Z

...ource/src/main/java/org/opensearch/dataprepper/plugins/source/rds/export/DataFileLoader.java

@@ -176,7 +176,7 @@ public void run() {
    private void transformEvent(final Event event, final String fullTableName) {
        Map<String, String> columnDataTypeMap = dbTableMetadata.getTableColumnDataTypeMap().get(fullTableName);
        for (Map.Entry<String, Object> entry : event.toMap().entrySet()) {
-            final Object data = DataTypeHelper.getDataByColumnType(MySQLDataType.byDataType(columnDataTypeMap.get(entry.getKey())), entry.getKey(),
+            final Object data = MySQLDataTypeHelper.getDataByColumnType(MySQLDataType.byDataType(columnDataTypeMap.get(entry.getKey())), entry.getKey(),


This class looks like generic but you are using MysqlDataTypeHelper here. Is this class specific to Mysql rds?

It's not specific for MySQL. Will add Postgres data type mapping here for export in a separate PR. This PR is for stream part.

san81 · 2025-02-05T20:03:25Z

...-source/src/main/java/org/opensearch/dataprepper/plugins/source/rds/model/TableMetadata.java

@@ -23,11 +24,21 @@ public TableMetadata(String tableName, String databaseName, List<String> columnN
        this(tableName, databaseName, columnNames, primaryKeys, Collections.emptyMap(), Collections.emptyMap());
    }

+    public TableMetadata(String tableName, String databaseName, List<String> columnNames,  List<String> columnTypes, List<String> primaryKeys) {


Probably a builder pattern fits better here instead of multiple overloaded constructors. Just an idea. Feel free to ignore it if you think it doesn't add much value here.

Made the change

san81 · 2025-02-05T20:08:17Z

...-source/src/main/java/org/opensearch/dataprepper/plugins/source/rds/resync/ResyncWorker.java

@@ -158,7 +158,7 @@ private Map<String, Object> mapDataType(final Map<String, Object> rowData, final
        Map<String, String> columnDataTypeMap = dbTableMetadata.getTableColumnDataTypeMap().get(fullTableName);
        Map<String, Object> rowDataAfterMapping = new HashMap<>();
        for (Map.Entry<String, Object> entry : rowData.entrySet()) {
-            final Object data = DataTypeHelper.getDataByColumnType(MySQLDataType.byDataType(columnDataTypeMap.get(entry.getKey())), entry.getKey(),
+            final Object data = MySQLDataTypeHelper.getDataByColumnType(MySQLDataType.byDataType(columnDataTypeMap.get(entry.getKey())), entry.getKey(),


It may be my understanding of the overall approach here but I got the same question here. This class looks like generic but you are using MysqlDataTypeHelper here. Is this class specific to Mysql rds?

This ResyncWorker is specific for MySQL

Then may be using a better name (or better package to segregate things related to Mysql under one) - will make this obvious.

Let's add MySql to the class name.

oeyh

Thanks for working on this! A few comments below

oeyh · 2025-02-06T20:36:39Z

.../opensearch/dataprepper/plugins/source/rds/datatype/postgres/handler/NumericTypeHandler.java

+            case DOUBLE_PRECISION:
+                return Double.parseDouble(textValue);
+            case NUMERIC:
+            case MONEY:


Is it possible to convert Money to an object (map) with two fields currency and amount? I think that's a better representation.

oeyh · 2025-02-06T20:52:57Z

.../opensearch/dataprepper/plugins/source/rds/datatype/postgres/handler/SpatialTypeHandler.java

+
+        // Case 1: Solve for y when x = 0
+        double x1 = 0.0;
+        double y1 = c == 0 ? 0 : -c / b;


What if b == 0?

a and b will never be zero as stated in the docs.
https://www.postgresql.org/docs/16/datatype-geometric.html#DATATYPE-LINE

It says a and b will not be both zero at the same time.

Lines are represented by the linear equation Ax + By + C = 0, where A and B are not both zero. Am I inferring this wrong? If so, I can change my approach

A and B are not both zero

I think this means A and B are not zero at the same time. B can still be zero if A is not and similarly the other way around. So you can still. have A = 0 so By + C = 0, or B = 0 so Ax + C = 0.

oeyh · 2025-02-06T21:51:26Z

...-source/src/main/java/org/opensearch/dataprepper/plugins/source/rds/resync/ResyncWorker.java

@@ -158,7 +158,7 @@ private Map<String, Object> mapDataType(final Map<String, Object> rowData, final
        Map<String, String> columnDataTypeMap = dbTableMetadata.getTableColumnDataTypeMap().get(fullTableName);
        Map<String, Object> rowDataAfterMapping = new HashMap<>();
        for (Map.Entry<String, Object> entry : rowData.entrySet()) {
-            final Object data = DataTypeHelper.getDataByColumnType(MySQLDataType.byDataType(columnDataTypeMap.get(entry.getKey())), entry.getKey(),
+            final Object data = MySQLDataTypeHelper.getDataByColumnType(MySQLDataType.byDataType(columnDataTypeMap.get(entry.getKey())), entry.getKey(),


Let's add MySql to the class name.

Signed-off-by: Hai Yan <oeyh@amazon.com>

Data type mapping for postgres Data type mapping for postgres stream Data type mapping for postgres stream Data type mapping for postgres stream Data type mapping for postgres stream Data type mapping for postgres stream Signed-off-by: Divyansh Bokadia <dbokadia@amazon.com>

Signed-off-by: Divyansh Bokadia <dbokadia@amazon.com>

oeyh · 2025-02-10T18:15:17Z

...arch/dataprepper/plugins/source/rds/datatype/postgres/handler/NetworkAddressTypeHandler.java

+        if (!columnType.isNetworkAddress()) {
+            throw new IllegalArgumentException("ColumnType is not Network Address: " + columnType);
+        }
+        return value.toString();


I thought we are converting Json to a map so that it will be an object in OpenSearch documents. But on the other hand, users can configure parse_json processor to parse it if necessary. Is it consistent with how we handle json in MySQL data type mapping?

I'm referring to the JsonTypeHandler. Somehow added at the wrong place.

Yes in MySQL data type mapping it is left as a json string, it is not being converted to an object.

oeyh · 2025-02-10T18:15:42Z

.../opensearch/dataprepper/plugins/source/rds/datatype/postgres/handler/SpatialTypeHandler.java

+        }
+    }
+
+    private PGpoint[] getPointsOnLine(PGline line) {


Thanks for making these changes.

divbok requested review from sb2k16, chenqi0805, engechas, san81, srikanthjg, graytaylor0, dinujoh, kkondaka, KarstenSchnitter, dlvenable and oeyh as code owners February 4, 2025 19:36

divbok force-pushed the postgres_data_type_mapping branch 2 times, most recently from 441070f to 3d58a1d Compare February 4, 2025 20:18

oeyh changed the title ~~Postgres data type mapping~~ Postgres data type mapping in stream processing Feb 4, 2025

san81 reviewed Feb 5, 2025

View reviewed changes

divbok force-pushed the postgres_data_type_mapping branch from 3d0d83d to 1f0cb70 Compare February 6, 2025 17:41

oeyh reviewed Feb 6, 2025

View reviewed changes

divbok force-pushed the postgres_data_type_mapping branch 3 times, most recently from e0ecc07 to fa7c849 Compare February 9, 2025 19:28

oeyh and others added 3 commits February 10, 2025 10:33

Add more tests

448bb2a

Signed-off-by: Hai Yan <oeyh@amazon.com>

Data type mapping for postgres stream

77a1bc3

Signed-off-by: Divyansh Bokadia <dbokadia@amazon.com>

divbok force-pushed the postgres_data_type_mapping branch from fa7c849 to 77a1bc3 Compare February 10, 2025 16:36

oeyh reviewed Feb 10, 2025

View reviewed changes

oeyh approved these changes Feb 12, 2025

View reviewed changes

san81 approved these changes Feb 12, 2025

View reviewed changes

oeyh merged commit 1a41705 into opensearch-project:main Feb 12, 2025
46 of 47 checks passed

oeyh mentioned this pull request Feb 12, 2025

Postgres schema name fix #5432

Merged

4 tasks

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Postgres data type mapping in stream processing #5410

Postgres data type mapping in stream processing #5410

divbok commented Feb 4, 2025 •

edited

Loading

san81 Feb 5, 2025

oeyh Feb 5, 2025

san81 Feb 5, 2025

divbok Feb 6, 2025

san81 Feb 5, 2025

oeyh Feb 5, 2025

san81 Feb 5, 2025

oeyh Feb 6, 2025

oeyh left a comment

oeyh Feb 6, 2025

divbok Feb 6, 2025

oeyh Feb 6, 2025

divbok Feb 6, 2025

oeyh Feb 6, 2025

divbok Feb 7, 2025

oeyh Feb 7, 2025 •

edited

Loading

oeyh Feb 6, 2025

oeyh Feb 10, 2025

oeyh Feb 10, 2025

divbok Feb 12, 2025

oeyh Feb 10, 2025

Postgres data type mapping in stream processing #5410

Postgres data type mapping in stream processing #5410

Conversation

divbok commented Feb 4, 2025 • edited Loading

Description

Issues Resolved

Check List

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

oeyh left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

oeyh Feb 7, 2025 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

divbok commented Feb 4, 2025 •

edited

Loading

oeyh Feb 7, 2025 •

edited

Loading