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/api/src/main/scala/za/co/absa/pramen/api/sql/SqlGeneratorBase.scala b/pramen/api/src/main/scala/za/co/absa/pramen/api/sql/SqlGeneratorBase.scala
index 1c1b7de91..9a2d15a4b 100644
--- a/pramen/api/src/main/scala/za/co/absa/pramen/api/sql/SqlGeneratorBase.scala
+++ b/pramen/api/src/main/scala/za/co/absa/pramen/api/sql/SqlGeneratorBase.scala
@@ -69,7 +69,7 @@ abstract class SqlGeneratorBase(sqlConfig: SqlConfig) extends SqlGenerator {
     if (columns.isEmpty) {
       "*"
     } else {
-      columns.map(col => escape(col)) .mkString(", ")
+      columns.map(col => escape(col)).mkString(", ")
     }
   }
 
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 8e7c4038a..7a865763a 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
@@ -16,19 +16,22 @@
 
 package za.co.absa.pramen.core.sql
 
-import za.co.absa.pramen.api.sql.{SqlColumnType, SqlConfig, SqlGeneratorBase}
+import za.co.absa.pramen.api.sql.SqlGeneratorBase.{needsEscaping, validateIdentifier}
+import za.co.absa.pramen.api.sql.{SqlColumnType, SqlConfig, SqlGenerator}
 
 import java.time.LocalDate
 import java.time.format.DateTimeFormatter
+import scala.collection.mutable.ListBuffer
 
-class SqlGeneratorMicrosoft(sqlConfig: SqlConfig) extends SqlGeneratorBase(sqlConfig) {
+class SqlGeneratorMicrosoft(sqlConfig: SqlConfig) extends SqlGenerator {
   private val dateFormatterApp = DateTimeFormatter.ofPattern(sqlConfig.dateFormatApp)
   private val isIso = sqlConfig.dateFormatApp.toLowerCase.startsWith("yyyy-mm-dd")
 
   // 23 is "yyyy-MM-dd", see https://www.mssqltips.com/sqlservertip/1145/date-and-time-conversions-using-sql-server/
   private val isoFormatMsSqlRef = 23
 
-  override val beginEndEscapeChars: (Char, Char) = ('[', ']')
+  val beginEndEscapeChars: (Char, Char) = ('[', ']')
+  val escapeChar2 = '\"'
 
   override def getDtable(sql: String): String = {
     if (sql.exists(_ == ' ')) {
@@ -97,7 +100,119 @@ class SqlGeneratorMicrosoft(sqlConfig: SqlConfig) extends SqlGeneratorBase(sqlCo
     }
   }
 
+  override def getAliasExpression(expression: String, alias: String): String = {
+    s"$expression AS ${escape(alias)}"
+  }
+
+  override def quote(identifier: String): String = {
+    validateIdentifier(identifier)
+    splitComplexIdentifier(identifier).map(quoteSingleIdentifier).mkString(".")
+  }
+
+  override def escape(identifier: String): String = {
+    if (needsEscaping(sqlConfig.identifierQuotingPolicy, identifier)) {
+      quote(identifier)
+    } else {
+      identifier
+    }
+  }
+
   private def getLimit(limit: Option[Int]): String = {
     limit.map(n => s"TOP $n ").getOrElse("")
   }
+
+  private def columnExpr(columns: Seq[String]): String = {
+    if (columns.isEmpty) {
+      "*"
+    } else {
+      columns.map(col => escape(col)).mkString(", ")
+    }
+  }
+
+  private def infoDateColumn: String = {
+    escape(sqlConfig.infoDateColumn)
+  }
+
+  private def quoteSingleIdentifier(identifier: String): String = {
+    val (escapeBegin, escapeEnd) = beginEndEscapeChars
+
+    if (
+      (identifier.startsWith(s"$escapeBegin") && identifier.endsWith(s"$escapeEnd")) ||
+        (identifier.startsWith(s"$escapeChar2") && identifier.endsWith(s"$escapeChar2"))
+    ) {
+      identifier
+    } else {
+      s"$escapeBegin$identifier$escapeEnd"
+    }
+  }
+
+  private[core] def splitComplexIdentifier(identifier: String): Seq[String] = {
+    val trimmedIdentifier = identifier.trim
+
+    if (trimmedIdentifier.isEmpty) {
+      throw new IllegalArgumentException(f"Found an empty table name or column name ('$identifier').")
+    }
+
+    val (escapeBegin1, escapeEnd1) = beginEndEscapeChars
+
+    val output = new ListBuffer[String]
+    val curColumn = new StringBuffer()
+    val len = trimmedIdentifier.length
+    var nestingChar = new ListBuffer[Char]
+    var i = 0
+
+    while (i < len) {
+      val c = trimmedIdentifier(i)
+      val nextChar = if (i == len - 1) ' ' else trimmedIdentifier(i + 1)
+
+      if (nestingChar.isEmpty && c == '.') {
+        output += curColumn.toString
+        curColumn.setLength(0)
+      } else {
+        curColumn.append(c)
+      }
+
+      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
+          }
+        }
+      } else if (c == escapeBegin1) {
+        if (nestingChar.nonEmpty && nestingChar.head == 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)
+      } 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
+          }
+        }
+      }
+      i += 1
+    }
+
+    if (nestingChar.nonEmpty) {
+      val c = nestingChar.head
+      throw new IllegalArgumentException(f"Found not matching '$c' in the identifier '$identifier'.")
+    }
+
+    if (curColumn.toString.nonEmpty)
+      output += curColumn.toString
+
+    output.toSeq
+  }
 }


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 b8ef45e7a..17edd9678 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
@@ -206,6 +206,14 @@ class SqlGeneratorLoaderSuite extends AnyWordSpec with RelationalDbFixture {
       assert(genDate.getDataQuery("A", Nil, Some(100)) == "SELECT TOP 100 * FROM A WITH (NOLOCK)")
     }
 
+    "generate data queries with limit clause date ranges when table is quoted" in {
+      assert(genDate.getDataQuery("\"A\"", Nil, Some(100)) == "SELECT TOP 100 * FROM \"A\" WITH (NOLOCK)")
+    }
+
+    "generate data queries with limit clause date ranges when table is escaped in brackets" in {
+      assert(genDate.getDataQuery("[A]", Nil, Some(100)) == "SELECT TOP 100 * FROM [A] WITH (NOLOCK)")
+    }
+
     "generate ranged count queries" when {
       "date is in DATE format" in {
         assert(genDate.getCountQuery("A", date1, date1) ==
@@ -324,51 +332,87 @@ class SqlGeneratorLoaderSuite extends AnyWordSpec with RelationalDbFixture {
     "splitComplexIdentifier" should {
       "throw on an empty identifier" in {
         assertThrows[IllegalArgumentException] {
-          genDate.asInstanceOf[SqlGeneratorBase].splitComplexIdentifier(" ")
+          genDate.asInstanceOf[SqlGeneratorMicrosoft].splitComplexIdentifier(" ")
         }
       }
 
       "keep original column name as is" in {
-        val actual = genDate.asInstanceOf[SqlGeneratorBase].splitComplexIdentifier("System User")
+        val actual = genDate.asInstanceOf[SqlGeneratorMicrosoft].splitComplexIdentifier("System User")
 
         assert(actual == Seq("System User"))
       }
 
-      "split a complex column" in {
-        val actual = genDate.asInstanceOf[SqlGeneratorBase].splitComplexIdentifier("System User.[Table Name]")
+      "split a complex column with brackets" in {
+        val actual = genDate.asInstanceOf[SqlGeneratorMicrosoft].splitComplexIdentifier("System User.[Table Name]")
 
         assert(actual == Seq("System User", "[Table Name]"))
       }
 
-      "handle escaped dots" in {
-        val actual = genDate.asInstanceOf[SqlGeneratorBase].splitComplexIdentifier("[System User.Table Name]")
+      "split a complex column with quotes" in {
+        val actual = genDate.asInstanceOf[SqlGeneratorMicrosoft].splitComplexIdentifier("[System User].\"Table Name\"")
+
+        assert(actual == Seq("[System User]", "\"Table Name\""))
+      }
+
+      "handle escaped dots with brackets" in {
+        val actual = genDate.asInstanceOf[SqlGeneratorMicrosoft].splitComplexIdentifier("[System User.Table Name]")
 
         assert(actual == Seq("[System User.Table Name]"))
       }
 
+      "handle escaped dots with quotes" in {
+        val actual = genDate.asInstanceOf[SqlGeneratorMicrosoft].splitComplexIdentifier("\"System User.Table Name\"")
+
+        assert(actual == Seq("\"System User.Table Name\""))
+      }
+
       "throw an exception if brackets are found inside the identifier" in {
         val ex = intercept[IllegalArgumentException] {
-          genDate.asInstanceOf[SqlGeneratorBase].splitComplexIdentifier("System Use[r.T]able Name")
+          genDate.asInstanceOf[SqlGeneratorMicrosoft].splitComplexIdentifier("System Use[r.T]able Name")
         }
 
         assert(ex.getMessage.contains("Invalid character '[' in the identifier 'System Use[r.T]able Name', position 10."))
       }
 
+      "throw an exception if quotes are found inside the identifier" in {
+        val ex = intercept[IllegalArgumentException] {
+          genDate.asInstanceOf[SqlGeneratorMicrosoft].splitComplexIdentifier("System Use\"r.T\"able Name")
+        }
+
+        assert(ex.getMessage.contains("Invalid character '\"' in the identifier 'System Use\"r.T\"able Name', position 10."))
+      }
+
       "throw on unmatched open bracket" in {
         val ex = intercept[IllegalArgumentException] {
-          genDate.asInstanceOf[SqlGeneratorBase].splitComplexIdentifier("System User.[Table Name")
+          genDate.asInstanceOf[SqlGeneratorMicrosoft].splitComplexIdentifier("System User.[Table Name")
         }
 
         assert(ex.getMessage.contains("Found not matching '[' in the identifier 'System User.[Table Name'"))
       }
 
+      "throw on unmatched open double quote" in {
+        val ex = intercept[IllegalArgumentException] {
+          genDate.asInstanceOf[SqlGeneratorMicrosoft].splitComplexIdentifier("System User.\"Table Name")
+        }
+
+        assert(ex.getMessage.contains("Found not matching '\"' in the identifier 'System User.\"Table Name'"))
+      }
+
       "throw on unmatched closing bracket" in {
         val ex = intercept[IllegalArgumentException] {
-          genDate.asInstanceOf[SqlGeneratorBase].splitComplexIdentifier("System User.Table Name]")
+          genDate.asInstanceOf[SqlGeneratorMicrosoft].splitComplexIdentifier("System User.Table Name]")
         }
 
         assert(ex.getMessage.contains("Found not matching ']' in the identifier 'System User.Table Name]'"))
       }
+
+      "throw on unmatched double quote at the and" in {
+        val ex = intercept[IllegalArgumentException] {
+          genDate.asInstanceOf[SqlGeneratorMicrosoft].splitComplexIdentifier("System User.Table Name\"")
+        }
+
+        assert(ex.getMessage.contains("Found not matching '\"' in the identifier 'System User.Table Name\"'"))
+      }
     }
   }