JingsongLi commented on code in PR #7715:
URL: https://github.com/apache/paimon/pull/7715#discussion_r4291189295
##########
paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/procedure/SparkOrphanFilesClean.scala:
##########
@@ -69,29 +82,57 @@ case class SparkOrphanFilesClean(
val usedManifestFiles = spark.sparkContext
.parallelize(branches.asScala.toSeq, maxBranchParallelism)
.mapPartitions(_.flatMap {
- branch => safelyGetAllSnapshots(branch).asScala.map(snapshot =>
(branch, snapshot.toJson))
+ branch =>
+ val liveSnapshots = safelyGetLiveSnapshots(branch)
+ snapshotsIncludingTagsAndChangelogs(branch,
liveSnapshots).asScala.map(
+ snapshot => (branch, snapshot.toJson,
liveSnapshots.contains(snapshot)))
})
.repartition(parallelism)
.flatMap {
- case (branch, snapshotJson) =>
+ case (branch, snapshotJson, isLive) =>
val usedFileBuffer = new ArrayBuffer[BranchAndManifestFile]()
val usedFileConsumer =
new Consumer[org.apache.paimon.utils.Pair[String,
java.lang.Boolean]] {
override def accept(pair: utils.Pair[String,
java.lang.Boolean]): Unit = {
- usedFileBuffer.append(BranchAndManifestFile(branch,
pair.getLeft, pair.getRight))
+ usedFileBuffer.append(
+ BranchAndManifestFile(
+ branch,
+ pair.getLeft,
+ pair.getRight,
+ isMissing = false,
+ isLiveSnapshot = isLive))
}
}
+ val missingManifest = if (isLive) new AtomicBoolean(false) else null
val snapshot = Snapshot.fromJson(snapshotJson)
- collectWithoutDataFileWithManifestFlag(branch, snapshot,
usedFileConsumer)
+ collectWithoutDataFileWithManifestFlag(
+ branch,
+ snapshot,
+ usedFileConsumer,
+ missingManifest)
+ if (missingManifest != null && missingManifest.get()) {
+ usedFileBuffer.append(
+ BranchAndManifestFile(branch, "", isManifestFile = false,
isMissing = true))
+ }
usedFileBuffer
}
.toDS()
.cache()
+ cachedDatasets += usedManifestFiles
+
+ if (!usedManifestFiles.filter(_.isMissing).isEmpty) {
+ logWarning("Detected missing manifest during used-files collection,
aborting clean.")
+ return (
+ spark.createDataset(
+ Seq((deletedFilesCountInLocal.get(),
deletedFilesLenInBytesInLocal.get()))),
+ cachedDatasets.toSeq)
+ }
- // find all data files
- val dataFiles = usedManifestFiles
- .filter(_.isManifestFile)
- .distinct()
+ val dataFilesWithFlag = usedManifestFiles
+ .filter(f => f.isManifestFile && !f.isMissing)
Review Comment:
[P1] Check missing metadata before filtering the data-reference branch
This filter still discards the missing-manifest marker when only the
data-reference branch is recomputed. The two branches of
`usedFiles.union(dataFiles)` can reuse different shuffle outputs: if the
metadata branch has already produced its output, loss of the reference caches
and the manifest-grouping shuffle can rerun only the data branch from the
surviving captured-snapshot shuffle. After concurrent commit/expiration removes
the old manifest list, its `isMissing` record is filtered out here. The
metadata branch reuses its completed output, so its later missing-marker check
does not run again. The anti-join can then delete data files still referenced
by the current snapshot.
I reproduced this on `09a86e021779569d7ea79bd50f58729a26fec170` with Spark
3.5.8, real commits/expiration, and selective cache/shuffle-loss injection.
After preserving the completed metadata-branch output and invalidating only the
manifest-grouping shuffle map output plus the two reference caches, Spark
retried the grouping stage following `MetadataFetchFailedException`.
`usedManifestFiles` contained missing markers, but `dataFilesWithFlag` was
empty; cleanup returned `(2, 594)`, deleted both old live data files, and a
subsequent table read failed with `FileNotFoundException`. A control that
throws on `isMissing` before this filter aborted cleanup and preserved all
three rows.
Please check `isMissing` and fail before filtering out non-manifest records,
or propagate the abort signal through this branch. Add a regression test that
retains the completed metadata-branch shuffle output while recomputing the
data-reference branch; the current test evicts all caches before execution and
therefore reruns the other branch's guard.
--
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]
