adding _source to index_mappings

Signed-off-by: Kai Huang <ahkcs@amazon.com>

Author: ahkcs

flint-commons/src/main/scala/org/opensearch/flint/common/metadata/FlintMetadata.scala

@@ -41,7 +41,11 @@ case class FlintMetadata(
      */
     latestLogEntry: Option[FlintMetadataLogEntry] = None,
     /** Optional Flint index settings. TODO: move elsewhere? */
-    indexSettings: Option[String]) {
+    indexSettings: Option[String],
+    /** Optional Flint index mappings.*/
+    indexMappings: Option[String],
+    /** Optional Flint index mappings _source field*/
+    indexMappingsSourceEnabled: Boolean) {
 
   require(version != null, "version is required")
   require(name != null, "name is required")
@@ -69,6 +73,8 @@ object FlintMetadata {
     private var latestId: Option[String] = None
     private var latestLogEntry: Option[FlintMetadataLogEntry] = None
     private var indexSettings: Option[String] = None
+    private var indexMappings: Option[String] = None
+    private var indexMappingsSourceEnabled = true
 
     def version(version: FlintVersion): this.type = {
       this.version = version
@@ -131,6 +137,16 @@ object FlintMetadata {
       this
     }
 
+    def indexMappings(indexMappings: String): this.type = {
+      this.indexMappings = Option(indexMappings)
+      this
+    }
+
+    def indexMappingsSourceEnabled(indexMappingsSourceEnabled: Boolean): this.type = {
+      this.indexMappingsSourceEnabled = indexMappingsSourceEnabled
+      this
+    }
+
     // Build method to create the FlintMetadata instance
     def build(): FlintMetadata = {
       FlintMetadata(
@@ -143,6 +159,8 @@ object FlintMetadata {
         properties = properties,
         schema = schema,
         indexSettings = indexSettings,
+        indexMappings = indexMappings,
+        indexMappingsSourceEnabled = indexMappingsSourceEnabled,
         latestId = latestId,
         latestLogEntry = latestLogEntry)
     }

flint-core/src/main/scala/org/opensearch/flint/core/storage/FlintOpenSearchIndexMetadataService.scala

@@ -134,7 +134,10 @@ object FlintOpenSearchIndexMetadataService {
             optionalObjectField(builder, "properties", metadata.properties)
           }
         }
-
+        // add _source field
+        objectField(builder, "_source") {
+          builder.field("enabled", metadata.indexMappingsSourceEnabled)
+        }
         // Add properties (schema) field
         builder.field("properties", metadata.schema)
       })
@@ -191,6 +194,14 @@ object FlintOpenSearchIndexMetadataService {
                   }
                 }
               }
+            case "_source" =>
+              parseObjectField(parser) { (parser, innerFieldName) =>
+              {
+                innerFieldName match {
+                  case "enabled" => builder.indexMappingsSourceEnabled(parser.booleanValue())
+                  case _ => // Handle other fields as needed
+                }
+              }}
             case "properties" =>
               builder.schema(parser.map())
             case _ => // Ignore other fields, for instance, dynamic.

flint-spark-integration/src/main/scala/org/opensearch/flint/spark/FlintSparkIndex.scala

@@ -180,6 +180,32 @@ object FlintSparkIndex extends Logging {
       builder.indexSettings(settings.get)
     }
 
+    //optional index mappings
+    val mappings = index.options.indexMappings()
+    // parse from json to object
+    // extract sourceEnabled
+    if(mappings.isDefined) {
+      parseJson(mappings.getOrElse("")) { (parser, fieldName) =>
+      {
+        fieldName match {
+          case "_source" =>
+            parseObjectField(parser) { (parser, innerFieldName) =>
+            {
+              innerFieldName match {
+                case "enabled" => builder.indexMappingsSourceEnabled(parser.booleanValue())
+                case _ => // Handle other fields as needed
+              }
+            }
+            }
+          case _ => // Ignore other fields, for instance, dynamic.
+        }
+      }
+      }
+    }
+    if (mappings.isDefined) {
+      builder.indexMappings(mappings.get)
+    }
+
     // Optional latest metadata log entry
     val latestLogEntry = index.latestLogEntry
     if (latestLogEntry.isDefined) {

flint-spark-integration/src/main/scala/org/opensearch/flint/spark/FlintSparkIndexOptions.scala

@@ -10,7 +10,7 @@ import java.util.{Collections, UUID}
 import org.json4s.{Formats, NoTypeHints}
 import org.json4s.native.JsonMethods._
 import org.json4s.native.Serialization
-import org.opensearch.flint.spark.FlintSparkIndexOptions.OptionName.{AUTO_REFRESH, CHECKPOINT_LOCATION, EXTRA_OPTIONS, ID_EXPRESSION, INCREMENTAL_REFRESH, INDEX_SETTINGS, OptionName, OUTPUT_MODE, REFRESH_INTERVAL, SCHEDULER_MODE, WATERMARK_DELAY}
+import org.opensearch.flint.spark.FlintSparkIndexOptions.OptionName.{AUTO_REFRESH, CHECKPOINT_LOCATION, EXTRA_OPTIONS, ID_EXPRESSION, INCREMENTAL_REFRESH, INDEX_MAPPINGS, INDEX_SETTINGS, OptionName, OUTPUT_MODE, REFRESH_INTERVAL, SCHEDULER_MODE, WATERMARK_DELAY}
 import org.opensearch.flint.spark.FlintSparkIndexOptions.validateOptionNames
 import org.opensearch.flint.spark.refresh.FlintSparkIndexRefresh.SchedulerMode
 import org.opensearch.flint.spark.scheduler.util.IntervalSchedulerParser
@@ -96,6 +96,14 @@ case class FlintSparkIndexOptions(options: Map[String, String]) {
    */
   def indexSettings(): Option[String] = getOptionValue(INDEX_SETTINGS)
 
+  /**
+   * The index mappings for OpenSearch index created.
+   *
+   * @return
+   *   index mapping JSON
+   */
+  def indexMappings(): Option[String] = getOptionValue(INDEX_MAPPINGS)
+
   /**
    * An expression that generates unique value as index data row ID.
    *
@@ -195,6 +203,7 @@ object FlintSparkIndexOptions {
     val WATERMARK_DELAY: OptionName.Value = Value("watermark_delay")
     val OUTPUT_MODE: OptionName.Value = Value("output_mode")
     val INDEX_SETTINGS: OptionName.Value = Value("index_settings")
+    val INDEX_MAPPINGS: OptionName.Value = Value("index_mappings")
     val ID_EXPRESSION: OptionName.Value = Value("id_expression")
     val EXTRA_OPTIONS: OptionName.Value = Value("extra_options")
   }

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

@@ -144,6 +144,22 @@ class FlintSparkCoveringIndexSqlITSuite extends FlintSparkSuite {
     (settings \ "index.number_of_replicas").extract[String] shouldBe "3"
   }
 
+  test("create covering index with index mappings") {
+    sql(s"""
+           | CREATE INDEX $testIndex ON $testTable ( name )
+           | WITH (
+           |   index_mappings = '{ "_source": { "enabled": false } }',
+           | )
+           |""".stripMargin)
+
+    // Check if the _source in index mappings option is set to OS index mappings
+    val flintIndexMetadataService =
+      new FlintOpenSearchIndexMetadataService(new FlintOptions(openSearchOptions.asJava))
+
+    val mappingsSourceEnabled = flintIndexMetadataService.getIndexMetadata(testFlintIndex).indexMappingsSourceEnabled
+    mappingsSourceEnabled shouldBe false
+  }
+
   test("create covering index with invalid option") {
     the[IllegalArgumentException] thrownBy
       sql(s"""

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

@@ -211,7 +211,7 @@ class FlintSparkMaterializedViewSqlITSuite extends FlintSparkSuite {
              | CREATE MATERIALIZED VIEW $testMvName
              | AS $testQuery
              | WITH (
-             |   index_settings = '{"number_of_shards": 3, "number_of_replicas": 2}'
+             |   index_settings = '{"number_of_shards": 4, "number_of_replicas": 3}'
              | )
              |""".stripMargin)
 
@@ -222,8 +222,25 @@ class FlintSparkMaterializedViewSqlITSuite extends FlintSparkSuite {
     implicit val formats: Formats = Serialization.formats(NoTypeHints)
     val settings =
       parse(flintIndexMetadataService.getIndexMetadata(testFlintIndex).indexSettings.get)
-    (settings \ "index.number_of_shards").extract[String] shouldBe "3"
-    (settings \ "index.number_of_replicas").extract[String] shouldBe "2"
+    (settings \ "index.number_of_shards").extract[String] shouldBe "4"
+    (settings \ "index.number_of_replicas").extract[String] shouldBe "3"
+  }
+
+  test("create materialized view with index mappings") {
+    sql(s"""
+           | CREATE MATERIALIZED VIEW $testMvName
+           | AS $testQuery
+           | WITH (
+           |   index_mappings = '{ "_source": { "enabled": false } }',
+           | )
+           |""".stripMargin)
+
+    // Check if the _source in index mappings option is set to OS index mappings
+    val flintIndexMetadataService =
+      new FlintOpenSearchIndexMetadataService(new FlintOptions(openSearchOptions.asJava))
+
+    val mappingsSourceEnabled = flintIndexMetadataService.getIndexMetadata(testFlintIndex).indexMappingsSourceEnabled
+    mappingsSourceEnabled shouldBe false
   }
 
   test("create materialized view with full refresh") {

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

@@ -209,6 +209,23 @@ class FlintSparkSkippingIndexSqlITSuite extends FlintSparkSuite with ExplainSuit
     (settings \ "index.number_of_replicas").extract[String] shouldBe "2"
   }
 
+  test("create skipping index with index mappings") {
+    sql(s"""
+           | CREATE SKIPPING INDEX ON $testTable
+           | ( year PARTITION )
+           | WITH (
+           |   index_mappings = '{ "_source": { "enabled": false } }',
+           | )
+           |""".stripMargin)
+
+    // Check if the _source in index mappings option is set to OS index mappings
+    val flintIndexMetadataService =
+      new FlintOpenSearchIndexMetadataService(new FlintOptions(openSearchOptions.asJava))
+
+    val mappingsSourceEnabled = flintIndexMetadataService.getIndexMetadata(testIndex).indexMappingsSourceEnabled
+    mappingsSourceEnabled shouldBe false
+  }
+
   Seq(
     "struct_col.field1.subfield VALUE_SET, struct_col.field2 MIN_MAX",
     "`struct_col.field1.subfield` VALUE_SET, `struct_col.field2` MIN_MAX", // ensure previous hack still works