[FLINK-40196][python] Add missing-value handling to DataFrame API - #28979
[FLINK-40196][python] Add missing-value handling to DataFrame API#28979MattBelle wants to merge 1 commit into
Conversation
| .. versionadded:: 2.4.0 | ||
| """ | ||
| subset = self._validate_subset(subset) | ||
| conditions = [table_col(col_name).is_not_nan for col_name in subset] |
There was a problem hiding this comment.
For NULL input, IS_NOT_NAN(NULL) returns NULL, and Table.filter only retains rows for which the predicate is TRUE. Therefore, the current implementation incorrectly drops NULL rows which violates the doc: NULL values are preserved.
| .. versionadded:: 2.4.0 | ||
| """ | ||
| subset = self._validate_subset(subset) | ||
| conditions = [table_col(col_name).is_not_nan for col_name in subset] |
There was a problem hiding this comment.
With subset=None, the current implementation applies IS_NAN/IS_NOT_NAN to every column for drop_nan/fill_nan. A common mixed schema containing STRING or BOOLEAN columns will therefore fail during type inference.
I think we could select only FLOAT/DOUBLE columns from the schema and return an equivalent DataFrame when no floating-point columns exist. I checked that Polars, Daft follow this behavior which seem reasonable for me.
| return self._fill_values(value, subset, lambda col: col.is_null) | ||
|
|
||
| @PublicEvolving() | ||
| def fill_nan(self, value: Any, subset: Optional[List[str]] = None) -> "DataFrame": |
There was a problem hiding this comment.
For drop_nan/fill_nan API, we should define clearly the behavior of non-number columns, eg. String, Boolean, etc. I suggest validating column existence first and then ignoring non-floating-point columns, consistent with Daft, and Polars’ default behavior.
| raise TypeError("subset must be a list of strings") | ||
|
|
||
| if not subset: | ||
| raise ValueError("subset cannot be empty") |
There was a problem hiding this comment.
Nit: Polars and Pandas treat empty subset as a no-op instead of raising exceptions. I have no preference. Just comment here for your reference.
| col_expr = table_col(col_name) | ||
| if col_name in subset_set: | ||
| col_type = schema.get_field_data_type(col_name) | ||
| typed_value = table_lit(value).cast(col_type) |
There was a problem hiding this comment.
The current implementation casts the replacement to every target column type. For example, fill_null(0) may replace a STRING NULL with "0" and may fail for ARRAY, ROW, or TIMESTAMP columns.
It only handles the columns which supports the type cast in Spark and Daft.
| @PublicEvolving() | ||
| def fill_nan(self, value: Any, subset: Optional[List[str]] = None) -> "DataFrame": | ||
| """ | ||
| Replace NaN values with a specified value (for float/double columns). |
There was a problem hiding this comment.
The current implementation appears to support fill_nan(None) which converts NaN values to NULL. I think this is reasonable. What about documenting this behavior explicitly?
|
|
||
| /** Implementation of {@link BuiltInFunctionDefinitions#IS_NOT_NAN}. */ | ||
| @Internal | ||
| public final class IsNotNanFunction extends BuiltInScalarFunction { |
There was a problem hiding this comment.
Do we really need IsNotNan?
3 participants
What is the purpose of the change
This pull request adds missing-value handling methods to the PyFlink DataFrame API, enabling users to easily drop or fill NULL and NaN values in their data processing pipelines.
Brief change log
drop_null()method to remove rows containing NULL valuesdrop_nan()method to remove rows containing NaN valuesfill_null()method to replace NULL values with specified valuesfill_nan()method to replace NaN values with specified valuesIS_NANandIS_NOT_NANbuilt-in functions in Flink Table API (required infrastructure for NaN detection)is_nanandis_not_nanhelper functions to Expression API (mirrorsis_null/is_not_nullpattern)_validate_subset()and_fill_values()for DRY code organizationVerifying this change
This change added tests and can be verified as follows:
DataFrameDropNullTests,DataFrameDropNanTests,DataFrameFillNullTests,DataFrameFillNanTests) that verify:ValueErrorValueErrorwith clear error messagesTypeErrorDataFrameNullNanITTests) that verify correct behavior with real data:drop_null()correctly removes rows with NULL valuesdrop_nan()correctly removes rows with NaN valuesfill_null()correctly replaces NULL with specified values (numeric and string)fill_nan()correctly replaces NaN with specified valuessubsetparameter to target specific columnsDoes this pull request potentially affect one of the following parts:
@Public(Evolving): yes (new@PublicEvolvingDataFrame methods)Documentation
Was generative AI tooling used to co-author this PR?
Generated-by: Bob Shell 1.0.6