[SPARK-58556][CONNECT][UI] Show ML cache status in Spark Connect UI

[zhengruifeng]

What changes were proposed in this pull request?

This PR adds a live ML cache status column to the session table in the Spark Connect UI.

For active sessions, the column shows the number of objects currently in memory and the current and maximum in-memory and total cache sizes. It also reports when memory control is disabled or the ML cache has not been used. Closed sessions and History Server views show N/A because their live cache state is unavailable.

The UI reads the cache status without updating the session's last-access time or initializing an unused ML cache.

Why are the changes needed?

Spark Connect ML models are cached per session, but operators currently cannot inspect cache utilization in the Spark UI. Showing the live status makes it easier to understand model retention, memory pressure, and disk offloading for each session.

Does this PR introduce any user-facing change?

Yes. The Spark Connect session table has a new ML Cache column showing live per-session cache utilization.

How was this patch tested?

Added unit coverage for ML cache status collection and Connect UI rendering.

Tests have not been run yet; this is a draft PR. git diff --check passes.

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

Generated-by: Codex (GPT-5)

[zhengruifeng]

Here is a simple end-to-end PySpark ML example for manually checking the UI.

Start a Connect server from the built Spark checkout:

./sbin/start-connect-server.sh --master local[2]

In another terminal, start a Connect PySpark shell:

./bin/pyspark --remote sc://localhost:15002

Then run:

from pyspark.ml.classification import LogisticRegression
from pyspark.ml.linalg import Vectors

training = spark.createDataFrame(
    [
        (1.0, Vectors.dense([0.0, 1.1, 0.1])),
        (0.0, Vectors.dense([2.0, 1.0, -1.0])),
        (0.0, Vectors.dense([2.0, 1.3, 1.0])),
        (1.0, Vectors.dense([0.0, 1.2, -0.5])),
    ],
    ["label", "features"],
)

model = LogisticRegression(maxIter=5, regParam=0.01).fit(training)
model.transform(training).select("label", "prediction", "probability").show(
    truncate=False
)

Keep the PySpark shell open and visit http://localhost:4040/connect/ (or the Spark UI URL printed by the server). Before fit, the session's ML Cache cell should say Not used. After fit, reload the page; it should show one in-memory object and nonzero in-memory/total cache usage.

[zhengruifeng]