-
Notifications
You must be signed in to change notification settings - Fork 214
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Initial PR for stream support for Postgres in Rds source (#5310)
* First working version Signed-off-by: Hai Yan <[email protected]> * More progress and update existing unit tests Signed-off-by: Hai Yan <[email protected]> * Add unit tests Signed-off-by: Hai Yan <[email protected]> * Remove and rename classes Signed-off-by: Hai Yan <[email protected]> * Remove test code Signed-off-by: Hai Yan <[email protected]> * Address review comments Signed-off-by: Hai Yan <[email protected]> * Address minor issues Signed-off-by: Hai Yan <[email protected]> * Group MySQL and Postgres stream states Signed-off-by: Hai Yan <[email protected]> * Address more comments Signed-off-by: Hai Yan <[email protected]> * Fix Java21 build Signed-off-by: Hai Yan <[email protected]> --------- Signed-off-by: Hai Yan <[email protected]>
- Loading branch information
Showing
48 changed files
with
2,299 additions
and
464 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
41 changes: 41 additions & 0 deletions
41
...va/org/opensearch/dataprepper/plugins/source/rds/coordination/state/MySqlStreamState.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,41 @@ | ||
| /* | ||
| * Copyright OpenSearch Contributors | ||
| * SPDX-License-Identifier: Apache-2.0 | ||
| * | ||
| * The OpenSearch Contributors require contributions made to | ||
| * this file be licensed under the Apache-2.0 license or a | ||
| * compatible open source license. | ||
| */ | ||
|
|
||
| package org.opensearch.dataprepper.plugins.source.rds.coordination.state; | ||
|
|
||
| import com.fasterxml.jackson.annotation.JsonProperty; | ||
| import org.opensearch.dataprepper.plugins.source.rds.model.BinlogCoordinate; | ||
| import org.opensearch.dataprepper.plugins.source.rds.model.ForeignKeyRelation; | ||
|
|
||
| import java.util.List; | ||
|
|
||
| public class MySqlStreamState { | ||
|
|
||
| @JsonProperty("currentPosition") | ||
| private BinlogCoordinate currentPosition; | ||
|
|
||
| @JsonProperty("foreignKeyRelations") | ||
| private List<ForeignKeyRelation> foreignKeyRelations; | ||
|
|
||
| public BinlogCoordinate getCurrentPosition() { | ||
| return currentPosition; | ||
| } | ||
|
|
||
| public void setCurrentPosition(BinlogCoordinate currentPosition) { | ||
| this.currentPosition = currentPosition; | ||
| } | ||
|
|
||
| public List<ForeignKeyRelation> getForeignKeyRelations() { | ||
| return foreignKeyRelations; | ||
| } | ||
|
|
||
| public void setForeignKeyRelations(List<ForeignKeyRelation> foreignKeyRelations) { | ||
| this.foreignKeyRelations = foreignKeyRelations; | ||
| } | ||
| } |
48 changes: 48 additions & 0 deletions
48
...org/opensearch/dataprepper/plugins/source/rds/coordination/state/PostgresStreamState.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,48 @@ | ||
| /* | ||
| * Copyright OpenSearch Contributors | ||
| * SPDX-License-Identifier: Apache-2.0 | ||
| * | ||
| * The OpenSearch Contributors require contributions made to | ||
| * this file be licensed under the Apache-2.0 license or a | ||
| * compatible open source license. | ||
| */ | ||
|
|
||
| package org.opensearch.dataprepper.plugins.source.rds.coordination.state; | ||
|
|
||
| import com.fasterxml.jackson.annotation.JsonProperty; | ||
|
|
||
| public class PostgresStreamState { | ||
|
|
||
| @JsonProperty("currentLsn") | ||
| private String currentLsn; | ||
|
|
||
| @JsonProperty("publicationName") | ||
| private String publicationName; | ||
|
|
||
| @JsonProperty("replicationSlotName") | ||
| private String replicationSlotName; | ||
|
|
||
| public String getCurrentLsn() { | ||
| return currentLsn; | ||
| } | ||
|
|
||
| public void setCurrentLsn(String currentLsn) { | ||
| this.currentLsn = currentLsn; | ||
| } | ||
|
|
||
| public String getPublicationName() { | ||
| return publicationName; | ||
| } | ||
|
|
||
| public void setPublicationName(String publicationName) { | ||
| this.publicationName = publicationName; | ||
| } | ||
|
|
||
| public String getReplicationSlotName() { | ||
| return replicationSlotName; | ||
| } | ||
|
|
||
| public void setReplicationSlotName(String replicationSlotName) { | ||
| this.replicationSlotName = replicationSlotName; | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
68 changes: 68 additions & 0 deletions
68
...main/java/org/opensearch/dataprepper/plugins/source/rds/datatype/postgres/ColumnType.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,68 @@ | ||
| /* | ||
| * Copyright OpenSearch Contributors | ||
| * SPDX-License-Identifier: Apache-2.0 | ||
| * | ||
| * The OpenSearch Contributors require contributions made to | ||
| * this file be licensed under the Apache-2.0 license or a | ||
| * compatible open source license. | ||
| * | ||
| */ | ||
|
|
||
| package org.opensearch.dataprepper.plugins.source.rds.datatype.postgres; | ||
|
|
||
| import java.util.HashMap; | ||
| import java.util.Map; | ||
|
|
||
| public enum ColumnType { | ||
| BOOLEAN(16, "boolean"), | ||
| SMALLINT(21, "smallint"), | ||
| INTEGER(23, "integer"), | ||
| BIGINT(20, "bigint"), | ||
| REAL(700, "real"), | ||
| DOUBLE_PRECISION(701, "double precision"), | ||
| NUMERIC(1700, "numeric"), | ||
| TEXT(25, "text"), | ||
| VARCHAR(1043, "varchar"), | ||
| DATE(1082, "date"), | ||
| TIME(1083, "time"), | ||
| TIMESTAMP(1114, "timestamp"), | ||
| TIMESTAMPTZ(1184, "timestamptz"), | ||
| UUID(2950, "uuid"), | ||
| JSON(114, "json"), | ||
| JSONB(3802, "jsonb"); | ||
|
|
||
| private final int typeId; | ||
| private final String typeName; | ||
|
|
||
| private static final Map<Integer, ColumnType> TYPE_ID_MAP = new HashMap<>(); | ||
|
|
||
| static { | ||
| for (ColumnType type : values()) { | ||
| TYPE_ID_MAP.put(type.typeId, type); | ||
| } | ||
| } | ||
|
|
||
| ColumnType(int typeId, String typeName) { | ||
| this.typeId = typeId; | ||
| this.typeName = typeName; | ||
| } | ||
|
|
||
| public int getTypeId() { | ||
| return typeId; | ||
| } | ||
|
|
||
| public String getTypeName() { | ||
| return typeName; | ||
| } | ||
|
|
||
| public static ColumnType getByTypeId(int typeId) { | ||
| if (!TYPE_ID_MAP.containsKey(typeId)) { | ||
| throw new IllegalArgumentException("Unsupported column type id: " + typeId); | ||
| } | ||
| return TYPE_ID_MAP.get(typeId); | ||
| } | ||
|
|
||
| public static String getTypeNameByEnum(ColumnType columnType) { | ||
| return columnType.getTypeName(); | ||
| } | ||
| } |
Oops, something went wrong.