[SPARK-42856][GRAPHX] Break tie with highest vertex id in label propagation

[awadhesh14]

What changes were proposed in this pull request?

Update the tie breaking logic in label propagation to consider the max vertex id.

Why are the changes needed?

This is a bug reported in SPARK-42856

Does this PR introduce any user-facing change?

No

How was this patch tested?

./dev/run-tests runs successfully

Was this patch authored or co-authored using generative AI tooling?

No

[zhengruifeng]

shall we add a test for it?

[awadhesh14]

@zhengruifeng I am looking into creating a unit test case.

Meanwhile Proposal: Using only active vertices for label propagation in each superstep.
Here are the proposed changed methods:

// Updates the vertex value only if it has changed.

def vertexProgram(vid: VertexId, attr: Long, message: Map[VertexId, Long]): VertexId = {
  val newAttr = if (message.isEmpty) attr else message.maxBy(_._2)._1
  if (newAttr == attr) {
    attr // Vertex value remains unchanged
  } else {
    newAttr // Vertex value has changed
  }
}

// Sends messages only if the source and destination vertex values are different.

def sendMessage(e: EdgeTriplet[VertexId, ED]): Iterator[(VertexId, Map[VertexId, Long])] = {
  if (e.srcAttr != e.dstAttr) {
    Iterator((e.srcId, Map(e.dstAttr -> 1L)), (e.dstId, Map(e.srcAttr -> 1L)))
  } else {
    Iterator.empty
  }
}

// Adding `activeDirection = EdgeDirection.Either` to ensure only active vertices are processed.

val initialMessage = Map[VertexId, Long]()
Pregel(lpaGraph, initialMessage, maxIterations = maxSteps, activeDirection = EdgeDirection.Either)(
  vprog = vertexProgram,
  sendMsg = sendMessage,
  mergeMsg = mergeMessage)

@zhengruifeng Can you review it?

[zhengruifeng]

@awadhesh14 The fix itself seems good: when there are multiple key have the same maximum value, the result is not deterministic.

I think we should add an end-to-end test in LabelPropagationSuite,

the test could be the mini-reproducer of the issue described in https://issues.apache.org/jira/browse/SPARK-42856, and we assert the results.