#62: Implement splitPath (#64)

* `splitPath` function implemented and used
* Updated UTs
* `keepEmptyFields: Boolean` to `splitPath`

Author: benedeki

README.md

@@ -74,6 +74,14 @@ _SchemaUtils_ provides methods for working with schemas, its comparison and alig
       SchemaUtils.appendPath(path, fieldName)
     ```
 
+5. Separates the field name components of a fully qualified column name as their hierarchy goes from root down to the
+deepest one.
+
+    ```scala
+      SchemaUtils.splitPath(columnName, keepEmptyFields = True)
+    ```
+
+
 ### JsonUtils
 
 _Json Utils_ provides methods for working with Json, both on input and output.

spark-commons/src/main/scala/za/co/absa/spark/commons/implicits/StructTypeImplicits.scala

@@ -22,6 +22,7 @@ import org.apache.spark.sql.types._
 import za.co.absa.spark.commons.adapters.TransformAdapter
 import za.co.absa.spark.commons.implicits.DataTypeImplicits.DataTypeEnhancements
 import za.co.absa.spark.commons.implicits.StructFieldImplicits.StructFieldEnhancements
+import za.co.absa.spark.commons.utils.SchemaUtils
 import za.co.absa.spark.commons.utils.SchemaUtils.{getAllArraySubPaths, isCommonSubPath}
 
 import scala.annotation.tailrec
@@ -102,7 +103,7 @@ object StructTypeImplicits {
         }
       }
 
-      val pathTokens = path.split('.').toList
+      val pathTokens = SchemaUtils.splitPath(path)
       Try{
         examineStructField(pathTokens.tail, schema(pathTokens.head))
       }.getOrElse(None)
@@ -192,7 +193,7 @@ object StructTypeImplicits {
      * @return true if the column is the only column in a struct
      */
     def isOnlyField(path: String): Boolean = {
-      val pathSegments = path.split('.')
+      val pathSegments = SchemaUtils.splitPath(path)
       evaluateConditionsForField(schema, pathSegments, path, applyArrayHelper = false, applyLeafCondition = true,
         field => field.fields.length == 1)
     }
@@ -330,7 +331,7 @@ object StructTypeImplicits {
         }
       }
 
-      val pathToks = path.split('.')
+      val pathToks = SchemaUtils.splitPath(path)
       helper(pathToks, Seq()).mkString(".")
     }
 
@@ -358,7 +359,7 @@ object StructTypeImplicits {
         }
       }
 
-      val pathToks = path.split("\\.")
+      val pathToks = SchemaUtils.splitPath(path)
       helper(pathToks, Seq(), Seq())
     }
 
@@ -411,7 +412,7 @@ object StructTypeImplicits {
      * @return true if a field is an array that is not nested in another array
      */
     def isNonNestedArray(path: String): Boolean = {
-      val pathSegments = path.split('.')
+      val pathSegments = SchemaUtils.splitPath(path)
       evaluateConditionsForField(schema, pathSegments, path, applyArrayHelper = false)
     }
   }

spark-commons/src/main/scala/za/co/absa/spark/commons/utils/ExplodeTools.scala

@@ -252,7 +252,7 @@ object ExplodeTools {
     }
 
     val newFieldName = inputDf.schema.getClosestUniqueName(deconstructedColumnName)
-    val resultDf = inputDf.select(processStruct(inputDf.schema, columnName.split('.'), None)
+    val resultDf = inputDf.select(processStruct(inputDf.schema, SchemaUtils.splitPath(columnName), None)
       :+ col(columnName).as(newFieldName): _*)
     DeconstructedNestedField(resultDf, newFieldName, transientColName)
   }
@@ -287,7 +287,7 @@ object ExplodeTools {
       val newFields2 = if (isColumnToFound) newFields else newFields :+ col(columnFrom).as(columnTo)
       inputDf.select(newFields2: _*)
     } else {
-      putFieldIntoNestedStruct(inputDf, columnFrom, columnTo.split('.'), positionColumn)
+      putFieldIntoNestedStruct(inputDf, columnFrom, SchemaUtils.splitPath(columnTo), positionColumn)
     }
   }
 
@@ -342,7 +342,7 @@ object ExplodeTools {
 
   private def addSuperTransientField(inputDf: DataFrame, arrayColPathName: String): (DataFrame, String) = {
     val colName = inputDf.schema.getClosestUniqueName(superTransientColumnName)
-    val nestedColName = (arrayColPathName.split('.').dropRight(1) :+ colName).mkString(".")
+    val nestedColName = (SchemaUtils.splitPath(arrayColPathName).dropRight(1) :+ colName).mkString(".")
     val df = inputDf.nestedWithColumn(nestedColName, lit(null))
     (df, nestedColName)
   }

spark-commons/src/main/scala/za/co/absa/spark/commons/utils/SchemaUtils.scala

@@ -74,7 +74,7 @@ object SchemaUtils {
     }
 
     var isParentCommon = true // For Seq() the property holds by [my] convention
-    var restOfPaths: Seq[Seq[String]] = paths.map(_.split('.').toSeq).filter(_.nonEmpty)
+    var restOfPaths: Seq[Seq[String]] = paths.map(SchemaUtils.splitPath).filter(_.nonEmpty)
     while (isParentCommon && restOfPaths.nonEmpty) {
       val parent = restOfPaths.head.head
       isParentCommon = restOfPaths.forall(path => path.head == parent)
@@ -100,6 +100,44 @@ object SchemaUtils {
     }
   }
 
+
+  /**
+   * Separates the field name components of a fully qualified column name as their hierarchy goes from root down to the
+   * deepest one. No validation on the field names is done
+   * Example: `"com.my.package.xyz"` -> `List("com", "my", "package", "xyz")`
+   * Trailing '.' is ignored, leading one not.
+   *
+   * @param columnName      A fully qualified column name
+   * @return Each level field name in sequence how they go from root to the lowest one
+   */
+  def splitPath(columnName: String): List[String] = splitPath(columnName, keepEmptyFields = true)
+
+  /**
+   * Separates the field name components of a fully qualified column name as their hierarchy goes from root down to the
+   * deepest one. No validation on the field names is done
+   * Function is rather overloaded than using default parameter for easier use in functions like `map`
+   * Example: `"com.my.package.xyz"` -> `List("com", "my", "package", "xyz")`
+   * Trailing '.' is ignored, leading one not.
+   *
+   * @param columnName      A fully qualified column name
+   * @param keepEmptyFields If `false` any empty field names are removed from the result list, otherwise kept
+   * @return Each level field name in sequence how they go from root to the lowest one
+   */
+  def splitPath(columnName: String, keepEmptyFields: Boolean): List[String] = {
+    val stripped = columnName.stripSuffix(".")
+
+    if (stripped.isEmpty) {
+      List.empty
+    } else {
+      val segments = stripped.split('.').toList
+      if (keepEmptyFields) {
+        segments
+      } else {
+        segments.filterNot(_.isEmpty)
+      }
+    }
+  }
+
   private def columnPathAndCore(columnName: String): (String, String) = {
     val index = columnName.lastIndexOf('.')
     if (index >= 0) {

spark-commons/src/test/scala/za/co/absa/spark/commons/utils/SchemaUtilsTest.scala

@@ -46,4 +46,21 @@ class SchemaUtilsTest extends AnyFunSuite with Matchers {
     assert (!isCommonSubPath("a.b.c.d.e.f", "a.b.c.x", "a.b.c", "a.b", "a"))
   }
 
+  test("Test splitPath") {
+    assertResult(List("a", "b", "c", "d", "e"))(splitPath("a.b.c.d.e"))
+    assertResult(List("a"))(splitPath("a"))
+    assertResult(List("a", "bcd"))(splitPath("a.bcd"))
+    assertResult(List("a", "bcd"))(splitPath("a.bcd."))
+    assertResult(List("", "a", "bcd"))(splitPath(".a.bcd"))
+    assertResult(List.empty[String])(splitPath(""))
+    assertResult(List.empty[String])(splitPath("."))
+  }
+
+  test("Test splitPath with removing empty fields") {
+    assertResult(List("a", "b", "c", "d", "e"))(splitPath("a.b.c.d.e", keepEmptyFields = false))
+    assertResult(List("a", "e"))(splitPath("a....e", keepEmptyFields = false))
+    assertResult(List("a", "bcd"))(splitPath(".a.bcd", keepEmptyFields = false))
+    assertResult(List.empty[String])(splitPath("", keepEmptyFields = false))
+    assertResult(List.empty[String])(splitPath(".", keepEmptyFields = false))
+  }
 }