AbsaOSS · yruslan · Sep 4, 2026 · Sep 1, 2026 · Sep 2, 2026 · Sep 3, 2026
@@ -150,8 +150,10 @@ pramen {
 
   # When runtime.run.mode = bulk, specifies the size of bulk load
   #runtime.run.bulk.batch.size = monthly
+  # You can enable repartitioning for bulk jobs
+  #runtime.enable.repartitioning = true
 
-  #runtime.run.bulk.curreent {
+  #runtime.run.bulk.current {
   #  date.from = ""
   #  date.to = ""
   #  output.information.date = ""

@@ -19,7 +19,8 @@ package za.co.absa.pramen.core.app.config
 import java.time.LocalDate
 
 case class BulkRunConfig(
-                          dateFrom: LocalDate,
-                          dateTo: LocalDate,
+                          dataDateFrom: LocalDate,
+                          dataDateTo: LocalDate,
+                          infoDateColumn: Option[String],
                           outputInfoDate: LocalDate
                         )
@@ -49,6 +49,7 @@ case class RuntimeConfig(
                           historicalRunMode: RunMode,
                           bulkBatchSize: BulkBatchSize,
                           bulkLoadCurrent: Option[BulkRunConfig],
+                          enableRepartitioning: Boolean,
                           sparkAppDescriptionTemplate: Option[String],
                           attempt: Int, // Current attempt number for the pipeline run (for auto-retry automation)
                           maxAttempts: Int, // Maximum number of attempts allowed for the pipeline run
@@ -65,10 +66,12 @@ object RuntimeConfig {
   val IS_RERUN = "pramen.runtime.is.rerun"
   val IS_INVERSE_ORDER = "pramen.runtime.inverse.order"
   val RUN_MODE = "pramen.runtime.run.mode"
-  val RUN_BULK_BATCH_SIZE = "pramen.runtime.run.bulk.batch.size"
-  val BULK_CURRENT_DATE_FROM = "pramen.runtime.run.bulk.curreent.date.from"
-  val BULK_CURRENT_DATE_TO = "pramen.runtime.run.bulk.curreent.date.to"
-  val BULK_CURRENT_OUTPUT_INFO_DATE = "pramen.runtime.run.bulk.curreent.output.information.date"
+  val RUN_BULK_BATCH_SIZE = "pramen.runtime.bulk.batch.size"
+  val RUN_ENABLE_REPARTITIONING = "pramen.runtime.enable.repartitioning"
+  val INFO_DATE_COLUMN = "pramen.runtime.info.date.column"
+  val BULK_CURRENT_DATE_FROM = "pramen.runtime.run.bulk.current.date.from"
+  val BULK_CURRENT_DATE_TO = "pramen.runtime.run.bulk.current.date.to"
+  val BULK_CURRENT_OUTPUT_INFO_DATE = "pramen.runtime.run.bulk.current.output.information.date"
   val RUN_TABLES = "pramen.runtime.run.tables"
   val UNDERCOVER = "pramen.undercover"
   val USE_LOCK = "pramen.use.lock"
@@ -161,9 +164,11 @@ object RuntimeConfig {
     val bulkCurrentDateFrom = ConfigUtils.getOptionString(conf, BULK_CURRENT_DATE_FROM).map(getDate)
     val bulkCurrentDateTo = ConfigUtils.getOptionString(conf, BULK_CURRENT_DATE_TO).map(getDate)
     val bulkCurrentOutputInfoDate = ConfigUtils.getOptionString(conf, BULK_CURRENT_OUTPUT_INFO_DATE).map(getDate)
+    val infoDateColumn = ConfigUtils.getOptionString(conf, INFO_DATE_COLUMN)
+    val enableRepartitioning = ConfigUtils.getOptionBoolean(conf, RUN_ENABLE_REPARTITIONING).getOrElse(false)
 
     val bulkLoadCurrent = if (bulkCurrentDateFrom.isDefined && bulkCurrentDateTo.isDefined && bulkCurrentOutputInfoDate.isDefined) {
-      Some(BulkRunConfig(bulkCurrentDateFrom.get, bulkCurrentDateTo.get, bulkCurrentOutputInfoDate.get))
+      Some(BulkRunConfig(bulkCurrentDateFrom.get, bulkCurrentDateTo.get, infoDateColumn, bulkCurrentOutputInfoDate.get))
     } else {
       None
     }
@@ -192,6 +197,7 @@ object RuntimeConfig {
       attempt = attempt,
       maxAttempts = maxAttempts,
       forceReCreateHiveTables =  ConfigUtils.getOptionBoolean(conf, FORCE_RECREATE_HIVE_TABLES).getOrElse(false),
+      enableRepartitioning = enableRepartitioning,
       executionOptions = executionOptions
     )
   }
@@ -217,6 +223,7 @@ object RuntimeConfig {
       historicalRunMode = RunMode.CheckUpdates,
       bulkBatchSize = Monthly,
       bulkLoadCurrent = None,
+      enableRepartitioning = false,
       sparkAppDescriptionTemplate = None,
       attempt = 1,
       maxAttempts = 1,

@@ -44,6 +44,7 @@ case class CmdLineConfig(
                           dateTo: Option[LocalDate] = None,
                           mode: Option[String] = None,
                           bulkSize: Option[String] = None,
+                          infoDateColumn: Option[String] = None,
                           inverseOrder: Option[Boolean] = None,
                           verbose: Option[Boolean] = None,
                           overrideLogLevel: Option[String] = None,
@@ -133,6 +134,9 @@ object CmdLineConfig {
     for (bulkSize <- cmd.bulkSize)
       accumulatedConfig = accumulatedConfig.withValue(RUN_BULK_BATCH_SIZE, ConfigValueFactory.fromAnyRef(bulkSize))
 
+    for (infoDateColumn <- cmd.infoDateColumn)
+      accumulatedConfig = accumulatedConfig.withValue(INFO_DATE_COLUMN, ConfigValueFactory.fromAnyRef(infoDateColumn))
+
     for (logEffectiveConfig <- cmd.logEffectiveConfig)
       accumulatedConfig = accumulatedConfig.withValue(LOG_EFFECTIVE_CONFIG, ConfigValueFactory.fromAnyRef(logEffectiveConfig))
 
@@ -213,7 +217,13 @@ object CmdLineConfig {
           .text("The bulk size for processing date ranges.")
           .validate(v =>
             if (v == "monthly" || v == "quarterly" || v == "yearly") success
-            else failure("Invalid bulk size. Must be one of 'monthly', 'quarterly', 'yearly'"))
+            else failure("Invalid bulk size. Must be one of 'monthly', 'quarterly', 'yearly'")),
+        opt[String]("info-date-column").optional().action((value, config) =>
+            config.copy(infoDateColumn = Option(value)))
+          .text("The information date column name to use for repartitioning.")
+          .validate(v =>
+            if (v.nonEmpty) success
+            else failure("Invalid information date column name. Must be a non-empty string."))
       )
 
     opt[Boolean]("inverse-order").optional().action((value, config) =>

@@ -40,6 +40,12 @@ trait MetastorePersistence {
   def repairHiveTable(hiveTableName: String,
                       queryExecutor: QueryExecutor,
                       hiveConfig: HiveConfig): Unit
+
+  def isRepartitioningSupported: Boolean
+
+  def repartitionPhase1(infoDateColumn: String, infoDateFrom: LocalDate, infoDateTo: LocalDate, outputInfoDate: LocalDate): Unit = {}
+
+  def repartitionPhase2(infoDateColumn: String, infoDateFrom: LocalDate, infoDateTo: LocalDate, outputInfoDate: LocalDate): Unit = {}
 }
 
 object MetastorePersistence {

@@ -215,6 +215,8 @@ class MetastorePersistenceDelta(query: Query,
     throw new UnsupportedOperationException("Delta format does not support Hive tables at the moment.")
   }
 
+  override def isRepartitioningSupported: Boolean = false
+
   def getFilter(infoDateFrom: Option[LocalDate], infoDateTo: Option[LocalDate]): Column = {
     if (partitionScheme == PartitionScheme.Overwrite) {
       if (infoDateFrom.isDefined || infoDateTo.isDefined) {

@@ -18,6 +18,7 @@ package za.co.absa.pramen.core.metastore.peristence
 
 import org.apache.spark.sql._
 import org.apache.spark.sql.functions._
+import org.apache.spark.sql.types.DateType
 import org.slf4j.LoggerFactory
 import za.co.absa.pramen.api.{CatalogTable, PartitionScheme}
 import za.co.absa.pramen.core.metastore.MetaTableStats
@@ -112,6 +113,29 @@ class MetastorePersistenceIceberg(table: CatalogTable,
     throw new UnsupportedOperationException("Iceberg only operates on tables in a catalog. Separate Hive options are not supported.")
   }
 
+  override def isRepartitioningSupported: Boolean = true
+
+  override def repartitionPhase1(infoDateDataColumn: String, infoDateFrom: LocalDate, infoDateTo: LocalDate, outputInfoDate: LocalDate): Unit = {
+     if (infoDateColumn.equalsIgnoreCase(infoDateDataColumn))
+       throw new IllegalArgumentException(s"Cannot repartition a table if the metastore info date column is the same as the data info date column ($infoDateDataColumn)")
+
+    val fullTableName = table.getFullTableName
+    val df = spark.table(fullTableName)
+      .filter(getFilter(Some(outputInfoDate), Some(outputInfoDate)))
+
+    log.info(s"Running Iceberg repartitioning: UPDATE $fullTableName SET $infoDateColumn = CAST($infoDateDataColumn AS DATE) " +
+      s"WHERE $infoDateColumn = '$outputInfoDate' AND $infoDateDataColumn >= '$infoDateFrom' AND $infoDateDataColumn <= '$infoDateTo'")
+
+    val dfToWrite = df.withColumn(infoDateColumn, col(infoDateDataColumn).cast(DateType))
+
+    writeRepartitionedDf(dfToWrite, fullTableName, infoDateColumn, infoDateFrom, infoDateTo, writeOptions)
+  }
+
+  override def repartitionPhase2(infoDateDataColumn: String, infoDateFrom: LocalDate, infoDateTo: LocalDate, outputInfoDate: LocalDate): Unit = {
+    if (infoDateColumn.equalsIgnoreCase(infoDateDataColumn))
+      throw new IllegalArgumentException(s"Cannot repartition a table if the metastore info date column is the same as the data info date column ($infoDateDataColumn)")
+  }
+
   def getFilter(infoDateFrom: Option[LocalDate], infoDateTo: Option[LocalDate]): Column = {
     if (partitionScheme == PartitionScheme.Overwrite) {
       if (infoDateFrom.isDefined || infoDateTo.isDefined) {

@@ -48,4 +48,6 @@ class MetastorePersistenceNull(implicit spark: SparkSession) extends MetastorePe
                                hiveConfig: HiveConfig): Unit = {
     throw new UnsupportedOperationException("Parquet format does not support Hive tables at the moment.")
   }
+
+  override def isRepartitioningSupported: Boolean = false
 }
@@ -145,6 +145,8 @@ class MetastorePersistenceParquet(path: String,
     throw new UnsupportedOperationException("Parquet format does not support Hive tables at the moment.")
   }
 
+  override def isRepartitioningSupported: Boolean = false
+
   def loadPartitionDirectly(infoDate: LocalDate): DataFrame = {
     val dateStr = dateFormatter.format(infoDate)
     val partitionPath = SparkUtils.getPartitionPath(infoDate, infoDateColumn, infoDateFormat, path)

@@ -172,6 +172,8 @@ class MetastorePersistenceRaw(path: String,
     throw new UnsupportedOperationException("Raw format does not support Hive tables.")
   }
 
+  override def isRepartitioningSupported: Boolean = false
+
   private def getListOfFilesRange(infoDateFrom: LocalDate, infoDateTo: LocalDate): Seq[FileStatus] = {
     if (infoDateFrom.isAfter(infoDateTo))
       Seq.empty[FileStatus]

@@ -60,4 +60,6 @@ class MetastorePersistenceTransient(tempPath: Option[String],
   override def repairHiveTable(hiveTableName: String, queryExecutor: QueryExecutor, hiveConfig: HiveConfig): Unit = {
     throw new UnsupportedOperationException("The 'transient' format does not support Hive tables.")
   }
+
+  override def isRepartitioningSupported: Boolean = false
 }
@@ -86,4 +86,6 @@ class MetastorePersistenceTransientEager(tempPathOpt: Option[String],
                                hiveConfig: HiveConfig): Unit = {
     throw new UnsupportedOperationException("Transient format does not support Hive tables.")
   }
+
+  override def isRepartitioningSupported: Boolean = false
 }
@@ -222,7 +222,7 @@ class IncrementalIngestionJob(operationDef: OperationDef,
             Reason.Ready
           case None =>
             log.info(s"Offsets not found for '${outputTable.name}' at '$infoDate'.")
-            Reason.SkipOnce("No offsets registered")
+            Reason.SkipOnce(s"Unable to re-run: No offsets registered for $infoDate")
         }
       case (true, false) =>
         Reason.Ready

@@ -87,7 +87,7 @@ class IngestionJob(operationDef: OperationDef,
     val dataChunk = bookkeeper.getLatestDataChunk(sourceTable.metaTableName, infoDate)
 
     val (from, to) = bulkLoadCurrent match {
-      case Some(bulk) => (bulk.dateFrom, bulk.dateTo)
+      case Some(bulk) => (bulk.dataDateFrom, bulk.dataDateTo)
       case None => getInfoDateRange(infoDate, sourceTable.rangeFromExpr, sourceTable.rangeToExpr)
     }
 
@@ -155,7 +155,7 @@ class IngestionJob(operationDef: OperationDef,
     val dfTransformed = applyTransformations(df, sourceTable.transformations)
 
     val (from, to) = bulkLoadCurrent match {
-      case Some(bulk) => (bulk.dateFrom, bulk.dateTo)
+      case Some(bulk) => (bulk.dataDateFrom, bulk.dataDateTo)
       case None => getInfoDateRange(infoDate, sourceTable.rangeFromExpr, sourceTable.rangeToExpr)
     }
 
@@ -278,7 +278,7 @@ class IngestionJob(operationDef: OperationDef,
 
   private def getSourcingResult(infoDate: LocalDate): SourceResult = {
     val (from, to) = bulkLoadCurrent match {
-      case Some(bulk) => (bulk.dateFrom, bulk.dateTo)
+      case Some(bulk) => (bulk.dataDateFrom, bulk.dataDateTo)
       case None => getInfoDateRange(infoDate, sourceTable.rangeFromExpr, sourceTable.rangeToExpr)
     }
 

@@ -41,6 +41,7 @@ case class OperationDef(
                          ignoreSchemaChange: Boolean,
                          isCritical: Boolean,
                          doNotWriteOutput: Boolean,
+                         enableRepartitioning: Boolean,
                          consumeThreads: Int,
                          dependencies: Seq[MetastoreDependency],
                          outputInfoDateExpression: String,
@@ -69,6 +70,7 @@ object OperationDef {
   val IGNORE_SCHEMA_CHANGE_KEY = "ignore.schema.change"
   val IS_CRITICAL_KEY = "critical"
   val DO_NOT_WRITE_OUTPUT_KEY = "do.not.write.output"
+  val ENABLE_REPARTITIONING_KEY = "enable.repartitioning"
   val CONSUME_THREADS_KEY = "consume.threads"
   val DEPENDENCIES_KEY = "dependencies"
   val STRICT_DEPENDENCY_MANAGEMENT_KEY = "pramen.strict.dependency.management"
@@ -109,6 +111,7 @@ object OperationDef {
     val ignoreSchemaChange = ConfigUtils.getOptionBoolean(conf, IGNORE_SCHEMA_CHANGE_KEY).getOrElse(false)
     val isCritical = ConfigUtils.getOptionBoolean(conf, IS_CRITICAL_KEY).getOrElse(false)
     val doNotWriteOutput = ConfigUtils.getOptionBoolean(conf, DO_NOT_WRITE_OUTPUT_KEY).getOrElse(false)
+    val enableRepartitioning = ConfigUtils.getOptionBoolean(conf, ENABLE_REPARTITIONING_KEY).getOrElse(true)
     val alwaysAttempt = ConfigUtils.getOptionBoolean(conf, ALWAYS_ATTEMPT_KEY).getOrElse(false)
     val dependencies = getDependencies(conf, parent, strictDependencyManagement)
     val outputInfoDateExpressionOpt = ConfigUtils.getOptionString(conf, OUTPUT_INFO_DATE_EXPRESSION_KEY)
@@ -158,6 +161,7 @@ object OperationDef {
       ignoreSchemaChange,
       isCritical,
       doNotWriteOutput,
+      enableRepartitioning,
       consumeThreads,
       dependencies,
       outputInfoDateExpression,

@@ -30,6 +30,7 @@ import za.co.absa.pramen.core.metastore.peristence.{TransientJobManager, Transie
 import za.co.absa.pramen.core.pipeline._
 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.repartitioner.{JobRepartitioner, JobRepartitionerImpl, JobRepartitionerNull}
 import za.co.absa.pramen.core.runner.task.{TaskRunner, TaskRunnerMultithreaded}
 import za.co.absa.pramen.core.state.{PipelineState, PipelineStateImpl, SystemExitCatcherSecurityManager}
 import za.co.absa.pramen.core.utils.Emoji._
@@ -455,6 +456,11 @@ object AppRunner {
         taskRunner,
         spark.sparkContext.applicationId)
 
+      implicit val repartitioner: JobRepartitioner = appContext.appConfig.runtimeConfig.bulkLoadCurrent match {
+        case Some(bulkConfig) => new JobRepartitionerImpl(bulkConfig, appContext.bulkLoadStateManager, appContext.metastore, conf, spark.sparkContext.applicationId, state.getBatchId)(spark)
+        case None => new JobRepartitionerNull
+      }
+
       TransientJobManager.setTaskRunner(taskRunner)
 
       val orchestrator = new OrchestratorImpl()

@@ -21,6 +21,7 @@ import org.apache.spark.sql.SparkSession
 import za.co.absa.pramen.core.app.AppContext
 import za.co.absa.pramen.core.pipeline.Job
 import za.co.absa.pramen.core.runner.jobrunner.ConcurrentJobRunner
+import za.co.absa.pramen.core.runner.repartitioner.JobRepartitioner
 import za.co.absa.pramen.core.state.PipelineState
 
 trait Orchestrator {
@@ -31,5 +32,6 @@ trait Orchestrator {
                               state: PipelineState,
                               appContext: AppContext,
                               jobRunner: ConcurrentJobRunner,
+                              repartitioner: JobRepartitioner,
                               spark: SparkSession): Unit
 }
@@ -26,6 +26,7 @@ import za.co.absa.pramen.core.exceptions.{FatalErrorWrapper, ValidationException
 import za.co.absa.pramen.core.metastore.peristence.TransientJobManager
 import za.co.absa.pramen.core.pipeline.{Job, JobBase, JobDependency, OperationType}
 import za.co.absa.pramen.core.runner.jobrunner.ConcurrentJobRunner
+import za.co.absa.pramen.core.runner.repartitioner.JobRepartitioner
 import za.co.absa.pramen.core.runner.splitter.ScheduleStrategyUtils.evaluateRunDate
 import za.co.absa.pramen.core.state.PipelineState
 import za.co.absa.pramen.core.utils.Emoji._
@@ -57,6 +58,7 @@ class OrchestratorImpl extends Orchestrator {
                                        state: PipelineState,
                                        appContext: AppContext,
                                        jobRunner: ConcurrentJobRunner,
+                                       repartitioner: JobRepartitioner,
                                        spark: SparkSession): Unit = {
     val applicationId = spark.sparkContext.applicationId
     val allOutputTables = jobs.map(_.outputTable.name)
@@ -93,13 +95,15 @@ class OrchestratorImpl extends Orchestrator {
     jobRunner.startWorkerLoop(runJobChannel)
 
     val atLeastOneStarted = sendPendingJobs(runJobChannel, dependencyResolver)
+    var hasFailures = false
     var hasFatalErrors = false
     var hasCriticalJobFailures = false
 
     if (atLeastOneStarted) {
       completedJobsChannel.foreach { case (finishedJob, taskResults, isSucceeded) =>
         runningJobs.remove(finishedJob)
 
+        hasFailures = hasFailures || !isSucceeded
         hasFatalErrors = hasFatalErrors || taskResults.exists(status => isFatalFailure(status.runStatus))
         hasCriticalJobFailures = hasCriticalJobFailures || TransientJobManager.hasCriticalLazyJobFailed || (!isSucceeded && finishedJob.operation.isCritical)
 
@@ -130,6 +134,20 @@ class OrchestratorImpl extends Orchestrator {
           }
         }
       }
+
+      if (!hasFailures && pendingJobs.isEmpty && appContext.appConfig.runtimeConfig.enableRepartitioning) {
+        log.info("Starting repartitioning of the completed jobs...")
+        appContext.appConfig.runtimeConfig.bulkLoadCurrent.foreach { bulkConfig =>
+          jobs.filter(job =>
+            !job.taskDef.outputTable.format.isLazy &&
+              !job.taskDef.outputTable.format.isTransient &&
+              job.operation.enableRepartitioning
+          ).foreach { job =>
+            val taskResults = repartitioner.repartition(job)
+            state.addTaskCompletion(taskResults)
+          }
+        }
+      }
     }
 
     pendingJobs.foreach(job => {

@@ -0,0 +1,24 @@
+/*
+ * Copyright 2022 ABSA Group Limited
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package za.co.absa.pramen.core.runner.repartitioner
+
+import za.co.absa.pramen.api.status.TaskResult
+import za.co.absa.pramen.core.pipeline.Job
+
+trait JobRepartitioner {
+  def repartition(job: Job): Seq[TaskResult]
+}