Move analyze skipping index rules to config

build.sbt

@@ -130,7 +130,8 @@ lazy val flintSparkIntegration = (project in file("flint-spark-integration"))
       "org.scalatest" %% "scalatest-flatspec" % "3.2.15" % "test",
       "org.scalatestplus" %% "mockito-4-6" % "3.2.15.0" % "test",
       "com.stephenn" %% "scalatest-json-jsonassert" % "0.2.5" % "test",
-      "com.github.sbt" % "junit-interface" % "0.13.3" % "test"),
+      "com.github.sbt" % "junit-interface" % "0.13.3" % "test",
+      "com.typesafe" % "config" % "1.3.3"),
     libraryDependencies ++= deps(sparkVersion),
     // ANTLR settings
     Antlr4 / antlr4Version := "4.8",

flint-spark-integration/src/main/resources/skipping_index_recommendation.conf

@@ -0,0 +1,48 @@
+recommendation {
+    data_type_rules {
+        PARTITION {
+            skipping_type = PARTITION,
+            reason = "PARTITION data structure is recommended for partition columns"
+        },
+        BooleanType {
+            skipping_type = VALUE_SET,
+            reason = "VALUE_SET data structure is recommended for BooleanType columns"
+        },
+        IntegerType {
+            skipping_type = MIN_MAX,
+            reason = "MIN_MAX data structure is recommended for IntegerType columns"
+        },
+        LongType {
+            skipping_type = MIN_MAX,
+            reason = "MIN_MAX data structure is recommended for LongType columns"
+        },
+        ShortType {
+            skipping_type = MIN_MAX,
+            reason = "MIN_MAX data structure is recommended for ShortType columns"
+        },
+        DateType {
+            skipping_type = BLOOM_FILTER,
+            reason = "BLOOM_FILTER data structure is recommended for DateType columns"
+        },
+        TimestampType {
+            skipping_type = BLOOM_FILTER,
+            reason = "BLOOM_FILTER data structure is recommended for TimestampType columns"
+        },
+        StringType {
+            skipping_type = BLOOM_FILTER,
+            reason = "BLOOM_FILTER data structure is recommended for StringType columns"
+        },
+        VarcharType {
+            skipping_type = BLOOM_FILTER,
+            reason = "BLOOM_FILTER data structure is recommended for VarcharType columns"
+        },
+        CharType {
+            skipping_type = BLOOM_FILTER,
+            reason = "BLOOM_FILTER data structure is recommended for CharType columns"
+        },
+        StructType {
+            skipping_type = BLOOM_FILTER,
+            reason = "BLOOM_FILTER data structure is recommended for StructType columns"
+        }
+    }
+}

flint-spark-integration/src/main/scala/org/opensearch/flint/spark/skipping/recommendations/DataTypeSkippingStrategy.scala

@@ -7,27 +7,16 @@ package org.opensearch.flint.spark.skipping.recommendations
 
 import scala.collection.mutable.ArrayBuffer
 
-import org.opensearch.flint.spark.skipping.FlintSparkSkippingStrategy.SkippingKind.{BLOOM_FILTER, MIN_MAX, PARTITION, VALUE_SET}
+import com.typesafe.config.{Config, ConfigFactory}
 
 import org.apache.spark.sql.{Row, SparkSession}
 import org.apache.spark.sql.flint.{loadTable, parseTableName}
 
 class DataTypeSkippingStrategy extends AnalyzeSkippingStrategy {
 
-  val rules = Map(
-    "PARTITION" -> (PARTITION.toString, "PARTITION data structure is recommended for partition columns"),
-    "BooleanType" -> (VALUE_SET.toString, "VALUE_SET data structure is recommended for BooleanType columns"),
-    "IntegerType" -> (MIN_MAX.toString, "MIN_MAX data structure is recommended for IntegerType columns"),
-    "LongType" -> (MIN_MAX.toString, "MIN_MAX data structure is recommended for LongType columns"),
-    "ShortType" -> (MIN_MAX.toString, "MIN_MAX data structure is recommended for ShortType columns"),
-    "DateType" -> (BLOOM_FILTER.toString, "BLOOM_FILTER data structure is recommended for DateType columns"),
-    "TimestampType" -> (BLOOM_FILTER.toString, "BLOOM_FILTER data structure is recommended for TimestampType columns"),
-    "StringType" -> (BLOOM_FILTER.toString, "BLOOM_FILTER data structure is recommended for StringType columns"),
-    "VarcharType" -> (BLOOM_FILTER.toString, "BLOOM_FILTER data structure is recommended for VarcharType columns"),
-    "CharType" -> (BLOOM_FILTER.toString, "BLOOM_FILTER data structure is recommended for CharType columns"),
-    "StructType" -> (BLOOM_FILTER.toString, "BLOOM_FILTER data structure is recommended for StructType columns"))
-
   override def analyzeSkippingIndexColumns(tableName: String, spark: SparkSession): Seq[Row] = {
+    val rules: Config = ConfigFactory.load("skipping_index_recommendation.conf")
+
     val (catalog, ident) = parseTableName(spark, tableName)
     val table = loadTable(catalog, ident).getOrElse(
       throw new IllegalStateException(s"Table $tableName is not found"))
@@ -48,14 +37,16 @@ class DataTypeSkippingStrategy extends AnalyzeSkippingStrategy {
         result += Row(
           field.name,
           field.dataType.typeName,
-          rules("PARTITION")._1,
-          rules("PARTITION")._2)
-      } else if (rules.contains(field.dataType.toString)) {
+          rules.getString("recommendation.data_type_rules.PARTITION.skipping_type"),
+          rules.getString("recommendation.data_type_rules.PARTITION.reason"))
+      } else if (rules.hasPath("recommendation.data_type_rules." + field.dataType.toString)) {
         result += Row(
           field.name,
           field.dataType.typeName,
-          rules(field.dataType.toString)._1,
-          rules(field.dataType.toString)._2)
+          rules.getString(
+            "recommendation.data_type_rules." + field.dataType.toString + ".skipping_type"),
+          rules.getString(
+            "recommendation.data_type_rules." + field.dataType.toString + ".reason"))
       }
     }
     result