#38: ArrayTransformation might fail if column name us a number (#39)

* Upgrade to spark-commons 0.4.0
* method `ArrayTransformations.arrCol` replaced by _Spark Commons_' `col_of_path` function.

Author: benedeki

project/Dependencies.scala

@@ -33,8 +33,8 @@ object Dependencies {
     List(
       "org.apache.spark" %% "spark-core" % sparkVersion % Provided,
       "org.apache.spark" %% "spark-sql" % sparkVersion % Provided,
-      "za.co.absa" %% s"spark-commons-spark$sparkVersionUpToMinor" % "0.3.2" % Provided,
-      "za.co.absa" %% "spark-commons-test" % "0.3.2" % Test,
+      "za.co.absa" %% s"spark-commons-spark$sparkVersionUpToMinor" % "0.4.0" % Provided,
+      "za.co.absa" %% "spark-commons-test" % "0.4.0" % Test,
       "com.typesafe" % "config" % "1.4.1",
       "com.github.mrpowers" %% "spark-fast-tests" % sparkFastTestsVersion(scalaVersion) % Test,
       "org.scalatest" %% "scalatest" % "3.2.2" % Test

src/main/scala/za/co/absa/standardization/ArrayTransformations.scala

@@ -23,6 +23,7 @@ import org.apache.spark.sql.{Column, Dataset, Row, SparkSession}
 import org.slf4j.LoggerFactory
 import za.co.absa.spark.commons.implicits.StructTypeImplicits.StructTypeEnhancements
 import za.co.absa.spark.commons.utils.SchemaUtils
+import za.co.absa.spark.commons.sql.functions.col_of_path
 
 object ArrayTransformations {
   private val logger = LoggerFactory.getLogger(this.getClass)
@@ -61,7 +62,7 @@ object ArrayTransformations {
         column as tokens.head
       } // some other attribute
       else if (!columnName.startsWith(currPath)) {
-        arrCol(currPath)
+        col_of_path(currPath)
       } // partial match, keep going
       else if (topType.isEmpty) {
         struct(helper(tokens.tail, pathAcc ++ List(tokens.head))) as tokens.head
@@ -76,23 +77,12 @@ object ArrayTransformations {
             }
             struct(fields.map(field => helper((List(field) ++ tokens.tail).distinct, pathAcc :+ tokens.head) as field): _*) as tokens.head
           case _: ArrayType => throw new IllegalStateException("Cannot reconstruct array columns. Please use this within arrayTransform.")
-          case _: DataType  => arrCol(currPath) as tokens.head
+          case _: DataType  => col_of_path(currPath) as tokens.head
         }
       }
     }
 
     ds.withColumn(toks.head, helper(toks, Seq()))
   }
 
-  def arrCol(any: String): Column = {
-    val toks = any.replaceAll("\\[(\\d+)\\]", "\\.$1").split("\\.")
-    toks.tail.foldLeft(col(toks.head)){
-      case (acc, tok) =>
-        if (tok.matches("\\d+")) {
-          acc(tok.toInt)
-        } else {
-          acc(tok)
-        }
-    }
-  }
 }

src/test/scala/za/co/absa/standardization/ArrayTransformationsSuite.scala

@@ -79,10 +79,10 @@ class ArrayTransformationsSuite extends AnyFunSuite with SparkTestBase {
 
     val res = ArrayTransformations.flattenArrays(df, "a")
 
-    val exp = List(
+    val exp = Seq(
       Nested1Level(List(Some(1), None, Some(2), Some(3), Some(4), Some(5), Some(6))),
       Nested1Level(List()),
-      Nested1Level(null)).toSeq
+      Nested1Level(null))
 
     val resLocal = res.as[Nested1Level].collect().toSeq