Repository: AbsaOSS/pramen (language: python)

Issue:
Add support of delta table for bookkeeper
Add support of delta table for bookkeeper. Could be used to maintain metastore in databricks.

https://github.com/AbsaOSS/pramen/blob/c0bc31219fdcbfe9398cf6a2f0e414278712ec55/pramen/core/src/main/scala/za/co/absa/pramen/core/bookkeeper/Bookkeeper.scala#L110


Discussion / hints:
We had such an implementation, actually 😄. It was quite slow, so we removed it. But it was a couple of years ago. Maybe now is a good time to revive it.
Found classes for Delta. I want to restore them in next Pramen version. Just, currently, it uses Delta paths, not tables. This is because it requires several different subpaths to save different stuff. Do you want to add Delta Lake table support or a path is fine?

Fix patch:
diff --git a/README.md b/README.md
index fc8508d0f..3913485c6 100644
--- a/README.md
+++ b/README.md
@@ -2243,6 +2243,74 @@ You specify:
 
 Let's take a look at an example based on the Enceladus sink.
 
+## Bookeeping
+
+In order to support auto-recovery from failures, schema tracking and all other nice features, Pramen requires to use a database
+or a storage for keeping the state of the pipeline.
+
+### PostgreSQL database (recommended)
+This is highly recommended way of storing bookkeeping data since it is the most efficient and feature rich.
+
+Configuration:
+```hocon
+pramen {
+  bookkeeping.enabled = "true"
+  
+  bookkeeping.jdbc {
+    driver = "org.postgresql.Driver"
+    url = "jdbc:postgresql://host:5433/pramen"
+    user = "username"
+    password = "password"
+  }
+}
+```
+
+### MongoDb database
+Here is how you can use a MongoDB database for storing bookkeeping information:
+
+```hocon
+pramen {
+  bookkeeping.enabled = "true"
+
+  bookkeeping.mongodb {
+    connection.string = "mongodb://aaabbb"
+    database = "mydb"
+  }
+}
+```
+
+### Hadoop (CSV+JSON)
+This is less recommended way, and is quite slow. But the advantage is that you don't need a database.
+
+```hocon
+pramen.bookkeeping {
+  enabled = "true"
+  location = "hdfs://path"
+}
+```
+
+### Delta Lake
+This requires Delta Lake format support from the cluster you are running pipelines at.
+
+You can use wither a path:
+```hocon
+pramen.bookkeeping {
+  enabled = "true"
+  hadoop.format = "delta"
+  location = "s3://path"
+}
+```
+
+or a set of managed tables:
+```hocon
+pramen.bookkeeping {
+  enabled = "true"
+  hadoop.format = "delta"
+  delta.database = "my_db"  # Optional. 'default' will be used if not speified
+  delta.table.prefix = "bk_"
+}
+```
+
 #### Enceladus ingestion pipelines for the Data Lake
 Pramen can help with ingesting data for data lake pipelines of [Enceladus](https://github.com/AbsaOSS/enceladus).
 A special sink (`EnceladusSink`) is used to save data to Enceladus' raw folder.
diff --git a/pramen/core/src/main/scala/za/co/absa/pramen/core/app/config/BookkeeperConfig.scala b/pramen/core/src/main/scala/za/co/absa/pramen/core/app/config/BookkeeperConfig.scala
index 08eb899bf..b0ab6910a 100644
--- a/pramen/core/src/main/scala/za/co/absa/pramen/core/app/config/BookkeeperConfig.scala
+++ b/pramen/core/src/main/scala/za/co/absa/pramen/core/app/config/BookkeeperConfig.scala
@@ -26,7 +26,9 @@ case class BookkeeperConfig(
                               bookkeepingHadoopFormat: HadoopFormat,
                               bookkeepingConnectionString: Option[String],
                               bookkeepingDbName: Option[String],
-                              bookkeepingJdbcConfig: Option[JdbcConfig]
+                              bookkeepingJdbcConfig: Option[JdbcConfig],
+                              deltaDatabase: Option[String],
+                              deltaTablePrefix: Option[String]
                             )
 
 object BookkeeperConfig {
@@ -36,6 +38,8 @@ object BookkeeperConfig {
   val BOOKKEEPING_HADOOP_FORMAT = "pramen.bookkeeping.hadoop.format"
   val BOOKKEEPING_CONNECTION_STRING = "pramen.bookkeeping.mongodb.connection.string"
   val BOOKKEEPING_DB_NAME = "pramen.bookkeeping.mongodb.database"
+  val BOOKKEEPING_DELTA_DB_NAME = "pramen.bookkeeping.delta.database"
+  val BOOKKEEPING_DELTA_TABLE_PREFIX = "pramen.bookkeeping.delta.table.prefix"
 
   def fromConfig(conf: Config): BookkeeperConfig = {
     val bookkeepingEnabled = conf.getBoolean(BOOKKEEPING_ENABLED)
@@ -44,14 +48,23 @@ object BookkeeperConfig {
     val bookkeepingConnectionString = ConfigUtils.getOptionString(conf, BOOKKEEPING_CONNECTION_STRING)
     val bookkeepingDbName = ConfigUtils.getOptionString(conf, BOOKKEEPING_DB_NAME)
     val bookkeepingJdbcConfig = JdbcConfig.loadOption(conf.getConfig(BOOKKEEPING_PARENT), BOOKKEEPING_PARENT)
+    val deltaDatabase = ConfigUtils.getOptionString(conf, BOOKKEEPING_DELTA_DB_NAME)
+    val deltaTablePrefix = ConfigUtils.getOptionString(conf, BOOKKEEPING_DELTA_TABLE_PREFIX)
 
-    if (bookkeepingEnabled && bookkeepingConnectionString.isEmpty && bookkeepingLocation.isEmpty && bookkeepingJdbcConfig.isEmpty) {
-      throw new RuntimeException(s"One of the following should be defined: $BOOKKEEPING_PARENT.jdbc.url, $BOOKKEEPING_CONNECTION_STRING or $BOOKKEEPING_LOCATION" +
-        s" when bookkeeping is enabled. You can disable bookkeeping by setting $BOOKKEEPING_ENABLED = false.")
-    }
+    if (bookkeepingEnabled && bookkeepingJdbcConfig.isEmpty && bookkeepingHadoopFormat == HadoopFormat.Delta) {
+      if (bookkeepingLocation.isEmpty && deltaTablePrefix.isEmpty) {
+        throw new RuntimeException(s"In order to use Delta Lake for bookkeeping, either $BOOKKEEPING_LOCATION or $BOOKKEEPING_DELTA_TABLE_PREFIX must be defined. " +
+        s"Preferably $BOOKKEEPING_DELTA_DB_NAME should be defined as well for managed Delta Lake tables.")
+      }
+    } else {
+      if (bookkeepingEnabled && bookkeepingConnectionString.isEmpty && bookkeepingLocation.isEmpty && bookkeepingJdbcConfig.isEmpty) {
+        throw new RuntimeException(s"One of the following should be defined: $BOOKKEEPING_PARENT.jdbc.url, $BOOKKEEPING_CONNECTION_STRING or $BOOKKEEPING_LOCATION" +
+          s" when bookkeeping is enabled. You can disable bookkeeping by setting $BOOKKEEPING_ENABLED = false.")
+      }
 
-    if (bookkeepingConnectionString.isDefined && bookkeepingDbName.isEmpty) {
-      throw new RuntimeException(s"Database name is not defined. Please, define $BOOKKEEPING_DB_NAME.")
+      if (bookkeepingConnectionString.isDefined && bookkeepingDbName.isEmpty) {
+        throw new RuntimeException(s"Database name is not defined. Please, define $BOOKKEEPING_DB_NAME.")
+      }
     }
 
     BookkeeperConfig(
@@ -60,7 +73,9 @@ object BookkeeperConfig {
       bookkeepingHadoopFormat,
       bookkeepingConnectionString,
       bookkeepingDbName,
-      bookkeepingJdbcConfig
+      bookkeepingJdbcConfig,
+      deltaDatabase,
+      deltaTablePrefix
     )
   }
 }
\ No newline at end of file
diff --git a/pramen/core/src/main/scala/za/co/absa/pramen/core/app/config/HadoopFormat.scala b/pramen/core/src/main/scala/za/co/absa/pramen/core/app/config/HadoopFormat.scala
index b27350085..174992c30 100644
--- a/pramen/core/src/main/scala/za/co/absa/pramen/core/app/config/HadoopFormat.scala
+++ b/pramen/core/src/main/scala/za/co/absa/pramen/core/app/config/HadoopFormat.scala
@@ -20,9 +20,11 @@ trait HadoopFormat
 
 case object HadoopFormat {
   case object Text extends HadoopFormat
+  case object Delta extends HadoopFormat
 
   def apply(format: String): HadoopFormat = format.toLowerCase match {
     case "text" => Text
+    case "delta" => Delta
     case _ => throw new IllegalArgumentException(s"Unknown Hadoop format: $format")
   }
 }
diff --git a/pramen/core/src/main/scala/za/co/absa/pramen/core/bookkeeper/Bookkeeper.scala b/pramen/core/src/main/scala/za/co/absa/pramen/core/bookkeeper/Bookkeeper.scala
index 8d11ce5c0..933cc4b70 100644
--- a/pramen/core/src/main/scala/za/co/absa/pramen/core/bookkeeper/Bookkeeper.scala
+++ b/pramen/core/src/main/scala/za/co/absa/pramen/core/bookkeeper/Bookkeeper.scala
@@ -87,8 +87,14 @@ object Bookkeeper {
             log.info(s"Using MongoDB for lock management.")
             new TokenLockFactoryMongoDb(connection)
           case None =>
-            log.info(s"Using HadoopFS for lock management.")
-            new TokenLockFactoryHadoop(spark.sparkContext.hadoopConfiguration, bookkeepingConfig.bookkeepingLocation.get + "/locks")
+            bookkeepingConfig.bookkeepingLocation match {
+              case Some(path) =>
+                log.info(s"Using HadoopFS for lock management at $path/locks")
+                new TokenLockFactoryHadoopPath(spark.sparkContext.hadoopConfiguration, path + "/locks")
+              case None =>
+                log.warn(s"Locking is DISABLED.")
+                new TokenLockFactoryAllow
+            }
         }
       }
     } else {
@@ -109,8 +115,20 @@ object Bookkeeper {
         case None =>
           bookkeepingConfig.bookkeepingHadoopFormat match {
             case HadoopFormat.Text =>
-              log.info(s"Using Hadoop (CSV for records, JSON for schemas) for bookkeeping.")
-              new BookkeeperText(bookkeepingConfig.bookkeepingLocation.get)
+              val path = bookkeepingConfig.bookkeepingLocation.get
+              log.info(s"Using Hadoop (CSV for records, JSON for schemas) for bookkeeping at $path")
+              new BookkeeperText(path)
+            case HadoopFormat.Delta =>
+              bookkeepingConfig.deltaTablePrefix match {
+                case Some(tablePrefix) =>
+                  val fullTableName = BookkeeperDeltaTable.getFullTableName(bookkeepingConfig.deltaDatabase, tablePrefix, "*")
+                  log.info(s"Using Delta Lake managed table '$fullTableName' for bookkeeping.")
+                  new BookkeeperDeltaTable(bookkeepingConfig.deltaDatabase, tablePrefix)
+                case None =>
+                  val path = bookkeepingConfig.bookkeepingLocation.get
+                  log.info(s"Using Delta Lake for bookkeeping at $path")
+                  new BookkeeperDeltaPath(path)
+              }
           }
       }
     }
@@ -127,8 +145,24 @@ object Bookkeeper {
           log.info(s"Using MongoDB to keep journal of executed jobs.")
           new JournalMongoDb(connection)
         case None =>
-          log.info(s"Using HadoopFS to keep journal of executed jobs.")
-          new JournalHadoop(bookkeepingConfig.bookkeepingLocation.get + "/journal")
+          bookkeepingConfig.bookkeepingHadoopFormat match {
+            case HadoopFormat.Text =>
+              val path = bookkeepingConfig.bookkeepingLocation.get + "/journal"
+              log.info(s"Using HadoopFS to keep journal of executed jobs at $path")
+              new JournalHadoopCsv(path)
+            case HadoopFormat.Delta =>
+              bookkeepingConfig.deltaTablePrefix match {
+                case Some(tablePrefix) =>
+                  val fullTableName = JournalHadoopDeltaTable.getFullTableName(bookkeepingConfig.deltaDatabase, tablePrefix)
+                  log.info(s"Using Delta Lake managed table '$fullTableName' for the journal.")
+                  new JournalHadoopDeltaTable(bookkeepingConfig.deltaDatabase, tablePrefix)
+                case None =>
+                  val path = bookkeepingConfig.bookkeepingLocation.get + "/journal"
+                  log.info(s"Using Delta Lake for the journal at $path")
+                  new JournalHadoopDeltaPath(path)
+              }
+          }
+
       }
     }
 
diff --git a/pramen/core/src/main/scala/za/co/absa/pramen/core/bookkeeper/BookkeeperDeltaBase.scala b/pramen/core/src/main/scala/za/co/absa/pramen/core/bookkeeper/BookkeeperDeltaBase.scala
new file mode 100644
index 000000000..72c06b188
--- /dev/null
+++ b/pramen/core/src/main/scala/za/co/absa/pramen/core/bookkeeper/BookkeeperDeltaBase.scala
@@ -0,0 +1,124 @@
+/*
+ * 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.bookkeeper
+
+import org.apache.spark.sql.functions._
+import org.apache.spark.sql.types.StructType
+import org.apache.spark.sql.{Column, Dataset}
+import za.co.absa.pramen.core.bookkeeper.model.TableSchemaJson
+import za.co.absa.pramen.core.model.{DataChunk, TableSchema}
+
+import java.time.LocalDate
+import scala.reflect.ClassTag
+import scala.reflect.runtime.universe
+
+abstract class BookkeeperDeltaBase extends BookkeeperHadoop {
+
+  def getBkDf(filter: Column): Dataset[DataChunk]
+
+  def saveRecordCountDelta(dataChunk: DataChunk): Unit
+
+  def getSchemasDeltaDf: Dataset[TableSchemaJson]
+
+  def saveSchemaDelta(schemas: TableSchema): Unit
+
+  def writeEmptyDataset[T <: Product : universe.TypeTag : ClassTag](pathOrTable: String): Unit
+
+  final override val bookkeepingEnabled: Boolean = true
+
+  final override def getLatestProcessedDateFromStorage(tableName: String, until: Option[LocalDate]): Option[LocalDate] = {
+    val filter = until match {
+      case Some(endDate) =>
+        val endDateStr = getDateStr(endDate)
+        col("tableName") === tableName && col("infoDate") <= endDateStr
+      case None =>
+        col("tableName") === tableName
+    }
+
+    val chunks = getBkData(filter)
+
+    if (chunks.isEmpty) {
+      None
+    } else {
+      val chunk = chunks.maxBy(_.infoDateEnd)
+      Option(LocalDate.parse(chunk.infoDateEnd, DataChunk.dateFormatter))
+    }
+  }
+
+  final override def getLatestDataChunkFromStorage(table: String, dateBegin: LocalDate, dateEnd: LocalDate): Option[DataChunk] = {
+    getDataChunks(table, dateBegin, dateEnd).lastOption
+  }
+
+  final override def getDataChunksFromStorage(tableName: String, infoDateBegin: LocalDate, infoDateEnd: LocalDate): Seq[DataChunk] = {
+    val infoDateFilter = getFilter(tableName, Option(infoDateBegin), Option(infoDateEnd))
+
+    getBkData(infoDateFilter)
+  }
+
+  final def getDataChunksCountFromStorage(table: String, dateBegin: Option[LocalDate], dateEnd: Option[LocalDate]): Long = {
+    getBkDf(getFilter(table, dateBegin, dateEnd)).count()
+  }
+
+  final private[pramen] override def saveRecordCountToStorage(table: String,
+                                                        infoDate: LocalDate,
+                                                        infoDateBegin: LocalDate,
+                                                        infoDateEnd: LocalDate,
+                                                        inputRecordCount: Long,
+                                                        outputRecordCount: Long,
+                                                        jobStarted: Long,
+                                                        jobFinished: Long): Unit = {
+    val dateStr = getDateStr(infoDate)
+    val dateBeginStr = getDateStr(infoDateBegin)
+    val dateEndStr = getDateStr(infoDateEnd)
+
+    val chunk = DataChunk(table, dateStr, dateBeginStr, dateEndStr, inputRecordCount, outputRecordCount, jobStarted, jobFinished)
+
+    saveRecordCountDelta(chunk)
+  }
+
+  final override def getLatestSchema(table: String, until: LocalDate): Option[(StructType, LocalDate)] = {
+    val filter = getFilter(table, None, Option(until))
+
+    val df = getSchemasDeltaDf
+
+    val tableSchemaOpt = df.filter(filter)
+      .orderBy(col("infoDate").desc, col("updatedTimestamp").desc)
+      .take(1)
+      .headOption
+
+    tableSchemaOpt.flatMap(s => {
+      TableSchema.toSchemaAndDate(TableSchema(s.tableName, s.infoDate, s.schemaJson))
+    })
+  }
+
+  private[pramen] override def saveSchema(table: String, infoDate: LocalDate, schema: StructType): Unit = {
+    val tableSchema = TableSchema(table, infoDate.toString, schema.json)
+
+    saveSchemaDelta(tableSchema)
+  }
+
+  private[core] def getBkData(filter: Column): Seq[DataChunk] = {
+    getBkDf(filter)
+      .collect()
+      .groupBy(v => (v.tableName, v.infoDate))
+      .map { case (_, listChunks) =>
+        listChunks.maxBy(c => c.jobFinished)
+      }
+      .toArray[DataChunk]
+      .sortBy(_.infoDate)
+  }
+}
diff --git a/pramen/core/src/main/scala/za/co/absa/pramen/core/bookkeeper/BookkeeperDeltaPath.scala b/pramen/core/src/main/scala/za/co/absa/pramen/core/bookkeeper/BookkeeperDeltaPath.scala
new file mode 100644
index 000000000..952418f93
--- /dev/null
+++ b/pramen/core/src/main/scala/za/co/absa/pramen/core/bookkeeper/BookkeeperDeltaPath.scala
@@ -0,0 +1,124 @@
+/*
+ * 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.bookkeeper
+
+import org.apache.hadoop.fs.Path
+import org.apache.spark.sql.functions._
+import org.apache.spark.sql.{Column, Dataset, SaveMode, SparkSession}
+import za.co.absa.pramen.core.bookkeeper.model.TableSchemaJson
+import za.co.absa.pramen.core.model.{DataChunk, TableSchema}
+import za.co.absa.pramen.core.utils.FsUtils
+
+import java.time.Instant
+import scala.reflect.ClassTag
+import scala.reflect.runtime.universe
+
+object BookkeeperDeltaPath {
+  val bookkeepingRootPath = "bk"
+  val recordsDirName = "records_delta"
+  val schemasDirName = "schemas_delta"
+  val locksDirName = "locks"
+}
+
+class BookkeeperDeltaPath(bookkeepingPath: String)(implicit spark: SparkSession) extends BookkeeperDeltaBase {
+  import BookkeeperDeltaPath._
+  import spark.implicits._
+
+  private val fsUtils = new FsUtils(spark.sparkContext.hadoopConfiguration, bookkeepingPath)
+  private val bkPath = new Path(bookkeepingPath, bookkeepingRootPath)
+  private val recordsPath = new Path(bkPath, recordsDirName)
+  private val schemasPath = new Path(bkPath, schemasDirName)
+  private val locksPath = new Path(bookkeepingPath, locksDirName)
+
+  init()
+
+  override def getBkDf(filter: Column): Dataset[DataChunk] = {
+    val df = spark
+      .read
+      .format("delta")
+      .load(recordsPath.toUri.toString)
+
+    df.filter(filter)
+      .orderBy(col("jobFinished"))
+      .as[DataChunk]
+  }
+
+  override def saveRecordCountDelta(dataChunk: DataChunk): Unit = {
+    val df = Seq(dataChunk).toDF()
+
+    df.write
+      .mode(SaveMode.Append)
+      .format("delta")
+      .option("mergeSchema", "true")
+      .save(recordsPath.toUri.toString)
+  }
+
+  override def getSchemasDeltaDf: Dataset[TableSchemaJson] = {
+    spark
+      .read
+      .format("delta")
+      .load(schemasPath.toUri.toString)
+      .as[TableSchemaJson]
+  }
+
+  override def saveSchemaDelta(schema: TableSchema): Unit = {
+    val df = Seq(
+      TableSchemaJson(schema.tableName, schema.infoDate, schema.schemaJson, Instant.now().toEpochMilli)
+    ).toDF()
+
+    df.write
+      .mode(SaveMode.Append)
+      .format("delta")
+      .option("mergeSchema", "true")
+      .save(schemasPath.toUri.toString)
+  }
+
+  override def writeEmptyDataset[T <: Product : universe.TypeTag : ClassTag](pathOrTable: String): Unit = {
+    val df = Seq.empty[T].toDS
+
+    df.write
+      .mode(SaveMode.Overwrite)
+      .format("delta")
+      .save(pathOrTable)
+  }
+
+  private def init(): Unit = {
+    initRecordsDirectory(recordsPath)
+    initSchemasDirectory(schemasPath)
+    initDirectory(locksPath)
+  }
+
+  private def initDirectory(path: Path): Unit = {
+    if (!fsUtils.exists(path)) {
+      fsUtils.createDirectoryRecursive(path)
+    }
+  }
+
+  private def initRecordsDirectory(path: Path): Unit = {
+    if (!fsUtils.exists(path)) {
+      fsUtils.createDirectoryRecursive(path)
+      writeEmptyDataset[DataChunk](path.toUri.toString)
+    }
+  }
+
+  private def initSchemasDirectory(path: Path): Unit = {
+    if (!fsUtils.exists(path)) {
+      fsUtils.createDirectoryRecursive(path)
+      writeEmptyDataset[TableSchemaJson](path.toUri.toString)
+    }
+  }
+}
diff --git a/pramen/core/src/main/scala/za/co/absa/pramen/core/bookkeeper/BookkeeperDeltaTable.scala b/pramen/core/src/main/scala/za/co/absa/pramen/core/bookkeeper/BookkeeperDeltaTable.scala
new file mode 100644
index 000000000..a5b0ca25f
--- /dev/null
+++ b/pramen/core/src/main/scala/za/co/absa/pramen/core/bookkeeper/BookkeeperDeltaTable.scala
@@ -0,0 +1,107 @@
+/*
+ * 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.bookkeeper
+
+import org.apache.spark.sql.functions.col
+import org.apache.spark.sql.{Column, Dataset, SaveMode, SparkSession}
+import za.co.absa.pramen.core.bookkeeper.model.TableSchemaJson
+import za.co.absa.pramen.core.model.{DataChunk, TableSchema}
+
+import java.time.Instant
+import scala.reflect.ClassTag
+import scala.reflect.runtime.universe
+
+object BookkeeperDeltaTable {
+  val recordsTable = "bookkeeping"
+  val schemasTable = "schemas"
+
+  def getFullTableName(databaseOpt: Option[String], tablePrefix: String, tableName: String): String = {
+    databaseOpt match {
+      case Some(db) => s"$db.$tablePrefix$tableName"
+      case None     => s"$tablePrefix$tableName"
+    }
+  }
+}
+
+class BookkeeperDeltaTable(database: Option[String],
+                           tablePrefix: String)
+                          (implicit spark: SparkSession) extends BookkeeperDeltaBase {
+  import BookkeeperDeltaTable._
+  import spark.implicits._
+
+  private val recordsFullTableName = getFullTableName(database, tablePrefix, recordsTable)
+  private val schemasFullTableName = getFullTableName(database, tablePrefix, schemasTable)
+
+  init()
+
+  override def getBkDf(filter: Column): Dataset[DataChunk] = {
+    val df = spark.table(recordsFullTableName).as[DataChunk]
+
+    df.filter(filter)
+      .orderBy(col("jobFinished"))
+      .as[DataChunk]
+  }
+
+  override def saveRecordCountDelta(dataChunks: DataChunk): Unit = {
+    val df = Seq(dataChunks).toDF()
+
+    df.write
+      .mode(SaveMode.Append)
+      .option("mergeSchema", "true")
+      .saveAsTable(recordsFullTableName)
+  }
+
+  override def getSchemasDeltaDf: Dataset[TableSchemaJson] = {
+    spark.table(schemasFullTableName).as[TableSchemaJson]
+  }
+
+  override def saveSchemaDelta(schema: TableSchema): Unit = {
+    val df = Seq(
+      TableSchemaJson(schema.tableName, schema.infoDate, schema.schemaJson, Instant.now().toEpochMilli)
+    ).toDF()
+
+    df.write
+      .mode(SaveMode.Append)
+      .option("mergeSchema", "true")
+      .saveAsTable(schemasFullTableName)
+  }
+
+  override def writeEmptyDataset[T <: Product : universe.TypeTag : ClassTag](pathOrTable: String): Unit = {
+    val df = Seq.empty[T].toDS
+
+    df.write
+      .mode(SaveMode.Overwrite)
+      .saveAsTable(pathOrTable)
+  }
+
+  def init(): Unit = {
+    initRecordsDirectory()
+    initSchemasDirectory()
+  }
+
+  private def initRecordsDirectory(): Unit = {
+    if (!spark.catalog.tableExists(recordsFullTableName)) {
+      writeEmptyDataset[DataChunk](recordsFullTableName)
+    }
+  }
+
+  private def initSchemasDirectory(): Unit = {
+    if (!spark.catalog.tableExists(schemasTable)) {
+      writeEmptyDataset[TableSchemaJson](schemasTable)
+    }
+  }
+}
diff --git a/pramen/core/src/main/scala/za/co/absa/pramen/core/bookkeeper/BookkeeperText.scala b/pramen/core/src/main/scala/za/co/absa/pramen/core/bookkeeper/BookkeeperText.scala
index aaf727cb6..2d91ec34d 100644
--- a/pramen/core/src/main/scala/za/co/absa/pramen/core/bookkeeper/BookkeeperText.scala
+++ b/pramen/core/src/main/scala/za/co/absa/pramen/core/bookkeeper/BookkeeperText.scala
@@ -21,7 +21,7 @@ import org.apache.spark.sql.functions._
 import org.apache.spark.sql.types.StructType
 import org.apache.spark.sql.{Column, Dataset, SparkSession}
 import za.co.absa.pramen.core.bookkeeper.model.TableSchemaJson
-import za.co.absa.pramen.core.lock.TokenLockHadoop
+import za.co.absa.pramen.core.lock.TokenLockHadoopPath
 import za.co.absa.pramen.core.model.{DataChunk, TableSchema}
 import za.co.absa.pramen.core.utils.{CsvUtils, FsUtils, JsonUtils, SparkUtils}
 
@@ -99,7 +99,7 @@ class BookkeeperText(bookkeepingPath: String)(implicit spark: SparkSession) exte
                                                         outputRecordCount: Long,
                                                         jobStarted: Long,
                                                         jobFinished: Long): Unit = {
-    val lock: TokenLockHadoop = getLock
+    val lock: TokenLockHadoopPath = getLock
 
     try {
       val dateStr = getDateStr(infoDate)
@@ -115,8 +115,8 @@ class BookkeeperText(bookkeepingPath: String)(implicit spark: SparkSession) exte
     }
   }
 
-  private def getLock: TokenLockHadoop = {
-    val lock = new TokenLockHadoop("bookkeeping",
+  private def getLock: TokenLockHadoopPath = {
+    val lock = new TokenLockHadoopPath("bookkeeping",
       spark.sparkContext.hadoopConfiguration,
       locksPath.toUri.toString,
       30L)
@@ -193,7 +193,7 @@ class BookkeeperText(bookkeepingPath: String)(implicit spark: SparkSession) exte
   private[pramen] override def saveSchema(table: String, infoDate: LocalDate, schema: StructType): Unit = {
     val tableSchema = TableSchemaJson(table, infoDate.toString, schema.json, Instant.now().toEpochMilli)
 
-    val lock: TokenLockHadoop = getLock
+    val lock: TokenLockHadoopPath = getLock
 
     try {
       val json = JsonUtils.asJson(tableSchema)
diff --git a/pramen/core/src/main/scala/za/co/absa/pramen/core/journal/JournalHadoop.scala b/pramen/core/src/main/scala/za/co/absa/pramen/core/journal/JournalHadoopCsv.scala
similarity index 94%
rename from pramen/core/src/main/scala/za/co/absa/pramen/core/journal/JournalHadoop.scala
rename to pramen/core/src/main/scala/za/co/absa/pramen/core/journal/JournalHadoopCsv.scala
index 45488735d..f31509ca9 100644
--- a/pramen/core/src/main/scala/za/co/absa/pramen/core/journal/JournalHadoop.scala
+++ b/pramen/core/src/main/scala/za/co/absa/pramen/core/journal/JournalHadoopCsv.scala
@@ -26,18 +26,18 @@ import za.co.absa.pramen.core.utils.{CsvUtils, FsUtils, SparkUtils}
 
 import java.time.{Instant, LocalDate}
 
-object JournalHadoop {
+object JournalHadoopCsv {
   val journalFileName = "journal.csv"
   val separator = '|'
 }
 
-class JournalHadoop(journalPath: String)
+class JournalHadoopCsv(journalPath: String)
                    (implicit spark: SparkSession) extends Journal {
 
-  import JournalHadoop._
+  import JournalHadoopCsv._
 
-  private val hdfsConfig: Configuration = spark.sparkContext.hadoopConfiguration
-  private val fsUtils = new FsUtils(hdfsConfig, journalPath)
+  private val hadoopConfig: Configuration = spark.sparkContext.hadoopConfiguration
+  private val fsUtils = new FsUtils(hadoopConfig, journalPath)
   private val journalFilePath = new Path(journalPath, journalFileName)
 
   private val headers = CsvUtils.getHeaders[TaskCompletedCsv](separator)
diff --git a/pramen/core/src/main/scala/za/co/absa/pramen/core/journal/JournalHadoopDeltaPath.scala b/pramen/core/src/main/scala/za/co/absa/pramen/core/journal/JournalHadoopDeltaPath.scala
new file mode 100644
index 000000000..6493029a8
--- /dev/null
+++ b/pramen/core/src/main/scala/za/co/absa/pramen/core/journal/JournalHadoopDeltaPath.scala
@@ -0,0 +1,67 @@
+/*
+ * 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.journal
+
+import org.apache.hadoop.conf.Configuration
+import org.apache.hadoop.fs.Path
+import org.apache.spark.sql.functions._
+import org.apache.spark.sql.{SaveMode, SparkSession}
+import za.co.absa.pramen.core.journal.model.TaskCompleted
+import za.co.absa.pramen.core.utils.FsUtils
+
+import java.time.Instant
+
+class JournalHadoopDeltaPath(journalPath: String)
+                            (implicit spark: SparkSession) extends Journal {
+  import spark.implicits._
+
+  private val hadoopConfig: Configuration = spark.sparkContext.hadoopConfiguration
+  private val fsUtils = new FsUtils(hadoopConfig, journalPath)
+
+  override def addEntry(entry: TaskCompleted): Unit = {
+    val recordDf = Seq(entry).toDS().toDF()
+
+    if (spark.version.split('.').head.toInt < 3) {
+      throw new IllegalArgumentException("Delta Lake for bookkeeping is only available in Spark 3+")
+    }
+
+    recordDf
+      .write
+      .mode(SaveMode.Append)
+      .option("format", "delta")
+      .option("mergeSchema", "true")
+      .save(journalPath)
+  }
+
+  override def getEntries(from: Instant, to: Instant): Seq[TaskCompleted] = {
+    import spark.implicits._
+
+    if (!fsUtils.exists(new Path(journalPath))) {
+      return Seq.empty[TaskCompleted]
+    }
+
+    val df = spark
+      .read
+      .option("format", "delta")
+      .load(journalPath)
+
+    df.filter(col("finishedAt") >= from.getEpochSecond && col("finishedAt") <= to.getEpochSecond)
+      .orderBy(col("finishedAt"))
+      .as[TaskCompleted]
+      .collect()
+  }
+}
diff --git a/pramen/core/src/main/scala/za/co/absa/pramen/core/journal/JournalHadoopDeltaTable.scala b/pramen/core/src/main/scala/za/co/absa/pramen/core/journal/JournalHadoopDeltaTable.scala
new file mode 100644
index 000000000..18e096671
--- /dev/null
+++ b/pramen/core/src/main/scala/za/co/absa/pramen/core/journal/JournalHadoopDeltaTable.scala
@@ -0,0 +1,70 @@
+/*
+ * 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.journal
+
+import org.apache.spark.sql.functions._
+import org.apache.spark.sql.{SaveMode, SparkSession}
+import za.co.absa.pramen.core.journal.model.TaskCompleted
+
+import java.time.Instant
+
+class JournalHadoopDeltaTable(database: Option[String],
+                              tablePrefix: String)
+                             (implicit spark: SparkSession) extends Journal {
+  import JournalHadoopDeltaTable._
+  import spark.implicits._
+
+  override def addEntry(entry: TaskCompleted): Unit = {
+    val recordDf = Seq(entry).toDS().toDF()
+
+    if (spark.version.split('.').head.toInt < 3) {
+      throw new IllegalArgumentException("Delta Lake for bookkeeping is only available in Spark 3+")
+    }
+
+    recordDf
+      .write
+      .mode(SaveMode.Append)
+      .option("format", "delta")
+      .option("mergeSchema", "true")
+      .saveAsTable(getFullTableName(database, tablePrefix))
+  }
+
+  override def getEntries(from: Instant, to: Instant): Seq[TaskCompleted] = {
+    import spark.implicits._
+
+    if (!spark.catalog.tableExists(getFullTableName(database, tablePrefix))) {
+      return Seq.empty[TaskCompleted]
+    }
+
+    val df = spark
+      .table(getFullTableName(database, tablePrefix))
+
+    df.filter(col("finishedAt") >= from.getEpochSecond && col("finishedAt") <= to.getEpochSecond)
+      .orderBy(col("finishedAt"))
+      .as[TaskCompleted]
+      .collect()
+  }
+}
+
+object JournalHadoopDeltaTable {
+  def getFullTableName(databaseOpt: Option[String], tablePrefix: String): String = {
+    databaseOpt match {
+      case Some(db) => s"$db.${tablePrefix}journal"
+      case None     => s"${tablePrefix}journal"
+    }
+  }
+}
\ No newline at end of file
diff --git a/pramen/core/src/main/scala/za/co/absa/pramen/core/lock/TokenLockFactoryHadoop.scala b/pramen/core/src/main/scala/za/co/absa/pramen/core/lock/TokenLockFactoryHadoopPath.scala
similarity index 81%
rename from pramen/core/src/main/scala/za/co/absa/pramen/core/lock/TokenLockFactoryHadoop.scala
rename to pramen/core/src/main/scala/za/co/absa/pramen/core/lock/TokenLockFactoryHadoopPath.scala
index 0c39f4f6f..7b36332b7 100644
--- a/pramen/core/src/main/scala/za/co/absa/pramen/core/lock/TokenLockFactoryHadoop.scala
+++ b/pramen/core/src/main/scala/za/co/absa/pramen/core/lock/TokenLockFactoryHadoopPath.scala
@@ -18,8 +18,8 @@ package za.co.absa.pramen.core.lock
 
 import org.apache.hadoop.conf.Configuration
 
-class TokenLockFactoryHadoop(hadoopConf: Configuration, lockLocation: String) extends TokenLockFactory {
+class TokenLockFactoryHadoopPath(hadoopConf: Configuration, lockLocation: String) extends TokenLockFactory {
   override def getLock(token: String): TokenLock = {
-    new TokenLockHadoop(token, hadoopConf, lockLocation)
+    new TokenLockHadoopPath(token, hadoopConf, lockLocation)
   }
 }
diff --git a/pramen/core/src/main/scala/za/co/absa/pramen/core/lock/TokenLockHadoop.scala b/pramen/core/src/main/scala/za/co/absa/pramen/core/lock/TokenLockHadoopPath.scala
similarity index 93%
rename from pramen/core/src/main/scala/za/co/absa/pramen/core/lock/TokenLockHadoop.scala
rename to pramen/core/src/main/scala/za/co/absa/pramen/core/lock/TokenLockHadoopPath.scala
index 5a3b041ac..853ed1c37 100644
--- a/pramen/core/src/main/scala/za/co/absa/pramen/core/lock/TokenLockHadoop.scala
+++ b/pramen/core/src/main/scala/za/co/absa/pramen/core/lock/TokenLockHadoopPath.scala
@@ -23,10 +23,10 @@ import za.co.absa.pramen.core.utils.{FsUtils, StringUtils}
 
 import java.time.Instant
 
-class TokenLockHadoop(token: String,
-                      hadoopConf: Configuration,
-                      locksPath: String,
-                      tokenExpireTimeSeconds: Long = 10L * 60L) extends TokenLock {
+class TokenLockHadoopPath(token: String,
+                          hadoopConf: Configuration,
+                          locksPath: String,
+                          tokenExpireTimeSeconds: Long = 10L * 60L) extends TokenLock {
   // Dependencies
   private val log = LoggerFactory.getLogger(this.getClass)
   private val fsUtils = new FsUtils(hadoopConf, locksPath)


Test patch:
diff --git a/pramen/core/src/test/scala/za/co/absa/pramen/core/BookkeepingConfigFactory.scala b/pramen/core/src/test/scala/za/co/absa/pramen/core/BookkeepingConfigFactory.scala
index aeb6debcf..34804699d 100644
--- a/pramen/core/src/test/scala/za/co/absa/pramen/core/BookkeepingConfigFactory.scala
+++ b/pramen/core/src/test/scala/za/co/absa/pramen/core/BookkeepingConfigFactory.scala
@@ -25,14 +25,18 @@ object BookkeepingConfigFactory {
                                 bookkeepingHadoopFormat: HadoopFormat = HadoopFormat.Text,
                                 bookkeepingConnectionString: Option[String] = None,
                                 bookkeepingDbName: Option[String] = None,
-                                bookkeepingJdbcConfig: Option[JdbcConfig] = None): BookkeeperConfig = {
+                                bookkeepingJdbcConfig: Option[JdbcConfig] = None,
+                                deltaDatabase: Option[String] = None,
+                                deltaTablePrefix: Option[String] = None): BookkeeperConfig = {
     BookkeeperConfig(
       bookkeepingEnabled,
       bookkeepingLocation,
       bookkeepingHadoopFormat,
       bookkeepingConnectionString,
       bookkeepingDbName,
-      bookkeepingJdbcConfig
+      bookkeepingJdbcConfig,
+      deltaDatabase,
+      deltaTablePrefix
     )
   }
 
diff --git a/pramen/core/src/test/scala/za/co/absa/pramen/core/app/config/BookkeeperConfigSuite.scala b/pramen/core/src/test/scala/za/co/absa/pramen/core/app/config/BookkeeperConfigSuite.scala
index e4bb2e874..83158c2d0 100644
--- a/pramen/core/src/test/scala/za/co/absa/pramen/core/app/config/BookkeeperConfigSuite.scala
+++ b/pramen/core/src/test/scala/za/co/absa/pramen/core/app/config/BookkeeperConfigSuite.scala
@@ -84,6 +84,47 @@ class BookkeeperConfigSuite extends AnyWordSpec {
       assert(bookkeeperConfig.bookkeepingLocation.contains("hdfs://dummy_path"))
     }
 
+    "deserialize the config properly for Hadoop Delta Path" in {
+      val configStr =
+        s"""pramen {
+           |  bookkeeping.enabled = true
+           |  bookkeeping.location = "hdfs://dummy_path"
+           |  bookkeeping.hadoop.format = "delta"
+           |}
+           |""".stripMargin
+
+      val config = ConfigFactory.parseString(configStr)
+        .withFallback(ConfigFactory.load())
+        .resolve()
+
+      val bookkeeperConfig = BookkeeperConfig.fromConfig(config)
+
+      assert(bookkeeperConfig.bookkeepingEnabled)
+      assert(bookkeeperConfig.bookkeepingLocation.contains("hdfs://dummy_path"))
+      assert(bookkeeperConfig.bookkeepingHadoopFormat == HadoopFormat.Delta)
+    }
+
+    "deserialize the config properly for Hadoop Delta Table" in {
+      val configStr =
+        s"""pramen {
+           |  bookkeeping.enabled = true
+           |  bookkeeping.hadoop.format = "delta"
+           |  bookkeeping.delta.table.prefix = "tbl_"
+           |}
+           |""".stripMargin
+
+      val config = ConfigFactory.parseString(configStr)
+        .withFallback(ConfigFactory.load())
+        .resolve()
+
+      val bookkeeperConfig = BookkeeperConfig.fromConfig(config)
+
+      assert(bookkeeperConfig.bookkeepingEnabled)
+      assert(bookkeeperConfig.bookkeepingLocation.isEmpty)
+      assert(bookkeeperConfig.bookkeepingHadoopFormat == HadoopFormat.Delta)
+      assert(bookkeeperConfig.deltaTablePrefix.contains("tbl_"))
+    }
+
     "deserialize the config properly for MongoDB" in {
       val configStr =
         s"""pramen {
diff --git a/pramen/core/src/test/scala/za/co/absa/pramen/core/tests/bookkeeper/BookkeeperCommonSuite.scala b/pramen/core/src/test/scala/za/co/absa/pramen/core/tests/bookkeeper/BookkeeperCommonSuite.scala
index e7d874351..5cfdc25b5 100644
--- a/pramen/core/src/test/scala/za/co/absa/pramen/core/tests/bookkeeper/BookkeeperCommonSuite.scala
+++ b/pramen/core/src/test/scala/za/co/absa/pramen/core/tests/bookkeeper/BookkeeperCommonSuite.scala
@@ -18,7 +18,7 @@ package za.co.absa.pramen.core.tests.bookkeeper
 
 import org.apache.spark.sql.types._
 import org.scalatest.wordspec.AnyWordSpec
-import za.co.absa.pramen.core.bookkeeper.Bookkeeper
+import za.co.absa.pramen.core.bookkeeper.{Bookkeeper, BookkeeperDeltaTable}
 import za.co.absa.pramen.core.model.DataChunk
 
 import java.time.LocalDate
@@ -190,6 +190,7 @@ class BookkeeperCommonSuite extends AnyWordSpec {
         val bk = getBookkeeper()
 
         bk.saveSchema("table", infoDate2, schema1)
+        Thread.sleep(10)
         bk.saveSchema("table", infoDate2, schema2)
 
         val actualSchema = bk.getLatestSchema("table", infoDate2)
@@ -235,8 +236,10 @@ class BookkeeperCommonSuite extends AnyWordSpec {
     "Multiple bookkeepers" should {
       "share the state" in {
         val bk = getBookkeeper()
-        val bk2 = getBookkeeper()
-        val bk3 = getBookkeeper()
+
+        // A workaround for BookkeeperDeltaTable which outputs to different tables in unit tests
+        val bk2 = if (bk.isInstanceOf[BookkeeperDeltaTable]) bk else getBookkeeper()
+        val bk3 = if (bk.isInstanceOf[BookkeeperDeltaTable]) bk else getBookkeeper()
 
         bk.setRecordCount("table", infoDate1, infoDate1, infoDate1, 100, 10, 1597318833, 1597318837, isTableTransient = false)
         bk.setRecordCount("table", infoDate1, infoDate1, infoDate1, 200, 20, 1597318838, 1597318839, isTableTransient = false)
diff --git a/pramen/core/src/test/scala/za/co/absa/pramen/core/tests/bookkeeper/BookkeeperDeltaPathLongSuite.scala b/pramen/core/src/test/scala/za/co/absa/pramen/core/tests/bookkeeper/BookkeeperDeltaPathLongSuite.scala
new file mode 100644
index 000000000..f7cb07561
--- /dev/null
+++ b/pramen/core/src/test/scala/za/co/absa/pramen/core/tests/bookkeeper/BookkeeperDeltaPathLongSuite.scala
@@ -0,0 +1,57 @@
+/*
+ * 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.tests.bookkeeper
+
+import org.apache.hadoop.fs.Path
+import org.scalatest.BeforeAndAfter
+import za.co.absa.pramen.core.base.SparkTestBase
+import za.co.absa.pramen.core.bookkeeper.BookkeeperDeltaPath
+import za.co.absa.pramen.core.fixtures.TempDirFixture
+import za.co.absa.pramen.core.utils.FsUtils
+
+class BookkeeperDeltaPathLongSuite extends BookkeeperCommonSuite with SparkTestBase with BeforeAndAfter with TempDirFixture {
+  import za.co.absa.pramen.core.bookkeeper.BookkeeperDeltaPath._
+
+  private val fsUtils = new FsUtils(spark.sparkContext.hadoopConfiguration, "/tmp")
+
+  var tmpDir: String = _
+
+  before {
+    tmpDir = createTempDir("bkHadoopDeltaPathSuite")
+  }
+
+  after {
+    deleteDir(tmpDir)
+  }
+
+  def getBookkeeper: BookkeeperDeltaPath = {
+    new BookkeeperDeltaPath(tmpDir)
+  }
+
+  "BookkeeperHadoopDeltaPath" when {
+    "initialized" should {
+      "Initialize bookkeeping directory" in {
+        getBookkeeper
+
+        assert(fsUtils.exists(new Path(tmpDir, s"$bookkeepingRootPath/$recordsDirName")))
+        assert(fsUtils.exists(new Path(tmpDir, locksDirName)))
+      }
+    }
+
+    testBookKeeper(() => getBookkeeper)
+  }
+}
diff --git a/pramen/core/src/test/scala/za/co/absa/pramen/core/tests/bookkeeper/BookkeeperDeltaTableLongSuite.scala b/pramen/core/src/test/scala/za/co/absa/pramen/core/tests/bookkeeper/BookkeeperDeltaTableLongSuite.scala
new file mode 100644
index 000000000..7b3748c03
--- /dev/null
+++ b/pramen/core/src/test/scala/za/co/absa/pramen/core/tests/bookkeeper/BookkeeperDeltaTableLongSuite.scala
@@ -0,0 +1,60 @@
+/*
+ * 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.tests.bookkeeper
+
+import org.scalatest.BeforeAndAfterAll
+import za.co.absa.pramen.core.base.SparkTestBase
+import za.co.absa.pramen.core.bookkeeper.BookkeeperDeltaTable
+
+import java.io.File
+import scala.util.Random
+
+class BookkeeperDeltaTableLongSuite extends BookkeeperCommonSuite with SparkTestBase with BeforeAndAfterAll {
+  override protected def beforeAll(): Unit = {
+    super.beforeAll()
+    cleanUpWarehouse()
+  }
+
+  override protected def afterAll(): Unit = {
+    cleanUpWarehouse()
+    super.afterAll()
+  }
+
+  def getBookkeeper(prefix: String): BookkeeperDeltaTable = {
+    new BookkeeperDeltaTable(None, prefix)
+  }
+
+  "BookkeeperHadoopDeltaTable" when {
+    testBookKeeper { () =>
+      val rndInt = Math.abs(Random.nextInt())
+      getBookkeeper(s"tbl${rndInt}_")
+    }
+  }
+
+  private def cleanUpWarehouse(): Unit = {
+    val warehouseDir = new File("spark-warehouse")
+    if (warehouseDir.exists()) {
+      warehouseDir.listFiles().foreach(f => {
+        if (f.isDirectory) {
+          f.listFiles().foreach(ff => ff.delete())
+        }
+        f.delete()
+      })
+      warehouseDir.delete()
+    }
+  }
+}
diff --git a/pramen/core/src/test/scala/za/co/absa/pramen/core/tests/bookkeeper/BookkeeperSuite.scala b/pramen/core/src/test/scala/za/co/absa/pramen/core/tests/bookkeeper/BookkeeperSuite.scala
index 4f92e2db2..7fbbc505e 100644
--- a/pramen/core/src/test/scala/za/co/absa/pramen/core/tests/bookkeeper/BookkeeperSuite.scala
+++ b/pramen/core/src/test/scala/za/co/absa/pramen/core/tests/bookkeeper/BookkeeperSuite.scala
@@ -18,13 +18,13 @@ package za.co.absa.pramen.core.tests.bookkeeper
 
 import org.scalatest.wordspec.AnyWordSpec
 import org.scalatest.{BeforeAndAfter, BeforeAndAfterAll}
-import za.co.absa.pramen.core.RuntimeConfigFactory
+import za.co.absa.pramen.core.{BookkeepingConfigFactory, RuntimeConfigFactory}
 import za.co.absa.pramen.core.app.config.{BookkeeperConfig, HadoopFormat, RuntimeConfig}
 import za.co.absa.pramen.core.base.SparkTestBase
-import za.co.absa.pramen.core.bookkeeper.{Bookkeeper, BookkeeperJdbc, BookkeeperMongoDb, BookkeeperText}
+import za.co.absa.pramen.core.bookkeeper.{Bookkeeper, BookkeeperDeltaPath, BookkeeperJdbc, BookkeeperMongoDb, BookkeeperText}
 import za.co.absa.pramen.core.fixtures.{MongoDbFixture, RelationalDbFixture, TempDirFixture}
-import za.co.absa.pramen.core.journal.{JournalHadoop, JournalJdbc, JournalMongoDb}
-import za.co.absa.pramen.core.lock.{TokenLockFactoryHadoop, TokenLockFactoryJdbc, TokenLockFactoryMongoDb}
+import za.co.absa.pramen.core.journal.{JournalHadoopCsv, JournalHadoopDeltaPath, JournalJdbc, JournalMongoDb}
+import za.co.absa.pramen.core.lock.{TokenLockFactoryHadoopPath, TokenLockFactoryJdbc, TokenLockFactoryMongoDb}
 import za.co.absa.pramen.core.metadata.{MetadataManagerJdbc, MetadataManagerNull}
 import za.co.absa.pramen.core.rdb.PramenDb
 import za.co.absa.pramen.core.reader.model.JdbcConfig
@@ -63,13 +63,9 @@ class BookkeeperSuite extends AnyWordSpec
 
   "config factory" should {
     "build bookkeeper, token lock, journal, and closable object for JDBC" in {
-      val bookkeepingConfig = BookkeeperConfig(
+      val bookkeepingConfig = BookkeepingConfigFactory.getDummyBookkeepingConfig(
         bookkeepingEnabled = true,
-        None,
-        HadoopFormat.Text,
-        None,
-        None,
-        Some(jdbcConfig)
+        bookkeepingJdbcConfig = Some(jdbcConfig)
       )
 
       val (bk, tf, journal, metadataManager, closable) = Bookkeeper.fromConfig(bookkeepingConfig, runtimeConfig)
@@ -83,13 +79,10 @@ class BookkeeperSuite extends AnyWordSpec
 
     if (db != null) {
       "build bookkeeper, token lock, journal, and closable object for MongoDB" in {
-        val bookkeepingConfig = BookkeeperConfig(
+        val bookkeepingConfig = BookkeepingConfigFactory.getDummyBookkeepingConfig(
           bookkeepingEnabled = true,
-          None,
-          HadoopFormat.Text,
-          Some(connectionString),
-          Some(dbName),
-          None
+          bookkeepingConnectionString = Some(connectionString),
+          bookkeepingDbName = Some(dbName)
         )
 
         val (bk, tf, journal, metadataManager, closable) = Bookkeeper.fromConfig(bookkeepingConfig, runtimeConfig)
@@ -106,22 +99,36 @@ class BookkeeperSuite extends AnyWordSpec
       }
     }
 
-    "build bookkeeper, token lock, journal, and closable object for Hadoop" in {
+    "build bookkeeper, token lock, journal, and closable object for Hadoop CSV" in {
       withTempDirectory("bk_hadoop") { tempDir =>
-        val bookkeepingConfig = BookkeeperConfig(
+        val bookkeepingConfig = BookkeepingConfigFactory.getDummyBookkeepingConfig(
           bookkeepingEnabled = true,
-          Some(tempDir),
-          HadoopFormat.Text,
-          None,
-          None,
-          None
+          bookkeepingLocation = Some(tempDir)
         )
 
         val (bk, tf, journal, metadataManager, closable) = Bookkeeper.fromConfig(bookkeepingConfig, runtimeConfig)
 
         assert(bk.isInstanceOf[BookkeeperText])
-        assert(tf.isInstanceOf[TokenLockFactoryHadoop])
-        assert(journal.isInstanceOf[JournalHadoop])
+        assert(tf.isInstanceOf[TokenLockFactoryHadoopPath])
+        assert(journal.isInstanceOf[JournalHadoopCsv])
+        assert(metadataManager.isInstanceOf[MetadataManagerNull])
+        closable.close()
+      }
+    }
+
+    "build bookkeeper, token lock, journal, and closable object for Hadoop Delta Path" in {
+      withTempDirectory("bk_hadoop_delta_path") { tempDir =>
+        val bookkeepingConfig = BookkeepingConfigFactory.getDummyBookkeepingConfig(
+          bookkeepingEnabled = true,
+          bookkeepingLocation = Some(tempDir),
+          bookkeepingHadoopFormat = HadoopFormat.Delta
+        )
+
+        val (bk, tf, journal, metadataManager, closable) = Bookkeeper.fromConfig(bookkeepingConfig, runtimeConfig)
+
+        assert(bk.isInstanceOf[BookkeeperDeltaPath])
+        assert(tf.isInstanceOf[TokenLockFactoryHadoopPath])
+        assert(journal.isInstanceOf[JournalHadoopDeltaPath])
         assert(metadataManager.isInstanceOf[MetadataManagerNull])
         closable.close()
       }
diff --git a/pramen/core/src/test/scala/za/co/absa/pramen/core/tests/journal/JournalHadoopSuite.scala b/pramen/core/src/test/scala/za/co/absa/pramen/core/tests/journal/JournalHadoopCsvSuite.scala
similarity index 88%
rename from pramen/core/src/test/scala/za/co/absa/pramen/core/tests/journal/JournalHadoopSuite.scala
rename to pramen/core/src/test/scala/za/co/absa/pramen/core/tests/journal/JournalHadoopCsvSuite.scala
index 78aad9f7a..6e9ce7083 100644
--- a/pramen/core/src/test/scala/za/co/absa/pramen/core/tests/journal/JournalHadoopSuite.scala
+++ b/pramen/core/src/test/scala/za/co/absa/pramen/core/tests/journal/JournalHadoopCsvSuite.scala
@@ -21,12 +21,12 @@ import org.scalatest.BeforeAndAfterAll
 import org.scalatest.wordspec.AnyWordSpec
 import za.co.absa.pramen.core.base.SparkTestBase
 import za.co.absa.pramen.core.fixtures.TempDirFixture
-import za.co.absa.pramen.core.journal.{Journal, JournalHadoop}
+import za.co.absa.pramen.core.journal.{Journal, JournalHadoopCsv}
 import za.co.absa.pramen.core.utils.FsUtils
 
-class JournalHadoopSuite extends AnyWordSpec with SparkTestBase with BeforeAndAfterAll with TempDirFixture {
+class JournalHadoopCsvSuite extends AnyWordSpec with SparkTestBase with BeforeAndAfterAll with TempDirFixture {
   import TestCases._
-  import za.co.absa.pramen.core.journal.JournalHadoop._
+  import za.co.absa.pramen.core.journal.JournalHadoopCsv._
 
   private val fsUtils = new FsUtils(spark.sparkContext.hadoopConfiguration, "/tmp")
 
@@ -73,7 +73,7 @@ class JournalHadoopSuite extends AnyWordSpec with SparkTestBase with BeforeAndAf
   }
 
   private def getJournal(path: String): Journal = {
-    new JournalHadoop(path)
+    new JournalHadoopCsv(path)
   }
 
 }
diff --git a/pramen/core/src/test/scala/za/co/absa/pramen/core/tests/journal/JournalHadoopDeltaPathLongSuite.scala b/pramen/core/src/test/scala/za/co/absa/pramen/core/tests/journal/JournalHadoopDeltaPathLongSuite.scala
new file mode 100644
index 000000000..56e332131
--- /dev/null
+++ b/pramen/core/src/test/scala/za/co/absa/pramen/core/tests/journal/JournalHadoopDeltaPathLongSuite.scala
@@ -0,0 +1,78 @@
+/*
+ * 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.tests.journal
+
+import org.apache.hadoop.fs.Path
+import org.scalatest.BeforeAndAfterAll
+import org.scalatest.wordspec.AnyWordSpec
+import za.co.absa.pramen.core.base.SparkTestBase
+import za.co.absa.pramen.core.fixtures.TempDirFixture
+import za.co.absa.pramen.core.journal.{Journal, JournalHadoopDeltaPath}
+
+class JournalHadoopDeltaPathLongSuite extends AnyWordSpec with SparkTestBase with BeforeAndAfterAll with TempDirFixture {
+
+  import TestCases._
+
+  var tmpDir: String = _
+
+  override def beforeAll(): Unit = {
+    super.beforeAll()
+    tmpDir = createTempDir("journalSuite")
+  }
+
+  override def afterAll(): Unit = {
+    deleteDir(tmpDir)
+    super.afterAll()
+  }
+
+  "Journal" should {
+    "Make sure the journal works even with empty path" in {
+      val journal = getJournal(tmpDir)
+
+      assert(journal.getEntries(instant1, instant3).isEmpty)
+    }
+
+    "addEntry()" should {
+      "return Nil if there are no entries" in {
+        val journal = getJournal(tmpDir)
+
+        assert(journal.getEntries(instant1, instant3).isEmpty)
+      }
+
+      "return entries if there are entries" in {
+        if (spark.version.split('.').head.toInt >= 3) {
+          val journal = getJournal(tmpDir)
+
+          journal.addEntry(task1)
+          journal.addEntry(task2)
+          journal.addEntry(task3)
+
+
+          val entries = journal.getEntries(instant2, instant3).sortBy(_.informationDate.toString)
+
+          assert(entries.nonEmpty)
+          assert(entries == task2 :: task3 :: Nil)
+        }
+      }
+    }
+  }
+
+  private def getJournal(path: String): Journal = {
+    new JournalHadoopDeltaPath(new Path(path, "journal").toString)
+  }
+
+}
diff --git a/pramen/core/src/test/scala/za/co/absa/pramen/core/tests/journal/JournalHadoopDeltaTableSuite.scala b/pramen/core/src/test/scala/za/co/absa/pramen/core/tests/journal/JournalHadoopDeltaTableSuite.scala
new file mode 100644
index 000000000..0ef814087
--- /dev/null
+++ b/pramen/core/src/test/scala/za/co/absa/pramen/core/tests/journal/JournalHadoopDeltaTableSuite.scala
@@ -0,0 +1,88 @@
+/*
+ * 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.tests.journal
+
+import org.scalatest.BeforeAndAfterAll
+import org.scalatest.wordspec.AnyWordSpec
+import za.co.absa.pramen.core.base.SparkTestBase
+import za.co.absa.pramen.core.journal.{Journal, JournalHadoopDeltaTable}
+
+import java.io.File
+
+class JournalHadoopDeltaTableSuite extends AnyWordSpec with BeforeAndAfterAll with SparkTestBase {
+
+  import TestCases._
+
+  override def beforeAll(): Unit = {
+    super.beforeAll()
+    cleanUpWarehouse()
+  }
+
+  override def afterAll(): Unit = {
+    cleanUpWarehouse()
+    super.afterAll()
+  }
+
+  "Journal" should {
+    "Make sure the journal works even with empty path" in {
+      val journal = getJournal("tbl1_")
+
+      assert(journal.getEntries(instant1, instant3).isEmpty)
+    }
+
+    "addEntry()" should {
+      "return Nil if there are no entries" in {
+        val journal = getJournal("tbl2_")
+
+        assert(journal.getEntries(instant1, instant3).isEmpty)
+      }
+
+      "return entries if there are entries" in {
+        if (spark.version.split('.').head.toInt >= 3) {
+          val journal = getJournal("tbl3_")
+
+          journal.addEntry(task1)
+          journal.addEntry(task2)
+          journal.addEntry(task3)
+
+
+          val entries = journal.getEntries(instant2, instant3).sortBy(_.informationDate.toString)
+
+          assert(entries.nonEmpty)
+          assert(entries == task2 :: task3 :: Nil)
+        }
+      }
+    }
+  }
+
+  private def getJournal(prefix: String): Journal = {
+    new JournalHadoopDeltaTable(None, prefix)
+  }
+
+  private def cleanUpWarehouse(): Unit = {
+    val warehouseDir = new File("spark-warehouse")
+    if (warehouseDir.exists()) {
+      warehouseDir.listFiles().foreach(f => {
+        if (f.isDirectory) {
+          f.listFiles().foreach(ff => ff.delete())
+        }
+        f.delete()
+      })
+      warehouseDir.delete()
+    }
+  }
+}
diff --git a/pramen/core/src/test/scala/za/co/absa/pramen/core/tests/lock/TokenLockHadoopSuite.scala b/pramen/core/src/test/scala/za/co/absa/pramen/core/tests/lock/TokenLockHadoopPathSuite.scala
similarity index 76%
rename from pramen/core/src/test/scala/za/co/absa/pramen/core/tests/lock/TokenLockHadoopSuite.scala
rename to pramen/core/src/test/scala/za/co/absa/pramen/core/tests/lock/TokenLockHadoopPathSuite.scala
index 64b461dc7..b9aea9964 100644
--- a/pramen/core/src/test/scala/za/co/absa/pramen/core/tests/lock/TokenLockHadoopSuite.scala
+++ b/pramen/core/src/test/scala/za/co/absa/pramen/core/tests/lock/TokenLockHadoopPathSuite.scala
@@ -19,20 +19,20 @@ package za.co.absa.pramen.core.tests.lock
 import org.scalatest.wordspec.AnyWordSpec
 import za.co.absa.pramen.core.base.SparkTestBase
 import za.co.absa.pramen.core.fixtures.TempDirFixture
-import za.co.absa.pramen.core.lock.TokenLockHadoop
+import za.co.absa.pramen.core.lock.TokenLockHadoopPath
 
-class TokenLockHadoopSuite extends AnyWordSpec with SparkTestBase with TempDirFixture {
+class TokenLockHadoopPathSuite extends AnyWordSpec with SparkTestBase with TempDirFixture {
   private val hdfsConfig = spark.sparkContext.hadoopConfiguration
 
   "Token lock" should {
     "be able to acquire and release locks" in {
       withTempDirectory("simpleLock") { tempDir =>
-        val lock1 = new TokenLockHadoop("token1", hdfsConfig, tempDir)
+        val lock1 = new TokenLockHadoopPath("token1", hdfsConfig, tempDir)
 
         assert(lock1.tryAcquire())
         assert(!lock1.tryAcquire())
 
-        val lock2 = new TokenLockHadoop("token1", hdfsConfig, tempDir)
+        val lock2 = new TokenLockHadoopPath("token1", hdfsConfig, tempDir)
         assert(!lock2.tryAcquire())
 
         lock1.release()
@@ -46,8 +46,8 @@ class TokenLockHadoopSuite extends AnyWordSpec with SparkTestBase with TempDirFi
 
     "multiple token locks should not affect each other" in {
       withTempDirectory("simpleLock") { tempDir =>
-        val lock1 = new TokenLockHadoop("token1", hdfsConfig, tempDir)
-        val lock2 = new TokenLockHadoop("token2", hdfsConfig, tempDir)
+        val lock1 = new TokenLockHadoopPath("token1", hdfsConfig, tempDir)
+        val lock2 = new TokenLockHadoopPath("token2", hdfsConfig, tempDir)
 
         assert(lock1.tryAcquire())
         assert(lock2.tryAcquire())
@@ -67,8 +67,8 @@ class TokenLockHadoopSuite extends AnyWordSpec with SparkTestBase with TempDirFi
 
     "lock pramen should constantly update lock ticket" ignore {
       withTempDirectory("simpleLock") { tempDir =>
-        val lock1 = new TokenLockHadoop("token1", hdfsConfig, tempDir, 3L)
-        val lock2 = new TokenLockHadoop("token1", hdfsConfig, tempDir)
+        val lock1 = new TokenLockHadoopPath("token1", hdfsConfig, tempDir, 3L)
+        val lock2 = new TokenLockHadoopPath("token1", hdfsConfig, tempDir)
         assert(lock1.tryAcquire())
         Thread.sleep(4000)
         assert(!lock2.tryAcquire())