scalafmtAll

Signed-off-by: Sean Kao <[email protected]>
opensearch-project · Mar 20, 2024 · 0762842 · 0762842
1 parent 2101d34
commit 0762842
Show file tree

Hide file tree

Showing 5 changed files with 59 additions and 102 deletions.
diff --git a/...k-integration/src/main/scala/org/opensearch/flint/spark/sql/FlintSparkSqlAstBuilder.scala b/...k-integration/src/main/scala/org/opensearch/flint/spark/sql/FlintSparkSqlAstBuilder.scala
@@ -88,15 +88,20 @@ object FlintSparkSqlAstBuilder {
    * @param updateOptions
    *   options to update
    */
-  def updateIndex(flint: FlintSpark, indexName: String, updateOptions: Map[String, String]): Option[String] = {
-    val oldIndex = flint.describeIndex(indexName)
+  def updateIndex(
+      flint: FlintSpark,
+      indexName: String,
+      updateOptions: Map[String, String]): Option[String] = {
+    val oldIndex = flint
+      .describeIndex(indexName)
       .getOrElse(throw new IllegalStateException(s"Index $indexName doesn't exist"))
 
     val oldOptions = oldIndex.options.options
     validateUpdateOptions(oldOptions, updateOptions, oldIndex.kind)
 
     val mergedOptions = oldOptions ++ updateOptions
-    val newMetadata = oldIndex.metadata().copy(options = mergedOptions.mapValues(_.asInstanceOf[AnyRef]).asJava)
+    val newMetadata =
+      oldIndex.metadata().copy(options = mergedOptions.mapValues(_.asInstanceOf[AnyRef]).asJava)
     val newIndex = FlintSparkIndexFactory.create(newMetadata)
 
     val updateMode = newIndex.options.autoRefresh() match {
@@ -108,9 +113,8 @@ object FlintSparkSqlAstBuilder {
   }
 
   /**
-   * Validate update options.
-   * These are rules specific for updating index, validating the update is allowed.
-   * It doesn't check whether the resulting index options will be valid.
+   * Validate update options. These are rules specific for updating index, validating the update
+   * is allowed. It doesn't check whether the resulting index options will be valid.
    *
    * @param oldOptions
    *   existing options
@@ -119,7 +123,10 @@ object FlintSparkSqlAstBuilder {
    * @param indexKind
    *   index kind
    */
-  private def validateUpdateOptions(oldOptions: Map[String, String], updateOptions: Map[String, String], indexKind: String): Unit = {
+  private def validateUpdateOptions(
+      oldOptions: Map[String, String],
+      updateOptions: Map[String, String],
+      indexKind: String): Unit = {
     val mergedOptions = oldOptions ++ updateOptions
     val newAutoRefresh = mergedOptions.getOrElse(AUTO_REFRESH.toString, "false")
     val oldAutoRefresh = oldOptions.getOrElse(AUTO_REFRESH.toString, "false")
@@ -135,17 +142,24 @@ object FlintSparkSqlAstBuilder {
       case ("false", "false") => FULL
       case ("false", "true") => INCREMENTAL
       case ("true", "true") =>
-        throw new IllegalArgumentException("auto_refresh and incremental_refresh options cannot both be true")
+        throw new IllegalArgumentException(
+          "auto_refresh and incremental_refresh options cannot both be true")
     }
 
     // validate allowed options depending on refresh mode
     val allowedOptions = refreshMode match {
       case FULL => Set(AUTO_REFRESH, INCREMENTAL_REFRESH)
       case AUTO | INCREMENTAL =>
-        Set(AUTO_REFRESH, INCREMENTAL_REFRESH, REFRESH_INTERVAL, CHECKPOINT_LOCATION, WATERMARK_DELAY)
+        Set(
+          AUTO_REFRESH,
+          INCREMENTAL_REFRESH,
+          REFRESH_INTERVAL,
+          CHECKPOINT_LOCATION,
+          WATERMARK_DELAY)
     }
     if (!updateOptions.keys.forall(allowedOptions.map(_.toString).contains)) {
-      throw new IllegalArgumentException(s"Altering ${indexKind} index to ${refreshMode} refresh only allows options: ${allowedOptions}")
+      throw new IllegalArgumentException(
+        s"Altering ${indexKind} index to ${refreshMode} refresh only allows options: ${allowedOptions}")
     }
   }
 }
diff --git a/integ-test/src/test/scala/org/opensearch/flint/spark/FlintSparkCoveringIndexSqlITSuite.scala b/integ-test/src/test/scala/org/opensearch/flint/spark/FlintSparkCoveringIndexSqlITSuite.scala
@@ -308,8 +308,7 @@ class FlintSparkCoveringIndexSqlITSuite extends FlintSparkSuite {
     flint.describeIndex(testFlintIndex) shouldBe defined
     flint.queryIndex(testFlintIndex).count() shouldBe 0
 
-    sql(
-      s"""
+    sql(s"""
          | ALTER INDEX $testIndex ON $testTable
          | WITH (auto_refresh = true)
          | """.stripMargin)

diff --git a/...test/src/test/scala/org/opensearch/flint/spark/FlintSparkMaterializedViewSqlITSuite.scala b/...test/src/test/scala/org/opensearch/flint/spark/FlintSparkMaterializedViewSqlITSuite.scala
@@ -296,8 +296,7 @@ class FlintSparkMaterializedViewSqlITSuite extends FlintSparkSuite {
       flint.describeIndex(testFlintIndex) shouldBe defined
       flint.queryIndex(testFlintIndex).count() shouldBe 0
 
-      sql(
-        s"""
+      sql(s"""
            | ALTER MATERIALIZED VIEW $testMvName
            | WITH (
            |   auto_refresh = true,

diff --git a/integ-test/src/test/scala/org/opensearch/flint/spark/FlintSparkSkippingIndexSqlITSuite.scala b/integ-test/src/test/scala/org/opensearch/flint/spark/FlintSparkSkippingIndexSqlITSuite.scala
@@ -355,8 +355,7 @@ class FlintSparkSkippingIndexSqlITSuite extends FlintSparkSuite {
   }
 
   test("update full refresh skipping index to auto refresh") {
-    sql(
-      s"""
+    sql(s"""
          | CREATE SKIPPING INDEX ON $testTable
          | (
          |   year PARTITION,
@@ -368,8 +367,7 @@ class FlintSparkSkippingIndexSqlITSuite extends FlintSparkSuite {
     flint.describeIndex(testIndex) shouldBe defined
     flint.queryIndex(testIndex).count() shouldBe 0
 
-    sql(
-      s"""
+    sql(s"""
          | ALTER SKIPPING INDEX ON $testTable
          | WITH (auto_refresh = true)
          | """.stripMargin)
@@ -382,8 +380,7 @@ class FlintSparkSkippingIndexSqlITSuite extends FlintSparkSuite {
 
   test("update incremental refresh skipping index to auto refresh") {
     withTempDir { checkpointDir =>
-      sql(
-        s"""
+      sql(s"""
            | CREATE SKIPPING INDEX ON $testTable
            | ( year PARTITION )
            | WITH (
@@ -397,15 +394,13 @@ class FlintSparkSkippingIndexSqlITSuite extends FlintSparkSuite {
       flint.queryIndex(testIndex).count() shouldBe 2
 
       // New data will be refreshed after updating index to auto refresh
-      sql(
-        s"""
+      sql(s"""
            | INSERT INTO $testTable
            | PARTITION (year=2023, month=5)
            | VALUES ('Hello', 50, 'Vancouver')
            |""".stripMargin)
 
-      sql(
-        s"""
+      sql(s"""
            | ALTER SKIPPING INDEX ON $testTable
            | WITH (
            |   auto_refresh = true,
@@ -421,8 +416,7 @@ class FlintSparkSkippingIndexSqlITSuite extends FlintSparkSuite {
   }
 
   test("update auto refresh skipping index to full refresh") {
-    sql(
-      s"""
+    sql(s"""
          | CREATE SKIPPING INDEX ON $testTable
          | ( year PARTITION )
          | WITH (auto_refresh = true)
@@ -435,17 +429,15 @@ class FlintSparkSkippingIndexSqlITSuite extends FlintSparkSuite {
     flint.describeIndex(testIndex) shouldBe defined
     flint.queryIndex(testIndex).count() shouldBe 2
 
-    sql(
-      s"""
+    sql(s"""
          | ALTER SKIPPING INDEX ON $testTable
          | WITH (auto_refresh = false)
          | """.stripMargin)
 
     spark.streams.active.find(_.name == testIndex) shouldBe empty
 
     // New data won't be refreshed until refresh statement triggered
-    sql(
-      s"""
+    sql(s"""
          | INSERT INTO $testTable
          | PARTITION (year=2023, month=5)
          | VALUES ('Hello', 50, 'Vancouver')
@@ -458,8 +450,7 @@ class FlintSparkSkippingIndexSqlITSuite extends FlintSparkSuite {
 
   test("update auto refresh skipping index to incremental refresh") {
     withTempDir { checkpointDir =>
-      sql(
-        s"""
+      sql(s"""
            | CREATE SKIPPING INDEX ON $testTable
            | ( year PARTITION )
            | WITH (
@@ -475,8 +466,7 @@ class FlintSparkSkippingIndexSqlITSuite extends FlintSparkSuite {
       flint.describeIndex(testIndex) shouldBe defined
       flint.queryIndex(testIndex).count() shouldBe 2
 
-      sql(
-        s"""
+      sql(s"""
            | ALTER SKIPPING INDEX ON $testTable
            | WITH (
            |   auto_refresh = false,
@@ -487,8 +477,7 @@ class FlintSparkSkippingIndexSqlITSuite extends FlintSparkSuite {
       spark.streams.active.find(_.name == testIndex) shouldBe empty
 
       // New data won't be refreshed until refresh statement triggered
-      sql(
-        s"""
+      sql(s"""
            | INSERT INTO $testTable
            | PARTITION (year=2023, month=5)
            | VALUES ('Hello', 50, 'Vancouver')