This is an automated email from the ASF dual-hosted git repository. XiaoHongbo-Hope pushed a commit to branch release-2.1 in repository https://gitbox.apache.org/repos/asf/paimon.git
commit b2a8ab824c5fd451ca3d910f6449bf4a7b58ed4a Author: Wenchao Wu <[email protected]> AuthorDate: Thu Sep 17 17:07:59 2026 +0800 [spark] Support remove_orphan_blobs procedure. (#9812) --- docs/docs/primary-key-table/blob-storage.md | 4 +- docs/docs/spark/procedures.md | 1 + docs/docs/spark/procedures/maintenance.md | 37 ++ .../org/apache/paimon/spark/SparkProcedures.java | 2 + .../procedure/RemoveOrphanBlobsProcedure.java | 171 +++++ .../SparkManagedBlobOrphanFilesCleanBase.java | 43 ++ .../SparkManagedBlobOrphanFilesClean.scala | 365 +++++++++++ .../procedure/RemoveOrphanBlobsProcedureTest.scala | 691 +++++++++++++++++++++ 8 files changed, 1313 insertions(+), 1 deletion(-) diff --git a/docs/docs/primary-key-table/blob-storage.md b/docs/docs/primary-key-table/blob-storage.md index 2b20099434..e7fd8f21f3 100644 --- a/docs/docs/primary-key-table/blob-storage.md +++ b/docs/docs/primary-key-table/blob-storage.md @@ -258,7 +258,9 @@ extra files because more than one retained data file can reference the same pack ## Garbage Collection -Unreferenced `.managed.blob` packs are reclaimed by `LocalManagedBlobOrphanFilesClean`. +Unreferenced `.managed.blob` packs are reclaimed by managed blob orphan cleanup. +Local cleanup is `LocalManagedBlobOrphanFilesClean`; Spark exposes the same cleanup as +[`remove_orphan_blobs`](../spark/procedures/maintenance#remove_orphan_blobs). The cleaner reads every retained data file's `.blobref` sidecar across snapshots, tags, and branches, then deletes packs that are not referenced and whose modification time is earlier than the absolute `older_than` cutoff (1 day before the run starts by default). diff --git a/docs/docs/spark/procedures.md b/docs/docs/spark/procedures.md index 02fa0485df..045f4bce92 100644 --- a/docs/docs/spark/procedures.md +++ b/docs/docs/spark/procedures.md @@ -67,6 +67,7 @@ Choose a group, then use the page contents to jump to a procedure: [`expire_snapshots`](./procedures/maintenance#expire_snapshots), [`expire_partitions`](./procedures/maintenance#expire_partitions), [`remove_orphan_files`](./procedures/maintenance#remove_orphan_files), +[`remove_orphan_blobs`](./procedures/maintenance#remove_orphan_blobs), [`remove_unexisting_files`](./procedures/maintenance#remove_unexisting_files), [`purge_files`](./procedures/maintenance#purge_files), [`repair`](./procedures/maintenance#repair), diff --git a/docs/docs/spark/procedures/maintenance.md b/docs/docs/spark/procedures/maintenance.md index 9952a70cc5..1b2a77a507 100644 --- a/docs/docs/spark/procedures/maintenance.md +++ b/docs/docs/spark/procedures/maintenance.md @@ -236,6 +236,8 @@ Remove the orphan data files and metadata files. - `parallelism` (`INT`, optional): The maximum number of concurrent deleting files. By default is the number of processors available to the Java virtual machine. - `mode` (`STRING`, optional): The mode of remove orphan clean procedure (local or distributed) . By default is distributed. +This procedure does not delete primary-key `.managed.blob` packs. Use [`remove_orphan_blobs`](#remove_orphan_blobs). + ```sql CALL sys.remove_orphan_files(table => 'default.T', older_than => '2023-10-31 12:00:00'); @@ -259,6 +261,41 @@ CALL sys.remove_orphan_files( ); ``` +## remove_orphan_blobs + +Remove unreferenced primary-key `.managed.blob` packs. + +**Arguments** + +- `table` (`STRING`, required): the target table identifier. Use `database_name.*` to process the whole database. +- `older_than` (`STRING`, optional): an absolute timestamp cutoff. Only packs whose modification time is earlier than this timestamp are candidates. The default cutoff is 1 day before the procedure starts. +- `dry_run` (`BOOLEAN`, optional): when true, calculate the candidate file count and total bytes without deleting files. The procedure returns aggregate counts, not individual pack paths. Default is false. +- `parallelism` (`INT`, optional): per-table concurrency. In `distributed` mode this is the Spark task parallelism of each table job (default: the larger of Spark's default parallelism and `spark.sql.shuffle.partitions`). In `local` mode this is the per-table file-operation thread limit (default: the number of processors available to the Java virtual machine). For `database_name.*`, `distributed` mode runs tables one Spark job at a time, so cluster concurrency stays within this per-table [...] +- `mode` (`STRING`, optional): The mode of remove orphan blob procedure (`local` or `distributed`). By default is `distributed`. + +```sql +CALL sys.remove_orphan_blobs(table => 'default.T', older_than => '2023-10-31 12:00:00'); + +CALL sys.remove_orphan_blobs(table => 'default.*', older_than => '2023-10-31 12:00:00'); + +CALL sys.remove_orphan_blobs(table => 'default.T', older_than => '2023-10-31 12:00:00', dry_run => true); + +CALL sys.remove_orphan_blobs( + table => 'default.T', + older_than => '2023-10-31 12:00:00', + dry_run => false, + parallelism => 5 +); + +CALL sys.remove_orphan_blobs( + table => 'default.T', + older_than => '2023-10-31 12:00:00', + dry_run => false, + parallelism => 5, + mode => 'local' +); +``` + ## remove_unexisting_files Procedure to remove unexisting data files from manifest entries. See [Java docs](https://paimon.apac diff --git a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/SparkProcedures.java b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/SparkProcedures.java index 866a54e802..672099d293 100644 --- a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/SparkProcedures.java +++ b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/SparkProcedures.java @@ -53,6 +53,7 @@ import org.apache.paimon.spark.procedure.Procedure; import org.apache.paimon.spark.procedure.ProcedureBuilder; import org.apache.paimon.spark.procedure.PurgeFilesProcedure; import org.apache.paimon.spark.procedure.ReassignRowIdProcedure; +import org.apache.paimon.spark.procedure.RemoveOrphanBlobsProcedure; import org.apache.paimon.spark.procedure.RemoveOrphanFilesProcedure; import org.apache.paimon.spark.procedure.RemoveUnexistingFilesProcedure; import org.apache.paimon.spark.procedure.RenameBranchProcedure; @@ -120,6 +121,7 @@ public class SparkProcedures { procedureBuilders.put("migrate_database", MigrateDatabaseProcedure::builder); procedureBuilders.put("migrate_table", MigrateTableProcedure::builder); procedureBuilders.put("remove_orphan_files", RemoveOrphanFilesProcedure::builder); + procedureBuilders.put("remove_orphan_blobs", RemoveOrphanBlobsProcedure::builder); procedureBuilders.put("remove_unexisting_files", RemoveUnexistingFilesProcedure::builder); procedureBuilders.put("expire_snapshots", ExpireSnapshotsProcedure::builder); procedureBuilders.put("expire_partitions", ExpirePartitionsProcedure::builder); diff --git a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/RemoveOrphanBlobsProcedure.java b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/RemoveOrphanBlobsProcedure.java new file mode 100644 index 0000000000..7d52f24ef0 --- /dev/null +++ b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/RemoveOrphanBlobsProcedure.java @@ -0,0 +1,171 @@ +/* + * 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.paimon.spark.procedure; + +import org.apache.paimon.catalog.Catalog; +import org.apache.paimon.operation.CleanOrphanFilesResult; +import org.apache.paimon.operation.LocalManagedBlobOrphanFilesClean; +import org.apache.paimon.operation.OrphanFilesClean; +import org.apache.paimon.spark.catalog.WithPaimonCatalog; +import org.apache.paimon.utils.Preconditions; + +import org.apache.spark.sql.catalyst.InternalRow; +import org.apache.spark.sql.connector.catalog.TableCatalog; +import org.apache.spark.sql.types.Metadata; +import org.apache.spark.sql.types.StructField; +import org.apache.spark.sql.types.StructType; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.util.Locale; + +import static org.apache.spark.sql.types.DataTypes.BooleanType; +import static org.apache.spark.sql.types.DataTypes.IntegerType; +import static org.apache.spark.sql.types.DataTypes.LongType; +import static org.apache.spark.sql.types.DataTypes.StringType; + +/** + * Remove orphan managed BLOB packs procedure. Usage: + * + * <pre><code> + * CALL sys.remove_orphan_blobs(table => 'tableId', [older_than => '2023-10-31 12:00:00']) + * + * CALL sys.remove_orphan_blobs(table => 'databaseName.*', [older_than => '2023-10-31 12:00:00']) + * </code></pre> + */ +public class RemoveOrphanBlobsProcedure extends BaseProcedure { + + private static final Logger LOG = + LoggerFactory.getLogger(RemoveOrphanBlobsProcedure.class.getName()); + + private static final ProcedureParameter[] PARAMETERS = + new ProcedureParameter[] { + ProcedureParameter.required("table", StringType), + ProcedureParameter.optional("older_than", StringType), + ProcedureParameter.optional("dry_run", BooleanType), + ProcedureParameter.optional("parallelism", IntegerType), + ProcedureParameter.optional("mode", StringType) + }; + + private static final StructType OUTPUT_TYPE = + new StructType( + new StructField[] { + new StructField("deletedFileCount", LongType, true, Metadata.empty()), + new StructField( + "deletedFileTotalLenInBytes", LongType, true, Metadata.empty()) + }); + + private RemoveOrphanBlobsProcedure(TableCatalog tableCatalog) { + super(tableCatalog); + } + + @Override + public ProcedureParameter[] parameters() { + return PARAMETERS; + } + + @Override + public StructType outputType() { + return OUTPUT_TYPE; + } + + @Override + public InternalRow[] call(InternalRow args) { + org.apache.paimon.catalog.Identifier identifier; + String tableId = args.getString(0); + String olderThan = args.isNullAt(1) ? null : args.getString(1); + boolean dryRun = !args.isNullAt(2) && args.getBoolean(2); + Integer parallelism = args.isNullAt(3) ? null : args.getInt(3); + + Preconditions.checkArgument( + tableId != null && !tableId.isEmpty(), + "Cannot handle an empty tableId for argument %s", + PARAMETERS[0].name()); + + if (tableId.endsWith(".*")) { + identifier = org.apache.paimon.catalog.Identifier.fromString(tableId); + } else { + identifier = + org.apache.paimon.catalog.Identifier.fromString( + toIdentifier(args.getString(0), PARAMETERS[0].name()).toString()); + } + LOG.info("identifier is {}.", identifier); + + if (parallelism != null) { + Preconditions.checkArgument( + parallelism > 0, + "Parallelism must be greater than 0, but was %s.", + parallelism); + } + + Catalog catalog = ((WithPaimonCatalog) tableCatalog()).paimonCatalog(); + String mode = args.isNullAt(4) ? "DISTRIBUTED" : args.getString(4); + + CleanOrphanFilesResult result; + try { + switch (mode.toUpperCase(Locale.ROOT)) { + case "LOCAL": + result = + LocalManagedBlobOrphanFilesClean.executeDatabase( + catalog, + identifier.getDatabaseName(), + identifier.getTableName(), + OrphanFilesClean.olderThanMillis(olderThan), + parallelism, + dryRun); + break; + case "DISTRIBUTED": + result = + SparkManagedBlobOrphanFilesClean.executeDatabase( + catalog, + identifier.getDatabaseName(), + identifier.getTableName(), + OrphanFilesClean.olderThanMillis(olderThan), + parallelism, + dryRun); + break; + default: + throw new IllegalArgumentException( + "Unknown mode: " + + mode + + ". Only 'DISTRIBUTED' and 'LOCAL' are supported."); + } + + return new InternalRow[] { + newInternalRow(result.getDeletedFileCount(), result.getDeletedFileTotalLenInBytes()) + }; + } catch (Exception e) { + throw new RuntimeException(e); + } + } + + public static ProcedureBuilder builder() { + return new BaseProcedure.Builder<RemoveOrphanBlobsProcedure>() { + @Override + public RemoveOrphanBlobsProcedure doBuild() { + return new RemoveOrphanBlobsProcedure(tableCatalog()); + } + }; + } + + @Override + public String description() { + return "RemoveOrphanBlobsProcedure"; + } +} diff --git a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/SparkManagedBlobOrphanFilesCleanBase.java b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/SparkManagedBlobOrphanFilesCleanBase.java new file mode 100644 index 0000000000..67cf3386cd --- /dev/null +++ b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/SparkManagedBlobOrphanFilesCleanBase.java @@ -0,0 +1,43 @@ +/* + * 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.paimon.spark.procedure; + +import org.apache.paimon.fs.Path; +import org.apache.paimon.operation.ManagedBlobOrphanFilesClean; +import org.apache.paimon.table.FileStoreTable; + +import java.util.function.Consumer; + +/** Java bridge for FileIO-aware managed blob candidate identities. */ +abstract class SparkManagedBlobOrphanFilesCleanBase extends ManagedBlobOrphanFilesClean { + + SparkManagedBlobOrphanFilesCleanBase( + FileStoreTable table, long olderThanMillis, boolean dryRun) { + super(table, olderThanMillis, dryRun); + } + + final String packIdentityForCandidate(Path path) { + return packIdentityForCleanup(path).orElse(SKIP_MANAGED_BLOB_GC); + } + + final void emitUsedPacksForSpark( + SidecarWorkItem workItem, ReachabilityScan scan, Consumer<String> used) { + emitUsedPacks(workItem, scan, used); + } +} diff --git a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/procedure/SparkManagedBlobOrphanFilesClean.scala b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/procedure/SparkManagedBlobOrphanFilesClean.scala new file mode 100644 index 0000000000..1a66995e62 --- /dev/null +++ b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/procedure/SparkManagedBlobOrphanFilesClean.scala @@ -0,0 +1,365 @@ +/* + * 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.paimon.spark.procedure + +import org.apache.paimon.catalog.{Catalog, Identifier} +import org.apache.paimon.fs.Path +import org.apache.paimon.manifest.{ManifestFile, ManifestFileMeta, ManifestList} +import org.apache.paimon.operation.{CleanOrphanFilesResult, ManagedBlobOrphanFilesClean} +import org.apache.paimon.operation.ManagedBlobOrphanFilesClean.SidecarWorkItem +import org.apache.paimon.operation.OrphanFilesClean.retryReadingFiles +import org.apache.paimon.table.FileStoreTable +import org.apache.paimon.utils.DataFilePathFactories +import org.apache.paimon.utils.FileStorePathFactory.BUCKET_PATH_PREFIX +import org.apache.paimon.utils.Preconditions + +import org.apache.spark.internal.Logging +import org.apache.spark.rdd.RDD +import org.apache.spark.sql.{functions, DataFrame, Dataset, PaimonSparkSession, SparkSession} +import org.apache.spark.sql.catalyst.SQLConfHelper +import org.apache.spark.storage.StorageLevel + +import java.util +import java.util.function.Consumer + +import scala.collection.JavaConverters._ +import scala.collection.mutable + +case class SparkManagedBlobOrphanFilesClean( + specifiedTable: FileStoreTable, + specifiedOlderThanMillis: Long, + parallelism: Int, + dryRunPara: Boolean, + @transient spark: SparkSession) + extends SparkManagedBlobOrphanFilesCleanBase(specifiedTable, specifiedOlderThanMillis, dryRunPara) + with SQLConfHelper + with Logging { + + def doClean(): (Dataset[(Long, Long)], Seq[Dataset[_]]) = { + import spark.implicits._ + + SparkManagedBlobOrphanFilesClean.checkParallelism(parallelism) + val cached = new mutable.ArrayBuffer[Dataset[_]]() + try { + val topologyBefore = snapshotTopology() + val usedPacks = collectUsedPacksDf().persist(StorageLevel.MEMORY_AND_DISK) + cached += usedPacks + val skipGc = usedPacks + .filter($"used_name" === ManagedBlobOrphanFilesClean.SKIP_MANAGED_BLOB_GC) + .limit(1) + .collect() + .nonEmpty + + val fileDirs = listPaimonFileDirs.asScala.map(_.toString).toSeq + val maxFileDirsParallelism = Math.min(Math.max(fileDirs.size, 1), parallelism) + val candidates = spark.sparkContext + .parallelize(fileDirs, maxFileDirsParallelism) + .flatMap { + dir => + tryBestListingDirs(new Path(dir)).asScala + .filter(file => !file.isDir) + .filter(oldEnough) + .filter( + file => ManagedBlobOrphanFilesClean.isManagedBlobPackName(file.getPath.getName)) + .map { + file => + val path = file.getPath + val parent = path.getParent + ( + packIdentityForCandidate(path), + path.toString, + file.getLen, + if (parent == null) "" else parent.toString) + } + } + .toDF("name", "path", "len", "dataDir") + .dropDuplicates("name") + .repartition(parallelism) + .persist(StorageLevel.MEMORY_AND_DISK) + cached += candidates + val candidateSkipGc = candidates + .filter($"name" === ManagedBlobOrphanFilesClean.SKIP_MANAGED_BLOB_GC) + .limit(1) + .collect() + .nonEmpty + val canonicalCandidates = candidates + .filter($"name" =!= ManagedBlobOrphanFilesClean.SKIP_MANAGED_BLOB_GC) + + betweenUsedCollections() + val usedPacks2 = collectUsedPacksDf().persist(StorageLevel.MEMORY_AND_DISK) + cached += usedPacks2 + val skipGc2 = usedPacks2 + .filter($"used_name" === ManagedBlobOrphanFilesClean.SKIP_MANAGED_BLOB_GC) + .limit(1) + .collect() + .nonEmpty + val topologyAfter = snapshotTopology() + val used1Packs = + usedPacks.filter($"used_name" =!= ManagedBlobOrphanFilesClean.SKIP_MANAGED_BLOB_GC) + val used2Packs = + usedPacks2.filter($"used_name" =!= ManagedBlobOrphanFilesClean.SKIP_MANAGED_BLOB_GC) + + val topologyChanged = topologyBefore != topologyAfter + val usedSetDifferences = used1Packs + .toDF() + .except(used2Packs.toDF()) + .union(used2Packs.toDF().except(used1Packs.toDF())) + val usedSetChanged = usedSetDifferences.limit(1).collect().nonEmpty + val frozenAbort = + skipGc || skipGc2 || candidateSkipGc || topologyChanged || usedSetChanged + + val deleted: Dataset[(Long, Long)] = if (frozenAbort) { + val reason = + if (usedSetChanged) { + "the used pack set changed during collection" + } else { + "sidecars, manifests, or candidate identities cannot be trusted, or snapshot topology changed during collection" + } + logWarning(s"Skip managed blob pack GC for table ${table.fullName()} because $reason.") + Seq.empty[(Long, Long)].toDS() + } else { + // persist() does not cut sidecar lineage. Collect the validated used names onto a + // driver-backed Dataset so the deletion join cannot recompute from stale sidecar shuffle + // after a cache miss, drop a reused pack, and still agree. + // Cost: driver memory grows with the used-pack set. An OOM fails the procedure instead of + // deleting live packs. Do not join the persisted used2Packs Dataset again; that puts sidecar + // reads back into the deletion lineage. + val frozenUsedNames = used2Packs.collect().map(_.getString(0)) + logInfo( + s"Frozen ${frozenUsedNames.length} used managed blob pack names for table ${table.fullName()} before deletion.") + val frozenUsed = frozenUsedNames.toSeq.toDF("used_name") + canonicalCandidates + .join(frozenUsed, $"name" === $"used_name", "left_anti") + .repartition(parallelism, $"dataDir") + .mapPartitions { + it => + var deletedFilesCount = 0L + var deletedFilesLenInBytes = 0L + val dataDirs = new mutable.HashSet[String]() + while (it.hasNext) { + val fileInfo = it.next() + val pathToClean = fileInfo.getString(1) + val deletedPath = new Path(pathToClean) + if (cleanManagedBlobFileIdempotently(deletedPath)) { + deletedFilesLenInBytes += fileInfo.getLong(2) + logInfo(s"Cleaned managed blob pack: $pathToClean") + dataDirs.add(fileInfo.getString(3)) + deletedFilesCount += 1 + } + } + if (!dryRun) { + val bucketDirs = dataDirs + .filter(_.contains(BUCKET_PATH_PREFIX)) + .map(new Path(_)) + tryCleanDataDirectory(bucketDirs.asJava, partitionKeysNum + 1) + } + Iterator.single((deletedFilesCount, deletedFilesLenInBytes)) + } + } + + (deleted, cached.toSeq) + } catch { + case t: Throwable => + cached.foreach(_.unpersist()) + throw t + } + } + + private[procedure] def collectUsedPacksDf(): DataFrame = { + import spark.implicits._ + val branches = validBranches() + val maxBranchParallelism = Math.min(branches.size(), parallelism) + val manifestLists = spark.sparkContext + .parallelize(branches.asScala.toSeq, maxBranchParallelism) + .flatMap { + branch => + safelyGetAllSnapshots(branch).asScala.flatMap { + snapshot => + Seq( + snapshot.changelogManifestList(), + snapshot.deltaManifestList(), + snapshot.baseManifestList()) + .filter(_ != null) + .map((branch, _)) + } + } + .distinct(parallelism) + + val manifests = manifestLists + .mapPartitions { + lists => + val branchManifestLists = new util.HashMap[String, ManifestList]() + lists.flatMap { + case (branch, listName) => + val manifestList = branchManifestLists.computeIfAbsent( + branch, + (key: String) => + specifiedTable.switchToBranch(key).store.manifestListFactory.create) + val metas = retryReadingFiles[java.util.List[ManifestFileMeta]]( + () => manifestList.readWithIOException(listName), + null) + if (metas == null) { + logWarning( + s"Manifest list $listName is missing while collecting used managed blob packs. Skip pack GC this run.") + Iterator.single((true, branch, listName)) + } else { + metas.asScala.iterator.map(meta => (false, branch, meta.fileName())) + } + } + } + .distinct(parallelism) + + val sidecarWorkItems = manifests + .mapPartitions { + records => + val branchManifestFiles = new util.HashMap[String, ManifestFile]() + val branchPathFactories = new util.HashMap[String, DataFilePathFactories]() + records.flatMap { + case (unsafe, _, _) if unsafe => + Iterator.single( + (ManagedBlobOrphanFilesClean.SKIP_MANAGED_BLOB_GC, null: SidecarWorkItem)) + case (_, branch, manifestName) => + val branchTable = specifiedTable.switchToBranch(branch) + val manifestFile = branchManifestFiles.computeIfAbsent( + branch, + (_: String) => branchTable.store.manifestFileFactory.create) + val pathFactories = branchPathFactories.computeIfAbsent( + branch, + (_: String) => new DataFilePathFactories(branchTable.store.pathFactory)) + val entries = + retryReadingFiles(() => manifestFile.readWithIOException(manifestName), null) + if (entries == null) { + logWarning( + s"Manifest $manifestName is missing while collecting used managed blob packs. Skip pack GC this run.") + Iterator.single( + (ManagedBlobOrphanFilesClean.SKIP_MANAGED_BLOB_GC, null: SidecarWorkItem)) + } else { + entries.asScala.iterator.flatMap { + entry => + createSidecarWorkItems( + entry, + pathFactories.get(entry.partition(), entry.bucket())).asScala.iterator + .map(workItem => (workItem.dedupIdentity(), workItem)) + } + } + } + } + .distinct(parallelism) + + val rawUsedPackNames = sidecarWorkItems + .mapPartitions { + records => + val scan = newReachabilityScan() + records.flatMap { + case (_, null) => + Iterator.single(ManagedBlobOrphanFilesClean.SKIP_MANAGED_BLOB_GC) + case (_, workItem) => + val names = new util.ArrayList[String]() + emitUsedPacksForSpark( + workItem, + scan, + new Consumer[String] { + override def accept(name: String): Unit = names.add(name) + }) + names.iterator().asScala + } + } + distinctUsedPackNames(rawUsedPackNames) + } + + private[procedure] def distinctUsedPackNames(raw: RDD[String]): DataFrame = { + import spark.implicits._ + raw.distinct(parallelism).toDF("used_name") + } + +} + +object SparkManagedBlobOrphanFilesClean extends SQLConfHelper { + + private def checkParallelism(parallelism: Int): Unit = { + Preconditions.checkArgument( + parallelism > 0, + "Parallelism must be greater than 0, but was %s.", + Int.box(parallelism)) + } + + def executeDatabase( + catalog: Catalog, + databaseName: String, + tableName: String, + olderThanMillis: Long, + parallelismOpt: Integer, + dryRun: Boolean): CleanOrphanFilesResult = { + val spark = PaimonSparkSession.active + val parallelism = if (parallelismOpt == null) { + Math.max(spark.sparkContext.defaultParallelism, conf.numShufflePartitions) + } else { + parallelismOpt.intValue() + } + checkParallelism(parallelism) + + val tableNames = if (tableName == null || "*" == tableName) { + catalog.listTables(databaseName).asScala + } else { + tableName :: Nil + } + val tables = tableNames.map { + tableName => + val identifier = new Identifier(databaseName, tableName) + val table = catalog.getTable(identifier) + assert( + table.isInstanceOf[FileStoreTable], + s"Only FileStoreTable supports remove-orphan-blobs action. The table type is '${table.getClass.getName}'.") + table.asInstanceOf[FileStoreTable] + } + if (tables.isEmpty) { + return new CleanOrphanFilesResult(0, 0) + } + var deletedFilesCount = 0L + var deletedFilesLenInBytes = 0L + // Run one table at a time and unpersist its marks before the next. Unlike + // SparkOrphanFilesClean, abort gating is per-table, so unioning every table's + // deletion Dataset into one job would keep all mark caches alive for the whole + // database and mix independent skip/used-set signals. + tables.foreach { + table => + val (tableDeleted, tableCached) = new SparkManagedBlobOrphanFilesClean( + table, + olderThanMillis, + parallelism, + dryRun, + spark + ).doClean() + try { + val result = tableDeleted + .toDF("deletedFilesCount", "deletedFilesLenInBytes") + .agg(functions.sum("deletedFilesCount"), functions.sum("deletedFilesLenInBytes")) + .head() + assert(result.schema.size == 2, result.schema) + if (!result.isNullAt(0)) { + deletedFilesCount += result.getLong(0) + deletedFilesLenInBytes += result.getLong(1) + } + } finally { + tableCached.foreach(_.unpersist()) + } + } + new CleanOrphanFilesResult(deletedFilesCount, deletedFilesLenInBytes) + } +} diff --git a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/procedure/RemoveOrphanBlobsProcedureTest.scala b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/procedure/RemoveOrphanBlobsProcedureTest.scala new file mode 100644 index 0000000000..b28d6e639a --- /dev/null +++ b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/procedure/RemoveOrphanBlobsProcedureTest.scala @@ -0,0 +1,691 @@ +/* + * 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.paimon.spark.procedure + +import org.apache.paimon.blob.ManagedBlobReferenceFile +import org.apache.paimon.catalog.Identifier +import org.apache.paimon.data.BinaryRow +import org.apache.paimon.fs.{FileStatus, Path, SeekableInputStream} +import org.apache.paimon.fs.local.LocalFileIO +import org.apache.paimon.manifest.FileKind +import org.apache.paimon.operation.ManagedBlobOrphanFilesClean +import org.apache.paimon.spark.PaimonSparkTestBase +import org.apache.paimon.table.{FileStoreTable, FileStoreTableFactory} +import org.apache.paimon.utils.{DataFilePathFactories, DateTimeUtils, TraceableFileIO} + +import org.apache.spark.sql.Row + +import java.io.{File, IOException} +import java.util.concurrent.ConcurrentHashMap +import java.util.concurrent.atomic.{AtomicBoolean, AtomicInteger} + +import scala.collection.JavaConverters._ + +class RemoveOrphanBlobsProcedureTest extends PaimonSparkTestBase { + + Seq("local", "distributed").foreach { + mode => + test(s"Paimon procedure: remove unreferenced managed blob pack ($mode)") { + createManagedBlobTable() + spark.sql("INSERT INTO T VALUES (1, 'a', X'0102')") + + val table = loadTable("T") + val orphan = writeOrphanPack(table, "orphan.managed.blob") + Thread.sleep(2000) + + val referenced = filesWithSuffix(table, ManagedBlobReferenceFile.MANAGED_BLOB_SUFFIX) + .filterNot(_.getName == orphan.getName) + assert(referenced.nonEmpty) + + val olderThan = DateTimeUtils.formatLocalDateTime( + DateTimeUtils.toLocalDateTime(System.currentTimeMillis()), + 3) + checkAnswer( + spark.sql( + s"CALL sys.remove_orphan_blobs(table => 'T', older_than => '$olderThan', mode => '$mode')"), + Row(1, 3) :: Nil) + + assert(!table.fileIO().exists(orphan)) + referenced.foreach(pack => assert(table.fileIO().exists(pack))) + } + + test(s"Paimon procedure: dry run remove orphan blobs ($mode)") { + createManagedBlobTable() + spark.sql("INSERT INTO T VALUES (1, 'a', X'0102')") + + val table = loadTable("T") + val orphan = writeOrphanPack(table, "orphan.managed.blob") + Thread.sleep(2000) + + val olderThan = DateTimeUtils.formatLocalDateTime( + DateTimeUtils.toLocalDateTime(System.currentTimeMillis()), + 3) + checkAnswer( + spark.sql( + s"CALL sys.remove_orphan_blobs(table => 'T', older_than => '$olderThan', dry_run => true, mode => '$mode')"), + Row(1, 3) :: Nil) + assert(table.fileIO().exists(orphan)) + } + + test(s"Paimon procedure: skip managed blob pack gc when sidecar missing ($mode)") { + createManagedBlobTable() + spark.sql("INSERT INTO T VALUES (1, 'a', X'0102')") + + val table = loadTable("T") + val orphanPack = new Path(bucketPath(table), "orphan.managed.blob") + val orphanOther = new Path(bucketPath(table), "orphan.txt") + table.fileIO().newOutputStream(orphanPack, false).close() + table.fileIO().writeFile(orphanOther, "x", true) + Thread.sleep(2000) + + val referenced = filesWithSuffix(table, ManagedBlobReferenceFile.MANAGED_BLOB_SUFFIX) + .filterNot(_.getName == orphanPack.getName) + assert(referenced.nonEmpty) + filesWithSuffix(table, ManagedBlobReferenceFile.REFERENCE_FILE_SUFFIX) + .foreach(table.fileIO().deleteQuietly) + + val olderThan = DateTimeUtils.formatLocalDateTime( + DateTimeUtils.toLocalDateTime(System.currentTimeMillis()), + 3) + spark.sql( + s"CALL sys.remove_orphan_blobs(table => 'T', older_than => '$olderThan', mode => '$mode')") + + assert(table.fileIO().exists(orphanPack)) + assert(table.fileIO().exists(orphanOther)) + referenced.foreach(pack => assert(table.fileIO().exists(pack))) + } + } + + test("Paimon procedure: preserve distributed deletion parallelism") { + createManagedBlobTable() + spark.sql("INSERT INTO T VALUES (1, 'a', X'0102')") + + val table = loadTable("T") + val orphan = new Path(bucketPath(table), "orphan.managed.blob") + table.fileIO().newOutputStream(orphan, false).close() + Thread.sleep(2000) + + withSQLConf("spark.sql.shuffle.partitions" -> "7") { + val cleaner = + SparkManagedBlobOrphanFilesClean(table, System.currentTimeMillis(), 2, true, spark) + val (deleted, cached) = cleaner.doClean() + try { + assert(deleted.rdd.getNumPartitions == 2, deleted.queryExecution.executedPlan) + } finally { + cached.foreach(_.unpersist()) + } + } + } + + Seq("local", "distributed").foreach { + mode => + test(s"Paimon procedure: reject non-positive parallelism ($mode)") { + createManagedBlobTable() + + val error = intercept[Exception] { + spark + .sql(s"CALL sys.remove_orphan_blobs(table => 'T', parallelism => 0, mode => '$mode')") + .collect() + } + assert(causeMessages(error).contains("Parallelism must be greater than 0, but was 0.")) + } + } + + Seq("local", "distributed").foreach { + mode => + test(s"Paimon procedure: remove database orphan blobs ($mode)") { + createManagedBlobTable("T1") + spark.sql("INSERT INTO T1 VALUES (1, 'a', X'0102')") + createManagedBlobTable("T2") + spark.sql("INSERT INTO T2 VALUES (1, 'a', X'0102')") + + try { + val table1 = loadTable("T1") + val table2 = loadTable("T2") + val orphan1 = writeOrphanPack(table1, "orphan.managed.blob") + val orphan2 = writeOrphanPack(table2, "orphan.managed.blob") + Thread.sleep(2000) + + val olderThan = DateTimeUtils.formatLocalDateTime( + DateTimeUtils.toLocalDateTime(System.currentTimeMillis()), + 3) + // beforeEach only drops T. Row(2, 6) is the exact sum of these two 3-byte orphans. + checkAnswer( + spark.sql( + s"CALL sys.remove_orphan_blobs(table => 'test.*', older_than => '$olderThan', mode => '$mode')"), + Row(2, 6) :: Nil) + assert(!table1.fileIO().exists(orphan1)) + assert(!table2.fileIO().exists(orphan2)) + } finally { + spark.sql("DROP TABLE IF EXISTS T1") + spark.sql("DROP TABLE IF EXISTS T2") + } + } + } + + test("Paimon procedure: release cached marks after distributed failure") { + createManagedBlobTable() + spark.sql("INSERT INTO T VALUES (1, 'a', X'0102')") + + val persistentBefore = spark.sparkContext.getPersistentRDDs.keySet + val cleaner = new SparkManagedBlobOrphanFilesClean( + loadTable("T"), + System.currentTimeMillis(), + 2, + true, + spark) { + override protected def betweenUsedCollections(): Unit = + throw new RuntimeException("Expected failure after the first mark.") + } + + val error = intercept[RuntimeException] { + cleaner.doClean() + } + assert(error.getMessage == "Expected failure after the first mark.") + assert(spark.sparkContext.getPersistentRDDs.keySet == persistentBefore) + } + + test("Paimon procedure: log and freeze used pack set changes") { + val sparkSession = spark + import sparkSession.implicits._ + createManagedBlobTable() + + val table = loadTable("T") + val orphan = new Path(bucketPath(table), "used-set-changed.managed.blob") + table.fileIO().mkdirs(orphan.getParent) + table.fileIO().newOutputStream(orphan, false).close() + val markPass = new AtomicInteger() + val warnings = new scala.collection.mutable.ArrayBuffer[String]() + val cleaner = new SparkManagedBlobOrphanFilesClean(table, Long.MaxValue, 2, false, spark) { + override private[procedure] def collectUsedPacksDf(): org.apache.spark.sql.DataFrame = + Seq(if (markPass.getAndIncrement() == 0) "first-pack" else "second-pack") + .toDF("used_name") + + override protected def logWarning(msg: => String): Unit = warnings += msg + } + + val (deleted, cached) = cleaner.doClean() + try { + assert(deleted.collect().map(_._1).sum == 0) + } finally { + cached.foreach(_.unpersist()) + } + assert(table.fileIO().exists(orphan)) + assert(warnings.exists(_.contains("used pack set changed during collection"))) + } + + test("Paimon procedure: frozen used mark survives stale sidecar recompute") { + val sparkSession = spark + import sparkSession.implicits._ + createManagedBlobTable() + spark.sql("INSERT INTO T VALUES (1, 'a', X'0102')") + + val table = loadTable("T") + val referenced = filesWithSuffix(table, ManagedBlobReferenceFile.MANAGED_BLOB_SUFFIX) + assert(referenced.nonEmpty) + val orphan = writeOrphanPack(table, "orphan.managed.blob") + val liveIdentities = referenced.map(ManagedBlobOrphanFilesClean.packIdentity) + val markPass = new AtomicInteger() + val cleaner = new SparkManagedBlobOrphanFilesClean(table, Long.MaxValue, 2, false, spark) { + override private[procedure] def collectUsedPacksDf(): org.apache.spark.sql.DataFrame = { + if (markPass.incrementAndGet() <= 2) { + liveIdentities.toDF("used_name") + } else { + Seq.empty[String].toDF("used_name") + } + } + } + + val (deleted, cached) = cleaner.doClean() + try { + cached.foreach(_.unpersist(true)) + clearShuffleOutputs() + assert(deleted.collect().map(_._1).sum == 1) + assert(markPass.get() == 2) + } finally { + cached.foreach(_.unpersist()) + } + referenced.foreach(pack => assert(table.fileIO().exists(pack))) + assert(!table.fileIO().exists(orphan)) + } + + test("Paimon procedure: skip cleanup when candidate identity cannot be canonicalized") { + createManagedBlobTable() + spark.sql("INSERT INTO T VALUES (1, 'a', X'0102')") + + val table = loadTable("T") + val orphan = new Path(bucketPath(table), "orphan.managed.blob") + table.fileIO().newOutputStream(orphan, false).close() + val relativeTable = FileStoreTableFactory.create( + new RemoveOrphanBlobsProcedureTest.RelativeListingFileIO, + table.location(), + table.schema()) + val cleaner = SparkManagedBlobOrphanFilesClean(relativeTable, Long.MaxValue, 2, false, spark) + + val (deleted, cached) = cleaner.doClean() + try { + assert(deleted.collect().map(_._1).sum == 0) + } finally { + cached.foreach(_.unpersist(true)) + } + assert(table.fileIO().exists(orphan)) + } + + test("Paimon procedure: frozen used mark still deletes after sidecar reads start failing") { + createManagedBlobTable() + spark.sql("INSERT INTO T VALUES (1, 'a', X'0102')") + + val table = loadTable("T") + val orphan = writeOrphanPack(table, "orphan.managed.blob") + val referenced = filesWithSuffix(table, ManagedBlobReferenceFile.MANAGED_BLOB_SUFFIX) + .filterNot(_.getName == orphan.getName) + assert(referenced.nonEmpty) + + RemoveOrphanBlobsProcedureTest.failSidecarReads.set(false) + val failingTable = FileStoreTableFactory.create( + new RemoveOrphanBlobsProcedureTest.FailingSidecarFileIO, + table.location(), + table.schema()) + val cleaner = + SparkManagedBlobOrphanFilesClean(failingTable, Long.MaxValue, 2, false, spark) + val (deleted, cached) = cleaner.doClean() + try { + RemoveOrphanBlobsProcedureTest.failedSidecarReadCount.set(0) + cached.foreach(_.unpersist(true)) + clearShuffleOutputs() + RemoveOrphanBlobsProcedureTest.failSidecarReads.set(true) + assert(deleted.collect().map(_._1).sum == 1) + assert(RemoveOrphanBlobsProcedureTest.failedSidecarReadCount.get() == 0) + } finally { + cached.foreach(_.unpersist()) + RemoveOrphanBlobsProcedureTest.failSidecarReads.set(false) + } + assert(!table.fileIO().exists(orphan)) + referenced.foreach(pack => assert(table.fileIO().exists(pack))) + } + + test("Paimon procedure: observed unsafe marks survive cache recomputation") { + createManagedBlobTable() + spark.sql("INSERT INTO T VALUES (1, 'a', X'0102')") + + val table = loadTable("T") + val orphan = new Path(bucketPath(table), "orphan.managed.blob") + table.fileIO().newOutputStream(orphan, false).close() + val referenced = filesWithSuffix(table, ManagedBlobReferenceFile.MANAGED_BLOB_SUFFIX) + .filterNot(_.getName == orphan.getName) + assert(referenced.nonEmpty) + + RemoveOrphanBlobsProcedureTest.failSidecarReads.set(true) + val failingTable = FileStoreTableFactory.create( + new RemoveOrphanBlobsProcedureTest.FailingSidecarFileIO, + table.location(), + table.schema()) + val cleaner = + SparkManagedBlobOrphanFilesClean(failingTable, Long.MaxValue, 2, false, spark) + val (deleted, cached) = cleaner.doClean() + try { + cached.foreach(_.unpersist(true)) + RemoveOrphanBlobsProcedureTest.failSidecarReads.set(false) + assert(deleted.collect().map(_._1).sum == 0) + } finally { + cached.foreach(_.unpersist()) + RemoveOrphanBlobsProcedureTest.failSidecarReads.set(false) + } + assert(table.fileIO().exists(orphan)) + referenced.foreach(pack => assert(table.fileIO().exists(pack))) + } + + test("Paimon procedure: distinct used pack identities before caching the mark") { + createManagedBlobTable() + spark.sql("INSERT INTO T VALUES (1, 'a', X'0102')") + spark.sql("INSERT INTO T VALUES (2, 'b', X'0304')") + spark.sql("CALL sys.compact(table => 'T')") + + val table = loadTable("T") + val cleaner = SparkManagedBlobOrphanFilesClean(table, Long.MaxValue, 2, true, spark) + val identity = cleaner.collectUsedPacksDf().head().getString(0) + val raw = spark.sparkContext.parallelize(Seq(identity, identity), 2) + assert(raw.count() == 2) + assert(raw.distinct().count() == 1) + assert( + cleaner + .distinctUsedPackNames(raw) + .collect() + .map(_.getString(0)) + .sameElements(Array(identity))) + } + + test("Paimon procedure: distinct used pack identities from duplicate sidecars") { + val sparkSession = spark + import sparkSession.implicits._ + createManagedBlobTable() + spark.sql("INSERT INTO T VALUES (1, 'a', X'0102')") + spark.sql("INSERT INTO T VALUES (2, 'b', X'0304')") + + val table = loadTable("T") + val sidecars = filesWithSuffix(table, ManagedBlobReferenceFile.REFERENCE_FILE_SUFFIX) + assert(sidecars.size >= 2, s"Expected two managed BLOB sidecars: $sidecars") + + val firstSidecar = sidecars.head + val secondSidecar = sidecars.tail.head + val firstReferences = ManagedBlobReferenceFile.read(table.fileIO(), firstSidecar) + assert(!firstReferences.isEmpty) + table.fileIO().delete(secondSidecar, false) + ManagedBlobReferenceFile.write(table.fileIO(), secondSidecar, firstReferences) + + val sharedPack = ManagedBlobOrphanFilesClean.packIdentity(firstReferences.get(0).toPath()) + val sidecarsReferencingSharedPack = Seq(firstSidecar, secondSidecar).count { + sidecar => + ManagedBlobReferenceFile + .read(table.fileIO(), sidecar) + .asScala + .exists( + reference => ManagedBlobOrphanFilesClean.packIdentity(reference.toPath()) == sharedPack) + } + assert(sidecarsReferencingSharedPack == 2) + + val cleaner = SparkManagedBlobOrphanFilesClean(table, Long.MaxValue, 2, true, spark) + val used = cleaner.collectUsedPacksDf() + assert(used.filter($"used_name" === sharedPack).count() == 1) + } + + test("Paimon procedure: canonical alias candidates are counted once") { + createManagedBlobTable() + val table = loadTable("T") + val orphan = new Path(bucketPath(table), "duplicate.managed.blob") + table.fileIO().mkdirs(orphan.getParent) + table.fileIO().newOutputStream(orphan, false).close() + + val duplicateListingTable = FileStoreTableFactory.create( + new RemoveOrphanBlobsProcedureTest.CanonicalAliasListingFileIO, + table.location(), + table.schema()) + val aliases = duplicateListingTable.fileIO + .listStatus(orphan.getParent) + .filter(_.getPath.getName == orphan.getName) + assert(aliases.map(_.getPath.toString).distinct.length == 2) + assert( + aliases + .map(status => ManagedBlobOrphanFilesClean.packIdentity(status.getPath)) + .distinct + .length == 1) + val cleaner = + SparkManagedBlobOrphanFilesClean(duplicateListingTable, Long.MaxValue, 2, false, spark) + val (deleted, cached) = cleaner.doClean() + try { + assert(deleted.collect().map(_._1).sum == 1) + } finally { + cached.foreach(_.unpersist()) + } + assert(!table.fileIO().exists(orphan)) + } + + test("Paimon procedure: deduplicate shared sidecar globally in each mark pass") { + createManagedBlobTable() + spark.sql("INSERT INTO T VALUES (1, 'a', X'0102')") + spark.sql("INSERT INTO T VALUES (2, 'b', X'0304')") + + val table = loadTable("T") + val commit = table.newBatchWriteBuilder().newCommit() + try { + commit.compactManifests() + } finally { + commit.close() + } + + val occurrences = sidecarManifestOccurrences(table) + val (sidecar, containingManifests) = occurrences + .find(_._2.distinct.size >= 2) + .getOrElse(fail(s"No sidecar is shared by distinct manifests: $occurrences")) + val parallelism = (2 to 64) + .find { + candidate => + containingManifests + .map { + manifest => + Math.floorMod( + (false, Identifier.DEFAULT_MAIN_BRANCH, manifest).hashCode(), + candidate) + } + .distinct + .size >= 2 + } + .getOrElse(fail(s"Shared manifests cannot be assigned to different partitions: $occurrences")) + + RemoveOrphanBlobsProcedureTest.resetSidecarReadCounts() + val countingTable = FileStoreTableFactory.create( + new RemoveOrphanBlobsProcedureTest.CountingSidecarFileIO, + table.location(), + table.schema()) + val cleaner = + SparkManagedBlobOrphanFilesClean(countingTable, Long.MaxValue, parallelism, true, spark) + + val (deleted, cached) = cleaner.doClean() + try { + deleted.collect() + } finally { + cached.foreach(_.unpersist()) + } + assert(RemoveOrphanBlobsProcedureTest.sidecarReadCount(sidecar) == 2) + } + + private def createManagedBlobTable(name: String = "T"): Unit = { + spark.sql(s""" + |CREATE TABLE $name (id INT, name STRING, payload BINARY) + |USING PAIMON + |TBLPROPERTIES ( + | 'primary-key'='id', + | 'bucket'='1', + | 'changelog-producer'='none', + | 'blob-field'='payload') + |""".stripMargin) + } + + private def writeOrphanPack( + table: FileStoreTable, + fileName: String, + payload: Array[Byte] = Array[Byte](1, 2, 3)): Path = { + val orphan = new Path(bucketPath(table), fileName) + val out = table.fileIO().newOutputStream(orphan, false) + try { + out.write(payload) + } finally { + out.close() + } + orphan + } + + private def bucketPath(table: FileStoreTable): Path = { + table.store().pathFactory().bucketPath(BinaryRow.EMPTY_ROW, 0) + } + + private def filesWithSuffix(table: FileStoreTable, suffix: String): Seq[Path] = { + val statuses = table.fileIO().listStatus(bucketPath(table)) + if (statuses == null) { + Seq.empty + } else { + statuses.map(_.getPath).filter(_.getName.endsWith(suffix)) + } + } + + private def sidecarManifestOccurrences(table: FileStoreTable): Map[String, Seq[String]] = { + val manifestList = table.store().manifestListFactory().create() + val manifestFile = table.store().manifestFileFactory().create() + val pathFactories = new DataFilePathFactories(table.store().pathFactory()) + val manifests = table + .snapshotManager() + .safelyGetAllSnapshots() + .asScala + .flatMap(snapshot => manifestList.readDataManifests(snapshot).asScala) + .map(_.fileName()) + .distinct + manifests + .flatMap { + manifest => + manifestFile + .read(manifest) + .asScala + .filter(_.kind() == FileKind.ADD) + .flatMap { + entry => + val dataFile = pathFactories + .get(entry.partition(), entry.bucket()) + .toPath(entry) + Option(entry.file().extraFiles()).toSeq + .flatMap(_.asScala) + .filter(_.endsWith(ManagedBlobReferenceFile.REFERENCE_FILE_SUFFIX)) + .map(extra => (new Path(dataFile.getParent, extra).toUri.getPath, manifest)) + } + } + .groupBy(_._1) + .map { case (sidecar, values) => (sidecar, values.map(_._2).toList) } + } + + private def causeMessages(error: Throwable): String = { + Iterator + .iterate(error)(_.getCause) + .takeWhile(_ != null) + .flatMap(e => Option(e.getMessage)) + .mkString("\n") + } + + // Drop Spark shuffle map outputs so a later action recomputes `deleted` instead of replaying + // cached shuffle blocks. Used to prove the frozen used-pack Dataset does not fall back to + // sidecar lineage after cache eviction. Relies on Spark's private + // MapOutputTracker.unregisterAllMapAndMergeOutput. + private def clearShuffleOutputs(): Unit = { + val sparkEnv = spark.sparkContext.getClass.getMethod("env").invoke(spark.sparkContext) + val mapOutputTracker = + sparkEnv.getClass.getMethod("mapOutputTracker").invoke(sparkEnv) + val shuffleStatuses = mapOutputTracker.getClass + .getMethod("shuffleStatuses") + .invoke(mapOutputTracker) + .asInstanceOf[scala.collection.Map[Int, _]] + val unregister = mapOutputTracker.getClass + .getMethod("unregisterAllMapAndMergeOutput", Integer.TYPE) + shuffleStatuses.keys.toSeq.foreach(id => unregister.invoke(mapOutputTracker, Int.box(id))) + } +} + +private object RemoveOrphanBlobsProcedureTest { + + private val sidecarReadCounts = new ConcurrentHashMap[String, AtomicInteger]() + private val failSidecarReads = new AtomicBoolean() + private val failedSidecarReadCount = new AtomicInteger() + + private def resetSidecarReadCounts(): Unit = sidecarReadCounts.clear() + + private def sidecarReadCount(path: String): Int = + Option(sidecarReadCounts.get(path)).map(_.get()).getOrElse(0) + + private class CountingSidecarFileIO extends LocalFileIO { + + override def newInputStream(path: Path): SeekableInputStream = { + if (path.getName.endsWith(ManagedBlobReferenceFile.REFERENCE_FILE_SUFFIX)) { + sidecarReadCounts + .computeIfAbsent(path.toUri.getPath, (_: String) => new AtomicInteger()) + .incrementAndGet() + } + super.newInputStream(path) + } + } + + private class FailingSidecarFileIO extends LocalFileIO { + + override def newInputStream(path: Path): SeekableInputStream = { + if ( + failSidecarReads.get() && path.getName.endsWith( + ManagedBlobReferenceFile.REFERENCE_FILE_SUFFIX) + ) { + failedSidecarReadCount.incrementAndGet() + throw new IOException("Injected sidecar read failure.") + } + super.newInputStream(path) + } + } + + private class CanonicalAliasListingFileIO extends TraceableFileIO { + + override def listStatus(path: Path): Array[FileStatus] = { + val statuses = super.listStatus(path) + if (statuses == null) { + null + } else { + statuses.flatMap { + status => + if (status.getPath.getName.endsWith(ManagedBlobReferenceFile.MANAGED_BLOB_SUFFIX)) { + val alias = + new Path(s"hdfs://duplicate-listing${status.getPath.toUri.getPath}") + Array(status, withPath(status, alias)) + } else { + Array(status) + } + } + } + } + + private def withPath(status: FileStatus, path: Path): FileStatus = new FileStatus { + override def getLen: Long = status.getLen + + override def isDir: Boolean = status.isDir + + override def getPath: Path = path + + override def getModificationTime: Long = status.getModificationTime + } + } + + private class RelativeListingFileIO extends TraceableFileIO { + + override def listStatus(path: Path): Array[FileStatus] = { + val statuses = super.listStatus(path) + if (statuses == null) { + null + } else { + statuses.map { + status => + if (status.getPath.getName.endsWith(ManagedBlobReferenceFile.MANAGED_BLOB_SUFFIX)) { + withPath(status, new Path(status.getPath.getName)) + } else { + status + } + } + } + } + + override def getFileStatus(path: Path): FileStatus = { + val status = super.getFileStatus(path) + if (new File(path.toUri.getPath).isAbsolute) { + status + } else { + withPath(status, path) + } + } + + private def withPath(status: FileStatus, path: Path): FileStatus = new FileStatus { + override def getLen: Long = status.getLen + + override def isDir: Boolean = status.isDir + + override def getPath: Path = path + + override def getModificationTime: Long = status.getModificationTime + } + } +}
