Use indexes subdirectory for custom index system path

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

@@ -152,7 +152,7 @@ class IndexCollectionManager(
 
   private def indexLogManagers: Seq[IndexLogManager] = {
     val hadoopConf = spark.sessionState.newHadoopConf()
-    val rootPath = PathResolver(conf, hadoopConf).systemPath
+    val rootPath = PathResolver(conf, hadoopConf).indexLocationDir
     val fs = fileSystemFactory.create(rootPath, hadoopConf)
     val indexPaths: Seq[Path] = if (fs.exists(rootPath)) {
       fs.listStatus(rootPath).map(_.getPath)

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

@@ -19,12 +19,14 @@ package com.microsoft.hyperspace.index
 import org.apache.spark.sql.internal.SQLConf
 
 object IndexConstants {
-  val INDEXES_DIR = "indexes"
-
   // Config used for setting the system path, which is considered as a "root" path for Hyperspace;
   // e.g, indexes are created under the system path.
   val INDEX_SYSTEM_PATH = "spark.hyperspace.system.path"
 
+  // Config used for subdirectory name under the system path.
+  val INDEX_DIR_NAME = "spark.hyperspace.system.indexDirName"
+  val INDEX_DIR_NAME_DEFAULT = "hyperspace"
+
   // Config used to set the number of buckets for the index.
   val INDEX_NUM_BUCKETS_LEGACY = "spark.hyperspace.index.num.buckets"
   val INDEX_NUM_BUCKETS = "spark.hyperspace.index.numBuckets"

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

@@ -37,7 +37,7 @@ private[hyperspace] class PathResolver(conf: SQLConf, hadoopConf: Configuration)
    * @return resolved index path
    */
   def getIndexPath(name: String): Path = {
-    val root = systemPath
+    val root = indexLocationDir
     val fs = root.getFileSystem(hadoopConf)
     if (fs.exists(root)) {
       // Note that fs.exists() is case-sensitive in some platforms and case-insensitive
@@ -58,14 +58,21 @@ private[hyperspace] class PathResolver(conf: SQLConf, hadoopConf: Configuration)
   }
 
   /**
-   * Get the Hyperspace index system path.
+   * Get the Hyperspace index location dir path.
    *
-   * @return Hyperspace index system path.
+   * @return Hyperspace index location dir path.
    */
-  def systemPath: Path = {
-    val defaultIndexesPath =
-      new Path(conf.getConfString("spark.sql.warehouse.dir"), "indexes")
-    new Path(conf.getConfString(IndexConstants.INDEX_SYSTEM_PATH, defaultIndexesPath.toString))
+  def indexLocationDir: Path = {
+    val indexDirName =
+      conf.getConfString(IndexConstants.INDEX_DIR_NAME, IndexConstants.INDEX_DIR_NAME_DEFAULT)
+    val indexSystemPath = conf.getConfString(
+      IndexConstants.INDEX_SYSTEM_PATH,
+      conf.getConfString("spark.sql.warehouse.dir"))
+    if (indexDirName.isEmpty) {
+      new Path(indexSystemPath)
+    } else {
+      new Path(indexSystemPath, indexDirName)
+    }
   }
 }
 

src/test/scala/com/microsoft/hyperspace/index/HyperspaceSuite.scala

@@ -34,6 +34,7 @@ trait HyperspaceSuite extends SparkFunSuite with SparkInvolvedSuite {
     super.beforeAll()
     FileUtils.delete(systemPath)
     spark.conf.set(IndexConstants.INDEX_SYSTEM_PATH, systemPath.toUri.toString)
+    spark.conf.set(IndexConstants.INDEX_DIR_NAME, "")
     clearCache()
   }
 

src/test/scala/com/microsoft/hyperspace/index/IndexCollectionManagerTest.scala

@@ -71,6 +71,7 @@ class IndexCollectionManagerTest extends SparkFunSuite with SparkInvolvedSuite {
   override def beforeAll(): Unit = {
     super.beforeAll()
     spark.conf.set(IndexConstants.INDEX_SYSTEM_PATH, indexSystemPath)
+    spark.conf.set(IndexConstants.INDEX_DIR_NAME, "")
     when(mockFileSystemFactory.create(any[Path], any[Configuration])).thenReturn(mockFileSystem)
 
     indexCollectionManager = new IndexCollectionManager(