Fix reruns so that they are not reported as running based on retrospe…

…ctive updates.
AbsaOSS · Oct 31, 2024 · 3623a87 · 3623a87
1 parent bf58920
commit 3623a87
Show file tree

Hide file tree

Showing 3 changed files with 10 additions and 6 deletions.
diff --git a/pramen/core/src/main/scala/za/co/absa/pramen/core/pipeline/JobBase.scala b/pramen/core/src/main/scala/za/co/absa/pramen/core/pipeline/JobBase.scala
@@ -106,10 +106,14 @@ abstract class JobBase(operationDef: OperationDef,
     }
   }
 
-  protected def preRunTransformationCheck(infoDate: LocalDate, dependencyWarnings: Seq[DependencyWarning]): JobPreRunResult = {
-    validateTransformationAlreadyRanCases(infoDate, dependencyWarnings) match {
-      case Some(result) => result
-      case None => JobPreRunResult(JobPreRunStatus.Ready, None, dependencyWarnings, Seq.empty[String])
+  protected def preRunTransformationCheck(infoDate: LocalDate, dependencyWarnings: Seq[DependencyWarning], runReason: TaskRunReason): JobPreRunResult = {
+    if (runReason == TaskRunReason.Rerun) {
+      JobPreRunResult(JobPreRunStatus.Ready, None, dependencyWarnings, Seq.empty[String])
+    } else {
+      validateTransformationAlreadyRanCases(infoDate, dependencyWarnings) match {
+        case Some(result) => result
+        case None         => JobPreRunResult(JobPreRunStatus.Ready, None, dependencyWarnings, Seq.empty[String])
+      }
     }
   }
 

diff --git a/pramen/core/src/main/scala/za/co/absa/pramen/core/pipeline/SinkJob.scala b/pramen/core/src/main/scala/za/co/absa/pramen/core/pipeline/SinkJob.scala
@@ -48,7 +48,7 @@ class SinkJob(operationDef: OperationDef,
   override val scheduleStrategy: ScheduleStrategy = new ScheduleStrategySourcing
 
   override def preRunCheckJob(infoDate: LocalDate, runReason: TaskRunReason, jobConfig: Config, dependencyWarnings: Seq[DependencyWarning]): JobPreRunResult = {
-    val alreadyRanStatus = preRunTransformationCheck(infoDate, dependencyWarnings)
+    val alreadyRanStatus = preRunTransformationCheck(infoDate, dependencyWarnings, runReason)
 
     alreadyRanStatus.status match {
       case JobPreRunStatus.Ready => JobPreRunResult(Ready, Some(getDataDf(infoDate).count()), dependencyWarnings, alreadyRanStatus.warnings)

diff --git a/pramen/core/src/main/scala/za/co/absa/pramen/core/pipeline/TransformationJob.scala b/pramen/core/src/main/scala/za/co/absa/pramen/core/pipeline/TransformationJob.scala
@@ -41,7 +41,7 @@ class TransformationJob(operationDef: OperationDef,
   override val scheduleStrategy: ScheduleStrategy = new ScheduleStrategySourcing
 
   override def preRunCheckJob(infoDate: LocalDate, runReason: TaskRunReason, jobConfig: Config, dependencyWarnings: Seq[DependencyWarning]): JobPreRunResult = {
-    preRunTransformationCheck(infoDate, dependencyWarnings)
+    preRunTransformationCheck(infoDate, dependencyWarnings, runReason)
   }
 
   override def validate(infoDate: LocalDate, jobConfig: Config): Reason = {