forked from opensearch-project/opensearch-spark
-
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.
Fix FlintMetadata conversion issue in FlintOpenSearchClient (opensear…
…ch-project#75) * Fix FlintMetadata conversion issue in FlintOpenSearchClient In `flint-core/src/main/scala/org/opensearch/flint/core/storage/FlintOpenSearchClient.java`, we encountered a conversion problem when transforming the index mapping to `FlintMetadata`. The root cause can be traced back to changes made in [this PR](https://github.com/opensearch-project/opensearch-spark/pull/70/files#diff-c3bdd10ec081d200e375aa601acff4c39b10bb4c34862634f56859f0054fedbf). Previously, `FlintMetadata` stored its content as a simple string. However, the recent changes have made its structure more complex, introducing fields like `version`, `name`, `kind`, `source`, `indexedColumns`, `options`, `properties`, `schema`, and `indexSettings`. Given that Flint index is specialized for storing covering index and materialized view metadata, it's not a typical data index. To address this, we've opted to use the OpenSearch rest client directly for the conversion. Testing done: 1. manual verification Signed-off-by: Kaituo Li <[email protected]> * throw exception in client call Signed-off-by: Kaituo Li <[email protected]> --------- Signed-off-by: Kaituo Li <[email protected]>
- Loading branch information
Showing
5 changed files
with
124 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
85 changes: 85 additions & 0 deletions
85
spark-sql-application/src/main/scala/org/apache/spark/sql/OSClient.scala
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,85 @@ | ||
| /* | ||
| * Copyright OpenSearch Contributors | ||
| * SPDX-License-Identifier: Apache-2.0 | ||
| */ | ||
|
|
||
| package org.apache.spark.sql | ||
|
|
||
| import org.opensearch.client.RequestOptions | ||
| import org.opensearch.client.indices.{CreateIndexRequest, GetIndexRequest, GetIndexResponse} | ||
| import org.opensearch.client.indices.CreateIndexRequest | ||
| import org.opensearch.common.xcontent.XContentType | ||
| import org.opensearch.flint.core.{FlintClientBuilder, FlintOptions} | ||
|
|
||
| import org.apache.spark.internal.Logging | ||
|
|
||
| class OSClient(val flintOptions: FlintOptions) extends Logging { | ||
|
|
||
| def getIndexMetadata(osIndexName: String): String = { | ||
|
|
||
| using(FlintClientBuilder.build(flintOptions).createClient()) { client => | ||
| val request = new GetIndexRequest(osIndexName) | ||
| try { | ||
| val response = client.indices.get(request, RequestOptions.DEFAULT) | ||
| response.getMappings.get(osIndexName).source.string | ||
| } catch { | ||
| case e: Exception => | ||
| throw new IllegalStateException( | ||
| s"Failed to get OpenSearch index mapping for $osIndexName", | ||
| e) | ||
| } | ||
| } | ||
| } | ||
|
|
||
| /** | ||
| * Create a new index with given mapping. | ||
| * | ||
| * @param osIndexName | ||
| * the name of the index | ||
| * @param mapping | ||
| * the mapping of the index | ||
| * @return | ||
| * use Either for representing success or failure. A Right value indicates success, while a | ||
| * Left value indicates an error. | ||
| */ | ||
| def createIndex(osIndexName: String, mapping: String): Unit = { | ||
| logInfo(s"create $osIndexName") | ||
|
|
||
| using(FlintClientBuilder.build(flintOptions).createClient()) { client => | ||
| val request = new CreateIndexRequest(osIndexName) | ||
| request.mapping(mapping, XContentType.JSON) | ||
|
|
||
| try { | ||
| client.indices.create(request, RequestOptions.DEFAULT) | ||
| logInfo(s"create $osIndexName successfully") | ||
| } catch { | ||
| case e: Exception => | ||
| throw new IllegalStateException(s"Failed to create index $osIndexName", e); | ||
| } | ||
| } | ||
| } | ||
|
|
||
| /** | ||
| * the loan pattern to manage resource. | ||
| * | ||
| * @param resource | ||
| * the resource to be managed | ||
| * @param f | ||
| * the function to be applied to the resource | ||
| * @tparam A | ||
| * the type of the resource | ||
| * @tparam B | ||
| * the type of the result | ||
| * @return | ||
| * the result of the function | ||
| */ | ||
| def using[A <: AutoCloseable, B](resource: A)(f: A => B): B = { | ||
| try { | ||
| f(resource) | ||
| } finally { | ||
| // client is guaranteed to be non-null | ||
| resource.close() | ||
| } | ||
| } | ||
|
|
||
| } |