dianfu commented on code in PR #29498:
URL: https://github.com/apache/flink/pull/29498#discussion_r4388361048
##########
flink-python/pyflink/table/typehints.py:
##########
@@ -74,48 +90,156 @@ def _is_typed_dict(type_hint: Any) -> bool:
)
+def _is_dataclass_type(type_hint: Any) -> bool:
+ return isinstance(type_hint, type) and dataclasses.is_dataclass(type_hint)
+
+
+def _is_named_tuple_type(type_hint: Any) -> bool:
+ return (
+ isinstance(type_hint, type)
+ and issubclass(type_hint, tuple)
+ and hasattr(type_hint, "_fields")
+ )
+
+
+def _is_composite_type(type_hint: Any) -> bool:
+ return (
+ _is_typed_dict(type_hint)
+ or _is_dataclass_type(type_hint)
+ or _is_named_tuple_type(type_hint)
+ )
+
+
+def _generic_composite_error(type_hint: Any) -> TypeError:
+ return TypeError(
+ f"Cannot infer DataType from type hint '{type_hint}'. Generic
composite "
+ "types are not supported. Please specify the data type explicitly."
+ )
+
+
+def _no_fields_error(type_hint: Any) -> TypeError:
+ name = type_hint.__name__ if isinstance(type_hint, type) else type_hint
+ return TypeError(
+ f"Cannot infer DataType from '{name}': it has no fields. "
+ "Please specify the data type explicitly."
+ )
+
+
+def _unresolved_forward_reference_error(hint: Union[ForwardRef, str]) ->
TypeError:
+ name = hint.__forward_arg__ if isinstance(hint, ForwardRef) else hint
+ advice = "Please specify the data type explicitly."
+ # typing generics wrap quoted names in ForwardRef, built-in generics keep
the bare str
+ if isinstance(hint, str) and sys.version_info < (3, 11):
+ advice = (
+ "Before Python 3.11, quoted names inside built-in generics such as
"
+ f"list['{name}'] are not resolved. Use typing.List['{name}'] or
specify "
+ "the data type explicitly."
+ )
+ return TypeError(
+ f"Cannot infer DataType from unresolved forward reference '{name}'.
{advice}"
+ )
+
+
+def _resolve_field_hints(composite: type) -> Dict[str, Any]:
+ try:
+ return get_type_hints(composite)
+ except (NameError, AttributeError, SyntaxError, TypeError) as exc:
+ raise TypeError(
+ f"Cannot resolve the field type hints of '{composite.__name__}': "
+ f"{exc}. Please specify the data type explicitly."
+ ) from exc
+
+
+def _composite_field_hints(composite: Any) -> List[Tuple[str, Any]]:
+ field_hints = _resolve_field_hints(composite)
+ if _is_typed_dict(composite):
+ return [
+ (name, _typed_dict_field_hint(hint, name in
composite.__optional_keys__))
+ for name, hint in field_hints.items()
+ ]
+ if _is_dataclass_type(composite):
Review Comment:
Could we also update the UDTF result normalization to handle single-field
dataclasses?
For example:
```python
@dataclass
class Output:
value: int
@pf.udtf
def emit(value: int) -> Iterator[Output]:
yield Output(value)
```
The inferred schema is `ROW<value BIGINT NOT NULL>`, but evaluating
`emit(42)` through the worker adapter produces `Row(value=Output(value=42))`
instead of `Row(value=42)`.
`_DataFrameTableFunctionAdapter._bind_func()` treats the dataclass instance
as a scalar and wraps it in `Row(item)` because it is neither a `Row`, tuple,
nor mapping. BIGINT serialization then fails with `required argument is not an
integer`.
##########
flink-python/pyflink/dataframe/udtf.py:
##########
@@ -300,7 +300,9 @@ def _resolve_udtf(
def _infer_udtf_return_dtype(
context: _UDFDeclarationContext, return_dtype: Optional[_DataTypeLike]
) -> Tuple[DataType, bool]:
- if return_dtype is not None:
+ if return_dtype is not None and get_origin(return_dtype) is tuple:
Review Comment:
Could we preserve positional tuple naming and validation when the tuple is
wrapped in `Optional`?
For example:
```python
def emit(row) -> Iterator[Tuple[int]]:
yield (row["x"],)
def emit_nullable(row) -> Iterator[Optional[Tuple[int]]]:
yield (row["x"],)
source = pf.from_dict({"x": [1]})
source.flat_map(emit).columns # ["f0"]
source.flat_map(emit_nullable).columns # ["_1"]
```
Adding `Optional` changes the output column name, so downstream expressions
referencing `f0` stop resolving even though the tuple's structure has not
changed.
This also affects validation: `Tuple[int, str]` is rejected by `flat_map`
because it lacks named output fields, whereas `Optional[Tuple[int, str]]` is
accepted with `_1`/`_2`. The latter takes the generic ROW inference path and is
incorrectly marked as having named fields. Both explicit `return_dtype` and
inferred return annotations are affected.
Please recognize tuples inside `Optional` and retain the same positional
naming and alias requirements while preserving nullability.
--
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]
