This is an automated email from the ASF dual-hosted git repository.
danny0405 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/hudi.git
The following commit(s) were added to refs/heads/master by this push:
new 63a1ab19254f feat(spark): add repair_orphan_files stored procedure
(#19121)
63a1ab19254f is described below
commit 63a1ab19254fe46f395b4b3ae81ae0c088499626
Author: Mahsood Ebrahim <[email protected]>
AuthorDate: Tue Jun 30 22:55:30 2026 -0700
feat(spark): add repair_orphan_files stored procedure (#19121)
Add a `repair_orphan_files` Spark SQL stored procedure that finds and
optionally removes orphan data files - files present on the filesystem but
not referenced by any commit (active or archived). This makes orphan-file
detection and cleanup accessible from any Spark SQL session, reusing the
same detection logic (org.apache.hudi.table.repair.RepairUtils) that backs
the HoodieRepairTool spark-submit utility.
Highlights:
- Dry-run (view) mode by default; cleanup mode moves orphans to backup_path.
- Per-partition scoping via `partition =>` for very large tables; detection
runs one Spark task per partition (listing + classification) so only
orphan
candidates are collected to the driver.
- Optional archived_start_ts / archived_end_ts to scope the instants
considered.
- max_orphans cap (default 100000) that fails fast with an actionable
message
instead of risking driver OOM when collecting candidates.
- Metadata-table safety cross-check: candidates still visible in the MDT are
surfaced as SKIPPED_PRESENT_IN_MDT rows rather than removed.
- Structured logging of BACKUP_FAILED root causes.
Handles COW (base files) and MOR (base + log files) and all commit action
types (COMMIT, DELTA_COMMIT, REPLACE_COMMIT). Registered in
HoodieProcedures.
Adds TestRepairOrphanFilesProcedure covering dry-run, cleanup + backup, MOR
log-file detection, inflight-commit exclusion, backup_path validation,
partition scoping, and the max_orphans cap.
Co-authored-by: mahsoode <[email protected]>
Co-authored-by: Claude Opus 4.8 (1M context) <[email protected]>
---
.../hudi/command/procedures/HoodieProcedures.scala | 1 +
.../procedures/RepairOrphanFilesProcedure.scala | 336 +++++++++++++++++++++
.../procedure/TestRepairOrphanFilesProcedure.scala | 256 ++++++++++++++++
3 files changed, 593 insertions(+)
diff --git
a/hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/command/procedures/HoodieProcedures.scala
b/hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/command/procedures/HoodieProcedures.scala
index 4fefef5ec65c..f602b6506264 100644
---
a/hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/command/procedures/HoodieProcedures.scala
+++
b/hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/command/procedures/HoodieProcedures.scala
@@ -86,6 +86,7 @@ object HoodieProcedures {
,(RepairDeduplicateProcedure.NAME, RepairDeduplicateProcedure.builder)
,(RepairMigratePartitionMetaProcedure.NAME,
RepairMigratePartitionMetaProcedure.builder)
,(RepairOverwriteHoodiePropsProcedure.NAME,
RepairOverwriteHoodiePropsProcedure.builder)
+ ,(RepairOrphanFilesProcedure.NAME, RepairOrphanFilesProcedure.builder)
,(RunCleanProcedure.NAME, RunCleanProcedure.builder)
,(ValidateHoodieSyncProcedure.NAME, ValidateHoodieSyncProcedure.builder)
,(ShowInvalidParquetProcedure.NAME, ShowInvalidParquetProcedure.builder)
diff --git
a/hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/command/procedures/RepairOrphanFilesProcedure.scala
b/hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/command/procedures/RepairOrphanFilesProcedure.scala
new file mode 100644
index 000000000000..2d15215f0c66
--- /dev/null
+++
b/hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/command/procedures/RepairOrphanFilesProcedure.scala
@@ -0,0 +1,336 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.hudi.command.procedures
+
+import org.apache.hudi.common.config.HoodieMetadataConfig
+import org.apache.hudi.common.engine.HoodieLocalEngineContext
+import org.apache.hudi.common.fs.FSUtils
+import org.apache.hudi.common.table.HoodieTableMetaClient
+import org.apache.hudi.metadata.{FileSystemBackedTableMetadata,
HoodieBackedTableMetadata}
+import org.apache.hudi.storage.{HoodieStorageUtils, StoragePath,
StoragePathInfo}
+import org.apache.hudi.table.repair.RepairUtils
+
+import org.apache.spark.internal.Logging
+import org.apache.spark.sql.Row
+import org.apache.spark.sql.types.{DataTypes, Metadata, StructField,
StructType}
+
+import java.util.function.Supplier
+
+import scala.collection.JavaConverters._
+
+/**
+ * Spark SQL stored procedure that finds and optionally removes orphan data
files — files
+ * that exist on the filesystem but are not referenced by any commit (active
or archived).
+ *
+ * Handles COW (base files) and MOR (base + log files), and all commit action
types
+ * (COMMIT, DELTA_COMMIT, REPLACE_COMMIT). The detection reuses
+ * [[org.apache.hudi.table.repair.RepairUtils]], the same logic that backs the
+ * `HoodieRepairTool` spark-submit utility, so results are consistent with
that tool.
+ *
+ * Usage:
+ * {{{
+ * -- View mode (default): list orphan files without touching them
+ * CALL repair_orphan_files(table => 'my_table')
+ *
+ * -- Scoped to one partition
+ * CALL repair_orphan_files(table => 'my_table', partition => '2024/01/15')
+ *
+ * -- Cleanup: move orphan files to a backup location
+ * CALL repair_orphan_files(
+ * table => 'my_table',
+ * dry_run => false,
+ * backup_path => '/user/hudi/orphan_files_backup'
+ * )
+ * }}}
+ *
+ * For very large tables, scope to one partition at a time using `partition
=>` to avoid
+ * collecting all orphan paths to the driver at once. The `max_orphans`
parameter (default
+ * 100,000) acts as a safety cap: if the detected count exceeds it the
procedure fails with
+ * a clear error instead of silently causing a driver OOM.
+ */
+class RepairOrphanFilesProcedure extends BaseProcedure with ProcedureBuilder
with Logging {
+
+ private val PARAMETERS = Array[ProcedureParameter](
+ ProcedureParameter.optional(0, "table", DataTypes.StringType,
null),
+ ProcedureParameter.optional(1, "path", DataTypes.StringType,
null),
+ ProcedureParameter.optional(2, "partition", DataTypes.StringType,
""),
+ ProcedureParameter.optional(3, "dry_run",
DataTypes.BooleanType, true),
+ ProcedureParameter.optional(4, "backup_path", DataTypes.StringType,
""),
+ ProcedureParameter.optional(5, "archived_start_ts", DataTypes.StringType,
""),
+ ProcedureParameter.optional(6, "archived_end_ts", DataTypes.StringType,
""),
+ ProcedureParameter.optional(7, "max_orphans",
DataTypes.IntegerType, 100000)
+ )
+
+ private val OUTPUT_TYPE = new StructType(Array[StructField](
+ StructField("partition", DataTypes.StringType, nullable = true,
Metadata.empty),
+ StructField("file_name", DataTypes.StringType, nullable = true,
Metadata.empty),
+ StructField("instant_time", DataTypes.StringType, nullable = true,
Metadata.empty),
+ StructField("backup_path", DataTypes.StringType, nullable = true,
Metadata.empty),
+ StructField("status", DataTypes.StringType, nullable = true,
Metadata.empty)
+ ))
+
+ def parameters: Array[ProcedureParameter] = PARAMETERS
+
+ def outputType: StructType = OUTPUT_TYPE
+
+ override def call(args: ProcedureArgs): Seq[Row] = {
+ super.checkArgs(PARAMETERS, args)
+
+ val tableName = getArgValueOrDefault(args, PARAMETERS(0))
+ val tablePathOpt = getArgValueOrDefault(args, PARAMETERS(1))
+ val partition = getArgValueOrDefault(args,
PARAMETERS(2)).get.asInstanceOf[String]
+ val dryRun = getArgValueOrDefault(args,
PARAMETERS(3)).get.asInstanceOf[Boolean]
+ val backupPath = getArgValueOrDefault(args,
PARAMETERS(4)).get.asInstanceOf[String]
+ val archivedStartTs = getArgValueOrDefault(args,
PARAMETERS(5)).get.asInstanceOf[String]
+ val archivedEndTs = getArgValueOrDefault(args,
PARAMETERS(6)).get.asInstanceOf[String]
+ val maxOrphans = getArgValueOrDefault(args,
PARAMETERS(7)).get.asInstanceOf[Int]
+
+ if (!dryRun && backupPath.isEmpty) {
+ throw new IllegalArgumentException("backup_path is required when dry_run
is false")
+ }
+
+ // Phase 1: Partition listing (driver)
+ val basePath = getBasePath(tableName, tablePathOpt)
+ val metaClient = createMetaClient(jsc, basePath)
+
+ val partitions: java.util.List[String] =
+ if (partition.nonEmpty) {
+ java.util.Collections.singletonList(partition)
+ } else {
+ // Use FileSystemBackedTableMetadata (filesystem listing, no MDT) for
partition discovery.
+ // This avoids any reliance on the metadata table being
present/consistent — the same
+ // approach HoodieRepairTool uses.
+ new FileSystemBackedTableMetadata(
+ new HoodieLocalEngineContext(metaClient.getStorageConf),
+ metaClient.getTableConfig, metaClient.getStorage,
basePath).getAllPartitionPaths
+ }
+
+ if (partitions.isEmpty) {
+ Seq.empty
+ } else {
+ doRepairOrphanFiles(basePath, metaClient, partitions, dryRun, backupPath,
+ archivedStartTs, archivedEndTs, maxOrphans)
+ }
+ }
+
+ private def doRepairOrphanFiles(
+ basePath: String,
+ metaClient: HoodieTableMetaClient,
+ partitions: java.util.List[String],
+ dryRun: Boolean,
+ backupPath: String,
+ archivedStartTs: String,
+ archivedEndTs: String,
+ maxOrphans: Int): Seq[Row] = {
+ // Build the active and archived timelines once on the driver, loading
completed-instant
+ // details into memory so executors can read commit metadata without
further I/O. This is
+ // the same pattern as HoodieRepairTool: the loaded timelines are
serializable and captured
+ // by the RDD closure below. The HoodieTableMetaClient itself is not
captured (not needed
+ // on executors once details are loaded).
+ val activeTimeline = metaClient.getActiveTimeline
+ val archivedTimeline =
+ if (archivedStartTs.nonEmpty)
metaClient.getArchivedTimeline(archivedStartTs)
+ else metaClient.getArchivedTimeline()
+ archivedTimeline.loadCompletedInstantDetailsInMemory()
+
+ // StorageConfiguration is Serializable and is the only stateful value
captured into the
+ // closure; storage handles are rebuilt per task from it.
+ val storageConf = metaClient.getStorageConf
+ val basePathStr = basePath
+ val archStartTs = archivedStartTs
+ val archEndTs = archivedEndTs
+
+ // Phase 2: Parallel orphan file detection (Spark RDD, one task per
partition). Each task
+ // lists its own partition and runs detection locally, so only the (small)
set of orphan
+ // candidates is collected back to the driver rather than the full file
listing.
+ val orphanRelPaths: List[String] = jsc.parallelize(partitions,
partitions.size())
+ .rdd
+ .flatMap { partitionStr =>
+ val storage = HoodieStorageUtils.getStorage(basePathStr,
storageConf)
+ val partPath = FSUtils.getAbsolutePartitionPath(new
StoragePath(basePathStr), partitionStr)
+ // getAllDataFilesInPartition handles FileNotFoundException (partition
deleted between
+ // listing and task execution) by returning an empty list rather than
throwing.
+ val allStatuses = FSUtils.getAllDataFilesInPartition(storage, partPath)
+ val allPaths = allStatuses.asScala.map((info: StoragePathInfo) =>
info.getPath).asJava
+
+ val instantToFilesMap =
RepairUtils.tagInstantsOfBaseAndLogFiles(basePathStr, allPaths)
+
+ if (instantToFilesMap.isEmpty) {
+ Iterator.empty
+ } else {
+ // Optionally scope detection to instants within [archived_start_ts,
archived_end_ts].
+ // Instants outside the range are left untouched (not reported as
orphans).
+ val instants = instantToFilesMap.keySet.asScala.toSeq.sorted.filter
{ instant =>
+ (archStartTs.isEmpty || instant >= archStartTs) &&
+ (archEndTs.isEmpty || instant <= archEndTs)
+ }
+
+ instants.flatMap { instant =>
+ RepairUtils.findInstantFilesToRemove(
+ instant,
+ instantToFilesMap.get(instant),
+ activeTimeline,
+ archivedTimeline
+ ).asScala
+ }.iterator
+ }
+ }
+ .collect()
+ .toList
+
+ if (orphanRelPaths.size > maxOrphans) {
+ throw new IllegalStateException(
+ s"Found ${orphanRelPaths.size} orphan candidates, which exceeds
max_orphans=$maxOrphans. " +
+ s"Re-run with partition => '<partition>' to scope to one partition at
a time, " +
+ s"or raise max_orphans if you are sure the driver has enough memory.")
+ }
+
+ // Phase 3: Metadata table (MDT) safety check (driver).
+ // Files still visible in the MDT are not true orphans — surface them as
+ // SKIPPED_PRESENT_IN_MDT (per-file exclusion) so the operator sees which
candidates the
+ // safety check held back, rather than silently dropping them.
+ val mdtConfig = HoodieMetadataConfig.newBuilder
+ .enable(true).ignoreSpuriousDeletes(true).build
+ val mdtReader = new HoodieBackedTableMetadata(
+ new HoodieLocalEngineContext(metaClient.getStorageConf),
metaClient.getStorage, mdtConfig, basePath)
+
+ val mdtUnsafePaths: Set[String] =
+ if (mdtReader.enabled) {
+ val byPartition = orphanRelPaths.groupBy(partitionOf)
+ val unsafe = byPartition.flatMap { case (partRel, paths) =>
+ val mdtPartPath = FSUtils.getAbsolutePartitionPath(new
StoragePath(basePath), partRel)
+ val mdtNames = mdtReader.getAllFilesInPartition(mdtPartPath).asScala
+ .map(_.getPath.getName).toSet
+ paths.filter(p => mdtNames.contains(new StoragePath(p).getName))
+ }.toSet
+ if (unsafe.nonEmpty) {
+ logWarning(s"Found ${unsafe.size} orphan candidate(s) still visible
in MDT — " +
+ s"emitting as SKIPPED_PRESENT_IN_MDT (per-file exclusion):
$unsafe")
+ }
+ unsafe
+ } else {
+ logWarning("Metadata table not enabled; skipping MDT safety
cross-check")
+ Set.empty[String]
+ }
+
+ val safeOrphanPaths = orphanRelPaths.filterNot(mdtUnsafePaths.contains)
+
+ // Phase 4: Build result rows.
+ val skippedRows: Seq[Row] = mdtUnsafePaths.toSeq.map { relPath =>
+ val fileName = new StoragePath(relPath).getName
+ val partRel = partitionOf(relPath)
+ val instantTime = FSUtils.getCommitTime(fileName)
+ Row(partRel, fileName, instantTime, "", "SKIPPED_PRESENT_IN_MDT")
+ }
+
+ val safeRows: Seq[Row] = if (dryRun) {
+ safeOrphanPaths.map { relPath =>
+ val fileName = new StoragePath(relPath).getName
+ val partRel = partitionOf(relPath)
+ val instantTime = FSUtils.getCommitTime(fileName)
+ Row(partRel, fileName, instantTime, "", "IDENTIFIED")
+ }
+ } else {
+ val storage = metaClient.getStorage
+ val tableNm = metaClient.getTableConfig.getTableName // always from
metaClient — the 'table'
+ // proc arg is
null when invoked via path =>
+ safeOrphanPaths.map { relPath =>
+ val fileName = new StoragePath(relPath).getName
+ val partRel = partitionOf(relPath)
+ val instantTime = FSUtils.getCommitTime(fileName)
+ val srcPath = new StoragePath(basePath, relPath)
+ // Non-partitioned tables have partRel="" — back up directly under
<backup>/<table>.
+ val destDir =
+ if (partRel.isEmpty) new StoragePath(s"$backupPath/$tableNm")
+ else new StoragePath(s"$backupPath/$tableNm/$partRel")
+ val destPath = new StoragePath(destDir, fileName)
+
+ // Each storage op is wrapped to capture exception class+message — a
backup can fail for
+ // permissions, missing parent, RPC, or concurrent-delete reasons, and
a status of
+ // BACKUP_FAILED with no log line leaves the operator with nothing to
diagnose. Causes
+ // are accumulated and logged once at the end iff the final outcome is
a failure.
+ val causes = scala.collection.mutable.ArrayBuffer.empty[String]
+
+ val dirCreated: Boolean =
+ try {
+ val ok = storage.createDirectory(destDir)
+ if (!ok) causes += s"createDirectory($destDir)=false (likely
permissions or destDir exists as a file)"
+ ok
+ } catch {
+ case t: Throwable =>
+ causes += s"createDirectory($destDir) threw
${t.getClass.getSimpleName}: ${t.getMessage}"
+ false
+ }
+
+ val moved: Boolean =
+ if (!dirCreated) {
+ false
+ } else {
+ try {
+ val ok = storage.rename(srcPath, destPath)
+ if (!ok) causes += s"rename returned false (srcPath missing,
destPath exists, or cross-volume rename)"
+ ok
+ } catch {
+ case t: Throwable =>
+ causes += s"rename threw ${t.getClass.getSimpleName}:
${t.getMessage}"
+ false
+ }
+ }
+
+ // A concurrent cleaner may have already deleted srcPath — that is
success, not failure.
+ val srcStillExists: Boolean =
+ try {
+ storage.exists(srcPath)
+ } catch {
+ case t: Throwable =>
+ // Cannot confirm the concurrent-delete recovery path; treat as
failure.
+ causes += s"exists($srcPath) threw ${t.getClass.getSimpleName}:
${t.getMessage}"
+ true
+ }
+ val succeeded = moved || !srcStillExists
+
+ if (!succeeded) {
+ logWarning(s"BACKUP_FAILED for $srcPath -> $destPath; causes:
${causes.mkString("; ")}")
+ }
+ val status = if (succeeded) "BACKED_UP" else "BACKUP_FAILED"
+ val backupOut = if (succeeded) destPath.toString else ""
+ Row(partRel, fileName, instantTime, backupOut, status)
+ }
+ }
+
+ safeRows ++ skippedRows
+ }
+
+ // Extract the relative partition path from a relative file path using
string ops. This avoids
+ // any path-normalization surprises and works uniformly for partitioned and
non-partitioned
+ // tables (root files yield "").
+ private def partitionOf(relPath: String): String = {
+ val s = relPath.lastIndexOf('/')
+ if (s < 0) "" else relPath.substring(0, s)
+ }
+
+ override def build: Procedure = new RepairOrphanFilesProcedure()
+}
+
+object RepairOrphanFilesProcedure {
+ val NAME = "repair_orphan_files"
+
+ def builder: Supplier[ProcedureBuilder] = new Supplier[ProcedureBuilder] {
+ override def get(): ProcedureBuilder = new RepairOrphanFilesProcedure()
+ }
+}
diff --git
a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/procedure/TestRepairOrphanFilesProcedure.scala
b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/procedure/TestRepairOrphanFilesProcedure.scala
new file mode 100644
index 000000000000..b77c3bd11e43
--- /dev/null
+++
b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/procedure/TestRepairOrphanFilesProcedure.scala
@@ -0,0 +1,256 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.hudi.procedure
+
+import org.apache.hudi.common.table.HoodieTableMetaClient
+import org.apache.hudi.common.testutils.FileCreateUtils
+import org.apache.hudi.hadoop.fs.HadoopFSUtils
+
+import org.apache.hadoop.fs.Path
+
+import java.util.UUID
+
+class TestRepairOrphanFilesProcedure extends HoodieSparkProcedureTestBase {
+
+ private val ORPHAN_INSTANT = "20000101000000000" // Year 2000; never in any
test timeline
+
+ private def metaClientFor(tablePath: String): HoodieTableMetaClient = {
+ HoodieTableMetaClient.builder
+
.setConf(HadoopFSUtils.getStorageConfWithCopy(spark.sparkContext.hadoopConfiguration))
+ .setBasePath(tablePath)
+ .build
+ }
+
+ // Test 1 — dry run detects a base-file orphan in a non-partitioned COW table
+ test("Test Call repair_orphan_files dry run finds base file orphan") {
+ withTempDir { tmp =>
+ val tableName = generateTableName
+ val tablePath = s"${tmp.getCanonicalPath}/$tableName"
+
+ // Create a non-partitioned COW table and write one real commit
+ spark.sql(
+ s"""create table $tableName (id int, name string, price double, ts
long)
+ |using hudi
+ |location '$tablePath'
+ |tblproperties (primaryKey = 'id', preCombineField = 'ts')
+ |""".stripMargin)
+ spark.sql(s"insert into $tableName select 1, 'a1', 10, 1000")
+
+ // Inject orphan base file directly onto the filesystem with a stale
instant timestamp
+ val orphanFileId = UUID.randomUUID().toString
+ FileCreateUtils.createBaseFile(metaClientFor(tablePath), "",
ORPHAN_INSTANT, orphanFileId)
+
+ // dry_run=true (default): should return exactly 1 row for the orphan,
touch nothing
+ val result = spark.sql(s"call repair_orphan_files(table =>
'$tableName')").collect()
+ assertResult(1)(result.length)
+
+ val row = result(0)
+ assertResult("")(row.getString(0)) // partition (root = ""
for non-partitioned)
+ assert(row.getString(1).contains(ORPHAN_INSTANT), s"file_name should
contain orphan instant: ${row.getString(1)}")
+ assertResult(ORPHAN_INSTANT)(row.getString(2)) // instant_time
+ assertResult("")(row.getString(3)) // backup_path is empty
in dry run
+ assertResult("IDENTIFIED")(row.getString(4)) // status
+ }
+ }
+
+ // Test 2 — cleanup mode backs up the orphan file and removes it from the
table path
+ test("Test Call repair_orphan_files cleanup backs up base file orphan") {
+ withTempDir { tmp =>
+ val tableName = generateTableName
+ val tablePath = s"${tmp.getCanonicalPath}/$tableName"
+ val backupDir = s"${tmp.getCanonicalPath}/backup"
+
+ spark.sql(
+ s"""create table $tableName (id int, name string, price double, ts
long)
+ |using hudi
+ |location '$tablePath'
+ |tblproperties (primaryKey = 'id', preCombineField = 'ts')
+ |""".stripMargin)
+ spark.sql(s"insert into $tableName select 1, 'a1', 10, 1000")
+
+ val orphanFileId = UUID.randomUUID().toString
+ // FileCreateUtils.createBaseFile returns the absolute path of the
created file
+ val orphanAbsPath =
FileCreateUtils.createBaseFile(metaClientFor(tablePath), "", ORPHAN_INSTANT,
orphanFileId)
+ val orphanFilePath = new Path(orphanAbsPath)
+
+ val hadoopConf = spark.sparkContext.hadoopConfiguration
+ val fs = HadoopFSUtils.getFs(tablePath, hadoopConf)
+ assert(fs.exists(orphanFilePath), "Orphan file should exist before
cleanup")
+
+ // dry_run=false: orphan should be moved to backup
+ val result = spark.sql(
+ s"""call repair_orphan_files(
+ | table => '$tableName',
+ | dry_run => false,
+ | backup_path => '$backupDir'
+ |)""".stripMargin).collect()
+
+ assertResult(1)(result.length)
+ assertResult("BACKED_UP")(result(0).getString(4))
+
+ // Orphan file must be gone from the table path
+ assert(!fs.exists(orphanFilePath), "Orphan file should no longer exist
at original path")
+
+ // Orphan file must exist at the backup path
+ val backedUpPath = new Path(result(0).getString(3))
+ assert(fs.exists(backedUpPath), "Orphan file should exist at backup
path")
+
+ // Real data must still be readable
+ assertResult(1)(spark.sql(s"select id from $tableName").collect().length)
+ }
+ }
+
+ // Test 3 — inflight commit files are skipped (RepairUtils skips
non-completed instants)
+ test("Test Call repair_orphan_files skips inflight commit files") {
+ withTempDir { tmp =>
+ val tableName = generateTableName
+ val tablePath = s"${tmp.getCanonicalPath}/$tableName"
+ val inflightTs = "20010101000000000"
+
+ spark.sql(
+ s"""create table $tableName (id int, name string, price double, ts
long)
+ |using hudi
+ |location '$tablePath'
+ |tblproperties (primaryKey = 'id', preCombineField = 'ts')
+ |""".stripMargin)
+ spark.sql(s"insert into $tableName select 1, 'a1', 10, 1000")
+
+ val metaClient = metaClientFor(tablePath)
+ // Create the inflight marker in .hoodie so the active timeline sees it
as non-completed
+ FileCreateUtils.createInflightCommit(metaClient, inflightTs)
+
+ // Place a data file with the inflight instant timestamp on disk
+ FileCreateUtils.createBaseFile(metaClient, "", inflightTs,
UUID.randomUUID().toString)
+
+ // Procedure must return 0 rows — inflight instant is excluded by
RepairUtils
+ val result = spark.sql(s"call repair_orphan_files(table =>
'$tableName')").collect()
+ assertResult(0)(result.length)
+ }
+ }
+
+ // Test 4 — backup_path validation: error when dry_run=false and no
backup_path given
+ test("Test Call repair_orphan_files requires backup_path when not dry run") {
+ withTempDir { tmp =>
+ val tableName = generateTableName
+ val tablePath = s"${tmp.getCanonicalPath}/$tableName"
+
+ spark.sql(
+ s"""create table $tableName (id int, name string, price double, ts
long)
+ |using hudi
+ |location '$tablePath'
+ |tblproperties (primaryKey = 'id', preCombineField = 'ts')
+ |""".stripMargin)
+
+ checkExceptionContain(
+ s"call repair_orphan_files(table => '$tableName', dry_run => false)"
+ )("backup_path is required")
+ }
+ }
+
+ // Test 5 — partition filter scopes the scan to a specific partition
+ test("Test Call repair_orphan_files partition filter scopes scan") {
+ withTempDir { tmp =>
+ val tableName = generateTableName
+ val tablePath = s"${tmp.getCanonicalPath}/$tableName"
+
+ spark.sql(
+ s"""create table $tableName (id int, name string, price double, ts
long)
+ |using hudi
+ |location '$tablePath'
+ |tblproperties (primaryKey = 'id', preCombineField = 'ts')
+ |""".stripMargin)
+ spark.sql(s"insert into $tableName select 1, 'a1', 10, 1000")
+
+ // Inject orphan at table root (partition = "")
+ FileCreateUtils.createBaseFile(metaClientFor(tablePath), "",
ORPHAN_INSTANT, UUID.randomUUID().toString)
+
+ // Filtering to a non-existent partition should find nothing
+ val filtered = spark.sql(
+ s"call repair_orphan_files(table => '$tableName', partition =>
'nonexistent')").collect()
+ assertResult(0)(filtered.length)
+
+ // Filtering to the root partition ("") should find the orphan
+ val root = spark.sql(
+ s"call repair_orphan_files(table => '$tableName', partition =>
'')").collect()
+ assertResult(1)(root.length)
+ assertResult("IDENTIFIED")(root(0).getString(4))
+ }
+ }
+
+ // Test 6 — log file orphan detected on a MOR table
+ test("Test Call repair_orphan_files detects log file orphan on MOR table") {
+ withTempDir { tmp =>
+ val tableName = generateTableName
+ val tablePath = s"${tmp.getCanonicalPath}/$tableName"
+
+ spark.sql(
+ s"""create table $tableName (id int, name string, price double, ts
long)
+ |using hudi
+ |location '$tablePath'
+ |tblproperties (primaryKey = 'id', preCombineField = 'ts', type =
'mor')
+ |""".stripMargin)
+ spark.sql(s"insert into $tableName select 1, 'a1', 10, 1000")
+
+ // Inject an orphan log file with a stale instant timestamp
+ val orphanFileId = UUID.randomUUID().toString
+ FileCreateUtils.createLogFile(metaClientFor(tablePath), "",
ORPHAN_INSTANT, orphanFileId, 1)
+
+ val result = spark.sql(s"call repair_orphan_files(table =>
'$tableName')").collect()
+
+ // At least the orphan log file must appear in the result
+ val orphanRows = result.filter(_.getString(2) == ORPHAN_INSTANT)
+ assert(orphanRows.length >= 1,
+ s"Expected at least 1 orphan row with instant $ORPHAN_INSTANT, got:
${result.mkString(", ")}")
+ assert(orphanRows.forall(_.getString(4) == "IDENTIFIED"))
+ }
+ }
+
+ // Note: the SKIPPED_PRESENT_IN_MDT branch (surfaces MDT-visible orphan
candidates instead of
+ // silently dropping them) is verified by inspection rather than an
end-to-end test.
+ // HoodieBackedTableMetadata.getAllFilesInPartition dynamically filters by
the data table's
+ // timeline state — deleting a timeline file immediately removes the
corresponding data file
+ // from the MDT's view as well. As a result, the (orphan ∧ in-MDT) state
cannot be constructed
+ // by manipulating the timeline alone; only direct MDT writes could produce
it, and the setup
+ // cost outweighs the value for what is a defense-in-depth path.
+
+ // Test 7 — max_orphans cap prevents driver OOM: error when detected count
exceeds the cap
+ test("Test Call repair_orphan_files max_orphans cap triggers error when
exceeded") {
+ withTempDir { tmp =>
+ val tableName = generateTableName
+ val tablePath = s"${tmp.getCanonicalPath}/$tableName"
+
+ spark.sql(
+ s"""create table $tableName (id int, name string, price double, ts
long)
+ |using hudi
+ |location '$tablePath'
+ |tblproperties (primaryKey = 'id', preCombineField = 'ts')
+ |""".stripMargin)
+ spark.sql(s"insert into $tableName select 1, 'a1', 10, 1000")
+
+ // Inject 3 orphan files, then cap at 2 — should throw
+ val metaClient = metaClientFor(tablePath)
+ FileCreateUtils.createBaseFile(metaClient, "", ORPHAN_INSTANT,
UUID.randomUUID().toString)
+ FileCreateUtils.createBaseFile(metaClient, "", ORPHAN_INSTANT,
UUID.randomUUID().toString)
+ FileCreateUtils.createBaseFile(metaClient, "", ORPHAN_INSTANT,
UUID.randomUUID().toString)
+
+ checkExceptionContain(
+ s"call repair_orphan_files(table => '$tableName', max_orphans => 2)"
+ )("exceeds max_orphans=2")
+ }
+ }
+}
