MattBelle commented on code in PR #28979:
URL: https://github.com/apache/flink/pull/28979#discussion_r4222120339
##########
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 = {
+ bool: (BooleanType,),
+ int: (IntegralType, FractionalType, DecimalType),
+ float: (FractionalType, DecimalType),
+ str: (CharType, VarCharType),
+ }
+
+ compatible_types = type_map.get(type(value))
+ if compatible_types is None:
+ return False
+
+ return isinstance(col_type, compatible_types)
+
+ def _fill_values(
+ self,
+ value: Any,
+ subset: Optional[List[str]],
+ condition_fn: Callable[[Expression], Expression]
+ ) -> "DataFrame":
+ """
+ Helper method to fill values based on a condition.
+
+ :param value: The value to use as replacement.
+ :param subset: Column names to fill, or None for all columns.
+ :param condition_fn: Function that takes a column expression and
returns
+ a boolean expression indicating when to replace.
+ :return: A new DataFrame with values replaced.
+ """
+ subset = self._validate_subset(subset)
+
+ # Empty subset is a no-op - return self unchanged
+ if not subset:
+ return self
+
+ subset_set = set(subset)
+
+ schema = self._table.get_schema()
+ all_columns = schema.get_field_names()
+
+ expressions = []
+ for col_name in all_columns:
+ col_expr = table_col(col_name)
+ if col_name in subset_set:
+ col_type = schema.get_field_data_type(col_name)
+
+ # Only fill type-compatible columns
+ if self._is_type_compatible(value, col_type):
Review Comment:
Fixed. `_is_type_compatible` now short-circuits on `value is None` and
returns `True` for any `AtomicType` column
--
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]
