diff --git a/docs/docs/primary-key-table/blob-storage.md b/docs/docs/primary-key-table/blob-storage.md index 44ccd638d3a9..9dc04ba3b05e 100644 --- a/docs/docs/primary-key-table/blob-storage.md +++ b/docs/docs/primary-key-table/blob-storage.md @@ -183,9 +183,8 @@ participates in aggregation or retraction, even when its sequence value is older the field for both newer and older retract records. Managed BLOB partial updates externalize each non-null scalar BLOB, array element, or map value into a -`.managed.blob` pack. Empty collections and collections containing only null values write no payload. BLOB garbage -collection for orphaned packs is not implemented yet; repeated updates can leave unreachable storage until a future -collector is available. +`.managed.blob` pack. Empty collections and collections containing only null values write no payload. Unreachable packs +from repeated updates are reclaimed by `remove_orphan_blobs` after they are older than `older_than`. `blob-view-field` columns store serialized view structs inline. Reads resolve upstream blob bytes through the catalog when `blob-view.resolve.enabled` is true (default). Append upstream tables used by `sys.blob_view(...)` must enable @@ -240,16 +239,33 @@ extra files because more than one retained data file can reference the same pack ## Garbage Collection -Garbage collection of unreferenced `.managed.blob` packs is not implemented yet. Updates, deletes, compaction, or an -ambiguous writer failure can therefore leave payload packs that are no longer reachable from current rows. - -The ordinary orphan-file cleaner intentionally preserves all `.managed.blob` files. This fail-safe behavior prevents it -from deleting a payload that is still reachable from a snapshot, tag, branch, or another retained root, but it also -means unused BLOB storage can grow until a root-aware BLOB garbage collector is available. - -A future collector must compute reachability across all retained roots and treat a missing, corrupt, or unsupported -`.blobref` sidecar as unsafe to delete. An empty, valid sidecar is different from a missing sidecar: it explicitly states -that the data file references no managed payload pack. +Unreferenced `.managed.blob` packs are reclaimed by `LocalManagedBlobOrphanFilesClean`. +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). +`remove_orphan_files` never deletes `.managed.blob` packs. + +This cleanup is best-effort. It lists snapshots, collects used packs twice, and aborts the run (deletes +nothing) if the snapshot topology or used-pack set changed between those collections. That shrinks the +window in which a concurrent commit can change reachability. Standard Paimon compaction does not make a pack +that was unreachable at the final collection reachable afterward: it only reuses packs referenced by its +input data files, and deletion-conflict detection rejects a stale compact whose inputs have already been +removed. Under these standard compaction invariants, no separate commit lease is required for that +compaction path. + +`older_than` provides a grace period for packs created by a writer but not yet referenced by a committed +snapshot. Standard writers create new UUID-named packs; choose a cutoff far enough behind the current time +for writes, commits, and retries to finish. The one-day default assumes those operations complete within one +day. Writers that publish references to pre-existing old packs, or commit implementations that bypass normal +deletion-conflict detection, are outside this safety model. + +A missing, corrupt, or unsupported `.blobref` sidecar on a data file that still exists is unsafe: that run skips +deleting every `.managed.blob` file. ADD entries left in unmerged manifests after snapshot expire, whose data files +are already gone, are ignored. An empty, valid sidecar is different from a missing sidecar: it explicitly states that +the data file references no managed payload pack. + +Snapshot expiration still deletes only the data file and its `.blobref` extra file. Pack bytes are reclaimed on the +next managed blob orphan cleanup run after they become unreachable. ## Reference Metadata diff --git a/paimon-core/src/main/java/org/apache/paimon/blob/ManagedBlobReachabilityCollector.java b/paimon-core/src/main/java/org/apache/paimon/blob/ManagedBlobReachabilityCollector.java new file mode 100644 index 000000000000..d639c19c71b1 --- /dev/null +++ b/paimon-core/src/main/java/org/apache/paimon/blob/ManagedBlobReachabilityCollector.java @@ -0,0 +1,273 @@ +/* + * 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.blob; + +import org.apache.paimon.blob.ManagedBlobReferenceFile.Reference; +import org.apache.paimon.fs.FileIO; +import org.apache.paimon.fs.Path; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.io.FileNotFoundException; +import java.io.IOException; +import java.util.Collections; +import java.util.HashSet; +import java.util.List; +import java.util.Objects; +import java.util.Set; +import java.util.concurrent.TimeUnit; + +/** + * Collects managed BLOB pack reachability from data-file {@code .blobref} sidecars. + * + *

This collector does not scan snapshots or delete files. Callers such as orphan-file cleanup + * (and later snapshot expiration) supply data files and decide what to delete from {@link Result}. + */ +public class ManagedBlobReachabilityCollector { + + private static final Logger LOG = + LoggerFactory.getLogger(ManagedBlobReachabilityCollector.class); + + private static final int READ_RETRY_NUM = 3; + private static final int READ_RETRY_INTERVAL_MS = 5; + + private final FileIO fileIO; + + public ManagedBlobReachabilityCollector(FileIO fileIO) { + this.fileIO = fileIO; + } + + /** + * Reads blobref extras of one data file. Extra files without a {@code .blobref} suffix are + * ignored. A listed sidecar that cannot be trusted marks the result unsafe, unless the data + * file itself is already gone: unmerged snapshot manifests can still contain {@code ADD} + * entries that snapshot expire has deleted, and those must not abort pack GC. + * + *

Orphan cleanup resolves sidecars itself and calls {@link #fromSidecar(Path, Path)}, so + * this whole-entry variant currently has no production caller. It is retained as the + * entry-level reachability oracle for tests and for snapshot expiration, which needs to walk + * {@code extraFiles} rather than pre-resolved sidecar paths. + */ + public Result fromDataFile(Path dataFile, List extraFiles) { + Result result = Result.empty(); + if (extraFiles == null || extraFiles.isEmpty()) { + return result; + } + Path parent = dataFile.getParent(); + Boolean dataFileExists = null; + for (String extra : extraFiles) { + if (extra == null || !extra.endsWith(ManagedBlobReferenceFile.REFERENCE_FILE_SUFFIX)) { + continue; + } + Path sidecar = new Path(parent, extra); + try { + result = result.merge(Result.of(readWithRetry(sidecar))); + } catch (IOException e) { + if (dataFileExists == null) { + dataFileExists = checkDataFileExists(dataFile); + } + if (!dataFileExists) { + LOG.debug( + "Ignore unreadable blobref {} because data file {} is already gone.", + sidecar, + dataFile); + continue; + } + LOG.warn( + "Failed to read managed BLOB reference file {}. Skip managed blob GC this run.", + sidecar, + e); + return Result.unsafe(); + } + } + return result; + } + + private boolean checkDataFileExists(Path dataFile) { + try { + return fileIO.exists(dataFile); + } catch (IOException e) { + LOG.warn( + "Failed to check existence of {}, treat as present for managed blob GC.", + dataFile, + e); + return true; + } + } + + /** + * Reads one sidecar. Missing, corrupt, or unsupported files are unsafe rather than thrown to + * the caller. + */ + public Result fromSidecar(Path sidecar) { + try { + return Result.of(readWithRetry(sidecar)); + } catch (IOException e) { + LOG.warn( + "Failed to read managed BLOB reference file {}. Skip managed blob GC this run.", + sidecar, + e); + return Result.unsafe(); + } + } + + /** + * Reads one resolved sidecar while preserving data-file-aware orphan cleanup semantics. An + * unreadable sidecar is ignored only when its data file is already gone. + */ + public Result fromSidecar(Path dataFile, Path sidecar) { + try { + return Result.of(readWithRetry(sidecar)); + } catch (IOException e) { + if (!checkDataFileExists(dataFile)) { + LOG.debug( + "Ignore unreadable blobref {} because data file {} is already gone.", + sidecar, + dataFile); + return Result.empty(); + } + LOG.warn( + "Failed to read managed BLOB reference file {}. Skip managed blob GC this run.", + sidecar, + e); + return Result.unsafe(); + } + } + + private List readWithRetry(Path sidecar) throws IOException { + IOException caught = null; + for (int retry = 0; retry < READ_RETRY_NUM; retry++) { + try { + return ManagedBlobReferenceFile.read(fileIO, sidecar); + } catch (FileNotFoundException e) { + throw e; + } catch (IOException e) { + caught = e; + } + try { + TimeUnit.MILLISECONDS.sleep(READ_RETRY_INTERVAL_MS); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new IOException("Interrupted while reading " + sidecar, e); + } + } + throw caught; + } + + /** Reachability of managed BLOB packs from one or more data files. */ + public static final class Result { + + private static final Result EMPTY = new Result(Collections.emptySet(), false); + private static final Result UNSAFE = new Result(Collections.emptySet(), true); + + private final Set referenced; + private final boolean unsafe; + + private Result(Set referenced, boolean unsafe) { + this.referenced = referenced; + this.unsafe = unsafe; + } + + public static Result empty() { + return EMPTY; + } + + public static Result unsafe() { + return UNSAFE; + } + + public static Result of(List refs) { + if (refs == null || refs.isEmpty()) { + return empty(); + } + return new Result(Collections.unmodifiableSet(new HashSet<>(refs)), false); + } + + public Set referenced() { + return referenced; + } + + public boolean isUnsafe() { + return unsafe; + } + + public boolean contains(Reference ref) { + return referenced.contains(ref); + } + + public boolean containsPackName(String fileName) { + for (Reference reference : referenced) { + if (reference.relativePath().equals(fileName)) { + return true; + } + } + return false; + } + + public Result merge(Result other) { + if (other == null) { + return this; + } + boolean mergedUnsafe = unsafe || other.unsafe; + if (referenced.isEmpty() && other.referenced.isEmpty()) { + return mergedUnsafe ? unsafe() : empty(); + } + Set refs; + if (referenced.isEmpty()) { + refs = other.referenced; + } else if (other.referenced.isEmpty()) { + refs = referenced; + } else { + refs = new HashSet<>(referenced); + refs.addAll(other.referenced); + refs = Collections.unmodifiableSet(refs); + } + if (mergedUnsafe == unsafe && refs == referenced) { + return this; + } + if (mergedUnsafe == other.unsafe && refs == other.referenced) { + return other; + } + return new Result(refs, mergedUnsafe); + } + + @Override + public boolean equals(Object o) { + if (this == o) { + return true; + } + if (o == null || getClass() != o.getClass()) { + return false; + } + Result result = (Result) o; + return unsafe == result.unsafe && Objects.equals(referenced, result.referenced); + } + + @Override + public int hashCode() { + return Objects.hash(referenced, unsafe); + } + + @Override + public String toString() { + return "Result{unsafe=" + unsafe + ", referenced=" + referenced + '}'; + } + } +} diff --git a/paimon-core/src/main/java/org/apache/paimon/blob/ManagedBlobReferenceFile.java b/paimon-core/src/main/java/org/apache/paimon/blob/ManagedBlobReferenceFile.java index f7b10512c65b..e540fb05cd94 100644 --- a/paimon-core/src/main/java/org/apache/paimon/blob/ManagedBlobReferenceFile.java +++ b/paimon-core/src/main/java/org/apache/paimon/blob/ManagedBlobReferenceFile.java @@ -170,6 +170,10 @@ public String relativePath() { return relativePath; } + public Path toPath() { + return new Path(storageRootId, relativePath); + } + @Override public boolean equals(Object o) { if (this == o) { diff --git a/paimon-core/src/main/java/org/apache/paimon/operation/LocalManagedBlobOrphanFilesClean.java b/paimon-core/src/main/java/org/apache/paimon/operation/LocalManagedBlobOrphanFilesClean.java new file mode 100644 index 000000000000..712bf9164e4d --- /dev/null +++ b/paimon-core/src/main/java/org/apache/paimon/operation/LocalManagedBlobOrphanFilesClean.java @@ -0,0 +1,415 @@ +/* + * 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.Collection; +import java.util.Collections; +import java.util.HashMap; +import java.util.Iterator; +import java.util.List; +import java.util.Map; +import java.util.Optional; +import java.util.Set; +import java.util.concurrent.CompletionService; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.ExecutorCompletionService; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.Future; +import java.util.concurrent.ThreadPoolExecutor; +import java.util.concurrent.TimeUnit; +import java.util.function.Consumer; +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; +import static org.apache.paimon.utils.ThreadUtils.newDaemonThreadFactory; + +/** Local {@link ManagedBlobOrphanFilesClean}. */ +public class LocalManagedBlobOrphanFilesClean extends ManagedBlobOrphanFilesClean + implements AutoCloseable { + + /** + * Upper bound for waiting cancelled table cleanups after a database-wide failure. {@code + * shutdownNow()} only requests interruption; a FileIO call may ignore it until a socket + * timeout. Waiting forever would hide the original failure. + */ + private static final long TERMINATION_TIMEOUT_MS = TimeUnit.SECONDS.toMillis(120); + + private final ThreadPoolExecutor executor; + + public LocalManagedBlobOrphanFilesClean( + FileStoreTable table, long olderThanMillis, boolean dryRun) { + super(table, olderThanMillis, dryRun); + this.executor = + createCachedThreadPool( + table.coreOptions().fileOperationThreadNum(), + "MANAGED_BLOB_ORPHAN_FILES_CLEAN"); + } + + /** + * Cleans unreferenced managed BLOB packs for this table. + * + *

Used-pack collection waits without a timeout. A FileIO call that ignores interruption can + * block this method indefinitely. {@link #executeDatabase} applies a bounded termination wait + * only after a sibling table has already failed and remaining tasks are cancelled. + */ + public CleanOrphanFilesResult clean() throws IOException { + List deleteFiles = new ArrayList<>(); + long deletedFilesLenInBytes = 0; + Map> candidates = getCandidatePacks(); + if (candidates.isEmpty()) { + return new CleanOrphanFilesResult(0, 0, deleteFiles); + } + if (candidates.containsKey(SKIP_MANAGED_BLOB_GC)) { + LOG.warn( + "Skip managed blob pack GC for table {} because a listed pack path cannot be resolved safely.", + table.fullName()); + return new CleanOrphanFilesResult(0, 0, deleteFiles); + } + + List topologyBefore = snapshotTopology(); + Set usedPacks = collectUsedPacks(); + betweenUsedCollections(); + Set usedPacks2 = collectUsedPacks(); + if (shouldAbortPackGc(topologyBefore, usedPacks, usedPacks2)) { + return new CleanOrphanFilesResult(0, 0, deleteFiles); + } + + for (Map.Entry> candidate : candidates.entrySet()) { + throwIfInterrupted(); + if (usedPacks2.contains(candidate.getKey())) { + continue; + } + Pair info = candidate.getValue(); + if (cleanManagedBlobFile(info.getLeft())) { + deletedFilesLenInBytes += info.getRight(); + deleteFiles.add(info.getLeft()); + } + } + + throwIfInterrupted(); + if (!dryRun) { + cleanEmptyDataDirectory(deleteFiles); + } + return new CleanOrphanFilesResult(deleteFiles.size(), deletedFilesLenInBytes, deleteFiles); + } + + private static void throwIfInterrupted() throws IOException { + if (Thread.currentThread().isInterrupted()) { + throw new IOException("Interrupted while cleaning managed blob orphan files."); + } + } + + @Override + protected Set collectUsedPacks() { + ReachabilityScan scan = newReachabilityScan(); + return validBranches().stream() + .flatMap(branch -> getUsedPacks(branch, scan).stream()) + .collect(Collectors.toSet()); + } + + private Set getUsedPacks(String branch, ReachabilityScan scan) { + Set used = ConcurrentHashMap.newKeySet(); + try { + executeSnapshotsInCompletionOrder( + executor, + snapshot -> { + try { + emitUsedPacks(branch, snapshot, scan, used::add); + } catch (IOException e) { + LOG.warn( + "Failed to collect used managed blob packs for table {} branch {} snapshot {}.", + table.fullName(), + branch, + snapshot.id(), + e); + throw new RuntimeException(e); + } + }, + safelyGetAllSnapshots(branch)); + } catch (IOException e) { + LOG.warn( + "Failed to list snapshots while collecting used managed blob packs for table {} branch {}.", + table.fullName(), + branch, + e); + throw new RuntimeException(e); + } + return used; + } + + /** + * Waits in completion order and cancels remaining snapshot tasks on the first failure so a + * later failed snapshot is not hidden by an earlier one stuck in uninterruptible I/O. + * + *

{@link Future#cancel(boolean)} only requests interruption, so a task blocked in a FileIO + * call can keep running after this method throws. {@code take()} itself has no timeout, so a + * hung FileIO call also blocks {@link #clean()} until that call returns. This does not leak: + * every caller reaches {@link #close()} (directly, or via {@link #executeDatabase}, which + * additionally awaits termination up to {@link #TERMINATION_TIMEOUT_MS}), and that shuts the + * pool down. Any new caller must preserve that guarantee. + */ + static void executeSnapshotsInCompletionOrder( + ExecutorService executor, Consumer processor, Collection input) { + if (input.isEmpty()) { + return; + } + CompletionService completionService = new ExecutorCompletionService<>(executor); + ClassLoader cl = Thread.currentThread().getContextClassLoader(); + List> futures = new ArrayList<>(input.size()); + for (U u : input) { + futures.add( + completionService.submit( + () -> { + Thread.currentThread().setContextClassLoader(cl); + processor.accept(u); + return null; + })); + } + try { + for (int i = 0; i < futures.size(); i++) { + completionService.take().get(); + } + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + cancelAll(futures); + throw new RuntimeException(e); + } catch (ExecutionException e) { + cancelAll(futures); + throw new RuntimeException(e); + } + } + + private static void cancelAll(List> futures) { + for (Future future : futures) { + future.cancel(true); + } + } + + private Map> getCandidatePacks() { + List fileDirs = listPaimonFileDirs(); + Iterator> packs = + randomlyExecuteSequentialReturn(executor, packLister(), fileDirs); + Map> result = new HashMap<>(); + while (packs.hasNext()) { + Pair fileInfo = packs.next(); + Optional identity = packIdentityForCleanup(fileInfo.getLeft()); + if (!identity.isPresent()) { + result.clear(); + result.put(SKIP_MANAGED_BLOB_GC, fileInfo); + return result; + } + result.put(identity.get(), fileInfo); + } + return result; + } + + private Function>> 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 deleted) { + if (deleted.isEmpty()) { + return; + } + Set bucketDirs = + deleted.stream() + .map(Path::getParent) + .filter(path -> path.toString().contains(BUCKET_PATH_PREFIX)) + .collect(Collectors.toSet()); + randomlyOnlyExecute(executor, this::tryDeleteEmptyDirectory, bucketDirs); + Set partitionDirs = + bucketDirs.stream().map(Path::getParent).collect(Collectors.toSet()); + tryCleanDataDirectory(partitionDirs, partitionKeysNum); + } + + public static List createCleans( + Catalog catalog, + String databaseName, + @Nullable String tableName, + long olderThanMillis, + @Nullable Integer parallelism, + boolean dryRun) + throws Catalog.DatabaseNotExistException, Catalog.TableNotExistException { + List tableNames = Collections.singletonList(tableName); + if (tableName == null || "*".equals(tableName)) { + tableNames = catalog.listTables(databaseName); + } + + Map dynamicOptions = + parallelism == null + ? Collections.emptyMap() + : new HashMap() { + { + put( + CoreOptions.FILE_OPERATION_THREAD_NUM.key(), + parallelism.toString()); + } + }; + + List 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 tableCleans = + createCleans( + catalog, databaseName, tableName, olderThanMillis, parallelism, dryRun); + ExecutorService executorService = + Executors.newFixedThreadPool( + Runtime.getRuntime().availableProcessors(), + newDaemonThreadFactory("MANAGED-BLOB-ORPHAN-DB-CLEAN")); + return executeDatabase(tableCleans, executorService, TERMINATION_TIMEOUT_MS); + } + + static CleanOrphanFilesResult executeDatabase( + List tableCleans, ExecutorService executorService) { + return executeDatabase(tableCleans, executorService, TERMINATION_TIMEOUT_MS); + } + + static CleanOrphanFilesResult executeDatabase( + List tableCleans, + ExecutorService executorService, + long terminationTimeoutMs) { + List> tasks = new ArrayList<>(tableCleans.size()); + CompletionService completionService = + new ExecutorCompletionService<>(executorService); + try { + for (LocalManagedBlobOrphanFilesClean clean : tableCleans) { + tasks.add(completionService.submit(clean::clean)); + } + + long deletedFileCount = 0; + long deletedFileTotalLenInBytes = 0; + for (int i = 0; i < tasks.size(); i++) { + try { + Future task = completionService.take(); + CleanOrphanFilesResult result = task.get(); + deletedFileCount += result.getDeletedFileCount(); + deletedFileTotalLenInBytes += result.getDeletedFileTotalLenInBytes(); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new RuntimeException(e); + } catch (ExecutionException e) { + throw new RuntimeException(e); + } + } + return new CleanOrphanFilesResult(deletedFileCount, deletedFileTotalLenInBytes); + } finally { + for (Future task : tasks) { + task.cancel(true); + } + for (LocalManagedBlobOrphanFilesClean clean : tableCleans) { + clean.close(); + } + executorService.shutdownNow(); + + List toAwait = new ArrayList<>(tableCleans.size() + 1); + toAwait.add(executorService); + for (LocalManagedBlobOrphanFilesClean clean : tableCleans) { + toAwait.add(clean.executor); + } + + boolean restoreInterrupt = Thread.interrupted(); + restoreInterrupt |= awaitTermination(terminationTimeoutMs, toAwait); + if (restoreInterrupt) { + Thread.currentThread().interrupt(); + } + } + } + + /** + * Waits until every executor terminates or {@code timeoutMs} elapses, whichever is first. Does + * not throw, so a database-wide failure still surfaces after a stuck FileIO call. + */ + private static boolean awaitTermination( + long timeoutMs, List executorServices) { + boolean interrupted = false; + long timeoutNanos = TimeUnit.MILLISECONDS.toNanos(Math.max(0L, timeoutMs)); + long startNanos = System.nanoTime(); + for (ExecutorService executorService : executorServices) { + long remainingNanos = timeoutNanos - (System.nanoTime() - startNanos); + if (remainingNanos <= 0) { + LOG.warn( + "Timed out waiting for managed blob orphan cleanup executors to terminate. " + + "A leftover FileIO call may still be running."); + break; + } + try { + if (!executorService.awaitTermination(remainingNanos, TimeUnit.NANOSECONDS)) { + LOG.warn( + "Timed out waiting for managed blob orphan cleanup executors to terminate. " + + "A leftover FileIO call may still be running."); + break; + } + } catch (InterruptedException e) { + interrupted = true; + break; + } + } + return interrupted; + } + + @Override + public void close() { + executor.shutdownNow(); + } +} diff --git a/paimon-core/src/main/java/org/apache/paimon/operation/ManagedBlobOrphanFilesClean.java b/paimon-core/src/main/java/org/apache/paimon/operation/ManagedBlobOrphanFilesClean.java new file mode 100644 index 000000000000..d362260dcb0d --- /dev/null +++ b/paimon-core/src/main/java/org/apache/paimon/operation/ManagedBlobOrphanFilesClean.java @@ -0,0 +1,508 @@ +/* + * 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.FileIO; +import org.apache.paimon.fs.Path; +import org.apache.paimon.fs.local.LocalFileIO; +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.File; +import java.io.FileNotFoundException; +import java.io.IOException; +import java.io.Serializable; +import java.util.ArrayList; +import java.util.Collections; +import java.util.HashSet; +import java.util.List; +import java.util.Objects; +import java.util.Optional; +import java.util.Set; +import java.util.concurrent.ConcurrentHashMap; +import java.util.function.Consumer; + +/** + * Cleans unreferenced primary-key {@code .managed.blob} packs. + * + *

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. + * + *

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. File-system qualification is omitted so a reference written as + * {@code hdfs:///warehouse/...} still matches a listing returned as {@code + * hdfs://namenode:8020/warehouse/...}. A collision between distinct storage authorities can + * only retain an orphan: a candidate sharing its URI path with any live pack is conservatively + * treated as used. Relative paths are intentionally left unchanged because resolving them + * requires the semantics of the table's {@link FileIO}. + */ + public static String packIdentity(Path packPath) { + return packPath.toUri().getPath(); + } + + 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 snapshotTopology() throws IOException { + List 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 collectUsedPacks() throws IOException { + Set used = new HashSet<>(); + ReachabilityScan scan = newReachabilityScan(); + for (String branch : validBranches()) { + for (Snapshot snapshot : safelyGetAllSnapshots(branch)) { + emitUsedPacks(branch, snapshot, scan, used::add); + } + } + return used; + } + + /** Creates independent deduplication state for one complete reachability scan. */ + protected final ReachabilityScan newReachabilityScan() { + return new ReachabilityScan(); + } + + /** Test hook between the two used-pack collections. Production cleaners leave this empty. */ + protected void betweenUsedCollections() {} + + /** + * Aborts this run when sidecars are untrusted, the snapshot topology changed, or the two + * used-pack collections disagree. Callers must not delete any pack when this returns true. + */ + protected boolean shouldAbortPackGc( + List topologyBefore, Set used, Set used2) throws IOException { + if (used.contains(SKIP_MANAGED_BLOB_GC) || used2.contains(SKIP_MANAGED_BLOB_GC)) { + LOG.warn( + "Skip managed blob pack GC for table {} because some sidecars or manifests cannot be trusted.", + table.fullName()); + return true; + } + List topologyAfter = snapshotTopology(); + if (!topologyBefore.equals(topologyAfter)) { + LOG.warn( + "Skip managed blob pack GC for table {} because snapshot topology changed during used-pack collection.", + table.fullName()); + return true; + } + if (!used.equals(used2)) { + LOG.warn( + "Skip managed blob pack GC for table {} because the used pack set changed during used-pack collection.", + table.fullName()); + return true; + } + return false; + } + + /** + * Emits referenced pack identities from {@code entry}. Pack reachability is collected only from + * {@link FileKind#ADD} files: {@link FileKind#DELETE} entries remain in delta manifests after + * compaction, while snapshot expire may already have removed their {@code .blobref} sidecars. + */ + protected void emitUsedPacks( + ManifestEntry entry, DataFilePathFactory pathFactory, Consumer used) { + emitUsedPacks(entry, pathFactory, newReachabilityScan(), used); + } + + /** Emits referenced packs while reading each sidecar at most once in this scan. */ + protected void emitUsedPacks( + ManifestEntry entry, + DataFilePathFactory pathFactory, + ReachabilityScan scan, + Consumer used) { + for (SidecarWorkItem workItem : createSidecarWorkItems(entry, pathFactory)) { + if (scan.markSidecar(workItem.dedupIdentity())) { + emitUsedPacks(workItem, scan, used); + } + } + } + + /** + * Creates one serializable work item per managed BLOB reference sidecar of an {@link + * FileKind#ADD} entry. No sidecar is read by this method. + */ + protected final List createSidecarWorkItems( + ManifestEntry entry, DataFilePathFactory pathFactory) { + if (entry.kind() != FileKind.ADD) { + return Collections.emptyList(); + } + List extraFiles = entry.file().extraFiles(); + if (extraFiles == null || extraFiles.isEmpty()) { + return Collections.emptyList(); + } + Path dataFile = pathFactory.toPath(entry); + List workItems = new ArrayList<>(); + for (String extraFile : extraFiles) { + if (extraFile != null + && extraFile.endsWith(ManagedBlobReferenceFile.REFERENCE_FILE_SUFFIX)) { + Path sidecar = new Path(dataFile.getParent(), extraFile); + workItems.add( + new SidecarWorkItem( + dataFile, sidecar, extraFile, sidecarDedupIdentity(sidecar))); + } + } + return workItems; + } + + /** + * Reads one globally deduplicated sidecar work item and emits canonical pack identities. + * Callers are responsible for deduplicating {@link SidecarWorkItem#dedupIdentity()} within one + * reachability pass. + */ + protected final void emitUsedPacks( + SidecarWorkItem workItem, ReachabilityScan scan, Consumer used) { + Result reachability = + new ManagedBlobReachabilityCollector(fileIO) + .fromSidecar(workItem.dataFile(), workItem.sidecar()); + if (reachability.isUnsafe()) { + used.accept(SKIP_MANAGED_BLOB_GC); + return; + } + for (Reference reference : reachability.referenced()) { + Optional identity = packIdentity(reference, scan); + used.accept(identity.orElse(SKIP_MANAGED_BLOB_GC)); + } + } + + private String sidecarDedupIdentity(Path sidecar) { + String path = sidecar.toUri().getPath(); + if (fileIO instanceof LocalFileIO && path != null && !new File(path).isAbsolute()) { + return new File(path).toPath().toAbsolutePath().normalize().toUri().toString(); + } + return sidecar.toUri().normalize().toString(); + } + + /** Serializable unit of work for one managed BLOB reference sidecar. */ + public static final class SidecarWorkItem implements Serializable { + + private static final long serialVersionUID = 1L; + + private final Path dataFile; + private final Path sidecar; + private final String extraFile; + private final String dedupIdentity; + + private SidecarWorkItem( + Path dataFile, Path sidecar, String extraFile, String dedupIdentity) { + this.dataFile = dataFile; + this.sidecar = sidecar; + this.extraFile = extraFile; + this.dedupIdentity = dedupIdentity; + } + + public Path dataFile() { + return dataFile; + } + + public Path sidecar() { + return sidecar; + } + + public String extraFile() { + return extraFile; + } + + public String dedupIdentity() { + return dedupIdentity; + } + + @Override + public boolean equals(Object o) { + if (this == o) { + return true; + } + if (o == null || getClass() != o.getClass()) { + return false; + } + SidecarWorkItem that = (SidecarWorkItem) o; + return Objects.equals(dedupIdentity, that.dedupIdentity); + } + + @Override + public int hashCode() { + return Objects.hash(dedupIdentity); + } + + @Override + public String toString() { + return dedupIdentity; + } + } + + /** Returns a canonical identity or empty when a relative path cannot be resolved safely. */ + protected final Optional packIdentityForCleanup(Path packPath) { + if (isAbsolute(packPath)) { + return Optional.of(packIdentity(packPath)); + } + if (fileIO instanceof LocalFileIO) { + String path = packPath.toUri().getPath(); + return Optional.of( + new File(path).toPath().toAbsolutePath().normalize().toUri().getPath()); + } + try { + Path canonical = fileIO.getFileStatus(packPath).getPath(); + return isAbsolute(canonical) ? Optional.of(packIdentity(canonical)) : Optional.empty(); + } catch (IOException e) { + LOG.warn("Cannot safely resolve relative managed blob path {}.", packPath, e); + return Optional.empty(); + } + } + + private Optional packIdentity(Reference reference, ReachabilityScan scan) { + Path packPath = reference.toPath(); + if (isAbsolute(packPath) || fileIO instanceof LocalFileIO) { + return packIdentityForCleanup(packPath); + } + Optional canonicalRoot = scan.canonicalStorageRoot(reference.storageRootId(), fileIO); + if (!canonicalRoot.isPresent()) { + LOG.warn( + "Cannot safely resolve relative managed blob storage root {}. Skip pack GC this run.", + reference.storageRootId()); + return Optional.empty(); + } + return Optional.of(packIdentity(new Path(canonicalRoot.get(), reference.relativePath()))); + } + + private static boolean isAbsolute(Path path) { + String uriPath = path.toUri().getPath(); + return uriPath != null && new File(uriPath).isAbsolute(); + } + + /** + * Reads {@code manifestName} and emits used packs. A missing manifest is treated as unsafe: + * {@link java.io.FileNotFoundException} would otherwise look like an empty used set. + */ + protected void emitUsedPacks( + String manifestName, + ManifestFile manifestFile, + DataFilePathFactories pathFactories, + Consumer used) + throws IOException { + emitUsedPacks("", manifestName, manifestFile, pathFactories, newReachabilityScan(), used); + } + + private void emitUsedPacks( + String branch, + String manifestName, + ManifestFile manifestFile, + DataFilePathFactories pathFactories, + ReachabilityScan scan, + Consumer used) + throws IOException { + if (!scan.markManifest(branch, manifestName)) { + return; + } + List entries = + retryReadingFiles(() -> manifestFile.readWithIOException(manifestName), null); + if (entries == null) { + LOG.warn( + "Manifest {} is missing while collecting used managed blob packs. Skip pack GC this run.", + manifestName); + used.accept(SKIP_MANAGED_BLOB_GC); + return; + } + for (ManifestEntry entry : entries) { + emitUsedPacks(entry, pathFactories.get(entry.partition(), entry.bucket()), scan, used); + } + } + + /** + * Reads data manifests of {@code snapshot} and emits used packs. A missing manifest list is + * treated as unsafe for the same reason as a missing manifest. + */ + protected void emitUsedPacks(String branch, Snapshot snapshot, Consumer used) + throws IOException { + emitUsedPacks(branch, snapshot, newReachabilityScan(), used); + } + + /** Reads a snapshot while deduplicating manifests and sidecars across the whole scan. */ + protected void emitUsedPacks( + String branch, Snapshot snapshot, ReachabilityScan scan, Consumer used) + throws IOException { + FileStoreTable branchTable = table.switchToBranch(branch); + ManifestList manifestList = branchTable.store().manifestListFactory().create(); + ManifestFile manifestFile = branchTable.store().manifestFileFactory().create(); + DataFilePathFactories pathFactories = + new DataFilePathFactories(branchTable.store().pathFactory()); + List metas = new ArrayList<>(); + if (!addManifestList(manifestList, snapshot.changelogManifestList(), metas, used) + || !addManifestList(manifestList, snapshot.deltaManifestList(), metas, used) + || !addManifestList(manifestList, snapshot.baseManifestList(), metas, used)) { + return; + } + for (ManifestFileMeta meta : metas) { + emitUsedPacks(branch, meta.fileName(), manifestFile, pathFactories, scan, used); + } + } + + /** Thread-safe read-deduplication state scoped to one reachability scan. */ + protected static final class ReachabilityScan { + + private final Set manifests = ConcurrentHashMap.newKeySet(); + private final Set sidecars = ConcurrentHashMap.newKeySet(); + private final ConcurrentHashMap> canonicalStorageRoots = + new ConcurrentHashMap<>(); + + private boolean markManifest(String branch, String manifestName) { + return manifests.add(branch + '\0' + manifestName); + } + + private boolean markSidecar(String sidecarIdentity) { + return sidecars.add(sidecarIdentity); + } + + private Optional canonicalStorageRoot(String storageRootId, FileIO fileIO) { + return canonicalStorageRoots.computeIfAbsent( + storageRootId, + root -> { + try { + Path canonical = fileIO.getFileStatus(new Path(root)).getPath(); + return isAbsolute(canonical) + ? Optional.of(canonical) + : Optional.empty(); + } catch (IOException e) { + return Optional.empty(); + } + }); + } + } + + private boolean addManifestList( + ManifestList manifestList, + String listFileName, + List metas, + Consumer used) + throws IOException { + if (listFileName == null) { + return true; + } + List listed = + retryReadingFiles(() -> manifestList.readWithIOException(listFileName), null); + if (listed == null) { + LOG.warn( + "Manifest list {} is missing while collecting used managed blob packs. Skip pack GC this run.", + listFileName); + used.accept(SKIP_MANAGED_BLOB_GC); + return false; + } + metas.addAll(listed); + return true; + } + + /** Deletes a managed BLOB pack and returns whether this invocation deleted it. */ + protected boolean cleanManagedBlobFile(Path path) { + return cleanManagedBlobFile(path, false); + } + + /** + * Idempotently completes deletion of a managed BLOB pack. + * + *

Distributed attempts use this method so a pack deleted by a failed earlier attempt is + * still included in the successful attempt's logical cleanup result. + */ + protected boolean cleanManagedBlobFileIdempotently(Path path) { + return cleanManagedBlobFile(path, true); + } + + private boolean cleanManagedBlobFile(Path path, boolean missingIsSuccess) { + // Same convention as OrphanFilesClean: dry-run reports would-be deletes without removing + // files. + if (dryRun) { + return true; + } + try { + if (fileIO.isDir(path)) { + LOG.error( + "Refusing to delete directory {} in managed blob orphan cleanup. " + + "This indicates a bug in candidate collection.", + path); + return false; + } + } catch (FileNotFoundException e) { + return missingIsSuccess; + } catch (IOException e) { + LOG.warn("Failed to check whether managed blob pack {} is a directory.", path, e); + return false; + } + try { + if (fileIO.delete(path, false)) { + return true; + } + if (missingIsSuccess && !fileIO.exists(path)) { + return true; + } + LOG.warn("Failed to delete managed blob pack {}.", path); + return false; + } catch (FileNotFoundException e) { + return missingIsSuccess; + } catch (IOException e) { + LOG.warn("Failed to delete managed blob pack {}.", path, e); + return false; + } + } + + public static boolean isManagedBlobPackName(String fileName) { + return fileName.endsWith(ManagedBlobReferenceFile.MANAGED_BLOB_SUFFIX); + } +} diff --git a/paimon-core/src/test/java/org/apache/paimon/blob/ManagedBlobReachabilityCollectorTest.java b/paimon-core/src/test/java/org/apache/paimon/blob/ManagedBlobReachabilityCollectorTest.java new file mode 100644 index 000000000000..069df3b26ba5 --- /dev/null +++ b/paimon-core/src/test/java/org/apache/paimon/blob/ManagedBlobReachabilityCollectorTest.java @@ -0,0 +1,210 @@ +/* + * 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.blob; + +import org.apache.paimon.blob.ManagedBlobReferenceFile.Reference; +import org.apache.paimon.fs.Path; +import org.apache.paimon.fs.local.LocalFileIO; + +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.io.TempDir; + +import java.io.DataOutputStream; +import java.util.Arrays; +import java.util.Collections; + +import static org.assertj.core.api.Assertions.assertThat; + +/** Tests for {@link ManagedBlobReachabilityCollector}. */ +class ManagedBlobReachabilityCollectorTest { + + @TempDir java.nio.file.Path tempDir; + + @Test + void testEmptyExtraFiles() { + LocalFileIO fileIO = LocalFileIO.create(); + Path dataFile = new Path(tempDir.resolve("data.avro").toUri()); + ManagedBlobReachabilityCollector collector = new ManagedBlobReachabilityCollector(fileIO); + + ManagedBlobReachabilityCollector.Result result = + collector.fromDataFile(dataFile, Collections.emptyList()); + + assertThat(result.isUnsafe()).isFalse(); + assertThat(result.referenced()).isEmpty(); + } + + @Test + void testEmptySidecar() throws Exception { + LocalFileIO fileIO = LocalFileIO.create(); + Path dataFile = new Path(tempDir.resolve("data.avro").toUri()); + Path sidecar = ManagedBlobReferenceFile.sidecarPath(dataFile); + ManagedBlobReferenceFile.write(fileIO, sidecar, Collections.emptyList()); + ManagedBlobReachabilityCollector collector = new ManagedBlobReachabilityCollector(fileIO); + + ManagedBlobReachabilityCollector.Result result = + collector.fromDataFile(dataFile, Collections.singletonList(sidecar.getName())); + + assertThat(result.isUnsafe()).isFalse(); + assertThat(result.referenced()).isEmpty(); + } + + @Test + void testReferencedPacks() throws Exception { + LocalFileIO fileIO = LocalFileIO.create(); + Path dataFile = new Path(tempDir.resolve("data.avro").toUri()); + Path sidecar = ManagedBlobReferenceFile.sidecarPath(dataFile); + Reference first = + new Reference( + tempDir.resolve("bucket-0").toUri().toString(), "data-a.managed.blob"); + Reference second = + new Reference( + tempDir.resolve("bucket-0").toUri().toString(), "data-b.managed.blob"); + ManagedBlobReferenceFile.write(fileIO, sidecar, Arrays.asList(first, second)); + ManagedBlobReachabilityCollector collector = new ManagedBlobReachabilityCollector(fileIO); + + ManagedBlobReachabilityCollector.Result result = + collector.fromDataFile(dataFile, Collections.singletonList(sidecar.getName())); + + assertThat(result.isUnsafe()).isFalse(); + assertThat(result.referenced()).containsExactlyInAnyOrder(first, second); + assertThat(result.contains(first)).isTrue(); + assertThat(result.containsPackName("data-b.managed.blob")).isTrue(); + assertThat(result.containsPackName("missing.managed.blob")).isFalse(); + assertThat(first.toPath()) + .isEqualTo(new Path(tempDir.resolve("bucket-0/data-a.managed.blob").toUri())); + } + + @Test + void testReadSingleSidecar() throws Exception { + LocalFileIO fileIO = LocalFileIO.create(); + Path sidecar = new Path(tempDir.resolve("data.avro.blobref").toUri()); + Reference reference = + new Reference(tempDir.resolve("bucket-0").toUri().toString(), "data.managed.blob"); + ManagedBlobReferenceFile.write(fileIO, sidecar, Collections.singletonList(reference)); + + ManagedBlobReachabilityCollector.Result result = + new ManagedBlobReachabilityCollector(fileIO).fromSidecar(sidecar); + + assertThat(result.isUnsafe()).isFalse(); + assertThat(result.referenced()).containsExactly(reference); + } + + @Test + void testMissingSingleSidecarIsUnsafe() { + LocalFileIO fileIO = LocalFileIO.create(); + Path sidecar = new Path(tempDir.resolve("missing.blobref").toUri()); + + ManagedBlobReachabilityCollector.Result result = + new ManagedBlobReachabilityCollector(fileIO).fromSidecar(sidecar); + + assertThat(result.isUnsafe()).isTrue(); + assertThat(result.referenced()).isEmpty(); + } + + @Test + void testMissingSidecarUnsafeWhenDataFileExists() throws Exception { + LocalFileIO fileIO = LocalFileIO.create(); + Path dataFile = new Path(tempDir.resolve("data.avro").toUri()); + fileIO.newOutputStream(dataFile, false).close(); + ManagedBlobReachabilityCollector collector = new ManagedBlobReachabilityCollector(fileIO); + + ManagedBlobReachabilityCollector.Result result = + collector.fromDataFile(dataFile, Collections.singletonList("data.avro.blobref")); + + assertThat(result.isUnsafe()).isTrue(); + assertThat(result.referenced()).isEmpty(); + } + + @Test + void testMissingSidecarIgnoredWhenDataFileGone() { + LocalFileIO fileIO = LocalFileIO.create(); + Path dataFile = new Path(tempDir.resolve("expired.avro").toUri()); + ManagedBlobReachabilityCollector collector = new ManagedBlobReachabilityCollector(fileIO); + + ManagedBlobReachabilityCollector.Result result = + collector.fromDataFile(dataFile, Collections.singletonList("expired.avro.blobref")); + + assertThat(result.isUnsafe()).isFalse(); + assertThat(result.referenced()).isEmpty(); + } + + @Test + void testCorruptSidecarUnsafe() throws Exception { + LocalFileIO fileIO = LocalFileIO.create(); + Path dataFile = new Path(tempDir.resolve("data.avro").toUri()); + fileIO.newOutputStream(dataFile, false).close(); + Path sidecar = ManagedBlobReferenceFile.sidecarPath(dataFile); + try (DataOutputStream out = new DataOutputStream(fileIO.newOutputStream(sidecar, false))) { + out.writeInt(0x50424C52); + out.writeByte(1); + out.writeInt(0); + out.writeInt(12345); + } + ManagedBlobReachabilityCollector collector = new ManagedBlobReachabilityCollector(fileIO); + + ManagedBlobReachabilityCollector.Result result = + collector.fromDataFile(dataFile, Collections.singletonList(sidecar.getName())); + + assertThat(result.isUnsafe()).isTrue(); + } + + @Test + void testUnsupportedVersionUnsafe() throws Exception { + LocalFileIO fileIO = LocalFileIO.create(); + Path dataFile = new Path(tempDir.resolve("data.avro").toUri()); + fileIO.newOutputStream(dataFile, false).close(); + Path sidecar = ManagedBlobReferenceFile.sidecarPath(dataFile); + try (DataOutputStream out = new DataOutputStream(fileIO.newOutputStream(sidecar, false))) { + out.writeInt(0x50424C52); + out.writeByte(99); + out.writeInt(0); + } + ManagedBlobReachabilityCollector collector = new ManagedBlobReachabilityCollector(fileIO); + + ManagedBlobReachabilityCollector.Result result = + collector.fromDataFile(dataFile, Collections.singletonList(sidecar.getName())); + + assertThat(result.isUnsafe()).isTrue(); + } + + @Test + void testMergePropagatesUnsafe() throws Exception { + LocalFileIO fileIO = LocalFileIO.create(); + Path dataFile = new Path(tempDir.resolve("data.avro").toUri()); + Path sidecar = ManagedBlobReferenceFile.sidecarPath(dataFile); + Reference referenced = + new Reference( + tempDir.resolve("bucket-0").toUri().toString(), "data-a.managed.blob"); + ManagedBlobReferenceFile.write(fileIO, sidecar, Collections.singletonList(referenced)); + ManagedBlobReachabilityCollector collector = new ManagedBlobReachabilityCollector(fileIO); + + ManagedBlobReachabilityCollector.Result safe = + collector.fromDataFile(dataFile, Collections.singletonList(sidecar.getName())); + ManagedBlobReachabilityCollector.Result merged = + safe.merge(ManagedBlobReachabilityCollector.Result.unsafe()); + + assertThat(merged.isUnsafe()).isTrue(); + assertThat(merged.referenced()).containsExactly(referenced); + assertThat( + ManagedBlobReachabilityCollector.Result.empty() + .merge(ManagedBlobReachabilityCollector.Result.unsafe()) + .isUnsafe()) + .isTrue(); + } +} diff --git a/paimon-core/src/test/java/org/apache/paimon/operation/ManagedBlobOrphanFilesCleanTest.java b/paimon-core/src/test/java/org/apache/paimon/operation/ManagedBlobOrphanFilesCleanTest.java new file mode 100644 index 000000000000..bdc323d1dc90 --- /dev/null +++ b/paimon-core/src/test/java/org/apache/paimon/operation/ManagedBlobOrphanFilesCleanTest.java @@ -0,0 +1,1645 @@ +/* + * 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.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.data.BinaryRow; +import org.apache.paimon.data.BinaryString; +import org.apache.paimon.data.BlobData; +import org.apache.paimon.data.GenericRow; +import org.apache.paimon.fs.FileIO; +import org.apache.paimon.fs.FileStatus; +import org.apache.paimon.fs.Path; +import org.apache.paimon.fs.SeekableInputStream; +import org.apache.paimon.fs.local.LocalFileIO; +import org.apache.paimon.io.DataFileMeta; +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.schema.Schema; +import org.apache.paimon.table.FileStoreTable; +import org.apache.paimon.table.FileStoreTableFactory; +import org.apache.paimon.table.TableTestBase; +import org.apache.paimon.table.sink.BatchTableCommit; +import org.apache.paimon.table.sink.BatchTableWrite; +import org.apache.paimon.table.sink.BatchWriteBuilder; +import org.apache.paimon.table.sink.CommitMessage; +import org.apache.paimon.table.sink.CommitMessageImpl; +import org.apache.paimon.types.DataTypes; +import org.apache.paimon.utils.DataFilePathFactories; +import org.apache.paimon.utils.InstantiationUtil; +import org.apache.paimon.utils.TraceableFileIO; + +import org.junit.jupiter.api.Test; + +import java.io.DataOutputStream; +import java.io.FileNotFoundException; +import java.io.IOException; +import java.nio.file.Paths; +import java.util.ArrayList; +import java.util.HashMap; +import java.util.HashSet; +import java.util.List; +import java.util.Map; +import java.util.Set; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.AtomicReference; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.catchThrowable; + +/** Tests orphan-file cleanup of unreferenced primary-key managed BLOB packs. */ +public class ManagedBlobOrphanFilesCleanTest extends TableTestBase { + + @Test + public void testDeleteUnreferencedPack() throws Exception { + FileStoreTable table = createManagedBlobTable("orphan_pack"); + write( + table, + GenericRow.of(1, BinaryString.fromString("a"), new BlobData(new byte[] {1, 2}))); + + Path orphan = new Path(bucketPath(table), "orphan.managed.blob"); + table.fileIO().newOutputStream(orphan, false).close(); + + List deleted = clean(table); + assertThat(table.fileIO().exists(orphan)).isFalse(); + assertThat(deleted).extracting(Path::getName).contains("orphan.managed.blob"); + assertThat(managedBlobs(table)).isNotEmpty(); + } + + @Test + public void testDeleteFailureIsNotReported() throws Exception { + FileStoreTable table = createManagedBlobTable("delete_failure"); + Path orphan = new Path(bucketPath(table), "delete-failure.managed.blob"); + table.fileIO().newOutputStream(orphan, false).close(); + + FileIO deleteFailingFileIO = + new LocalFileIO() { + @Override + public boolean delete(Path path, boolean recursive) throws IOException { + if (path.getName().equals(orphan.getName())) { + return false; + } + return super.delete(path, recursive); + } + }; + FileStoreTable failingTable = + FileStoreTableFactory.create(deleteFailingFileIO, table.location(), table.schema()); + + CleanOrphanFilesResult result = + new LocalManagedBlobOrphanFilesClean( + failingTable, + System.currentTimeMillis() + TimeUnit.SECONDS.toMillis(2), + false) + .clean(); + + assertThat(result.getDeletedFileCount()).isZero(); + assertThat(result.getDeletedFileTotalLenInBytes()).isZero(); + assertThat(result.getDeletedFilesPath()).isEmpty(); + assertThat(table.fileIO().exists(orphan)).isTrue(); + } + + @Test + public void testAlreadyAbsentPackIsNotReportedAsDeletedLocally() throws Exception { + FileStoreTable table = createManagedBlobTable("already_absent"); + Path orphan = new Path(bucketPath(table), "already-absent.managed.blob"); + table.fileIO().mkdirs(orphan.getParent()); + table.fileIO().newOutputStream(orphan, false).close(); + + FileIO alreadyDeletedFileIO = + new LocalFileIO() { + @Override + public boolean delete(Path path, boolean recursive) throws IOException { + if (path.getName().equals(orphan.getName())) { + super.delete(path, recursive); + return false; + } + return super.delete(path, recursive); + } + }; + FileStoreTable alreadyDeletedTable = + FileStoreTableFactory.create( + alreadyDeletedFileIO, table.location(), table.schema()); + + CleanOrphanFilesResult result = + new LocalManagedBlobOrphanFilesClean( + alreadyDeletedTable, + System.currentTimeMillis() + TimeUnit.SECONDS.toMillis(2), + false) + .clean(); + + assertThat(result.getDeletedFileCount()).isZero(); + assertThat(result.getDeletedFileTotalLenInBytes()).isZero(); + assertThat(result.getDeletedFilesPath()).isEmpty(); + assertThat(table.fileIO().exists(orphan)).isFalse(); + } + + @Test + public void testPackVanishingBeforeDirectoryCheckIsNotReportedAsDeletedLocally() + throws Exception { + FileStoreTable table = createManagedBlobTable("vanished_before_check"); + Path orphan = new Path(bucketPath(table), "vanished.managed.blob"); + table.fileIO().mkdirs(orphan.getParent()); + table.fileIO().newOutputStream(orphan, false).close(); + + // Another worker removes the pack between candidate listing and the directory check, so + // isDir raises FileNotFoundException instead of returning. Overriding isDir rather than + // getFileStatus keeps candidate listing intact: LocalFileIO.listStatus resolves each entry + // through getFileStatus and silently drops the ones that raise FileNotFoundException. + FileIO vanishingFileIO = + new LocalFileIO() { + @Override + public boolean isDir(Path path) throws IOException { + if (path.getName().equals(orphan.getName())) { + super.delete(path, false); + } + return super.isDir(path); + } + }; + FileStoreTable vanishingTable = + FileStoreTableFactory.create(vanishingFileIO, table.location(), table.schema()); + + CleanOrphanFilesResult result = + new LocalManagedBlobOrphanFilesClean( + vanishingTable, + System.currentTimeMillis() + TimeUnit.SECONDS.toMillis(2), + false) + .clean(); + + assertThat(result.getDeletedFileCount()).isZero(); + assertThat(result.getDeletedFileTotalLenInBytes()).isZero(); + assertThat(result.getDeletedFilesPath()).isEmpty(); + assertThat(table.fileIO().exists(orphan)).isFalse(); + } + + @Test + public void testIdempotentDeleteCountsPackRemovedBeforeExistsCheck() throws Exception { + FileStoreTable table = createManagedBlobTable("idempotent_delete_false"); + Path orphan = new Path(bucketPath(table), "idempotent-delete-false.managed.blob"); + table.fileIO().newOutputStream(orphan, false).close(); + + FileIO disappearingFileIO = + new LocalFileIO() { + @Override + public boolean delete(Path path, boolean recursive) throws IOException { + if (path.getName().equals(orphan.getName())) { + super.delete(path, recursive); + return false; + } + return super.delete(path, recursive); + } + }; + FileStoreTable disappearingTable = + FileStoreTableFactory.create(disappearingFileIO, table.location(), table.schema()); + LocalManagedBlobOrphanFilesClean cleaner = + new LocalManagedBlobOrphanFilesClean( + disappearingTable, + System.currentTimeMillis() + TimeUnit.SECONDS.toMillis(2), + false); + + assertThat(cleaner.cleanManagedBlobFileIdempotently(orphan)).isTrue(); + assertThat(table.fileIO().exists(orphan)).isFalse(); + } + + @Test + public void testIdempotentDeleteCountsFileNotFoundDuringDelete() throws Exception { + FileStoreTable table = createManagedBlobTable("idempotent_delete_not_found"); + Path orphan = new Path(bucketPath(table), "idempotent-delete-not-found.managed.blob"); + table.fileIO().newOutputStream(orphan, false).close(); + + FileIO disappearingFileIO = + new LocalFileIO() { + @Override + public boolean delete(Path path, boolean recursive) throws IOException { + if (path.getName().equals(orphan.getName())) { + super.delete(path, recursive); + throw new FileNotFoundException(path.toString()); + } + return super.delete(path, recursive); + } + }; + FileStoreTable disappearingTable = + FileStoreTableFactory.create(disappearingFileIO, table.location(), table.schema()); + LocalManagedBlobOrphanFilesClean cleaner = + new LocalManagedBlobOrphanFilesClean( + disappearingTable, + System.currentTimeMillis() + TimeUnit.SECONDS.toMillis(2), + false); + + assertThat(cleaner.cleanManagedBlobFileIdempotently(orphan)).isTrue(); + assertThat(table.fileIO().exists(orphan)).isFalse(); + } + + @Test + public void testKeepReferencedPack() throws Exception { + FileStoreTable table = createManagedBlobTable("keep_pack"); + write( + table, + GenericRow.of(1, BinaryString.fromString("a"), new BlobData(new byte[] {9, 8, 7}))); + List before = managedBlobs(table); + assertThat(before).isNotEmpty(); + + List deleted = clean(table); + + assertThat(deleted).doesNotContainAnyElementsOf(before); + for (Path pack : before) { + assertThat(table.fileIO().exists(pack)).isTrue(); + } + assertThat(read(table)).hasSize(1); + assertThat(read(table).get(0).getBlob(2).toData()) + .containsExactly((byte) 9, (byte) 8, (byte) 7); + } + + @Test + public void testPackIdentityIgnoresFileSystemQualification() { + Path listed = new Path("file:/tmp/table/bucket-0/data-a.managed.blob"); + Path referenced = new Path("traceable:/tmp/table/bucket-0/data-a.managed.blob"); + Path unqualifiedHdfs = new Path("hdfs:///tmp/table/bucket-0/data-a.managed.blob"); + Path qualifiedHdfs = + new Path("hdfs://namenode:8020/tmp/table/bucket-0/data-a.managed.blob"); + Path otherBucket = new Path("file:/tmp/table/bucket-1/data-a.managed.blob"); + assertThat(ManagedBlobOrphanFilesClean.packIdentity(referenced)) + .isEqualTo(ManagedBlobOrphanFilesClean.packIdentity(listed)); + assertThat(ManagedBlobOrphanFilesClean.packIdentity(unqualifiedHdfs)) + .isEqualTo(ManagedBlobOrphanFilesClean.packIdentity(qualifiedHdfs)); + assertThat(ManagedBlobOrphanFilesClean.packIdentity(otherBucket)) + .isNotEqualTo(ManagedBlobOrphanFilesClean.packIdentity(listed)); + } + + @Test + public void testRelativeTablePathDoesNotDeleteReferencedPack() throws Exception { + FileStoreTable table = createManagedBlobTable("relative_table_path"); + java.nio.file.Path absoluteLocation = + Paths.get(table.location().toUri().getPath()).toAbsolutePath(); + String relativeLocation = + Paths.get("").toAbsolutePath().relativize(absoluteLocation).toString(); + FileStoreTable relativeTable = + FileStoreTableFactory.create( + new LocalFileIO(), new Path(relativeLocation), table.schema()); + write( + relativeTable, + GenericRow.of(1, BinaryString.fromString("a"), new BlobData(new byte[] {1, 2}))); + List referenced = managedBlobs(relativeTable); + assertThat(referenced).isNotEmpty(); + + CleanOrphanFilesResult result; + try (LocalManagedBlobOrphanFilesClean cleaner = + new LocalManagedBlobOrphanFilesClean( + relativeTable, + System.currentTimeMillis() + TimeUnit.SECONDS.toMillis(2), + false)) { + result = cleaner.clean(); + } + + assertThat(result.getDeletedFilesPath()).isEmpty(); + for (Path pack : referenced) { + assertThat(relativeTable.fileIO().exists(pack)).isTrue(); + } + assertThat(read(relativeTable)).hasSize(1); + } + + @Test + public void testUnresolvedNonLocalRelativeListingSkipsPackGc() throws Exception { + FileStoreTable table = createManagedBlobTable("non_local_relative_listing"); + write( + table, + GenericRow.of(1, BinaryString.fromString("a"), new BlobData(new byte[] {1, 2}))); + Path orphan = new Path(bucketPath(table), "orphan.managed.blob"); + table.fileIO().newOutputStream(orphan, false).close(); + List packs = managedBlobs(table); + + RelativeListingFileIO relativeListingFileIO = new RelativeListingFileIO(); + FileStoreTable relativeListingTable = + FileStoreTableFactory.create( + relativeListingFileIO, table.location(), table.schema()); + CleanOrphanFilesResult result; + try (LocalManagedBlobOrphanFilesClean cleaner = + new LocalManagedBlobOrphanFilesClean(relativeListingTable, Long.MAX_VALUE, false)) { + result = cleaner.clean(); + } + + assertThat(result.getDeletedFilesPath()).isEmpty(); + assertThat(relativeListingFileIO.managedBlobDeleteAttempts()).isZero(); + for (Path pack : packs) { + assertThat(table.fileIO().exists(pack)).isTrue(); + } + } + + @Test + public void testRepeatedCleanDoesNotAccumulateResults() throws Exception { + FileStoreTable table = createManagedBlobTable("repeated_clean"); + Path orphan = new Path(bucketPath(table), "orphan.managed.blob"); + table.fileIO().mkdirs(orphan.getParent()); + table.fileIO().newOutputStream(orphan, false).close(); + + try (LocalManagedBlobOrphanFilesClean cleaner = + new LocalManagedBlobOrphanFilesClean(table, Long.MAX_VALUE, true)) { + CleanOrphanFilesResult first = cleaner.clean(); + CleanOrphanFilesResult second = cleaner.clean(); + + assertThat(first.getDeletedFilesPath()) + .extracting(Path::getName) + .containsExactly(orphan.getName()); + assertThat(second.getDeletedFilesPath()) + .extracting(Path::getName) + .containsExactly(orphan.getName()); + assertThat(first.getDeletedFileCount()).isEqualTo(1); + assertThat(second.getDeletedFileCount()).isEqualTo(1); + } + } + + @Test + public void testQualifiedListingDoesNotDeleteReferencedPack() throws Exception { + FileStoreTable table = createManagedBlobTable("qualified_listing"); + write( + table, + GenericRow.of(1, BinaryString.fromString("a"), new BlobData(new byte[] {1, 2}))); + List referenced = managedBlobs(table); + assertThat(referenced).isNotEmpty(); + + QualifiedListingFileIO qualifiedFileIO = new QualifiedListingFileIO(); + FileStoreTable qualifiedTable = + FileStoreTableFactory.create(qualifiedFileIO, table.location(), table.schema()); + CleanOrphanFilesResult result; + try (LocalManagedBlobOrphanFilesClean cleaner = + new LocalManagedBlobOrphanFilesClean( + qualifiedTable, + System.currentTimeMillis() + TimeUnit.SECONDS.toMillis(2), + false)) { + result = cleaner.clean(); + } + + assertThat(result.getDeletedFilesPath()).isEmpty(); + assertThat(qualifiedFileIO.managedBlobDeleteAttempts()).isZero(); + for (Path pack : referenced) { + assertThat(table.fileIO().exists(pack)).isTrue(); + } + } + + @Test + public void testReachabilityScanDeduplicatesReadsPerPass() throws Exception { + FileStoreTable table = createManagedBlobTable("deduplicate_reachability_reads"); + write( + table, + GenericRow.of(1, BinaryString.fromString("a"), new BlobData(new byte[] {1, 2}))); + write( + table, + GenericRow.of(2, BinaryString.fromString("b"), new BlobData(new byte[] {3, 4}))); + + CountingInputFileIO countingFileIO = new CountingInputFileIO(); + FileStoreTable countingTable = + FileStoreTableFactory.create(countingFileIO, table.location(), table.schema()); + try (LocalManagedBlobOrphanFilesClean cleaner = + new LocalManagedBlobOrphanFilesClean( + countingTable, + System.currentTimeMillis() + TimeUnit.SECONDS.toMillis(2), + false)) { + cleaner.collectUsedPacks(); + assertReadOncePerPass(countingFileIO.readCounts()); + + countingFileIO.reset(); + cleaner.collectUsedPacks(); + assertReadOncePerPass(countingFileIO.readCounts()); + } + } + + @Test + public void testSidecarWorkItemIsStableSerializableAndAddOnly() throws Exception { + FileStoreTable table = createManagedBlobTable("sidecar_work_item"); + write( + table, + GenericRow.of(1, BinaryString.fromString("a"), new BlobData(new byte[] {1, 2}))); + ManifestEntry add = table.store().newScan().plan().files().get(0); + DataFilePathFactory pathFactory = + new DataFilePathFactories(table.store().pathFactory()) + .get(add.partition(), add.bucket()); + + try (LocalManagedBlobOrphanFilesClean cleaner = + new LocalManagedBlobOrphanFilesClean(table, Long.MAX_VALUE, true)) { + List workItems = + cleaner.createSidecarWorkItems(add, pathFactory); + assertThat(workItems).isNotEmpty(); + assertThat(cleaner.createSidecarWorkItems(add, pathFactory)) + .extracting(ManagedBlobOrphanFilesClean.SidecarWorkItem::dedupIdentity) + .containsExactlyElementsOf( + workItems.stream() + .map(ManagedBlobOrphanFilesClean.SidecarWorkItem::dedupIdentity) + .collect(java.util.stream.Collectors.toList())); + + ManagedBlobOrphanFilesClean.SidecarWorkItem workItem = workItems.get(0); + ManagedBlobOrphanFilesClean.SidecarWorkItem restored = + InstantiationUtil.clone(workItem); + assertThat(restored).isEqualTo(workItem); + assertThat(restored.dataFile()).isEqualTo(workItem.dataFile()); + assertThat(restored.sidecar()).isEqualTo(workItem.sidecar()); + assertThat(restored.extraFile()).isEqualTo(workItem.extraFile()); + + ManifestEntry delete = + ManifestEntry.create( + FileKind.DELETE, + add.partition(), + add.bucket(), + add.totalBuckets(), + add.file()); + assertThat(cleaner.createSidecarWorkItems(delete, pathFactory)).isEmpty(); + } + } + + @Test + public void testExecuteDatabaseClosesExecutorsAfterTaskFailure() throws Exception { + FileStoreTable table = createManagedBlobTable("database_failure_cleanup"); + CountDownLatch secondStarted = new CountDownLatch(1); + CountDownLatch secondInterrupted = new CountDownLatch(1); + AtomicBoolean firstClosed = new AtomicBoolean(); + AtomicBoolean secondClosed = new AtomicBoolean(); + LocalManagedBlobOrphanFilesClean first = + new LocalManagedBlobOrphanFilesClean(table, Long.MAX_VALUE, false) { + @Override + public CleanOrphanFilesResult clean() throws IOException { + try { + if (!secondStarted.await(10, TimeUnit.SECONDS)) { + throw new IOException("Second cleanup did not start."); + } + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new IOException(e); + } + throw new IOException("Expected cleanup failure."); + } + + @Override + public void close() { + super.close(); + firstClosed.set(true); + } + }; + LocalManagedBlobOrphanFilesClean second = + new LocalManagedBlobOrphanFilesClean(table, Long.MAX_VALUE, false) { + @Override + public CleanOrphanFilesResult clean() throws IOException { + secondStarted.countDown(); + try { + new CountDownLatch(1).await(); + throw new IOException("Unexpected completion."); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + secondInterrupted.countDown(); + throw new IOException(e); + } + } + + @Override + public void close() { + super.close(); + secondClosed.set(true); + } + }; + ExecutorService databaseExecutor = Executors.newFixedThreadPool(2); + + Throwable failure = + catchThrowable( + () -> + LocalManagedBlobOrphanFilesClean.executeDatabase( + java.util.Arrays.asList(first, second), databaseExecutor)); + + assertThat(failure).isInstanceOf(RuntimeException.class); + assertThat(databaseExecutor.isShutdown()).isTrue(); + assertThat(firstClosed).isTrue(); + assertThat(secondClosed).isTrue(); + assertThat(secondInterrupted.await(10, TimeUnit.SECONDS)).isTrue(); + } + + @Test + public void testExecuteDatabaseObservesLaterFailureBeforeEarlierTaskCompletes() + throws Exception { + FileStoreTable table = createManagedBlobTable("database_completion_order"); + CountDownLatch blockedStarted = new CountDownLatch(1); + CountDownLatch blockedInterrupted = new CountDownLatch(1); + LocalManagedBlobOrphanFilesClean blocked = + new LocalManagedBlobOrphanFilesClean(table, Long.MAX_VALUE, false) { + @Override + public CleanOrphanFilesResult clean() throws IOException { + blockedStarted.countDown(); + try { + new CountDownLatch(1).await(); + throw new IOException("Unexpected completion."); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + blockedInterrupted.countDown(); + throw new IOException(e); + } + } + }; + LocalManagedBlobOrphanFilesClean failing = + new LocalManagedBlobOrphanFilesClean(table, Long.MAX_VALUE, false) { + @Override + public CleanOrphanFilesResult clean() throws IOException { + try { + if (!blockedStarted.await(10, TimeUnit.SECONDS)) { + throw new IOException("Earlier cleanup did not start."); + } + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new IOException(e); + } + throw new IOException("Expected later cleanup failure."); + } + }; + ExecutorService databaseExecutor = Executors.newFixedThreadPool(2); + CountDownLatch executeReturned = new CountDownLatch(1); + AtomicReference failure = new AtomicReference<>(); + Thread caller = + new Thread( + () -> { + failure.set( + catchThrowable( + () -> + LocalManagedBlobOrphanFilesClean + .executeDatabase( + java.util.Arrays.asList( + blocked, failing), + databaseExecutor))); + executeReturned.countDown(); + }); + caller.start(); + + boolean returnedInTime = executeReturned.await(10, TimeUnit.SECONDS); + if (!returnedInTime) { + caller.interrupt(); + } + caller.join(TimeUnit.SECONDS.toMillis(10)); + + assertThat(returnedInTime).isTrue(); + assertThat(failure.get()).isInstanceOf(RuntimeException.class); + assertThat(blockedInterrupted.await(10, TimeUnit.SECONDS)).isTrue(); + assertThat(databaseExecutor.isTerminated()).isTrue(); + } + + @Test + public void testCollectUsedPacksObservesLaterSnapshotFailureBeforeEarlierSnapshotCompletes() + throws Exception { + CountDownLatch firstTaskStarted = new CountDownLatch(1); + AtomicBoolean releaseStuckTask = new AtomicBoolean(); + ExecutorService snapshotExecutor = Executors.newFixedThreadPool(2); + CountDownLatch executeReturned = new CountDownLatch(1); + AtomicReference failure = new AtomicReference<>(); + Thread caller = + new Thread( + () -> { + failure.set( + catchThrowable( + () -> + LocalManagedBlobOrphanFilesClean + .executeSnapshotsInCompletionOrder( + snapshotExecutor, + id -> { + if (id == 1) { + firstTaskStarted + .countDown(); + while (!releaseStuckTask + .get()) { + try { + Thread.sleep( + 20); + } catch ( + InterruptedException + ignored) { + // Keep the + // earlier task + // stuck so the + // later + // failure must + // be observed + // first. + } + } + } else { + // Do not fail until the + // earlier snapshot is + // running. Otherwise + // cancelAll() can drop + // the queued first task + // and the test never + // sees it start. + try { + if (!firstTaskStarted + .await( + 10, + TimeUnit + .SECONDS)) { + throw new IllegalStateException( + "Earlier snapshot task did not start."); + } + } catch ( + InterruptedException + e) { + Thread + .currentThread() + .interrupt(); + throw new RuntimeException( + e); + } + throw new RuntimeException( + new IOException( + "Expected later snapshot sidecar failure.")); + } + }, + java.util.Arrays.asList( + 1, 2)))); + executeReturned.countDown(); + }); + caller.start(); + + try { + assertThat(firstTaskStarted.await(10, TimeUnit.SECONDS)).isTrue(); + boolean returnedInTime = executeReturned.await(10, TimeUnit.SECONDS); + if (!returnedInTime) { + caller.interrupt(); + } + caller.join(TimeUnit.SECONDS.toMillis(10)); + + assertThat(returnedInTime).isTrue(); + assertThat(failure.get()) + .isInstanceOf(RuntimeException.class) + .hasRootCauseMessage("Expected later snapshot sidecar failure."); + } finally { + releaseStuckTask.set(true); + caller.join(TimeUnit.SECONDS.toMillis(10)); + snapshotExecutor.shutdownNow(); + snapshotExecutor.awaitTermination(10, TimeUnit.SECONDS); + } + } + + @Test + public void testExecuteDatabasePreservesInterruptAndClosesExecutors() throws Exception { + FileStoreTable table = createManagedBlobTable("database_interrupt_cleanup"); + CountDownLatch cleanStarted = new CountDownLatch(1); + CountDownLatch cleanInterrupted = new CountDownLatch(1); + AtomicBoolean cleanerClosed = new AtomicBoolean(); + AtomicBoolean callerInterrupted = new AtomicBoolean(); + LocalManagedBlobOrphanFilesClean cleaner = + new LocalManagedBlobOrphanFilesClean(table, Long.MAX_VALUE, false) { + @Override + public CleanOrphanFilesResult clean() throws IOException { + cleanStarted.countDown(); + try { + new CountDownLatch(1).await(); + throw new IOException("Unexpected completion."); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + cleanInterrupted.countDown(); + throw new IOException(e); + } + } + + @Override + public void close() { + super.close(); + cleanerClosed.set(true); + } + }; + ExecutorService databaseExecutor = Executors.newSingleThreadExecutor(); + Thread caller = + new Thread( + () -> { + try { + LocalManagedBlobOrphanFilesClean.executeDatabase( + java.util.Collections.singletonList(cleaner), + databaseExecutor); + } catch (RuntimeException e) { + callerInterrupted.set(Thread.currentThread().isInterrupted()); + } + }); + caller.start(); + assertThat(cleanStarted.await(10, TimeUnit.SECONDS)).isTrue(); + + caller.interrupt(); + caller.join(TimeUnit.SECONDS.toMillis(10)); + + assertThat(caller.isAlive()).isFalse(); + assertThat(callerInterrupted).isTrue(); + assertThat(databaseExecutor.isShutdown()).isTrue(); + assertThat(cleanerClosed).isTrue(); + assertThat(cleanInterrupted.await(10, TimeUnit.SECONDS)).isTrue(); + } + + @Test + public void testExecuteDatabaseWaitsForInterruptedDeleteLoop() throws Exception { + FileStoreTable table = createManagedBlobTable("database_delete_loop_cleanup"); + Path firstOrphan = new Path(bucketPath(table), "first.managed.blob"); + Path secondOrphan = new Path(bucketPath(table), "second.managed.blob"); + table.fileIO().mkdirs(firstOrphan.getParent()); + table.fileIO().newOutputStream(firstOrphan, false).close(); + table.fileIO().newOutputStream(secondOrphan, false).close(); + + CountDownLatch firstDeleteStarted = new CountDownLatch(1); + CountDownLatch releaseUnexpectedSecondDelete = new CountDownLatch(1); + AtomicInteger deleteAttempts = new AtomicInteger(); + FileIO interruptibleDeleteFileIO = + new LocalFileIO() { + @Override + public boolean delete(Path path, boolean recursive) throws IOException { + if (!path.getName() + .endsWith(ManagedBlobReferenceFile.MANAGED_BLOB_SUFFIX)) { + return super.delete(path, recursive); + } + int attempt = deleteAttempts.incrementAndGet(); + if (attempt == 1) { + firstDeleteStarted.countDown(); + try { + new CountDownLatch(1).await(); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + return false; + } + while (releaseUnexpectedSecondDelete.getCount() > 0) { + try { + releaseUnexpectedSecondDelete.await(); + } catch (InterruptedException ignored) { + // Keep the unexpected delete observable until the assertion. + } + } + return false; + } + }; + FileStoreTable interruptibleTable = + FileStoreTableFactory.create( + interruptibleDeleteFileIO, table.location(), table.schema()); + AtomicBoolean actualCleanerReturned = new AtomicBoolean(); + LocalManagedBlobOrphanFilesClean actualCleaner = + new LocalManagedBlobOrphanFilesClean(interruptibleTable, Long.MAX_VALUE, false) { + @Override + public CleanOrphanFilesResult clean() throws IOException { + try { + return super.clean(); + } finally { + actualCleanerReturned.set(true); + } + } + }; + LocalManagedBlobOrphanFilesClean failingCleaner = + new LocalManagedBlobOrphanFilesClean(table, Long.MAX_VALUE, false) { + @Override + public CleanOrphanFilesResult clean() throws IOException { + try { + if (!firstDeleteStarted.await(10, TimeUnit.SECONDS)) { + throw new IOException("Managed blob deletion did not start."); + } + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new IOException(e); + } + throw new IOException("Expected cleanup failure."); + } + }; + ExecutorService databaseExecutor = Executors.newFixedThreadPool(2); + + try { + Throwable failure = + catchThrowable( + () -> + LocalManagedBlobOrphanFilesClean.executeDatabase( + java.util.Arrays.asList(failingCleaner, actualCleaner), + databaseExecutor)); + + assertThat(failure).isInstanceOf(RuntimeException.class); + assertThat(actualCleanerReturned).isTrue(); + assertThat(deleteAttempts).hasValue(1); + assertThat(databaseExecutor.isTerminated()).isTrue(); + assertThat(table.fileIO().exists(firstOrphan)).isTrue(); + assertThat(table.fileIO().exists(secondOrphan)).isTrue(); + } finally { + releaseUnexpectedSecondDelete.countDown(); + } + } + + @Test + public void testExecuteDatabaseDoesNotHangOnUninterruptibleDelete() throws Exception { + FileStoreTable table = createManagedBlobTable("database_uninterruptible_delete"); + Path firstOrphan = new Path(bucketPath(table), "first.managed.blob"); + Path secondOrphan = new Path(bucketPath(table), "second.managed.blob"); + table.fileIO().mkdirs(firstOrphan.getParent()); + table.fileIO().newOutputStream(firstOrphan, false).close(); + table.fileIO().newOutputStream(secondOrphan, false).close(); + + CountDownLatch firstDeleteStarted = new CountDownLatch(1); + AtomicBoolean releaseStuckDelete = new AtomicBoolean(); + AtomicInteger deleteAttempts = new AtomicInteger(); + FileIO uninterruptibleDeleteFileIO = + new LocalFileIO() { + @Override + public boolean delete(Path path, boolean recursive) throws IOException { + if (!path.getName() + .endsWith(ManagedBlobReferenceFile.MANAGED_BLOB_SUFFIX)) { + return super.delete(path, recursive); + } + int attempt = deleteAttempts.incrementAndGet(); + if (attempt == 1) { + firstDeleteStarted.countDown(); + boolean interrupted = false; + while (!releaseStuckDelete.get()) { + try { + Thread.sleep(20); + } catch (InterruptedException e) { + interrupted = true; + } + } + if (interrupted) { + Thread.currentThread().interrupt(); + } + return false; + } + return false; + } + }; + FileStoreTable uninterruptibleTable = + FileStoreTableFactory.create( + uninterruptibleDeleteFileIO, table.location(), table.schema()); + LocalManagedBlobOrphanFilesClean stuckCleaner = + new LocalManagedBlobOrphanFilesClean(uninterruptibleTable, Long.MAX_VALUE, false); + LocalManagedBlobOrphanFilesClean failingCleaner = + new LocalManagedBlobOrphanFilesClean(table, Long.MAX_VALUE, false) { + @Override + public CleanOrphanFilesResult clean() throws IOException { + try { + if (!firstDeleteStarted.await(10, TimeUnit.SECONDS)) { + throw new IOException("Managed blob deletion did not start."); + } + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new IOException(e); + } + throw new IOException("Expected cleanup failure."); + } + }; + ExecutorService databaseExecutor = Executors.newFixedThreadPool(2); + CountDownLatch executeReturned = new CountDownLatch(1); + AtomicReference failure = new AtomicReference<>(); + Thread caller = + new Thread( + () -> { + failure.set( + catchThrowable( + () -> + LocalManagedBlobOrphanFilesClean + .executeDatabase( + java.util.Arrays.asList( + failingCleaner, + stuckCleaner), + databaseExecutor, + 200L))); + executeReturned.countDown(); + }); + caller.start(); + + try { + assertThat(executeReturned.await(10, TimeUnit.SECONDS)).isTrue(); + assertThat(failure.get()) + .isInstanceOf(RuntimeException.class) + .hasRootCauseMessage("Expected cleanup failure."); + assertThat(deleteAttempts).hasValue(1); + assertThat(table.fileIO().exists(firstOrphan)).isTrue(); + assertThat(table.fileIO().exists(secondOrphan)).isTrue(); + assertThat(databaseExecutor.isTerminated()).isFalse(); + } finally { + releaseStuckDelete.set(true); + caller.join(TimeUnit.SECONDS.toMillis(10)); + databaseExecutor.awaitTermination(10, TimeUnit.SECONDS); + } + assertThat(deleteAttempts).hasValue(1); + assertThat(table.fileIO().exists(firstOrphan)).isTrue(); + assertThat(table.fileIO().exists(secondOrphan)).isTrue(); + } + + @Test + public void testEmptySidecarDoesNotBlockOthers() throws Exception { + FileStoreTable table = createManagedBlobTable("empty_sidecar"); + write(table, GenericRow.of(1, BinaryString.fromString("a"), null)); + + Path orphan = new Path(bucketPath(table), "orphan.managed.blob"); + table.fileIO().newOutputStream(orphan, false).close(); + + List deleted = clean(table); + assertThat(table.fileIO().exists(orphan)).isFalse(); + assertThat(deleted).extracting(Path::getName).contains("orphan.managed.blob"); + } + + @Test + public void testMissingSidecarSkipsAllPacks() throws Exception { + FileStoreTable table = createManagedBlobTable("missing_sidecar"); + write(table, GenericRow.of(1, BinaryString.fromString("a"), new BlobData(new byte[] {1}))); + Path orphan = new Path(bucketPath(table), "orphan.managed.blob"); + table.fileIO().newOutputStream(orphan, false).close(); + + deleteSidecars(table); + List referenced = managedBlobs(table); + referenced.remove(orphan); + + clean(table); + + assertThat(table.fileIO().exists(orphan)).isTrue(); + for (Path pack : referenced) { + assertThat(table.fileIO().exists(pack)).isTrue(); + } + } + + @Test + public void testCorruptSidecarSkipsAllPacks() throws Exception { + FileStoreTable table = createManagedBlobTable("corrupt_sidecar"); + write(table, GenericRow.of(1, BinaryString.fromString("a"), new BlobData(new byte[] {1}))); + Path orphan = new Path(bucketPath(table), "orphan.managed.blob"); + table.fileIO().newOutputStream(orphan, false).close(); + + overwriteSidecars( + table, + out -> { + out.writeInt(0x50424C52); + out.writeByte(1); + out.writeInt(0); + out.writeInt(12345); + }); + + clean(table); + assertThat(table.fileIO().exists(orphan)).isTrue(); + } + + @Test + public void testUnsupportedVersionSkipsAllPacks() throws Exception { + FileStoreTable table = createManagedBlobTable("unsupported_sidecar"); + write(table, GenericRow.of(1, BinaryString.fromString("a"), new BlobData(new byte[] {1}))); + Path orphan = new Path(bucketPath(table), "orphan.managed.blob"); + table.fileIO().newOutputStream(orphan, false).close(); + + overwriteSidecars( + table, + out -> { + out.writeInt(0x50424C52); + out.writeByte(99); + out.writeInt(0); + }); + + clean(table); + assertThat(table.fileIO().exists(orphan)).isTrue(); + } + + @Test + public void testUnreferencedAfterUpdateAndExpire() throws Exception { + FileStoreTable table = createManagedBlobTable("update_expire"); + write( + table, + GenericRow.of(1, BinaryString.fromString("old"), new BlobData(new byte[] {1, 1}))); + write( + table, + GenericRow.of(1, BinaryString.fromString("new"), new BlobData(new byte[] {2, 2}))); + compact(table, BinaryRow.EMPTY_ROW, 0, ioManager, true); + + Map expire = new HashMap<>(); + expire.put(CoreOptions.SNAPSHOT_NUM_RETAINED_MIN.key(), "1"); + expire.put(CoreOptions.SNAPSHOT_NUM_RETAINED_MAX.key(), "1"); + expire.put(CoreOptions.SNAPSHOT_EXPIRE_LIMIT.key(), "10"); + try (org.apache.paimon.table.sink.TableCommitImpl commit = + table.copy(expire).newCommit("")) { + commit.expireSnapshots(); + } + + Set live = livePackNames(table); + assertThat(live).isNotEmpty(); + Path orphan = new Path(bucketPath(table), "orphan.managed.blob"); + table.fileIO().newOutputStream(orphan, false).close(); + + List deleted = clean(table); + + assertThat(table.fileIO().exists(orphan)).isFalse(); + assertThat(deleted).extracting(Path::getName).contains("orphan.managed.blob"); + for (Path pack : managedBlobs(table)) { + assertThat(live).contains(pack.getName()); + } + assertThat(read(table)).hasSize(1); + assertThat(read(table).get(0).getBlob(2).toData()).containsExactly((byte) 2, (byte) 2); + } + + /** + * Compaction can commit after orphan GC has listed snapshots. Expire then deletes + * compact-before data files and blobrefs while those snapshots' manifests are still readable. A + * used-file scan of the stale list therefore neither skips nor retains the reused pack. + * Production GC collects used packs twice and aborts when the used set or snapshot topology + * changes; this test keeps the scan-only interleaving. A compaction prepared from inputs that + * are later removed cannot fill the remaining window after the second collection because + * conflict detection rejects its stale commit; see {@link + * #testStaleCompactionCannotCommitAfterFinalMark()}. + */ + @Test + public void testStaleSnapshotListMissesReusedPackAfterCompactBeforeDeleted() throws Exception { + FileStoreTable table = createManagedBlobTable("stale_list_compact"); + write( + table, + GenericRow.of(1, BinaryString.fromString("old"), new BlobData(new byte[] {3, 3}))); + write( + table, + GenericRow.of(1, BinaryString.fromString("new"), new BlobData(new byte[] {4, 4}))); + List listed = new ArrayList<>(table.snapshotManager().safelyGetAllSnapshots()); + assertThat(listed).isNotEmpty(); + + compact(table, BinaryRow.EMPTY_ROW, 0, ioManager, true); + Set liveAfterCompact = livePackNames(table); + assertThat(liveAfterCompact).isNotEmpty(); + + List compactBefore = + table.store() + .newSnapshotDeletion() + .planDeletedInDeltaManifest( + table.snapshotManager().latestSnapshot(), entry -> false); + assertThat(compactBefore).isNotEmpty(); + for (Path path : compactBefore) { + table.fileIO().deleteQuietly(path); + } + + StaleScan stale = collectUsedPacks(table, listed); + assertThat(stale.skip).isFalse(); + assertThat(stale.packs).doesNotContainAnyElementsOf(liveAfterCompact); + assertThat(read(table)).hasSize(1); + assertThat(read(table).get(0).getBlob(2).toData()).containsExactly((byte) 4, (byte) 4); + } + + @Test + public void testAbortWhenUsedSetChangesBetweenCollections() throws Exception { + FileStoreTable table = createManagedBlobTable("abort_used_change"); + write( + table, + GenericRow.of(1, BinaryString.fromString("old"), new BlobData(new byte[] {3, 3}))); + write( + table, + GenericRow.of(1, BinaryString.fromString("new"), new BlobData(new byte[] {4, 4}))); + Path orphan = new Path(bucketPath(table), "orphan.managed.blob"); + table.fileIO().newOutputStream(orphan, false).close(); + + List deleted = + new LocalManagedBlobOrphanFilesClean( + table, System.currentTimeMillis() + TimeUnit.SECONDS.toMillis(2), false) { + @Override + protected void betweenUsedCollections() { + try { + compact(table, BinaryRow.EMPTY_ROW, 0, ioManager, true); + List compactBefore = + table.store() + .newSnapshotDeletion() + .planDeletedInDeltaManifest( + table.snapshotManager().latestSnapshot(), + entry -> false); + for (Path path : compactBefore) { + table.fileIO().deleteQuietly(path); + } + } catch (Exception e) { + throw new RuntimeException(e); + } + } + }.clean().getDeletedFilesPath(); + + assertThat(deleted).isEmpty(); + assertThat(table.fileIO().exists(orphan)).isTrue(); + Set live = livePackNames(table); + assertThat(live).isNotEmpty(); + for (String name : live) { + assertThat(table.fileIO().exists(new Path(bucketPath(table), name))).isTrue(); + } + assertThat(read(table)).hasSize(1); + assertThat(read(table).get(0).getBlob(2).toData()).containsExactly((byte) 4, (byte) 4); + } + + @Test + public void testStaleCompactionCannotCommitAfterFinalMark() throws Exception { + FileStoreTable table = createManagedBlobTable("stale_compaction_commit"); + write( + table, + GenericRow.of( + 1, BinaryString.fromString("old-1"), new BlobData(new byte[] {1, 1}))); + write( + table, + GenericRow.of( + 2, BinaryString.fromString("old-2"), new BlobData(new byte[] {2, 2}))); + Set oldPacks = livePackNames(table); + assertThat(oldPacks).isNotEmpty(); + + BatchWriteBuilder staleBuilder = table.newBatchWriteBuilder(); + List staleMessages; + try (BatchTableWrite staleWrite = staleBuilder.newWrite()) { + staleWrite.withIOManager(ioManager); + staleWrite.compact(BinaryRow.EMPTY_ROW, 0, true); + staleMessages = staleWrite.prepareCommit(); + } + assertThat(staleMessages).isNotEmpty(); + Set reusedPacks = compactAfterPackNames(table, staleMessages); + assertThat(reusedPacks).isNotEmpty(); + assertThat(oldPacks).containsAll(reusedPacks); + + write( + table, + GenericRow.of(1, BinaryString.fromString("new-1"), new BlobData(new byte[] {3, 3})), + GenericRow.of( + 2, BinaryString.fromString("new-2"), new BlobData(new byte[] {4, 4}))); + compact(table, BinaryRow.EMPTY_ROW, 0, ioManager, true); + + Map expire = new HashMap<>(); + expire.put(CoreOptions.SNAPSHOT_NUM_RETAINED_MIN.key(), "1"); + expire.put(CoreOptions.SNAPSHOT_NUM_RETAINED_MAX.key(), "1"); + expire.put(CoreOptions.SNAPSHOT_EXPIRE_LIMIT.key(), "10"); + try (org.apache.paimon.table.sink.TableCommitImpl commit = + table.copy(expire).newCommit("")) { + commit.expireSnapshots(); + } + + Set currentPacks = livePackNames(table); + assertThat(currentPacks).isNotEmpty(); + assertThat(currentPacks).doesNotContainAnyElementsOf(oldPacks); + + AtomicBoolean commitAttempted = new AtomicBoolean(); + AtomicReference commitFailure = new AtomicReference<>(); + long snapshotIdBeforeClean = table.snapshotManager().latestSnapshotId(); + try (BatchTableCommit staleCommit = staleBuilder.newCommit()) { + List deleted = + new LocalManagedBlobOrphanFilesClean( + table, + System.currentTimeMillis() + TimeUnit.SECONDS.toMillis(2), + false) { + @Override + protected boolean cleanManagedBlobFile(Path path) { + if (commitAttempted.compareAndSet(false, true)) { + commitFailure.set( + catchThrowable(() -> staleCommit.commit(staleMessages))); + } + return super.cleanManagedBlobFile(path); + } + }.clean().getDeletedFilesPath(); + + assertThat(commitAttempted).isTrue(); + assertThat(commitFailure.get()) + .isNotNull() + .hasStackTraceContaining("File deletion conflicts detected"); + assertThat(deleted).extracting(Path::getName).containsAll(reusedPacks); + } + + assertThat(table.snapshotManager().latestSnapshotId()).isEqualTo(snapshotIdBeforeClean); + for (String pack : currentPacks) { + assertThat(table.fileIO().exists(new Path(bucketPath(table), pack))).isTrue(); + } + assertThat(read(table)) + .extracting(row -> row.getString(1).toString()) + .containsExactlyInAnyOrder("new-1", "new-2"); + } + + @Test + public void testSuccessfulCompactionAfterFinalMarkKeepsReusedPacks() throws Exception { + FileStoreTable table = createManagedBlobTable("successful_compaction_after_mark"); + write( + table, + GenericRow.of( + 1, BinaryString.fromString("value-1"), new BlobData(new byte[] {1, 1}))); + write( + table, + GenericRow.of( + 2, BinaryString.fromString("value-2"), new BlobData(new byte[] {2, 2}))); + Set liveBeforeCompact = livePackNames(table); + assertThat(liveBeforeCompact).isNotEmpty(); + + Path orphan = new Path(bucketPath(table), "orphan-after-final-mark.managed.blob"); + table.fileIO().newOutputStream(orphan, false).close(); + + AtomicBoolean compacted = new AtomicBoolean(); + long snapshotIdBeforeClean = table.snapshotManager().latestSnapshotId(); + List deleted = + new LocalManagedBlobOrphanFilesClean( + table, System.currentTimeMillis() + TimeUnit.SECONDS.toMillis(2), false) { + @Override + protected boolean cleanManagedBlobFile(Path path) { + if (compacted.compareAndSet(false, true)) { + try { + compact(table, BinaryRow.EMPTY_ROW, 0, ioManager, true); + } catch (Exception e) { + throw new RuntimeException(e); + } + } + return super.cleanManagedBlobFile(path); + } + }.clean().getDeletedFilesPath(); + + assertThat(compacted).isTrue(); + assertThat(table.snapshotManager().latestSnapshotId()).isGreaterThan(snapshotIdBeforeClean); + assertThat(deleted).extracting(Path::getName).containsExactly(orphan.getName()); + + Set liveAfterCompact = livePackNames(table); + assertThat(liveAfterCompact).isNotEmpty(); + assertThat(liveBeforeCompact).containsAll(liveAfterCompact); + for (String pack : liveAfterCompact) { + assertThat(table.fileIO().exists(new Path(bucketPath(table), pack))).isTrue(); + } + assertThat(read(table)) + .extracting(row -> row.getString(1).toString()) + .containsExactlyInAnyOrder("value-1", "value-2"); + } + + @Test + public void testJoinByFullPackPath() throws Exception { + FileStoreTable table = createManagedBlobTable("full_path_join"); + write( + table, + GenericRow.of(1, BinaryString.fromString("a"), new BlobData(new byte[] {1, 2}))); + List live = managedBlobs(table); + assertThat(live).isNotEmpty(); + String liveName = live.get(0).getName(); + Path otherBucket = new Path(bucketPath(table).getParent(), "bucket-1"); + Path other = new Path(otherBucket, liveName); + table.fileIO().mkdirs(otherBucket); + table.fileIO().newOutputStream(other, false).close(); + + List deleted = clean(table); + + assertThat(table.fileIO().exists(other)).isFalse(); + assertThat(deleted).extracting(Path::getName).contains(liveName); + for (Path pack : live) { + assertThat(table.fileIO().exists(pack)).isTrue(); + } + } + + private FileStoreTable createManagedBlobTable(String name) throws Exception { + Schema schema = + Schema.newBuilder() + .column("id", DataTypes.INT()) + .column("name", DataTypes.STRING()) + .column("payload", DataTypes.BLOB()) + .primaryKey("id") + .option(CoreOptions.BLOB_FIELD.key(), "payload") + .option(CoreOptions.CHANGELOG_PRODUCER.key(), "none") + .option(CoreOptions.BUCKET.key(), "1") + .build(); + catalog.createTable(identifier(name), schema, true); + return getTable(identifier(name)); + } + + private static List clean(FileStoreTable table) throws Exception { + return new LocalManagedBlobOrphanFilesClean( + table, System.currentTimeMillis() + TimeUnit.SECONDS.toMillis(2), false) + .clean() + .getDeletedFilesPath(); + } + + private static Path bucketPath(FileStoreTable table) { + return table.store().pathFactory().bucketPath(BinaryRow.EMPTY_ROW, 0); + } + + private static Set livePackNames(FileStoreTable table) throws IOException { + Set names = new HashSet<>(); + FileIO fileIO = table.fileIO(); + DataFilePathFactories factories = new DataFilePathFactories(table.store().pathFactory()); + for (ManifestEntry entry : table.store().newScan().plan().files()) { + DataFilePathFactory pathFactory = factories.get(entry.partition(), entry.bucket()); + DataFileMeta file = entry.file(); + for (String extra : file.extraFiles()) { + if (!extra.endsWith(ManagedBlobReferenceFile.REFERENCE_FILE_SUFFIX)) { + continue; + } + Path sidecar = pathFactory.toAlignedPath(extra, file); + for (ManagedBlobReferenceFile.Reference ref : + ManagedBlobReferenceFile.read(fileIO, sidecar)) { + names.add(ref.relativePath()); + } + } + } + return names; + } + + private static List managedBlobs(FileStoreTable table) throws IOException { + List packs = new ArrayList<>(); + FileStatus[] statuses = table.fileIO().listStatus(bucketPath(table)); + if (statuses == null) { + return packs; + } + for (FileStatus status : statuses) { + if (status.getPath().getName().endsWith(ManagedBlobReferenceFile.MANAGED_BLOB_SUFFIX)) { + packs.add(status.getPath()); + } + } + return packs; + } + + private static Set compactAfterPackNames( + FileStoreTable table, List messages) throws IOException { + Set packs = new HashSet<>(); + FileIO fileIO = table.fileIO(); + DataFilePathFactories factories = new DataFilePathFactories(table.store().pathFactory()); + for (CommitMessage message : messages) { + CommitMessageImpl messageImpl = (CommitMessageImpl) message; + DataFilePathFactory pathFactory = factories.get(message.partition(), message.bucket()); + for (DataFileMeta file : messageImpl.compactIncrement().compactAfter()) { + for (String extra : file.extraFiles()) { + if (!extra.endsWith(ManagedBlobReferenceFile.REFERENCE_FILE_SUFFIX)) { + continue; + } + Path sidecar = pathFactory.toAlignedPath(extra, file); + for (Reference reference : ManagedBlobReferenceFile.read(fileIO, sidecar)) { + packs.add(reference.relativePath()); + } + } + } + } + return packs; + } + + private static StaleScan collectUsedPacks(FileStoreTable table, Iterable snapshots) + throws IOException { + StaleScan scan = new StaleScan(); + ManifestFile manifestFile = table.store().manifestFileFactory().create(); + ManifestList manifestList = table.store().manifestListFactory().create(); + DataFilePathFactories factories = new DataFilePathFactories(table.store().pathFactory()); + ManagedBlobReachabilityCollector collector = + new ManagedBlobReachabilityCollector(table.fileIO()); + for (Snapshot snapshot : snapshots) { + List metas; + try { + metas = manifestList.readDataManifests(snapshot); + } catch (Exception e) { + scan.skip = true; + return scan; + } + for (ManifestFileMeta meta : metas) { + List entries; + try { + entries = manifestFile.read(meta.fileName()); + } catch (Exception e) { + scan.skip = true; + return scan; + } + for (ManifestEntry entry : entries) { + if (entry.kind() != FileKind.ADD) { + continue; + } + Result result = + collector.fromDataFile( + factories.get(entry.partition(), entry.bucket()).toPath(entry), + entry.file().extraFiles()); + if (result.isUnsafe()) { + scan.skip = true; + return scan; + } + for (Reference reference : result.referenced()) { + scan.packs.add(reference.relativePath()); + } + } + } + } + return scan; + } + + private static final class StaleScan { + private boolean skip; + private final Set packs = new HashSet<>(); + } + + private static void deleteSidecars(FileStoreTable table) throws IOException { + FileIO fileIO = table.fileIO(); + DataFilePathFactories factories = new DataFilePathFactories(table.store().pathFactory()); + for (ManifestEntry entry : table.store().newScan().plan().files()) { + DataFilePathFactory pathFactory = factories.get(entry.partition(), entry.bucket()); + DataFileMeta file = entry.file(); + for (String extra : file.extraFiles()) { + if (extra.endsWith(ManagedBlobReferenceFile.REFERENCE_FILE_SUFFIX)) { + fileIO.deleteQuietly(pathFactory.toAlignedPath(extra, file)); + } + } + } + } + + private static void assertReadOncePerPass(Map readCounts) { + assertThat(readCounts) + .anySatisfy( + (path, count) -> { + assertThat(path) + .endsWith(ManagedBlobReferenceFile.REFERENCE_FILE_SUFFIX); + assertThat(count).isEqualTo(1); + }); + assertThat(readCounts) + .anySatisfy( + (path, count) -> { + assertThat(new Path(path).getName()) + .startsWith("manifest-") + .doesNotStartWith("manifest-list-"); + assertThat(count).isEqualTo(1); + }); + assertThat(readCounts.values()).allMatch(count -> count == 1); + } + + private static class RelativeListingFileIO extends TraceableFileIO { + + private final AtomicInteger managedBlobDeleteAttempts = new AtomicInteger(); + + @Override + public FileStatus[] listStatus(Path path) throws IOException { + FileStatus[] statuses = super.listStatus(path); + if (statuses == null) { + return null; + } + FileStatus[] relative = new FileStatus[statuses.length]; + for (int i = 0; i < statuses.length; i++) { + FileStatus status = statuses[i]; + relative[i] = + status.getPath() + .getName() + .endsWith(ManagedBlobReferenceFile.MANAGED_BLOB_SUFFIX) + ? withPath(status, toRelativePath(status.getPath())) + : status; + } + return relative; + } + + @Override + public FileStatus getFileStatus(Path path) throws IOException { + FileStatus status = super.getFileStatus(path); + return path.toUri().getPath().startsWith("/") ? status : withPath(status, path); + } + + @Override + public boolean delete(Path path, boolean recursive) throws IOException { + if (path.getName().endsWith(ManagedBlobReferenceFile.MANAGED_BLOB_SUFFIX)) { + managedBlobDeleteAttempts.incrementAndGet(); + } + return super.delete(path, recursive); + } + + private int managedBlobDeleteAttempts() { + return managedBlobDeleteAttempts.get(); + } + + private static Path toRelativePath(Path path) { + java.nio.file.Path absolute = Paths.get(path.toUri().getPath()).toAbsolutePath(); + return new Path(Paths.get("").toAbsolutePath().relativize(absolute).toString()); + } + + private static FileStatus withPath(FileStatus status, Path path) { + return new FileStatus() { + @Override + public long getLen() { + return status.getLen(); + } + + @Override + public boolean isDir() { + return status.isDir(); + } + + @Override + public Path getPath() { + return path; + } + + @Override + public long getModificationTime() { + return status.getModificationTime(); + } + }; + } + } + + private static class QualifiedListingFileIO extends LocalFileIO { + + private final AtomicInteger managedBlobDeleteAttempts = new AtomicInteger(); + + @Override + public FileStatus[] listStatus(Path path) throws IOException { + FileStatus[] statuses = super.listStatus(path); + if (statuses == null) { + return null; + } + FileStatus[] qualified = new FileStatus[statuses.length]; + for (int i = 0; i < statuses.length; i++) { + FileStatus status = statuses[i]; + if (!status.getPath() + .getName() + .endsWith(ManagedBlobReferenceFile.MANAGED_BLOB_SUFFIX)) { + qualified[i] = status; + continue; + } + qualified[i] = + new FileStatus() { + @Override + public long getLen() { + return status.getLen(); + } + + @Override + public boolean isDir() { + return status.isDir(); + } + + @Override + public Path getPath() { + return new Path( + "hdfs://namenode:8020" + + status.getPath().toUri().getPath()); + } + + @Override + public long getModificationTime() { + return status.getModificationTime(); + } + }; + } + return qualified; + } + + @Override + public boolean delete(Path path, boolean recursive) throws IOException { + if (path.getName().endsWith(ManagedBlobReferenceFile.MANAGED_BLOB_SUFFIX)) { + managedBlobDeleteAttempts.incrementAndGet(); + return super.delete(new Path(path.toUri().getPath()), recursive); + } + return super.delete(path, recursive); + } + + private int managedBlobDeleteAttempts() { + return managedBlobDeleteAttempts.get(); + } + } + + private static class CountingInputFileIO extends LocalFileIO { + + private final Map readCounts = new ConcurrentHashMap<>(); + + @Override + public SeekableInputStream newInputStream(Path path) throws IOException { + String fileName = path.getName(); + if (fileName.endsWith(ManagedBlobReferenceFile.REFERENCE_FILE_SUFFIX) + || (fileName.startsWith("manifest-") + && !fileName.startsWith("manifest-list-"))) { + readCounts + .computeIfAbsent(path.toUri().getPath(), ignored -> new AtomicInteger()) + .incrementAndGet(); + } + return super.newInputStream(path); + } + + private Map readCounts() { + Map result = new HashMap<>(); + readCounts.forEach((path, count) -> result.put(path, count.get())); + return result; + } + + private void reset() { + readCounts.clear(); + } + } + + private interface SidecarOverwriter { + void write(DataOutputStream out) throws IOException; + } + + private static void overwriteSidecars(FileStoreTable table, SidecarOverwriter overwriter) + throws IOException { + FileIO fileIO = table.fileIO(); + DataFilePathFactories factories = new DataFilePathFactories(table.store().pathFactory()); + for (ManifestEntry entry : table.store().newScan().plan().files()) { + DataFilePathFactory pathFactory = factories.get(entry.partition(), entry.bucket()); + DataFileMeta file = entry.file(); + for (String extra : file.extraFiles()) { + if (!extra.endsWith(ManagedBlobReferenceFile.REFERENCE_FILE_SUFFIX)) { + continue; + } + Path sidecar = pathFactory.toAlignedPath(extra, file); + fileIO.deleteQuietly(sidecar); + try (DataOutputStream out = + new DataOutputStream(fileIO.newOutputStream(sidecar, false))) { + overwriter.write(out); + } + } + } + } +}