At the moment, even if the result of that expression (Pregel.scala) is empty:
val newAggMsgDF = msgDF
.filter(Pregel.msg.isNotNull)
.groupBy(ID)
.agg(aggMsgsCol.as(Pregel.MSG_COL_NAME))
we are not stopping the Pregel and continue to do nested joins.
I would like to add a check that if the msgDF.filter(Pregel.msg.isNotNull).isEmpty, we are breaking iterations even if maxIter is not reached yet.
It means, that if there is no new messages, we are not doing join-persist-join iterations.
At the moment, even if the result of that expression (
Pregel.scala) is empty:we are not stopping the Pregel and continue to do nested joins.
I would like to add a check that if the
msgDF.filter(Pregel.msg.isNotNull).isEmpty, we are breaking iterations even ifmaxIteris not reached yet.It means, that if there is no new messages, we are not doing join-persist-join iterations.