diff --git a/pramen/core/src/main/scala/za/co/absa/pramen/core/metastore/MetastoreImpl.scala b/pramen/core/src/main/scala/za/co/absa/pramen/core/metastore/MetastoreImpl.scala index 9b0d4309..3c72cf47 100644 --- a/pramen/core/src/main/scala/za/co/absa/pramen/core/metastore/MetastoreImpl.scala +++ b/pramen/core/src/main/scala/za/co/absa/pramen/core/metastore/MetastoreImpl.scala @@ -206,7 +206,7 @@ class MetastoreImpl(appConfig: Config, if (recreate) { log.info(s"Recreating Hive table '$fullTableName'") - hiveHelper.createOrUpdateHiveTable(effectivePath, format, effectiveSchema, Seq(mt.infoDateColumn), mt.hiveConfig.database, hiveTable, neverRepairPartitions) + hiveHelper.createOrUpdateHiveTable(effectivePath, format, effectiveSchema, Seq(mt.infoDateColumn), mt.hiveConfig.database, hiveTable, !neverRepairPartitions) } else { if (hiveHelper.doesTableExist(mt.hiveConfig.database, hiveTable)) { if (updateSchema) { @@ -218,7 +218,7 @@ class MetastoreImpl(appConfig: Config, } catch { case NonFatal(ex) => log.warn(s"Could not update Hive schema via ${hiveHelper.getClass.getName}. Recreating Hive table '$fullTableName'", ex) - hiveHelper.createOrUpdateHiveTable(effectivePath, format, effectiveSchema, Seq(mt.infoDateColumn), mt.hiveConfig.database, hiveTable, neverRepairPartitions) + hiveHelper.createOrUpdateHiveTable(effectivePath, format, effectiveSchema, Seq(mt.infoDateColumn), mt.hiveConfig.database, hiveTable, !neverRepairPartitions) } } else { // Schema didn't change, but we need to add the new partition @@ -226,7 +226,7 @@ class MetastoreImpl(appConfig: Config, } } else { log.info(s"The table '$fullTableName' does not exist. Creating it.") - hiveHelper.createOrUpdateHiveTable(effectivePath, format, effectiveSchema, Seq(mt.infoDateColumn), mt.hiveConfig.database, hiveTable, neverRepairPartitions) + hiveHelper.createOrUpdateHiveTable(effectivePath, format, effectiveSchema, Seq(mt.infoDateColumn), mt.hiveConfig.database, hiveTable, !neverRepairPartitions) } } diff --git a/pramen/core/src/main/scala/za/co/absa/pramen/core/utils/hive/HiveHelper.scala b/pramen/core/src/main/scala/za/co/absa/pramen/core/utils/hive/HiveHelper.scala index acb6ef39..45119da5 100644 --- a/pramen/core/src/main/scala/za/co/absa/pramen/core/utils/hive/HiveHelper.scala +++ b/pramen/core/src/main/scala/za/co/absa/pramen/core/utils/hive/HiveHelper.scala @@ -29,7 +29,7 @@ abstract class HiveHelper { partitionBy: Seq[String], databaseName: Option[String], tableName: String, - neverRepairPartitions: Boolean = false): Unit + autoRepairPartitions: Boolean = true): Unit def createOrUpdateHiveTable(path: String, format: HiveFormat, @@ -37,7 +37,7 @@ abstract class HiveHelper { partitionBy: Seq[String], databaseName: Option[String], tableName: String, - neverRepairPartitions: Boolean = false): Unit + autoRepairPartitions: Boolean = true): Unit def replaceHiveTableSchema(schema: StructType, partitionBy: Seq[String], diff --git a/pramen/core/src/main/scala/za/co/absa/pramen/core/utils/hive/HiveHelperSparkCatalog.scala b/pramen/core/src/main/scala/za/co/absa/pramen/core/utils/hive/HiveHelperSparkCatalog.scala index e5131466..875e1d2c 100644 --- a/pramen/core/src/main/scala/za/co/absa/pramen/core/utils/hive/HiveHelperSparkCatalog.scala +++ b/pramen/core/src/main/scala/za/co/absa/pramen/core/utils/hive/HiveHelperSparkCatalog.scala @@ -33,18 +33,18 @@ class HiveHelperSparkCatalog(spark: SparkSession) extends HiveHelper { partitionBy: Seq[String], databaseName: Option[String], tableName: String, - neverRepairPartitions: Boolean): Unit = { + autoRepairPartitions: Boolean): Unit = { val fullTableName = HiveHelper.getFullTable(databaseName, tableName) createCatalogTable(fullTableName, path, format) - if (partitionBy.nonEmpty && !neverRepairPartitions) { + if (partitionBy.nonEmpty && autoRepairPartitions) { repairHiveTable(databaseName, tableName, format) } else { if (partitionBy.isEmpty) log.info(s"Skipping repairing partition for $fullTableName because the table is not partitioned.") else - log.info(s"Skipping repairing partition for $fullTableName because repairing partitions is disabled.") + log.info(s"Skipping repairing partition for $fullTableName as requested.") } if (!doesTableExist(databaseName, tableName)) { @@ -58,7 +58,7 @@ class HiveHelperSparkCatalog(spark: SparkSession) extends HiveHelper { partitionBy: Seq[String], databaseName: Option[String], tableName: String, - neverRepairPartitions: Boolean): Unit = { + autoRepairPartitions: Boolean): Unit = { val fullTableName = HiveHelper.getFullTable(databaseName, tableName) if (doesTableExist(databaseName, tableName)) { @@ -68,13 +68,13 @@ class HiveHelperSparkCatalog(spark: SparkSession) extends HiveHelper { createCatalogTable(fullTableName, path, format) - if (partitionBy.nonEmpty && !neverRepairPartitions) { + if (partitionBy.nonEmpty && autoRepairPartitions) { repairHiveTable(databaseName, tableName, format) } else { if (partitionBy.isEmpty) log.info(s"Skipping repairing partition for $fullTableName because the table is not partitioned.") else - log.info(s"Skipping repairing partition for $fullTableName because repairing partitions is disabled.") + log.info(s"Skipping repairing partition for $fullTableName as requested.") } if (!doesTableExist(databaseName, tableName)) { diff --git a/pramen/core/src/main/scala/za/co/absa/pramen/core/utils/hive/HiveHelperSql.scala b/pramen/core/src/main/scala/za/co/absa/pramen/core/utils/hive/HiveHelperSql.scala index 82ef849c..4dee7c5e 100644 --- a/pramen/core/src/main/scala/za/co/absa/pramen/core/utils/hive/HiveHelperSql.scala +++ b/pramen/core/src/main/scala/za/co/absa/pramen/core/utils/hive/HiveHelperSql.scala @@ -31,17 +31,17 @@ class HiveHelperSql(val queryExecutor: QueryExecutor, partitionBy: Seq[String], databaseName: Option[String], tableName: String, - neverRepairPartitions: Boolean): Unit = { + autoRepairPartitions: Boolean): Unit = { val fullTableName = HiveHelper.getFullTable(databaseName, tableName) createHiveTable(fullTableName, path, format, schema, partitionBy, failIfExists = true) - if (partitionBy.nonEmpty && !neverRepairPartitions) { + if (partitionBy.nonEmpty && autoRepairPartitions) { repairHiveTable(fullTableName) } else { if (partitionBy.isEmpty) log.info(s"Skipping repairing partition for $fullTableName because the table is not partitioned.") else - log.info(s"Skipping repairing partition for $fullTableName because repairing partitions is disabled.") + log.info(s"Skipping repairing partition for $fullTableName as requested.") } } @@ -51,18 +51,18 @@ class HiveHelperSql(val queryExecutor: QueryExecutor, partitionBy: Seq[String], databaseName: Option[String], tableName: String, - neverRepairPartitions: Boolean): Unit = { + autoRepairPartitions: Boolean): Unit = { val fullTableName = HiveHelper.getFullTable(databaseName, tableName) dropHiveTable(fullTableName) createHiveTable(fullTableName, path, format, schema, partitionBy, failIfExists = false) - if (partitionBy.nonEmpty && !neverRepairPartitions) { + if (partitionBy.nonEmpty && autoRepairPartitions) { repairHiveTable(fullTableName) } else { if (partitionBy.isEmpty) log.info(s"Skipping repairing partition for $fullTableName because the table is not partitioned.") else - log.info(s"Skipping repairing partition for $fullTableName because repairing partitions is disabled.") + log.info(s"Skipping repairing partition for $fullTableName as requested.") } } diff --git a/pramen/core/src/test/scala/za/co/absa/pramen/core/tests/utils/hive/HiveHelperSparkCatalogSuite.scala b/pramen/core/src/test/scala/za/co/absa/pramen/core/tests/utils/hive/HiveHelperSparkCatalogSuite.scala index 9541cacf..d82c5af5 100644 --- a/pramen/core/src/test/scala/za/co/absa/pramen/core/tests/utils/hive/HiveHelperSparkCatalogSuite.scala +++ b/pramen/core/src/test/scala/za/co/absa/pramen/core/tests/utils/hive/HiveHelperSparkCatalogSuite.scala @@ -38,12 +38,12 @@ class HiveHelperSparkCatalogSuite extends AnyWordSpec with SparkTestBase with Te val hiveHelper = new HiveHelperSparkCatalog(spark) val schema = spark.read.parquet(path).schema - hiveHelper.createHiveTable(path, HiveFormat.Parquet, schema, Nil, Some("default"), "tbl1", neverRepairPartitions = false) + hiveHelper.createHiveTable(path, HiveFormat.Parquet, schema, Nil, Some("default"), "tbl1") assert(spark.catalog.tableExists("default.tbl1")) // If the table exists an exception should be thrown assertThrows[AnalysisException] { - hiveHelper.createHiveTable(path, HiveFormat.Parquet, schema, Nil, Some("default"), "tbl1", neverRepairPartitions = false) + hiveHelper.createHiveTable(path, HiveFormat.Parquet, schema, Nil, Some("default"), "tbl1") } } } @@ -55,13 +55,13 @@ class HiveHelperSparkCatalogSuite extends AnyWordSpec with SparkTestBase with Te val hiveHelper = new HiveHelperSparkCatalog(spark) val schema = spark.read.parquet(path).withColumn("b", lit(1)).schema - hiveHelper.createHiveTable(path, HiveFormat.Parquet, schema, "a" :: "b" :: Nil, Some("default"), "tbl2", neverRepairPartitions = false) + hiveHelper.createHiveTable(path, HiveFormat.Parquet, schema, "a" :: "b" :: Nil, Some("default"), "tbl2") assert(hiveHelper.doesTableExist(Some("default"), "tbl2")) spark.sql(s"DROP TABLE default.tbl2").collect() assert(!hiveHelper.doesTableExist(Some("default"), "tbl2")) - hiveHelper.createHiveTable(path, HiveFormat.Parquet, schema, "a" :: "b" :: Nil, Some("default"), "tbl2", neverRepairPartitions = false) + hiveHelper.createHiveTable(path, HiveFormat.Parquet, schema, "a" :: "b" :: Nil, Some("default"), "tbl2") assert(hiveHelper.doesTableExist(Some("default"), "tbl2")) } } @@ -73,7 +73,7 @@ class HiveHelperSparkCatalogSuite extends AnyWordSpec with SparkTestBase with Te val hiveHelper = new HiveHelperSparkCatalog(spark) val schema = spark.read.parquet(path).withColumn("b", lit(1)).schema - hiveHelper.createHiveTable(path, HiveFormat.Parquet, schema, "a" :: "b" :: Nil, Some("default"), "tbl3", neverRepairPartitions = false) + hiveHelper.createHiveTable(path, HiveFormat.Parquet, schema, "a" :: "b" :: Nil, Some("default"), "tbl3") assert(hiveHelper.doesTableExist(Some("default"), "tbl3")) assertThrows[AnalysisException] { @@ -91,11 +91,11 @@ class HiveHelperSparkCatalogSuite extends AnyWordSpec with SparkTestBase with Te val hiveHelper = new HiveHelperSparkCatalog(spark) val schema = spark.read.parquet(path).schema - hiveHelper.createOrUpdateHiveTable(path, HiveFormat.Parquet, schema, Nil, Some("default"), "tbl1", neverRepairPartitions = false) + hiveHelper.createOrUpdateHiveTable(path, HiveFormat.Parquet, schema, Nil, Some("default"), "tbl1") assert(spark.catalog.tableExists("default.tbl1")) // If the table exists it should be re-created - hiveHelper.createOrUpdateHiveTable(path, HiveFormat.Parquet, schema, Nil, Some("default"), "tbl1", neverRepairPartitions = false) + hiveHelper.createOrUpdateHiveTable(path, HiveFormat.Parquet, schema, Nil, Some("default"), "tbl1") assert(spark.catalog.tableExists("default.tbl1")) } } @@ -107,13 +107,13 @@ class HiveHelperSparkCatalogSuite extends AnyWordSpec with SparkTestBase with Te val hiveHelper = new HiveHelperSparkCatalog(spark) val schema = spark.read.parquet(path).withColumn("b", lit(1)).schema - hiveHelper.createOrUpdateHiveTable(path, HiveFormat.Parquet, schema, "a" :: "b" :: Nil, Some("default"), "tbl2", neverRepairPartitions = false) + hiveHelper.createOrUpdateHiveTable(path, HiveFormat.Parquet, schema, "a" :: "b" :: Nil, Some("default"), "tbl2") assert(hiveHelper.doesTableExist(Some("default"),"tbl2")) spark.sql(s"DROP TABLE default.tbl2").collect() assert(!hiveHelper.doesTableExist(Some("default"),"tbl2")) - hiveHelper.createOrUpdateHiveTable(path, HiveFormat.Parquet, schema, "a" :: "b" :: Nil, Some("default"), "tbl2", neverRepairPartitions = false) + hiveHelper.createOrUpdateHiveTable(path, HiveFormat.Parquet, schema, "a" :: "b" :: Nil, Some("default"), "tbl2") assert(hiveHelper.doesTableExist(Some("default"),"tbl2")) } } @@ -125,7 +125,7 @@ class HiveHelperSparkCatalogSuite extends AnyWordSpec with SparkTestBase with Te val hiveHelper = new HiveHelperSparkCatalog(spark) val schema = spark.read.format("delta").load(path).withColumn("b", lit(1)).schema - hiveHelper.createOrUpdateHiveTable(path, HiveFormat.Delta, schema, "b" :: Nil, Some("default"), "tbl3", neverRepairPartitions = false) + hiveHelper.createOrUpdateHiveTable(path, HiveFormat.Delta, schema, "b" :: Nil, Some("default"), "tbl3") assert(hiveHelper.doesTableExist(Some("default"), "tbl3")) assert(spark.table("default.tbl3").count() == 3) @@ -139,7 +139,7 @@ class HiveHelperSparkCatalogSuite extends AnyWordSpec with SparkTestBase with Te val hiveHelper = new HiveHelperSparkCatalog(spark) val schema = spark.read.parquet(path).withColumn("b", lit(1)).schema - hiveHelper.createOrUpdateHiveTable(path, HiveFormat.Parquet, schema, "a" :: "b" :: Nil, Some("default"), "tbl4", neverRepairPartitions = true) + hiveHelper.createOrUpdateHiveTable(path, HiveFormat.Parquet, schema, "a" :: "b" :: Nil, Some("default"), "tbl4") assert(hiveHelper.doesTableExist(Some("default"), "tbl4")) hiveHelper.dropTable(Some("default"), "tbl4") @@ -155,7 +155,7 @@ class HiveHelperSparkCatalogSuite extends AnyWordSpec with SparkTestBase with Te val hiveHelper = new HiveHelperSparkCatalog(spark) val schema = spark.read.parquet(path).withColumn("b", lit(1)).schema - hiveHelper.createOrUpdateHiveTable(path, HiveFormat.Parquet, schema, "a" :: "b" :: Nil, Some("default"), "tbl5", neverRepairPartitions = false) + hiveHelper.createOrUpdateHiveTable(path, HiveFormat.Parquet, schema, "a" :: "b" :: Nil, Some("default"), "tbl5") assert(hiveHelper.doesTableExist(Some("default"),"tbl5")) val df = List(("D", 40)).toDF("a", "c") diff --git a/pramen/core/src/test/scala/za/co/absa/pramen/core/tests/utils/hive/HiveHelperSqlSuite.scala b/pramen/core/src/test/scala/za/co/absa/pramen/core/tests/utils/hive/HiveHelperSqlSuite.scala index 0ba92217..243b3fd5 100644 --- a/pramen/core/src/test/scala/za/co/absa/pramen/core/tests/utils/hive/HiveHelperSqlSuite.scala +++ b/pramen/core/src/test/scala/za/co/absa/pramen/core/tests/utils/hive/HiveHelperSqlSuite.scala @@ -55,7 +55,7 @@ class HiveHelperSqlSuite extends AnyWordSpec with SparkTestBase with TempDirFixt val hiveHelper = new HiveHelperSql(qe, defaultHiveConfig, false) val schema = spark.read.parquet(path).schema - hiveHelper.createOrUpdateHiveTable(path, HiveFormat.Parquet, schema, Nil, Some("db"), "tbl", neverRepairPartitions = false) + hiveHelper.createOrUpdateHiveTable(path, HiveFormat.Parquet, schema, Nil, Some("db"), "tbl", autoRepairPartitions = false) qe.close() @@ -86,7 +86,7 @@ class HiveHelperSqlSuite extends AnyWordSpec with SparkTestBase with TempDirFixt val hiveHelper = new HiveHelperSql(qe, defaultHiveConfig, true) val schema = spark.read.parquet(path).withColumn("b", lit(1)).schema - hiveHelper.createOrUpdateHiveTable(path, HiveFormat.Parquet, schema, "a" :: "b" :: Nil, Some("db"), "tbl", neverRepairPartitions = false) + hiveHelper.createOrUpdateHiveTable(path, HiveFormat.Parquet, schema, "a" :: "b" :: Nil, Some("db"), "tbl") val actual = qe.queries.mkString("\n") @@ -113,7 +113,7 @@ class HiveHelperSqlSuite extends AnyWordSpec with SparkTestBase with TempDirFixt val hiveHelper = new HiveHelperSql(qe, defaultHiveConfig, true) val schema = spark.read.parquet(path).withColumn("b", lit(1)).schema - hiveHelper.createOrUpdateHiveTable(path, HiveFormat.Parquet, schema, "a" :: "b" :: Nil, Some("db"), "tbl", neverRepairPartitions = true) + hiveHelper.createOrUpdateHiveTable(path, HiveFormat.Parquet, schema, "a" :: "b" :: Nil, Some("db"), "tbl", autoRepairPartitions = false) val actual = qe.queries.mkString("\n") @@ -141,7 +141,7 @@ class HiveHelperSqlSuite extends AnyWordSpec with SparkTestBase with TempDirFixt val hiveHelper = new HiveHelperSql(qe, defaultHiveConfig, true) val schema = spark.read.parquet(path).withColumn("b", lit(1)).schema - hiveHelper.createHiveTable(path, HiveFormat.Parquet, schema, "a" :: "b" :: Nil, Some("db"), "tbl", neverRepairPartitions = false) + hiveHelper.createHiveTable(path, HiveFormat.Parquet, schema, "a" :: "b" :: Nil, Some("db"), "tbl") val actual = qe.queries.mkString("\n") diff --git a/pramen/extras/src/main/scala/za/co/absa/pramen/extras/sink/EnceladusSink.scala b/pramen/extras/src/main/scala/za/co/absa/pramen/extras/sink/EnceladusSink.scala index c73ed815..9c027009 100644 --- a/pramen/extras/src/main/scala/za/co/absa/pramen/extras/sink/EnceladusSink.scala +++ b/pramen/extras/src/main/scala/za/co/absa/pramen/extras/sink/EnceladusSink.scala @@ -451,7 +451,7 @@ class EnceladusSink(sinkConfig: Config, val schema = dfForHiveSchema.schema - hiveHelper.createOrUpdateHiveTable(publishBase, HiveFormat.Parquet, schema, Seq("enceladus_info_date", "enceladus_info_version"), enceladusConfig.hiveDatabase, hiveTable, neverRepairPartitions) + hiveHelper.createOrUpdateHiveTable(publishBase, HiveFormat.Parquet, schema, Seq("enceladus_info_date", "enceladus_info_version"), enceladusConfig.hiveDatabase, hiveTable, !neverRepairPartitions) } } diff --git a/pramen/extras/src/main/scala/za/co/absa/pramen/extras/sink/StandardizationSink.scala b/pramen/extras/src/main/scala/za/co/absa/pramen/extras/sink/StandardizationSink.scala index 758ab70b..5d65ba0e 100644 --- a/pramen/extras/src/main/scala/za/co/absa/pramen/extras/sink/StandardizationSink.scala +++ b/pramen/extras/src/main/scala/za/co/absa/pramen/extras/sink/StandardizationSink.scala @@ -234,8 +234,7 @@ class StandardizationSink(sinkConfig: Config, fullSchema, partitionBy, standardizationConfig.hiveDatabase, - hiveTable, - neverRepairPartitions = false) + hiveTable) (Seq.empty[String], Seq(fullTableName)) } catch { case NonFatal(ex) =>