#374 Fix the way retrospective updates are determined.

AbsaOSS · Sep 30, 2024 · 6f13121 · 6f13121
1 parent 7da2627
commit 6f13121
Show file tree

Hide file tree

Showing 3 changed files with 15 additions and 8 deletions.
diff --git a/pramen/core/src/main/scala/za/co/absa/pramen/core/metastore/MetastoreImpl.scala b/pramen/core/src/main/scala/za/co/absa/pramen/core/metastore/MetastoreImpl.scala
@@ -120,7 +120,9 @@ class MetastoreImpl(appConfig: Config,
 
     val finish = Instant.now.getEpochSecond
 
-    if (!skipBookKeepingUpdates) {
+    val nothingAppended = stats.recordCountAppended.contains(0)
+
+    if (!skipBookKeepingUpdates && !nothingAppended) {
       bookkeeper.setRecordCount(tableName, infoDate, infoDate, infoDate, inputRecordCount.getOrElse(stats.recordCount), stats.recordCount, start, finish, isTransient)
     }
 

diff --git a/...c/main/scala/za/co/absa/pramen/core/notify/pipeline/PipelineNotificationBuilderHtml.scala b/...c/main/scala/za/co/absa/pramen/core/notify/pipeline/PipelineNotificationBuilderHtml.scala
@@ -680,8 +680,12 @@ class PipelineNotificationBuilderHtml(implicit conf: Config) extends PipelineNot
     } else {
       if (status.warnings.nonEmpty)
         TextElement("Warning", style)
-      else
-        TextElement("Success", style)
+      else {
+        if (status.recordsAppended.contains(0))
+          TextElement("No new data", style)
+        else
+          TextElement("Success", style)
+      }
     }
   }
 

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
@@ -26,7 +26,7 @@ import za.co.absa.pramen.core.metastore.Metastore
 import za.co.absa.pramen.core.metastore.model.MetaTable
 import za.co.absa.pramen.core.schedule.Schedule
 import za.co.absa.pramen.core.utils.Emoji._
-import za.co.absa.pramen.core.utils.TimeUtils
+import za.co.absa.pramen.core.utils.{Emoji, TimeUtils}
 
 import java.time.{Instant, LocalDate}
 import scala.util.{Failure, Success, Try}
@@ -126,7 +126,8 @@ abstract class JobBase(operationDef: OperationDef,
         val outOfDateTables = getOutdatedTables(infoDate, chunk.jobFinished)
         if (outOfDateTables.nonEmpty) {
           log.info(s"Job for table ${outputTableDef.name} as already ran for $infoDate, but has outdated tables: ${outOfDateTables.mkString(", ")}")
-          Some(JobPreRunResult(JobPreRunStatus.NeedsUpdate, None, dependencyWarnings, Seq.empty[String]))
+          val warning = s"Based on outdated tables: ${outOfDateTables.mkString(", ")}"
+          Some(JobPreRunResult(JobPreRunStatus.NeedsUpdate, None, dependencyWarnings, Seq(warning)))
         } else {
           log.info(s"Job for table ${outputTableDef.name} as already ran for $infoDate.")
           Some(JobPreRunResult(JobPreRunStatus.AlreadyRan, None, dependencyWarnings, Seq.empty[String]))
@@ -143,12 +144,12 @@ abstract class JobBase(operationDef: OperationDef,
       .flatMap(_.tables)
       .distinct
       .filter { table =>
-        bookkeeper.getLatestDataChunk(outputTableDef.name, infoDate, infoDate) match {
+        bookkeeper.getLatestDataChunk(table, infoDate, infoDate) match {
           case Some(chunk) if chunk.jobFinished >= targetJobFinishedSeconds =>
-            log.warn(s"The dependent table '$table' has been updated at ${Instant.ofEpochSecond(chunk.jobFinished)} retrospectively " +
+            log.warn(s"${Emoji.WARNING} The dependent table '$table' has been updated at ${Instant.ofEpochSecond(chunk.jobFinished)} retrospectively " +
               s"after the transformation at ${Instant.ofEpochSecond(targetJobFinishedSeconds)} .")
             true
-          case Some(chunk) =>
+          case Some(_) =>
             false
         }
       }