MattBelle commented on code in PR #28979:
URL: https://github.com/apache/flink/pull/28979#discussion_r4222687681
##########
flink-python/pyflink/dataframe/dataframe.py:
##########
@@ -638,6 +654,279 @@ def pipe(
"""
return func(self, *args, **kwargs)
+ # ======================== Missing Value Handling ========================
+
+ def _validate_subset(self, subset: Optional[List[str]]) -> List[str]:
+ """
+ Validate and normalize the subset parameter.
+
+ :param subset: Column names to validate, or None for all columns.
+ :return: Validated list of column names.
+ :raises ValueError: If subset contains invalid column names.
+ :raises TypeError: If subset is not a list of strings.
+ """
+ schema = self._table.get_schema()
+ all_columns = schema.get_field_names()
+
+ if subset is None:
+ return all_columns
+
+ if not isinstance(subset, list):
+ raise TypeError("subset must be a list of strings")
+
+ # Empty subset is allowed - it's a no-op
+ if not subset:
+ return []
+
+ # Validate all column names exist
+ all_columns_set = set(all_columns)
+ invalid_columns = set(subset) - all_columns_set
+ if invalid_columns:
+ raise ValueError(f"Columns not found in DataFrame:
{sorted(invalid_columns)}")
+
+ return subset
+
+ def _is_type_compatible(self, value: Any, col_type: DataType) -> bool:
+ """
+ Check if a value's type is compatible with a column's data type.
+
+ Only atomic (primitive) types are supported. Complex types (arrays,
maps, rows)
+ are not compatible and will return False.
+
+ :param value: The value to check.
+ :param col_type: The column's Flink data type.
+ :return: True if types are compatible, False otherwise.
+ """
+ # Skip complex types
+ if not isinstance(col_type, AtomicType):
+ return False
+
+ # Map Python types to compatible Flink types
+ type_map = {
Review Comment:
Fixed. Extended the `type_map` in `_is_type_compatible()` to also cover:
- `Decimal` -> `DecimalType`
- `bytes`/`bytearray` -> `BinaryType`/`VarBinaryType`
- `datetime.date` -> `DateType`
- `datetime.time` -> `TimeType`
- `datetime.datetime` -> `TimestampType`/`LocalZonedTimestampType`
Added `test_fill_null_with_decimal_value` (and equivalent binary/temporal
cases) reproducing your exact example, it now returns
```
[
Row(1, Decimal(“1.25”)),
Row(2, Decimal(“0.00”))
]
```
as expected.
--
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]
