You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@hudi.apache.org by vh...@apache.org on 2026/08/05 16:02:10 UTC
(hudi) 19/19: feat(spark): add repair_orphan_files stored procedure (#19121)
This is an automated email from the ASF dual-hosted git repository.
voonhous pushed a commit to branch release-1.2.1
in repository https://gitbox.apache.org/repos/asf/hudi.git
commit a132f0af254ee6f90e54c7f58c0bf9b7264aaadb
Author: Mahsood Ebrahim <ma...@gmail.com>
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 <ma...@uber.com>
Co-authored-by: Claude Opus 4.8 (1M context) <no...@anthropic.com>
(cherry picked from commit 63a1ab19254fe46f395b4b3ae81ae0c088499626)
---
.../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 fd084ec0bf6e..fad6d39ed06c 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
@@ -85,6 +85,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")
+ }
+ }
+}