hvanhovell commented on code in PR #53852:
URL: https://github.com/apache/spark/pull/53852#discussion_r2708152968
##########
sql/core/src/main/scala/org/apache/spark/sql/artifact/ArtifactManager.scala:
##########
@@ -183,25 +183,34 @@ class ArtifactManager(session: SparkSession) extends
AutoCloseable with Logging
if (normalizedRemoteRelativePath.startsWith(s"cache${File.separator}")) {
val tmpFile = serverLocalStagingPath.toFile
Utils.tryWithSafeFinallyAndFailureCallbacks {
+ val hash =
normalizedRemoteRelativePath.toString.stripPrefix(s"cache${File.separator}")
val blockManager = session.sparkContext.env.blockManager
val blockId = CacheId(
sessionUUID = session.sessionUUID,
- hash =
normalizedRemoteRelativePath.toString.stripPrefix(s"cache${File.separator}"))
- val updater = blockManager.TempFileBasedBlockStoreUpdater(
- blockId = blockId,
- level = StorageLevel.MEMORY_AND_DISK_SER,
- classTag = implicitly[ClassTag[Array[Byte]]],
- tmpFile = tmpFile,
- blockSize = tmpFile.length(),
- tellMaster = false)
- updater.save()
- val oldBlock = hashToCachedIdMap.put(blockId.hash, new
RefCountedCacheId(blockId))
- if (oldBlock != null) {
- logWarning(
- log"Replacing existing cache artifact with hash
${MDC(LogKeys.BLOCK_ID, blockId)} " +
- log"in session ${MDC(LogKeys.SESSION_ID, session.sessionUUID)}.
" +
- log"This may indicate duplicate artifact addition.")
- oldBlock.release(blockManager)
+ hash = hash)
+ // If the exact same block (same CacheId) already exists, skip
re-adding.
+ // This prevents incorrectly removing the existing block from
BlockManager.
+ // Note: We only skip if the CacheId matches - if it's a different
session's block
+ // (e.g., after clone), we should replace it.
+ val existingBlock = hashToCachedIdMap.get(hash)
+ if (existingBlock == null || existingBlock.id != blockId) {
+ val updater = blockManager.TempFileBasedBlockStoreUpdater(
+ blockId = blockId,
+ level = StorageLevel.MEMORY_AND_DISK_SER,
+ classTag = implicitly[ClassTag[Array[Byte]]],
+ tmpFile = tmpFile,
+ blockSize = tmpFile.length(),
+ tellMaster = false)
+ updater.save()
+ val oldBlock = hashToCachedIdMap.put(blockId.hash, new
RefCountedCacheId(blockId))
+ if (oldBlock != null) {
Review Comment:
`oldBlock` is the same as `existingBlock` right? It might be better to use
`existingBlock`...
--
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]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
