Create sink only once (#528)

* Create sink only once

* Create sources only once as well

Author: kevinwallimann

pramen/core/src/main/scala/za/co/absa/pramen/core/pipeline/OperationSplitter.scala

@@ -56,13 +56,9 @@ class OperationSplitter(conf: Config,
                       sourceTables: Seq[SourceTable])(implicit spark: SparkSession): Seq[Job] = {
     val specialCharacters = conf.getString(SPECIAL_CHARACTERS_IN_COLUMN_NAMES)
     val temporaryDirectory = ConfigUtils.getOptionString(conf, TEMPORARY_DIRECTORY_KEY)
-    val sourceBase = SourceManager.getSourceByName(sourceName, conf, None)
 
     sourceTables.map(sourceTable => {
-      val source = sourceTable.overrideConf match {
-        case Some(confOverride) => SourceManager.getSourceByName(sourceName, conf, Some(confOverride))
-        case None               => sourceBase
-      }
+      val source = SourceManager.getSourceByName(sourceName, conf, sourceTable.overrideConf)
 
       val disableCountQuery = ConfigUtils.getOptionBoolean(source.config, DISABLE_COUNT_QUERY).getOrElse(false)
       val outputTable = metastore.getTableDef(sourceTable.metaTableName)
@@ -85,19 +81,10 @@ class OperationSplitter(conf: Config,
                      tables: Seq[TransferTable])(implicit spark: SparkSession): Seq[Job] = {
     val specialCharacters = conf.getString(SPECIAL_CHARACTERS_IN_COLUMN_NAMES)
     val temporaryDirectory = ConfigUtils.getOptionString(conf, TEMPORARY_DIRECTORY_KEY)
-    val sourceBase = SourceManager.getSourceByName(sourceName, conf, None)
-    val sinkBase = SinkManager.getSinkByName(sinkName, conf, None)
 
     tables.map(transferTable => {
-      val source = transferTable.sourceOverrideConf match {
-        case Some(confOverride) => SourceManager.getSourceByName(sourceName, conf, Some(confOverride))
-        case None               => sourceBase
-      }
-
-      val sink = transferTable.sinkOverrideConf match {
-        case Some(confOverride) => SinkManager.getSinkByName(sinkName, conf, Some(confOverride))
-        case None               => sinkBase
-      }
+      val source = SourceManager.getSourceByName(sourceName, conf, transferTable.sourceOverrideConf)
+      val sink = SinkManager.getSinkByName(sinkName, conf, transferTable.sinkOverrideConf)
 
       val disableCountQuery = ConfigUtils.getOptionBoolean(source.config, DISABLE_COUNT_QUERY).getOrElse(false)
       val outputTable = TransferTableParser.getMetaTable(transferTable)
@@ -146,15 +133,10 @@ class OperationSplitter(conf: Config,
                  sinkName: String,
                  sinkTables: Seq[SinkTable])
                 (implicit spark: SparkSession): Seq[Job] = {
-    val sinkBase = SinkManager.getSinkByName(sinkName, conf, None)
-
     sinkTables.map(sinkTable => {
       val inputTable = metastore.getTableDef(sinkTable.metaTableName)
 
-      val sink = sinkTable.overrideConf match {
-        case Some(confOverride) => SinkManager.getSinkByName(sinkName, conf, Some(confOverride))
-        case None               => sinkBase
-      }
+      val sink = SinkManager.getSinkByName(sinkName, conf, sinkTable.overrideConf)
 
       val outputTableName = sinkTable.outputTableName.getOrElse(s"${sinkTable.metaTableName}->$sinkName")