You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@gluten.apache.org by "JkSelf (via GitHub)" <gi...@apache.org> on 2026/07/10 03:53:09 UTC
[PR] [VL] Release all the hash table when SQLExecutionEnd end [gluten]
JkSelf opened a new pull request, #12487:
URL: https://github.com/apache/gluten/pull/12487
<!--
Thank you for submitting a pull request! Here are some tips:
1. For first-time contributors, please read our contributing guide:
https://github.com/apache/gluten/blob/main/CONTRIBUTING.md
2. If necessary, create a GitHub issue for discussion beforehand to avoid duplicate work.
3. If the PR is specific to a single backend, include [VL] or [CH] in the PR title to indicate the
Velox or ClickHouse backend, respectively.
4. If the PR is not ready for review, please mark it as a draft.
-->
## What changes are proposed in this pull request?
<!--
Provide a clear and concise description of the changes introduced in this PR.
Ensure the PR description aligns with the code changes, especially after updates.
If applicable, include "Fixes #<GitHub_Issue_ID>" to automatically close the corresponding issue
when the PR is merged.
-->
Fix the following unstable error:
```
2026-07-09T05:01:36.4980422Z E20260709 05:01:36.497267 1021214 VeloxMemoryManager.cc:446] Failed to release Velox memory manager as there are still outstanding memory resources.
2026-07-09T05:01:36.4983663Z E20260709 05:01:36.497301 1021214 MemoryPool.cpp:481] [MEM] Memory leak (Used memory): Memory Pool[cached_table_15560717 LEAF root[root] parent[root] MALLOC track-usage thread-safe]<unlimited max capacity capacity 8.00MB used 68.00KB available 956.00KB reservation [used 68.00KB, reserved 1.00MB, min 0B] counters [allocs 3, frees 1, reserves 0, releases 0, collisions 0, external-allocs 0, external-frees 0, cumulative-external 0B])>
2026-07-09T05:01:36.4987303Z E20260709 05:01:36.497359 1021214 Exceptions.h:87] Line: /work/cpp/velox/memory/VeloxMemoryManager.cc:108, Function:removePool, Expression: pool->reservedBytes() == 0 (1048576 vs. 0), Source: RUNTIME, ErrorCode: INVALID_STATE
2026-07-09T05:01:36.4989039Z terminate called after throwing an instance of 'facebook::velox::VeloxRuntimeError'
2026-07-09T05:01:36.4992594Z what(): Exception: VeloxRuntimeError
2026-07-09T05:01:36.4993041Z Error Source: RUNTIME
2026-07-09T05:01:36.4993272Z Error Code: INVALID_STATE
2026-07-09T05:01:36.4993498Z Reason: (1048576 vs. 0)
2026-07-09T05:01:36.4993711Z Retriable: False
2026-07-09T05:01:36.4994093Z Expression: pool->reservedBytes() == 0
2026-07-09T05:01:36.4994432Z Function: removePool
2026-07-09T05:01:36.4994702Z File: /work/cpp/velox/memory/VeloxMemoryManager.cc
2026-07-09T05:01:36.4995003Z Line: 108
2026-07-09T05:01:36.4995183Z Stack trace:
2026-07-09T05:01:36.4995357Z # 0
2026-07-09T05:01:36.4995516Z # 1
2026-07-09T05:01:36.4995671Z # 2
2026-07-09T05:01:36.4995823Z # 3
2026-07-09T05:01:36.4995978Z # 4
2026-07-09T05:01:36.4996133Z # 5
2026-07-09T05:01:36.4996290Z # 6
2026-07-09T05:01:36.4996446Z # 7
2026-07-09T05:01:36.4996597Z # 8
2026-07-09T05:01:36.4996749Z # 9
2026-07-09T05:01:36.4996901Z # 10
2026-07-09T05:01:36.4997059Z # 11
2026-07-09T05:01:36.4997218Z # 12
```
## How was this patch tested?
Existing tests
<!--
Describe how the changes were tested, if applicable.
Include new tests to validate the functionality, if necessary.
For UI-related changes, attach screenshots to demonstrate the updates.
-->
## Was this patch authored or co-authored using generative AI tooling?
No
<!--
If generative AI tooling has been used in the process of authoring this patch, please include the
phrase: 'Generated-by: ' followed by the name of the tool and its version.
If no, write 'No'.
Please refer to the [ASF Generative Tooling Guidance](https://www.apache.org/legal/generative-tooling.html) for details.
-->
--
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: commits-unsubscribe@gluten.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For additional commands, e-mail: commits-help@gluten.apache.org
Re: [PR] [VL] Release all the hash table when SQLExecutionEnd end [gluten]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #12487:
URL: https://github.com/apache/gluten/pull/12487#discussion_r3556377527
##########
backends-velox/src/main/scala/org/apache/gluten/execution/VeloxBroadcastBuildSideCache.scala:
##########
@@ -249,4 +262,43 @@ object VeloxBroadcastBuildSideCache
HashJoinBuilder.clearHashTable(key, value.pointer)
}
}
+
+ 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
+ }
+ }
+ )
+ }
Review Comment:
`localReferenceCounts` is updated concurrently from multiple tasks, but the increment path isn’t synchronized with `releaseLocalReference`. Because `releaseLocalReference` uses `remove(key, refCount)` on the same `AtomicInteger` instance, a concurrent increment can still be lost and the entry can be invalidated even though another task has already increased the count. Synchronize the refcount increment with the decrement/invalidate path.
##########
backends-velox/src/main/scala/org/apache/gluten/execution/VeloxBroadcastBuildSideCache.scala:
##########
@@ -249,4 +262,43 @@ object VeloxBroadcastBuildSideCache
HashJoinBuilder.clearHashTable(key, value.pointer)
}
}
+
+ 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
+ }
+ }
+ )
+ }
+
+ 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)
+ }
+ }
Review Comment:
`releaseLocalReference` performs decrement / conditional map removal / cache invalidation without synchronization. With concurrent tasks, this can invalidate the broadcast hash table even if another task is in the middle of registering/incrementing the same `AtomicInteger` (since `remove(key, refCount)` only checks object identity). Wrap the whole decrement + invalidate decision in the same lock used by the increment path so the refcount logic is consistent.
##########
backends-velox/src/main/scala/org/apache/gluten/execution/VeloxBroadcastBuildSideCache.scala:
##########
@@ -249,4 +262,43 @@ object VeloxBroadcastBuildSideCache
HashJoinBuilder.clearHashTable(key, value.pointer)
}
}
+
+ private def registerLocalCleanupIfNeeded(
+ broadcastHashTableId: String,
+ enableLocalHashTableCleanup: Boolean): Unit = {
+ if (!enableLocalHashTableCleanup || !TaskResources.inSparkTask()) {
Review Comment:
The new task-local cleanup path (via `enableLocalHashTableCleanup`, TaskResources registration, and refcounted invalidation) changes lifecycle behavior for broadcast hash tables but isn’t covered by existing tests in the Velox backend. Adding a regression test that exercises the `executionId == null` path and asserts the cache is cleared at task completion would help prevent reintroducing the memory-leak / outstanding-pool failure this PR targets.
--
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: commits-unsubscribe@gluten.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For additional commands, e-mail: commits-help@gluten.apache.org
Re: [PR] [VL] Disable useHashTableCache if the hash table is not pre-built during HashJoinNode creation [gluten]
Posted by "JkSelf (via GitHub)" <gi...@apache.org>.
JkSelf merged PR #12487:
URL: https://github.com/apache/gluten/pull/12487
--
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: commits-unsubscribe@gluten.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For additional commands, e-mail: commits-help@gluten.apache.org
Re: [PR] [VL] Release all the hash table when SQLExecutionEnd end [gluten]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #12487:
URL: https://github.com/apache/gluten/pull/12487#discussion_r3557246020
##########
cpp/velox/substrait/SubstraitToVeloxPlan.cc:
##########
@@ -458,6 +458,21 @@ core::PlanNodePtr SubstraitToVeloxPlanConverter::toVeloxPlan(const ::substrait::
} else if (
sJoin.has_advanced_extension() &&
SubstraitParser::configSetInOptimization(sJoin.advanced_extension(), "isBHJ=")) {
+ 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);
Review Comment:
This new logic introduces a JNI call via getJoin(hashTableId). JniHashTableContext::callJavaGet currently creates a jstring with env->NewStringUTF but never deletes the local reference, and it also doesn’t check/clear pending Java exceptions after CallStaticLongMethod. With this new call site, repeated plan conversions in a single JNI frame can accumulate local refs or silently proceed with a pending exception; consider fixing callJavaGet to DeleteLocalRef(s) (or use a scoped local ref helper) and handle JNI exceptions explicitly.
##########
cpp/velox/substrait/SubstraitToVeloxPlan.cc:
##########
@@ -458,6 +458,21 @@ core::PlanNodePtr SubstraitToVeloxPlanConverter::toVeloxPlan(const ::substrait::
} else if (
sJoin.has_advanced_extension() &&
SubstraitParser::configSetInOptimization(sJoin.advanced_extension(), "isBHJ=")) {
+ const std::string hashTableId = sJoin.hashtableid();
+ bool useHashTableCache = false;
+ if (!hashTableId.empty()) {
+ try {
+ const auto handle = getJoin(hashTableId);
Review Comment:
The new cache-selection behavior (useHashTableCache depends on looking up the hash table in VeloxBroadcastBuildSideCache/ObjectStore) appears to be the core of the leak fix, but there’s no regression test coverage ensuring the plan conversion correctly toggles hash-table caching when the cache key is missing/evicted or when the handle is stale. Adding a targeted test (e.g., in existing Velox hash join suites) for the cache-miss path would help prevent future regressions of the SQLExecutionEnd memory-leak scenario.
--
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: commits-unsubscribe@gluten.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For additional commands, e-mail: commits-help@gluten.apache.org
Re: [PR] [VL] Release all the hash table when SQLExecutionEnd end [gluten]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #12487:
URL: https://github.com/apache/gluten/pull/12487#discussion_r3556271555
##########
backends-velox/src/main/scala/org/apache/spark/listener/VeloxGlutenSQLAppStatusListener.scala:
##########
@@ -70,6 +74,21 @@ class VeloxGlutenSQLAppStatusListener(val driverEndpointRef: RpcEndpointRef)
// logWarning(element.toString);
// }
val executionId = event.executionId.toString
+
+ val qe = event.qe
+ if (qe != null) {
+ val bhjIds = qe.executedPlan.collect { case bhj: BroadcastHashJoinExecTransformer =>
+ bhj.buildBroadcastTableId
+ }.toSet
Review Comment:
This adds new SQLExecutionEnd-based cleanup behavior that is meant to prevent Velox broadcast hash table leaks, but there’s no automated test covering the new contract (e.g., that broadcast hash tables are invalidated at SQLExecutionEnd even when SQLExecution.EXECUTION_ID_KEY was null during planning/execution). Adding a regression test would help prevent leaks from reappearing.
##########
backends-velox/src/main/scala/org/apache/spark/listener/VeloxGlutenSQLAppStatusListener.scala:
##########
@@ -70,6 +74,21 @@ class VeloxGlutenSQLAppStatusListener(val driverEndpointRef: RpcEndpointRef)
// logWarning(element.toString);
// }
val executionId = event.executionId.toString
+
+ val qe = event.qe
+ if (qe != null) {
+ val bhjIds = qe.executedPlan.collect { case bhj: BroadcastHashJoinExecTransformer =>
+ bhj.buildBroadcastTableId
+ }.toSet
+
+ if (bhjIds.nonEmpty) {
+ val idSet = new util.HashSet[String]()
+ bhjIds.foreach(idSet.add)
+ GlutenDriverEndpoint.executorDataMap.forEach(
+ (_, executor) => executor.executorEndpointRef.send(GlutenCleanExecutionResource(executionId, idSet)))
+ }
+ }
Review Comment:
This sends GlutenCleanExecutionResource directly to every executor from the listener, but GlutenDriverEndpoint already sends the same message via executionResourceRelation invalidation on GlutenOnExecutionEnd. That can cause duplicate per-executor RPCs on every SQLExecutionEnd (and splits cleanup logic across two places). Prefer adding the BHJ IDs into GlutenDriverEndpoint.collectResources(executionId, ...) here and let the existing invalidation/onRemoval path perform the single cleanup send.
##########
backends-velox/src/main/scala/org/apache/gluten/execution/HashJoinExecTransformer.scala:
##########
@@ -142,7 +142,7 @@ case class BroadcastHashJoinExecTransformer(
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.")
Review Comment:
This log line now states cleanup "will" happen on SQLExecutionEnd, but cleanup is best-effort (e.g., the SQLExecutionEnd listener path is conditional and may not run in all contexts). Consider softening the wording to avoid implying a guarantee.
--
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: commits-unsubscribe@gluten.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For additional commands, e-mail: commits-help@gluten.apache.org
Re: [PR] [VL] Release all the hash table when SQLExecutionEnd end [gluten]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #12487:
URL: https://github.com/apache/gluten/pull/12487#discussion_r3557490072
##########
cpp/velox/substrait/SubstraitToVeloxPlan.cc:
##########
@@ -458,6 +458,34 @@ core::PlanNodePtr SubstraitToVeloxPlanConverter::toVeloxPlan(const ::substrait::
} else if (
sJoin.has_advanced_extension() &&
SubstraitParser::configSetInOptimization(sJoin.advanced_extension(), "isBHJ=")) {
+ 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.";
+ }
+ }
Review Comment:
Deciding `useHashTableCache` by calling `getJoin(hashTableId)` / `ObjectStore::retrieve(...)` during plan conversion is unsafe for broadcast hash join: the broadcast hash table is typically built/deserialized later at runtime (e.g., `VeloxSerializedBroadcastRDD` and `VeloxBroadcastBuildSideRDD` return `Iterator.empty` after populating the cache). If `useHashTableCache` becomes false here, the `HashJoinNode` will fall back to building from the build-side child, which may be empty and can produce incorrect results.
Instead of checking cache presence here, enable hash table cache based on the presence of `hashTableId` (it should be non-empty for BHJ) and let runtime populate/lookup the cache by key.
--
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: commits-unsubscribe@gluten.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For additional commands, e-mail: commits-help@gluten.apache.org
Re: [PR] [VL] Release all the hash table when SQLExecutionEnd end [gluten]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #12487:
URL: https://github.com/apache/gluten/pull/12487#discussion_r3557036827
##########
cpp/velox/substrait/SubstraitToVeloxPlan.cc:
##########
@@ -469,11 +484,11 @@ core::PlanNodePtr SubstraitToVeloxPlanConverter::toVeloxPlan(const ::substrait::
leftNode,
rightNode,
getJoinOutputType(leftNode, rightNode, joinType),
- true,
+ useHashTableCache,
false,
false,
nullptr,
- sJoin.hashtableid());
+ hashTableId);
Review Comment:
The PR title/description talk about releasing hash tables at SQLExecutionEnd, but this change actually gates whether HashJoinNode is configured to use the broadcast hash table cache based on whether a pre-built HashTableBuilder can be retrieved. If the intent is primarily to avoid leaking cached hash tables, consider updating the PR title/description (or adding the missing cleanup changes) so reviewers/operators understand the actual mechanism being changed.
##########
cpp/velox/substrait/SubstraitToVeloxPlan.cc:
##########
@@ -469,11 +484,11 @@ core::PlanNodePtr SubstraitToVeloxPlanConverter::toVeloxPlan(const ::substrait::
leftNode,
rightNode,
getJoinOutputType(leftNode, rightNode, joinType),
- true,
+ useHashTableCache,
false,
false,
nullptr,
- sJoin.hashtableid());
+ hashTableId);
Review Comment:
This introduces new behavior for BHJ plan conversion (enabling/disabling hash table cache depending on whether a pre-built HashTableBuilder is present). There doesn’t appear to be a regression test exercising this branch (e.g., cache hit vs miss) to ensure correctness and to prevent the original memory-leak/cleanup issue from reappearing.
--
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: commits-unsubscribe@gluten.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For additional commands, e-mail: commits-help@gluten.apache.org