[SPARK-59402][ML] Reduce RandomForestRegressionModel broadcast size

[zhengruifeng]

What changes were proposed in this pull request?

This PR reduces the broadcast payload created by RandomForestRegressionModel.transform.

Instead of broadcasting the complete model, the transform broadcasts only the tree root nodes.
Prediction uses an unweighted root-node overload of TreeEnsembleModel.predictRaw, while leaf
prediction uses the existing TreeEnsembleModel.predictLeaf helper.

Why are the changes needed?

Prediction and leaf traversal need only the root nodes. Broadcasting the complete model also
serializes unrelated state such as the model and tree-model parameter graphs, UIDs, and metadata.
Avoiding that state reduces driver and executor memory pressure for long-lived Spark Connect
servers. The unweighted helper also avoids allocating an intermediate prediction array for each
row.

A temporary deterministic probe trained a 20-tree model with 220 total nodes and serialized each
payload using Spark's configured serializer. The full model required 67,381 bytes, while the root
nodes required 18,718 bytes, a 72.2% reduction.

Does this PR introduce any user-facing change?

No.

How was this patch tested?

The following checks passed:

build/sbt mllib/compile
build/sbt 'mllib/testOnly org.apache.spark.ml.regression.RandomForestRegressorSuite'

RandomForestRegressorSuite ran 11 tests covering transform prediction and leaf-index output. No
new test was added because the change only narrows the serialized state used by those existing code
paths.

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

Generated-by: OpenAI Codex (GPT-5)

[HyukjinKwon]

0 blocking, 0 non-blocking, 0 nits.
Clean, behavior-preserving memory optimization that mirrors the established GBTRegressionModel pattern; no issues found.

Verification

Confirmed transform output is unchanged. Prediction: the new UDF computes predictRaw(features, rootNodes) / rootNodes.length with rootNodes = _trees.map(_.rootNode); the model's predict is predictRaw(features, _trees) / getNumTrees. Both sum rootNode.predictImpl(features).prediction over the same trees and divide by the same count (getNumTrees == _trees.length), and the divisor is non-zero via require(_trees.nonEmpty). Leaf: predictLeaf(features, rootNodes) computes DecisionTreeModel.predictLeaf(features, rootNodes(i)) per tree, equal to the prior per-tree predictLeaf. Node is serializable and already broadcast/closed-over by GBTRegressor and RandomForestClassifier.

[zhengruifeng]

Merge Summary:

• merged into master 77e3af3
• merged into branch-4.x 49595b4

Posted by merge_spark_pr.py