Repository: AbsaOSS/pramen (language: python)

Issue:
Pramen fails to esacpe ids in Hive DDL properly when complex nested schemas are involved 
## Describe the bug
An example input schema:

```
root
 |-- id: long (nullable = true)
 |-- legs: array (nullable = true)
 |    |-- element: struct (containsNull = true)
 |    |    |-- conditions: array (nullable = true)
 |    |    |    |-- element: struct (containsNull = true)
 |    |    |    |    |-- amount: decimal(18,4) (nullable = true)
 |    |    |    |    |-- checks: array (nullable = true)
 |    |    |    |    |    |-- element: struct (containsNull = true)
 |    |    |    |    |    |    |-- checkNums: array (nullable = true)
 |    |    |    |    |    |    |    |-- element: string (containsNull = true)
 |    |    |-- legid: long (nullable = true)
```

DDL produced:
```sql
`id` BIGINT COMMENT 'This is my table',
`legs` ARRAY<STRUCT<`conditions`: ARRAY<STRUCT<`amount`: DECIMAL(18,4),`` `checks`: ARRAY<STRUCT<`checkNums`: ARRAY<`STRING>` COMMENT 'decimal(10,`` 12)'>>>>,
`` `legid`: BIGINT COMMENT 'This is a \'test\': long'>>
```

DDL Expected:
```sql
`id` BIGINT COMMENT 'This is my table',`
legs` ARRAY<STRUCT<`conditions`: ARRAY<STRUCT<`amount`: DECIMAL(18,4), `checks`: ARRAY<STRUCT<`checkNums`: ARRAY<STRING> COMMENT 'decimal(10, 12)'>>>>, 
`legid`: BIGINT COMMENT 'This is a \'test\': long'>>
```

## Expected behavior
DDL escaping should work properly for complex data types.

## Context
- Pramen version 1.8.10
- Spark version: 3.4.1
- Scala version: 2.12


Pramen fails to esacpe ids in Hive DDL properly when complex nested schemas are involved 
## Describe the bug
An example input schema:

```
root
 |-- id: long (nullable = true)
 |-- legs: array (nullable = true)
 |    |-- element: struct (containsNull = true)
 |    |    |-- conditions: array (nullable = true)
 |    |    |    |-- element: struct (containsNull = true)
 |    |    |    |    |-- amount: decimal(18,4) (nullable = true)
 |    |    |    |    |-- checks: array (nullable = true)
 |    |    |    |    |    |-- element: struct (containsNull = true)
 |    |    |    |    |    |    |-- checkNums: array (nullable = true)
 |    |    |    |    |    |    |    |-- element: string (containsNull = true)
 |    |    |-- legid: long (nullable = true)
```

DDL produced:
```sql
`id` BIGINT COMMENT 'This is my table',
`legs` ARRAY<STRUCT<`conditions`: ARRAY<STRUCT<`amount`: DECIMAL(18,4),`` `checks`: ARRAY<STRUCT<`checkNums`: ARRAY<`STRING>` COMMENT 'decimal(10,`` 12)'>>>>,
`` `legid`: BIGINT COMMENT 'This is a \'test\': long'>>
```

DDL Expected:
```sql
`id` BIGINT COMMENT 'This is my table',`
legs` ARRAY<STRUCT<`conditions`: ARRAY<STRUCT<`amount`: DECIMAL(18,4), `checks`: ARRAY<STRUCT<`checkNums`: ARRAY<STRING> COMMENT 'decimal(10, 12)'>>>>, 
`legid`: BIGINT COMMENT 'This is a \'test\': long'>>
```

## Expected behavior
DDL escaping should work properly for complex data types.

## Context
- Pramen version 1.8.10
- Spark version: 3.4.1
- Scala version: 2.12




Discussion / hints:

Fix patch:
diff --git a/pramen/core/src/main/scala/za/co/absa/pramen/core/utils/SparkUtils.scala b/pramen/core/src/main/scala/za/co/absa/pramen/core/utils/SparkUtils.scala
index 30e80c6c6..1aab4b3be 100644
--- a/pramen/core/src/main/scala/za/co/absa/pramen/core/utils/SparkUtils.scala
+++ b/pramen/core/src/main/scala/za/co/absa/pramen/core/utils/SparkUtils.scala
@@ -208,6 +208,7 @@ object SparkUtils {
     * Transforms the schema to the format compatible with Hive-like catalogs.
     * - Removes non-nullable flag since it is not compatible with catalogs
     * - Switches from string to varchar(n) when the maximum field length is known
+    *
     * @param schema The input schema.
     * @return The transformed schema.
     */
@@ -391,43 +392,111 @@ object SparkUtils {
     * {{{
     *   `Id` INT NOT NULL,`Name` STRING,`System Date` ARRAY<STRING>
     * }}}
+    *
     * @param sparkDdlExpr An expression as a Spark DDL (df.schema.toDDL)
     * @return The same DDL with all column names escaped.
     */
   def escapeColumnsSparkDDL(sparkDdlExpr: String): String = {
+    val STATE_WHITESPACE_OR_ID = 0
+    val STATE_POSSIBLE_ID = 1
+    val STATE_DATA_TYPE = 2
+    val STATE_PARENTHESIS = 3
+    val STATE_QUOTES = 4
+    val STATE_ESCAPE = 5
+
     val output = new StringBuilder()
     val token = new StringBuilder()
 
     var state = 0
+    var depth = 0
     var i = 0
     val len = sparkDdlExpr.length
 
     while (i < len) {
       val c = sparkDdlExpr(i)
-      if (state == 0) {
-        if (c == ' ' || c == ':') {
-          state = 1
-          val tokenStr = token.toString()
-          if (tokenStr.nonEmpty && tokenStr.head == '`') {
-            output.append(s"$token$c")
+      if (state == STATE_WHITESPACE_OR_ID) {
+        if (c == '<' || c == ':') {
+          output.append(s"$token")
+          token.clear()
+          state = STATE_DATA_TYPE
+        } else if (c == ' ') {
+          token.append(c)
+        } else {
+          output.append(s"$token")
+          token.clear()
+          state = STATE_POSSIBLE_ID
+        }
+      }
+
+      if (state == 1) {
+        if (c == '(') {
+          depth = 1
+          state = STATE_PARENTHESIS
+          token.append(c)
+        } else if (c == '\'') {
+          state = STATE_QUOTES
+          token.append(c)
+        } else if (c == '\\') {
+          state = STATE_ESCAPE
+          token.append(c)
+        } else if (c == ' ' || c == ':') {
+          state = STATE_DATA_TYPE
+          val tokenStr = token.toString().trim
+          if (tokenStr.isEmpty) {
+            output.append(s"$tokenStr$c")
+          } else if (tokenStr.head == '`') {
+            output.append(s"$tokenStr$c")
           } else {
-            output.append(s"`$token`$c")
+            output.append(s"`$tokenStr`$c")
           }
           token.clear()
         } else if (c == ',' || c == '<') {
           output.append(s"$token$c")
           token.clear()
+          state = STATE_WHITESPACE_OR_ID
+        } else if (c == '>') {
+          output.append(s"$token$c")
+          token.clear()
+          state = STATE_DATA_TYPE
         } else {
           token.append(c)
         }
-      } else if (state == 1) {
-        if (c == ',' || c == '<') {
-          state = 0
+      } else if (state == STATE_DATA_TYPE) {
+        if (c == '(') {
+          depth = 1
+          state = STATE_PARENTHESIS
+        } else if (c == '\'') {
+          state = STATE_QUOTES
+        } else if (c == '\\') {
+          state = STATE_ESCAPE
+        }
+        if (c == ',') {
+          state = STATE_WHITESPACE_OR_ID
+          output.append(s"$token$c")
+          token.clear()
+        } else if (c == '<') {
+          state = STATE_WHITESPACE_OR_ID
           output.append(s"$token$c")
           token.clear()
         } else {
           token.append(c)
         }
+      } else if (state == STATE_PARENTHESIS) {
+        if (c == ')') {
+          depth -= 1
+          if (depth == 0)
+            state = STATE_DATA_TYPE
+        } else if (c == '(')
+          depth += 1
+        token.append(c)
+      } else if (state == STATE_QUOTES) {
+        if (c == '\'') {
+          state = STATE_DATA_TYPE
+        }
+        token.append(c)
+      } else if (state == STATE_ESCAPE) {
+        state = STATE_DATA_TYPE
+        token.append(c)
       }
       i += 1
     }


Test patch:
diff --git a/pramen/core/src/test/scala/za/co/absa/pramen/core/NestedDataFrameFactory.scala b/pramen/core/src/test/scala/za/co/absa/pramen/core/NestedDataFrameFactory.scala
new file mode 100644
index 000000000..cffae6b01
--- /dev/null
+++ b/pramen/core/src/test/scala/za/co/absa/pramen/core/NestedDataFrameFactory.scala
@@ -0,0 +1,93 @@
+/*
+ * Copyright 2022 ABSA Group Limited
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package za.co.absa.pramen.core
+
+import org.apache.spark.sql.types._
+import org.apache.spark.sql.{DataFrame, Row, SparkSession}
+
+object NestedDataFrameFactory {
+  val testCaseSchema: StructType = StructType(
+    Array(
+      StructField("id", LongType),
+      StructField("key1", LongType),
+      StructField("key2", LongType),
+      StructField("struct1", StructType(Array(
+        StructField("key3", IntegerType),
+        StructField("key4", IntegerType)
+      ))),
+      StructField("struct2", StructType(Array(
+        StructField("inner1", StructType(Array(
+          StructField("key5", LongType),
+          StructField("key6", LongType),
+          StructField("skey1", StringType)
+        )))
+      ))),
+      StructField("struct3", StructType(Array(
+        StructField("inner3", StructType(Array(
+          StructField("array3", ArrayType(StructType(Array(
+            StructField("a1", LongType),
+            StructField("a2", LongType),
+            StructField("a3", StringType)
+          ))))
+        )))
+      ))),
+      StructField("array1", ArrayType(StructType(Array(
+        StructField("key7", LongType),
+        StructField("key8", LongType),
+        StructField("skey2", StringType)
+      )))),
+      StructField("array2", ArrayType(StructType(Array(
+        StructField("key2", LongType),
+        StructField("inner2", ArrayType(StructType(Array(
+          StructField("key9", LongType),
+          StructField("key10", LongType),
+          StructField("struct3", StructType(Array(
+            StructField("k1", IntegerType),
+            StructField("k2", IntegerType)
+          )))
+        ))))
+      ))))
+    ))
+
+  private val testCaseWithMapSchema = new StructType()
+    .add("name", StringType)
+    .add("addresses", ArrayType(new StructType()
+      .add("city", StringType)
+      .add("state", StringType)))
+    .add("properties", MapType(StringType, StringType))
+
+  private val tstCaseWithMapData = Seq(
+    Row("John", List(Row("Newark", "NY"), Row("Brooklyn", "NY")), Map("hair" -> "black", "eyes" -> "brown", "height" -> "178")),
+    Row("Kate", List(Row("San Jose", "CA"), Row("Sandiago", "CA")), Map("hair" -> "brown", "eyes" -> "black", "height" -> "165")),
+    Row("William", List(Row("Las Vegas", "NV")), Map("hair" -> "red", "eye" -> "gray", "height" -> "185")),
+    Row("Sarah", null, Map("hair" -> "blond", "eyes" -> "red", "height" -> "162")),
+    Row("Michael", List(Row("Sacramento", "CA"), Row("San Diego", "CA")), Map("white" -> "black", "eyes" -> "black", "height" -> "180"))
+  )
+
+  def getNestedTestCase(implicit spark: SparkSession): DataFrame = {
+    spark.read
+      .schema(testCaseSchema)
+      .json(getClass.getResource("/test/nestedDf1.json").getPath)
+  }
+
+  def getMapTestCase(implicit spark: SparkSession): DataFrame = {
+    spark.createDataFrame(
+      spark.sparkContext.parallelize(tstCaseWithMapData),
+      testCaseWithMapSchema
+    ).orderBy("name")
+  }
+}
diff --git a/pramen/core/src/test/scala/za/co/absa/pramen/core/tests/utils/SparkUtilsSuite.scala b/pramen/core/src/test/scala/za/co/absa/pramen/core/tests/utils/SparkUtilsSuite.scala
index c79063f4d..5cd6a4f50 100644
--- a/pramen/core/src/test/scala/za/co/absa/pramen/core/tests/utils/SparkUtilsSuite.scala
+++ b/pramen/core/src/test/scala/za/co/absa/pramen/core/tests/utils/SparkUtilsSuite.scala
@@ -22,6 +22,7 @@ import org.apache.spark.sql.types._
 import org.apache.spark.sql.{DataFrame, Row}
 import org.scalatest.wordspec.AnyWordSpec
 import za.co.absa.pramen.api.FieldChange._
+import za.co.absa.pramen.core.NestedDataFrameFactory
 import za.co.absa.pramen.core.base.SparkTestBase
 import za.co.absa.pramen.core.fixtures.{TempDirFixture, TextComparisonFixture}
 import za.co.absa.pramen.core.pipeline.TransformExpression
@@ -641,6 +642,59 @@ class SparkUtilsSuite extends AnyWordSpec with SparkTestBase with TempDirFixture
 
       assert(actual == expectedDDL)
     }
+
+    "work with decimals and nested arrays of struct" in {
+      val expectedDDLWithNestedComments = "`id` BIGINT COMMENT 'This is my table',`legs` ARRAY<STRUCT<`conditions`: ARRAY<STRUCT<`amount`: DECIMAL(18,4), `checks`: ARRAY<STRUCT<`checkNums`: ARRAY<STRING> COMMENT 'decimal(10, 12)'>>>>, `legid`: BIGINT COMMENT 'This is a \\'test\\': long'>>"
+      val expectedDDLWithoutNestedComments = "`id` BIGINT COMMENT 'This is my table',`legs` ARRAY<STRUCT<`conditions`: ARRAY<STRUCT<`amount`: DECIMAL(18,4), `checks`: ARRAY<STRUCT<`checkNums`: ARRAY<STRING>>>>>, `legid`: BIGINT>>"
+
+      val comment1 = new MetadataBuilder().putString("comment", "This is my table").build()
+      val comment2 = new MetadataBuilder().putString("comment", "decimal(10, 12)").build()
+      val comment3 = new MetadataBuilder().putString("comment", "This is a 'test': long").build()
+      val schema = StructType(Array(
+        StructField("id", LongType, nullable = true, metadata = comment1),
+        StructField("legs", ArrayType(StructType(List(
+          StructField("conditions", ArrayType(StructType(List(
+            StructField("amount", DecimalType(18, 4), nullable = true),
+            StructField("checks", ArrayType(StructType(List(
+              StructField("checkNums", ArrayType(StringType, containsNull = true), nullable = true, metadata = comment2)
+            )), containsNull = true), nullable = true))), containsNull = true), nullable = true),
+          StructField("legid", LongType, nullable = true, metadata = comment3))), containsNull = true), nullable = true)))
+
+      val actualDDL = escapeColumnsSparkDDL(schema.toDDL)
+
+      // Depends on the version of Spark
+      if (actualDDL.contains("ARRAY<STRING> COMMENT")) {
+        assert(actualDDL == expectedDDLWithNestedComments)
+      } else {
+        assert(actualDDL == expectedDDLWithoutNestedComments)
+      }
+    }
+
+    "work with decimals and nested arrays of struct when the input is not escaped" in {
+      val inputDDL = "id BIGINT COMMENT 'This is my table',legs ARRAY<STRUCT<conditions: ARRAY<STRUCT<amount: DECIMAL(18,4), checks: ARRAY<STRUCT<checkNums: ARRAY<STRING> COMMENT 'decimal(10, 12)'>>>>, legid: BIGINT COMMENT 'This is a \\'test\\': long'>>"
+      val expectedDDL = "`id` BIGINT COMMENT 'This is my table',`legs` ARRAY<STRUCT<`conditions`: ARRAY<STRUCT<`amount`: DECIMAL(18,4), `checks`: ARRAY<STRUCT<`checkNums`: ARRAY<STRING> COMMENT 'decimal(10, 12)'>>>>, `legid`: BIGINT COMMENT 'This is a \\'test\\': long'>>"
+
+      val actualDDL = escapeColumnsSparkDDL(inputDDL)
+
+      assert(actualDDL == expectedDDL)
+    }
+
+    "work with another complex example" in {
+      val expectedDDL = "`id` BIGINT,`key1` BIGINT,`key2` BIGINT,`struct1` STRUCT<`key3`: INT, `key4`: INT>,`struct2` STRUCT<`inner1`: STRUCT<`key5`: BIGINT, `key6`: BIGINT, `skey1`: STRING>>,`struct3` STRUCT<`inner3`: STRUCT<`array3`: ARRAY<STRUCT<`a1`: BIGINT, `a2`: BIGINT, `a3`: STRING>>>>,`array1` ARRAY<STRUCT<`key7`: BIGINT, `key8`: BIGINT, `skey2`: STRING>>,`array2` ARRAY<STRUCT<`key2`: BIGINT, `inner2`: ARRAY<STRUCT<`key9`: BIGINT, `key10`: BIGINT, `struct3`: STRUCT<`k1`: INT, `k2`: INT>>>>>"
+
+      val actualDDL = escapeColumnsSparkDDL(NestedDataFrameFactory.testCaseSchema.toDDL)
+
+      assert(actualDDL == expectedDDL)
+    }
+
+    "work with another complex example (unescaped)" in {
+      val inputDDL = "id BIGINT, key1 BIGINT, key2 BIGINT, struct1 STRUCT<key3: INT, key4: INT>, struct2 STRUCT< inner1: STRUCT<key5: BIGINT, key6: BIGINT, skey1: STRING>>, struct3 STRUCT<inner3: STRUCT<array3: ARRAY<STRUCT<a1: BIGINT, a2: BIGINT, a3: STRING>>>>,array1 ARRAY<STRUCT<key7: BIGINT, key8: BIGINT, skey2: STRING>>,array2 ARRAY<STRUCT<key2: BIGINT, inner2: ARRAY<STRUCT<key9: BIGINT, `key10`: BIGINT, struct3: STRUCT<k1: INT, k2: INT>>>>>"
+      val expectedDDL = "`id` BIGINT, `key1` BIGINT, `key2` BIGINT, `struct1` STRUCT<`key3`: INT, `key4`: INT>, `struct2` STRUCT< `inner1`: STRUCT<`key5`: BIGINT, `key6`: BIGINT, `skey1`: STRING>>, `struct3` STRUCT<`inner3`: STRUCT<`array3`: ARRAY<STRUCT<`a1`: BIGINT, `a2`: BIGINT, `a3`: STRING>>>>,`array1` ARRAY<STRUCT<`key7`: BIGINT, `key8`: BIGINT, `skey2`: STRING>>,`array2` ARRAY<STRUCT<`key2`: BIGINT, `inner2`: ARRAY<STRUCT<`key9`: BIGINT, `key10`: BIGINT, `struct3`: STRUCT<`k1`: INT, `k2`: INT>>>>>"
+
+      val actualDDL = escapeColumnsSparkDDL(inputDDL)
+
+      assert(actualDDL == expectedDDL)
+    }
   }
 
 }
diff --git a/pramen/extras/src/test/scala/za/co/absa/pramen/extras/NestedDataFrameFactory.scala b/pramen/extras/src/test/scala/za/co/absa/pramen/extras/NestedDataFrameFactory.scala
index 947eec471..e2bbb285c 100644
--- a/pramen/extras/src/test/scala/za/co/absa/pramen/extras/NestedDataFrameFactory.scala
+++ b/pramen/extras/src/test/scala/za/co/absa/pramen/extras/NestedDataFrameFactory.scala
@@ -14,7 +14,7 @@
  * limitations under the License.
  */
 
-package za.co.absa.pramen
+package za.co.absa.pramen.extras
 
 import org.apache.spark.sql.types._
 import org.apache.spark.sql.{DataFrame, Row, SparkSession}
diff --git a/pramen/extras/src/test/scala/za/co/absa/pramen/extras/tests/avro/AvroUtilsSuite.scala b/pramen/extras/src/test/scala/za/co/absa/pramen/extras/tests/avro/AvroUtilsSuite.scala
index b2872887a..8e0f1e111 100644
--- a/pramen/extras/src/test/scala/za/co/absa/pramen/extras/tests/avro/AvroUtilsSuite.scala
+++ b/pramen/extras/src/test/scala/za/co/absa/pramen/extras/tests/avro/AvroUtilsSuite.scala
@@ -18,7 +18,7 @@ package za.co.absa.pramen.extras.tests.avro
 
 import org.apache.spark.sql.functions.struct
 import org.scalatest.wordspec.AnyWordSpec
-import za.co.absa.pramen.NestedDataFrameFactory
+import za.co.absa.pramen.extras.NestedDataFrameFactory
 import za.co.absa.pramen.extras.utils.ResourceUtils.getResourceString
 import za.co.absa.pramen.extras.utils.JsonUtils
 import za.co.absa.pramen.extras.avro.AvroUtils