Repository: AbsaOSS/pramen (language: python)

Issue:
If database name and/or table name are quoted, Pramen is unable to fetch metadata from MS SQL Server
## Describe the bug
If database name and/or table name are quoted, Pramen is unable to fetch metadata from MS SQL Server.

## Code and/or configuration snippet that caused the issue

```hocon
pramen {
}
```

## Expected behavior
While searching for table metadata Pramen should try unquoting database and table name first.

## Context
- Pramen/pramen-py version:
- Spark version:
- Scala/Python version:
- Operating system:



Discussion / hints:

Fix patch:
diff --git a/pramen/api/src/main/scala/za/co/absa/pramen/api/sql/SqlGenerator.scala b/pramen/api/src/main/scala/za/co/absa/pramen/api/sql/SqlGenerator.scala
index 934aeaed5..947e331ed 100644
--- a/pramen/api/src/main/scala/za/co/absa/pramen/api/sql/SqlGenerator.scala
+++ b/pramen/api/src/main/scala/za/co/absa/pramen/api/sql/SqlGenerator.scala
@@ -85,6 +85,13 @@ trait SqlGenerator {
     */
   def quote(identifier: String): String
 
+  /**
+    * Unquotes an identifier name with characters specific to SQL dialects.
+    * If the identifier is already not quoted, nothing will be done.
+    * It supports partially quoted identifiers. E.g. '"my_catalog".my table' will be quoted as 'my_catalog.my table'.
+    */
+  def unquote(identifier: String): String
+
   /**
     * Returns true if the SQL generator can only work if it has an active connection.
     * This can be for database engines that does not support "SELECT * FROM table" and require explicit list of columns.
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 9a2d15a4b..15945a1c8 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
@@ -44,6 +44,16 @@ abstract class SqlGeneratorBase(sqlConfig: SqlConfig) extends SqlGenerator {
     }
   }
 
+  def unquoteSingleIdentifier(identifier: String): String = {
+    val (escapeBegin, escapeEnd) = beginEndEscapeChars
+
+    if (identifier.startsWith(s"$escapeBegin") && identifier.endsWith(s"$escapeEnd") && identifier.length > 2) {
+      identifier.substring(1, identifier.length - 1)
+    } else {
+      identifier
+    }
+  }
+
   override def getAliasExpression(expression: String, alias: String): String = {
     s"$expression AS ${escape(alias)}"
   }
@@ -53,6 +63,11 @@ abstract class SqlGeneratorBase(sqlConfig: SqlConfig) extends SqlGenerator {
     splitComplexIdentifier(identifier).map(quoteSingleIdentifier).mkString(".")
   }
 
+  override final def unquote(identifier: String): String = {
+    validateIdentifier(identifier)
+    splitComplexIdentifier(identifier).map(unquoteSingleIdentifier).mkString(".")
+  }
+
   override final def escape(identifier: String): String = {  
     if (needsEscaping(sqlConfig.identifierQuotingPolicy, identifier)) {
       quote(identifier)
diff --git a/pramen/core/src/main/scala/za/co/absa/pramen/core/reader/TableReaderJdbc.scala b/pramen/core/src/main/scala/za/co/absa/pramen/core/reader/TableReaderJdbc.scala
index 750ebcac3..d80f77ec6 100644
--- a/pramen/core/src/main/scala/za/co/absa/pramen/core/reader/TableReaderJdbc.scala
+++ b/pramen/core/src/main/scala/za/co/absa/pramen/core/reader/TableReaderJdbc.scala
@@ -192,8 +192,11 @@ class TableReaderJdbc(jdbcReaderConfig: TableReaderJdbcConfig,
       JdbcSparkUtils.withJdbcMetadata(jdbcReaderConfig.jdbcConfig, sql) { (connection, jdbcMetadata) =>
         val schemaWithMetadata = JdbcSparkUtils.addMetadataFromJdbc(df.schema, jdbcMetadata)
         val schemaWithColumnDescriptions = tableOpt match {
-          case Some(table) => JdbcSparkUtils.addColumnDescriptionsFromJdbc(schemaWithMetadata, table, connection)
-          case None => schemaWithMetadata
+          case Some(table) =>
+            log.info(s"Reading JDBC metadata descriptions the table: $table")
+            JdbcSparkUtils.addColumnDescriptionsFromJdbc(schemaWithMetadata, sqlGen.unquote(table), connection)
+          case None =>
+            schemaWithMetadata
         }
         df = spark.createDataFrame(df.rdd, schemaWithColumnDescriptions)
       }
diff --git a/pramen/core/src/main/scala/za/co/absa/pramen/core/reader/TableReaderJdbcNative.scala b/pramen/core/src/main/scala/za/co/absa/pramen/core/reader/TableReaderJdbcNative.scala
index 93b07b9d1..e7a1831bd 100644
--- a/pramen/core/src/main/scala/za/co/absa/pramen/core/reader/TableReaderJdbcNative.scala
+++ b/pramen/core/src/main/scala/za/co/absa/pramen/core/reader/TableReaderJdbcNative.scala
@@ -88,12 +88,13 @@ class TableReaderJdbcNative(jdbcReaderConfig: TableReaderJdbcConfig,
     }
 
     if (jdbcReaderConfig.enableSchemaMetadata) {
-      JdbcSparkUtils.withJdbcMetadata(jdbcReaderConfig.jdbcConfig, sql) { (connection, jdbcMetadata) =>
+      JdbcSparkUtils.withJdbcMetadata(jdbcReaderConfig.jdbcConfig, sql) { (connection, _) =>
         val schemaWithColumnDescriptions = tableOpt match {
           case Some(table) =>
-            log.info(s"Reading JDBC metadata descriptions the query: $sql")
-            JdbcSparkUtils.addColumnDescriptionsFromJdbc(df.schema, table, connection)
-          case None => df.schema
+            log.info(s"Reading JDBC metadata descriptions the table: $table")
+            JdbcSparkUtils.addColumnDescriptionsFromJdbc(df.schema, sqlGen.unquote(table), connection)
+          case None =>
+            df.schema
         }
         df = spark.createDataFrame(df.rdd, schemaWithColumnDescriptions)
       }
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 a5f285b7a..1d5064454 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
@@ -114,6 +114,11 @@ class SqlGeneratorMicrosoft(sqlConfig: SqlConfig) extends SqlGenerator {
     splitComplexIdentifier(identifier).map(quoteSingleIdentifier).mkString(".")
   }
 
+  override def unquote(identifier: String): String = {
+    validateIdentifier(identifier)
+    splitComplexIdentifier(identifier).map(unquoteSingleIdentifier).mkString(".")
+  }
+
   override def escape(identifier: String): String = {
     if (needsEscaping(sqlConfig.identifierQuotingPolicy, identifier)) {
       quote(identifier)
@@ -138,6 +143,18 @@ class SqlGeneratorMicrosoft(sqlConfig: SqlConfig) extends SqlGenerator {
     escape(sqlConfig.infoDateColumn)
   }
 
+  private def unquoteSingleIdentifier(identifier: String): String = {
+    val (escapeBegin, escapeEnd) = beginEndEscapeChars
+
+    if (identifier.startsWith(s"$escapeBegin") && identifier.endsWith(s"$escapeEnd") && identifier.length > 2) {
+      identifier.substring(1, identifier.length - 1)
+    } else if (identifier.startsWith(s"$escapeChar2") && identifier.endsWith(s"$escapeChar2") && identifier.length > 2) {
+      identifier.substring(1, identifier.length - 1)
+    } else {
+      identifier
+    }
+  }
+
   private def quoteSingleIdentifier(identifier: String): String = {
     val (escapeBegin, escapeEnd) = beginEndEscapeChars
 
diff --git a/pramen/core/src/main/scala/za/co/absa/pramen/core/utils/JdbcSparkUtils.scala b/pramen/core/src/main/scala/za/co/absa/pramen/core/utils/JdbcSparkUtils.scala
index 4a1a93b88..b89cfbda5 100644
--- a/pramen/core/src/main/scala/za/co/absa/pramen/core/utils/JdbcSparkUtils.scala
+++ b/pramen/core/src/main/scala/za/co/absa/pramen/core/utils/JdbcSparkUtils.scala
@@ -26,7 +26,6 @@ import za.co.absa.pramen.core.utils.impl.JdbcFieldMetadata
 
 import java.sql.{Connection, DatabaseMetaData, ResultSet, ResultSetMetaData}
 import scala.collection.mutable.ListBuffer
-import scala.util.control.NonFatal
 
 object JdbcSparkUtils {
   private val log = LoggerFactory.getLogger(this.getClass)


Test patch:
diff --git a/pramen/core/src/test/scala/za/co/absa/pramen/core/mocks/SqlGeneratorDummy.scala b/pramen/core/src/test/scala/za/co/absa/pramen/core/mocks/SqlGeneratorDummy.scala
index e439686ab..2aaea5ad3 100644
--- a/pramen/core/src/test/scala/za/co/absa/pramen/core/mocks/SqlGeneratorDummy.scala
+++ b/pramen/core/src/test/scala/za/co/absa/pramen/core/mocks/SqlGeneratorDummy.scala
@@ -44,4 +44,6 @@ class SqlGeneratorDummy(sqlConfig: SqlConfig) extends SqlGenerator {
   override def escape(identifier: String): String = null
 
   override def quote(identifier: String): String = null
+
+  override def unquote(identifier: String): String = null
 }
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 586f2ed3b..c420a1423 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
@@ -185,6 +185,14 @@ class SqlGeneratorLoaderSuite extends AnyWordSpec with RelationalDbFixture {
         assert(actual == "\"System User\".\"Table Name\"")
       }
     }
+
+    "unquote" should {
+      "quote each subfields separately" in {
+        val actual = gen.unquote("System User.\"Table Name\"")
+
+        assert(actual == "System User.Table Name")
+      }
+    }
   }
 
   "Microsoft SQL generator" should {
@@ -341,6 +349,20 @@ class SqlGeneratorLoaderSuite extends AnyWordSpec with RelationalDbFixture {
       }
     }
 
+    "unquote" should {
+      "quote each subfields separately using quotes" in {
+        val actual = genDate.unquote("System User.\"Table Name\"")
+
+        assert(actual == "System User.Table Name")
+      }
+
+      "quote each subfields separately using brackets" in {
+        val actual = genDate.unquote("[System User].[Table Name]")
+
+        assert(actual == "System User.Table Name")
+      }
+    }
+
     "splitComplexIdentifier" should {
       "throw on an empty identifier" in {
         assertThrows[IllegalArgumentException] {
@@ -576,6 +598,14 @@ class SqlGeneratorLoaderSuite extends AnyWordSpec with RelationalDbFixture {
       }
     }
 
+    "unquote" should {
+      "quote each subfields separately" in {
+        val actual = gen.unquote("System User.\"Table Name\"")
+
+        assert(actual == "System User.Table Name")
+      }
+    }
+
     "splitComplexIdentifier" should {
       "throw on an empty identifier" in {
         assertThrows[IllegalArgumentException] {
@@ -1025,6 +1055,14 @@ class SqlGeneratorLoaderSuite extends AnyWordSpec with RelationalDbFixture {
         assert(actual == "\"System User\".\"Table Name\"")
       }
     }
+
+    "unquote" should {
+      "quote each subfields separately" in {
+        val actual = genDate.unquote("System User.\"Table Name\"")
+
+        assert(actual == "System User.Table Name")
+      }
+    }
   }
 
   "MySQL SQL generator" should {
@@ -1147,6 +1185,14 @@ class SqlGeneratorLoaderSuite extends AnyWordSpec with RelationalDbFixture {
         assert(actual == "`System User`.`Table Name`")
       }
     }
+
+    "unquote" should {
+      "quote each subfields separately" in {
+        val actual = genDate.unquote("System User.`Table Name`")
+
+        assert(actual == "System User.Table Name")
+      }
+    }
   }
 
   "DB2 SQL generator" should {
@@ -1269,6 +1315,14 @@ class SqlGeneratorLoaderSuite extends AnyWordSpec with RelationalDbFixture {
         assert(actual == "\"System User\".\"Table Name\"")
       }
     }
+
+    "unquote" should {
+      "quote each subfields separately" in {
+        val actual = genDate.unquote("System User.\"Table Name\"")
+
+        assert(actual == "System User.Table Name")
+      }
+    }
   }
 
   "HSQL generator" should {
@@ -1560,6 +1614,14 @@ class SqlGeneratorLoaderSuite extends AnyWordSpec with RelationalDbFixture {
         }
       }
     }
+
+    "unquote" should {
+      "quote each subfields separately" in {
+        val actual = genDate.unquote("System User.\"Table Name\"")
+
+        assert(actual == "System User.Table Name")
+      }
+    }
   }
 
 }