zjuwangg commented on code in PR #8931:
URL: https://github.com/apache/incubator-gluten/pull/8931#discussion_r2043967721
##########
backends-velox/src/main/scala/org/apache/gluten/execution/HashJoinExecTransformer.scala:
##########
@@ -125,9 +129,40 @@ case class BroadcastHashJoinExecTransformer(
override def columnarInputRDDs: Seq[RDD[ColumnarBatch]] = {
val streamedRDD = getColumnarInputRDDs(streamedPlan)
+ val executionId =
sparkContext.getLocalProperty(SQLExecution.EXECUTION_ID_KEY)
+ if (executionId != null) {
+ GlutenDriverEndpoint.collectResources(executionId, buildBroadcastTableId)
+ } else {
+ logWarning(
+ s"Can not trace broadcast table data $buildBroadcastTableId" +
+ s" because execution id is null." +
+ s" Will clean up until expire time.")
+ }
+
val broadcast = buildPlan.executeBroadcast[BuildSideRelation]()
- val broadcastRDD = VeloxBroadcastBuildSideRDD(sparkContext, broadcast)
+ val context =
+ BroadCastHashJoinContext(
+ buildKeyExprs,
+ substraitJoinType,
+ buildSide == BuildRight,
+ condition.isDefined,
+ joinType.isInstanceOf[ExistenceJoin],
+ buildPlan.output,
+ buildBroadcastTableId,
+ isNullAwareAntiJoin
+ )
+ val broadcastRDD = VeloxBroadcastBuildSideRDD(sparkContext, broadcast,
context)
// FIXME: Do we have to make build side a RDD?
streamedRDD :+ broadcastRDD
}
}
+
+case class BroadCastHashJoinContext(
Review Comment:
minor: rename to `BroadcastHashJoinContext`
--
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]
