fix

Author: JkSelf

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

@@ -48,9 +48,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 +95,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 +108,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}")
     }
   }
 }

backends-velox/src/main/scala/org/apache/spark/sql/execution/ColumnarBuildSideRelation.scala

@@ -204,21 +204,25 @@ case class ColumnarBuildSideRelation(
         }.toArray
 
         // Build the hash table
-        hashTableData = HashJoinBuilder
-          .nativeBuild(
-            broadcastContext.buildHashTableId,
-            batchArray.toArray,
-            joinKeys,
-            broadcastContext.substraitJoinType.ordinal(),
-            broadcastContext.hasMixedFiltCondition,
-            broadcastContext.isExistenceJoin,
-            SubstraitUtil.toNameStruct(newOutput).toByteArray,
-            broadcastContext.isNullAwareAntiJoin,
-            broadcastContext.bloomFilterPushdownSize,
-            broadcastContext.broadcastHashTableBuildThreads
-          )
-
-        jniWrapper.close(serializeHandle)
+        try {
+          hashTableData = HashJoinBuilder
+            .nativeBuild(
+              broadcastContext.buildHashTableId,
+              batchArray.toArray,
+              joinKeys,
+              broadcastContext.substraitJoinType.ordinal(),
+              broadcastContext.hasMixedFiltCondition,
+              broadcastContext.isExistenceJoin,
+              SubstraitUtil.toNameStruct(newOutput).toByteArray,
+              broadcastContext.isNullAwareAntiJoin,
+              broadcastContext.bloomFilterPushdownSize,
+              broadcastContext.broadcastHashTableBuildThreads
+            )
+        } finally {
+          // Release batch handles after building hash table
+          batchArray.foreach(ColumnarBatches.release)
+          jniWrapper.close(serializeHandle)
+        }
         (hashTableData, this)
       } else {
         (HashJoinBuilder.cloneHashTable(hashTableData), null)

backends-velox/src/main/scala/org/apache/spark/sql/execution/unsafe/UnsafeColumnarBuildSideRelation.scala

@@ -174,21 +174,25 @@ class UnsafeColumnarBuildSideRelation(
         }.toArray
 
         // Build the hash table
-        hashTableData = HashJoinBuilder
-          .nativeBuild(
-            broadcastContext.buildHashTableId,
-            batchArray.toArray,
-            joinKeys,
-            broadcastContext.substraitJoinType.ordinal(),
-            broadcastContext.hasMixedFiltCondition,
-            broadcastContext.isExistenceJoin,
-            SubstraitUtil.toNameStruct(newOutput).toByteArray,
-            broadcastContext.isNullAwareAntiJoin,
-            broadcastContext.bloomFilterPushdownSize,
-            broadcastContext.broadcastHashTableBuildThreads
-          )
-
-        jniWrapper.close(serializeHandle)
+        try {
+          hashTableData = HashJoinBuilder
+            .nativeBuild(
+              broadcastContext.buildHashTableId,
+              batchArray.toArray,
+              joinKeys,
+              broadcastContext.substraitJoinType.ordinal(),
+              broadcastContext.hasMixedFiltCondition,
+              broadcastContext.isExistenceJoin,
+              SubstraitUtil.toNameStruct(newOutput).toByteArray,
+              broadcastContext.isNullAwareAntiJoin,
+              broadcastContext.bloomFilterPushdownSize,
+              broadcastContext.broadcastHashTableBuildThreads
+            )
+        } finally {
+          // Release batch handles after building hash table
+          batchArray.foreach(ColumnarBatches.release)
+          jniWrapper.close(serializeHandle)
+        }
         (hashTableData, this)
       } else {
         (HashJoinBuilder.cloneHashTable(hashTableData), null)