JingsongLi commented on code in PR #9207: URL: https://github.com/apache/paimon/pull/9207#discussion_r3791106328
########## paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/procedure/SparkManagedBlobOrphanFilesClean.scala: ########## @@ -0,0 +1,222 @@ +/* + * 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.Snapshot +import org.apache.paimon.catalog.{Catalog, Identifier} +import org.apache.paimon.fs.Path +import org.apache.paimon.operation.{CleanOrphanFilesResult, ManagedBlobOrphanFilesClean} +import org.apache.paimon.table.FileStoreTable +import org.apache.paimon.utils.FileStorePathFactory.BUCKET_PATH_PREFIX + +import org.apache.spark.internal.Logging +import org.apache.spark.sql.{functions, Dataset, PaimonSparkSession, SparkSession} +import org.apache.spark.sql.catalyst.SQLConfHelper + +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 ManagedBlobOrphanFilesClean(specifiedTable, specifiedOlderThanMillis, dryRunPara) + with SQLConfHelper + with Logging { + + def doClean(): (Dataset[(Long, Long)], Seq[Dataset[_]]) = { + import spark.implicits._ + + val topologyBefore = snapshotTopology() + val usedPacks = collectUsedPacksDf().cache() + 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 + ( + ManagedBlobOrphanFilesClean.packIdentity(path), + path.toString, + file.getLen, + path.getParent.toString) + } + } + .toDF("name", "path", "len", "dataDir") + .repartition(parallelism) + + betweenUsedCollections() + val usedPacks2 = collectUsedPacksDf().cache() + val skipGc2 = usedPacks2 + .filter($"used_name" === ManagedBlobOrphanFilesClean.SKIP_MANAGED_BLOB_GC) + .limit(1) + .collect() + .nonEmpty + val topologyAfter = snapshotTopology() + val used1Packs: Dataset[_] = + usedPacks.filter($"used_name" =!= ManagedBlobOrphanFilesClean.SKIP_MANAGED_BLOB_GC) + val used2Packs: Dataset[_] = + usedPacks2.filter($"used_name" =!= ManagedBlobOrphanFilesClean.SKIP_MANAGED_BLOB_GC) + val usedChanged = used1Packs + .toDF() + .except(used2Packs.toDF()) + .union(used2Packs.toDF().except(used1Packs.toDF())) + .limit(1) + .count() > 0 + val abort = skipGc || skipGc2 || topologyBefore != topologyAfter || usedChanged + if (abort) { + logWarning( + s"Skip managed blob pack GC for table ${table.fullName()} because sidecars cannot be trusted or used packs changed during collection.") + } + + val unused = candidates.join(used2Packs.toDF(), $"name" === $"used_name", "left_anti") + val toDelete = if (abort) unused.limit(0) else unused + + val deleted: Dataset[(Long, Long)] = toDelete + .repartition($"dataDir") Review Comment: [P2] Preserve the requested deletion parallelism The column-only `repartition($"dataDir")` overload uses `spark.sql.shuffle.partitions`, replacing the `parallelism` limit applied earlier. For example, `parallelism => 1` can still produce hundreds of concurrent deletion tasks when the session shuffle partition count is 200. Please use `repartition(parallelism, $"dataDir")` so the documented maximum concurrency is honored while records remain grouped by data directory. ########## paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/orphan/FlinkManagedBlobOrphanFilesClean.java: ########## @@ -0,0 +1,434 @@ +/* + * 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.flink.orphan; + +import org.apache.paimon.Snapshot; +import org.apache.paimon.catalog.Catalog; +import org.apache.paimon.catalog.Identifier; +import org.apache.paimon.flink.utils.BoundedOneInputOperator; +import org.apache.paimon.flink.utils.BoundedTwoInputOperator; +import org.apache.paimon.fs.FileStatus; +import org.apache.paimon.fs.Path; +import org.apache.paimon.operation.CleanOrphanFilesResult; +import org.apache.paimon.operation.ManagedBlobOrphanFilesClean; +import org.apache.paimon.table.FileStoreTable; +import org.apache.paimon.table.Table; +import org.apache.paimon.utils.FileStorePathFactory; + +import org.apache.flink.api.common.RuntimeExecutionMode; +import org.apache.flink.api.common.typeinfo.TypeInformation; +import org.apache.flink.api.java.tuple.Tuple2; +import org.apache.flink.configuration.Configuration; +import org.apache.flink.configuration.CoreOptions; +import org.apache.flink.configuration.ExecutionOptions; +import org.apache.flink.streaming.api.datastream.DataStream; +import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator; +import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment; +import org.apache.flink.streaming.api.functions.ProcessFunction; +import org.apache.flink.streaming.api.functions.sink.v2.DiscardingSink; +import org.apache.flink.streaming.api.operators.InputSelection; +import org.apache.flink.streaming.runtime.streamrecord.StreamRecord; +import org.apache.flink.util.CloseableIterator; +import org.apache.flink.util.Collector; +import org.apache.flink.util.OutputTag; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import javax.annotation.Nullable; + +import java.util.ArrayList; +import java.util.Collections; +import java.util.HashSet; +import java.util.List; +import java.util.Set; + +import static org.apache.flink.api.common.typeinfo.BasicTypeInfo.STRING_TYPE_INFO; +import static org.apache.flink.util.Preconditions.checkState; +import static org.apache.paimon.utils.FileStorePathFactory.BUCKET_PATH_PREFIX; +import static org.apache.paimon.utils.Preconditions.checkArgument; + +/** Flink {@link ManagedBlobOrphanFilesClean}. */ +public class FlinkManagedBlobOrphanFilesClean extends ManagedBlobOrphanFilesClean { + + private static final Logger LOG = + LoggerFactory.getLogger(FlinkManagedBlobOrphanFilesClean.class); + + @Nullable private final Integer parallelism; + + public FlinkManagedBlobOrphanFilesClean( + FileStoreTable table, + long olderThanMillis, + boolean dryRun, + @Nullable Integer parallelism) { + super(table, olderThanMillis, dryRun); + this.parallelism = parallelism; + } + + @Nullable + public DataStream<CleanOrphanFilesResult> doClean(StreamExecutionEnvironment env) { + List<String> topologyBefore; + try { + topologyBefore = snapshotTopology(); + } catch (java.io.IOException e) { + throw new RuntimeException(e); + } + + Configuration flinkConf = new Configuration(); + flinkConf.set(ExecutionOptions.RUNTIME_MODE, RuntimeExecutionMode.BATCH); + flinkConf.set(ExecutionOptions.SORT_INPUTS, false); + flinkConf.set(ExecutionOptions.USE_BATCH_STATE_BACKEND, false); + if (parallelism != null) { + flinkConf.set(CoreOptions.DEFAULT_PARALLELISM, parallelism); + } + flinkConf.setString("execution.batch.adaptive.auto-parallelism.enabled", "false"); + env.configure(flinkConf); + + List<String> branches = validBranches(); + final OutputTag<Boolean> skipGcTag = new OutputTag<Boolean>("managed-blob-gc-skip") {}; + SingleOutputStreamOperator<String> usedPacks = + env.fromCollection(branches) + .name("branch-source") + .process( + new ProcessFunction<String, String>() { + @Override + public void processElement( + String branch, + ProcessFunction<String, String>.Context ctx, + Collector<String> out) + throws Exception { + for (Snapshot snapshot : safelyGetAllSnapshots(branch)) { + emitUsedPacks( + branch, + snapshot, + identity -> { + if (SKIP_MANAGED_BLOB_GC.equals(identity)) { + ctx.output(skipGcTag, Boolean.TRUE); + } else { + out.collect(identity); + } + }); + } + } + }) + .name("collect-used-packs") + .returns(STRING_TYPE_INFO); + + SingleOutputStreamOperator<String> usedPacks2 = + usedPacks + .transform( + "re-mark-used-packs", + STRING_TYPE_INFO, + new BoundedOneInputOperator<String, String>() { + + private final Set<String> used = new HashSet<>(); + + @Override + public void processElement(StreamRecord<String> element) { + used.add(element.getValue()); + } + + @Override + public void endInput() throws Exception { + Set<String> used2 = collectUsedPacks(); Review Comment: [P2] Keep the second mark and set comparison distributed This operator is forced to parallelism 1, retains the entire first-pass live-pack set in `used`, and then builds a second complete `HashSet` through `collectUsedPacks()` on the same task. A large managed-BLOB table can therefore OOM or stall one TaskManager regardless of the requested parallelism, which defeats the distributed mode. Please collect the second pass as a keyed stream, compute the symmetric difference distributively, and broadcast only the final mismatch/skip flag. ########## paimon-core/src/main/java/org/apache/paimon/operation/ManagedBlobOrphanFilesClean.java: ########## @@ -0,0 +1,266 @@ +/* + * 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.operation; + +import org.apache.paimon.Snapshot; +import org.apache.paimon.blob.ManagedBlobReachabilityCollector; +import org.apache.paimon.blob.ManagedBlobReachabilityCollector.Result; +import org.apache.paimon.blob.ManagedBlobReferenceFile; +import org.apache.paimon.blob.ManagedBlobReferenceFile.Reference; +import org.apache.paimon.fs.Path; +import org.apache.paimon.io.DataFilePathFactory; +import org.apache.paimon.manifest.FileKind; +import org.apache.paimon.manifest.ManifestEntry; +import org.apache.paimon.manifest.ManifestFile; +import org.apache.paimon.manifest.ManifestFileMeta; +import org.apache.paimon.manifest.ManifestList; +import org.apache.paimon.table.FileStoreTable; +import org.apache.paimon.utils.DataFilePathFactories; + +import java.io.IOException; +import java.net.URI; +import java.util.ArrayList; +import java.util.Collections; +import java.util.HashSet; +import java.util.List; +import java.util.Set; +import java.util.function.Consumer; + +/** + * Cleans unreferenced primary-key {@code .managed.blob} packs. + * + * <p>Unlike {@link OrphanFilesClean}, this cleaner only lists and deletes managed BLOB packs. Pack + * reachability is collected from live {@link FileKind#ADD} data-file {@code .blobref} sidecars. + * Missing manifest lists or unreadable sidecars on a still-existing data file abort pack deletion + * for the rest of the run. + * + * <p>Used packs are collected twice. If the snapshot topology or the used-pack set changes between + * those collections, this run deletes nothing. That shrinks the race with compaction reuse; it is + * not a commit lease. + */ +public abstract class ManagedBlobOrphanFilesClean extends OrphanFilesClean { + + /** + * Marker emitted into the used-pack set when a {@code .blobref} sidecar or a required manifest + * cannot be trusted. Callers must skip deleting every {@code .managed.blob} pack. + */ + public static final String SKIP_MANAGED_BLOB_GC = "__paimon_skip_managed_blob_gc__"; + + public ManagedBlobOrphanFilesClean(FileStoreTable table, long olderThanMillis, boolean dryRun) { + super(table, olderThanMillis, dryRun); + } + + /** + * Join key for a managed pack. Identity is {@code storageRootId + relativePath}, reconstructed + * by {@link Reference#toPath()}. The FileIO scheme is omitted so a wrapper such as test {@code + * traceable:} still matches listed {@code file:} paths at the same location. + */ + public static String packIdentity(Path packPath) { + URI uri = packPath.toUri(); + String authority = uri.getAuthority(); + String path = uri.getPath(); + if (authority == null || authority.isEmpty()) { + return path; + } + return authority + path; + } + + public static String packIdentity(Reference reference) { + return packIdentity(reference.toPath()); + } + + /** + * Sorted {@code branch:snapshotId} pairs over every valid branch. Used to abort pack GC when + * the snapshot set changes between the two used-pack collections. + */ + protected List<String> snapshotTopology() throws IOException { + List<String> topology = new ArrayList<>(); + for (String branch : validBranches()) { + for (Snapshot snapshot : safelyGetAllSnapshots(branch)) { + topology.add(branch + ":" + snapshot.id()); + } + } + Collections.sort(topology); + return topology; + } + + /** + * Collects used pack identities from every valid branch. Subclasses may override to + * parallelize. + */ + protected Set<String> collectUsedPacks() throws IOException { + Set<String> used = new HashSet<>(); + for (String branch : validBranches()) { + for (Snapshot snapshot : safelyGetAllSnapshots(branch)) { Review Comment: [P2] Deduplicate manifests and sidecars within each mark pass Every retained snapshot/tag is processed independently here, and `emitUsedPacks` immediately rereads all referenced manifests and `.blobref` files. Successive snapshots commonly share immutable base manifests and data files, so the same remote objects are read repeatedly before the final `Set` deduplicates only their pack identities; the cleaner then repeats the entire mark pass a second time. On a large table this can amplify remote reads by the retained-root count and make GC time out or trigger storage throttling. Please deduplicate `(branch, manifestName)` and resolved sidecar paths within each pass, while keeping the two passes independent for the concurrency check. ########## paimon-core/src/main/java/org/apache/paimon/operation/ManagedBlobOrphanFilesClean.java: ########## @@ -0,0 +1,266 @@ +/* + * 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.operation; + +import org.apache.paimon.Snapshot; +import org.apache.paimon.blob.ManagedBlobReachabilityCollector; +import org.apache.paimon.blob.ManagedBlobReachabilityCollector.Result; +import org.apache.paimon.blob.ManagedBlobReferenceFile; +import org.apache.paimon.blob.ManagedBlobReferenceFile.Reference; +import org.apache.paimon.fs.Path; +import org.apache.paimon.io.DataFilePathFactory; +import org.apache.paimon.manifest.FileKind; +import org.apache.paimon.manifest.ManifestEntry; +import org.apache.paimon.manifest.ManifestFile; +import org.apache.paimon.manifest.ManifestFileMeta; +import org.apache.paimon.manifest.ManifestList; +import org.apache.paimon.table.FileStoreTable; +import org.apache.paimon.utils.DataFilePathFactories; + +import java.io.IOException; +import java.net.URI; +import java.util.ArrayList; +import java.util.Collections; +import java.util.HashSet; +import java.util.List; +import java.util.Set; +import java.util.function.Consumer; + +/** + * Cleans unreferenced primary-key {@code .managed.blob} packs. + * + * <p>Unlike {@link OrphanFilesClean}, this cleaner only lists and deletes managed BLOB packs. Pack + * reachability is collected from live {@link FileKind#ADD} data-file {@code .blobref} sidecars. + * Missing manifest lists or unreadable sidecars on a still-existing data file abort pack deletion + * for the rest of the run. + * + * <p>Used packs are collected twice. If the snapshot topology or the used-pack set changes between + * those collections, this run deletes nothing. That shrinks the race with compaction reuse; it is + * not a commit lease. + */ +public abstract class ManagedBlobOrphanFilesClean extends OrphanFilesClean { + + /** + * Marker emitted into the used-pack set when a {@code .blobref} sidecar or a required manifest + * cannot be trusted. Callers must skip deleting every {@code .managed.blob} pack. + */ + public static final String SKIP_MANAGED_BLOB_GC = "__paimon_skip_managed_blob_gc__"; + + public ManagedBlobOrphanFilesClean(FileStoreTable table, long olderThanMillis, boolean dryRun) { + super(table, olderThanMillis, dryRun); + } + + /** + * Join key for a managed pack. Identity is {@code storageRootId + relativePath}, reconstructed + * by {@link Reference#toPath()}. The FileIO scheme is omitted so a wrapper such as test {@code + * traceable:} still matches listed {@code file:} paths at the same location. + */ + public static String packIdentity(Path packPath) { + URI uri = packPath.toUri(); + String authority = uri.getAuthority(); Review Comment: [P1] Normalize qualified and unqualified pack paths identically This drops the URI scheme but keeps the authority. With the documented `hdfs:///warehouse` form, the descriptor writer records an authority-null path because `AbstractBlobElementWriter#setFile` receives the unqualified writer `Path`, while Hadoop `listStatus` returns a qualified path such as `hdfs://nn:8020/warehouse/...`. The resulting reference key (`/warehouse/.../x.managed.blob`) never matches the candidate key (`nn:8020/warehouse/.../x.managed.blob`), so a live pack older than `older_than` can be classified as orphan and deleted. Please canonicalize both sides with the same filesystem identity (for example, path-only while managed BLOB external paths remain unsupported, or qualify both through the same `FileIO`) and add a MiniDFS regression test for an authority-less HDFS warehouse. ########## paimon-core/src/main/java/org/apache/paimon/operation/LocalManagedBlobOrphanFilesClean.java: ########## @@ -0,0 +1,239 @@ +/* + * 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.operation; + +import org.apache.paimon.CoreOptions; +import org.apache.paimon.catalog.Catalog; +import org.apache.paimon.catalog.Identifier; +import org.apache.paimon.fs.Path; +import org.apache.paimon.table.FileStoreTable; +import org.apache.paimon.table.Table; +import org.apache.paimon.utils.Pair; + +import javax.annotation.Nullable; + +import java.io.IOException; +import java.util.ArrayList; +import java.util.Collections; +import java.util.HashMap; +import java.util.Iterator; +import java.util.List; +import java.util.Map; +import java.util.Set; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.Future; +import java.util.concurrent.ThreadPoolExecutor; +import java.util.concurrent.atomic.AtomicLong; +import java.util.function.Function; +import java.util.stream.Collectors; + +import static org.apache.paimon.utils.FileStorePathFactory.BUCKET_PATH_PREFIX; +import static org.apache.paimon.utils.Preconditions.checkArgument; +import static org.apache.paimon.utils.ThreadPoolUtils.createCachedThreadPool; +import static org.apache.paimon.utils.ThreadPoolUtils.randomlyExecuteSequentialReturn; +import static org.apache.paimon.utils.ThreadPoolUtils.randomlyOnlyExecute; + +/** Local {@link ManagedBlobOrphanFilesClean}. */ +public class LocalManagedBlobOrphanFilesClean extends ManagedBlobOrphanFilesClean { + + private final ThreadPoolExecutor executor; + private final List<Path> deleteFiles = new ArrayList<>(); + private final AtomicLong deletedFilesLenInBytes = new AtomicLong(0); + + public LocalManagedBlobOrphanFilesClean( + FileStoreTable table, long olderThanMillis, boolean dryRun) { + super(table, olderThanMillis, dryRun); + this.executor = + createCachedThreadPool( + table.coreOptions().fileOperationThreadNum(), + "MANAGED_BLOB_ORPHAN_FILES_CLEAN"); + } + + public CleanOrphanFilesResult clean() throws IOException { + Map<String, Pair<Path, Long>> candidates = getCandidatePacks(); + if (candidates.isEmpty()) { + return new CleanOrphanFilesResult(0, 0, deleteFiles); + } + + List<String> topologyBefore = snapshotTopology(); + Set<String> usedPacks = collectUsedPacks(); + betweenUsedCollections(); + Set<String> usedPacks2 = collectUsedPacks(); + if (shouldAbortPackGc(topologyBefore, usedPacks, usedPacks2)) { + return new CleanOrphanFilesResult(0, 0, deleteFiles); + } + + candidates.entrySet().stream() + .filter(e -> !usedPacks2.contains(e.getKey())) + .map(Map.Entry::getValue) + .forEach( + info -> { + if (cleanManagedBlobFile(info.getLeft())) { + deletedFilesLenInBytes.addAndGet(info.getRight()); + deleteFiles.add(info.getLeft()); + } + }); + + if (!dryRun) { + cleanEmptyDataDirectory(deleteFiles); + } + return new CleanOrphanFilesResult( + deleteFiles.size(), deletedFilesLenInBytes.get(), deleteFiles); + } + + @Override + protected Set<String> collectUsedPacks() { + return validBranches().stream() + .flatMap(branch -> getUsedPacks(branch).stream()) + .collect(Collectors.toSet()); + } + + private Set<String> getUsedPacks(String branch) { + Set<String> used = ConcurrentHashMap.newKeySet(); + try { + randomlyOnlyExecute( + executor, + snapshot -> { + try { + emitUsedPacks(branch, snapshot, used::add); + } catch (IOException e) { + throw new RuntimeException(e); + } + }, + safelyGetAllSnapshots(branch)); + } catch (IOException e) { + throw new RuntimeException(e); + } + return used; + } + + private Map<String, Pair<Path, Long>> getCandidatePacks() { + List<Path> fileDirs = listPaimonFileDirs(); + Iterator<Pair<Path, Long>> packs = + randomlyExecuteSequentialReturn(executor, packLister(), fileDirs); + Map<String, Pair<Path, Long>> result = new HashMap<>(); + while (packs.hasNext()) { + Pair<Path, Long> fileInfo = packs.next(); + result.put(packIdentity(fileInfo.getLeft()), fileInfo); + } + return result; + } + + private Function<Path, List<Pair<Path, Long>>> packLister() { + return path -> + tryBestListingDirs(path).stream() + .filter(status -> !status.isDir()) + .filter(this::oldEnough) + .filter(status -> isManagedBlobPackName(status.getPath().getName())) + .map(status -> Pair.of(status.getPath(), status.getLen())) + .collect(Collectors.toList()); + } + + private void cleanEmptyDataDirectory(List<Path> deleted) { + if (deleted.isEmpty()) { + return; + } + Set<Path> bucketDirs = + deleted.stream() + .map(Path::getParent) + .filter(path -> path.toString().contains(BUCKET_PATH_PREFIX)) + .collect(Collectors.toSet()); + randomlyOnlyExecute(executor, this::tryDeleteEmptyDirectory, bucketDirs); + Set<Path> partitionDirs = + bucketDirs.stream().map(Path::getParent).collect(Collectors.toSet()); + tryCleanDataDirectory(partitionDirs, partitionKeysNum); + } + + public static List<LocalManagedBlobOrphanFilesClean> createCleans( + Catalog catalog, + String databaseName, + @Nullable String tableName, + long olderThanMillis, + @Nullable Integer parallelism, + boolean dryRun) + throws Catalog.DatabaseNotExistException, Catalog.TableNotExistException { + List<String> tableNames = Collections.singletonList(tableName); + if (tableName == null || "*".equals(tableName)) { + tableNames = catalog.listTables(databaseName); + } + + Map<String, String> dynamicOptions = + parallelism == null + ? Collections.emptyMap() + : new HashMap<String, String>() { + { + put( + CoreOptions.FILE_OPERATION_THREAD_NUM.key(), + parallelism.toString()); + } + }; + + List<LocalManagedBlobOrphanFilesClean> cleans = new ArrayList<>(tableNames.size()); + for (String t : tableNames) { + Identifier identifier = new Identifier(databaseName, t); + Table table = catalog.getTable(identifier).copy(dynamicOptions); + checkArgument( + table instanceof FileStoreTable, + "Only FileStoreTable supports remove-orphan-blobs action. The table type is '%s'.", + table.getClass().getName()); + cleans.add( + new LocalManagedBlobOrphanFilesClean( + (FileStoreTable) table, olderThanMillis, dryRun)); + } + return cleans; + } + + public static CleanOrphanFilesResult executeDatabase( + Catalog catalog, + String databaseName, + @Nullable String tableName, + long olderThanMillis, + @Nullable Integer parallelism, + boolean dryRun) + throws Catalog.DatabaseNotExistException, Catalog.TableNotExistException { + List<LocalManagedBlobOrphanFilesClean> tableCleans = + createCleans( + catalog, databaseName, tableName, olderThanMillis, parallelism, dryRun); + ExecutorService executorService = Review Comment: [P2] Shut down the database executor on every exit `shutdownNow()` is reached only after every future completes successfully. If any table task throws (for example, because a manifest/object-store read fails), the `ExecutionException`/`InterruptedException` path exits without cancelling the other tasks or closing this fixed thread pool. These workers are non-daemon core threads with no timeout, so the caller can leak threads or fail to terminate, and already-submitted cleanup tasks may continue deleting after the procedure reports failure. Please cancel unfinished futures and close the executor in a `finally` block, preserving the interrupt status. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
