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.