Repository: AbsaOSS/pramen (language: python)

Issue:
Use more effective record count when JDBC source is used with an SQL query
## Background
This idea is reported by @filiphornak

Currently the record count is calculated this way if SQL expression (rather than table name) is used as an input to the JDBC source:
https://github.com/AbsaOSS/pramen/blob/0f8040a8bad151eccf5a6ee3403b2ae9c6a24b9e/pramen/core/src/main/scala/za/co/absa/pramen/core/reader/TableReaderJdbc.scala#L129-L129

This is not always effective since Spark does not always can get the record count without fetching all records.

## Feature
Use more effective record count when JDBC source is used with an SQL query.

## Proposed Solution
```
SELECT COUNT(*) AS CNT FROM (${query})
```


Discussion / hints:

Fix patch:
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 666b4abd6..0007dd0a1 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
@@ -22,7 +22,7 @@ import org.slf4j.LoggerFactory
 import za.co.absa.pramen.api.Query
 import za.co.absa.pramen.core.config.Keys
 import za.co.absa.pramen.core.reader.model.TableReaderJdbcConfig
-import za.co.absa.pramen.core.utils.{ConfigUtils, JdbcSparkUtils, TimeUtils}
+import za.co.absa.pramen.core.utils.{ConfigUtils, JdbcNativeUtils, JdbcSparkUtils, TimeUtils}
 
 import java.time.format.DateTimeFormatter
 import java.time.{Instant, LocalDate}
@@ -109,7 +109,18 @@ class TableReaderJdbc(jdbcReaderConfig: TableReaderJdbcConfig,
 
   private[core] def getCountForSql(sql: String, infoDateBegin: LocalDate, infoDateEnd: LocalDate): Long = {
     val filteredSql = TableReaderJdbcNative.getFilteredSql(sql, infoDateBegin, infoDateEnd)
-    getWithRetry[Long](filteredSql, isDataQuery = false, jdbcRetries, None)(df => df.count())
+    val countSql = s"SELECT COUNT(*) FROM ($filteredSql)"
+    var count = 0L
+
+    JdbcNativeUtils.withResultSet(jdbcUrlSelector, countSql, jdbcRetries) { rs =>
+      if (!rs.next())
+        throw new IllegalStateException(s"No rows returned by the count query: $countSql")
+      else {
+        count = rs.getLong(1)
+      }
+    }
+
+    count
   }
 
   private[core] def getDataForTable(tableName: String, infoDateBegin: LocalDate, infoDateEnd: LocalDate, columns: Seq[String]): DataFrame = {


Test patch:
diff --git a/pramen/core/src/test/scala/za/co/absa/pramen/core/tests/reader/TableReaderJdbcSuite.scala b/pramen/core/src/test/scala/za/co/absa/pramen/core/tests/reader/TableReaderJdbcSuite.scala
index 34bd07922..6f3a968f8 100644
--- a/pramen/core/src/test/scala/za/co/absa/pramen/core/tests/reader/TableReaderJdbcSuite.scala
+++ b/pramen/core/src/test/scala/za/co/absa/pramen/core/tests/reader/TableReaderJdbcSuite.scala
@@ -349,7 +349,7 @@ class TableReaderJdbcSuite extends AnyWordSpec with BeforeAndAfterAll with Spark
         assert(df.schema.fields.length == 7)
       }
 
-      "return selected columnd for a table snapshot-like query" in {
+      "return selected column for a table snapshot-like query" in {
         val testConfig = conf
         val reader = TableReaderJdbc(testConfig.getConfig("reader"), "reader")