Repository: AbsaOSS/pramen (language: python)

Issue:
Add support for `"` delimiter for MS SQL databases
## Background
Both '[]' and '"' are valid, but we support only '[]'.

## Feature
Add support for `"` delimiter for MS SQL databases.

## Example [Optional]
--

## Proposed Solution [Optional]
--



Discussion / hints:
Currently, database identifiers escaped with `"` in the user-provided configuration are escaped again with `[` by Pramen, leading to errors.

For example, the config might contain the following section:
```
pramen.operations = [
  {
    name = "Land"
    type = "ingestion"
    schedule.type = "daily"
    source = "pramen_source"
    info.date.expr = "@runDate"
    tables = [
      {
        input.table = "\"mydb\".\"mytable\""
        output.metastore.table = "MYTABLE"
      },
    ]
  },
```

If mydb is a Microsoft SQL Database, Pramen will escape the input table as `["mydb"."mytable"]`, which leads to the error message
```
com.microsoft.sqlserver.jdbc.SQLServerException: Invalid object name '"mydbe"."mytable"'.
com.microsoft.sqlserver.jdbc.SQLServerException.makeFromDatabaseError(SQLServerException.java:265)
com.microsoft.sqlserver.jdbc.SQLServerStatement.getNextResult(SQLServerStatement.java:1662)
com.microsoft.sqlserver.jdbc.SQLServerPreparedStatement.doExecutePreparedStatement(SQLServerPreparedStatement.java:615)
com.microsoft.sqlserver.jdbc.SQLServerPreparedStatement$PrepStmtExecCmd.doExecute(SQLServerPreparedStatement.java:537)
  com.microsoft.sqlserver.jdbc.TDSCommand.execute(IOBuffer.java:7417)
  com.microsoft.sqlserver.jdbc.SQLServerConnection.executeCommand(SQLServerConnection.java:3455)
  com.microsoft.sqlserver.jdbc.SQLServerStatement.executeCommand(SQLServerStatement.java:262)
  com.microsoft.sqlserver.jdbc.SQLServerStatement.executeStatement(SQLServerStatement.java:237)
  com.microsoft.sqlserver.jdbc.SQLServerPreparedStatement.executeQuery(SQLServerPreparedStatement.java:456)
  org.apache.spark.sql.execution.datasources.jdbc.JDBCRDD$.getQueryOutputSchema(JDBCRDD.scala:68)
  org.apache.spark.sql.execution.datasources.jdbc.JDBCRDD$.resolveTable(JDBCRDD.scala:58)
  org.apache.spark.sql.execution.datasources.jdbc.JDBCRelation$.getSchema(JDBCRelation.scala:242)
  org.apache.spark.sql.execution.datasources.jdbc.JdbcRelationProvider.createRelation(JdbcRelationProvider.scala:37)
  org.apache.spark.sql.execution.datasources.DataSource.resolveRelation(DataSource.scala:350)
  org.apache.spark.sql.DataFrameReader.loadV1Source(DataFrameReader.scala:228)
  org.apache.spark.sql.DataFrameReader.$anonfun$load$2(DataFrameReader.scala:210)
  scala.Option.getOrElse(Option.scala:189)
  org.apache.spark.sql.DataFrameReader.load(DataFrameReader.scala:210)
  org.apache.spark.sql.DataFrameReader.load(DataFrameReader.scala:171)
  za.co.absa.pramen.core.reader.TableReaderJdbc.getDataFrame(TableReaderJdbc.scala:154)
```
> Delimited identifiers
> Are enclosed in double quotation marks (") or brackets ([ ])

Source: https://learn.microsoft.com/en-us/sql/relational-databases/databases/database-identifiers?view=sql-server-ver16#classes-of-identifiers
Thanks for adding the details, @kevinwallimann !

Fix patch:
diff --git a/pramen/core/src/main/scala/za/co/absa/pramen/core/sql/SqlGeneratorMicrosoft.scala b/pramen/core/src/main/scala/za/co/absa/pramen/core/sql/SqlGeneratorMicrosoft.scala
index 7a865763a..324abc745 100644
--- a/pramen/core/src/main/scala/za/co/absa/pramen/core/sql/SqlGeneratorMicrosoft.scala
+++ b/pramen/core/src/main/scala/za/co/absa/pramen/core/sql/SqlGeneratorMicrosoft.scala
@@ -18,6 +18,7 @@ package za.co.absa.pramen.core.sql
 
 import za.co.absa.pramen.api.sql.SqlGeneratorBase.{needsEscaping, validateIdentifier}
 import za.co.absa.pramen.api.sql.{SqlColumnType, SqlConfig, SqlGenerator}
+import za.co.absa.pramen.core.utils.MutableStack
 
 import java.time.LocalDate
 import java.time.format.DateTimeFormatter
@@ -158,7 +159,7 @@ class SqlGeneratorMicrosoft(sqlConfig: SqlConfig) extends SqlGenerator {
     val output = new ListBuffer[String]
     val curColumn = new StringBuffer()
     val len = trimmedIdentifier.length
-    var nestingChar = new ListBuffer[Char]
+    val nestingChar = new MutableStack[Char]
     var i = 0
 
     while (i < len) {
@@ -173,41 +174,36 @@ class SqlGeneratorMicrosoft(sqlConfig: SqlConfig) extends SqlGenerator {
       }
 
       if (c == escapeChar2) {
-        if (nestingChar.isEmpty) {
-          nestingChar.prepend(escapeChar2)
-          if (curColumn.length() > 1 && i < len - 1 && nextChar != '.')
-            throw new IllegalArgumentException(f"Invalid character '$escapeChar2' in the identifier '$identifier', position $i.")
-        } else {
-          if (nestingChar.head != escapeChar2) {
-            throw new IllegalArgumentException(f"Invalid character '$escapeChar2' in the identifier '$identifier', position $i.")
-          } else {
-            nestingChar = nestingChar.tail
-          }
+        if (curColumn.length() > 1 && i < len - 1 && nextChar != '.')
+          throw new IllegalArgumentException(f"Invalid character '$escapeChar2' in the identifier '$identifier', position $i.")
+        nestingChar.pop() match {
+          case Some(ch) =>
+            if (ch != escapeChar2)
+              throw new IllegalArgumentException(f"Invalid character '$escapeChar2' in the identifier '$identifier', position $i.")
+          case None =>
+            nestingChar.push(escapeChar2)
         }
       } else if (c == escapeBegin1) {
-        if (nestingChar.nonEmpty && nestingChar.head == escapeChar2) {
+        if (nestingChar.nonEmpty && nestingChar.peek() == escapeChar2) {
           throw new IllegalArgumentException(f"Invalid character '$escapeChar2' in the identifier '$identifier', position $i.")
         }
         if (curColumn.length() > 1 && i < len - 1 && nextChar != '.')
           throw new IllegalArgumentException(f"Invalid character '$escapeBegin1' in the identifier '$identifier', position $i.")
-        nestingChar.prepend(escapeBegin1)
+        nestingChar.push(escapeBegin1)
       } else if (c == escapeEnd1) {
-        if (nestingChar.nonEmpty && nestingChar.head == escapeChar2)
-          throw new IllegalArgumentException(f"Found not matching '$escapeChar2' in the identifier '$identifier'.")
-        else {
-          if (nestingChar.isEmpty) {
-            throw new IllegalArgumentException(f"Found not matching '$c' in the identifier '$identifier'.")
-          } else {
-            nestingChar = nestingChar.tail
-          }
+        nestingChar.pop() match {
+          case Some(ch) =>
+            if (ch == escapeChar2)
+              throw new IllegalArgumentException(f"Found not matching '$escapeChar2' in the identifier '$identifier'.")
+          case None =>
+            throw new IllegalArgumentException(f"Found not matching '$escapeEnd1' in the identifier '$identifier'.")
         }
       }
       i += 1
     }
 
-    if (nestingChar.nonEmpty) {
-      val c = nestingChar.head
-      throw new IllegalArgumentException(f"Found not matching '$c' in the identifier '$identifier'.")
+    nestingChar.pop().foreach{ch =>
+      throw new IllegalArgumentException(f"Found not matching '$ch' in the identifier '$identifier'.")
     }
 
     if (curColumn.toString.nonEmpty)
diff --git a/pramen/core/src/main/scala/za/co/absa/pramen/core/utils/MutableStack.scala b/pramen/core/src/main/scala/za/co/absa/pramen/core/utils/MutableStack.scala
new file mode 100644
index 000000000..539d68dcb
--- /dev/null
+++ b/pramen/core/src/main/scala/za/co/absa/pramen/core/utils/MutableStack.scala
@@ -0,0 +1,49 @@
+/*
+ * 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.utils
+
+class MutableStack[T] {
+  var content = List.empty[T]
+
+  def push(el: T): Unit = synchronized {
+    content = el :: content
+  }
+
+  def pop(): Option[T] = synchronized {
+    if (content.isEmpty)
+      None
+    else {
+      val el = content.head
+      content = content.tail
+      Some(el)
+    }
+  }
+
+  def peek(): T = synchronized {
+    if (content.isEmpty)
+      throw new NoSuchElementException
+    else content.head
+  }
+
+  def isEmpty: Boolean = synchronized {
+    content.isEmpty
+  }
+
+  def nonEmpty: Boolean = synchronized {
+    content.nonEmpty
+  }
+}


Test patch:
diff --git a/pramen/core/src/test/scala/za/co/absa/pramen/core/tests/sql/SqlGeneratorLoaderSuite.scala b/pramen/core/src/test/scala/za/co/absa/pramen/core/tests/sql/SqlGeneratorLoaderSuite.scala
index 17edd9678..be7c28672 100644
--- a/pramen/core/src/test/scala/za/co/absa/pramen/core/tests/sql/SqlGeneratorLoaderSuite.scala
+++ b/pramen/core/src/test/scala/za/co/absa/pramen/core/tests/sql/SqlGeneratorLoaderSuite.scala
@@ -348,7 +348,13 @@ class SqlGeneratorLoaderSuite extends AnyWordSpec with RelationalDbFixture {
         assert(actual == Seq("System User", "[Table Name]"))
       }
 
-      "split a complex column with quotes" in {
+      "split a complex column with quotes 1" in {
+        val actual = genDate.asInstanceOf[SqlGeneratorMicrosoft].splitComplexIdentifier("\"System User\".Table Name")
+
+        assert(actual == Seq("\"System User\"", "Table Name"))
+      }
+
+      "split a complex column with quotes 2" in {
         val actual = genDate.asInstanceOf[SqlGeneratorMicrosoft].splitComplexIdentifier("[System User].\"Table Name\"")
 
         assert(actual == Seq("[System User]", "\"Table Name\""))
@@ -413,6 +419,22 @@ class SqlGeneratorLoaderSuite extends AnyWordSpec with RelationalDbFixture {
 
         assert(ex.getMessage.contains("Found not matching '\"' in the identifier 'System User.Table Name\"'"))
       }
+
+      "throw on invalid open double-quote" in {
+        val ex = intercept[IllegalArgumentException] {
+          genDate.asInstanceOf[SqlGeneratorMicrosoft].splitComplexIdentifier("\"System User.\"Table Name")
+        }
+
+        assert(ex.getMessage.contains("Invalid character '\"' in the identifier '\"System User.\"Table Name'"))
+      }
+
+      "throw on invalid close double-quote" in {
+        val ex = intercept[IllegalArgumentException] {
+          genDate.asInstanceOf[SqlGeneratorMicrosoft].splitComplexIdentifier("\"System Use\"r.Table Name")
+        }
+
+        assert(ex.getMessage.contains("Invalid character '\"' in the identifier '\"System Use\"r.Table Name'"))
+      }
     }
   }
 
diff --git a/pramen/core/src/test/scala/za/co/absa/pramen/core/tests/utils/MutableStackSuite.scala b/pramen/core/src/test/scala/za/co/absa/pramen/core/tests/utils/MutableStackSuite.scala
new file mode 100644
index 000000000..7dcbde7dd
--- /dev/null
+++ b/pramen/core/src/test/scala/za/co/absa/pramen/core/tests/utils/MutableStackSuite.scala
@@ -0,0 +1,104 @@
+/*
+ * 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.tests.utils
+
+import org.scalatest.wordspec.AnyWordSpec
+import za.co.absa.pramen.core.utils.MutableStack
+
+class MutableStackSuite extends AnyWordSpec {
+  "push()" should {
+    "add an element to the stack" in {
+      val stack = new MutableStack[Int]
+      stack.push(1)
+      assert(stack.peek() == 1)
+    }
+
+    "add multiple elements to the stack" in {
+      val stack = new MutableStack[Int]
+      stack.push(1)
+      stack.push(2)
+      stack.push(3)
+      assert(stack.peek() == 3)
+    }
+  }
+
+  "pop()" should {
+    "remove the top element from the stack" in {
+      val stack = new MutableStack[Int]
+      stack.push(1)
+      stack.push(2)
+      stack.push(3)
+      assert(stack.pop().get == 3)
+      assert(stack.pop().get == 2)
+      assert(stack.pop().get == 1)
+    }
+
+    "return None if the stack is empty" in {
+      val stack = new MutableStack[Int]
+      assert(stack.pop().isEmpty)
+    }
+  }
+
+  "peek()" should {
+    "return the top element from the stack" in {
+      val stack = new MutableStack[Int]
+      stack.push(1)
+      stack.push(2)
+      stack.push(3)
+      assert(stack.peek() == 3)
+
+      stack.pop()
+      assert(stack.peek() == 2)
+
+      stack.pop()
+      assert(stack.peek() == 1)
+    }
+
+    "throw an exception if the stack is empty" in {
+      val stack = new MutableStack[Int]
+      assertThrows[NoSuchElementException] {
+        stack.peek()
+      }
+    }
+  }
+
+  "isEmpty()" should {
+    "return true if the stack is empty" in {
+      val stack = new MutableStack[Int]
+      assert(stack.isEmpty)
+    }
+
+    "return false if the stack is not empty" in {
+      val stack = new MutableStack[Int]
+      stack.push(1)
+      assert(!stack.isEmpty)
+    }
+  }
+
+  "nonEmpty()" should {
+    "return true if the stack is not empty" in {
+      val stack = new MutableStack[Int]
+      stack.push(1)
+      assert(stack.nonEmpty)
+    }
+
+    "return false if the stack is empty" in {
+      val stack = new MutableStack[Int]
+      assert(!stack.nonEmpty)
+    }
+  }
+}