[SPARK-53908][CONNECT] Fix observations on Spark Connect with plan cache

[ueshin]

What changes were proposed in this pull request?

Fixes observations on Spark Connect with plan cache enabled.

Why are the changes needed?

The observations on Spark Connect with plan cache enabled has cases that can't see the observed values.

>>> from pyspark.sql.observation import Observation
>>> from pyspark.sql import functions as F
>>>
>>> df = spark.range(10)
>>> observation = Observation()
>>> observed_df = df.observe(observation, F.count(F.lit(1)).alias("cnt"))
>>>
>>> observed_df.schema  # causes to cache
StructType([StructField('id', LongType(), False)])
>>>
>>> observed_df.show()
+---+
| id|
+---+
...
+---+

>>>
>>> observation.get
{}

This should be:

>>> spark.conf.set('spark.connect.session.planCache.enabled', False)
...
>>> observation.get
{'cnt': 10}

This is because the cached plan by the Analyze request for observed_df.schema doesn't register the new Observation for the actual execution.

The observations should be registered later.

Does this PR introduce any user-facing change?

Yes, the observations will be available.

How was this patch tested?

Modified the related tests.

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

No.

[ueshin]

Thanks! merging to master.