Merge append and delete actions on indexes on modified source data

src/main/scala/com/microsoft/hyperspace/Hyperspace.scala

@@ -19,6 +19,7 @@ package com.microsoft.hyperspace
 import org.apache.spark.sql.{DataFrame, SparkSession}
 
 import com.microsoft.hyperspace.index._
+import com.microsoft.hyperspace.index.IndexConstants.REFRESH_MODE_FULL
 import com.microsoft.hyperspace.index.plananalysis.PlanAnalyzer
 
 class Hyperspace(spark: SparkSession) {
@@ -74,7 +75,18 @@ class Hyperspace(spark: SparkSession) {
    * @param indexName Name of the index to refresh.
    */
   def refreshIndex(indexName: String): Unit = {
-    indexManager.refresh(indexName)
+    refreshIndex(indexName, REFRESH_MODE_FULL)
+  }
+
+  /**
+   * Update indexes for the latest version of the data. This api provides a few supported
+   * refresh modes as listed below.
+   *
+   * @param indexName Name of the index to refresh.
+   * @param mode Refresh mode. Currently supported modes are `incremental` and `full`.
+   */
+  def refreshIndex(indexName: String, mode: String): Unit = {
+    indexManager.refresh(indexName, mode)
   }
 
   /**

src/main/scala/com/microsoft/hyperspace/actions/RefreshAction.scala

@@ -37,9 +37,10 @@ class RefreshAction(
     extends RefreshActionBase(spark, logManager, dataManager) {
 
   final override def op(): Unit = {
-    // TODO: The current implementation picks the number of buckets from session config.
-    //   This should be user-configurable to allow maintain the existing bucket numbers
-    //   in the index log entry.
+    // TODO: The current implementation picks the number of buckets and
+    //  "spark.hyperspace.index.lineage.enabled" from session config. This should be
+    //  user-configurable to allow maintain the existing bucket numbers in the index log entry.
+    //  https://github.com/microsoft/hyperspace/issues/196.
     write(spark, df, indexConfig)
   }
 

src/main/scala/com/microsoft/hyperspace/actions/RefreshAppendAction.scala

@@ -41,9 +41,10 @@ class RefreshAppendAction(
     dataManager: IndexDataManager)
     extends RefreshActionBase(spark, logManager, dataManager) {
   final override def op(): Unit = {
-    // TODO: The current implementation picks the number of buckets from session config.
-    //   This should be user-configurable to allow maintain the existing bucket numbers
-    //   in the index log entry.
+    // TODO: The current implementation picks the number of buckets and
+    //  "spark.hyperspace.index.lineage.enabled" from session config. This should be
+    //  user-configurable to allow maintain the existing bucket numbers in the index log entry.
+    //  https://github.com/microsoft/hyperspace/issues/196.
     write(spark, dfWithAppendedFiles, indexConfig)
   }
 

src/main/scala/com/microsoft/hyperspace/index/CachingIndexCollectionManager.scala

@@ -92,9 +92,9 @@ class CachingIndexCollectionManager(
     super.vacuum(indexName)
   }
 
-  override def refresh(indexName: String): Unit = {
+  override def refresh(indexName: String, mode: String): Unit = {
     clearCache()
-    super.refresh(indexName)
+    super.refresh(indexName, mode)
   }
 }
 

src/main/scala/com/microsoft/hyperspace/index/IndexCollectionManager.scala

@@ -22,7 +22,7 @@ import org.apache.spark.sql.internal.SQLConf
 
 import com.microsoft.hyperspace.HyperspaceException
 import com.microsoft.hyperspace.actions._
-import com.microsoft.hyperspace.util.HyperspaceConf
+import com.microsoft.hyperspace.index.IndexConstants.{REFRESH_MODE_FULL, REFRESH_MODE_INCREMENTAL}
 
 class IndexCollectionManager(
     spark: SparkSession,
@@ -63,16 +63,17 @@ class IndexCollectionManager(
     }
   }
 
-  override def refresh(indexName: String): Unit = {
+  override def refresh(indexName: String, mode: String): Unit = {
     withLogManager(indexName) { logManager =>
       val indexPath = PathResolver(spark.sessionState.conf).getIndexPath(indexName)
       val dataManager = indexDataManagerFactory.create(indexPath)
-      if (HyperspaceConf.refreshDeleteEnabled(spark)) {
+      if (mode.equalsIgnoreCase(REFRESH_MODE_INCREMENTAL)) {
         new RefreshDeleteAction(spark, logManager, dataManager).run()
-      } else if (HyperspaceConf.refreshAppendEnabled(spark)) {
         new RefreshAppendAction(spark, logManager, dataManager).run()
-      } else {
+      } else if (mode.equalsIgnoreCase(REFRESH_MODE_FULL)) {
         new RefreshAction(spark, logManager, dataManager).run()
+      } else {
+        throw HyperspaceException(s"Unsupported refresh mode '$mode' found.")
       }
     }
   }

src/main/scala/com/microsoft/hyperspace/index/IndexConstants.scala

@@ -73,16 +73,6 @@ object IndexConstants {
   val INDEX_LINEAGE_ENABLED = "spark.hyperspace.index.lineage.enabled"
   val INDEX_LINEAGE_ENABLED_DEFAULT = "false"
 
-  val REFRESH_DELETE_ENABLED = "spark.hyperspace.index.refresh.delete.enabled"
-  val REFRESH_DELETE_ENABLED_DEFAULT = "false"
-
-  /**
-   * This flag enables refreshing index if additional data files are appended to the source. When
-   * set to false, the refresh call will not run RefreshAppendAction. It will instead go for full
-   * refresh.
-   * This flag is temporary, and will be removed when both Append and Delete actions are merged
-   * for refreshing indexes.
-   */
-  val REFRESH_APPEND_ENABLED = "spark.hyperspace.index.refresh.append.enabled"
-  val REFRESH_APPEND_ENABLED_DEFAULT = "false"
+  val REFRESH_MODE_INCREMENTAL = "incremental"
+  val REFRESH_MODE_FULL = "full"
 }

src/main/scala/com/microsoft/hyperspace/index/IndexManager.scala

@@ -65,7 +65,7 @@ trait IndexManager {
    *
    * @param indexName Name of the index to refresh.
    */
-  def refresh(indexName: String): Unit
+  def refresh(indexName: String, mode: String): Unit
 
   /**
    * Cancel api to bring back index from an inconsistent state to the last known stable state.

src/main/scala/com/microsoft/hyperspace/util/HyperspaceConf.scala

@@ -39,18 +39,4 @@ object HyperspaceConf {
         IndexConstants.INDEX_HYBRID_SCAN_DELETE_ENABLED_DEFAULT)
       .toBoolean
   }
-
-  def refreshDeleteEnabled(spark: SparkSession): Boolean = {
-    spark.conf
-      .get(IndexConstants.REFRESH_DELETE_ENABLED,
-        IndexConstants.REFRESH_DELETE_ENABLED_DEFAULT)
-      .toBoolean
-  }
-
-  def refreshAppendEnabled(spark: SparkSession): Boolean = {
-    spark.conf
-      .get(IndexConstants.REFRESH_APPEND_ENABLED,
-        IndexConstants.REFRESH_APPEND_ENABLED_DEFAULT)
-      .toBoolean
-  }
 }

src/test/scala/com/microsoft/hyperspace/TestUtils.scala

@@ -16,11 +16,13 @@
 
 package com.microsoft.hyperspace
 
+import org.apache.hadoop.conf.Configuration
 import org.apache.hadoop.fs.Path
 
 import com.microsoft.hyperspace.MockEventLogger.reset
 import com.microsoft.hyperspace.index.IndexLogEntry
 import com.microsoft.hyperspace.telemetry.{EventLogger, HyperspaceEvent}
+import com.microsoft.hyperspace.util.FileUtils
 
 object TestUtils {
   def copyWithState(index: IndexLogEntry, state: String): IndexLogEntry = {
@@ -46,6 +48,28 @@ object TestUtils {
       path.getName +: splitPath(path.getParent)
     }
   }
+
+  /**
+   * Delete files from a given path.
+   *
+   * @param path Path to the folder containing files.
+   * @param pattern File name pattern to delete.
+   * @param numFilesToDelete Number of files to delete.
+   * @return Paths to the deleted file.
+   */
+  def deleteFiles(path: String, pattern: String, numFilesToDelete: Int): Seq[Path] = {
+    val pathToDelete = new Path(path, pattern)
+    val fileNames = pathToDelete
+      .getFileSystem(new Configuration)
+      .globStatus(pathToDelete)
+      .map(_.getPath)
+
+    assert(fileNames.length >= numFilesToDelete)
+    val filesToDelete = fileNames.take(numFilesToDelete)
+    filesToDelete.foreach(FileUtils.delete(_))
+
+    filesToDelete
+  }
 }
 
 /**