-
Notifications
You must be signed in to change notification settings - Fork 202
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Detect cascading actions in RDS source (#5168)
* Detect cascades Signed-off-by: Hai Yan <[email protected]> * Address comments Signed-off-by: Hai Yan <[email protected]> --------- Signed-off-by: Hai Yan <[email protected]>
- Loading branch information
Showing
20 changed files
with
931 additions
and
21 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
56 changes: 56 additions & 0 deletions
56
...org/opensearch/dataprepper/plugins/source/rds/coordination/partition/ResyncPartition.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,56 @@ | ||
| /* | ||
| * Copyright OpenSearch Contributors | ||
| * SPDX-License-Identifier: Apache-2.0 | ||
| */ | ||
|
|
||
| package org.opensearch.dataprepper.plugins.source.rds.coordination.partition; | ||
|
|
||
| import org.opensearch.dataprepper.model.source.coordinator.SourcePartitionStoreItem; | ||
| import org.opensearch.dataprepper.model.source.coordinator.enhanced.EnhancedSourcePartition; | ||
| import org.opensearch.dataprepper.plugins.source.rds.coordination.state.ResyncProgressState; | ||
|
|
||
| import java.util.Optional; | ||
|
|
||
| public class ResyncPartition extends EnhancedSourcePartition<ResyncProgressState> { | ||
|
|
||
| public static final String PARTITION_TYPE = "RESYNC"; | ||
|
|
||
| private final String database; | ||
| private final String table; | ||
| private final long timestamp; | ||
| private final ResyncProgressState state; | ||
|
|
||
| public ResyncPartition(String database, String table, long timestamp, ResyncProgressState state) { | ||
| this.database = database; | ||
| this.table = table; | ||
| this.timestamp = timestamp; | ||
| this.state = state; | ||
| } | ||
|
|
||
| public ResyncPartition(SourcePartitionStoreItem sourcePartitionStoreItem) { | ||
| setSourcePartitionStoreItem(sourcePartitionStoreItem); | ||
| String[] keySplits = sourcePartitionStoreItem.getSourcePartitionKey().split("\\|"); | ||
| database = keySplits[0]; | ||
| table = keySplits[1]; | ||
| timestamp = Long.parseLong(keySplits[2]); | ||
| state = convertStringToPartitionProgressState(ResyncProgressState.class, sourcePartitionStoreItem.getPartitionProgressState()); | ||
| } | ||
|
|
||
| @Override | ||
| public String getPartitionType() { | ||
| return PARTITION_TYPE; | ||
| } | ||
|
|
||
| @Override | ||
| public String getPartitionKey() { | ||
| return database + "|" + table + "|" + timestamp; | ||
| } | ||
|
|
||
| @Override | ||
| public Optional<ResyncProgressState> getProgressState() { | ||
| if (state != null) { | ||
| return Optional.of(state); | ||
| } | ||
| return Optional.empty(); | ||
| } | ||
| } |
25 changes: 25 additions & 0 deletions
25
...org/opensearch/dataprepper/plugins/source/rds/coordination/state/ResyncProgressState.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,25 @@ | ||
| /* | ||
| * Copyright OpenSearch Contributors | ||
| * SPDX-License-Identifier: Apache-2.0 | ||
| */ | ||
|
|
||
| package org.opensearch.dataprepper.plugins.source.rds.coordination.state; | ||
|
|
||
| import com.fasterxml.jackson.annotation.JsonProperty; | ||
| import lombok.Getter; | ||
| import lombok.Setter; | ||
|
|
||
| import java.util.List; | ||
|
|
||
| @Setter | ||
| @Getter | ||
| public class ResyncProgressState { | ||
| @JsonProperty("foreignKeyName") | ||
| private String foreignKeyName; | ||
|
|
||
| @JsonProperty("updatedValue") | ||
| private Object updatedValue; | ||
|
|
||
| @JsonProperty("primaryKeys") | ||
| private List<String> primaryKeys; | ||
| } |
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
56 changes: 56 additions & 0 deletions
56
...e/src/main/java/org/opensearch/dataprepper/plugins/source/rds/model/ForeignKeyAction.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,56 @@ | ||
| /* | ||
| * Copyright OpenSearch Contributors | ||
| * SPDX-License-Identifier: Apache-2.0 | ||
| */ | ||
|
|
||
| package org.opensearch.dataprepper.plugins.source.rds.model; | ||
|
|
||
| import java.sql.DatabaseMetaData; | ||
| import java.util.Set; | ||
|
|
||
| public enum ForeignKeyAction { | ||
| CASCADE, | ||
| NO_ACTION, | ||
| RESTRICT, | ||
| SET_DEFAULT, | ||
| SET_NULL, | ||
| UNKNOWN; | ||
|
|
||
| private static final Set<ForeignKeyAction> CASCADING_ACTIONS = Set.of(CASCADE, SET_DEFAULT, SET_NULL); | ||
| /** | ||
| * Returns the corresponding ForeignKeyAction for the given metadata action value. | ||
| * | ||
| * @param action the metadata action value | ||
| * @return the corresponding ForeignKeyAction | ||
| */ | ||
| public static ForeignKeyAction getActionFromMetadata(short action) { | ||
| switch (action) { | ||
| case DatabaseMetaData.importedKeyCascade: | ||
| return CASCADE; | ||
| case DatabaseMetaData.importedKeySetNull: | ||
| return SET_NULL; | ||
| case DatabaseMetaData.importedKeySetDefault: | ||
| return SET_DEFAULT; | ||
| case DatabaseMetaData.importedKeyRestrict: | ||
| return RESTRICT; | ||
| case DatabaseMetaData.importedKeyNoAction: | ||
| return NO_ACTION; | ||
| default: | ||
| return UNKNOWN; | ||
| } | ||
| } | ||
|
|
||
| /** | ||
| * Checks if the foreign key action is one of the cascading actions (CASCADE, SET_DEFAULT, SET_NULL) | ||
| * that will result in changes to the foreign key value when referenced key in parent table changes. | ||
| * | ||
| * @param foreignKeyAction the foreign key action | ||
| * @return true if the foreign key action is a cascade action, false otherwise | ||
| */ | ||
| public static boolean isCascadingAction(ForeignKeyAction foreignKeyAction) { | ||
| if (foreignKeyAction == null) { | ||
| return false; | ||
| } | ||
| return CASCADING_ACTIONS.contains(foreignKeyAction); | ||
| } | ||
| } |
71 changes: 71 additions & 0 deletions
71
...src/main/java/org/opensearch/dataprepper/plugins/source/rds/model/ForeignKeyRelation.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,71 @@ | ||
| /* | ||
| * Copyright OpenSearch Contributors | ||
| * SPDX-License-Identifier: Apache-2.0 | ||
| */ | ||
|
|
||
| package org.opensearch.dataprepper.plugins.source.rds.model; | ||
|
|
||
| import com.fasterxml.jackson.annotation.JsonCreator; | ||
| import com.fasterxml.jackson.annotation.JsonProperty; | ||
| import lombok.Builder; | ||
| import lombok.Getter; | ||
|
|
||
| @Getter | ||
| @Builder | ||
| public class ForeignKeyRelation { | ||
| // TODO: add java docs | ||
| @JsonProperty("database_name") | ||
| private final String databaseName; | ||
|
|
||
| @JsonProperty("parent_table_name") | ||
| private final String parentTableName; | ||
|
|
||
| @JsonProperty("referenced_key_name") | ||
| private final String referencedKeyName; | ||
|
|
||
| @JsonProperty("child_table_name") | ||
| private final String childTableName; | ||
|
|
||
| @JsonProperty("foreign_key_name") | ||
| private final String foreignKeyName; | ||
|
|
||
| @JsonProperty("foreign_key_default_value") | ||
| @Builder.Default | ||
| private Object foreignKeyDefaultValue = null; | ||
|
|
||
| @JsonProperty("update_action") | ||
| private final ForeignKeyAction updateAction; | ||
|
|
||
| @JsonProperty("delete_action") | ||
| private final ForeignKeyAction deleteAction; | ||
|
|
||
| @JsonCreator | ||
| public ForeignKeyRelation(@JsonProperty("database_name") String databaseName, | ||
| @JsonProperty("parent_table_name") String parentTableName, | ||
| @JsonProperty("referenced_key_name") String referencedKeyName, | ||
| @JsonProperty("child_table_name") String childTableName, | ||
| @JsonProperty("foreign_key_name") String foreignKeyName, | ||
| @JsonProperty("foreign_key_default_value") Object foreignKeyDefaultValue, | ||
| @JsonProperty("update_action") ForeignKeyAction updateAction, | ||
| @JsonProperty("delete_action") ForeignKeyAction deleteAction) { | ||
| this.databaseName = databaseName; | ||
| this.parentTableName = parentTableName; | ||
| this.referencedKeyName = referencedKeyName; | ||
| this.childTableName = childTableName; | ||
| this.foreignKeyName = foreignKeyName; | ||
| this.foreignKeyDefaultValue = foreignKeyDefaultValue; | ||
| this.updateAction = updateAction; | ||
| this.deleteAction = deleteAction; | ||
| } | ||
|
|
||
| /** | ||
| * Checks either update action or delete action is one of the cascading actions (CASCADE, SET_DEFAULT, SET_NULL). | ||
| * | ||
| * @param foreignKeyRelation The foreign key relation. | ||
| * @return True if the foreign key relation contains a cascade action, false otherwise. | ||
| */ | ||
| public static boolean containsCascadingAction(ForeignKeyRelation foreignKeyRelation) { | ||
| return ForeignKeyAction.isCascadingAction(foreignKeyRelation.getUpdateAction()) || | ||
| ForeignKeyAction.isCascadingAction(foreignKeyRelation.getDeleteAction()); | ||
| } | ||
| } |
83 changes: 83 additions & 0 deletions
83
...source/src/main/java/org/opensearch/dataprepper/plugins/source/rds/model/ParentTable.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,83 @@ | ||
| /* | ||
| * Copyright OpenSearch Contributors | ||
| * SPDX-License-Identifier: Apache-2.0 | ||
| */ | ||
|
|
||
| package org.opensearch.dataprepper.plugins.source.rds.model; | ||
|
|
||
| import lombok.AccessLevel; | ||
| import lombok.Builder; | ||
| import lombok.Getter; | ||
|
|
||
| import java.util.ArrayList; | ||
| import java.util.HashMap; | ||
| import java.util.List; | ||
| import java.util.Map; | ||
|
|
||
| /** | ||
| * A data model for a parent table in a foreign key relationship | ||
| */ | ||
| @Getter | ||
| @Builder | ||
| public class ParentTable { | ||
| private final String databaseName; | ||
| private final String tableName; | ||
| /** | ||
| * Column name to a list of ForeignKeyRelation in which the column is referenced | ||
| */ | ||
| private final Map<String, List<ForeignKeyRelation>> referencedColumnMap; | ||
|
|
||
| @Getter(AccessLevel.NONE) | ||
| @Builder.Default | ||
| private Map<String, List<ForeignKeyRelation>> columnsWithCascadingUpdate = null; | ||
|
|
||
| @Getter(AccessLevel.NONE) | ||
| @Builder.Default | ||
| private Map<String, List<ForeignKeyRelation>> columnsWithCascadingDelete = null; | ||
|
|
||
| /** | ||
| * Returns a map of column name to a list of ForeignKeyRelation in which the column is referenced and the update action is cascading. | ||
| * @return a map of column name to a list of ForeignKeyRelation | ||
| */ | ||
| public Map<String, List<ForeignKeyRelation>> getColumnsWithCascadingUpdate() { | ||
| if (columnsWithCascadingUpdate != null) { | ||
| return columnsWithCascadingUpdate; | ||
| } | ||
|
|
||
| columnsWithCascadingUpdate = new HashMap<>(); | ||
| for (String column : referencedColumnMap.keySet()) { | ||
| for (ForeignKeyRelation foreignKeyRelation : referencedColumnMap.get(column)) { | ||
| if (ForeignKeyAction.isCascadingAction(foreignKeyRelation.getUpdateAction())) { | ||
| if (!columnsWithCascadingUpdate.containsKey(column)) { | ||
| columnsWithCascadingUpdate.put(column, new ArrayList<>()); | ||
| } | ||
| columnsWithCascadingUpdate.get(column).add(foreignKeyRelation); | ||
| } | ||
| } | ||
| } | ||
| return columnsWithCascadingUpdate; | ||
| } | ||
|
|
||
| /** | ||
| * Returns a map of column name to a list of ForeignKeyRelation in which the column is referenced and the delete action is cascading. | ||
| * @return a map of column name to a list of ForeignKeyRelation | ||
| */ | ||
| public Map<String, List<ForeignKeyRelation>> getColumnsWithCascadingDelete() { | ||
| if (columnsWithCascadingDelete != null) { | ||
| return columnsWithCascadingDelete; | ||
| } | ||
|
|
||
| columnsWithCascadingDelete = new HashMap<>(); | ||
| for (String column : referencedColumnMap.keySet()) { | ||
| for (ForeignKeyRelation foreignKeyRelation : referencedColumnMap.get(column)) { | ||
| if (ForeignKeyAction.isCascadingAction(foreignKeyRelation.getDeleteAction())) { | ||
| if (!columnsWithCascadingDelete.containsKey(column)) { | ||
| columnsWithCascadingDelete.put(column, new ArrayList<>()); | ||
| } | ||
| columnsWithCascadingDelete.get(column).add(foreignKeyRelation); | ||
| } | ||
| } | ||
| } | ||
| return columnsWithCascadingDelete; | ||
| } | ||
| } |
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
Oops, something went wrong.