Request index not exist handling #169
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
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -44,14 +44,15 @@ | |
| import org.opensearch.flint.core.auth.AWSRequestSigningApacheInterceptor; | ||
| import org.opensearch.flint.core.metadata.FlintMetadata; | ||
| import org.opensearch.flint.core.metadata.log.DefaultOptimisticTransaction; | ||
| import org.opensearch.flint.core.metadata.log.FlintMetadataLogEntry; | ||
| import org.opensearch.flint.core.metadata.log.OptimisticTransaction; | ||
| import org.opensearch.index.query.AbstractQueryBuilder; | ||
| import org.opensearch.index.query.MatchAllQueryBuilder; | ||
| import org.opensearch.index.query.QueryBuilder; | ||
| import org.opensearch.search.SearchModule; | ||
| import org.opensearch.search.builder.SearchSourceBuilder; | ||
| import scala.Option; | ||
| import scala.Some; | ||
|
|
||
| /** | ||
| * Flint client implementation for OpenSearch storage. | ||
|
|
@@ -80,34 +81,48 @@ public FlintOpenSearchClient(FlintOptions options) { | |
| } | ||
|
|
||
| @Override | ||
| public <T> OptimisticTransaction<T> startTransaction(String indexName, String dataSourceName, | ||
| boolean forceInit) { | ||
| LOG.info("Starting transaction on index " + indexName + " and data source " + dataSourceName); | ||
| String metaLogIndexName = dataSourceName.isEmpty() ? META_LOG_NAME_PREFIX | ||
| : META_LOG_NAME_PREFIX + "_" + dataSourceName; | ||
| try (RestHighLevelClient client = createClient()) { | ||
| if (client.indices().exists(new GetIndexRequest(metaLogIndexName), RequestOptions.DEFAULT)) { | ||
| LOG.info("Found metadata log index " + metaLogIndexName); | ||
| } else { | ||
| if (forceInit) { | ||
| createIndex(metaLogIndexName, FlintMetadataLogEntry.QUERY_EXECUTION_REQUEST_MAPPING(), | ||
| Some.apply(FlintMetadataLogEntry.QUERY_EXECUTION_REQUEST_SETTINGS())); | ||
| } else { | ||
| String errorMsg = "Metadata log index not found " + metaLogIndexName; | ||
| LOG.warning(errorMsg); | ||
| throw new IllegalStateException(errorMsg); | ||
|
There was a problem hiding this comment. where do you emit metrics QueryExecutionRequestIndexNotFound ? There was a problem hiding this comment. will add after metrics sink ready. There was a problem hiding this comment. track in here. #117 |
||
| } | ||
| } | ||
| return new DefaultOptimisticTransaction<>(dataSourceName, | ||
| new FlintOpenSearchMetadataLog(this, indexName, metaLogIndexName)); | ||
| } catch (IOException e) { | ||
| throw new IllegalStateException("Failed to check if index metadata log index exists " + metaLogIndexName, e); | ||
| } | ||
| } | ||
|
|
||
| @Override | ||
| public <T> OptimisticTransaction<T> startTransaction(String indexName, String dataSourceName) { | ||
| return startTransaction(indexName, dataSourceName, false); | ||
| } | ||
|
|
||
| @Override | ||
| public void createIndex(String indexName, FlintMetadata metadata) { | ||
| LOG.info("Creating Flint index " + indexName + " with metadata " + metadata); | ||
| createIndex(indexName, metadata.getContent(), metadata.indexSettings()); | ||
| } | ||
|
|
||
| protected void createIndex(String indexName, String mapping, Option<String> settings) { | ||
| LOG.info("Creating Flint index " + indexName); | ||
| String osIndexName = toLowercase(indexName); | ||
| try (RestHighLevelClient client = createClient()) { | ||
| CreateIndexRequest request = new CreateIndexRequest(osIndexName); | ||
| request.mapping(mapping, XContentType.JSON); | ||
| if (settings.isDefined()) { | ||
| request.settings(settings.get(), XContentType.JSON); | ||
| } | ||
|
|
||
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
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -4,14 +4,17 @@ | |
| import org.opensearch.action.update.UpdateRequest; | ||
| import org.opensearch.client.RequestOptions; | ||
| import org.opensearch.client.RestHighLevelClient; | ||
| import org.opensearch.client.indices.GetIndexRequest; | ||
| import org.opensearch.common.xcontent.XContentType; | ||
| import org.opensearch.flint.core.FlintClient; | ||
|
|
||
| import java.io.IOException; | ||
| import java.util.logging.Level; | ||
| import java.util.logging.Logger; | ||
|
|
||
| public class OpenSearchUpdater { | ||
| private static final Logger LOG = Logger.getLogger(OpenSearchUpdater.class.getName()); | ||
|
|
||
| private final String indexName; | ||
|
|
||
| private final FlintClient flintClient; | ||
|
|
@@ -28,6 +31,7 @@ public void upsert(String id, String doc) { | |
| // also, failure to close the client causes the job to be stuck in the running state as the client resource | ||
| // is not released. | ||
| try (RestHighLevelClient client = flintClient.createClient()) { | ||
| assertIndexExist(client, indexName); | ||
|
There was a problem hiding this comment. @kaituo any other test cases we need to cover when write to query_execution_request index? There was a problem hiding this comment. We may also need to check request indexing mapping as we did in the result index mapping. This can be done once during REPL lifetime. |
||
| UpdateRequest | ||
| updateRequest = | ||
| new UpdateRequest(indexName, id).doc(doc, XContentType.JSON) | ||
|
|
@@ -44,6 +48,7 @@ public void upsert(String id, String doc) { | |
|
|
||
| public void update(String id, String doc) { | ||
| try (RestHighLevelClient client = flintClient.createClient()) { | ||
| assertIndexExist(client, indexName); | ||
|
There was a problem hiding this comment. So this check is not required for Flint metadata log entry update? There was a problem hiding this comment. I found I didn't set refresh policy in create/update doc in There was a problem hiding this comment. |
||
| UpdateRequest | ||
| updateRequest = | ||
| new UpdateRequest(indexName, id).doc(doc, XContentType.JSON) | ||
|
|
@@ -59,6 +64,7 @@ public void update(String id, String doc) { | |
|
|
||
| public void updateIf(String id, String doc, long seqNo, long primaryTerm) { | ||
| try (RestHighLevelClient client = flintClient.createClient()) { | ||
| assertIndexExist(client, indexName); | ||
| UpdateRequest | ||
| updateRequest = | ||
| new UpdateRequest(indexName, id).doc(doc, XContentType.JSON) | ||
|
|
@@ -73,4 +79,13 @@ public void updateIf(String id, String doc, long seqNo, long primaryTerm) { | |
| id), e); | ||
| } | ||
| } | ||
|
|
||
| private void assertIndexExist(RestHighLevelClient client, String indexName) throws IOException { | ||
| LOG.info("Checking if index exists " + indexName); | ||
| if (!client.indices().exists(new GetIndexRequest(indexName), RequestOptions.DEFAULT)) { | ||
| String errorMsg = "Index not found " + indexName; | ||
| LOG.log(Level.SEVERE, errorMsg); | ||
| throw new IllegalStateException(errorMsg); | ||
| } | ||
| } | ||
| } | ||
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
Oops, something went wrong.
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Previously this was Long type? And should we add
jobStartTimehere too?There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
aligned with https://github.com/opensearch-project/sql/blob/main/spark/src/main/resources/query_execution_request_mapping.yml