Repository: AbsaOSS/pramen (language: python)

Issue:
Allow metastore tables having `delta` format not be partitioned
## Background
Partitioning of Delta Lake tables might actually worsen the efficiency of reads, especially for small tables.
https://delta.io/blog/pros-cons-hive-style-partionining/
https://delta.io/blog/2023-06-03-delta-lake-z-order/

This feature is about adding a flag to make metastore tables not partitioned. The information date column should still be added, probably as the first column so that Z-order or Liquid Clustering can take effect.

## Feature
Allow metastore tables having `delta` format not be partitioned.

## Example
```hocon
pramen.metastore {
  tables = [
    {
      name = "table1"
      format = "delta"
      path = "s3://bucket/prefix/table1"
      partitioned = false
      zorder = [ "info_date", "id" ]
    }
  ]
}
```

## Proposed Solution
Solution Ideas 
1. Make sure the logic is the same when working with partitioned and non-partitioned delta tables
2. Put the information date column at the beginning for partitioned tables so that it is part of Z-order or Liquid Clustering
3. Make sure parallel writes to different info dates is not possible when the data is not partitioned by information date
4. Use Z-order by the infomration date column and by batchid column by default, but allow users to specify z-order in config.



Discussion / hints:

Fix patch:
diff --git a/README.md b/README.md
index 1e2956de7..e0d689049 100644
--- a/README.md
+++ b/README.md
@@ -341,7 +341,8 @@ pramen.metastore {
       # (Experimental) Save mode to use when writing to partitions.
       # Supported: overwrite (default), append
       #save.mode = append
-      
+
+      information.date.partition.by = true
       information.date.column = "INFORMATION_DATE"
       information.date.format = "yyyy-MM-dd"
 
@@ -373,6 +374,7 @@ Metastore table options:
 | `table`                            | Delta Lake table name (if Delta Lake tables are the underlying storage).                                                                                                 |
 | `cache.ploicy`                     | For `transient` format only. Cache policy defines how to store transient tables for the duration of the pipeline. Available options: `cache`, `no_cache`, `persist`.---- |
 | `records.per.partition`            | Number of records per partition (in order to avoid small files problem).                                                                                                 |
+| `information.date.partition.by`    | If `true` (default) the table will be partitioned by the information date. If `false`, the table won't be partitioned (supported for `delta` format only).               |
 | `information.date.column`          | Name of the column that contains the information date. *                                                                                                                 |
 | `information.date.format`          | Format of the information date used for partitioning (in Java format notation). *                                                                                        |
 | `information.date.start`           | The earliest date the table contains data for. *                                                                                                                         |
diff --git a/pramen/api/src/main/scala/za/co/absa/pramen/api/MetaTableDef.scala b/pramen/api/src/main/scala/za/co/absa/pramen/api/MetaTableDef.scala
index 832149ec0..7f648aeb2 100644
--- a/pramen/api/src/main/scala/za/co/absa/pramen/api/MetaTableDef.scala
+++ b/pramen/api/src/main/scala/za/co/absa/pramen/api/MetaTableDef.scala
@@ -21,17 +21,18 @@ import java.time.LocalDate
 /**
   * This is metatable details available to read from the metastore.
   *
-  * @param name            The name of the table.
-  * @param description     The description of the table.
-  * @param format          The format of the table.
-  * @param infoDateColumn  The name of the column that contains the information date (partitioned by).
-  * @param infoDateFormat  The format of the information date.
-  * @param batchIdColumn   The name of the column that contains the batch id.
-  * @param hiveTable       The name of the Hive table.
-  * @param hivePath        The path of the Hive table (if it differs from the path in the underlying format).
-  * @param infoDateStart   The start date of the information date.
-  * @param readOptions     The read options for the table.
-  * @param writeOptions    The write options for the table.
+  * @param name                The name of the table.
+  * @param description         The description of the table.
+  * @param format              The format of the table.
+  * @param infoDateColumn      The name of the column that contains the information date (partitioned by).
+  * @param infoDateFormat      The format of the information date.
+  * @param partitionByInfoDate Should the table be partitioned by the information date column.
+  * @param batchIdColumn       The name of the column that contains the batch id.
+  * @param hiveTable           The name of the Hive table.
+  * @param hivePath            The path of the Hive table (if it differs from the path in the underlying format).
+  * @param infoDateStart       The start date of the information date.
+  * @param readOptions         The read options for the table.
+  * @param writeOptions        The write options for the table.
   */
 case class MetaTableDef(
                          name: String,
@@ -39,6 +40,7 @@ case class MetaTableDef(
                          format: DataFormat,
                          infoDateColumn: String,
                          infoDateFormat: String,
+                         partitionByInfoDate: Boolean,
                          batchIdColumn: String,
                          hiveTable: Option[String],
                          hivePath: Option[String],
diff --git a/pramen/core/src/main/resources/reference.conf b/pramen/core/src/main/resources/reference.conf
index 4a7112bb1..fd97e654b 100644
--- a/pramen/core/src/main/resources/reference.conf
+++ b/pramen/core/src/main/resources/reference.conf
@@ -40,6 +40,7 @@ pramen {
   # Default information column date field used for the metastore. Sourced tables will be partitioned by this field.
   information.date.column = "pramen_info_date"
   information.date.format = "yyyy-MM-dd"
+  information.date.partition.by = true
 
   # You can set this option so that Pramen never writes to partitions older than the specified date
   #information.date.start = "2010-01-01"
diff --git a/pramen/core/src/main/scala/za/co/absa/pramen/core/app/config/InfoDateConfig.scala b/pramen/core/src/main/scala/za/co/absa/pramen/core/app/config/InfoDateConfig.scala
index 761c4a0da..e76205393 100644
--- a/pramen/core/src/main/scala/za/co/absa/pramen/core/app/config/InfoDateConfig.scala
+++ b/pramen/core/src/main/scala/za/co/absa/pramen/core/app/config/InfoDateConfig.scala
@@ -26,6 +26,7 @@ import java.time.format.DateTimeFormatter
 case class InfoDateConfig(
                            columnName: String,
                            dateFormat: String,
+                           partitionByInfoDate: Boolean,
                            startDate: LocalDate,
                            defaultTrackDays: Int,
                            defaultDelayDays: Int,
@@ -43,6 +44,7 @@ object InfoDateConfig {
 
   val INFORMATION_DATE_COLUMN_KEY = "pramen.information.date.column"
   val INFORMATION_DATE_FORMAT_KEY = "pramen.information.date.format"
+  val INFORMATION_DATE_PARTITION_BY_KEY = "pramen.information.date.partition.by"
 
   val INFORMATION_DATE_START_KEY = "pramen.information.date.start"
   val INFORMATION_DATE_START_DAYS_KEY = "pramen.information.date.max.days.behind"
@@ -63,6 +65,7 @@ object InfoDateConfig {
 
   def fromConfig(conf: Config): InfoDateConfig = {
     val dateFormat = conf.getString(INFORMATION_DATE_FORMAT_KEY)
+    val partitionByInfoDate = conf.getBoolean(INFORMATION_DATE_PARTITION_BY_KEY)
 
     val columnName = conf.getString(INFORMATION_DATE_COLUMN_KEY)
     val expressionDaily = conf.getString(INFORMATION_DATE_EXPRESSION_DAILY_KEY)
@@ -93,6 +96,7 @@ object InfoDateConfig {
 
     InfoDateConfig(columnName,
       dateFormat,
+      partitionByInfoDate,
       startDate,
       defaultTrackDays,
       defaultDelayDays,
diff --git a/pramen/core/src/main/scala/za/co/absa/pramen/core/config/InfoDateOverride.scala b/pramen/core/src/main/scala/za/co/absa/pramen/core/config/InfoDateOverride.scala
index d0ae2ed75..cccc3381a 100644
--- a/pramen/core/src/main/scala/za/co/absa/pramen/core/config/InfoDateOverride.scala
+++ b/pramen/core/src/main/scala/za/co/absa/pramen/core/config/InfoDateOverride.scala
@@ -25,6 +25,7 @@ import java.time.LocalDate
 case class InfoDateOverride(
                              columnName: Option[String],
                              dateFormat: Option[String],
+                             partitionByInfoDate: Option[Boolean],
                              expression: Option[String],
                              startDate: Option[LocalDate]
                            )
@@ -32,6 +33,7 @@ case class InfoDateOverride(
 object InfoDateOverride {
   val INFORMATION_DATE_COLUMN_KEY = "information.date.column"
   val INFORMATION_DATE_FORMAT_KEY = "information.date.format"
+  val INFORMATION_DATE_PARTITION_BY_KEY = "information.date.partition.by"
   val INFORMATION_DATE_EXPRESSION_KEY = "information.date.expression"
   val INFORMATION_DATE_START_KEY = "information.date.start"
   val INFORMATION_DATE_MAX_DAYS_BEHIND_KEY = "information.date.max.days.behind"
@@ -39,6 +41,7 @@ object InfoDateOverride {
   def fromConfig(conf: Config): InfoDateOverride = {
     val columnNameOpt = ConfigUtils.getOptionString(conf, INFORMATION_DATE_COLUMN_KEY)
     val dateFormatOpt = ConfigUtils.getOptionString(conf, INFORMATION_DATE_FORMAT_KEY)
+    val infoDatePartitionByOpt = ConfigUtils.getOptionBoolean(conf, INFORMATION_DATE_PARTITION_BY_KEY)
     val expressionOpt = ConfigUtils.getOptionString(conf, INFORMATION_DATE_EXPRESSION_KEY)
 
     val startDateOpt = ConfigUtils.getDateOpt(conf, INFORMATION_DATE_START_KEY, DEFAULT_DATE_FORMAT)
@@ -56,6 +59,6 @@ object InfoDateOverride {
         None
     }
 
-    InfoDateOverride(columnNameOpt, dateFormatOpt, expressionOpt, startDate)
+    InfoDateOverride(columnNameOpt, dateFormatOpt, infoDatePartitionByOpt, expressionOpt, startDate)
   }
 }
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 d8bcddc7a..f3b3c42f7 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
@@ -35,6 +35,7 @@ import scala.util.{Failure, Success, Try}
   * @param format                 The format of the table.
   * @param infoDateColumn         The name of the column that contains the information date (partitioned by).
   * @param infoDateFormat         The format of the information date.
+  * @param partitionByInfoDate    Should the table be partitioned by the information date column.
   * @param batchIdColumn          The name of the column that contains the batch id.
   * @param hiveConfig             The effective Hive configuration to use for Hive operations.
   * @param hiveTable              The name of the Hive table.
@@ -54,6 +55,7 @@ case class MetaTable(
                       format: DataFormat,
                       infoDateColumn: String,
                       infoDateFormat: String,
+                      partitionByInfoDate: Boolean,
                       batchIdColumn: String,
                       hiveConfig: HiveConfig,
                       hiveTable: Option[String],
@@ -91,6 +93,7 @@ object MetaTable {
   def fromConfig(conf: Config, infoDateConfig: InfoDateConfig, key: String): Seq[MetaTable] = {
     val defaultInfoDateColumnName = infoDateConfig.columnName
     val defaultInfoDateFormat = infoDateConfig.dateFormat
+    val defaultPartitionByInfoDate = infoDateConfig.partitionByInfoDate
     val defaultStartDate = infoDateConfig.startDate
     val defaultTrackDays = infoDateConfig.defaultTrackDays
     val defaultHiveConfig = HiveDefaultConfig.fromConfig(ConfigUtils.getOptionConfig(conf, DEFAULT_HIVE_CONFIG_PREFIX))
@@ -104,7 +107,7 @@ object MetaTable {
     }
 
     val metatables = tableConfigs
-      .map(tableConfig => fromConfigSingleEntity(tableConfig, conf, defaultInfoDateColumnName, defaultInfoDateFormat, defaultStartDate, defaultTrackDays, defaultHiveConfig, defaultPreferAddPartition, defaultBatchIdColumnName))
+      .map(tableConfig => fromConfigSingleEntity(tableConfig, conf, defaultInfoDateColumnName, defaultInfoDateFormat, defaultPartitionByInfoDate, defaultStartDate, defaultTrackDays, defaultHiveConfig, defaultPreferAddPartition, defaultBatchIdColumnName))
       .toSeq
 
     val duplicates = AlgorithmUtils.findDuplicates(metatables.map(_.name))
@@ -118,6 +121,7 @@ object MetaTable {
                              appConf: Config,
                              defaultInfoColumnName: String,
                              defaultInfoDateFormat: String,
+                             defaultPartitionByInfoDate: Boolean,
                              defaultStartDate: LocalDate,
                              defaultTrackDays: Int,
                              defaultHiveConfig: HiveDefaultConfig,
@@ -129,6 +133,7 @@ object MetaTable {
     val infoDateColumn = infoDateOverride.columnName.getOrElse(defaultInfoColumnName)
     val infoDateFormat = infoDateOverride.dateFormat.getOrElse(defaultInfoDateFormat)
     val infoDateExpressionOpt = infoDateOverride.expression
+    val partitionByInfoDate = infoDateOverride.partitionByInfoDate.getOrElse(defaultPartitionByInfoDate)
     val startDate = infoDateOverride.startDate.getOrElse(defaultStartDate)
     val trackDays = ConfigUtils.getOptionInt(conf, TRACK_DAYS_KEY).getOrElse(defaultTrackDays)
     val trackDaysExplicitlySet = conf.hasPath(TRACK_DAYS_KEY)
@@ -161,6 +166,7 @@ object MetaTable {
       format,
       infoDateColumn,
       infoDateFormat,
+      partitionByInfoDate,
       batchIdColumn,
       hiveConfig,
       hiveTable,
@@ -183,6 +189,7 @@ object MetaTable {
       table.format,
       table.infoDateColumn,
       table.infoDateFormat,
+      table.partitionByInfoDate,
       table.batchIdColumn,
       table.hiveTable,
       table.hivePath,
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 2feb18b66..86a0d4777 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
@@ -53,7 +53,7 @@ object MetastorePersistence {
         )
       case DataFormat.Delta(query, recordsPerPartition)  =>
         new MetastorePersistenceDelta(
-          query, metaTable.infoDateColumn, metaTable.infoDateFormat, metaTable.batchIdColumn, batchId, recordsPerPartition, saveModeOpt, metaTable.readOptions, metaTable.writeOptions
+          query, metaTable.infoDateColumn, metaTable.infoDateFormat, metaTable.batchIdColumn, batchId, metaTable.partitionByInfoDate, recordsPerPartition, saveModeOpt, metaTable.readOptions, metaTable.writeOptions
         )
       case DataFormat.Raw(path)                          =>
         new MetastorePersistenceRaw(path, metaTable.infoDateColumn, metaTable.infoDateFormat, saveModeOpt)
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 273b18b9c..faa54ee94 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
@@ -37,6 +37,7 @@ class MetastorePersistenceDelta(query: Query,
                                 infoDateFormat: String,
                                 batchIdColumn: String,
                                 batchId: Long,
+                                partitionByInfoDate: Boolean,
                                 recordsPerPartition: Option[Long],
                                 saveModeOpt: Option[SaveMode],
                                 readOptions: Map[String, String],
@@ -64,19 +65,17 @@ class MetastorePersistenceDelta(query: Query,
   override def saveTable(infoDate: LocalDate, df: DataFrame, numberOfRecordsEstimate: Option[Long]): MetaTableStats = {
     val infoDateStr = dateFormatter.format(infoDate)
 
-    val dfIn = df.withColumn(infoDateColumn, lit(infoDateStr).cast(DateType))
-
     val whereCondition = s"$infoDateColumn='$infoDateStr'"
 
-    val dfRepartitioned = if (recordsPerPartition.nonEmpty) {
+    val dfRepartitioned = if (partitionByInfoDate && recordsPerPartition.nonEmpty) {
       val recordCount = numberOfRecordsEstimate match {
         case Some(count) => count
-        case None => dfIn.count()
+        case None => df.count()
       }
 
-      applyRepartitioning(dfIn, recordCount)
+      applyRepartitioning(df, recordCount)
     } else {
-      dfIn
+      df
     }
 
     val saveMode = saveModeOpt.getOrElse(SaveMode.Overwrite)
@@ -87,19 +86,36 @@ class MetastorePersistenceDelta(query: Query,
     }
 
     if (log.isDebugEnabled) {
-      log.debug(s"Schema: ${dfIn.schema.treeString}")
+      log.debug(s"Schema: ${df.schema.treeString}")
       log.debug(s"Info date column: $infoDateColumn")
       log.debug(s"Info date: $infoDateStr")
     }
 
-    val writer = dfRepartitioned
-      .write
-      .format("delta")
-      .mode(saveMode)
-      .partitionBy(infoDateColumn)
-      .option("mergeSchema", "true")
-      .option("replaceWhere", s"$infoDateColumn='$infoDateStr'")
-      .options(writeOptions)
+    val writer = if (partitionByInfoDate) {
+      val dfIn = dfRepartitioned.withColumn(infoDateColumn, lit(infoDateStr).cast(DateType))
+
+      dfIn
+        .write
+        .format("delta")
+        .mode(saveMode)
+        .partitionBy(infoDateColumn)
+        .option("mergeSchema", "true")
+        .option("replaceWhere", s"$infoDateColumn='$infoDateStr'")
+        .options(writeOptions)
+    } else {
+      val dfIn = dfRepartitioned.drop(infoDateColumn)
+
+      dfIn.select(
+          lit(infoDateStr).cast(DateType).as(infoDateColumn) +: dfIn.columns.map(col): _*
+        )
+        .withColumn(infoDateColumn, col(infoDateColumn)) // Move the column to the front for index purposes
+        .write
+        .format("delta")
+        .mode(saveMode)
+        .option("mergeSchema", "true")
+        .option("replaceWhere", s"$infoDateColumn='$infoDateStr'")
+        .options(writeOptions)
+    }
 
     query match {
       case Query.Path(path)   =>
@@ -142,7 +158,7 @@ class MetastorePersistenceDelta(query: Query,
     val df = loadTable(Option(infoDate), Option(infoDate))
 
     val sizeOpt = query match {
-      case Query.Path(path)   =>
+      case Query.Path(path) if partitionByInfoDate  =>
         val fsUtils = new FsUtils(spark.sparkContext.hadoopConfiguration, path)
         val size = Try {
           // When 0 records is saved to a Delta directory, the directory is not created.
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 f76810f46..6da982287 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, None, readOptions, writeOptions, sparkConfig)
+    MetaTable(jobMetaTableName, "", DataFormat.Null(), "", "", partitionByInfoDate = false, "", 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/InfoDateConfigFactory.scala b/pramen/core/src/test/scala/za/co/absa/pramen/core/InfoDateConfigFactory.scala
index 731490497..a993379ff 100644
--- a/pramen/core/src/test/scala/za/co/absa/pramen/core/InfoDateConfigFactory.scala
+++ b/pramen/core/src/test/scala/za/co/absa/pramen/core/InfoDateConfigFactory.scala
@@ -23,6 +23,7 @@ import java.time.LocalDate
 object InfoDateConfigFactory {
   def getDummyInfoDateConfig(columnName: String = "pramen_info_date",
                              dateFormat: String = "yyyy-MM-dd",
+                             partitionByInfoDate: Boolean = true,
                              startDate: LocalDate = LocalDate.parse("2010-01-01"),
                              defaultTrackDays: Int = 0,
                              defaultDelayDays: Int = 0,
@@ -35,6 +36,7 @@ object InfoDateConfigFactory {
     InfoDateConfig(
       columnName,
       dateFormat,
+      partitionByInfoDate,
       startDate,
       defaultTrackDays,
       defaultDelayDays,
diff --git a/pramen/core/src/test/scala/za/co/absa/pramen/core/MetaTableDefFactory.scala b/pramen/core/src/test/scala/za/co/absa/pramen/core/MetaTableDefFactory.scala
index f863c68b4..23d35ceb5 100644
--- a/pramen/core/src/test/scala/za/co/absa/pramen/core/MetaTableDefFactory.scala
+++ b/pramen/core/src/test/scala/za/co/absa/pramen/core/MetaTableDefFactory.scala
@@ -25,9 +25,10 @@ object MetaTableDefFactory {
   def getDummyMetaTableDef(
                             name: String = "table",
                             description: String = "",
-                            format: DataFormat  =DataFormat.Null(),
+                            format: DataFormat = DataFormat.Null(),
                             infoDateColumn: String = "info_date",
                             infoDateFormat: String = "yyyy-MM-dd",
+                            partitionByInfoDate: Boolean = true,
                             batchIdColumn: String = "pramen_batchid",
                             hiveTable: Option[String] = None,
                             hivePath: Option[String] = None,
@@ -41,6 +42,7 @@ object MetaTableDefFactory {
       format,
       infoDateColumn,
       infoDateFormat,
+      partitionByInfoDate,
       batchIdColumn,
       hiveTable,
       hivePath,
diff --git a/pramen/core/src/test/scala/za/co/absa/pramen/core/app/config/InfoDateConfigSuite.scala b/pramen/core/src/test/scala/za/co/absa/pramen/core/app/config/InfoDateConfigSuite.scala
index 08ea30211..ffe0210d4 100644
--- a/pramen/core/src/test/scala/za/co/absa/pramen/core/app/config/InfoDateConfigSuite.scala
+++ b/pramen/core/src/test/scala/za/co/absa/pramen/core/app/config/InfoDateConfigSuite.scala
@@ -24,11 +24,12 @@ import java.time.LocalDate
 
 class InfoDateConfigSuite extends AnyWordSpec {
   "InfoDateConfig" should {
-    "deserialize the config properly when set" in {
+    " set" in {
       val configStr =
         s"""pramen {
            |  information.date.column = "dummy_col"
            |  information.date.format = "dummy_format"
+           |  information.date.partition.by = true
            |  information.date.start = "2020-01-31"
            |
            |  track.days = 4
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 15803ba92..016350017 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
@@ -159,7 +159,7 @@ class MetaTableSuite extends AnyWordSpec {
 
       val defaultHiveConfig = HiveDefaultConfig.getNullConfig
 
-      val metaTable = MetaTable.fromConfigSingleEntity(conf, conf, "INFO_DATE", "dd-MM-yyyy", LocalDate.parse("2020-01-31"), 0, defaultHiveConfig, defaultPreferAddPartition = true, "batchid")
+      val metaTable = MetaTable.fromConfigSingleEntity(conf, conf, "INFO_DATE", "dd-MM-yyyy", defaultPartitionByInfoDate = true, LocalDate.parse("2020-01-31"), 0, defaultHiveConfig, defaultPreferAddPartition = true, "batchid")
 
       assert(metaTable.name == "my_table")
       assert(metaTable.format.name == "delta")
@@ -212,7 +212,7 @@ class MetaTableSuite extends AnyWordSpec {
 
       val appConf = ConfigFactory.parseString("pramen.default.records.per.partition = 100")
 
-      val metaTable = MetaTable.fromConfigSingleEntity(conf, appConf, "INFO_DATE", "dd-MM-yyyy", LocalDate.parse("2020-01-31"), 1, defaultHiveConfig, defaultPreferAddPartition = true, "batchid")
+      val metaTable = MetaTable.fromConfigSingleEntity(conf, appConf, "INFO_DATE", "dd-MM-yyyy", defaultPartitionByInfoDate = true, LocalDate.parse("2020-01-31"), 1, defaultHiveConfig, defaultPreferAddPartition = true, "batchid")
 
       assert(metaTable.name == "my_table")
       assert(metaTable.hiveConfig.hiveApi == HiveApi.Sql)
@@ -278,7 +278,7 @@ class MetaTableSuite extends AnyWordSpec {
 
       val appConf = ConfigFactory.parseString("pramen.default.records.per.partition = 100")
 
-      val metaTable = MetaTable.fromConfigSingleEntity(conf, appConf, "INFO_DATE", "dd-MM-yyyy", LocalDate.parse("2020-01-31"), 1, defaultHiveConfig, defaultPreferAddPartition = true, "batchid")
+      val metaTable = MetaTable.fromConfigSingleEntity(conf, appConf, "INFO_DATE", "dd-MM-yyyy", defaultPartitionByInfoDate = true, LocalDate.parse("2020-01-31"), 1, defaultHiveConfig, defaultPreferAddPartition = true, "batchid")
 
       assert(metaTable.name == "my_table")
       assert(metaTable.hiveConfig.hiveApi == HiveApi.SparkCatalog)
@@ -304,7 +304,7 @@ class MetaTableSuite extends AnyWordSpec {
       val defaultHiveConfig = HiveDefaultConfig.getNullConfig
 
       val ex = intercept[IllegalArgumentException] {
-        MetaTable.fromConfigSingleEntity(conf, conf, "", "", LocalDate.parse("2020-01-31"), 0, defaultHiveConfig, defaultPreferAddPartition = true, "batchid")
+        MetaTable.fromConfigSingleEntity(conf, conf, "", "", defaultPartitionByInfoDate = true, LocalDate.parse("2020-01-31"), 0, defaultHiveConfig, defaultPreferAddPartition = true, "batchid")
       }
 
       assert(ex.getMessage.contains("Mandatory option missing: name"))
@@ -322,7 +322,7 @@ class MetaTableSuite extends AnyWordSpec {
       val defaultHiveConfig = HiveDefaultConfig.getNullConfig
 
       val ex = intercept[IllegalArgumentException] {
-        MetaTable.fromConfigSingleEntity(conf, conf, "", "", LocalDate.parse("2020-01-31"), 0, defaultHiveConfig, defaultPreferAddPartition = true, "batchid")
+        MetaTable.fromConfigSingleEntity(conf, conf, "", "", defaultPartitionByInfoDate = true, LocalDate.parse("2020-01-31"), 0, defaultHiveConfig, defaultPreferAddPartition = true, "batchid")
       }
 
       assert(ex.getMessage.contains("Unable to read data format from config for the metastore table: table1"))
@@ -340,7 +340,7 @@ class MetaTableSuite extends AnyWordSpec {
       val defaultHiveConfig = HiveDefaultConfig.getNullConfig
 
       val ex = intercept[IllegalArgumentException] {
-        MetaTable.fromConfigSingleEntity(conf, conf, "", "", LocalDate.parse("2020-01-31"), 0, defaultHiveConfig, defaultPreferAddPartition = true, "batchid")
+        MetaTable.fromConfigSingleEntity(conf, conf, "", "", defaultPartitionByInfoDate = true, LocalDate.parse("2020-01-31"), 0, defaultHiveConfig, defaultPreferAddPartition = true, "batchid")
       }
 
       assert(ex.getMessage.contains("Invalid or unsupported save mode: 'ignore' for table 'table1'."))
@@ -358,7 +358,7 @@ class MetaTableSuite extends AnyWordSpec {
       val defaultHiveConfig = HiveDefaultConfig.getNullConfig
 
       val ex = intercept[IllegalArgumentException] {
-        MetaTable.fromConfigSingleEntity(conf, conf, "", "", LocalDate.parse("2020-01-31"), 0, defaultHiveConfig, defaultPreferAddPartition = true, "batchid")
+        MetaTable.fromConfigSingleEntity(conf, conf, "", "", defaultPartitionByInfoDate = true, LocalDate.parse("2020-01-31"), 0, defaultHiveConfig, defaultPreferAddPartition = true, "batchid")
       }
 
       assert(ex.getMessage.contains("Invalid or unsupported save mode: 'test' for table 'table1'."))
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 007d9f566..d77da4e0b 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
@@ -148,10 +148,10 @@ class MetastorePersistenceSuite extends AnyWordSpec with SparkTestBase with Temp
     mtp.saveTable(infoDate, getDf.withColumn("p", lit(1)), None)
     mtp.saveTable(infoDate.plusDays(1), getDf.withColumn("p", lit(2)), None)
 
-    val df1 = mtp.loadTable(Some(infoDate), Some(infoDate))
-    val df2 = mtp.loadTable(Some(infoDate.plusDays(1)), None)
-    val df3 = mtp.loadTable(None, Some(infoDate))
-    val df4 = mtp.loadTable(None, None)
+    val df1 = mtp.loadTable(Some(infoDate), Some(infoDate)).select("a", "b", "p", "info_date")
+    val df2 = mtp.loadTable(Some(infoDate.plusDays(1)), None).select("a", "b", "p", "info_date")
+    val df3 = mtp.loadTable(None, Some(infoDate)).select("a", "b", "p", "info_date")
+    val df4 = mtp.loadTable(None, None).select("a", "b", "p", "info_date")
 
     val actual1 = SparkUtils.dataFrameToJson(df1.orderBy("a"))
     val actual2 = SparkUtils.dataFrameToJson(df2.orderBy("a"))
@@ -221,6 +221,15 @@ class MetastorePersistenceSuite extends AnyWordSpec with SparkTestBase with Temp
     assert(stats.dataSizeBytes.exists(_ > 0))
   }
 
+  def testStatsEmptyForNonPartitionedTables(mtp: MetastorePersistence): Assertion = {
+    mtp.saveTable(infoDate, getDf, Some(4))
+
+    val stats = mtp.getStats(infoDate, onlyForCurrentBatchId = false)
+
+    assert(stats.recordCount.contains(3))
+    assert(stats.dataSizeBytes.isEmpty)
+  }
+
   def testStatsAvailableForEmptyTable(mtp: MetastorePersistence): Assertion = {
     mtp.saveTable(infoDate, getDf.filter(col("b") > 10), Some(0))
 
@@ -262,6 +271,7 @@ class MetastorePersistenceSuite extends AnyWordSpec with SparkTestBase with Temp
     mtp.saveTable(infoDate, getDf.withColumn("p", lit(2)), None)
 
     val df = mtp.loadTable(Some(infoDate), Some(infoDate))
+      .select("a", "b", "p", "info_date")
 
     val actual = SparkUtils.dataFrameToJson(df.orderBy("a"))
 
@@ -305,6 +315,7 @@ class MetastorePersistenceSuite extends AnyWordSpec with SparkTestBase with Temp
     mtp.saveTable(infoDate, getDf.withColumn("p", lit(2)), None)
 
     val df = mtp.loadTable(Some(infoDate), Some(infoDate))
+      .select("a", "b", "p", "info_date")
 
     val actual = SparkUtils.dataFrameToJson(df.orderBy("a", "p"))
 
@@ -467,16 +478,30 @@ class MetastorePersistenceSuite extends AnyWordSpec with SparkTestBase with Temp
           testLoadExistingTable(getDeltaMtPersistence(tempDir))
         }
       }
-      "load table periods" in {
+      "load partitioned table periods" in {
         withTempDirectory("mt_persist") { tempDir =>
           testLoadTablePeriods(getDeltaMtPersistence(tempDir))
         }
       }
-      "load empty table if wrong period" in {
+      "load non-partitioned table periods" in {
+        if (spark.version.split('.').head.toInt >= 3) {
+          withTempDirectory("mt_persist") { tempDir =>
+            testLoadTablePeriods(getDeltaMtPersistence(tempDir, partitionByInfoDate = false))
+          }
+        }
+      }
+      "load empty table if wrong period partitioned" in {
         withTempDirectory("mt_persist") { tempDir =>
           testLoadEmptyTable(getDeltaMtPersistence(tempDir))
         }
       }
+      "load empty table if wrong period non partitioned" in {
+        if (spark.version.split('.').head.toInt >= 3) {
+          withTempDirectory("mt_persist") { tempDir =>
+            testLoadEmptyTable(getDeltaMtPersistence(tempDir, partitionByInfoDate = false))
+          }
+        }
+      }
       "throw an exception is the folder does not exist" in {
         withTempDirectory("mt_persist") { tempDir =>
           testNoData(getDeltaMtPersistence(tempDir))
@@ -485,24 +510,48 @@ class MetastorePersistenceSuite extends AnyWordSpec with SparkTestBase with Temp
     }
 
     "saveTable()" should {
-      "supports fixing the existing info date column" in {
+      "supports fixing the existing info date column partitioned" in {
         withTempDirectory("mt_persist") { tempDir =>
           testInfoDateExists(getDeltaMtPersistence(tempDir))
         }
       }
 
-      "support partition overwrites" in {
+      "supports fixing the existing info date column not partitioned" in {
+        if (spark.version.split('.').head.toInt >= 3) {
+          withTempDirectory("mt_persist") { tempDir =>
+            testInfoDateExists(getDeltaMtPersistence(tempDir, partitionByInfoDate = false))
+          }
+        }
+      }
+
+      "support partition overwrites partitioned" in {
         withTempDirectory("mt_persist") { tempDir =>
           testOverwritePartition(getDeltaMtPersistence(tempDir))
         }
       }
 
-      "support partition appends" in {
+      "support partition overwrites not partitioned" in {
+        if (spark.version.split('.').head.toInt >= 3) {
+          withTempDirectory("mt_persist") { tempDir =>
+            testOverwritePartition(getDeltaMtPersistence(tempDir, partitionByInfoDate = false))
+          }
+        }
+      }
+
+      "support partition appends partitioned" in {
         withTempDirectory("mt_persist") { tempDir =>
           testAppendPartition(getDeltaMtPersistence(tempDir, saveModeOpt = Some(SaveMode.Append)))
         }
       }
 
+      "support partition appends not partitioned" in {
+        if (spark.version.split('.').head.toInt >= 3) {
+          withTempDirectory("mt_persist") { tempDir =>
+            testAppendPartition(getDeltaMtPersistence(tempDir, saveModeOpt = Some(SaveMode.Append), partitionByInfoDate = false))
+          }
+        }
+      }
+
       "fail of schema change" in {
         withTempDirectory("mt_persist") { tempDir =>
           val ex = intercept[AnalysisException] {
@@ -527,12 +576,20 @@ class MetastorePersistenceSuite extends AnyWordSpec with SparkTestBase with Temp
         }
       }
 
-      "supports schema merges" in {
+      "supports schema merges partitioned" in {
         withTempDirectory("mt_persist") { tempDir =>
           testSchemaMerge(getDeltaMtPersistence(tempDir))
         }
       }
 
+      "supports schema merges not partitioned" in {
+        if (spark.version.split('.').head.toInt >= 3) {
+          withTempDirectory("mt_persist") { tempDir =>
+            testSchemaMerge(getDeltaMtPersistence(tempDir, partitionByInfoDate = false))
+          }
+        }
+      }
+
       "supports records per partition" in {
         withTempDirectory("mt_persist") { tempDir =>
           testRecordsPerPartition(s"$tempDir/delta/info_date=2021-10-12", "*.parquet", getDeltaMtPersistence(tempDir, recordsPerPartition = Some(2)))
@@ -547,12 +604,20 @@ class MetastorePersistenceSuite extends AnyWordSpec with SparkTestBase with Temp
     }
 
     "getStats()" should {
-      "get stats after a save" in {
+      "get stats after a save partitioned" in {
         withTempDirectory("mt_persist") { tempDir =>
           testStatsAvailable(getDeltaMtPersistence(tempDir))
         }
       }
 
+      "get stats after a save not partitioned" in {
+        if (spark.version.split('.').head.toInt >= 3) {
+          withTempDirectory("mt_persist") { tempDir =>
+            testStatsEmptyForNonPartitionedTables(getDeltaMtPersistence(tempDir, partitionByInfoDate = false))
+          }
+        }
+      }
+
       "throw an exception if stats are not available" in {
         withTempDirectory("mt_persist") { tempDir =>
           testStatsNotAvailable(getDeltaMtPersistence(tempDir))
@@ -593,12 +658,14 @@ class MetastorePersistenceSuite extends AnyWordSpec with SparkTestBase with Temp
   }
 
   def getDeltaMtPersistence(tempDir: String,
+                            partitionByInfoDate: Boolean = true,
                             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),
+      partitionByInfoDate = partitionByInfoDate,
       infoDateColumn = infoDateColumn,
       infoDateFormat = infoDateFormat,
       saveModeOpt = saveModeOpt,
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 61f4f1404..9f25bf535 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
@@ -28,6 +28,7 @@ object MetaTableFactory {
                         format: DataFormat = DataFormat.Parquet("/tmp/dummy", None),
                         infoDateColumn: String = "INFO_DATE",
                         infoDateFormat: String = "yyyy-MM-dd",
+                        partitionByInfoDate: Boolean = true,
                         batchIdColumn: String = "pramen_batchid",
                         hiveConfig: HiveConfig = HiveConfig.getNullConfig,
                         hiveTable: Option[String] = None,
@@ -47,6 +48,7 @@ object MetaTableFactory {
       format,
       infoDateColumn,
       infoDateFormat,
+      partitionByInfoDate,
       batchIdColumn,
       hiveConfig,
       hiveTable,
diff --git a/pramen/core/src/test/scala/za/co/absa/pramen/core/mocks/metastore/MetastoreReaderMock.scala b/pramen/core/src/test/scala/za/co/absa/pramen/core/mocks/metastore/MetastoreReaderMock.scala
index d4bd0e8ca..bb8bcf9a4 100644
--- a/pramen/core/src/test/scala/za/co/absa/pramen/core/mocks/metastore/MetastoreReaderMock.scala
+++ b/pramen/core/src/test/scala/za/co/absa/pramen/core/mocks/metastore/MetastoreReaderMock.scala
@@ -20,7 +20,9 @@ import org.apache.spark.sql.DataFrame
 import za.co.absa.pramen.api.offset.DataOffset
 import za.co.absa.pramen.api.status.TaskRunReason
 import za.co.absa.pramen.api._
+import za.co.absa.pramen.core.MetaTableDefFactory
 import za.co.absa.pramen.core.metadata.MetadataManagerNull
+import za.co.absa.pramen.core.mocks.MetaTableFactory
 
 import java.time.LocalDate
 
@@ -58,7 +60,8 @@ class MetastoreReaderMock(tables: Seq[(String, DataFrame)], infoDate: LocalDate)
 
   override def getTableDef(tableName: String): MetaTableDef = {
     tables.find(_._1 == tableName) match {
-      case Some((name, _)) => MetaTableDef(name, "", DataFormat.Null(), "pramen_info_date", "yyyy-MM-dd", "pramen_batchid", None, None, null, Map.empty[String, String], Map.empty[String, String])
+      case Some((name, _)) =>
+        MetaTableDefFactory.getDummyMetaTableDef(name, format = DataFormat.Null(), infoDateColumn = "pramen_info_date")
       case None          => throw new IllegalArgumentException(s"Table $tableName not found")
     }
   }
diff --git a/pramen/core/src/test/scala/za/co/absa/pramen/core/mocks/metastore/MetastoreSpy.scala b/pramen/core/src/test/scala/za/co/absa/pramen/core/mocks/metastore/MetastoreSpy.scala
index d43749be3..88e0d117c 100644
--- a/pramen/core/src/test/scala/za/co/absa/pramen/core/mocks/metastore/MetastoreSpy.scala
+++ b/pramen/core/src/test/scala/za/co/absa/pramen/core/mocks/metastore/MetastoreSpy.scala
@@ -148,9 +148,10 @@ class MetastoreSpy(registeredTables: Seq[String] = Seq("table1", "table2"),
         MetaTableDef(table.name,
           table.description,
           table.format,
-          table.batchIdColumn,
           table.infoDateColumn,
           table.infoDateFormat,
+          table.partitionByInfoDate,
+          table.batchIdColumn,
           table.hiveTable,
           table.hivePath,
           table.infoDateStart,
diff --git a/pramen/core/src/test/scala/za/co/absa/pramen/core/pipeline/OperationDefSuite.scala b/pramen/core/src/test/scala/za/co/absa/pramen/core/pipeline/OperationDefSuite.scala
index 148a4efbf..7cc1c1bdc 100644
--- a/pramen/core/src/test/scala/za/co/absa/pramen/core/pipeline/OperationDefSuite.scala
+++ b/pramen/core/src/test/scala/za/co/absa/pramen/core/pipeline/OperationDefSuite.scala
@@ -28,6 +28,7 @@ import java.time.LocalDate
 class OperationDefSuite extends AnyWordSpec with TempDirFixture {
   private val defaults = InfoDateConfig("INFO_DATE",
     "yyyy-MM-dd",
+    partitionByInfoDate = true,
     LocalDate.of(2022, 1, 1),
     4,
     0,
diff --git a/pramen/core/src/test/scala/za/co/absa/pramen/core/pipeline/PipelineDefSuite.scala b/pramen/core/src/test/scala/za/co/absa/pramen/core/pipeline/PipelineDefSuite.scala
index fc59ef6b9..68b6fa4dc 100644
--- a/pramen/core/src/test/scala/za/co/absa/pramen/core/pipeline/PipelineDefSuite.scala
+++ b/pramen/core/src/test/scala/za/co/absa/pramen/core/pipeline/PipelineDefSuite.scala
@@ -32,6 +32,7 @@ import java.time.LocalDate
 class PipelineDefSuite extends AnyWordSpec with SparkTestBase with TempDirFixture {
   private val defaults = InfoDateConfig("INFO_DATE",
     "yyyy-MM-dd",
+    partitionByInfoDate = true,
     LocalDate.of(2022, 1, 1),
     4,
     0,
diff --git a/pramen/extras/src/test/scala/za/co/absa/pramen/extras/mocks/TestPrototypes.scala b/pramen/extras/src/test/scala/za/co/absa/pramen/extras/mocks/TestPrototypes.scala
index b9b65fffe..2f6cb1fff 100644
--- a/pramen/extras/src/test/scala/za/co/absa/pramen/extras/mocks/TestPrototypes.scala
+++ b/pramen/extras/src/test/scala/za/co/absa/pramen/extras/mocks/TestPrototypes.scala
@@ -33,6 +33,7 @@ object TestPrototypes {
     DataFormat.Null(),
     "pramen_info_date",
     "yyyy-MM-dd",
+    partitionByInfoDate = true,
     "pramen_batchid",
     None,
     None,