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 <[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]> (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") + } + } +}
