Adding enum and range datatype support for postgres #5458

divbok · 2025-02-24T16:43:06Z

Description

Adding enum and range datatype support for Postgres.
As enum does not have a fixed type_id, the enum columns are first fetched from the database and stored in Postgres Stream State and then the data type mapping is done. The enum type is not stored as user-defined enum type, instead we overwrite it to enum while storing.
Range datatype has been added without including numeric range data type and multirange. Opensource does not support the high precision and scale of numeric postgres data type hence the mapping cannot be done here.
No corresponding multirange data type is supported in Opensource.

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.

Signed-off-by: Divyansh Bokadia <dbokadia@amazon.com> Adding enum and range datatype support for postgres Adding enum and range datatype support for postgres

dlvenable · 2025-02-24T17:04:58Z

...rc/main/java/org/opensearch/dataprepper/plugins/source/rds/datatype/postgres/ColumnType.java

@@ -49,11 +50,19 @@ public enum ColumnType {
    XML(142, "xml"),
    UUID(2950, "uuid"),
    PG_LSN(3220, "pg_lsn"),
+    PG_SNAPSHOT(5038, "pg_snapshot"),


Beyond this PR, why do we have this enum mapping? Is this not provided by the Posgres library itself?

There is no such library which provides the type_id of these data types.

dlvenable · 2025-02-24T17:07:02Z

...va/org/opensearch/dataprepper/plugins/source/rds/coordination/state/PostgresStreamState.java

@@ -22,6 +25,9 @@ public class PostgresStreamState {
    @JsonProperty("replicationSlotName")
    private String replicationSlotName;

+    @JsonProperty("enumColumnsMap")


Can we have a clearer name here? What exactly is this storing?

It stores the table name against the set of all enum columns. That is why i went with this naming convention.

dlvenable · 2025-02-24T17:08:08Z

.../org/opensearch/dataprepper/plugins/source/rds/datatype/postgres/PostgresDataTypeHelper.java

+    private static final NumericTypeHandler numericTypeHandler = new NumericTypeHandler();
+    private static final TemporalTypeHandler temporalTypeHandler = new TemporalTypeHandler();
+
+    private static final Map<PostgresDataType.DataCategory, PostgresDataTypeHandler> typeHandlers = Map.ofEntries(


I'm curious why you moved away from Map.of to ofEntries? The original had a little less boilerplate code.

Map.of allows only 10 arguments, so I moved to Map.ofEntries

oeyh · 2025-02-24T22:45:17Z

...rg/opensearch/dataprepper/plugins/source/rds/datatype/postgres/handler/RangeTypeHandler.java

+
+    private Object parseRangeValue(PostgresDataType columnType,String columnName, String rangeString) {
+
+        if(rangeString.equals("empty"))


Let's make "empty" a constant.

oeyh · 2025-02-24T22:55:36Z

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

-    private static final String POSTGRES_TIMESTAMP_MAX_INFINITY = "9999-12-31 23:59:59";
-    private static final String POSTGRES_TIMESTAMPTZ_MIN_INFINITY = "1970-01-01 00:00:00+00";
-    private static final String POSTGRES_TIMESTAMPTZ_MAX_INFINITY = "9999-12-31 23:59:59+00";
+    public static final String POSTGRES_DATE_MIN_INFINITY = "1970-01-01";


Why this change?

oeyh · 2025-02-24T23:05:30Z

...rg/opensearch/dataprepper/plugins/source/rds/datatype/postgres/handler/RangeTypeHandler.java

+        Map<String, Object> rangeMap = new HashMap<>();
+        if(!lowerBound.isEmpty()) {
+            Object parsedLowerBound = numericTypeHandler.handle(columnType, columnName, lowerBound);
+            rangeMap.put(lowerBoundInclusivity.equals("[") ? "gte" : "gt", parsedLowerBound);


Make "gte", "gt", "lte", "lt" constants as well.

oeyh · 2025-02-24T23:17:30Z

...e/src/main/java/org/opensearch/dataprepper/plugins/source/rds/schema/MySqlSchemaManager.java

@@ -91,6 +91,7 @@ public Map<String, String> getColumnDataTypes(final String fullTableName) {
                        );
                    }
                }
+                return columnsToDataType;


Thanks for fixing these!

oeyh · 2025-02-24T23:20:17Z

...est/java/org/opensearch/dataprepper/plugins/source/rds/schema/PostgresSchemaManagerTest.java

@@ -229,6 +234,88 @@ void getColumnDataTypes_whenColumnsExist_shouldReturnValidMapping() throws SQLEx
        assertThat(result, equalTo(expectedColumnTypes));
    }

+    @Test
+    void getEnumColumns_successfully_returns_enum_columns() throws SQLException {


Can you add a test for getColumnDataTypes when enums are involved?

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

Adding enum and range datatype support for postgres

147ce19

Signed-off-by: Divyansh Bokadia <dbokadia@amazon.com> Adding enum and range datatype support for postgres Adding enum and range datatype support for postgres

divbok force-pushed the postgres_data branch from 05ae81a to 147ce19 Compare February 24, 2025 16:44

dlvenable requested changes Feb 24, 2025

View reviewed changes

oeyh reviewed Feb 24, 2025

View reviewed changes

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Adding enum and range datatype support for postgres #5458

Adding enum and range datatype support for postgres #5458

divbok commented Feb 24, 2025 •

edited

Loading

dlvenable Feb 24, 2025

divbok Feb 24, 2025

dlvenable Feb 24, 2025

divbok Feb 24, 2025 •

edited

Loading

dlvenable Feb 24, 2025

divbok Feb 24, 2025

oeyh Feb 24, 2025

oeyh Feb 24, 2025

oeyh Feb 24, 2025

oeyh Feb 24, 2025

oeyh Feb 24, 2025


		private Object parseRangeValue(PostgresDataType columnType,String columnName, String rangeString) {

		if(rangeString.equals("empty"))

Adding enum and range datatype support for postgres #5458

Are you sure you want to change the base?

Adding enum and range datatype support for postgres #5458

Conversation

divbok commented Feb 24, 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

divbok Feb 24, 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

Choose a reason for hiding this comment

Choose a reason for hiding this comment

divbok commented Feb 24, 2025 •

edited

Loading

divbok Feb 24, 2025 •

edited

Loading