Repository: AbsaOSS/pramen (language: python)

Issue:
When running for date range make all dependencies passive
## Background
When running a job for a date range, an absence of data makes dependent transformations fail, even for dates that are not affected. This can be quickly fixed by automatically making all dependencies passive when running a pipeline for a date range.

## Feature
When running for date range make all dependencies passive.

## Example
--

## Proposed Solution 
--



Discussion / hints:

Fix patch:
diff --git a/pramen/core/src/main/scala/za/co/absa/pramen/core/runner/AppRunner.scala b/pramen/core/src/main/scala/za/co/absa/pramen/core/runner/AppRunner.scala
index 0dcc38270..e65faafc2 100644
--- a/pramen/core/src/main/scala/za/co/absa/pramen/core/runner/AppRunner.scala
+++ b/pramen/core/src/main/scala/za/co/absa/pramen/core/runner/AppRunner.scala
@@ -25,7 +25,7 @@ import za.co.absa.pramen.core.app.{AppContext, AppContextImpl}
 import za.co.absa.pramen.core.config.Keys.LOG_EXECUTOR_NODES
 import za.co.absa.pramen.core.exceptions.ValidationException
 import za.co.absa.pramen.core.metastore.peristence.{TransientJobManager, TransientTableManager}
-import za.co.absa.pramen.core.pipeline.{Job, OperationSplitter, PipelineDef}
+import za.co.absa.pramen.core.pipeline.{Job, OperationDef, OperationSplitter, PipelineDef}
 import za.co.absa.pramen.core.runner.jobrunner.{ConcurrentJobRunner, ConcurrentJobRunnerImpl}
 import za.co.absa.pramen.core.runner.orchestrator.OrchestratorImpl
 import za.co.absa.pramen.core.runner.task.{TaskRunner, TaskRunnerMultithreaded}
@@ -182,12 +182,29 @@ object AppRunner {
                               appContext: AppContext,
                               spark: SparkSession): Try[Seq[Job]] = {
     handleFailure(Try {
+      val isHistoricalRun = appContext.appConfig.runtimeConfig.runDateTo.nonEmpty
       val splitter = new OperationSplitter(conf, appContext.metastore, appContext.bookkeeper)
 
-      pipelineDef.operations.flatMap(op => splitter.createJobs(op))
+      if (isHistoricalRun)
+        log.info("This is a historical run. Making all dependencies 'passive' for all jobs...")
+
+      pipelineDef.operations.flatMap { opOrig =>
+        val op = if (isHistoricalRun) {
+          preProcessOperationForHistoricalRun(opOrig)
+        } else {
+          opOrig
+        }
+
+        splitter.createJobs(op)
+      }
     }, state, "splitting of the pipeline into jobs")
   }
 
+  private[core] def preProcessOperationForHistoricalRun(ops: OperationDef): OperationDef = {
+    val newDep = ops.dependencies.map(dep => dep.copy(isPassive = true))
+    ops.copy(dependencies = newDep)
+  }
+
   private[core] def filterJobs(state: PipelineState,
                                jobs: Seq[Job],
                                runtimeConfig: RuntimeConfig): Try[Seq[Job]] = {


Test patch:
diff --git a/pramen/core/src/test/scala/za/co/absa/pramen/core/tests/runner/AppRunnerSuite.scala b/pramen/core/src/test/scala/za/co/absa/pramen/core/tests/runner/AppRunnerSuite.scala
index 3937d4cec..dceb5f6e7 100644
--- a/pramen/core/src/test/scala/za/co/absa/pramen/core/tests/runner/AppRunnerSuite.scala
+++ b/pramen/core/src/test/scala/za/co/absa/pramen/core/tests/runner/AppRunnerSuite.scala
@@ -23,6 +23,7 @@ import za.co.absa.pramen.core._
 import za.co.absa.pramen.core.app.AppContext
 import za.co.absa.pramen.core.app.config.HookConfig
 import za.co.absa.pramen.core.base.SparkTestBase
+import za.co.absa.pramen.core.metastore.model.MetastoreDependencyFactory
 import za.co.absa.pramen.core.mocks.RunnableSpy
 import za.co.absa.pramen.core.mocks.job.JobSpy
 import za.co.absa.pramen.core.mocks.state.PipelineStateSpy
@@ -55,6 +56,25 @@ class AppRunnerSuite extends AnyWordSpec with SparkTestBase {
     }
   }
 
+  "preProcessOperationForHistoricalRun" should {
+    "make all dependencies passive" in {
+      val conf = ConfigFactory.parseString(
+        """
+          |tables = [ "table1", "table2", "table3",]
+          |date.from = "2020-01-01"
+          |""".stripMargin
+      )
+      val dep1 = MetastoreDependencyFactory.fromConfig(conf, "")
+      val dep2 = dep1.copy(isOptional = true)
+      val dep3 = dep1.copy(isPassive = true)
+      val op = OperationDefFactory.getDummyOperationDef(dependencies = Seq(dep1, dep2, dep3))
+
+      val newOp = AppRunner.preProcessOperationForHistoricalRun(op)
+
+      assert(newOp.dependencies.forall(_.isPassive))
+    }
+  }
+
   "filterJobs()" should {
     val conf: Config = getTestConfig()