-
Notifications
You must be signed in to change notification settings - Fork 33
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Merge branch 'main' into feature-support-sigv4a-signer
Signed-off-by: Louis Chu <[email protected]>
- Loading branch information
Showing
45 changed files
with
972 additions
and
472 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
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
90 changes: 90 additions & 0 deletions
90
...rk-integration/src/main/scala/org/opensearch/flint/spark/FlintSparkValidationHelper.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,90 @@ | ||
| /* | ||
| * Copyright OpenSearch Contributors | ||
| * SPDX-License-Identifier: Apache-2.0 | ||
| */ | ||
|
|
||
| package org.opensearch.flint.spark | ||
|
|
||
| import java.io.IOException | ||
|
|
||
| import org.apache.hadoop.fs.Path | ||
| import org.opensearch.flint.spark.covering.FlintSparkCoveringIndex | ||
| import org.opensearch.flint.spark.mv.FlintSparkMaterializedView | ||
| import org.opensearch.flint.spark.skipping.FlintSparkSkippingIndex | ||
|
|
||
| import org.apache.spark.internal.Logging | ||
| import org.apache.spark.sql.SparkSession | ||
| import org.apache.spark.sql.catalyst.analysis.UnresolvedRelation | ||
| import org.apache.spark.sql.execution.command.DDLUtils | ||
| import org.apache.spark.sql.execution.streaming.CheckpointFileManager | ||
| import org.apache.spark.sql.flint.{loadTable, parseTableName, qualifyTableName} | ||
|
|
||
| /** | ||
| * Flint Spark validation helper. | ||
| */ | ||
| trait FlintSparkValidationHelper extends Logging { | ||
|
|
||
| /** | ||
| * Determines whether the source table(s) for a given Flint index are supported. | ||
| * | ||
| * @param spark | ||
| * Spark session | ||
| * @param index | ||
| * Flint index | ||
| * @return | ||
| * true if all non Hive, otherwise false | ||
| */ | ||
| def isTableProviderSupported(spark: SparkSession, index: FlintSparkIndex): Boolean = { | ||
| // Extract source table name (possibly more than one for MV query) | ||
| val tableNames = index match { | ||
| case skipping: FlintSparkSkippingIndex => Seq(skipping.tableName) | ||
| case covering: FlintSparkCoveringIndex => Seq(covering.tableName) | ||
| case mv: FlintSparkMaterializedView => | ||
| spark.sessionState.sqlParser | ||
| .parsePlan(mv.query) | ||
| .collect { case relation: UnresolvedRelation => | ||
| qualifyTableName(spark, relation.tableName) | ||
| } | ||
| } | ||
|
|
||
| // Validate if any source table is not supported (currently Hive only) | ||
| tableNames.exists { tableName => | ||
| val (catalog, ident) = parseTableName(spark, tableName) | ||
| val table = loadTable(catalog, ident).get | ||
|
|
||
| // TODO: add allowed table provider list | ||
| DDLUtils.isHiveTable(Option(table.properties().get("provider"))) | ||
| } | ||
| } | ||
|
|
||
| /** | ||
| * Checks whether a specified checkpoint location is accessible. Accessibility, in this context, | ||
| * means that the folder exists and the current Spark session has the necessary permissions to | ||
| * access it. | ||
| * | ||
| * @param spark | ||
| * Spark session | ||
| * @param checkpointLocation | ||
| * checkpoint location | ||
| * @return | ||
| * true if accessible, otherwise false | ||
| */ | ||
| def isCheckpointLocationAccessible(spark: SparkSession, checkpointLocation: String): Boolean = { | ||
| try { | ||
| val checkpointManager = | ||
| CheckpointFileManager.create( | ||
| new Path(checkpointLocation), | ||
| spark.sessionState.newHadoopConf()) | ||
|
|
||
| // The primary intent here is to catch any exceptions during the accessibility check. | ||
| // The actual result is ignored, as Spark can create any necessary sub-folders | ||
| // when the streaming job starts. | ||
| checkpointManager.exists(new Path(checkpointLocation)) | ||
| true | ||
| } catch { | ||
| case e: IOException => | ||
| logWarning(s"Failed to check if checkpoint location $checkpointLocation exists", e) | ||
| false | ||
| } | ||
| } | ||
| } |
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
Oops, something went wrong.