[Vended Dashboards] Enable config of meta data cache by default (#1123)

* [Vended Dashboards] Enable config of meta data cache for vended dashboards refresh

Signed-off-by: Jialiang Liang <jiallian@amazon.com>

* fix the IT

Signed-off-by: Jialiang Liang <jiallian@amazon.com>

* fix the lint

Signed-off-by: Jialiang Liang <jiallian@amazon.com>

---------

Signed-off-by: Jialiang Liang <jiallian@amazon.com>

Author: RyanL1997

flint-spark-integration/src/main/scala/org/apache/spark/sql/flint/config/FlintSparkConf.scala

@@ -314,7 +314,7 @@ object FlintSparkConf {
       .createOptional()
   val METADATA_CACHE_WRITE = FlintConfig("spark.flint.metadataCacheWrite.enabled")
     .doc("Enable Flint metadata cache write to Flint index mappings")
-    .createWithDefault("false")
+    .createWithDefault("true")
 
   val CUSTOM_SESSION_MANAGER =
     FlintConfig("spark.flint.job.customSessionManager")

flint-spark-integration/src/test/scala/org/apache/spark/FlintSuite.scala

@@ -63,12 +63,12 @@ trait FlintSuite extends SharedSparkSession {
     }
   }
 
-  protected def withMetadataCacheWriteEnabled(block: => Unit): Unit = {
-    setFlintSparkConf(METADATA_CACHE_WRITE, "true")
+  protected def withMetadataCacheWriteDisabled(block: => Unit): Unit = {
+    setFlintSparkConf(METADATA_CACHE_WRITE, "false")
     try {
       block
     } finally {
-      setFlintSparkConf(METADATA_CACHE_WRITE, "false")
+      setFlintSparkConf(METADATA_CACHE_WRITE, "true")
     }
   }
 

integ-test/src/integration/scala/org/opensearch/flint/spark/metadatacache/FlintOpenSearchMetadataCacheWriterITSuite.scala

@@ -63,15 +63,15 @@ class FlintOpenSearchMetadataCacheWriterITSuite extends FlintSparkSuite with Mat
   }
 
   test("build disabled metadata cache writer") {
-    FlintMetadataCacheWriterBuilder
-      .build(FlintSparkConf()) shouldBe a[FlintDisabledMetadataCacheWriter]
+    withMetadataCacheWriteDisabled {
+      FlintMetadataCacheWriterBuilder
+        .build(FlintSparkConf()) shouldBe a[FlintDisabledMetadataCacheWriter]
+    }
   }
 
   test("build opensearch metadata cache writer") {
-    withMetadataCacheWriteEnabled {
-      FlintMetadataCacheWriterBuilder
-        .build(FlintSparkConf()) shouldBe a[FlintOpenSearchMetadataCacheWriter]
-    }
+    FlintMetadataCacheWriterBuilder
+      .build(FlintSparkConf()) shouldBe a[FlintOpenSearchMetadataCacheWriter]
   }
 
   test("write metadata cache to index mappings") {
@@ -345,67 +345,62 @@ class FlintOpenSearchMetadataCacheWriterITSuite extends FlintSparkSuite with Mat
          |""".stripMargin)).foreach { case (refreshMode, optionsMap, expectedJson) =>
     test(s"write metadata cache for $refreshMode") {
       withExternalSchedulerEnabled {
-        withMetadataCacheWriteEnabled {
-          val flint: FlintSpark = new FlintSpark(spark)
-          withTempDir { checkpointDir =>
-            // update checkpoint_location if available in optionsMap
-            val indexOptions = FlintSparkIndexOptions(
-              optionsMap
-                .get("checkpoint_location")
-                .map(_ =>
-                  optionsMap.updated("checkpoint_location", checkpointDir.getAbsolutePath))
-                .getOrElse(optionsMap))
-
-            flint
-              .skippingIndex()
-              .onTable(testTable)
-              .addMinMax("age")
-              .options(indexOptions, testFlintIndex)
-              .create()
-
-            val propertiesJson = compact(render(getPropertiesJValue(testFlintIndex)))
-            propertiesJson should matchJson(expectedJson)
-
-            flint.refreshIndex(testFlintIndex)
-            val lastRefreshTime =
-              compact(render(getPropertiesJValue(testFlintIndex) \ "lastRefreshTime")).toLong
-            lastRefreshTime should be > 0L
-          }
+        val flint: FlintSpark = new FlintSpark(spark)
+        withTempDir { checkpointDir =>
+          // update checkpoint_location if available in optionsMap
+          val indexOptions = FlintSparkIndexOptions(
+            optionsMap
+              .get("checkpoint_location")
+              .map(_ => optionsMap.updated("checkpoint_location", checkpointDir.getAbsolutePath))
+              .getOrElse(optionsMap))
+
+          flint
+            .skippingIndex()
+            .onTable(testTable)
+            .addMinMax("age")
+            .options(indexOptions, testFlintIndex)
+            .create()
+
+          val propertiesJson = compact(render(getPropertiesJValue(testFlintIndex)))
+          propertiesJson should matchJson(expectedJson)
+
+          flint.refreshIndex(testFlintIndex)
+          val lastRefreshTime =
+            compact(render(getPropertiesJValue(testFlintIndex) \ "lastRefreshTime")).toLong
+          lastRefreshTime should be > 0L
         }
       }
     }
   }
 
   test("write metadata cache for auto refresh index with internal scheduler") {
-    withMetadataCacheWriteEnabled {
-      val flint: FlintSpark = new FlintSpark(spark)
-      withTempDir { checkpointDir =>
-        flint
-          .skippingIndex()
-          .onTable(testTable)
-          .addMinMax("age")
-          .options(
-            FlintSparkIndexOptions(
-              Map(
-                "auto_refresh" -> "true",
-                "scheduler_mode" -> "internal",
-                "refresh_interval" -> "10 Minute",
-                "checkpoint_location" -> checkpointDir.getAbsolutePath)),
-            testFlintIndex)
-          .create()
-
-        val propertiesJson = compact(render(getPropertiesJValue(testFlintIndex)))
-        propertiesJson should matchJson(s"""
+    val flint: FlintSpark = new FlintSpark(spark)
+    withTempDir { checkpointDir =>
+      flint
+        .skippingIndex()
+        .onTable(testTable)
+        .addMinMax("age")
+        .options(
+          FlintSparkIndexOptions(
+            Map(
+              "auto_refresh" -> "true",
+              "scheduler_mode" -> "internal",
+              "refresh_interval" -> "10 Minute",
+              "checkpoint_location" -> checkpointDir.getAbsolutePath)),
+          testFlintIndex)
+        .create()
+
+      val propertiesJson = compact(render(getPropertiesJValue(testFlintIndex)))
+      propertiesJson should matchJson(s"""
             | {
             |   "metadataCacheVersion": "${FlintMetadataCache.metadataCacheVersion}",
             |   "refreshInterval": 600,
             |   "sourceTables": ["$testTable"]
             | }
             |""".stripMargin)
 
-        flint.refreshIndex(testFlintIndex)
-        compact(render(getPropertiesJValue(testFlintIndex))) should not include "lastRefreshTime"
-      }
+      flint.refreshIndex(testFlintIndex)
+      compact(render(getPropertiesJValue(testFlintIndex))) should not include "lastRefreshTime"
     }
   }