[SPARK-46212][CORE][SQL][SS][CONNECT][MLLIB][GRAPHX][DSTREAM][PROTOBUF][EXAMPLES] Use other functions to simplify the code pattern of s.c.MapOps#view.mapValues

[LuciferYang]

What changes were proposed in this pull request?

This pr simplifies s.c.MapOps.view.mapValues using the following approach:

• For the s.c.immutable.MapOps type, replace it with the s.c.immutable.MapOps#transform function.

def transform[W](f: (K, V) => W): CC[K, W] = map { case (k, v) => (k, f(k, v)) }

Like the case in CountMinSketchSuite:

spark/common/sketch/src/test/scala/org/apache/spark/util/sketch/CountMinSketchSuite.scala

Line 59 in 0d40b1a

sampledItems.groupBy(identity).view.mapValues(_.length.toLong)

• For the s.c.MapOps type, since the transform function does not exist for this type, replace it directly with the map function.

def map[K2, V2](f: ((K, V)) => (K2, V2)): CC[K2, V2] = mapFactory.from(new View.Map(this, f))

Like the case in KafkaTestUtils:

spark/connector/kafka-0-10-sql/src/test/scala/org/apache/spark/sql/kafka010/KafkaTestUtils.scala

Line 381 in 0d40b1a

zkClient.getPartitionsForTopics(zkClient.getAllTopicsInCluster()).view.mapValues(_.size).toSeq

• For the s.c.mutable.MapOps type, the transform function has also been deprecated. At the same time, the signature of transform and its replacement function mapValuesInPlace is as follows:

  @deprecated("Use mapValuesInPlace instead", "2.13.0")
  @inline final def transform(f: (K, V) => V): this.type = mapValuesInPlace(f)

  def mapValuesInPlace(f: (K, V) => V): this.type = {...}

The target type of the value in the function is V, which is different from the target type of the value in s.c.immutable.MapOps#transform, which is W. This does not meet the desired requirement. So in this scenario, it can be divided into two sub-scenarios for handling:

1. If the mutable.Map are using needs to be eventually converted to an immutable.Map, first convert it to an immutable.Map and then use the transform function for replacement. Like the case in SparkConnectPlanner:

spark/connector/connect/server/src/main/scala/org/apache/spark/sql/connect/planner/SparkConnectPlanner.scala

Line 292 in 0d40b1a

namedArguments.asScala.view.mapValues(transformExpression).toMap)

2. If the mutable.Map are using does not need to be converted to an immutable.Map in the end, directly use the map function from scala.collection.MapOps for replacement. Like the case in SparkSession:

spark/connector/connect/client/jvm/src/main/scala/org/apache/spark/sql/SparkSession.scala

Line 313 in 0d40b1a

.putAllNamedArguments(args.asScala.view.mapValues(lit(_).expr).toMap.asJava)))

In addition, there is a special case in PythonWorkerFactory:

spark/core/src/main/scala/org/apache/spark/api/python/PythonWorkerFactory.scala

Line 381 in 0d40b1a

simpleWorkers.view.mapValues(_.destroy())

For this case, it only needs to destroy each Process in values without returning any value. Therefore, it has been rewritten using .values.foreach.

Why are the changes needed?

The coding pattern of s.c.MapOps.view.mapValues seems verbose, it can be simplified using other functions.

Does this PR introduce any user-facing change?

No

How was this patch tested?

Pass GitHub Actions

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

No

[LuciferYang]

cc @cloud-fan

[dongjoon-hyun]

+1, LGTM (from my side). Thank you, @LuciferYang .

[dongjoon-hyun]

Merged to master for Apache Spark 4.

[cloud-fan]

late LGTM, thanks for simplifying it!