Skip to content

#765 Add neverRepairPartitions option to skip MSCK REPAIR TABLE in Hive helpers. #766

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
yruslan merged 2 commits into from
Jun 25, 2026
Merged

#765 Add neverRepairPartitions option to skip MSCK REPAIR TABLE in Hive helpers. #766

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
d330b9b
#765 Add `neverRepairPartitions` option to skip MSCK REPAIR TABLE in …
yruslan Jun 25, 2026
3972fea
#765 Pass `neverRepairPartitions` config to EnceladusSink via Externa…
yruslan Jun 25, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 3 additions & 0 deletions pramen/core/src/main/resources/reference.conf
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
1 change: 1 addition & 0 deletions pramen/core/src/main/scala/za/co/absa/pramen/core/config/Keys.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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"

Expand Down
9 changes: 6 additions & 3 deletions pramen/core/src/main/scala/za/co/absa/pramen/core/metastore/MetastoreImpl.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -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) {
Expand All @@ -215,15 +218,15 @@ 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
needAddPartition = true
}
} 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)
}
}

Expand Down
6 changes: 4 additions & 2 deletions pramen/core/src/main/scala/za/co/absa/pramen/core/utils/hive/HiveHelper.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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],
Expand Down
20 changes: 16 additions & 4 deletions pramen/core/src/main/scala/za/co/absa/pramen/core/utils/hive/HiveHelperSparkCatalog.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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)) {
Expand All @@ -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)) {
Expand All @@ -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)) {
Expand Down
20 changes: 16 additions & 4 deletions pramen/core/src/main/scala/za/co/absa/pramen/core/utils/hive/HiveHelperSql.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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.")
}
}

Expand All @@ -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.")
}
}

Expand Down
24 changes: 12 additions & 12 deletions .../src/test/scala/za/co/absa/pramen/core/tests/utils/hive/HiveHelperSparkCatalogSuite.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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)
}
}
}
Expand All @@ -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"))
}
}
Expand All @@ -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] {
Expand All @@ -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"))
}
}
Expand All @@ -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"))
}
}
Expand All @@ -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)
Expand All @@ -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")
Expand All @@ -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")
Expand Down
33 changes: 30 additions & 3 deletions pramen/core/src/test/scala/za/co/absa/pramen/core/tests/utils/hive/HiveHelperSqlSuite.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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()

Expand Down Expand Up @@ -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")

Expand Down Expand Up @@ -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")

Expand Down
Loading
Loading