adding _source to index_mappings

ahkcs · ahkcs · commit 20022d6f115b · 2025-04-09T14:27:40.000-07:00
Signed-off-by: Kai Huang &lt;ahkcs@amazon.com&gt;
diff --git a/flint-commons/src/main/scala/org/opensearch/flint/common/metadata/FlintMetadata.scala b/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)
     }
diff --git a/flint-core/src/main/scala/org/opensearch/flint/core/storage/FlintOpenSearchIndexMetadataService.scala b/flint-core/src/main/scala/org/opensearch/flint/core/storage/FlintOpenSearchIndexMetadataService.scala
@@ -133,9 +133,16 @@ object FlintOpenSearchIndexMetadataService {
             optionalObjectField(builder, "options", metadata.options)
             optionalObjectField(builder, "properties", metadata.properties)
           }
-        }
+          // extract index mappings options from metadata.options
+          // turn into object from string
+          // obj[_source] exists -> set the following
 
+          // mapping fields options / others
+        }
         // Add properties (schema) field
+        objectField(builder, "_source") {
+          builder.field("enabled", metadata.indexMappingsSourceEnabled)
+        }
         builder.field("properties", metadata.schema)
       })
     } catch {
@@ -191,6 +198,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.
diff --git a/flint-spark-integration/src/main/scala/org/opensearch/flint/spark/FlintSparkIndex.scala b/flint-spark-integration/src/main/scala/org/opensearch/flint/spark/FlintSparkIndex.scala
@@ -180,6 +180,40 @@ object FlintSparkIndex extends Logging {
       builder.indexSettings(settings.get)
     }
 
+    val mappings = index.options.indexMappings()
+    // parse into object
+    // extract source enabled
+    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 "source" => builder.source(parser.text())
+                //              case "indexedColumns" =>
+                //                parseArrayField(parser) {
+                //                  builder.addIndexedColumn(parser.map())
+                //                }
+                //              case "options" => builder.options(parser.map())
+                //              case "properties" => builder.properties(parser.map())
+                case _ => // Handle other fields as needed
+              }
+            }
+            }
+          //        case "_source" =>
+          //          builder.schema(parser.map())
+          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) {
diff --git a/flint-spark-integration/src/main/scala/org/opensearch/flint/spark/FlintSparkIndexOptions.scala b/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")
   }
diff --git a/integ-test/src/integration/scala/org/opensearch/flint/spark/FlintSparkCoveringIndexSqlITSuite.scala b/integ-test/src/integration/scala/org/opensearch/flint/spark/FlintSparkCoveringIndexSqlITSuite.scala
@@ -144,6 +144,23 @@ 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 } }',
+           |   index_settings = '{"number_of_shards": 2, "number_of_replicas": 3}'
+           | )
+           |""".stripMargin)
+
+    // Check if the index setting option is set to OS index setting
+    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"""
diff --git a/integ-test/src/integration/scala/org/opensearch/flint/spark/FlintSparkMaterializedViewSqlITSuite.scala b/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,26 @@ 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 } }',
+           |   index_settings = '{"number_of_shards": 3, "number_of_replicas": 2}'
+           | )
+           |""".stripMargin)
+
+    // Check if the index setting option is set to OS index setting
+    val flintIndexMetadataService =
+      new FlintOpenSearchIndexMetadataService(new FlintOptions(openSearchOptions.asJava))
+
+    val mappingsSourceEnabled = flintIndexMetadataService.getIndexMetadata(testFlintIndex).indexMappingsSourceEnabled
+    mappingsSourceEnabled shouldBe false
   }
 
   test("create materialized view with full refresh") {
diff --git a/integ-test/src/integration/scala/org/opensearch/flint/spark/FlintSparkSkippingIndexSqlITSuite.scala b/integ-test/src/integration/scala/org/opensearch/flint/spark/FlintSparkSkippingIndexSqlITSuite.scala
@@ -209,6 +209,24 @@ 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 } }',
+           |   index_settings = '{"number_of_shards": 3, "number_of_replicas": 2}'
+           | )
+           |""".stripMargin)
+
+    // Check if the index setting option is set to OS index setting
+    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
