Data type mapping for postgres export #5438

divbok · 2025-02-17T16:40:34Z

Data type mapping for postgres export

Description

Added support for data type mapping for postgres export.
Added conversion to hexadecimal representation for binary data type for uniformity between rds stream and export.
Refactored the numeric and boolean data type names in postgres to represent its internal names.

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.

oeyh

Looks like only Binary and Boolean type handler needed changes. Were you able to verify that for other types, the output between export and stream are consistent?

oeyh · 2025-02-17T20:45:26Z

...ugins/rds-source/src/main/java/org/opensearch/dataprepper/plugins/source/rds/RdsService.java

        if (sourceConfig.getEngine() == EngineType.MYSQL) {
-            final Map<String, Map<String, String>> tableColumnDataTypeMap = getColumnDataTypeMap(
+            final Map<String, Map<String, String>> tableColumnDataTypeMap = getMySQLColumnDataTypeMap(
                    (MySqlSchemaManager) schemaManager);
            dbTableMetadata  = new DbTableMetadata(dbMetadata, tableColumnDataTypeMap);
        } else {
-            dbTableMetadata = new DbTableMetadata(dbMetadata, Collections.emptyMap());
+            final Map<String, Map<String, String>> tableColumnDataTypeMap = getPostgresColumnDataTypeMap(
+                    (PostgresSchemaManager) schemaManager);
+            dbTableMetadata = new DbTableMetadata(dbMetadata, tableColumnDataTypeMap);


This if-else block can be moved to a method, so here we only have:

DbTableMetadata dbTableMetadata = getDbTableMetadata(dbMetadata);

oeyh · 2025-02-18T17:51:56Z

...rc/main/java/org/opensearch/dataprepper/plugins/source/rds/schema/PostgresSchemaManager.java

@@ -117,6 +122,38 @@ public List<String> getPrimaryKeys(final String fullTableName) {
        throw new RuntimeException("Failed to get primary keys for table " + fullTableName);
    }

+
+    public Map<String, String> getColumnDataTypes(final String fullTableName) {


Can you add this method declaration to the SchemaManager interface? This way, we may be able to get rid of the if-else condition in RdsService class where this method is used.

oeyh · 2025-02-18T17:55:17Z

...ugins/rds-source/src/main/java/org/opensearch/dataprepper/plugins/source/rds/RdsService.java

+        return sourceConfig.getTableNames().stream()
+                .collect(Collectors.toMap(
+                        fullTableName -> fullTableName,
+                        fullTableName -> schemaManager.getColumnDataTypes(fullTableName)


If we put getColumnDataTypes into SchemaManager interface, I think we can combine this method with the corresponding MySQL method and get rid of the if-else conditional.

Made the changes

divbok · 2025-02-18T19:22:28Z

Yes, I verified the consistency between stream and export, only binary and boolean had changes.

oeyh · 2025-02-18T19:56:36Z

...rc/main/java/org/opensearch/dataprepper/plugins/source/rds/schema/PostgresSchemaManager.java

+            try (Connection connection = connectionManager.getConnection()) {
+                final DatabaseMetaData metaData = connection.getMetaData();
+                // Retrieve column metadata
+                try (ResultSet columns = metaData.getColumns(database, null, table, null)) {


schema should be used here:

Suggested change

try (ResultSet columns = metaData.getColumns(database, null, table, null)) {

try (ResultSet columns = metaData.getColumns(database, schema, table, null)) {

oeyh · 2025-02-18T19:57:45Z

...rc/main/java/org/opensearch/dataprepper/plugins/source/rds/schema/PostgresSchemaManager.java

+                }
+                return columnsToDataType;
+            } catch (final Exception e) {
+                LOG.error("Failed to get dataTypes for database {} table {}, retrying", database, table, e);


It would be better to Include schema name in the error message as well.

oeyh · 2025-02-18T19:58:12Z

...rc/main/java/org/opensearch/dataprepper/plugins/source/rds/schema/PostgresSchemaManager.java

+            } catch (final Exception e) {
+                LOG.error("Failed to get dataTypes for database {} table {}, retrying", database, table, e);
+                if (retry == NUM_OF_RETRIES) {
+                    throw new RuntimeException(String.format("Failed to get dataTypes for database %s table %s after " +


Similarly here, include schema name in the error message as well.

oeyh · 2025-02-18T20:16:38Z

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

@@ -184,5 +186,13 @@ private void transformEvent(final Event event, final String fullTableName, final
                event.put(entry.getKey(), data);
            }
        }
+        if (engineType == EngineType.POSTGRES) {


Can you remove the TODO line around Line 180?

oeyh · 2025-02-18T20:27:24Z

...g/opensearch/dataprepper/plugins/source/rds/datatype/postgres/handler/BinaryTypeHandler.java

+        if (value instanceof Map) {
+            Object data = ((Map<?, ?>)value).get(BYTES_KEY);
+            byte[] bytes = ((String) data).getBytes(StandardCharsets.ISO_8859_1);
+            String hexString = "\\x" + bytesToHex(bytes);


Nit: return it directly, no need to create a variable.

oeyh · 2025-02-18T20:27:37Z

...g/opensearch/dataprepper/plugins/source/rds/datatype/postgres/handler/BinaryTypeHandler.java

        return value.toString();
    }
+
+    private static String bytesToHex(byte[] bytes) {


Why static?

san81 · 2025-02-18T22:19:57Z

...ugins/rds-source/src/main/java/org/opensearch/dataprepper/plugins/source/rds/RdsService.java

+        return new DbTableMetadata(dbMetadata, tableColumnDataTypeMap);
+    }
+
+    private Map<String, Map<String, String>> getColumnDataTypeMap(final SchemaManager schemaManager) {


nit: I think these method could be inside the config object itself.

san81 · 2025-02-18T22:23:44Z

...g/opensearch/dataprepper/plugins/source/rds/datatype/postgres/handler/BinaryTypeHandler.java

        return value.toString();
    }
+
+    private String bytesToHex(byte[] bytes) {


private methods makes it hard to unit test them. This looks like a static utility that can live in a utility class that enables better unit testing of the logic inside this method.

This method won't be used across classes mostly. Shouldn't we keep it here as it has more relevance?

Added a separate util class

Signed-off-by: Divyansh Bokadia <[email protected]> Signed-off-by: Divyansh Bokadia <[email protected]> Data type mapping for postgres export Signed-off-by: Divyansh Bokadia <[email protected]> Data type mapping for postgres export Data type mapping for postgres export

divbok requested review from sb2k16, chenqi0805, engechas, san81, srikanthjg, graytaylor0, dinujoh, kkondaka, KarstenSchnitter, dlvenable and oeyh as code owners February 17, 2025 16:40

divbok force-pushed the postgres_data_mapping branch from 1520ec1 to 958c4de Compare February 17, 2025 17:00

oeyh reviewed Feb 18, 2025

View reviewed changes

divbok force-pushed the postgres_data_mapping branch from 958c4de to 8cc0e79 Compare February 18, 2025 19:20

oeyh reviewed Feb 18, 2025

View reviewed changes

divbok force-pushed the postgres_data_mapping branch from 8cc0e79 to 52b108b Compare February 18, 2025 20:51

oeyh previously approved these changes Feb 18, 2025

View reviewed changes

san81 reviewed Feb 18, 2025

View reviewed changes

divbok dismissed oeyh’s stale review via 531f92f February 19, 2025 15:07

divbok force-pushed the postgres_data_mapping branch from 52b108b to 531f92f Compare February 19, 2025 15:07

san81 approved these changes Feb 19, 2025

View reviewed changes

oeyh approved these changes Feb 19, 2025

View reviewed changes

oeyh merged commit eba918b into opensearch-project:main Feb 19, 2025
46 of 47 checks passed

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Data type mapping for postgres export #5438

Data type mapping for postgres export #5438

divbok commented Feb 17, 2025

oeyh left a comment

oeyh Feb 17, 2025

oeyh Feb 18, 2025

divbok Feb 18, 2025

oeyh Feb 18, 2025 •

edited

Loading

divbok Feb 18, 2025

divbok commented Feb 18, 2025

oeyh Feb 18, 2025

oeyh Feb 18, 2025

oeyh Feb 18, 2025

oeyh Feb 18, 2025

oeyh Feb 18, 2025

oeyh Feb 18, 2025

divbok Feb 18, 2025

san81 Feb 18, 2025

san81 Feb 18, 2025 •

edited

Loading

divbok Feb 18, 2025

divbok Feb 19, 2025

	try (ResultSet columns = metaData.getColumns(database, null, table, null)) {
	try (ResultSet columns = metaData.getColumns(database, schema, table, null)) {

Data type mapping for postgres export #5438

Data type mapping for postgres export #5438

Conversation

divbok commented Feb 17, 2025

Description

Issues Resolved

Check List

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

oeyh Feb 18, 2025 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

divbok commented Feb 18, 2025

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

san81 Feb 18, 2025 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

oeyh Feb 18, 2025 •

edited

Loading

san81 Feb 18, 2025 •

edited

Loading