[VL] Disable useHashTableCache if the hash table is not pre-built during HashJoinNode creation#12487
Merged
Merged
Conversation
Contributor
There was a problem hiding this comment.
Pull request overview
This PR targets an unstable Velox memory leak at query end by triggering cleanup of broadcast hash-join hash tables when Spark fires SQLExecutionEnd, and by aligning logging to the new cleanup path.
Changes:
- On
SparkListenerSQLExecutionEnd, traverse the executed plan to collectBroadcastHashJoinExecTransformerbuild-table IDs and initiate cleanup. - Update the warning message in
BroadcastHashJoinExecTransformerfor the “execution id is null” scenario to referenceSQLExecutionEndcleanup.
Reviewed changes
Copilot reviewed 2 out of 2 changed files in this pull request and generated 3 comments.
| File | Description |
|---|---|
| backends-velox/src/main/scala/org/apache/spark/listener/VeloxGlutenSQLAppStatusListener.scala | Adds SQLExecutionEnd-based discovery of BHJ broadcast table IDs and triggers per-executor cleanup. |
| backends-velox/src/main/scala/org/apache/gluten/execution/HashJoinExecTransformer.scala | Updates the warning message to reflect the new cleanup trigger at SQL execution end. |
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
Comment on lines
+85
to
+90
| val idSet = new util.HashSet[String]() | ||
| bhjIds.foreach(idSet.add) | ||
| GlutenDriverEndpoint.executorDataMap.forEach( | ||
| (_, executor) => executor.executorEndpointRef.send(GlutenCleanExecutionResource(executionId, idSet))) | ||
| } | ||
| } |
Comment on lines
+78
to
+82
| val qe = event.qe | ||
| if (qe != null) { | ||
| val bhjIds = qe.executedPlan.collect { case bhj: BroadcastHashJoinExecTransformer => | ||
| bhj.buildBroadcastTableId | ||
| }.toSet |
Comment on lines
+142
to
+145
| logWarning( | ||
| s"Cannot trace broadcast table data $buildBroadcastTableId" + | ||
| s" because execution id is null." + | ||
| s" Will clean up until expire time.") | ||
| s" Will clean up on SQLExecutionEnd.") |
Contributor
There was a problem hiding this comment.
Pull request overview
Copilot reviewed 3 out of 3 changed files in this pull request and generated 3 comments.
Comments suppressed due to low confidence (2)
backends-velox/src/main/scala/org/apache/gluten/execution/VeloxBroadcastBuildSideCache.scala:98
registerLocalCleanupIfNeededis invoked afterbuildSideRelationCache.get(...)returns. That leaves a window where another task can decrement the refcount to zero and invalidate the cache entry while this task already obtained theBroadcastHashTablebut hasn’t registered its TaskResource yet (potential use-after-free of the native hash table). Registering the task-local cleanup should happen before exposing the hash table to the caller.
val hashTable = buildSideRelationCache
.get(
broadcastContext.buildHashTableId,
(_: String) => {
val (pointer, relation, droppedDuplicates) = broadcast.value match {
backends-velox/src/main/scala/org/apache/gluten/execution/VeloxBroadcastBuildSideCache.scala:209
deserializeOnExecutorhas the same race asgetOrBuildBroadcastHashTable: task-local cleanup is registered after the cache entry is returned. Another task can invalidate the broadcast hash table before this task registers its TaskResource, which can lead to clearing the native hash table while it is still in use. Register the cleanup before callingbuildSideRelationCache.get(...)so the refcount is bumped before the table is observable.
val hashTable = buildSideRelationCache.get(
broadcastHashTableId,
(_: String) => {
logInfo(s"Deserializing hash table on executor for broadcast ID: $broadcastHashTableId")
val startTime = System.currentTimeMillis()
Comment on lines
+266
to
+290
| private def registerLocalCleanupIfNeeded( | ||
| broadcastHashTableId: String, | ||
| enableLocalHashTableCleanup: Boolean): Unit = { | ||
| if (!enableLocalHashTableCleanup || !TaskResources.inSparkTask()) { | ||
| return | ||
| } | ||
|
|
||
| val resourceId = s"broadcast-hash-table-local-cleanup:$broadcastHashTableId" | ||
| TaskResources.addResourceIfNotRegistered( | ||
| resourceId, | ||
| () => { | ||
| localReferenceCounts | ||
| .computeIfAbsent(broadcastHashTableId, _ => new AtomicInteger(0)) | ||
| .incrementAndGet() | ||
|
|
||
| new TaskResource { | ||
| override def release(): Unit = releaseLocalReference(broadcastHashTableId) | ||
|
|
||
| override def priority(): Int = 200 | ||
|
|
||
| override def resourceName(): String = resourceId | ||
| } | ||
| } | ||
| ) | ||
| } |
Comment on lines
+292
to
+303
| private def releaseLocalReference(broadcastHashTableId: String): Unit = { | ||
| val refCount = localReferenceCounts.get(broadcastHashTableId) | ||
| if (refCount == null) { | ||
| return | ||
| } | ||
|
|
||
| val remaining = refCount.decrementAndGet() | ||
| if (remaining <= 0 && localReferenceCounts.remove(broadcastHashTableId, refCount)) { | ||
| logInfo(s"Cleaning up broadcast hash table with local fallback: $broadcastHashTableId") | ||
| invalidateBroadcastHashtable(broadcastHashTableId) | ||
| } | ||
| } |
Comment on lines
+266
to
+269
| private def registerLocalCleanupIfNeeded( | ||
| broadcastHashTableId: String, | ||
| enableLocalHashTableCleanup: Boolean): Unit = { | ||
| if (!enableLocalHashTableCleanup || !TaskResources.inSparkTask()) { |
Comment on lines
461
to
+491
| @@ -469,11 +484,11 @@ core::PlanNodePtr SubstraitToVeloxPlanConverter::toVeloxPlan(const ::substrait:: | |||
| leftNode, | |||
| rightNode, | |||
| getJoinOutputType(leftNode, rightNode, joinType), | |||
| true, | |||
| useHashTableCache, | |||
| false, | |||
| false, | |||
| nullptr, | |||
| sJoin.hashtableid()); | |||
| hashTableId); | |||
Comment on lines
461
to
+491
| @@ -469,11 +484,11 @@ core::PlanNodePtr SubstraitToVeloxPlanConverter::toVeloxPlan(const ::substrait:: | |||
| leftNode, | |||
| rightNode, | |||
| getJoinOutputType(leftNode, rightNode, joinType), | |||
| true, | |||
| useHashTableCache, | |||
| false, | |||
| false, | |||
| nullptr, | |||
| sJoin.hashtableid()); | |||
| hashTableId); | |||
Comment on lines
+464
to
+468
| try { | ||
| const auto handle = getJoin(hashTableId); | ||
| if (handle != 0) { | ||
| auto hashTableBuilder = ObjectStore::retrieve<gluten::HashTableBuilder>(handle); | ||
| useHashTableCache = (hashTableBuilder != nullptr); |
Comment on lines
+461
to
+465
| const std::string hashTableId = sJoin.hashtableid(); | ||
| bool useHashTableCache = false; | ||
| if (!hashTableId.empty()) { | ||
| try { | ||
| const auto handle = getJoin(hashTableId); |
Comment on lines
+461
to
+474
| const std::string hashTableId = sJoin.hashtableid(); | ||
| bool useHashTableCache = false; | ||
| if (!hashTableId.empty()) { | ||
| try { | ||
| const auto handle = getJoin(hashTableId); | ||
| if (handle != 0) { | ||
| auto hashTableBuilder = ObjectStore::retrieve<gluten::HashTableBuilder>(handle); | ||
| useHashTableCache = (hashTableBuilder != nullptr); | ||
| } | ||
| } catch (const std::exception& e) { | ||
| LOG(WARNING) << "Failed to retrieve pre-built HashTableBuilder for cache key: " << hashTableId | ||
| << ", error: " << e.what() << ". Disable hash table cache and build a new table."; | ||
| } | ||
| } |
zhztheplayer
approved these changes
Jul 10, 2026
JkSelf
changed the title
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
What changes are proposed in this pull request?
Because Velox does not release the hash tables in
HashTableCacheupon query completion (they are only released when the executor terminates), the CI encounters the following flaky error:This PR adds a check to determine whether the hash table is pre-built. If it is not,
useHashTableCachewill be disabled automatically.How was this patch tested?
Existing tests
Was this patch authored or co-authored using generative AI tooling?
No