#765 Pass neverRepairPartitions config to EnceladusSink via ExternalChannelFactoryV2 and add default value to HiveHelper methods.

Author: yruslan

pramen/core/src/main/scala/za/co/absa/pramen/core/utils/hive/HiveHelper.scala

@@ -29,15 +29,15 @@ abstract class HiveHelper {
                       partitionBy: Seq[String],
                       databaseName: Option[String],
                       tableName: String,
-                      neverRepairPartitions: Boolean): Unit
+                      neverRepairPartitions: Boolean = false): Unit
 
   def createOrUpdateHiveTable(path: String,
                               format: HiveFormat,
                               schema: StructType,
                               partitionBy: Seq[String],
                               databaseName: Option[String],
                               tableName: String,
-                              neverRepairPartitions: Boolean): Unit
+                              neverRepairPartitions: Boolean = false): Unit
 
   def replaceHiveTableSchema(schema: StructType,
                              partitionBy: Seq[String],

pramen/extras/src/main/scala/za/co/absa/pramen/extras/sink/EnceladusSink.scala

@@ -21,7 +21,8 @@ import org.apache.hadoop.fs.Path
 import org.apache.spark.sql.functions.lit
 import org.apache.spark.sql.{DataFrame, SparkSession}
 import org.slf4j.LoggerFactory
-import za.co.absa.pramen.api.{ExternalChannelFactory, MetastoreReader, Sink, SinkResult}
+import za.co.absa.pramen.api.{ExternalChannelFactoryV2, MetastoreReader, Sink, SinkResult}
+import za.co.absa.pramen.core.config.Keys.NEVER_REPAIR_PARTITIONS
 import za.co.absa.pramen.core.utils.ConfigUtils
 import za.co.absa.pramen.core.utils.hive.HiveQueryTemplates.TEMPLATES_DEFAULT_PREFIX
 import za.co.absa.pramen.core.utils.hive._
@@ -149,7 +150,8 @@ import scala.util.{Failure, Success, Try}
   */
 class EnceladusSink(sinkConfig: Config,
                     enceladusConfig: EnceladusConfig,
-                    hiveHelper: HiveHelper) extends Sink {
+                    hiveHelper: HiveHelper,
+                    neverRepairPartitions: Boolean) extends Sink {
 
   import za.co.absa.pramen.extras.sink.EnceladusSink._
 
@@ -449,7 +451,7 @@ class EnceladusSink(sinkConfig: Config,
 
       val schema = dfForHiveSchema.schema
 
-      hiveHelper.createOrUpdateHiveTable(publishBase, HiveFormat.Parquet, schema, Seq("enceladus_info_date", "enceladus_info_version"), enceladusConfig.hiveDatabase, hiveTable, neverRepairPartitions = false)
+      hiveHelper.createOrUpdateHiveTable(publishBase, HiveFormat.Parquet, schema, Seq("enceladus_info_date", "enceladus_info_version"), enceladusConfig.hiveDatabase, hiveTable, neverRepairPartitions)
     }
   }
 
@@ -486,7 +488,7 @@ class EnceladusSink(sinkConfig: Config,
   }
 }
 
-object EnceladusSink extends ExternalChannelFactory[EnceladusSink] {
+object EnceladusSink extends ExternalChannelFactoryV2[EnceladusSink] {
   val OUTPUT_PATH_KEY = "path"
   val INFO_VERSION_KEY = "info.version"
   val DATASET_NAME_KEY = "dataset.name"
@@ -500,15 +502,16 @@ object EnceladusSink extends ExternalChannelFactory[EnceladusSink] {
 
   val INFO_VERSION_AUTO_VALUE = "auto"
 
-  override def apply(conf: Config, parentPath: String, spark: SparkSession): EnceladusSink = {
+  override def apply(conf: Config, workflowConfig: Config, parentPath: String, spark: SparkSession): EnceladusSink = {
     val enceladusConfig = EnceladusConfig.fromConfig(conf)
     val alwaysEscapeColumnNames = ConfigUtils.getOptionBoolean(conf, HIVE_ALWAYS_ESCAPE_COLUMN_NAMES).getOrElse(true)
 
     val hiveTemplates = HiveQueryTemplates.fromConfig(ConfigUtils.getOptionConfig(conf, TEMPLATES_DEFAULT_PREFIX))
     val queryExecutor = QueryExecutorSpark(spark)
 
     val hiveHelper = new HiveHelperSql(queryExecutor, hiveTemplates, alwaysEscapeColumnNames)
+    val neverRepairPartitions = ConfigUtils.getOptionBoolean(workflowConfig, NEVER_REPAIR_PARTITIONS).getOrElse(false)
 
-    new EnceladusSink(conf, enceladusConfig, hiveHelper)
+    new EnceladusSink(conf, enceladusConfig, hiveHelper, neverRepairPartitions)
   }
 }

pramen/extras/src/test/scala/za/co/absa/pramen/extras/sink/EnceladusSinkSuite.scala

@@ -61,7 +61,7 @@ class EnceladusSinkSuite extends AnyWordSpec with SparkTestBase with TextCompari
       var sink: EnceladusSink = null
 
       "constructed from a config" in {
-        sink = EnceladusSink.apply(conf, "", spark)
+        sink = EnceladusSink.apply(conf, conf, "", spark)
 
         assert(sink.isInstanceOf[EnceladusSink])
       }
@@ -103,7 +103,7 @@ class EnceladusSinkSuite extends AnyWordSpec with SparkTestBase with TextCompari
     }
 
     "autoDetectVersionNumber" should {
-      val sink = EnceladusSink.apply(conf, "", spark)
+      val sink = EnceladusSink.apply(conf, conf, "", spark)
 
       withTempDirectory("enceladus_raw") { tempRaw =>
         withTempDirectory("enceladus_publish") { tempPublish =>
@@ -139,7 +139,7 @@ class EnceladusSinkSuite extends AnyWordSpec with SparkTestBase with TextCompari
 
       implicit val qe: QueryExecutorSpy = new QueryExecutorSpy
 
-      val sink = EnceladusSink.apply(conf, "", spark)
+      val sink = EnceladusSink.apply(conf, conf, "", spark)
 
       sink.runEnceladusIfNeeded("my_table",
         infoDate,
@@ -167,7 +167,7 @@ class EnceladusSinkSuite extends AnyWordSpec with SparkTestBase with TextCompari
 
       implicit val qe: QueryExecutorSpy = new QueryExecutorSpy(throwException = Some(new lang.IllegalStateException))
 
-      val sink = EnceladusSink.apply(conf, "", spark)
+      val sink = EnceladusSink.apply(conf, conf, "", spark)
 
       assertThrows[IllegalStateException] {
         sink.runEnceladusIfNeeded("my_table",
@@ -198,7 +198,7 @@ class EnceladusSinkSuite extends AnyWordSpec with SparkTestBase with TextCompari
 
       implicit val qe: QueryExecutorSpy = new QueryExecutorSpy
 
-      val sink = EnceladusSink.apply(conf, "", spark)
+      val sink = EnceladusSink.apply(conf, conf, "", spark)
 
       val ex = intercept[RuntimeException] {
         sink.runEnceladusIfNeeded("my_table",

pramen/project/build.properties

@@ -13,4 +13,4 @@
 # limitations under the License.
 #
 
-sbt.version=1.12.11
+sbt.version=1.12.13