#420 Fix Hive DDL escaping for complex schema.

Author: yruslan

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
     }

pramen/core/src/test/scala/za/co/absa/pramen/core/tests/utils/SparkUtilsSuite.scala

@@ -641,6 +641,36 @@ class SparkUtilsSuite extends AnyWordSpec with SparkTestBase with TempDirFixture
 
       assert(actual == expectedDDL)
     }
+
+    "work with decimals and nested arrays of struct" in {
+      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 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)
+
+      assert(actualDDL == expectedDDL)
+    }
+
+    "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)
+    }
   }
 
 }