#791 Add automatic repartitioning of parquet tables after bulk history loads

README.md

@@ -3138,6 +3138,49 @@ pramen.operations = [
 ```
 </details>
 
+
+### (Experimental) Bulk load of historical data
+Sometimes data needs to be loaded and processed for historical periods that may span several years. You can, of course, 
+run Pramen with `--date-from` and `--date-to` to load such data, but for daily datasets this can take a long time because 
+each day is processed independently. Bulk loading allows data to be loaded in `monthly`, `quarterly`, or `yearly` chunks.
+
+Bulk loading works as follows:
+- Each month, quarter, or year is loaded into a single info date partition corresponding to the first info date of that
+  period.
+- Data is loaded for each period independently. Pramen tracks progress in the `bulk_loads` table. If a job is interrupted 
+  and later restarted, Pramen resumes processing from where it left off.
+- After all data for a period has been loaded, you can enable repartitioning so that `pramen_info_date` matches the
+  daily dates.
+
+Example configuration options:
+```hocon
+pramen {
+  # The period of data to load
+  load.date.from = "2000-01-01"
+  load.date.to = "2020-12-31"
+
+  runtime.run.mode = bulk
+  runtime.run.bulk.batch.size = monthly # Can be quarterly or yearly as well
+  runtime.inverse.order = true
+
+  runtime.info.date.column = "transaction_timestamp"
+  runtime.info.date.format = "yyyy-MM-dd" # Only is the info date column data type is not 'date' or 'timestamp'  
+  runtime.enable.repartitioning = true # This is false by default - please use with caustion since this is an experimental feature
+}
+```
+
+Alternatively, you can use command line to run bulk loads without changing the config like this:
+```
+--workflow "dummy.config" \
+--date-from "2000-01-01" \
+--date-to "2020-12-31" \
+--inverse-order "true" \
+--run-mode "bulk" \
+--bulk-size "yearly" \
+--info-date-column "info_date" \
+--info-date-format "yyyyMMdd"
+```
+
 ## Pipeline Notifications
 Custom pipeline notification targets allow execution arbitrary actions after the pipeline is finished. Usually, it is 
 used to send custom notifications to external systems. A pipeline notification target can be created by implementing

pramen/core/src/main/scala/za/co/absa/pramen/core/app/config/BulkRunConfig.scala

@@ -22,5 +22,6 @@ case class BulkRunConfig(
                           dataDateFrom: LocalDate,
                           dataDateTo: LocalDate,
                           infoDateColumn: Option[String],
+                          infoDateFormat: String,
                           outputInfoDate: LocalDate
                         )

pramen/core/src/main/scala/za/co/absa/pramen/core/app/config/RuntimeConfig.scala

@@ -69,6 +69,7 @@ object RuntimeConfig {
   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 INFO_DATE_FORMAT = "pramen.runtime.info.date.format"
   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"
@@ -165,10 +166,11 @@ object RuntimeConfig {
     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 infoDateFormat2 = ConfigUtils.getOptionString(conf, INFO_DATE_FORMAT).getOrElse(infoDateFormat)
     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, infoDateColumn, bulkCurrentOutputInfoDate.get))
+      Some(BulkRunConfig(bulkCurrentDateFrom.get, bulkCurrentDateTo.get, infoDateColumn, infoDateFormat2, bulkCurrentOutputInfoDate.get))
     } else {
       None
     }

pramen/core/src/main/scala/za/co/absa/pramen/core/cmd/CmdLineConfig.scala

@@ -45,6 +45,7 @@ case class CmdLineConfig(
                           mode: Option[String] = None,
                           bulkSize: Option[String] = None,
                           infoDateColumn: Option[String] = None,
+                          infoDateFormat: Option[String] = None,
                           inverseOrder: Option[Boolean] = None,
                           verbose: Option[Boolean] = None,
                           overrideLogLevel: Option[String] = None,
@@ -137,6 +138,9 @@ object CmdLineConfig {
     for (infoDateColumn <- cmd.infoDateColumn)
       accumulatedConfig = accumulatedConfig.withValue(INFO_DATE_COLUMN, ConfigValueFactory.fromAnyRef(infoDateColumn))
 
+    for (infoDateColumn <- cmd.infoDateFormat)
+      accumulatedConfig = accumulatedConfig.withValue(INFO_DATE_FORMAT, ConfigValueFactory.fromAnyRef(infoDateColumn))
+
     for (logEffectiveConfig <- cmd.logEffectiveConfig)
       accumulatedConfig = accumulatedConfig.withValue(LOG_EFFECTIVE_CONFIG, ConfigValueFactory.fromAnyRef(logEffectiveConfig))
 
@@ -223,7 +227,13 @@ object CmdLineConfig {
           .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."))
+            else failure("Invalid information date column name. Must be a non-empty string.")),
+        opt[String]("info-date-format").optional().action((value, config) =>
+            config.copy(infoDateFormat = Option(value)))
+          .text("The format of the information date column if it is not of date or datetime/timestamp type, for repartitioning.")
+          .validate(v =>
+            if (v.nonEmpty) success
+            else failure("Invalid information date format. Must be a non-empty string."))
       )
 
     opt[Boolean]("inverse-order").optional().action((value, config) =>

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

@@ -43,9 +43,9 @@ trait MetastorePersistence {
 
   def isRepartitioningSupported: Boolean
 
-  def repartitionPhase1(infoDateColumn: String, infoDateFrom: LocalDate, infoDateTo: LocalDate, outputInfoDate: LocalDate): Unit = {}
+  def repartitionPhase1(infoDateColumn: String, infoDateFormat: String, infoDateFrom: LocalDate, infoDateTo: LocalDate, outputInfoDate: LocalDate): Unit = {}
 
-  def repartitionPhase2(infoDateColumn: String, infoDateFrom: LocalDate, infoDateTo: LocalDate, outputInfoDate: LocalDate): Unit = {}
+  def repartitionPhase2(infoDateColumn: String, infoDateFormat: String, infoDateFrom: LocalDate, infoDateTo: LocalDate, outputInfoDate: LocalDate): Unit = {}
 }
 
 object MetastorePersistence {

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

@@ -18,7 +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.apache.spark.sql.types.{DateType, StringType, TimestampType}
 import org.slf4j.LoggerFactory
 import za.co.absa.pramen.api.{CatalogTable, PartitionScheme}
 import za.co.absa.pramen.core.metastore.MetaTableStats
@@ -113,25 +113,39 @@ 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 isRepartitioningSupported: Boolean = partitionScheme != PartitionScheme.NotPartitioned
 
-  override def repartitionPhase1(infoDateDataColumn: String, infoDateFrom: LocalDate, infoDateTo: LocalDate, outputInfoDate: LocalDate): Unit = {
+  override def repartitionPhase1(infoDateDataColumn: String, infoDateDataFormat: 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)")
 
+    if (partitionScheme == PartitionScheme.Overwrite)
+      throw new IllegalArgumentException(s"Repartitioning is not supported for this partition scheme: ${partitionScheme.getClass.getSimpleName}")
+
     val fullTableName = table.getFullTableName
     val df = spark.table(fullTableName)
       .filter(getFilter(Some(outputInfoDate), Some(outputInfoDate)))
 
+    val dataInfoDateType = df.schema.fields
+      .find(_.name.equalsIgnoreCase(infoDateDataColumn))
+      .map(_.dataType)
+      .getOrElse(StringType)
+
+    val castExpression = dataInfoDateType match {
+      case _: DateType      => col(infoDateDataColumn)
+      case _: TimestampType => col(infoDateDataColumn).cast(DateType)
+      case _                => to_date(col(infoDateDataColumn).cast(StringType), infoDateDataFormat)
+    }
+
     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))
+    val dfToWrite = df.withColumn(infoDateColumn, castExpression)
 
     writeRepartitionedDf(dfToWrite, fullTableName, infoDateColumn, infoDateFrom, infoDateTo, writeOptions)
   }
 
-  override def repartitionPhase2(infoDateDataColumn: String, infoDateFrom: LocalDate, infoDateTo: LocalDate, outputInfoDate: LocalDate): Unit = {
+  override def repartitionPhase2(infoDateDataColumn: String, infoDateDataFormat: 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)")
   }

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

@@ -18,6 +18,7 @@ package za.co.absa.pramen.core.metastore.peristence
 
 import org.apache.hadoop.fs.Path
 import org.apache.spark.sql.functions._
+import org.apache.spark.sql.types.{DateType, StringType, TimestampType}
 import org.apache.spark.sql.{Column, DataFrame, SaveMode, SparkSession}
 import org.slf4j.LoggerFactory
 import za.co.absa.pramen.api.{PartitionInfo, PartitionScheme}
@@ -145,7 +146,86 @@ class MetastorePersistenceParquet(path: String,
     throw new UnsupportedOperationException("Parquet format does not support Hive tables at the moment.")
   }
 
-  override def isRepartitioningSupported: Boolean = false
+  override def isRepartitioningSupported: Boolean = partitionScheme == PartitionScheme.PartitionByDay
+
+  override def repartitionPhase1(infoDateDataColumn: String, infoDateDataFormat: String, infoDateDataFrom: LocalDate, infoDateDataTo: LocalDate, outputInfoDate: LocalDate): Unit = {
+    ensureRepartitioningSupported(infoDateDataColumn)
+
+    val infoDateFrom = outputInfoDate
+    val infoDateTo = outputInfoDate.plusYears(1000)
+    val pathFrom = SparkUtils.getPartitionPath(infoDateFrom, infoDateColumn, infoDateFormat, path)
+    val pathTo = SparkUtils.getPartitionPath(infoDateTo, infoDateColumn, infoDateFormat, path)
+
+    val fsUtils = new FsUtils(spark.sparkContext.hadoopConfiguration, path)
+
+    log.info(s"Repartitioning phase 1.1 - copying files to a temporary location ($pathTo)...")
+    fsUtils.copyDirectory(pathFrom, pathTo)
+    val sizeFrom = fsUtils.getDirectorySize(pathFrom.toUri.toString)
+    val sizeTo = fsUtils.getDirectorySize(pathTo.toUri.toString)
+    if (sizeFrom != sizeTo) {
+      throw new IllegalStateException(s"Repartitioning failed: size mismatch between $pathFrom ($sizeFrom) and $pathTo ($sizeTo)")
+    }
+  }
+
+  override def repartitionPhase2(infoDateDataColumn: String, infoDateDataFormat: String, infoDateDataFrom: LocalDate, infoDateDataTo: LocalDate, outputInfoDate: LocalDate): Unit = {
+    ensureRepartitioningSupported(infoDateDataColumn)
+
+    val infoDateFrom = outputInfoDate.plusYears(1000)
+    val pathFrom = SparkUtils.getPartitionPath(infoDateFrom, infoDateColumn, infoDateFormat, path)
+    val pathTo = SparkUtils.getPartitionPath(outputInfoDate, infoDateColumn, infoDateFormat, path)
+    val fsUtils = new FsUtils(spark.sparkContext.hadoopConfiguration, path)
+    if (!fsUtils.exists(pathFrom)) {
+      throw new IllegalArgumentException(s"Path does not exist: $pathFrom")
+    }
+
+    log.info(s"Repartitioning phase 2.1 - deleting data in the original partition ($pathTo)...")
+    fsUtils.deleteDirectoryRecursively(pathTo)
+
+    log.info(s"Repartitioning phase 2.2 - deleting data from target partitions ($infoDateDataFrom to $infoDateDataTo)...")
+    var date = infoDateDataFrom
+    while(date.isBefore(infoDateDataTo) || date.isEqual(infoDateDataTo)) {
+      val pathToDelete = SparkUtils.getPartitionPath(date, infoDateColumn, infoDateFormat, path)
+      fsUtils.deleteDirectoryRecursively(pathToDelete)
+      date = date.plusDays(1)
+    }
+
+    log.info(s"Repartitioning phase 2.3 - appending data to target partitions ($infoDateDataFrom to $infoDateDataTo)...")
+    val df = spark.read
+      .format("parquet")
+      .options(readOptions)
+      .load(pathFrom.toUri.toString)
+
+    val dataInfoDateType = df.schema.fields
+      .find(_.name.equalsIgnoreCase(infoDateDataColumn))
+      .map(_.dataType)
+      .getOrElse(StringType)
+
+    val castExpression = dataInfoDateType match {
+      case _: DateType      => col(infoDateDataColumn)
+      case _: TimestampType => col(infoDateDataColumn).cast(DateType)
+      case _                => to_date(col(infoDateDataColumn).cast(StringType), infoDateDataFormat)
+    }
+
+    df.withColumn(infoDateColumn, castExpression)
+      .write
+      .format("parquet")
+      .mode(SaveMode.Append)
+      .partitionBy(infoDateColumn)
+      .options(writeOptions)
+      .save(path)
+
+    log.info(s"Repartitioning phase 2.4 - deleting data from the temporary location ($pathFrom)...")
+    fsUtils.deleteDirectoryRecursively(pathFrom)
+  }
+
+  private def ensureRepartitioningSupported(infoDateDataColumn: String): 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)")
+
+    if (partitionScheme != PartitionScheme.PartitionByDay)
+      throw new IllegalArgumentException(s"Repartitioning is not supported for this partition scheme: ${partitionScheme.getClass.getSimpleName}")
+  }
+
 
   def loadPartitionDirectly(infoDate: LocalDate): DataFrame = {
     val dateStr = dateFormatter.format(infoDate)

pramen/core/src/main/scala/za/co/absa/pramen/core/runner/repartitioner/JobRepartitionerImpl.scala

@@ -85,7 +85,11 @@ class JobRepartitionerImpl(bulkLoadCurrent: BulkRunConfig,
       // Starting repartitioning phase 1
 
       if (persistence.isRepartitioningSupported) {
-        persistence.repartitionPhase1(infoDateColumn, bulkLoadCurrent.outputInfoDate, bulkLoadCurrent.outputInfoDate, bulkLoadCurrent.outputInfoDate)
+        persistence.repartitionPhase1(infoDateColumn,
+          bulkLoadCurrent.infoDateFormat,
+          bulkLoadCurrent.dataDateFrom,
+          bulkLoadCurrent.dataDateTo,
+          bulkLoadCurrent.outputInfoDate)
         val updatedState = bulkLoadState.copy(phase = BulkLoadPhase.Repartition1)
         bulkLoadStateManager.updatePhase(updatedState)
 
@@ -101,7 +105,11 @@ class JobRepartitionerImpl(bulkLoadCurrent: BulkRunConfig,
       // Starting repartitioning phase 2
 
       if (persistence.isRepartitioningSupported) {
-        persistence.repartitionPhase2(infoDateColumn, bulkLoadCurrent.outputInfoDate, bulkLoadCurrent.outputInfoDate, bulkLoadCurrent.outputInfoDate)
+        persistence.repartitionPhase2(infoDateColumn,
+          bulkLoadCurrent.infoDateFormat,
+          bulkLoadCurrent.dataDateFrom,
+          bulkLoadCurrent.dataDateTo,
+          bulkLoadCurrent.outputInfoDate)
         val updatedState = bulkLoadState.copy(phase = BulkLoadPhase.Done)
         bulkLoadStateManager.updatePhase(updatedState)
         BulkLoadPhase.Done

pramen/core/src/main/scala/za/co/absa/pramen/core/utils/FsUtils.scala

@@ -466,6 +466,38 @@ class FsUtils(conf: Configuration, pathBase: String) {
     }
   }
 
+  /**
+    * Copies the contents of a directory to another location.
+    *
+    * The source directory must exist, otherwise an `IllegalArgumentException` is thrown.
+    * If the destination directory already exists, it is deleted recursively and
+    * recreated before the copy operation starts. Every file found directly in the
+    * source directory is then copied into the destination directory, keeping its
+    * original name, using a retrying copy operation to tolerate transient failures.
+    *
+    * @param srcDir the path of the existing directory whose files are copied
+    * @param dstDir the path of the target directory, recreated from scratch before copying
+    * @return nothing, the method is executed only for its side effects on the file system
+    * @throws IllegalArgumentException if the source path does not exist
+    */
+  def copyDirectory(srcDir: Path, dstDir: Path): Unit = {
+    if (!exists(srcDir))
+      throw new IllegalArgumentException("No data in the source path: " + srcDir)
+
+    if (exists(dstDir))
+      deleteDirectoryRecursively(dstDir)
+
+    createDirectoryRecursive(dstDir)
+
+    val listOfFiles = getHadoopFiles(srcDir, includeHiddenFiles = true)
+
+    listOfFiles.foreach(file => {
+      val srcPath = file.getPath
+      val dstFile = new Path(dstDir, srcPath.getName)
+      copyFileWithRetry(srcPath, dstFile)
+    })
+  }
+
   def copyToLocal(srcFile: Path, targetFile: Path, overwrite: Boolean = false): Unit = {
     if (!overwrite && fs.exists(targetFile)) {
       throw new IllegalStateException(s"Target file $targetFile already exists.")

pramen/core/src/test/scala/za/co/absa/pramen/core/cmd/CmdLineConfigSuite.scala

@@ -244,7 +244,7 @@ class CmdLineConfigSuite extends AnyWordSpec {
     }
 
     "return a modified config if bulk mode for date-to override is specified" in {
-      val cmd = CmdLineConfig.parseCmdLine(Array("--workflow", "dummy.config", "--date-to", "2020-08-15", "--inverse-order", "true", "--run-mode", "bulk", "--bulk-size", "yearly", "--info-date-column", "info_date"))
+      val cmd = CmdLineConfig.parseCmdLine(Array("--workflow", "dummy.config", "--date-to", "2020-08-15", "--inverse-order", "true", "--run-mode", "bulk", "--bulk-size", "yearly", "--info-date-column", "info_date", "--info-date-format", "yyyyMMdd"))
       val config = CmdLineConfig.applyCmdLineToConfig(emptyConfig, cmd.get)
 
       assert(config.hasPath(LOAD_DATE_TO))
@@ -256,6 +256,7 @@ class CmdLineConfigSuite extends AnyWordSpec {
       assert(config.getString(RUN_MODE) == "bulk")
       assert(config.getString(RUN_BULK_BATCH_SIZE) == "yearly")
       assert(config.getString(INFO_DATE_COLUMN) == "info_date")
+      assert(config.getString(INFO_DATE_FORMAT) == "yyyyMMdd")
     }
 
     "return the original config if no cmd line arguments are provided" in {

pramen/core/src/test/scala/za/co/absa/pramen/core/integration/BulkLoadLongSuite.scala

@@ -154,16 +154,16 @@ class BulkLoadLongSuite extends AnyWordSpec
 
         assert(df.count() == 59)
 
-        // For now...
-        assert(!df.filter(col("dt") =!= col("pramen_info_date")).isEmpty)
+        // Ensure parquet directory was repartitioned
+        assert(df.filter(!(col("dt") <=> col("pramen_info_date"))).isEmpty)
 
-        // Running the job for the second time shouyld not change the output
+        // Running the job for the second time should not change the output
         val exitCode2 = AppRunner.runBulkPipelines(conf)
         assert(exitCode2 == 0)
 
         val df2 = spark.table(tableName)
         assert(df2.count() == 59)
-        assert(df2.filter(col("dt") =!= col("pramen_info_date")).isEmpty)
+        assert(df2.filter(!(col("dt") <=> col("pramen_info_date"))).isEmpty)
 
         spark.sql(s"DELETE FROM $tableName").count()
       }