Add more javadoc

Signed-off-by: Chen Dai <[email protected]>
opensearch-project · Oct 13, 2023 · 24f2296 · 24f2296
1 parent a6121fd
commit 24f2296
Show file tree

Hide file tree

Showing 4 changed files with 26 additions and 5 deletions.
diff --git a/flint-spark-integration/src/main/scala/org/opensearch/flint/spark/FlintSpark.scala b/flint-spark-integration/src/main/scala/org/opensearch/flint/spark/FlintSpark.scala
@@ -137,6 +137,7 @@ class FlintSpark(val spark: SparkSession) {
         batchRefresh()
         None
 
+      // Flint index has specialized logic and capability for incremental refresh
       case INCREMENTAL if index.isInstanceOf[StreamingRefresh] =>
         val job =
           index
@@ -152,6 +153,7 @@ class FlintSpark(val spark: SparkSession) {
             .start(indexName)
         Some(job.id.toString)
 
+      // Otherwise, fall back to foreachBatch + batch refresh
       case INCREMENTAL =>
         val job = spark.readStream
           .table(tableName)
@@ -256,6 +258,7 @@ class FlintSpark(val spark: SparkSession) {
     val indexOptions = FlintSparkIndexOptions(
       metadata.options.asScala.mapValues(_.asInstanceOf[String]).toMap)
 
+    // Convert generic Map[String,AnyRef] in metadata to specific data structure in Flint index
     metadata.kind match {
       case SKIPPING_INDEX_TYPE =>
         val strategies = metadata.indexedColumns.map { colInfo =>

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
@@ -120,7 +120,7 @@ object FlintSparkIndex {
     builder.kind(index.kind)
     builder.options(index.options.options.mapValues(_.asInstanceOf[AnyRef]).asJava)
 
-    // Index properties
+    // Optional index properties
     val envs = populateEnvToMetadata
     if (envs.nonEmpty) {
       builder.addProperty("env", envs.asJava)

diff --git a/...integration/src/main/scala/org/opensearch/flint/spark/mv/FlintSparkMaterializedView.scala b/...integration/src/main/scala/org/opensearch/flint/spark/mv/FlintSparkMaterializedView.scala
@@ -22,6 +22,18 @@ import org.apache.spark.sql.catalyst.util.IntervalUtils
 import org.apache.spark.sql.flint.logicalPlanToDataFrame
 import org.apache.spark.unsafe.types.UTF8String
 
+/**
+ * Flint materialized view in Spark.
+ *
+ * @param mvName
+ *   MV name
+ * @param query
+ *   source query that generates MV data
+ * @param outputSchema
+ *   output schema
+ * @param options
+ *   index options
+ */
 case class FlintSparkMaterializedView(
     mvName: String,
     query: String,

diff --git a/integ-test/src/test/scala/org/opensearch/flint/spark/FlintSparkMaterializedViewITSuite.scala b/integ-test/src/test/scala/org/opensearch/flint/spark/FlintSparkMaterializedViewITSuite.scala
@@ -42,10 +42,13 @@ class FlintSparkMaterializedViewITSuite extends FlintSparkSuite {
   }
 
   test("create materialized view with metadata successfully") {
+    val indexOptions =
+      FlintSparkIndexOptions(Map("auto_refresh" -> "true", "checkpoint_location" -> "s3://test/"))
     flint
       .materializedView()
       .name(testMvName)
       .query(testQuery)
+      .options(indexOptions)
       .create()
 
     val index = flint.describeIndex(testFlintIndex)
@@ -65,7 +68,10 @@ class FlintSparkMaterializedViewITSuite extends FlintSparkSuite {
          |      "columnName": "count",
          |      "columnType": "long"
          |    }],
-         |    "options": {},
+         |    "options": {
+         |      "auto_refresh": "true",
+         |      "checkpoint_location": "s3://test/"
+         |    },
          |    "properties": {}
          |  },
          |  "properties": {
@@ -102,13 +108,13 @@ class FlintSparkMaterializedViewITSuite extends FlintSparkSuite {
 
   test("incremental refresh materialized view") {
     withTempDir { checkpointDir =>
-      val checkpointOption =
-        FlintSparkIndexOptions(Map("checkpoint_location" -> checkpointDir.getAbsolutePath))
+      val indexOptions = FlintSparkIndexOptions(
+        Map("auto_refresh" -> "true", "checkpoint_location" -> checkpointDir.getAbsolutePath))
       flint
         .materializedView()
         .name(testMvName)
         .query(testQuery)
-        .options(checkpointOption)
+        .options(indexOptions)
         .create()
 
       flint