Repository: AbsaOSS/pramen (language: python)

Issue:
Move 'optimize.exist.query' from JDBC properties to Hive properties
## Background
Just a small refactoring.

## Feature
Move 'optimize.exist.query' from JDBC properties to Hive properties.

## Example [Optional]
--

## Proposed Solution [Optional]
--


Discussion / hints:

Fix patch:
diff --git a/README.md b/README.md
index 16b10385e..64b536aa0 100644
--- a/README.md
+++ b/README.md
@@ -2843,10 +2843,10 @@ pramen {
     escape.column.names = true
 
     # If 
-    #  - true, uses this query for checking Hive table existence:
+    #  - true (default), uses this query for checking Hive table existence:
     #    DESCRIBE my_db.my_table
     #    (this is faster since it never touhces data, but may depend on Hive dialect)
-    #  - false (default), uses this query for checking Hive table existence:
+    #  - false uses this query for checking Hive table existence:
     #    SELECT 1 FROM my_db.my_table WHERE 0 = 1  
     optimize.exist.query = true
 
diff --git a/pramen/core/src/main/scala/za/co/absa/pramen/core/metastore/model/HiveConfig.scala b/pramen/core/src/main/scala/za/co/absa/pramen/core/metastore/model/HiveConfig.scala
index 4c2b50746..5aad98cc1 100644
--- a/pramen/core/src/main/scala/za/co/absa/pramen/core/metastore/model/HiveConfig.scala
+++ b/pramen/core/src/main/scala/za/co/absa/pramen/core/metastore/model/HiveConfig.scala
@@ -58,6 +58,7 @@ import za.co.absa.pramen.core.utils.hive.HiveQueryTemplates._
   * @param jdbcConfig              Hive JDBC configuration to use instead of Spark metastore if needed
   * @param ignoreFailures          Whether to ignore errors when creating or repairing tables. If true, only warnings will be emitted on Hive errors.
   * @param alwaysEscapeColumnNames If true, column names are always escaped when executing SQL against Hive.
+  * @param optimizeExistQuery      If true, Pramen uses Hive-specific SQL dialect to check table existence to ensure data won't be touched.
   */
 case class HiveConfig(
                        hiveApi: HiveApi,
@@ -65,7 +66,8 @@ case class HiveConfig(
                        templates: HiveQueryTemplates,
                        jdbcConfig: Option[JdbcConfig],
                        ignoreFailures: Boolean,
-                       alwaysEscapeColumnNames: Boolean
+                       alwaysEscapeColumnNames: Boolean,
+                       optimizeExistQuery: Boolean
                      )
 
 object HiveConfig {
@@ -114,13 +116,17 @@ object HiveConfig {
     val dropTableTemplate = ConfigUtils.getOptionString(conf, s"$HIVE_TEMPLATE_CONFIG_PREFIX.$DROP_TABLE_TEMPLATE_KEY")
       .getOrElse(defaultTemplates.dropTableTemplate)
 
+    val hiveOptimizeExistQuery = ConfigUtils.getOptionBoolean(conf, s"$HIVE_TEMPLATE_CONFIG_PREFIX.$HIVE_OPTIMIZE_EXIST_QUERY_KEY")
+      .getOrElse(defaults.optimizeExistQuery)
+
     HiveConfig(
       hiveApi = hiveApi,
       database = database,
       templates = HiveQueryTemplates(createTableTemplate, repairTableTemplate, addPartitionTableTemplate, dropTableTemplate),
       jdbcConfig = jdbcConfig,
       ignoreFailures,
-      alwaysEscapeColumnNames
+      alwaysEscapeColumnNames,
+      hiveOptimizeExistQuery
     )
   }
 
@@ -139,7 +145,7 @@ object HiveConfig {
       DEFAULT_DROP_TABLE_TEMPLATE
     ))
 
-    HiveConfig(defaults.hiveApi, defaults.database, templates, defaults.jdbcConfig, defaults.ignoreFailures, alwaysEscapeColumnNames = true)
+    HiveConfig(defaults.hiveApi, defaults.database, templates, defaults.jdbcConfig, defaults.ignoreFailures, alwaysEscapeColumnNames = true, optimizeExistQuery = true)
   }
 
   def getNullConfig: HiveConfig = HiveConfig(
@@ -148,5 +154,6 @@ object HiveConfig {
     HiveQueryTemplates(DEFAULT_CREATE_TABLE_TEMPLATE, DEFAULT_REPAIR_TABLE_TEMPLATE, DEFAULT_ADD_PARTITION_TEMPLATE, DEFAULT_DROP_TABLE_TEMPLATE),
     None,
     ignoreFailures = false,
-    alwaysEscapeColumnNames = true)
+    alwaysEscapeColumnNames = true,
+    optimizeExistQuery = true)
 }
diff --git a/pramen/core/src/main/scala/za/co/absa/pramen/core/metastore/model/HiveDefaultConfig.scala b/pramen/core/src/main/scala/za/co/absa/pramen/core/metastore/model/HiveDefaultConfig.scala
index cab9a9095..169223bb8 100644
--- a/pramen/core/src/main/scala/za/co/absa/pramen/core/metastore/model/HiveDefaultConfig.scala
+++ b/pramen/core/src/main/scala/za/co/absa/pramen/core/metastore/model/HiveDefaultConfig.scala
@@ -84,11 +84,12 @@ import za.co.absa.pramen.core.utils.hive.HiveQueryTemplates
   * }
   * }}}
   *
-  * @param hiveApi        The Hive API to use (SQL or Spark Catalog).
-  * @param database       T he database database to use. If omitted, you can use full table names for each table.
-  * @param templates      Query templates for each output format
-  * @param jdbcConfig     Hive JDBC configuration to use instead of Spark metastore if needed
-  * @param ignoreFailures Whether to ignore errors when creating or repairing tables. If true, only warnings will be emitted on Hive errors.
+  * @param hiveApi             The Hive API to use (SQL or Spark Catalog).
+  * @param database            The database database to use. If omitted, you can use full table names for each table.
+  * @param templates           Query templates for each output format
+  * @param jdbcConfig          Hive JDBC configuration to use instead of Spark metastore if needed
+  * @param ignoreFailures      Whether to ignore errors when creating or repairing tables. If true, only warnings will be emitted on Hive errors.
+  * @param optimizeExistQuery  If true, Pramen uses Hive-specific SQL dialect to check table existence to ensure data won't be touched.
   */
 case class HiveDefaultConfig(
                               hiveApi: HiveApi,
@@ -96,7 +97,8 @@ case class HiveDefaultConfig(
                               templates: Map[String, HiveQueryTemplates],
                               jdbcConfig: Option[JdbcConfig],
                               ignoreFailures: Boolean,
-                              alwaysEscapeColumnNames: Boolean
+                              alwaysEscapeColumnNames: Boolean,
+                              optimizeExistQuery: Boolean
                             )
 
 object HiveDefaultConfig {
@@ -107,6 +109,7 @@ object HiveDefaultConfig {
   val HIVE_IGNORE_FAILURES_KEY = "ignore.failures"
   val HIVE_ALWAYS_ESCAPE_COLUMN_NAMES = "escape.column.names"
   val HIVE_DATABASE_KEY = "database"
+  val HIVE_OPTIMIZE_EXIST_QUERY_KEY = "optimize.exist.query"
 
   /**
     * Gets global Hive configuration and query templates for all supported formats.
@@ -122,6 +125,7 @@ object HiveDefaultConfig {
     val database = if (conf.hasPath(HIVE_DATABASE_KEY)) Some(conf.getString(HIVE_DATABASE_KEY)) else None
     val ignoreFailures = ConfigUtils.getOptionBoolean(conf, HIVE_IGNORE_FAILURES_KEY).getOrElse(false)
     val alwaysEscapeColumnNames = ConfigUtils.getOptionBoolean(conf, HIVE_ALWAYS_ESCAPE_COLUMN_NAMES).getOrElse(true)
+    val hiveOptimizeExistQuery = ConfigUtils.getOptionBoolean(conf, HIVE_OPTIMIZE_EXIST_QUERY_KEY).getOrElse(true)
 
     val jdbcConfig = if (conf.hasPath(HIVE_CONFIG_JDBC_PREFIX))
       Option(JdbcConfig.load(conf, parent))
@@ -136,8 +140,16 @@ object HiveDefaultConfig {
       (format, HiveQueryTemplates.fromConfig(ConfigUtils.getOptionConfig(conf, prefix)))
     }).toMap
 
-    HiveDefaultConfig(hiveApi, database, templates, jdbcConfig, ignoreFailures, alwaysEscapeColumnNames)
+    HiveDefaultConfig(hiveApi, database, templates, jdbcConfig, ignoreFailures, alwaysEscapeColumnNames, hiveOptimizeExistQuery)
   }
 
-  def getNullConfig: HiveDefaultConfig = HiveDefaultConfig(HiveApi.Sql, None, Map(), None, ignoreFailures = false, alwaysEscapeColumnNames = true)
+  def getNullConfig: HiveDefaultConfig = HiveDefaultConfig(
+    HiveApi.Sql,
+    None,
+    Map(),
+    None,
+    ignoreFailures = false,
+    alwaysEscapeColumnNames = true,
+    optimizeExistQuery = true
+  )
 }
diff --git a/pramen/core/src/main/scala/za/co/absa/pramen/core/reader/model/JdbcConfig.scala b/pramen/core/src/main/scala/za/co/absa/pramen/core/reader/model/JdbcConfig.scala
index 584f58553..55762f0e3 100644
--- a/pramen/core/src/main/scala/za/co/absa/pramen/core/reader/model/JdbcConfig.scala
+++ b/pramen/core/src/main/scala/za/co/absa/pramen/core/reader/model/JdbcConfig.scala
@@ -34,7 +34,6 @@ case class JdbcConfig(
                        connectionTimeoutSeconds: Option[Int] = None,
                        sanitizeDateTime: Boolean = true,
                        incorrectDecimalsAsString: Boolean = false,
-                       optimizedExistQuery: Boolean = false,
                        extraOptions: Map[String, String] = Map.empty[String, String]
                      )
 
@@ -53,7 +52,6 @@ object JdbcConfig {
   val JDBC_CONNECTION_TIMEOUT = "jdbc.connection.timeout"
   val JDBC_SANITIZE_DATETIME = "jdbc.sanitize.datetime"
   val JDBC_INCORRECT_PRECISION_AS_STRING = "jdbc.incorrect.precision.as.string"
-  val JDBC_OPTIMIZED_EXIST_QUERY = "jdbc.optimize.exist.query"
   val JDBC_EXTRA_OPTIONS_PREFIX = "jdbc.option"
 
   def load(conf: Config, parent: String = ""): JdbcConfig = {
@@ -86,7 +84,6 @@ object JdbcConfig {
       connectionTimeoutSeconds = ConfigUtils.getOptionInt(conf, JDBC_CONNECTION_TIMEOUT),
       sanitizeDateTime = ConfigUtils.getOptionBoolean(conf, JDBC_SANITIZE_DATETIME).getOrElse(true),
       incorrectDecimalsAsString = ConfigUtils.getOptionBoolean(conf, JDBC_INCORRECT_PRECISION_AS_STRING).getOrElse(false),
-      optimizedExistQuery = ConfigUtils.getOptionBoolean(conf, JDBC_OPTIMIZED_EXIST_QUERY).getOrElse(false),
       extraOptions = ConfigUtils.getExtraOptions(conf, JDBC_EXTRA_OPTIONS_PREFIX)
     )
   }
diff --git a/pramen/core/src/main/scala/za/co/absa/pramen/core/utils/ConfigUtils.scala b/pramen/core/src/main/scala/za/co/absa/pramen/core/utils/ConfigUtils.scala
index b704f19ae..349c15af2 100644
--- a/pramen/core/src/main/scala/za/co/absa/pramen/core/utils/ConfigUtils.scala
+++ b/pramen/core/src/main/scala/za/co/absa/pramen/core/utils/ConfigUtils.scala
@@ -199,7 +199,7 @@ object ConfigUtils {
           case list: ConfigList =>
             val lst = list.unwrapped()
 
-            if (!lst.isEmpty && lst.get(0).isInstanceOf[java.util.Map[String, AnyRef]]) {
+            if (!lst.isEmpty && lst.get(0).isInstanceOf[java.util.Map[_, _]]) {
               k -> v.render(renderOptions)
             } else  {
               k -> v.unwrapped()
diff --git a/pramen/core/src/main/scala/za/co/absa/pramen/core/utils/hive/HiveHelper.scala b/pramen/core/src/main/scala/za/co/absa/pramen/core/utils/hive/HiveHelper.scala
index d6de4e27d..87b9e50fd 100644
--- a/pramen/core/src/main/scala/za/co/absa/pramen/core/utils/hive/HiveHelper.scala
+++ b/pramen/core/src/main/scala/za/co/absa/pramen/core/utils/hive/HiveHelper.scala
@@ -57,7 +57,7 @@ object HiveHelper {
         val queryExecutor = hiveConfig.jdbcConfig match {
           case Some(jdbcConfig) =>
             log.info(s"Using Hive SQL API by connecting to the Hive metastore via JDBC.")
-            new QueryExecutorJdbc(JdbcUrlSelector(jdbcConfig))
+            new QueryExecutorJdbc(JdbcUrlSelector(jdbcConfig), hiveConfig.optimizeExistQuery)
           case None             =>
             log.info(s"Using Hive SQL API by connecting to the Hive metastore via Spark.")
             new QueryExecutorSpark()
diff --git a/pramen/core/src/main/scala/za/co/absa/pramen/core/utils/hive/QueryExecutorJdbc.scala b/pramen/core/src/main/scala/za/co/absa/pramen/core/utils/hive/QueryExecutorJdbc.scala
index be339c72d..be15e3f7e 100644
--- a/pramen/core/src/main/scala/za/co/absa/pramen/core/utils/hive/QueryExecutorJdbc.scala
+++ b/pramen/core/src/main/scala/za/co/absa/pramen/core/utils/hive/QueryExecutorJdbc.scala
@@ -24,7 +24,7 @@ import java.sql.{Connection, ResultSet, SQLException, SQLSyntaxErrorException}
 import scala.util.{Failure, Try}
 import scala.util.control.NonFatal
 
-class QueryExecutorJdbc(jdbcUrlSelector: JdbcUrlSelector) extends QueryExecutor {
+class QueryExecutorJdbc(jdbcUrlSelector: JdbcUrlSelector, optimizedExistQuery: Boolean) extends QueryExecutor {
   private val log = LoggerFactory.getLogger(this.getClass)
   private var connection: Connection = _
   private val defaultRetries = jdbcUrlSelector.getNumberOfUrls
@@ -34,7 +34,7 @@ class QueryExecutorJdbc(jdbcUrlSelector: JdbcUrlSelector) extends QueryExecutor
   override def doesTableExist(dbName: Option[String], tableName: String): Boolean = {
     val fullTableName = HiveHelper.getFullTable(dbName, tableName)
 
-    val query = if (jdbcUrlSelector.jdbcConfig.optimizedExistQuery) {
+    val query = if (optimizedExistQuery) {
       s"DESCRIBE $fullTableName"
     } else {
       s"SELECT 1 FROM $fullTableName WHERE 0 = 1"
@@ -97,9 +97,9 @@ class QueryExecutorJdbc(jdbcUrlSelector: JdbcUrlSelector) extends QueryExecutor
 }
 
 object QueryExecutorJdbc {
-  def fromJdbcConfig(jdbcConfig: JdbcConfig): QueryExecutorJdbc = {
+  def fromJdbcConfig(jdbcConfig: JdbcConfig, optimizedExistQuery: Boolean): QueryExecutorJdbc = {
     val jdbcUrlSelector = JdbcUrlSelector(jdbcConfig)
 
-    new QueryExecutorJdbc(jdbcUrlSelector)
+    new QueryExecutorJdbc(jdbcUrlSelector, optimizedExistQuery)
   }
 }
diff --git a/pramen/extras/src/main/scala/za/co/absa/pramen/extras/sink/StandardizationConfig.scala b/pramen/extras/src/main/scala/za/co/absa/pramen/extras/sink/StandardizationConfig.scala
index 06f8966fc..4a8479fb5 100644
--- a/pramen/extras/src/main/scala/za/co/absa/pramen/extras/sink/StandardizationConfig.scala
+++ b/pramen/extras/src/main/scala/za/co/absa/pramen/extras/sink/StandardizationConfig.scala
@@ -37,6 +37,7 @@ case class StandardizationConfig(
                                   hiveJdbcConfig: Option[JdbcConfig],
                                   hiveDatabase: Option[String],
                                   hiveIgnoreFailures: Boolean,
+                                  hiveOptimizeExistQuery: Boolean,
                                   infoDateColumn: String,
                                   infoDateStringColumn: String,
                                   infoVersionColumn: String
@@ -55,6 +56,7 @@ object StandardizationConfig {
   val HIVE_JDBC_PREFIX = "hive"
   val HIVE_DATABASE_KEY = "hive.database"
   val HIVE_IGNORE_FAILURES_KEY = "hive.ignore.failures"
+  val HIVE_OPTIMIZE_EXIST_QUERY_KEY = "hive.optimize.exist.query"
 
   val INFO_DATE_COLUMN_KEY = "info.date.column"
   val INFO_DATE_STRING_COLUMN_KEY = "info.date.str.column"
@@ -99,6 +101,7 @@ object StandardizationConfig {
       hiveJdbcConfig,
       ConfigUtils.getOptionString(conf, HIVE_DATABASE_KEY),
       ConfigUtils.getOptionBoolean(conf, HIVE_IGNORE_FAILURES_KEY).getOrElse(false),
+      hiveOptimizeExistQuery = ConfigUtils.getOptionBoolean(conf, HIVE_OPTIMIZE_EXIST_QUERY_KEY).getOrElse(true),
       infoDateColumn,
       infoDateStringColumn,
       infoVersionColumn
diff --git a/pramen/extras/src/main/scala/za/co/absa/pramen/extras/sink/StandardizationSink.scala b/pramen/extras/src/main/scala/za/co/absa/pramen/extras/sink/StandardizationSink.scala
index d75aa65c5..5d65ba0e6 100644
--- a/pramen/extras/src/main/scala/za/co/absa/pramen/extras/sink/StandardizationSink.scala
+++ b/pramen/extras/src/main/scala/za/co/absa/pramen/extras/sink/StandardizationSink.scala
@@ -432,7 +432,7 @@ object StandardizationSink extends ExternalChannelFactory[StandardizationSink] {
         val queryExecutor = standardizationConfig.hiveJdbcConfig match {
           case Some(hiveJdbcConfig) =>
             log.info(s"Using Hive SQL API by connecting to the Hive metastore via JDBC.")
-            QueryExecutorJdbc.fromJdbcConfig(hiveJdbcConfig)
+            QueryExecutorJdbc.fromJdbcConfig(hiveJdbcConfig, standardizationConfig.hiveOptimizeExistQuery)
           case None                 =>
             log.info(s"Using Hive SQL API by connecting to the Hive metastore via Spark.")
             QueryExecutorSpark(spark)


Test patch:
diff --git a/pramen/core/src/test/scala/za/co/absa/pramen/core/metastore/model/HiveConfigSuite.scala b/pramen/core/src/test/scala/za/co/absa/pramen/core/metastore/model/HiveConfigSuite.scala
index 566dba193..7edf8133e 100644
--- a/pramen/core/src/test/scala/za/co/absa/pramen/core/metastore/model/HiveConfigSuite.scala
+++ b/pramen/core/src/test/scala/za/co/absa/pramen/core/metastore/model/HiveConfigSuite.scala
@@ -32,7 +32,8 @@ class HiveConfigSuite extends AnyWordSpec {
         Map("parquet" -> HiveQueryTemplates("create1", "repair1", "add_partition1", "drop1")),
         None,
         ignoreFailures = true,
-        alwaysEscapeColumnNames = false)
+        alwaysEscapeColumnNames = false,
+        optimizeExistQuery = true)
 
       val hiveConfig = HiveConfig.fromConfigWithDefaults(conf, defaultConfig, DataFormat.Parquet("dummy", None))
 
@@ -41,6 +42,7 @@ class HiveConfigSuite extends AnyWordSpec {
       assert(hiveConfig.jdbcConfig.isEmpty)
       assert(hiveConfig.ignoreFailures)
       assert(!hiveConfig.alwaysEscapeColumnNames)
+      assert(hiveConfig.optimizeExistQuery)
       assert(hiveConfig.templates.createTableTemplate.contains("create1"))
       assert(hiveConfig.templates.repairTableTemplate.contains("repair1"))
       assert(hiveConfig.templates.addPartitionTemplate.contains("add_partition1"))
@@ -67,6 +69,7 @@ class HiveConfigSuite extends AnyWordSpec {
           |   repair.table.template = "repair2"
           |   add.partition.template = "add_partition2"
           |   drop.table.template = "drop2"
+          |   optimize.exist.query = false
           |}
           |""".stripMargin)
 
@@ -76,7 +79,8 @@ class HiveConfigSuite extends AnyWordSpec {
         Map("parquet" -> HiveQueryTemplates("create1", "repair1", "add_partition1", "drop1")),
         None,
         ignoreFailures = false,
-        alwaysEscapeColumnNames = false)
+        alwaysEscapeColumnNames = false,
+        optimizeExistQuery = true)
 
       val hiveConfig = HiveConfig.fromConfigWithDefaults(conf, defaultConfig, DataFormat.Parquet("dummy", None))
 
@@ -86,6 +90,7 @@ class HiveConfigSuite extends AnyWordSpec {
       assert(hiveConfig.jdbcConfig.map(_.driver).contains("driver2"))
       assert(hiveConfig.ignoreFailures)
       assert(hiveConfig.alwaysEscapeColumnNames)
+      assert(!hiveConfig.optimizeExistQuery)
       assert(hiveConfig.templates.createTableTemplate.contains("create2"))
       assert(hiveConfig.templates.repairTableTemplate.contains("repair2"))
       assert(hiveConfig.templates.addPartitionTemplate.contains("add_partition2"))
@@ -101,7 +106,8 @@ class HiveConfigSuite extends AnyWordSpec {
         Map("parquet" -> HiveQueryTemplates("create", "repair", "add_partition1", "drop")),
         None,
         ignoreFailures = true,
-        alwaysEscapeColumnNames = true)
+        alwaysEscapeColumnNames = true,
+        optimizeExistQuery = true)
 
       val hiveConfig = HiveConfig.fromDefaults(defaultConfig, DataFormat.Parquet("dummy", None))
 
@@ -110,6 +116,7 @@ class HiveConfigSuite extends AnyWordSpec {
       assert(hiveConfig.jdbcConfig.isEmpty)
       assert(hiveConfig.ignoreFailures)
       assert(hiveConfig.alwaysEscapeColumnNames)
+      assert(hiveConfig.optimizeExistQuery)
       assert(hiveConfig.templates.createTableTemplate.contains("create"))
       assert(hiveConfig.templates.repairTableTemplate.contains("repair"))
       assert(hiveConfig.templates.addPartitionTemplate.contains("add_partition1"))
diff --git a/pramen/core/src/test/scala/za/co/absa/pramen/core/metastore/model/MetaTableSuite.scala b/pramen/core/src/test/scala/za/co/absa/pramen/core/metastore/model/MetaTableSuite.scala
index 0145bc5e1..188184901 100644
--- a/pramen/core/src/test/scala/za/co/absa/pramen/core/metastore/model/MetaTableSuite.scala
+++ b/pramen/core/src/test/scala/za/co/absa/pramen/core/metastore/model/MetaTableSuite.scala
@@ -198,7 +198,7 @@ class MetaTableSuite extends AnyWordSpec {
         Some(JdbcConfig("driver", Some("url"),
           user = Some("user"),
           password = Some("pass")
-        )), ignoreFailures = true, alwaysEscapeColumnNames = false)
+        )), ignoreFailures = true, alwaysEscapeColumnNames = false, optimizeExistQuery = true)
 
       val appConf = ConfigFactory.parseString("pramen.default.records.per.partition = 100")
 
@@ -262,7 +262,7 @@ class MetaTableSuite extends AnyWordSpec {
         Some(JdbcConfig("driver1", Some("url1"),
           user = Some("user1"),
           password = Some("pass1")
-        )), ignoreFailures = false, alwaysEscapeColumnNames = false)
+        )), ignoreFailures = false, alwaysEscapeColumnNames = false, optimizeExistQuery = true)
 
       val appConf = ConfigFactory.parseString("pramen.default.records.per.partition = 100")
 
diff --git a/pramen/core/src/test/scala/za/co/absa/pramen/core/tests/utils/hive/QueryExecutorJdbcSuite.scala b/pramen/core/src/test/scala/za/co/absa/pramen/core/tests/utils/hive/QueryExecutorJdbcSuite.scala
index 3613aeba2..56519b736 100644
--- a/pramen/core/src/test/scala/za/co/absa/pramen/core/tests/utils/hive/QueryExecutorJdbcSuite.scala
+++ b/pramen/core/src/test/scala/za/co/absa/pramen/core/tests/utils/hive/QueryExecutorJdbcSuite.scala
@@ -48,21 +48,21 @@ class QueryExecutorJdbcSuite extends AnyWordSpec with BeforeAndAfterAll with Rel
 
   "QueryExecutorJdbc" should {
     "be constructed from JdbcConfig" in {
-      val qe = QueryExecutorJdbc.fromJdbcConfig(jdbcConfig)
+      val qe = QueryExecutorJdbc.fromJdbcConfig(jdbcConfig, optimizedExistQuery = false)
 
       qe.execute("UPDATE company SET id = 200 WHERE id = 100")
       qe.close()
     }
 
     "execute JDBC queries" in {
-      val qe = new QueryExecutorJdbc(JdbcUrlSelector(jdbcConfig))
+      val qe = new QueryExecutorJdbc(JdbcUrlSelector(jdbcConfig), optimizedExistQuery = true)
 
       qe.execute("SELECT * FROM company")
       qe.close()
     }
 
     "execute CREATE TABLE queries" in {
-      val qe = new QueryExecutorJdbc(JdbcUrlSelector(jdbcConfig))
+      val qe = new QueryExecutorJdbc(JdbcUrlSelector(jdbcConfig), optimizedExistQuery = false)
 
       qe.execute("CREATE TABLE my_table (id INT)")
 
@@ -74,7 +74,7 @@ class QueryExecutorJdbcSuite extends AnyWordSpec with BeforeAndAfterAll with Rel
     }
 
     "throw an exception on errors" in {
-      val qe = new QueryExecutorJdbc(JdbcUrlSelector(jdbcConfig))
+      val qe = new QueryExecutorJdbc(JdbcUrlSelector(jdbcConfig), optimizedExistQuery = false)
 
       val ex = intercept[SQLSyntaxErrorException] {
         qe.execute("SELECT * FROM does_not_exist")
@@ -84,7 +84,7 @@ class QueryExecutorJdbcSuite extends AnyWordSpec with BeforeAndAfterAll with Rel
     }
 
     "return true if the table is found" in {
-      val qe = new QueryExecutorJdbc(JdbcUrlSelector(jdbcConfig))
+      val qe = new QueryExecutorJdbc(JdbcUrlSelector(jdbcConfig), optimizedExistQuery = false)
 
       val exist = qe.doesTableExist(None, "company")
 
@@ -94,7 +94,7 @@ class QueryExecutorJdbcSuite extends AnyWordSpec with BeforeAndAfterAll with Rel
     }
 
     "return false if the table is not found" in {
-      val qe = new QueryExecutorJdbc(JdbcUrlSelector(jdbcConfig))
+      val qe = new QueryExecutorJdbc(JdbcUrlSelector(jdbcConfig), optimizedExistQuery = false)
 
       val exist = qe.doesTableExist(Option(database), "does_not_exist")
 
@@ -104,7 +104,7 @@ class QueryExecutorJdbcSuite extends AnyWordSpec with BeforeAndAfterAll with Rel
     }
 
     "return false if the table is not found in an optimized query" in {
-      val qe = new QueryExecutorJdbc(JdbcUrlSelector(jdbcConfig.copy(optimizedExistQuery = true)))
+      val qe = new QueryExecutorJdbc(JdbcUrlSelector(jdbcConfig), optimizedExistQuery = true)
 
       val exist = qe.doesTableExist(Option(database), "does_not_exist")
 
@@ -114,7 +114,7 @@ class QueryExecutorJdbcSuite extends AnyWordSpec with BeforeAndAfterAll with Rel
     }
 
     "return false if the table is not found in an optimized query without a database" in {
-      val qe = new QueryExecutorJdbc(JdbcUrlSelector(jdbcConfig.copy(optimizedExistQuery = true)))
+      val qe = new QueryExecutorJdbc(JdbcUrlSelector(jdbcConfig), optimizedExistQuery = true)
 
       val exist = qe.doesTableExist(None, "does_not_exist")
 
@@ -131,7 +131,7 @@ class QueryExecutorJdbcSuite extends AnyWordSpec with BeforeAndAfterAll with Rel
       whenMock(sel.jdbcConfig).thenReturn(jdbcConfig)
       whenMock(sel.getWorkingConnection(anyInt())).thenReturn((conn, "dummyurl"))
 
-      val qe = new QueryExecutorJdbc(sel)
+      val qe = new QueryExecutorJdbc(sel, true)
       qe.execute("SELECT * FROM company")
 
       var execution = 0
@@ -162,7 +162,7 @@ class QueryExecutorJdbcSuite extends AnyWordSpec with BeforeAndAfterAll with Rel
         .thenReturn((conn, "dummyurl"))
         .thenThrow(new RuntimeException("fail the second time"))
 
-      val qe = new QueryExecutorJdbc(sel)
+      val qe = new QueryExecutorJdbc(sel, true)
 
       var execution = 0
       var actionExecuted = false
@@ -197,7 +197,7 @@ class QueryExecutorJdbcSuite extends AnyWordSpec with BeforeAndAfterAll with Rel
         .thenThrow(new RuntimeException("fail the second time"))
         .thenReturn((conn, "dummyurl"))
 
-      val qe = new QueryExecutorJdbc(sel)
+      val qe = new QueryExecutorJdbc(sel, true)
 
       var execution = 0
       var actionExecuted = false