This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/paimon.git
The following commit(s) were added to refs/heads/master by this push:
new c698361110 [spark] Support remove_orphan_blobs procedure. (#9812)
c698361110 is described below
commit c698361110126ca9295c0939f209d14eb9711342
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
+ }
+ }
+}
