[SPARK-51920][SS][PYTHON] Fix type handling of namedTuple for transfromWithState

[zeruibao]

What changes were proposed in this pull request?

Fix type handling of namedTuple for transfromWithState

Why are the changes needed?

We hit the issue when using namedTuple as value of structType like

class Person(NamedTuple):
    age: Integer
    name: String
    
 def handleInputRows(
        self, 
        key: Any, 
        rows: Iterator[Row], 
        timerValues: TimerValues
    ) -> Iterator[Row]:             
        person: Person = Person(age = 1, name= "peter")
        person_list = []
        person_list.append(person)
        self.person_list.update((person_list,))    

The _serialize_to_bytes cannot construct the namedTuple correctly and hit

File "/databricks/spark/python/pyspark/sql/streaming/stateful_processor_api_client.py", line 575, in normalize_value
    return type(v)(normalize_value(e) for e in v)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
TypeError: Person.__new__() missing 2 required positional arguments: 'age' and 'name'

It's because NamedTuple cannot accept generator as parameter.

Does this PR introduce any user-facing change?

No

How was this patch tested?

UT

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

No

[gaogaotiantian]

I think checking the type is much safer than checking an attribute, especially considering that _fields is a not a rare attribute name. If we know what type we are targeting, we should just check type.

[zeruibao]

sounds good! You are so fast. Did not get a chance to add a UT yet 😛 just convert to draft.

[zeruibao]

Hey @gaogaotiantian, I use

                if (
                    isinstance(v, Row) or
                    (isinstance(v, tuple) and hasattr(v, "_fields"))
                ):

instead. isinstance(v, NamedTuple) won’t work because typing.NamedTuple is a class factory, not a runtime parent of instances. Checking isinstance(v, tuple) and _fields is the correct way. Please take another look. Thanks!

[bogao007]

LGTM overall, one minor comment on the test.

[gaogaotiantian]

Can we move this to the top of the test module? Is there a specific reason that you want it in the class definition?

[zeruibao]

you are right. Just moved it the top!