dianfu commented on code in PR #29498:
URL: https://github.com/apache/flink/pull/29498#discussion_r4489583892
##########
flink-python/pyflink/dataframe/udtf.py:
##########
@@ -543,7 +555,8 @@ def invoke(*args: Any) -> Iterator[Row]:
result = invoke_func(*args)
for item in _iter_user_results(result):
if is_struct and field_count == 1 and item is not None:
- if not isinstance(item, (Row, tuple,
collections.abc.Mapping)):
+ if not isinstance(item, (Row, tuple,
collections.abc.Mapping)) and \
Review Comment:
For the following example:
```python
@dataclass
class Point:
x: int
y: int
schema = pf.DataType.struct({
"point": pf.DataType.struct({
"x": pf.DataType.int64(),
"y": pf.DataType.int64(),
})
})
@pf.udtf(return_dtype=schema)
def emit():
yield Point(42, 7)
```
Previously it will produce `Row(point=Row(x=42, y=7))`, but the latest
revision produces `Row(point=None)`.
The new `not is_dataclass(item)` condition skips scalar wrapping for every
dataclass instance. The normalizer consequently treats `Point` as the complete
output row and looks for a `point` attribute. Since that attribute does not
exist, the value silently becomes NULL.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
