[SPARK-53544][PYTHON] Support complex types on observations

[ueshin]

What changes were proposed in this pull request?

Supports complex types on observations.

Why are the changes needed?

The observations didn't support complex types.

For example:

>>> observation = Observation("struct")
>>> df = spark.range(10).observe(
...     observation,
...     F.struct(F.count(F.lit(1)).alias("rows"), F.max("id").alias("maxid")).alias("struct"),
... )

• classic

>>> df.collect()
[Row(id=0), Row(id=1), Row(id=2), Row(id=3), Row(id=4), Row(id=5), Row(id=6), Row(id=7), Row(id=8), Row(id=9)]
>>> observation.get
{'struct': JavaObject id=o61}

• connect

>>> df.collect()
Traceback (most recent call last):
...
pyspark.errors.exceptions.base.PySparkTypeError: [UNSUPPORTED_LITERAL] Unsupported Literal 'struct {
...

Does this PR introduce any user-facing change?

Yes, complex types are available on observations.

>>> df.collect()
[Row(id=0), Row(id=1), Row(id=2), Row(id=3), Row(id=4), Row(id=5), Row(id=6), Row(id=7), Row(id=8), Row(id=9)]
>>>
>>> observation.get
{'struct': Row(rows=10, maxid=9)}

How was this patch tested?

Added the related tests.

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

No.

[ueshin]

cc @heyihong

[xinrong-meng]

Curious when does data_type exist and when does it not?

[ueshin]

For example of Array, data_type was added and element_type was deprecated.

https://github.com/apache/spark/blob/master/sql/connect/common/src/main/protobuf/spark/connect/expressions.proto#L227-L244

but the creator of the message may follow it or may not, so this should be able to handle both cases.

Same for Map and Struct.

[heyihong]

To minimize changes, it is not necessary to support these new data type fields. They are still under development and not fully stabilized yet. Currently, the new data type fields are only used in the requests from the Spark Connect Scala Client.

If we really want to support them, I think we need to consider:

• How can we enable the new data type fields in the responses while maintaining backward compatibility?
• How is this code path tested?

[ueshin]

Sure, I'll revert the changes and leave it to you.