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: [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]
