Repository: AbsaOSS/pramen (language: python)

Issue:
Add write modes to metastore tables
## Background
This is something that can be useful for incremental ingestions.

## Feature
Add write modes to metastore tables

## Example [Optional]
A simple example if applicable.

## Proposed Solution [Optional]
Solution Ideas 
1. 
2. 
3.



Discussion / hints:

Fix patch:
diff --git a/pramen/core/src/main/scala/za/co/absa/pramen/core/metastore/model/MetaTable.scala b/pramen/core/src/main/scala/za/co/absa/pramen/core/metastore/model/MetaTable.scala
index 3e57709b9..da4f9b8f6 100644
--- a/pramen/core/src/main/scala/za/co/absa/pramen/core/metastore/model/MetaTable.scala
+++ b/pramen/core/src/main/scala/za/co/absa/pramen/core/metastore/model/MetaTable.scala
@@ -17,6 +17,7 @@
 package za.co.absa.pramen.core.metastore.model
 
 import com.typesafe.config.Config
+import org.apache.spark.sql.SaveMode
 import org.slf4j.LoggerFactory
 import za.co.absa.pramen.api.{DataFormat, MetaTableDef}
 import za.co.absa.pramen.core.app.config.InfoDateConfig
@@ -60,6 +61,7 @@ case class MetaTable(
                       infoDateStart: LocalDate,
                       trackDays: Int,
                       trackDaysExplicitlySet: Boolean,
+                      saveModeOpt: Option[SaveMode],
                       readOptions: Map[String, String],
                       writeOptions: Map[String, String],
                       sparkConfig: Map[String, String]
@@ -74,6 +76,7 @@ object MetaTable {
   val HIVE_PATH_KEY = "hive.path"
   val HIVE_PREFER_ADD_PARTITION_KEY = "hive.prefer.add.partition"
   val TRACK_DAYS_KEY = "track.days"
+  val SAVE_MODE_OPTION_KEY = "save.mode"
   val READ_OPTION_KEY = "read.option"
   val WRITE_OPTION_KEY = "write.option"
   val TABLE_HIVE_CONFIG_PREFIX = "hive"
@@ -140,6 +143,7 @@ object MetaTable {
       HiveConfig.fromConfigWithDefaults(ConfigUtils.getOptionConfig(conf, TABLE_HIVE_CONFIG_PREFIX), defaultHiveConfig, format)
     }
 
+    val saveModeOpt = ConfigUtils.getOptionString(conf, SAVE_MODE_OPTION_KEY).map(getSaveMode(_, name))
     val readOptions = ConfigUtils.getExtraOptions(conf, READ_OPTION_KEY)
     val writeOptions = ConfigUtils.getExtraOptions(conf, WRITE_OPTION_KEY)
     val sparkConfig = ConfigUtils.getExtraOptions(conf, SPARK_CONFIG_PREFIX)
@@ -157,6 +161,7 @@ object MetaTable {
       startDate,
       trackDays,
       trackDaysExplicitlySet,
+      saveModeOpt,
       readOptions,
       writeOptions,
       sparkConfig)
@@ -176,4 +181,12 @@ object MetaTable {
       table.writeOptions
     )
   }
+
+  private[core] def getSaveMode(saveModeStr: String, table: String): SaveMode = {
+    saveModeStr.toLowerCase.trim match {
+      case "append" => SaveMode.Append
+      case "overwrite" => SaveMode.Overwrite
+      case _ => throw new IllegalArgumentException(s"Invalid or unsupported save mode: '$saveModeStr' for table '$table'.")
+    }
+  }
 }
diff --git a/pramen/core/src/main/scala/za/co/absa/pramen/core/metastore/peristence/MetastorePersistence.scala b/pramen/core/src/main/scala/za/co/absa/pramen/core/metastore/peristence/MetastorePersistence.scala
index 2f220d3d9..8a858ad08 100644
--- a/pramen/core/src/main/scala/za/co/absa/pramen/core/metastore/peristence/MetastorePersistence.scala
+++ b/pramen/core/src/main/scala/za/co/absa/pramen/core/metastore/peristence/MetastorePersistence.scala
@@ -47,14 +47,14 @@ object MetastorePersistence {
     metaTable.format match {
       case DataFormat.Parquet(path, recordsPerPartition) =>
         new MetastorePersistenceParquet(
-          path, metaTable.infoDateColumn, metaTable.infoDateFormat, recordsPerPartition, metaTable.readOptions, metaTable.writeOptions
+          path, metaTable.infoDateColumn, metaTable.infoDateFormat, recordsPerPartition, metaTable.saveModeOpt, metaTable.readOptions, metaTable.writeOptions
         )
       case DataFormat.Delta(query, recordsPerPartition)  =>
         new MetastorePersistenceDelta(
-          query, metaTable.infoDateColumn, metaTable.infoDateFormat, recordsPerPartition, metaTable.readOptions, metaTable.writeOptions
+          query, metaTable.infoDateColumn, metaTable.infoDateFormat, recordsPerPartition, metaTable.saveModeOpt, metaTable.readOptions, metaTable.writeOptions
         )
       case DataFormat.Raw(path)                          =>
-        new MetastorePersistenceRaw(path, metaTable.infoDateColumn, metaTable.infoDateFormat)
+        new MetastorePersistenceRaw(path, metaTable.infoDateColumn, metaTable.infoDateFormat, metaTable.saveModeOpt)
       case DataFormat.TransientEager(cachePolicy)             =>
         new MetastorePersistenceTransientEager(TransientTableManager.getTempDirectory(cachePolicy, conf), metaTable.name, cachePolicy)
       case DataFormat.Transient(cachePolicy) =>
diff --git a/pramen/core/src/main/scala/za/co/absa/pramen/core/metastore/peristence/MetastorePersistenceDelta.scala b/pramen/core/src/main/scala/za/co/absa/pramen/core/metastore/peristence/MetastorePersistenceDelta.scala
index aa8380e4a..9fdfce062 100644
--- a/pramen/core/src/main/scala/za/co/absa/pramen/core/metastore/peristence/MetastorePersistenceDelta.scala
+++ b/pramen/core/src/main/scala/za/co/absa/pramen/core/metastore/peristence/MetastorePersistenceDelta.scala
@@ -36,6 +36,7 @@ class MetastorePersistenceDelta(query: Query,
                                 infoDateColumn: String,
                                 infoDateFormat: String,
                                 recordsPerPartition: Option[Long],
+                                saveModeOpt: Option[SaveMode],
                                 readOptions: Map[String, String],
                                 writeOptions: Map[String, String]
                                  )(implicit spark: SparkSession) extends MetastorePersistence {
@@ -72,14 +73,23 @@ class MetastorePersistenceDelta(query: Query,
 
     val dfRepartitioned = applyRepartitioning(dfIn, recordCount)
 
-    log.debug(s"Schema: ${dfIn.schema.treeString}")
-    log.debug(s"Info date column: $infoDateColumn")
-    log.debug(s"Info date: $infoDateStr")
+    val saveMode = saveModeOpt.getOrElse(SaveMode.Overwrite)
+
+    val operationStr = saveMode match {
+      case SaveMode.Append => "Appending to"
+      case _               => "Writing to"
+    }
+
+    if (log.isDebugEnabled) {
+      log.debug(s"Schema: ${dfIn.schema.treeString}")
+      log.debug(s"Info date column: $infoDateColumn")
+      log.debug(s"Info date: $infoDateStr")
+    }
 
     val writer = dfRepartitioned
       .write
       .format("delta")
-      .mode(SaveMode.Overwrite)
+      .mode(saveMode)
       .partitionBy(infoDateColumn)
       .option("mergeSchema", "true")
       .option("replaceWhere", s"$infoDateColumn='$infoDateStr'")
@@ -89,11 +99,11 @@ class MetastorePersistenceDelta(query: Query,
       case Query.Path(path)   =>
         val fsUtils = new FsUtils(spark.sparkContext.hadoopConfiguration, path)
         fsUtils.createDirectoryRecursive(new Path(path))
-        log.info(s"Writing to path '$path' WHERE $whereCondition...")
+        log.info(s"$operationStr to path '$path' WHERE $whereCondition...")
         writer.save(path)
       case Query.Table(table) =>
         writer.saveAsTable(table)
-        log.info(s"Writing to table '$table' WHERE $whereCondition...")
+        log.info(s"$operationStr to table '$table' WHERE $whereCondition...")
       case q =>
         throw new IllegalStateException(s"The '${q.name}' option is not supported as a write target for Delta.")
     }
diff --git a/pramen/core/src/main/scala/za/co/absa/pramen/core/metastore/peristence/MetastorePersistenceParquet.scala b/pramen/core/src/main/scala/za/co/absa/pramen/core/metastore/peristence/MetastorePersistenceParquet.scala
index 7e4e2e909..ac082b6b0 100644
--- a/pramen/core/src/main/scala/za/co/absa/pramen/core/metastore/peristence/MetastorePersistenceParquet.scala
+++ b/pramen/core/src/main/scala/za/co/absa/pramen/core/metastore/peristence/MetastorePersistenceParquet.scala
@@ -37,6 +37,7 @@ class MetastorePersistenceParquet(path: String,
                                   infoDateColumn: String,
                                   infoDateFormat: String,
                                   recordsPerPartition: Option[Long],
+                                  saveModeOpt: Option[SaveMode],
                                   readOptions: Map[String, String],
                                   writeOptions: Map[String, String]
                                  )(implicit spark: SparkSession) extends MetastorePersistence {
@@ -69,7 +70,12 @@ class MetastorePersistenceParquet(path: String,
       df
     }
 
-    log.info(s"Writing to '$outputDirStr'...")
+    val saveMode = saveModeOpt.getOrElse(SaveMode.Overwrite)
+
+    saveMode match {
+      case SaveMode.Append => log.info(s"Appending to '$outputDirStr'...")
+      case _               => log.info(s"Writing to '$outputDirStr'...")
+    }
 
     val recordCount = numberOfRecordsEstimate match {
       case Some(count) => count
@@ -78,7 +84,7 @@ class MetastorePersistenceParquet(path: String,
 
     val dfRepartitioned = applyRepartitioning(dfIn, recordCount)
 
-    writeAndCleanOnFailure(dfRepartitioned, outputDirStr, fsUtils)
+    writeAndCleanOnFailure(dfRepartitioned, outputDirStr, fsUtils, saveMode)
 
     val stats = getStats(infoDate)
 
@@ -170,7 +176,8 @@ class MetastorePersistenceParquet(path: String,
 
   private[core] def writeAndCleanOnFailure(df: DataFrame,
                                            outputDirStr: String,
-                                           fsUtils: FsUtils): Unit = {
+                                           fsUtils: FsUtils,
+                                           saveMode: SaveMode): Unit = {
     if (writeOptions.nonEmpty) {
       log.info("Custom write options:")
       ConfigUtils.renderExtraOptions(writeOptions, Keys.KEYS_TO_REDACT)(log.info)
@@ -178,7 +185,7 @@ class MetastorePersistenceParquet(path: String,
 
     try {
       df.write
-        .mode(SaveMode.Overwrite)
+        .mode(saveMode)
         .format("parquet")
         .options(writeOptions)
         .save(outputDirStr)
@@ -188,7 +195,7 @@ class MetastorePersistenceParquet(path: String,
         // We need to delete it to avoid the an attempt to read it in the future.
         Try {
           val partitionPath = new Path(outputDirStr)
-          if (fsUtils.exists(partitionPath)) {
+          if (fsUtils.exists(partitionPath) && saveMode == SaveMode.Overwrite) {
             log.warn(s"The write failed. Deleting the empty directory: $partitionPath")
             fsUtils.deleteDirectoryRecursively(partitionPath)
           }
diff --git a/pramen/core/src/main/scala/za/co/absa/pramen/core/metastore/peristence/MetastorePersistenceRaw.scala b/pramen/core/src/main/scala/za/co/absa/pramen/core/metastore/peristence/MetastorePersistenceRaw.scala
index 7a84a89ce..9266b8bef 100644
--- a/pramen/core/src/main/scala/za/co/absa/pramen/core/metastore/peristence/MetastorePersistenceRaw.scala
+++ b/pramen/core/src/main/scala/za/co/absa/pramen/core/metastore/peristence/MetastorePersistenceRaw.scala
@@ -17,7 +17,7 @@
 package za.co.absa.pramen.core.metastore.peristence
 
 import org.apache.hadoop.fs.{FileStatus, Path}
-import org.apache.spark.sql.{DataFrame, SparkSession}
+import org.apache.spark.sql.{DataFrame, SaveMode, SparkSession}
 import org.slf4j.LoggerFactory
 import za.co.absa.pramen.core.metastore.MetaTableStats
 import za.co.absa.pramen.core.metastore.model.HiveConfig
@@ -29,7 +29,8 @@ import scala.collection.mutable
 
 class MetastorePersistenceRaw(path: String,
                               infoDateColumn: String,
-                              infoDateFormat: String
+                              infoDateFormat: String,
+                              saveModeOpt: Option[SaveMode]
                              )(implicit spark: SparkSession) extends MetastorePersistence {
 
   private val log = LoggerFactory.getLogger(this.getClass)
@@ -59,7 +60,12 @@ class MetastorePersistenceRaw(path: String,
     val fsUtilsTrg = new FsUtils(spark.sparkContext.hadoopConfiguration, outputDir.toString)
 
     if (fsUtilsTrg.exists(outputDir)) {
-      fsUtilsTrg.deleteDirectoryRecursively(outputDir)
+      if (saveModeOpt.contains(SaveMode.Append)) {
+        log.info(s"Appending to partition: $outputDir...")
+      } else {
+        log.info(s"Cleaning up the partition: $outputDir...")
+        fsUtilsTrg.deleteDirectoryRecursively(outputDir)
+      }
     }
 
     fsUtilsTrg.createDirectoryRecursive(outputDir)
diff --git a/pramen/core/src/main/scala/za/co/absa/pramen/core/pipeline/TransferTable.scala b/pramen/core/src/main/scala/za/co/absa/pramen/core/pipeline/TransferTable.scala
index b3058648e..888cfda90 100644
--- a/pramen/core/src/main/scala/za/co/absa/pramen/core/pipeline/TransferTable.scala
+++ b/pramen/core/src/main/scala/za/co/absa/pramen/core/pipeline/TransferTable.scala
@@ -57,7 +57,7 @@ case class TransferTable(
   }
 
   def getMetaTable: MetaTable = {
-    MetaTable(jobMetaTableName, "", DataFormat.Null(), "", "", HiveConfig.getNullConfig, None, None, hivePreferAddPartition = true, None, infoDateStart, trackDays, trackDaysExplicitlySet = trackDaysExplicitlySet, readOptions, writeOptions, sparkConfig)
+    MetaTable(jobMetaTableName, "", DataFormat.Null(), "", "", HiveConfig.getNullConfig, None, None, hivePreferAddPartition = true, None, infoDateStart, trackDays, trackDaysExplicitlySet = trackDaysExplicitlySet, None, readOptions, writeOptions, sparkConfig)
   }
 }
 


Test patch:
diff --git a/pramen/core/src/test/scala/za/co/absa/pramen/core/metastore/model/DataFormatSuite.scala b/pramen/core/src/test/scala/za/co/absa/pramen/core/metastore/model/DataFormatSuite.scala
index 73be1ae07..42fc39d14 100644
--- a/pramen/core/src/test/scala/za/co/absa/pramen/core/metastore/model/DataFormatSuite.scala
+++ b/pramen/core/src/test/scala/za/co/absa/pramen/core/metastore/model/DataFormatSuite.scala
@@ -155,13 +155,13 @@ class DataFormatSuite extends AnyWordSpec {
     }
 
     "throw an exception on unknown format" in {
-      val conf = ConfigFactory.parseString("format = isberg")
+      val conf = ConfigFactory.parseString("format = iceberg")
 
       val ex = intercept[IllegalArgumentException] {
         DataFormatParser.fromConfig(conf, conf)
       }
 
-      assert(ex.getMessage.contains("Unknown format: isberg"))
+      assert(ex.getMessage.contains("Unknown format: iceberg"))
     }
 
     "throw an exception on mandatory options missing" in {
diff --git a/pramen/core/src/test/scala/za/co/absa/pramen/core/metastore/model/MetaTableSuite.scala b/pramen/core/src/test/scala/za/co/absa/pramen/core/metastore/model/MetaTableSuite.scala
index b79714bb7..4be6b2e1d 100644
--- a/pramen/core/src/test/scala/za/co/absa/pramen/core/metastore/model/MetaTableSuite.scala
+++ b/pramen/core/src/test/scala/za/co/absa/pramen/core/metastore/model/MetaTableSuite.scala
@@ -17,6 +17,7 @@
 package za.co.absa.pramen.core.metastore.model
 
 import com.typesafe.config.ConfigFactory
+import org.apache.spark.sql.SaveMode
 import org.scalatest.wordspec.AnyWordSpec
 import za.co.absa.pramen.api.DataFormat.Parquet
 import za.co.absa.pramen.core.app.config.InfoDateConfig
@@ -153,6 +154,7 @@ class MetaTableSuite extends AnyWordSpec {
           |spark.conf = {
           |  key1 = value1
           |}
+          |save.mode = append
           |""".stripMargin)
 
       val defaultHiveConfig = HiveDefaultConfig.getNullConfig
@@ -173,6 +175,7 @@ class MetaTableSuite extends AnyWordSpec {
       assert(metaTable.infoDateFormat == "dd-MM-yyyy")
       assert(metaTable.infoDateStart.toString == "2020-01-31")
       assert(metaTable.sparkConfig("key1") == "value1")
+      assert(metaTable.saveModeOpt.contains(SaveMode.Append))
     }
 
     "load a metatable definition with hive table defined" in {
@@ -233,6 +236,7 @@ class MetaTableSuite extends AnyWordSpec {
       assert(metaTable.writeOptions("x") == "test2")
       assert(metaTable.writeOptions("y") == "101")
       assert(metaTable.sparkConfig("key1") == "value1")
+      assert(metaTable.saveModeOpt.isEmpty)
     }
 
     "load a metatable definition with hive overrides" in {
@@ -310,7 +314,7 @@ class MetaTableSuite extends AnyWordSpec {
       val conf = ConfigFactory.parseString(
         """
           |name = table1
-          |format = isberg
+          |format = iceberg
           |path = /a/b/c
           |hive.table = my_hive_table
           |""".stripMargin)
@@ -323,6 +327,70 @@ class MetaTableSuite extends AnyWordSpec {
 
       assert(ex.getMessage.contains("Unable to read data format from config for the metastore table: table1"))
     }
+
+    "throw an exception when the save mode is not supported" in {
+      val conf = ConfigFactory.parseString(
+        """
+          |name = table1
+          |format = parquet
+          |path = /a/b/c
+          |save.mode = "ignore"
+          |""".stripMargin)
+
+      val defaultHiveConfig = HiveDefaultConfig.getNullConfig
+
+      val ex = intercept[IllegalArgumentException] {
+        MetaTable.fromConfigSingleEntity(conf, conf, "", "", LocalDate.parse("2020-01-31"), 0, defaultHiveConfig, defaultPreferAddPartition = true)
+      }
+
+      assert(ex.getMessage.contains("Invalid or unsupported save mode: 'ignore' for table 'table1'."))
+    }
+
+    "throw an exception when the save mode is invalid" in {
+      val conf = ConfigFactory.parseString(
+        """
+          |name = table1
+          |format = parquet
+          |path = /a/b/c
+          |save.mode = "test"
+          |""".stripMargin)
+
+      val defaultHiveConfig = HiveDefaultConfig.getNullConfig
+
+      val ex = intercept[IllegalArgumentException] {
+        MetaTable.fromConfigSingleEntity(conf, conf, "", "", LocalDate.parse("2020-01-31"), 0, defaultHiveConfig, defaultPreferAddPartition = true)
+      }
+
+      assert(ex.getMessage.contains("Invalid or unsupported save mode: 'test' for table 'table1'."))
+    }
+  }
+
+  "getSaveMode()" should {
+    "return SaveMode.Append" in {
+      assert(MetaTable.getSaveMode("append", "table1") == SaveMode.Append)
+    }
+
+    "return SaveMode.Overwrite" in {
+      assert(MetaTable.getSaveMode("overwrite", "table1") == SaveMode.Overwrite)
+    }
+
+    "throw an exception when the save mode is not supported" in {
+      val ex = intercept[IllegalArgumentException] {
+        MetaTable.getSaveMode("error_if_exists", "table1")
+      }
+
+      assert(ex.getMessage.contains("Invalid or unsupported save mode: 'error_if_exists' for table 'table1'"))
+    }
+
+
+    "throw an exception when the save mode is invalid" in {
+      val ex = intercept[IllegalArgumentException] {
+        MetaTable.getSaveMode("invalid", "table1")
+      }
+
+      assert(ex.getMessage.contains("Invalid or unsupported save mode: 'invalid' for table 'table1'"))
+    }
+
   }
 
 }
diff --git a/pramen/core/src/test/scala/za/co/absa/pramen/core/metastore/persistence/MetastorePersistenceParquetSuite.scala b/pramen/core/src/test/scala/za/co/absa/pramen/core/metastore/persistence/MetastorePersistenceParquetSuite.scala
index 95fab3323..600afba44 100644
--- a/pramen/core/src/test/scala/za/co/absa/pramen/core/metastore/persistence/MetastorePersistenceParquetSuite.scala
+++ b/pramen/core/src/test/scala/za/co/absa/pramen/core/metastore/persistence/MetastorePersistenceParquetSuite.scala
@@ -17,7 +17,7 @@
 package za.co.absa.pramen.core.metastore.persistence
 
 import org.apache.hadoop.fs.Path
-import org.apache.spark.sql.DataFrame
+import org.apache.spark.sql.{DataFrame, SaveMode}
 import org.mockito.Mockito.{mock, when}
 import org.scalatest.wordspec.AnyWordSpec
 import za.co.absa.pramen.core.base.SparkTestBase
@@ -39,24 +39,42 @@ class MetastorePersistenceParquetSuite extends AnyWordSpec with SparkTestBase wi
       withTempDirectory("metastore_parquet") { tempDir =>
         val outputPath = new Path(tempDir, "partition=10")
 
-        val persistence = new MetastorePersistenceParquet(tempDir, "ignore", "yyyy-MM-dd", None, Map.empty, Map.empty)
+        val persistence = new MetastorePersistenceParquet(tempDir, "ignore", "yyyy-MM-dd", None, None, Map.empty, Map.empty)
 
-        persistence.writeAndCleanOnFailure(exampleDf, outputPath.toString, fsUtils)
+        persistence.writeAndCleanOnFailure(exampleDf, outputPath.toString, fsUtils, SaveMode.Overwrite)
 
         assert(fsUtils.exists(outputPath))
         assert(fsUtils.getFilesRecursive(outputPath, "*.parquet").nonEmpty)
       }
     }
 
+    "append data when save mode is append" in {
+      withTempDirectory("metastore_parquet") { tempDir =>
+        val outputPath = new Path(tempDir, "partition=10")
+
+        val persistence = new MetastorePersistenceParquet(tempDir, "ignore", "yyyy-MM-dd", None, Some(SaveMode.Append), Map.empty, Map.empty)
+
+        persistence.writeAndCleanOnFailure(exampleDf, outputPath.toString, fsUtils, SaveMode.Append)
+        persistence.writeAndCleanOnFailure(exampleDf, outputPath.toString, fsUtils, SaveMode.Append)
+
+        assert(fsUtils.exists(outputPath))
+        assert(fsUtils.getFilesRecursive(outputPath, "*.parquet").nonEmpty)
+
+        val df = spark.read.parquet(outputPath.toString)
+
+        assert(df.count() == 6)
+      }
+    }
+
     "re-throw on failure" in {
       val df = mock(classOf[DataFrame])
 
       when(df.write).thenThrow(new RuntimeException("test exception"))
 
-      val persistence = new MetastorePersistenceParquet("dummy", "ignore", "yyyy-MM-dd", None, Map.empty, Map.empty)
+      val persistence = new MetastorePersistenceParquet("dummy", "ignore", "yyyy-MM-dd", None, None, Map.empty, Map.empty)
 
       assertThrows[RuntimeException] {
-        persistence.writeAndCleanOnFailure(df, "dummy", fsUtils)
+        persistence.writeAndCleanOnFailure(df, "dummy", fsUtils, SaveMode.Overwrite)
       }
     }
 
@@ -70,10 +88,10 @@ class MetastorePersistenceParquetSuite extends AnyWordSpec with SparkTestBase wi
 
         fsUtils.createDirectoryRecursive(outputPath)
 
-        val persistence = new MetastorePersistenceParquet("dummy", "ignore", "yyyy-MM-dd", None, Map.empty, Map.empty)
+        val persistence = new MetastorePersistenceParquet("dummy", "ignore", "yyyy-MM-dd", None, None, Map.empty, Map.empty)
 
         val ex = intercept[RuntimeException] {
-          persistence.writeAndCleanOnFailure(df, outputPath.toString, fsUtils)
+          persistence.writeAndCleanOnFailure(df, outputPath.toString, fsUtils, SaveMode.Overwrite)
         }
 
         assert(ex.getMessage.contains("test exception"))
diff --git a/pramen/core/src/test/scala/za/co/absa/pramen/core/metastore/persistence/MetastorePersistenceRawSuite.scala b/pramen/core/src/test/scala/za/co/absa/pramen/core/metastore/persistence/MetastorePersistenceRawSuite.scala
index 9898e6325..5bb8e124f 100644
--- a/pramen/core/src/test/scala/za/co/absa/pramen/core/metastore/persistence/MetastorePersistenceRawSuite.scala
+++ b/pramen/core/src/test/scala/za/co/absa/pramen/core/metastore/persistence/MetastorePersistenceRawSuite.scala
@@ -17,6 +17,7 @@
 package za.co.absa.pramen.core.metastore.persistence
 
 import org.apache.hadoop.fs.Path
+import org.apache.spark.sql.SaveMode
 import org.scalatest.wordspec.AnyWordSpec
 import za.co.absa.pramen.core.base.SparkTestBase
 import za.co.absa.pramen.core.fixtures.{TempDirFixture, TextComparisonFixture}
@@ -47,7 +48,7 @@ class MetastorePersistenceRawSuite extends AnyWordSpec with SparkTestBase with T
         fsUtils.fs.create(new Path(dataPath, "1.dat")).close()
         fsUtils.fs.create(new Path(dataPath, "2.dat")).close()
 
-        val persistence = new MetastorePersistenceRaw(tempDir, infoDateColumn, infoDateFormat)
+        val persistence = new MetastorePersistenceRaw(tempDir, infoDateColumn, infoDateFormat, None)
 
         val actual = persistence.loadTable(Some(infoDate), Some(infoDate)).orderBy("path").collect().map(_.getString(0))
 
@@ -68,7 +69,7 @@ class MetastorePersistenceRawSuite extends AnyWordSpec with SparkTestBase with T
         fsUtils.fs.create(new Path(dataPath1, "1.dat")).close()
         fsUtils.fs.create(new Path(dataPath2, "2.dat")).close()
 
-        val persistence = new MetastorePersistenceRaw(tempDir, infoDateColumn, infoDateFormat)
+        val persistence = new MetastorePersistenceRaw(tempDir, infoDateColumn, infoDateFormat, None)
 
         val actual = persistence.loadTable(Some(infoDate), Some(infoDateTo)).orderBy("path").collect().map(_.getString(0))
 
@@ -87,7 +88,7 @@ class MetastorePersistenceRawSuite extends AnyWordSpec with SparkTestBase with T
         fsUtils.fs.create(new Path(dataPath, "1.dat")).close()
         fsUtils.fs.create(new Path(dataPath, "2.dat")).close()
 
-        val persistence = new MetastorePersistenceRaw(tempDir, infoDateColumn, infoDateFormat)
+        val persistence = new MetastorePersistenceRaw(tempDir, infoDateColumn, infoDateFormat, None)
 
         val actual = persistence.loadTable(Some(infoDate.plusDays(1)), Some(infoDate)).orderBy("path").collect().map(_.getString(0))
 
@@ -102,7 +103,7 @@ class MetastorePersistenceRawSuite extends AnyWordSpec with SparkTestBase with T
 
         fsUtils.createDirectoryRecursive(dataPath)
 
-        val persistence = new MetastorePersistenceRaw(tempDir, infoDateColumn, infoDateFormat)
+        val persistence = new MetastorePersistenceRaw(tempDir, infoDateColumn, infoDateFormat, None)
 
         assertThrows[IllegalArgumentException] {
           persistence.loadTable(None, None)
@@ -114,7 +115,7 @@ class MetastorePersistenceRawSuite extends AnyWordSpec with SparkTestBase with T
   "saveTable()" should {
     "do nothing on an empty dataset" in {
       withTempDirectory("metastore_raw") { tempDir =>
-        val persistence = new MetastorePersistenceRaw(tempDir, infoDateColumn, infoDateFormat)
+        val persistence = new MetastorePersistenceRaw(tempDir, infoDateColumn, infoDateFormat, None)
 
         persistence.saveTable(infoDate, Seq.empty[String].toDF("path"), None)
 
@@ -135,7 +136,7 @@ class MetastorePersistenceRawSuite extends AnyWordSpec with SparkTestBase with T
 
         val files = Seq(file1, file2).map(_.toUri.toString).toDF("path")
 
-        val persistence = new MetastorePersistenceRaw(tempDir, infoDateColumn, infoDateFormat)
+        val persistence = new MetastorePersistenceRaw(tempDir, infoDateColumn, infoDateFormat, Some(SaveMode.Overwrite))
 
         persistence.saveTable(infoDate, files, None)
 
@@ -160,7 +161,7 @@ class MetastorePersistenceRawSuite extends AnyWordSpec with SparkTestBase with T
 
         fsUtils.fs.create(new Path(partitionPath, "3.dat"))
 
-        val persistence = new MetastorePersistenceRaw(tempDir, infoDateColumn, infoDateFormat)
+        val persistence = new MetastorePersistenceRaw(tempDir, infoDateColumn, infoDateFormat, None)
 
         persistence.saveTable(infoDate, files, None)
 
@@ -174,10 +175,40 @@ class MetastorePersistenceRawSuite extends AnyWordSpec with SparkTestBase with T
       }
     }
 
+    "append existing partition directory if it exists for save mode append" in {
+      withTempDirectory("metastore_raw") { tempDir =>
+        val file1 = new Path(tempDir, "1.dat")
+        val file2 = new Path(tempDir, "2.dat")
+
+        fsUtils.fs.create(file1).close()
+        fsUtils.fs.create(file2).close()
+
+        val files = Seq(file1, file2).map(_.toUri.toString).toDF("path")
+
+        val partitionPath = new Path(tempDir, s"$infoDateColumn=$infoDate")
+
+        fsUtils.fs.create(new Path(partitionPath, "3.dat"))
+
+        val persistence = new MetastorePersistenceRaw(tempDir, infoDateColumn, infoDateFormat, Some(SaveMode.Append))
+
+        persistence.saveTable(infoDate, files, None)
+
+        assert(fsUtils.exists(new Path(partitionPath, "1.dat")))
+        assert(fsUtils.exists(new Path(partitionPath, "2.dat")))
+        assert(fsUtils.exists(new Path(partitionPath, "3.dat")))
+
+        persistence.saveTable(infoDate, Seq.empty[String].toDF("path"), None)
+
+        assert(fsUtils.exists(new Path(partitionPath, "1.dat")))
+        assert(fsUtils.exists(new Path(partitionPath, "2.dat")))
+        assert(fsUtils.exists(new Path(partitionPath, "3.dat")))
+      }
+    }
+
     "throw an exception if the dataframe does not contain the required column" in {
       withTempDirectory("metastore_raw") { tempDir =>
 
-        val persistence = new MetastorePersistenceRaw(tempDir, infoDateColumn, infoDateFormat)
+        val persistence = new MetastorePersistenceRaw(tempDir, infoDateColumn, infoDateFormat, None)
 
         assertThrows[IllegalArgumentException] {
           persistence.saveTable(infoDate, spark.emptyDataFrame, None)
@@ -197,7 +228,7 @@ class MetastorePersistenceRawSuite extends AnyWordSpec with SparkTestBase with T
         fsUtils.writeFile(file1, "123")
         fsUtils.writeFile(file2, "4567")
 
-        val persistence = new MetastorePersistenceRaw(tempDir, infoDateColumn, infoDateFormat)
+        val persistence = new MetastorePersistenceRaw(tempDir, infoDateColumn, infoDateFormat, None)
 
         val stats = persistence.getStats(infoDate)
 
@@ -209,7 +240,7 @@ class MetastorePersistenceRawSuite extends AnyWordSpec with SparkTestBase with T
 
   "createOrUpdateHiveTable" should {
     "throw the unsupported exception" in {
-      val persistence = new MetastorePersistenceRaw("", "", "")
+      val persistence = new MetastorePersistenceRaw("", "", "", None)
       assertThrows[UnsupportedOperationException] {
         persistence.createOrUpdateHiveTable(infoDate, "table", null, null)
       }
@@ -218,7 +249,7 @@ class MetastorePersistenceRawSuite extends AnyWordSpec with SparkTestBase with T
 
   "repairHiveTable" should {
     "throw the unsupported exception" in {
-      val persistence = new MetastorePersistenceRaw("", "", "")
+      val persistence = new MetastorePersistenceRaw("", "", "", None)
       assertThrows[UnsupportedOperationException] {
         persistence.repairHiveTable("table", null, null)
       }
diff --git a/pramen/core/src/test/scala/za/co/absa/pramen/core/metastore/persistence/MetastorePersistenceSuite.scala b/pramen/core/src/test/scala/za/co/absa/pramen/core/metastore/persistence/MetastorePersistenceSuite.scala
index f8460d684..f35234590 100644
--- a/pramen/core/src/test/scala/za/co/absa/pramen/core/metastore/persistence/MetastorePersistenceSuite.scala
+++ b/pramen/core/src/test/scala/za/co/absa/pramen/core/metastore/persistence/MetastorePersistenceSuite.scala
@@ -18,7 +18,7 @@ package za.co.absa.pramen.core.metastore.persistence
 
 import com.typesafe.config.ConfigFactory
 import org.apache.spark.sql.functions.{col, lit}
-import org.apache.spark.sql.{AnalysisException, DataFrame}
+import org.apache.spark.sql.{AnalysisException, DataFrame, SaveMode}
 import org.scalatest.Assertion
 import org.scalatest.wordspec.AnyWordSpec
 import za.co.absa.pramen.api.{CachePolicy, DataFormat, Query}
@@ -268,6 +268,49 @@ class MetastorePersistenceSuite extends AnyWordSpec with SparkTestBase with Temp
     compareText(actual, expected)
   }
 
+  def testAppendPartition(mtp: MetastorePersistence): Assertion = {
+    val expected =
+      """[ {
+        |  "a" : "A",
+        |  "b" : 1,
+        |  "p" : 1,
+        |  "info_date" : "2021-10-12"
+        |}, {
+        |  "a" : "A",
+        |  "b" : 1,
+        |  "p" : 2,
+        |  "info_date" : "2021-10-12"
+        |}, {
+        |  "a" : "B",
+        |  "b" : 2,
+        |  "p" : 1,
+        |  "info_date" : "2021-10-12"
+        |}, {
+        |  "a" : "B",
+        |  "b" : 2,
+        |  "p" : 2,
+        |  "info_date" : "2021-10-12"
+        |}, {
+        |  "a" : "C",
+        |  "b" : 3,
+        |  "p" : 1,
+        |  "info_date" : "2021-10-12"
+        |}, {
+        |  "a" : "C",
+        |  "b" : 3,
+        |  "p" : 2,
+        |  "info_date" : "2021-10-12"
+        |} ]""".stripMargin
+    mtp.saveTable(infoDate, getDf.withColumn("p", lit(1)), None)
+    mtp.saveTable(infoDate, getDf.withColumn("p", lit(2)), None)
+
+    val df = mtp.loadTable(Some(infoDate), Some(infoDate))
+
+    val actual = SparkUtils.dataFrameToJson(df.orderBy("a", "p"))
+
+    compareText(actual, expected)
+  }
+
   def testSchemaMerge(mtp: MetastorePersistence): Assertion = {
     val expected =
       """[ {
@@ -377,6 +420,12 @@ class MetastorePersistenceSuite extends AnyWordSpec with SparkTestBase with Temp
         }
       }
 
+      "support partition appends" in {
+        withTempDirectory("mt_persist") { tempDir =>
+          testAppendPartition(getParquetMtPersistence(tempDir, saveModeOpt = Some(SaveMode.Append)))
+        }
+      }
+
       "support schema overwrites" in {
         withTempDirectory("mt_persist") { tempDir =>
           testSchemaMerge(getParquetMtPersistence(tempDir))
@@ -448,6 +497,12 @@ class MetastorePersistenceSuite extends AnyWordSpec with SparkTestBase with Temp
         }
       }
 
+      "support partition appends" in {
+        withTempDirectory("mt_persist") { tempDir =>
+          testAppendPartition(getDeltaMtPersistence(tempDir, saveModeOpt = Some(SaveMode.Append)))
+        }
+      }
+
       "fail of schema change" in {
         withTempDirectory("mt_persist") { tempDir =>
           val ex = intercept[AnalysisException] {
@@ -524,12 +579,14 @@ class MetastorePersistenceSuite extends AnyWordSpec with SparkTestBase with Temp
 
   def getParquetMtPersistence(tempDir: String,
                               recordsPerPartition: Option[Long] = None,
-                              pathSuffix: String = "parquet"): MetastorePersistence = {
+                              pathSuffix: String = "parquet",
+                              saveModeOpt: Option[SaveMode] = None): MetastorePersistence = {
 
     val mt = MetaTableFactory.getDummyMetaTable(name = "table1",
       format = DataFormat.Parquet(s"$tempDir/$pathSuffix", recordsPerPartition),
       infoDateColumn = infoDateColumn,
-      infoDateFormat = infoDateFormat
+      infoDateFormat = infoDateFormat,
+      saveModeOpt = saveModeOpt
     )
 
     MetastorePersistence.fromMetaTable(mt, null)
@@ -538,11 +595,13 @@ class MetastorePersistenceSuite extends AnyWordSpec with SparkTestBase with Temp
   def getDeltaMtPersistence(tempDir: String,
                             recordsPerPartition: Option[Long] = None,
                             pathSuffix: String = "delta",
+                            saveModeOpt: Option[SaveMode] = None,
                             writeOptions: Map[String, String] = Map.empty[String, String]): MetastorePersistence = {
     val mt = MetaTableFactory.getDummyMetaTable(name = "table1",
       format = DataFormat.Delta(Query.Path(s"$tempDir/$pathSuffix"), recordsPerPartition),
       infoDateColumn = infoDateColumn,
       infoDateFormat = infoDateFormat,
+      saveModeOpt = saveModeOpt,
       writeOptions = writeOptions
     )
 
diff --git a/pramen/core/src/test/scala/za/co/absa/pramen/core/mocks/MetaTableFactory.scala b/pramen/core/src/test/scala/za/co/absa/pramen/core/mocks/MetaTableFactory.scala
index 218c81f03..2151794ea 100644
--- a/pramen/core/src/test/scala/za/co/absa/pramen/core/mocks/MetaTableFactory.scala
+++ b/pramen/core/src/test/scala/za/co/absa/pramen/core/mocks/MetaTableFactory.scala
@@ -16,6 +16,7 @@
 
 package za.co.absa.pramen.core.mocks
 
+import org.apache.spark.sql.SaveMode
 import za.co.absa.pramen.api.DataFormat
 import za.co.absa.pramen.core.metastore.model.{HiveConfig, MetaTable}
 
@@ -35,6 +36,7 @@ object MetaTableFactory {
                         infoDateStart: LocalDate = LocalDate.of(2020, 1, 31),
                         trackDays: Int = 0,
                         trackDaysExplicitlySet: Boolean = false,
+                        saveModeOpt: Option[SaveMode] = None,
                         readOptions: Map[String, String] = Map.empty[String, String],
                         writeOptions: Map[String, String] = Map.empty[String, String],
                         sparkConfig: Map[String, String] = Map.empty[String, String]
@@ -52,6 +54,7 @@ object MetaTableFactory {
       infoDateStart,
       trackDays,
       trackDaysExplicitlySet,
+      saveModeOpt,
       readOptions,
       writeOptions,
       sparkConfig)