JingsongLi commented on code in PR #7715:
URL: https://github.com/apache/paimon/pull/7715#discussion_r3888769060
##########
paimon-core/src/main/java/org/apache/paimon/operation/LocalOrphanFilesClean.java:
##########
@@ -157,35 +173,59 @@ private void cleanEmptyDataDirectory(List<Path>
deleteFiles) {
}
private void collectWithoutDataFile(
- String branch, Consumer<String> usedFileConsumer, Consumer<String>
manifestConsumer)
+ String branch,
+ Consumer<String> usedFileConsumer,
+ Consumer<String> manifestConsumer,
+ Consumer<String> liveManifestConsumer,
+ AtomicBoolean missingManifest)
throws IOException {
+ Set<Snapshot> liveSnapshots =
Review Comment:
[P1] Reuse one active-snapshot view instead of rescanning it
This captures `liveSnapshots` here, but line 207 independently calls
`safelyGetAllSnapshots(branch)` again. That helper fixes the path list first
and silently skips a snapshot whose file is deleted before it is read. A
concurrent S1 -> S2 commit plus retention-1 expiration can therefore make the
second scan drop S1 without seeing S2 or setting `missingManifest`. If another
branch contributes any used files, the global `usedFiles.isEmpty()` guard is
bypassed and old files still referenced by S2 can be deleted. Please process
the captured live snapshot objects (adding tags/changelogs separately), or
propagate a per-branch incomplete-enumeration signal and abort globally. A
deterministic two-branch race test would cover this gap.
##########
paimon-core/src/main/java/org/apache/paimon/operation/LocalOrphanFilesClean.java:
##########
@@ -108,12 +110,26 @@ public CleanOrphanFilesResult clean()
}
candidateDeletes = new HashSet<>(candidates.keySet());
+ AtomicBoolean missingManifest = new AtomicBoolean(false);
+
// find used files
Set<String> usedFiles =
branches.stream()
- .flatMap(branch -> getUsedFiles(branch).stream())
+ .flatMap(branch -> getUsedFiles(branch,
missingManifest).stream())
.collect(Collectors.toSet());
+ if (usedFiles.isEmpty()) {
Review Comment:
[P2] Preserve cleanup for a valid zero-snapshot table
An empty `usedFiles` set is also the successful result for a never-committed
table. If an aborted first write leaves old bucket or manifest files, local
cleanup now returns here forever, while Flink and Spark still clean against
zero references. Please base the abort on an explicit per-branch incomplete or
missing signal instead of the empty set itself, so legitimate zero-reference
cleanup still works.
##########
paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/procedure/SparkOrphanFilesClean.scala:
##########
@@ -102,18 +136,41 @@ case class SparkOrphanFilesClean(
(key: String) =>
specifiedTable.switchToBranch(key).store.manifestFileFactory.create)
- retryReadingFiles(
+ val manifestMissing =
+ if (branchAndManifestFile.isLiveSnapshot) new
AtomicBoolean(false) else null
+ val entries = retryReadingFiles(
() =>
manifestFile.readWithIOException(branchAndManifestFile.manifestName),
- Collections.emptyList[ManifestEntry]
- ).asScala.flatMap {
- manifestEntry =>
- manifestEntry.fileName() +:
manifestEntry.file().extraFiles().asScala
+ Collections.emptyList[ManifestEntry],
+ manifestMissing
+ ).asScala
+ if (manifestMissing != null && manifestMissing.get()) {
+ Iterator.single(("", true))
+ } else {
+ entries
+ .flatMap {
+ manifestEntry =>
+ manifestEntry.fileName() +:
manifestEntry.file().extraFiles().asScala
+ }
+ .map(name => (name, false))
+ .iterator
}
}
}
+ .cache()
Review Comment:
[P2] Release this cached dataset
`dataFilesWithFlag` is materialized by the `isEmpty` check and later reused
by deletion planning, but `doOrphanClean` returns only `usedManifestFiles`;
`executeDatabaseOrphanFiles` therefore unpersists only that dataset in its
`finally` block. Normal runs and the early return at line 163 leave this cache
registered in the Spark session, so repeated or database-wide cleanup can
retain large executor memory/local-disk blocks. Please return and release all
cached datasets, including every early-return path.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
