fix

Author: JkSelf

backends-velox/src/main/scala/org/apache/gluten/execution/VeloxBroadcastBuildSideCache.scala

@@ -28,7 +28,6 @@ import org.apache.spark.sql.execution.unsafe.UnsafeColumnarBuildSideRelation
 
 import com.github.benmanes.caffeine.cache.{Cache, Caffeine, RemovalCause, RemovalListener}
 
-import java.util.concurrent.ConcurrentHashMap
 import java.util.concurrent.TimeUnit
 
 case class BroadcastHashTable(pointer: Long, relation: BuildSideRelation)
@@ -48,9 +47,6 @@ object VeloxBroadcastBuildSideCache
     VeloxBackendSettings.GLUTEN_VELOX_BROADCAST_CACHE_EXPIRED_TIME_DEFAULT
   )
 
-  // Track released pointers to prevent double free
-  private val releasedPointers = ConcurrentHashMap.newKeySet[Long]()
-
   // Use for controlling to build bhj hash table once.
   // key: hashtable id, value is hashtable backend pointer(long to string).
   private val buildSideRelationCache: Cache[String, BroadcastHashTable] =
@@ -98,9 +94,9 @@ object VeloxBroadcastBuildSideCache
   def cleanAll(): Unit = buildSideRelationCache.invalidateAll()
 
   override def onRemoval(key: String, value: BroadcastHashTable, cause: RemovalCause): Unit = {
-    // Use ConcurrentHashMap.add() which returns false if already present
-    // This ensures only one thread can successfully mark the pointer as released
-    if (releasedPointers.add(value.pointer)) {
+    // Synchronize on the value object to ensure only one thread can release this specific hash table
+    // This prevents concurrent calls to clearHashTable which can corrupt memory pool state
+    value.synchronized {
       logWarning(s"Remove bhj $key = ${value.pointer}")
       if (value.relation != null) {
         value.relation match {
@@ -111,9 +107,9 @@ object VeloxBroadcastBuildSideCache
         }
       }
 
+      // clearHashTable must be called under synchronization to prevent
+      // concurrent access to the same hash table's memory pool
       HashJoinBuilder.clearHashTable(value.pointer)
-    } else {
-      logWarning(s"Skip already released bhj $key = ${value.pointer}")
     }
   }
 }