diff --git a/pramen/core/src/main/resources/reference.conf b/pramen/core/src/main/resources/reference.conf index 82a34d5e..9fa1316f 100644 --- a/pramen/core/src/main/resources/reference.conf +++ b/pramen/core/src/main/resources/reference.conf @@ -67,6 +67,9 @@ pramen { # This option can be overridden per metatable. hive.prefer.add.partition = true + # If true, hive helpers won't repair partitions via MSCK REPAIR TABLE unless explicitly called. + hive.never.repair.partitions = false + # If enabled, the job will wait for the output table to become available before running a job # If the number of seconds <=0 the waiting will be infinite wait.for.output.table.enabled = false diff --git a/pramen/core/src/main/scala/za/co/absa/pramen/core/config/Keys.scala b/pramen/core/src/main/scala/za/co/absa/pramen/core/config/Keys.scala index 681dee03..16b39f0e 100644 --- a/pramen/core/src/main/scala/za/co/absa/pramen/core/config/Keys.scala +++ b/pramen/core/src/main/scala/za/co/absa/pramen/core/config/Keys.scala @@ -40,6 +40,7 @@ object Keys { val EXTRA_OPTIONS_PREFIX_V2 = "pramen.spark.conf" val ENABLE_HIVE_SUPPORT = "pramen.enable.hive" + val NEVER_REPAIR_PARTITIONS = "pramen.hive.never.repair.partitions" val STOP_SPARK_SESSION = "pramen.stop.spark.session" 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 53e11837..9b0d4309 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 @@ -30,6 +30,7 @@ import za.co.absa.pramen.core.app.config.{InfoDateConfig, RuntimeConfig} import za.co.absa.pramen.core.bookkeeper.model.OffsetCommitRequest import za.co.absa.pramen.core.bookkeeper.{Bookkeeper, OffsetManagerUtils} import za.co.absa.pramen.core.config.Keys +import za.co.absa.pramen.core.config.Keys.NEVER_REPAIR_PARTITIONS import za.co.absa.pramen.core.metastore.model.{MetaTable, ReaderMode, TrackingTable} import za.co.absa.pramen.core.metastore.peristence.{MetastorePersistence, TransientJobManager} import za.co.absa.pramen.core.utils.ConfigUtils @@ -57,6 +58,8 @@ class MetastoreImpl(appConfig: Config, private val incrementalTables = new mutable.HashSet[String] + private val neverRepairPartitions = ConfigUtils.getOptionBoolean(appConfig, NEVER_REPAIR_PARTITIONS).getOrElse(false) + override def getRegisteredTables: Seq[String] = tableDefs.map(_.name) override def getRegisteredMetaTables: Seq[MetaTable] = tableDefs @@ -203,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) + hiveHelper.createOrUpdateHiveTable(effectivePath, format, effectiveSchema, Seq(mt.infoDateColumn), mt.hiveConfig.database, hiveTable, neverRepairPartitions) } else { if (hiveHelper.doesTableExist(mt.hiveConfig.database, hiveTable)) { if (updateSchema) { @@ -215,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) + 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 @@ -223,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) + 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 541cfc51..acb6ef39 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 @@ -28,14 +28,16 @@ abstract class HiveHelper { schema: StructType, partitionBy: Seq[String], databaseName: Option[String], - tableName: String): Unit + tableName: String, + neverRepairPartitions: Boolean = false): Unit def createOrUpdateHiveTable(path: String, format: HiveFormat, schema: StructType, partitionBy: Seq[String], databaseName: Option[String], - tableName: String): Unit + tableName: String, + neverRepairPartitions: Boolean = false): 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 9ac102fd..e5131466 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 @@ -32,13 +32,19 @@ class HiveHelperSparkCatalog(spark: SparkSession) extends HiveHelper { schema: StructType, partitionBy: Seq[String], databaseName: Option[String], - tableName: String): Unit = { + tableName: String, + neverRepairPartitions: Boolean): Unit = { val fullTableName = HiveHelper.getFullTable(databaseName, tableName) createCatalogTable(fullTableName, path, format) - if (partitionBy.nonEmpty) { + if (partitionBy.nonEmpty && !neverRepairPartitions) { 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.") } if (!doesTableExist(databaseName, tableName)) { @@ -51,7 +57,8 @@ class HiveHelperSparkCatalog(spark: SparkSession) extends HiveHelper { schema: StructType, partitionBy: Seq[String], databaseName: Option[String], - tableName: String): Unit = { + tableName: String, + neverRepairPartitions: Boolean): Unit = { val fullTableName = HiveHelper.getFullTable(databaseName, tableName) if (doesTableExist(databaseName, tableName)) { @@ -61,8 +68,13 @@ class HiveHelperSparkCatalog(spark: SparkSession) extends HiveHelper { createCatalogTable(fullTableName, path, format) - if (partitionBy.nonEmpty) { + if (partitionBy.nonEmpty && !neverRepairPartitions) { 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.") } 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 a784deb6..82ef849c 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 @@ -30,12 +30,18 @@ class HiveHelperSql(val queryExecutor: QueryExecutor, schema: StructType, partitionBy: Seq[String], databaseName: Option[String], - tableName: String): Unit = { + tableName: String, + neverRepairPartitions: Boolean): Unit = { val fullTableName = HiveHelper.getFullTable(databaseName, tableName) createHiveTable(fullTableName, path, format, schema, partitionBy, failIfExists = true) - if (partitionBy.nonEmpty) { + if (partitionBy.nonEmpty && !neverRepairPartitions) { 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.") } } @@ -44,13 +50,19 @@ class HiveHelperSql(val queryExecutor: QueryExecutor, schema: StructType, partitionBy: Seq[String], databaseName: Option[String], - tableName: String): Unit = { + tableName: String, + neverRepairPartitions: Boolean): Unit = { val fullTableName = HiveHelper.getFullTable(databaseName, tableName) dropHiveTable(fullTableName) createHiveTable(fullTableName, path, format, schema, partitionBy, failIfExists = false) - if (partitionBy.nonEmpty) { + if (partitionBy.nonEmpty && !neverRepairPartitions) { 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.") } } 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 d82c5af5..9541cacf 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") + hiveHelper.createHiveTable(path, HiveFormat.Parquet, schema, Nil, Some("default"), "tbl1", neverRepairPartitions = false) 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") + hiveHelper.createHiveTable(path, HiveFormat.Parquet, schema, Nil, Some("default"), "tbl1", neverRepairPartitions = false) } } } @@ -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") + hiveHelper.createHiveTable(path, HiveFormat.Parquet, schema, "a" :: "b" :: Nil, Some("default"), "tbl2", neverRepairPartitions = false) 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") + hiveHelper.createHiveTable(path, HiveFormat.Parquet, schema, "a" :: "b" :: Nil, Some("default"), "tbl2", neverRepairPartitions = false) 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") + hiveHelper.createHiveTable(path, HiveFormat.Parquet, schema, "a" :: "b" :: Nil, Some("default"), "tbl3", neverRepairPartitions = false) 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") + hiveHelper.createOrUpdateHiveTable(path, HiveFormat.Parquet, schema, Nil, Some("default"), "tbl1", neverRepairPartitions = false) 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") + hiveHelper.createOrUpdateHiveTable(path, HiveFormat.Parquet, schema, Nil, Some("default"), "tbl1", neverRepairPartitions = false) 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") + hiveHelper.createOrUpdateHiveTable(path, HiveFormat.Parquet, schema, "a" :: "b" :: Nil, Some("default"), "tbl2", neverRepairPartitions = false) 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") + hiveHelper.createOrUpdateHiveTable(path, HiveFormat.Parquet, schema, "a" :: "b" :: Nil, Some("default"), "tbl2", neverRepairPartitions = false) 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") + hiveHelper.createOrUpdateHiveTable(path, HiveFormat.Delta, schema, "b" :: Nil, Some("default"), "tbl3", neverRepairPartitions = false) 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") + hiveHelper.createOrUpdateHiveTable(path, HiveFormat.Parquet, schema, "a" :: "b" :: Nil, Some("default"), "tbl4", neverRepairPartitions = true) 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") + hiveHelper.createOrUpdateHiveTable(path, HiveFormat.Parquet, schema, "a" :: "b" :: Nil, Some("default"), "tbl5", neverRepairPartitions = false) 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 d7cbd873..0ba92217 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") + hiveHelper.createOrUpdateHiveTable(path, HiveFormat.Parquet, schema, Nil, Some("db"), "tbl", neverRepairPartitions = false) qe.close() @@ -86,7 +86,34 @@ 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") + hiveHelper.createOrUpdateHiveTable(path, HiveFormat.Parquet, schema, "a" :: "b" :: Nil, Some("db"), "tbl", neverRepairPartitions = false) + + val actual = qe.queries.mkString("\n") + + compareText(actual, expected) + } + } + + "execute expected queries for partitioned table without repair table" in { + withTempDirectory("hive_test") { tempDir => + val path = getParquetPath(tempDir) + + val expected = + s"""DROP TABLE IF EXISTS `db`.`tbl` + |CREATE EXTERNAL TABLE IF NOT EXISTS + |`db`.`tbl` ( `c` INT ) + |PARTITIONED BY (`a` STRING,`b` INT) + |ROW FORMAT SERDE 'org.apache.hadoop.hive.ql.io.parquet.serde.ParquetHiveSerDe' + |STORED AS INPUTFORMAT 'org.apache.hadoop.hive.ql.io.parquet.MapredParquetInputFormat' + |OUTPUTFORMAT 'org.apache.hadoop.hive.ql.io.parquet.MapredParquetOutputFormat' + |LOCATION '$path'; + |""".stripMargin + + val qe = new QueryExecutorMock(tableExists = false) + 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) val actual = qe.queries.mkString("\n") @@ -114,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") + hiveHelper.createHiveTable(path, HiveFormat.Parquet, schema, "a" :: "b" :: Nil, Some("db"), "tbl", neverRepairPartitions = false) 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 e831fcbc..c73ed815 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 @@ -21,7 +21,8 @@ import org.apache.hadoop.fs.Path import org.apache.spark.sql.functions.lit import org.apache.spark.sql.{DataFrame, SparkSession} import org.slf4j.LoggerFactory -import za.co.absa.pramen.api.{ExternalChannelFactory, MetastoreReader, Sink, SinkResult} +import za.co.absa.pramen.api.{ExternalChannelFactoryV2, MetastoreReader, Sink, SinkResult} +import za.co.absa.pramen.core.config.Keys.NEVER_REPAIR_PARTITIONS import za.co.absa.pramen.core.utils.ConfigUtils import za.co.absa.pramen.core.utils.hive.HiveQueryTemplates.TEMPLATES_DEFAULT_PREFIX import za.co.absa.pramen.core.utils.hive._ @@ -149,7 +150,8 @@ import scala.util.{Failure, Success, Try} */ class EnceladusSink(sinkConfig: Config, enceladusConfig: EnceladusConfig, - hiveHelper: HiveHelper) extends Sink { + hiveHelper: HiveHelper, + neverRepairPartitions: Boolean) extends Sink { import za.co.absa.pramen.extras.sink.EnceladusSink._ @@ -449,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) + hiveHelper.createOrUpdateHiveTable(publishBase, HiveFormat.Parquet, schema, Seq("enceladus_info_date", "enceladus_info_version"), enceladusConfig.hiveDatabase, hiveTable, neverRepairPartitions) } } @@ -486,7 +488,7 @@ class EnceladusSink(sinkConfig: Config, } } -object EnceladusSink extends ExternalChannelFactory[EnceladusSink] { +object EnceladusSink extends ExternalChannelFactoryV2[EnceladusSink] { val OUTPUT_PATH_KEY = "path" val INFO_VERSION_KEY = "info.version" val DATASET_NAME_KEY = "dataset.name" @@ -500,7 +502,7 @@ object EnceladusSink extends ExternalChannelFactory[EnceladusSink] { val INFO_VERSION_AUTO_VALUE = "auto" - override def apply(conf: Config, parentPath: String, spark: SparkSession): EnceladusSink = { + override def apply(conf: Config, workflowConfig: Config, parentPath: String, spark: SparkSession): EnceladusSink = { val enceladusConfig = EnceladusConfig.fromConfig(conf) val alwaysEscapeColumnNames = ConfigUtils.getOptionBoolean(conf, HIVE_ALWAYS_ESCAPE_COLUMN_NAMES).getOrElse(true) @@ -508,7 +510,8 @@ object EnceladusSink extends ExternalChannelFactory[EnceladusSink] { val queryExecutor = QueryExecutorSpark(spark) val hiveHelper = new HiveHelperSql(queryExecutor, hiveTemplates, alwaysEscapeColumnNames) + val neverRepairPartitions = ConfigUtils.getOptionBoolean(workflowConfig, NEVER_REPAIR_PARTITIONS).getOrElse(false) - new EnceladusSink(conf, enceladusConfig, hiveHelper) + new EnceladusSink(conf, enceladusConfig, hiveHelper, 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 5d65ba0e..758ab70b 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,7 +234,8 @@ class StandardizationSink(sinkConfig: Config, fullSchema, partitionBy, standardizationConfig.hiveDatabase, - hiveTable) + hiveTable, + neverRepairPartitions = false) (Seq.empty[String], Seq(fullTableName)) } catch { case NonFatal(ex) => diff --git a/pramen/extras/src/test/scala/za/co/absa/pramen/extras/sink/EnceladusSinkSuite.scala b/pramen/extras/src/test/scala/za/co/absa/pramen/extras/sink/EnceladusSinkSuite.scala index f41becec..cb8c49ca 100644 --- a/pramen/extras/src/test/scala/za/co/absa/pramen/extras/sink/EnceladusSinkSuite.scala +++ b/pramen/extras/src/test/scala/za/co/absa/pramen/extras/sink/EnceladusSinkSuite.scala @@ -61,7 +61,7 @@ class EnceladusSinkSuite extends AnyWordSpec with SparkTestBase with TextCompari var sink: EnceladusSink = null "constructed from a config" in { - sink = EnceladusSink.apply(conf, "", spark) + sink = EnceladusSink.apply(conf, conf, "", spark) assert(sink.isInstanceOf[EnceladusSink]) } @@ -103,7 +103,7 @@ class EnceladusSinkSuite extends AnyWordSpec with SparkTestBase with TextCompari } "autoDetectVersionNumber" should { - val sink = EnceladusSink.apply(conf, "", spark) + val sink = EnceladusSink.apply(conf, conf, "", spark) withTempDirectory("enceladus_raw") { tempRaw => withTempDirectory("enceladus_publish") { tempPublish => @@ -139,7 +139,7 @@ class EnceladusSinkSuite extends AnyWordSpec with SparkTestBase with TextCompari implicit val qe: QueryExecutorSpy = new QueryExecutorSpy - val sink = EnceladusSink.apply(conf, "", spark) + val sink = EnceladusSink.apply(conf, conf, "", spark) sink.runEnceladusIfNeeded("my_table", infoDate, @@ -167,7 +167,7 @@ class EnceladusSinkSuite extends AnyWordSpec with SparkTestBase with TextCompari implicit val qe: QueryExecutorSpy = new QueryExecutorSpy(throwException = Some(new lang.IllegalStateException)) - val sink = EnceladusSink.apply(conf, "", spark) + val sink = EnceladusSink.apply(conf, conf, "", spark) assertThrows[IllegalStateException] { sink.runEnceladusIfNeeded("my_table", @@ -198,7 +198,7 @@ class EnceladusSinkSuite extends AnyWordSpec with SparkTestBase with TextCompari implicit val qe: QueryExecutorSpy = new QueryExecutorSpy - val sink = EnceladusSink.apply(conf, "", spark) + val sink = EnceladusSink.apply(conf, conf, "", spark) val ex = intercept[RuntimeException] { sink.runEnceladusIfNeeded("my_table", diff --git a/pramen/project/build.properties b/pramen/project/build.properties index 7fecd510..6766effe 100644 --- a/pramen/project/build.properties +++ b/pramen/project/build.properties @@ -13,4 +13,4 @@ # limitations under the License. # -sbt.version=1.12.11 +sbt.version=1.12.13