Repository: AbsaOSS/pramen (language: python)

Issue:
Add support for metadata of PostgreSQL `UUID` column type
## Background
PostgreSQL has `UUID` column type. When Pramen loads the schema, metadata of `varchar(n)` fields is extracted as `maxLength` metadata key. But for `UUID` this is not happening, so the resulting data type is just `string`.

## Feature
Add support for metadata of PostgreSQL `UUID` column type.

## Example [Optional]
--

## Proposed Solution [Optional]
Need to investigate if it is possible to get the metadata information for `UUID` columns in the first place



Discussion / hints:

Fix patch:
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 92a91ae54..4a1a93b88 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
@@ -32,6 +32,7 @@ object JdbcSparkUtils {
   private val log = LoggerFactory.getLogger(this.getClass)
 
   val MAXIMUM_VARCHAR_LENGTH = 8192
+  val MAXIMUM_UUID_LENGTH = 50
 
   /**
     * Adds metadata to Spark fields based on JDBC metadata.
@@ -61,7 +62,10 @@ object JdbcSparkUtils {
           field.dataType match {
             case StringType if fieldToMetadataMap.contains(fieldNameLowerCase) =>
               val jdbcMetadata = fieldToMetadataMap(fieldNameLowerCase)
-              val maxLength = Math.max(jdbcMetadata.displaySize, jdbcMetadata.precision)
+              val maxLength = jdbcMetadata.sqlTypeName.toLowerCase() match {
+                case "uuid" => MAXIMUM_UUID_LENGTH
+                case _      => Math.max(jdbcMetadata.displaySize, jdbcMetadata.precision)
+              }
               if (maxLength > 0 && maxLength < MAXIMUM_VARCHAR_LENGTH) {
                 val metadata = new MetadataBuilder
                 metadata.withMetadata(field.metadata)
@@ -180,10 +184,12 @@ object JdbcSparkUtils {
                       (action: (Connection, ResultSetMetaData) => Unit): Unit = {
     val (url, connection) = JdbcNativeUtils.getConnection(jdbcConfig)
 
+    connection.setAutoCommit(false)
+
     log.info(s"Successfully connected to JDBC URL: $url")
 
     try {
-      withResultSet(connection, nativeQuery) { rs =>
+      withMetadataResultSet(connection, nativeQuery) { rs =>
         action(connection, rs.getMetaData)
       }
     } finally {
@@ -195,23 +201,16 @@ object JdbcSparkUtils {
     * Executes a query against a JDBC connection and allows running an action on the result set.
     * Handles the closure of created objects.
     *
+    * This method is opinionated on the cursor type and is used exclusively for metadata extraction.
+    *
     * @param connection  a JDBC connection.
     * @param query       a SQL query in the dialect native to the database.
     * @param action      the action to execute on the result set.
     */
-  def withResultSet(connection: Connection,
-                    query: String)
-                   (action: ResultSet => Unit): Unit = {
-    val statement = try {
-      connection.createStatement(ResultSet.TYPE_SCROLL_INSENSITIVE, ResultSet.CONCUR_READ_ONLY)
-    } catch {
-      case _: java.sql.SQLException =>
-        // Fallback with more permissible result type.
-        // JDBC sources should automatically downgrade result type, but Denodo driver doesn't do that.
-        connection.createStatement(ResultSet.TYPE_FORWARD_ONLY, ResultSet.CONCUR_READ_ONLY)
-      case NonFatal(ex) =>
-        throw ex
-    }
+  private[core] def withMetadataResultSet(connection: Connection,
+                                          query: String)
+                                         (action: ResultSet => Unit): Unit = {
+    val statement = connection.createStatement(ResultSet.TYPE_FORWARD_ONLY, ResultSet.CONCUR_READ_ONLY)
 
     try {
       val resultSet = statement.executeQuery(query)


Test patch:
diff --git a/pramen/core/src/test/scala/za/co/absa/pramen/core/tests/utils/JdbcSparkUtilsSuite.scala b/pramen/core/src/test/scala/za/co/absa/pramen/core/tests/utils/JdbcSparkUtilsSuite.scala
index 386727777..b5723126b 100644
--- a/pramen/core/src/test/scala/za/co/absa/pramen/core/tests/utils/JdbcSparkUtilsSuite.scala
+++ b/pramen/core/src/test/scala/za/co/absa/pramen/core/tests/utils/JdbcSparkUtilsSuite.scala
@@ -143,9 +143,9 @@ class JdbcSparkUtilsSuite extends AnyWordSpec with BeforeAndAfterAll with SparkT
     }
   }
 
-  "withResultSet" should {
+  "withMetadataResultSet" should {
     "provide the resultset object for the query" in {
-      JdbcSparkUtils.withResultSet(getConnection, s"SELECT * FROM ${RdbExampleTable.Company.tableName}") { rs =>
+      JdbcSparkUtils.withMetadataResultSet(getConnection, s"SELECT * FROM ${RdbExampleTable.Company.tableName}") { rs =>
         assert(rs.getMetaData.getColumnName(2) == "NAME")
       }
     }