forked from opensearch-project/sql
-
Notifications
You must be signed in to change notification settings - Fork 0
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Signed-off-by: Peng Huo <[email protected]>
- Loading branch information
Showing
12 changed files
with
508 additions
and
30 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
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
52 changes: 52 additions & 0 deletions
52
spark/src/main/java/org/opensearch/sql/spark/flint/FlintIndexState.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,52 @@ | ||
| /* | ||
| * Copyright OpenSearch Contributors | ||
| * SPDX-License-Identifier: Apache-2.0 | ||
| */ | ||
|
|
||
| package org.opensearch.sql.spark.flint; | ||
|
|
||
| import java.util.Arrays; | ||
| import java.util.Locale; | ||
| import java.util.Map; | ||
| import java.util.stream.Collectors; | ||
| import lombok.Getter; | ||
|
|
||
| /** | ||
| * Flint index state. | ||
| */ | ||
| @Getter | ||
| public enum FlintIndexState { | ||
| // stable state | ||
| EMPTY("empty"), | ||
| // transitioning state | ||
| CREATING("creating"), | ||
| // transitioning state | ||
| REFRESHING("refreshing"), | ||
| // transitioning state | ||
| CANCELLING("cancelling"), | ||
| // stable state | ||
| ACTIVE("active"), | ||
| // transitioning state | ||
| DELETING("deleting"), | ||
| // stable state | ||
| DELETED("deleted"); | ||
|
|
||
| private final String state; | ||
|
|
||
| FlintIndexState(String state) { | ||
| this.state = state; | ||
| } | ||
|
|
||
| private static Map<String, FlintIndexState> STATES = | ||
| Arrays.stream(FlintIndexState.values()) | ||
| .collect(Collectors.toMap(t -> t.name().toLowerCase(), t -> t)); | ||
|
|
||
| public static FlintIndexState fromString(String key) { | ||
| for (FlintIndexState ss : FlintIndexState.values()) { | ||
| if (ss.getState().toLowerCase(Locale.ROOT).equals(key)) { | ||
| return ss; | ||
| } | ||
| } | ||
| throw new IllegalArgumentException("Invalid flint index state: " + key); | ||
| } | ||
| } |
128 changes: 128 additions & 0 deletions
128
spark/src/main/java/org/opensearch/sql/spark/flint/FlintIndexStateModel.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,128 @@ | ||
| /* | ||
| * Copyright OpenSearch Contributors | ||
| * SPDX-License-Identifier: Apache-2.0 | ||
| */ | ||
|
|
||
| package org.opensearch.sql.spark.flint; | ||
|
|
||
| import static org.opensearch.sql.spark.execution.session.SessionModel.APPLICATION_ID; | ||
| import static org.opensearch.sql.spark.execution.session.SessionModel.DATASOURCE_NAME; | ||
| import static org.opensearch.sql.spark.execution.session.SessionModel.JOB_ID; | ||
| import static org.opensearch.sql.spark.execution.statement.StatementModel.ERROR; | ||
| import static org.opensearch.sql.spark.execution.statement.StatementModel.VERSION; | ||
|
|
||
| import java.io.IOException; | ||
| import lombok.Builder; | ||
| import lombok.EqualsAndHashCode; | ||
| import lombok.Getter; | ||
| import lombok.SneakyThrows; | ||
| import org.opensearch.core.xcontent.XContentBuilder; | ||
| import org.opensearch.core.xcontent.XContentParser; | ||
| import org.opensearch.core.xcontent.XContentParserUtils; | ||
| import org.opensearch.sql.spark.execution.statestore.StateModel; | ||
|
|
||
| /** | ||
| * Flint Index Model maintain the index state. | ||
| */ | ||
| @Getter | ||
| @Builder | ||
| @EqualsAndHashCode(callSuper = false) | ||
| public class FlintIndexStateModel extends StateModel { | ||
| public static final String FLINT_INDEX_DOC_TYPE = "flintindexstate"; | ||
| public static final String DOC_ID_PREFIX = "flint"; | ||
|
|
||
| private final FlintIndexState indexState; | ||
| private final String applicationId; | ||
| private final String jobId; | ||
| private final String datasourceName; | ||
| private final long lastUpdateTime; | ||
| private final String error; | ||
|
|
||
| @EqualsAndHashCode.Exclude | ||
| private final long seqNo; | ||
| @EqualsAndHashCode.Exclude | ||
| private final long primaryTerm; | ||
|
|
||
| public FlintIndexStateModel(FlintIndexState indexState, String applicationId, | ||
| String jobId, String datasourceName, long lastUpdateTime, String error, | ||
| long seqNo, | ||
| long primaryTerm) { | ||
| this.indexState = indexState; | ||
| this.applicationId = applicationId; | ||
| this.jobId = jobId; | ||
| this.datasourceName = datasourceName; | ||
| this.lastUpdateTime = lastUpdateTime; | ||
| this.error = error; | ||
| this.seqNo = seqNo; | ||
| this.primaryTerm = primaryTerm; | ||
| } | ||
|
|
||
| public static FlintIndexStateModel copy(FlintIndexStateModel copy, long seqNo, long primaryTerm) { | ||
| return new FlintIndexStateModel(copy.indexState, copy.applicationId, copy.jobId, | ||
| copy.datasourceName, copy.lastUpdateTime, copy.error, seqNo, primaryTerm); | ||
| } | ||
|
|
||
| public static FlintIndexStateModel copyWithState( | ||
| FlintIndexStateModel copy, FlintIndexState state, long seqNo, long primaryTerm) { | ||
| return new FlintIndexStateModel(state, copy.applicationId, copy.jobId, | ||
| copy.datasourceName, copy.lastUpdateTime, copy.error, seqNo, primaryTerm); | ||
| } | ||
|
|
||
| @SneakyThrows | ||
| public static FlintIndexStateModel fromXContent(XContentParser parser, long seqNo, long primaryTerm) { | ||
| FlintIndexStateModelBuilder builder = FlintIndexStateModel.builder(); | ||
| XContentParserUtils.ensureExpectedToken( | ||
| XContentParser.Token.START_OBJECT, parser.currentToken(), parser); | ||
| while (!XContentParser.Token.END_OBJECT.equals(parser.nextToken())) { | ||
| String fieldName = parser.currentName(); | ||
| parser.nextToken(); | ||
| switch (fieldName) { | ||
| case VERSION: | ||
| case TYPE: | ||
| // do nothing | ||
| break; | ||
| case STATE: | ||
| builder.indexState(FlintIndexState.fromString(parser.text())); | ||
| case APPLICATION_ID: | ||
| builder.applicationId(parser.text()); | ||
| break; | ||
| case JOB_ID: | ||
| builder.jobId(parser.text()); | ||
| break; | ||
| case DATASOURCE_NAME: | ||
| builder.datasourceName(parser.text()); | ||
| break; | ||
| case LAST_UPDATE_TIME: | ||
| builder.lastUpdateTime(parser.longValue()); | ||
| break; | ||
| case ERROR: | ||
| builder.error(parser.text()); | ||
| break; | ||
| } | ||
| } | ||
| builder.seqNo(seqNo); | ||
| builder.primaryTerm(primaryTerm); | ||
| return builder.build(); | ||
| } | ||
|
|
||
| @Override | ||
| public String getId() { | ||
| return DOC_ID_PREFIX + jobId; | ||
| } | ||
|
|
||
| @Override | ||
| public XContentBuilder toXContent(XContentBuilder builder, Params params) throws IOException { | ||
| builder | ||
| .startObject() | ||
| .field(VERSION, VERSION_1_0) | ||
| .field(TYPE, FLINT_INDEX_DOC_TYPE) | ||
| .field(STATE, indexState.getState()) | ||
| .field(APPLICATION_ID, applicationId) | ||
| .field(JOB_ID, jobId) | ||
| .field(DATASOURCE_NAME, datasourceName) | ||
| .field(LAST_UPDATE_TIME, lastUpdateTime) | ||
| .field(ERROR, error) | ||
| .endObject(); | ||
| return builder; | ||
| } | ||
| } |
Oops, something went wrong.