#245 OnDemand -> Lazy everywhere since 'lazy' is a known concept. We could use either everywhere and chose 'lazy'.

Author: yruslan

README.md

@@ -2294,7 +2294,8 @@ The cache policy can be:
 - `persist` - the table is persisted in the temporary directory for the duration of the pipeline run.
 
 Transient tables are calculated on-demand by executing the operation that outputs to that table. This occurs when a 
-transformer or a sink invokes metastore.getTable() or metastore.getLatest(). Pramen ensures that if the same data is 
+transformer or a sink invokes metastore.getTable() or metastore.getLatest(). This is also known as 'lazy evaluation'
+therefore jobs that output to transient tables are also known as 'lazy jobs'. Pramen ensures that if the same data is 
 required by multiple transformers (for the same job for the same date), the job will run only once.
 
 ### File-based sourcing

pramen/core/src/main/scala/za/co/absa/pramen/core/metastore/MetastoreImpl.scala

@@ -49,9 +49,9 @@ class MetastoreImpl(appConfig: Config,
 
   override def isDataAvailable(tableName: String, infoDateFromOpt: Option[LocalDate], infoDateToOpt: Option[LocalDate]): Boolean = {
     val mt = getTableDef(tableName)
-    val isOnDemand = mt.format.isLazy
+    val isLazy = mt.format.isLazy
 
-    if (isOnDemand) {
+    if (isLazy) {
       (infoDateFromOpt, infoDateToOpt) match {
         case (Some(infoDateFrom), Some(infoDateTo)) =>
           TransientJobManager.selectInfoDatesToExecute(tableName, infoDateFrom, infoDateTo).nonEmpty
@@ -76,8 +76,8 @@ class MetastoreImpl(appConfig: Config,
 
   override def getLatest(tableName: String, until: Option[LocalDate]): DataFrame = {
     val mt = getTableDef(tableName)
-    val isOnDemand = mt.format.isLazy
-    if (isOnDemand) {
+    val isLazy = mt.format.isLazy
+    if (isLazy) {
       MetastorePersistence.fromMetaTable(mt, appConfig).loadTable(None, until)
     } else {
       bookkeeper.getLatestProcessedDate(tableName, until) match {

pramen/core/src/main/scala/za/co/absa/pramen/core/metastore/peristence/MetastorePersistenceTransient.scala

@@ -35,13 +35,13 @@ class MetastorePersistenceTransient(tempPath: Option[String],
   override def loadTable(infoDateFrom: Option[LocalDate], infoDateTo: Option[LocalDate]): DataFrame = {
     (infoDateFrom, infoDateTo) match {
       case (Some(from), Some(to)) if from == to =>
-        runOnDemandTask(tableName, from)
+        runLazyTask(tableName, from)
       case (Some(from), Some(to)) =>
-        runOnDemandTasks(tableName, selectInfoDatesToExecute(tableName, from, to))
+        runLazyTasks(tableName, selectInfoDatesToExecute(tableName, from, to))
       case (None, Some(until)) =>
-        runOnDemandTask(tableName, selectLatestOnDemandSnapshot(tableName, until))
+        runLazyTask(tableName, selectLatestLazySnapshot(tableName, until))
       case _ =>
-        throw new IllegalArgumentException("Metastore 'on_demand' format requires info date for querying its contents.")
+        throw new IllegalArgumentException("Metastore 'transient' format requires info date for querying its contents.")
     }
   }
 
@@ -50,14 +50,14 @@ class MetastorePersistenceTransient(tempPath: Option[String],
   }
 
   override def getStats(infoDate: LocalDate): MetaTableStats = {
-    throw new UnsupportedOperationException("On demand format does not support getting record count and size statistics.")
+    throw new UnsupportedOperationException("The 'transient' format does not support getting record count and size statistics.")
   }
 
   override def createOrUpdateHiveTable(infoDate: LocalDate, hiveTableName: String, queryExecutor: QueryExecutor, hiveConfig: HiveConfig): Unit = {
-    throw new UnsupportedOperationException("On demand format does not support Hive tables.")
+    throw new UnsupportedOperationException("The 'transient' format does not support Hive tables.")
   }
 
   override def repairHiveTable(hiveTableName: String, queryExecutor: QueryExecutor, hiveConfig: HiveConfig): Unit = {
-    throw new UnsupportedOperationException("On demand format does not support Hive tables.")
+    throw new UnsupportedOperationException("The 'transient' format does not support Hive tables.")
   }
 }

pramen/core/src/main/scala/za/co/absa/pramen/core/metastore/peristence/TransientJobManager.scala

@@ -34,7 +34,7 @@ object TransientJobManager {
   val MAXIMUM_UNIONS = 50
 
   private val log = LoggerFactory.getLogger(this.getClass)
-  private val onDemandJobs = new mutable.HashMap[String, Job]()
+  private val lazyJobs = new mutable.HashMap[String, Job]()
   private val runningJobs = new mutable.HashMap[MetastorePartition, Future[DataFrame]]()
   private var taskRunnerOpt: Option[TaskRunner] = None
 
@@ -46,8 +46,8 @@ object TransientJobManager {
     taskRunnerOpt.isDefined
   }
 
-  private[core] def addOnDemandJob(job: Job): Unit = synchronized {
-    onDemandJobs += job.outputTable.name.toLowerCase -> job
+  private[core] def addLazyJob(job: Job): Unit = synchronized {
+    lazyJobs += job.outputTable.name.toLowerCase -> job
   }
 
   private[core] def selectInfoDatesToExecute(outputTableName: String,
@@ -62,8 +62,8 @@ object TransientJobManager {
       job.operation.schedule)
   }
 
-  private[core] def selectLatestOnDemandSnapshot(outputTableName: String,
-                                                 infoDateUntil: LocalDate): LocalDate = {
+  private[core] def selectLatestLazySnapshot(outputTableName: String,
+                                             infoDateUntil: LocalDate): LocalDate = {
     val job = getJob(outputTableName)
 
     ScheduleStrategyUtils.getLatestActiveInfoDate(outputTableName,
@@ -72,10 +72,10 @@ object TransientJobManager {
       job.operation.schedule)
   }
 
-  private[core] def runOnDemandTasks(outputTableName: String,
-                                     infoDates: Seq[LocalDate])
-                                    (implicit spark: SparkSession): DataFrame = {
-    val dfs = infoDates.map(infoDate => runOnDemandTask(outputTableName, infoDate))
+  private[core] def runLazyTasks(outputTableName: String,
+                                 infoDates: Seq[LocalDate])
+                                (implicit spark: SparkSession): DataFrame = {
+    val dfs = infoDates.map(infoDate => runLazyTask(outputTableName, infoDate))
 
     if (dfs.isEmpty) {
       TransientTableManager.getEmptyDfForTable(outputTableName).getOrElse(spark.emptyDataFrame)
@@ -85,13 +85,13 @@ object TransientJobManager {
           s"since the number of dataframe unions is too big (${infoDates.length} > $WARN_UNIONS)")
       }
       if (infoDates.length > MAXIMUM_UNIONS) {
-        throw new IllegalArgumentException(s"The number of subtasks requested for the on-demand job contains too many " +
+        throw new IllegalArgumentException(s"The number of subtasks requested for the lazy job contains too many " +
           s"dataframe unions (${infoDates.length} > $MAXIMUM_UNIONS)")
       }
 
-      infoDates.tail.foldLeft(runOnDemandTask(outputTableName, infoDates.head))(
+      infoDates.tail.foldLeft(runLazyTask(outputTableName, infoDates.head))(
         (acc, infoDate) => {
-          val df = runOnDemandTask(outputTableName, infoDate)
+          val df = runLazyTask(outputTableName, infoDate)
           safeUnion(acc, df)
         }
       )
@@ -113,11 +113,11 @@ object TransientJobManager {
     }
   }
 
-  private[core] def runOnDemandTask(outputTableName: String,
-                                    infoDate: LocalDate)
-                                   (implicit spark: SparkSession): DataFrame = {
+  private[core] def runLazyTask(outputTableName: String,
+                                infoDate: LocalDate)
+                               (implicit spark: SparkSession): DataFrame = {
     val start = Instant.now()
-    val fut = getOnDemandTaskFuture(outputTableName, infoDate)
+    val fut = getLazyTaskFuture(outputTableName, infoDate)
 
     if (!fut.isCompleted) {
       log.info(s"Waiting for the dependent task to finish ($outputTableName for $infoDate)...")
@@ -130,9 +130,9 @@ object TransientJobManager {
     df
   }
 
-  private[core] def getOnDemandTaskFuture(outputTableName: String,
-                                          infoDate: LocalDate)
-                                         (implicit spark: SparkSession): Future[DataFrame] = {
+  private[core] def getLazyTaskFuture(outputTableName: String,
+                                      infoDate: LocalDate)
+                                     (implicit spark: SparkSession): Future[DataFrame] = {
 
     val promise = Promise[DataFrame]()
 
@@ -186,7 +186,7 @@ object TransientJobManager {
       log.info(s"The task ($outputTableName for $infoDate) is already running. Waiting for results...")
       Some(runningJobs(metastorePartition))
     } else {
-      log.info(s"Running the on-demand task ($outputTableName for $infoDate)...")
+      log.info(s"Running (materializing) the lazy task ($outputTableName for $infoDate)...")
       runningJobs += metastorePartition -> newJonFuture
       None
     }
@@ -201,27 +201,27 @@ object TransientJobManager {
   }
 
   private[core] def getJob(outputTableName: String): Job = {
-    val jobOpt = onDemandJobs.get(outputTableName.toLowerCase)
+    val jobOpt = lazyJobs.get(outputTableName.toLowerCase)
 
     jobOpt match {
       case Some(job) => job
-      case None => throw new IllegalArgumentException(s"On-demand job with output table name '$outputTableName' not found or haven't registered yet.")
+      case None => throw new IllegalArgumentException(s"Lazy job with output table name '$outputTableName' not found or haven't registered yet.")
     }
   }
 
   private[core] def reset(): Unit = synchronized {
-    onDemandJobs.clear()
+    lazyJobs.clear()
     runningJobs.clear()
     taskRunnerOpt = None
   }
 
   private[core] def runJob(job: Job,
                            infoDate: LocalDate)
                           (implicit spark: SparkSession): DataFrame = {
-    val jobStr = s"On-demand job outputting to '${job.outputTable.name}' for '$infoDate'"
+    val jobStr = s"Lazy job outputting to '${job.outputTable.name}' for '$infoDate'"
     taskRunnerOpt match {
       case Some(taskRunner) =>
-        taskRunner.runOnDemand(job, infoDate) match {
+        taskRunner.runLazyTask(job, infoDate) match {
           case _: RunStatus.Succeeded         => TransientTableManager.getDataForTheDate(job.outputTable.name, infoDate)
           case _: RunStatus.Skipped           => spark.emptyDataFrame
           case RunStatus.ValidationFailed(ex) => throw new IllegalStateException(s"$jobStr validation failed.", ex)

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

@@ -39,7 +39,7 @@ object TaskRunReason {
     override def toString: String = "Skip"
   }
 
-  case object OnDemand extends TaskRunReason {
-    override def toString: String = "On Demand"
+  case object OnRequest extends TaskRunReason {
+    override def toString: String = "On Request"
   }
 }

pramen/core/src/main/scala/za/co/absa/pramen/core/runner/jobrunner/ConcurrentJobRunnerImpl.scala

@@ -146,7 +146,7 @@ class ConcurrentJobRunnerImpl(runtimeConfig: RuntimeConfig,
   }
 
   private[core] def runLazyJob(job: Job): Boolean = {
-    TransientJobManager.addOnDemandJob(job)
+    TransientJobManager.addLazyJob(job)
     true
   }
 

pramen/core/src/main/scala/za/co/absa/pramen/core/runner/orchestrator/OrchestratorImpl.scala

@@ -20,7 +20,6 @@ import com.github.yruslan.channel.Channel
 import com.typesafe.config.Config
 import org.apache.spark.sql.SparkSession
 import org.slf4j.LoggerFactory
-import za.co.absa.pramen.api.DataFormat
 import za.co.absa.pramen.core.app.AppContext
 import za.co.absa.pramen.core.exceptions.{FatalErrorWrapper, ValidationException}
 import za.co.absa.pramen.core.pipeline.{Job, JobDependency, OperationType}
@@ -92,10 +91,10 @@ class OrchestratorImpl extends Orchestrator {
           log.info(s"There is another job outputting to ${finishedJob.outputTable.name}. Waiting for it to finish before marking the table as finished.")
         }
 
-        val isOnDemand = finishedJob.outputTable.format.isInstanceOf[DataFormat.Transient]
+        val isLazy = finishedJob.outputTable.format.isLazy
 
         if (!hasAnotherUnfinishedJob || !isSucceeded) {
-          updateDependencyResolver(dependencyResolver, finishedJob, isSucceeded, isOnDemand)
+          updateDependencyResolver(dependencyResolver, finishedJob, isSucceeded, isLazy)
         }
 
         state.addTaskCompletion(taskResults)
@@ -188,12 +187,12 @@ class OrchestratorImpl extends Orchestrator {
   private def updateDependencyResolver(dependencyResolver: DependencyResolver,
                                        job: Job,
                                        isSucceeded: Boolean,
-                                       isOnDemand: Boolean): Unit = {
+                                       isLazy: Boolean): Unit = {
     val outputTable = job.outputTable
 
     if (isSucceeded) {
-      if (isOnDemand) {
-        log.info(s"On-demand job '${job.name}' outputting to '${outputTable.name}' has been registered for the future use.")
+      if (isLazy) {
+        log.info(s"Lazy job '${job.name}' outputting to '${outputTable.name}' has been registered for the future use.")
       } else {
         log.info(s"$SUCCESS Job '${job.name}' outputting to '${outputTable.name}' has SUCCEEDED.")
       }

pramen/core/src/main/scala/za/co/absa/pramen/core/runner/task/TaskRunner.scala

@@ -25,6 +25,6 @@ trait TaskRunner extends AutoCloseable {
   /** Run a job for specified information dates as a part of pipeline execution. */
   def runJobTasks(job: Job, infoDates: Seq[TaskPreDef]): Future[Seq[RunStatus]]
 
-  /** Run a job for specified information date on-demand from another job. */
-  def runOnDemand(job: Job, infoDate: LocalDate): RunStatus
+  /** Run a job for specified information date when requested from another job. */
+  def runLazyTask(job: Job, infoDate: LocalDate): RunStatus
 }

pramen/core/src/main/scala/za/co/absa/pramen/core/runner/task/TaskRunnerBase.scala

@@ -83,15 +83,15 @@ abstract class TaskRunnerBase(conf: Config,
     }
   }
 
-  override def runOnDemand(job: Job, infoDate: LocalDate): RunStatus = {
+  override def runLazyTask(job: Job, infoDate: LocalDate): RunStatus = {
     val started = Instant.now()
-    val task = Task(job, infoDate, TaskRunReason.OnDemand)
+    val task = Task(job, infoDate, TaskRunReason.OnRequest)
     val result: TaskResult = validate(task, started) match {
       case Left(failedResult) => failedResult
       case Right(validationResult) => run(task, started, validationResult)
     }
 
-    onTaskCompletion(task, result, isOnDemand = true)
+    onTaskCompletion(task, result, isLazy = true)
   }
 
   /** Runs multiple tasks in the single thread in the order of info dates. If one task fails, the rest will be skipped. */
@@ -125,7 +125,7 @@ abstract class TaskRunnerBase(conf: Config,
           case Left(failedResult) => failedResult
           case Right(validationResult) => run(task, started, validationResult)
         }
-        onTaskCompletion(task, result, isOnDemand = false)
+        onTaskCompletion(task, result, isLazy = false)
     }
   }
 
@@ -137,7 +137,7 @@ abstract class TaskRunnerBase(conf: Config,
     val isTransient = task.job.outputTable.format.isTransient
     val taskResult = TaskResult(task.job, runStatus, Some(runInfo), applicationId, isTransient, Nil, Nil, Nil)
 
-    onTaskCompletion(task, taskResult, isOnDemand = false)
+    onTaskCompletion(task, taskResult, isLazy = false)
   }
 
   /**
@@ -362,11 +362,11 @@ abstract class TaskRunnerBase(conf: Config,
   }
 
   /** Logs task completion and sends corresponding notifications. */
-  private def onTaskCompletion(task: Task, taskResult: TaskResult, isOnDemand: Boolean): RunStatus = {
+  private def onTaskCompletion(task: Task, taskResult: TaskResult, isLazy: Boolean): RunStatus = {
     val notificationTargetErrors = sendNotifications(task, taskResult)
     val updatedResult = taskResult.copy(notificationTargetErrors = notificationTargetErrors)
 
-    logTaskResult(updatedResult, isOnDemand)
+    logTaskResult(updatedResult, isLazy)
     pipelineState.addTaskCompletion(Seq(updatedResult))
     addJournalEntry(task, updatedResult)
 
@@ -447,14 +447,14 @@ abstract class TaskRunnerBase(conf: Config,
     Some(RunInfo(infoDate, started, Instant.now()))
   }
 
-  private def logTaskResult(result: TaskResult, isOnDemand: Boolean): Unit = synchronized {
+  private def logTaskResult(result: TaskResult, isLazy: Boolean): Unit = synchronized {
     val infoDateMsg = result.runInfo match {
       case Some(date) => s" for ${date.infoDate}"
       case None => ""
     }
 
-    val taskStr = if (isOnDemand) {
-      "On demand task"
+    val taskStr = if (isLazy) {
+      "Requested lazy task"
     } else {
       "Task"
     }

pramen/core/src/test/resources/test/config/integration_transient_transformer.conf

@@ -64,7 +64,7 @@ pramen.operations = [
     info.date.expr = "@runDate - 1"
   },
   {
-    name = "Identity transformer (on-demand)"
+    name = "Identity transformer (lazy)"
     type = "transformation"
 
     class = "za.co.absa.pramen.core.transformers.IdentityTransformer"