dianfu commented on code in PR #28979:
URL: https://github.com/apache/flink/pull/28979#discussion_r3870629920
##########
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:
`_is_type_compatible(None, DoubleType()) == False`, which makes
`fill_nan(None)` do nothing which is not as expected.
##########
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):
+ typed_value = table_lit(value).cast(col_type)
+ filled_expr = if_then_else(
+ condition_fn(col_expr),
+ typed_value,
+ col_expr
+ ).alias(col_name)
+ expressions.append(filled_expr)
+ else:
+ expressions.append(col_expr)
+ else:
+ expressions.append(col_expr)
+
+ return DataFrame(self._table.select(*expressions))
+
+ @PublicEvolving()
+ def drop_null(self, subset: Optional[List[str]] = None) -> "DataFrame":
+ """
+ Remove rows containing NULL values.
+
+ This method uses three-valued logic: NULL values in the specified
columns
+ will cause the row to be filtered out. Rows where all checked columns
are
+ non-NULL will be retained.
+
+ :param subset: Column names to check. If None, checks all columns.
+ :return: A new DataFrame with rows containing NULL values removed.
+ :raises ValueError: If subset is empty or contains invalid column
names.
+ :raises TypeError: If subset is not a list of strings.
+
+ Example::
+
+ >>> import pyflink.dataframe as pf
+ >>> df = pf.from_records([
+ ... {"id": 1, "name": "Alice", "age": 30},
+ ... {"id": 2, "name": None, "age": 25},
+ ... {"id": 3, "name": "Bob", "age": None},
+ ... ])
+ >>> df.drop_null() # Drop rows with any NULL
+ >>> df.drop_null(subset=["age"]) # Drop rows where "age" is NULL
+
+ .. versionadded:: 2.4.0
+ """
+ subset = self._validate_subset(subset)
+
+ # Empty subset is a no-op - return self unchanged
+ if not subset:
+ return self
+
+ conditions = [table_col(col_name).is_not_null for col_name in subset]
+ condition = and_(*conditions) if len(conditions) > 1 else conditions[0]
+ return DataFrame(self._table.filter(condition))
+
+ @PublicEvolving()
+ def drop_nan(self, subset: Optional[List[str]] = None) -> "DataFrame":
+ """
+ Remove rows containing NaN values (for float/double columns).
+
+ This method uses three-valued logic: NaN values in the specified
columns
+ will cause the row to be filtered out. NULL values are preserved (not
+ treated as NaN). Only applies to floating-point numeric types.
+
+ :param subset: Column names to check. If None, checks all columns.
+ :return: A new DataFrame with rows containing NaN values removed.
+ :raises ValueError: If subset is empty or contains invalid column
names.
+ :raises TypeError: If subset is not a list of strings.
+
+ Example::
+
+ >>> import pyflink.dataframe as pf
+ >>> df = pf.from_records([
+ ... {"id": 1, "score": 0.95},
+ ... {"id": 2, "score": float('nan')},
+ ... ])
+ >>> df.drop_nan() # Drop rows with any NaN
+ >>> df.drop_nan(subset=["score"]) # Drop rows where "score" is NaN
+
+ .. versionadded:: 2.4.0
+ """
+ schema = self._table.get_schema()
+
+ if subset is None:
+ # Auto-filter to only floating-point columns
+ subset = [
+ col_name for col_name in schema.get_field_names()
+ if isinstance(schema.get_field_data_type(col_name),
(FloatType, DoubleType))
+ ]
+
+ # If no floating-point columns, return unchanged DataFrame
+ if not subset:
+ return self
+ else:
+ subset = self._validate_subset(subset)
Review Comment:
Only the `subset=None` path filters FLOAT/DOUBLE. Explicit
INTEGER/DECIMAL/STRING/BOOLEAN columns still have inconsistent behavior. We
need apply the same FLOAT/DOUBLE filtering to both implicit and explicit
subsets.
##########
flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/scalar/IsNanFunction.java:
##########
@@ -0,0 +1,64 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.runtime.functions.scalar;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.table.functions.BuiltInFunctionDefinitions;
+import org.apache.flink.table.functions.SpecializedFunction.SpecializedContext;
+
+import javax.annotation.Nullable;
+
+import java.math.BigDecimal;
+
+/** Implementation of {@link BuiltInFunctionDefinitions#IS_NAN}. */
+@Internal
+public final class IsNanFunction extends BuiltInScalarFunction {
+
+ public IsNanFunction(SpecializedContext context) {
+ super(BuiltInFunctionDefinitions.IS_NAN, context);
+ }
+
+ public @Nullable Boolean eval(final @Nullable Byte value) {
+ return value == null ? null : false;
+ }
+
+ public @Nullable Boolean eval(final @Nullable Short value) {
+ return value == null ? null : false;
+ }
+
+ public @Nullable Boolean eval(final @Nullable Integer value) {
+ return value == null ? null : false;
+ }
+
+ public @Nullable Boolean eval(final @Nullable Long value) {
+ return value == null ? null : false;
+ }
+
+ public @Nullable Boolean eval(final @Nullable Float value) {
+ return value == null ? null : Float.isNaN(value);
+ }
+
+ public @Nullable Boolean eval(final @Nullable Double value) {
+ return value == null ? null : Double.isNaN(value);
+ }
+
+ public @Nullable Boolean eval(final @Nullable BigDecimal value) {
Review Comment:
Should use DecimalData. For BuiltInScalarFunction, its arguments use
internal representations by default. DECIMAL therefore reaches runtime as
DecimalData, which cannot match eval(BigDecimal).
##########
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:
The compatibility map handles only bool/int/float/str, while the API accepts
Any and PyFlink literals also support Decimal, binary, and temporal values.
See the following example:
```
from decimal import Decimal
import pyflink.dataframe as pf
df = pf.from_records(
[
{"id": 1, "amount": Decimal("1.25")},
{"id": 2, "amount": None},
]
)
result = df.fill_null(
Decimal("0.00"),
subset=["amount"],
)
print(result.collect())
```
The expected output should be:
```
[
Row(1, Decimal("1.25")),
Row(2, Decimal("0.00")),
]
```
Actual result:
```
[
Row(1, Decimal("1.25")),
Row(2, None),
]
```
##########
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):
+ typed_value = table_lit(value).cast(col_type)
+ filled_expr = if_then_else(
+ condition_fn(col_expr),
+ typed_value,
+ col_expr
+ ).alias(col_name)
+ expressions.append(filled_expr)
+ else:
+ expressions.append(col_expr)
+ else:
+ expressions.append(col_expr)
+
+ return DataFrame(self._table.select(*expressions))
+
+ @PublicEvolving()
+ def drop_null(self, subset: Optional[List[str]] = None) -> "DataFrame":
+ """
+ Remove rows containing NULL values.
+
+ This method uses three-valued logic: NULL values in the specified
columns
+ will cause the row to be filtered out. Rows where all checked columns
are
+ non-NULL will be retained.
+
+ :param subset: Column names to check. If None, checks all columns.
+ :return: A new DataFrame with rows containing NULL values removed.
+ :raises ValueError: If subset is empty or contains invalid column
names.
+ :raises TypeError: If subset is not a list of strings.
+
+ Example::
+
+ >>> import pyflink.dataframe as pf
+ >>> df = pf.from_records([
+ ... {"id": 1, "name": "Alice", "age": 30},
+ ... {"id": 2, "name": None, "age": 25},
+ ... {"id": 3, "name": "Bob", "age": None},
+ ... ])
+ >>> df.drop_null() # Drop rows with any NULL
+ >>> df.drop_null(subset=["age"]) # Drop rows where "age" is NULL
+
+ .. versionadded:: 2.4.0
+ """
+ subset = self._validate_subset(subset)
+
+ # Empty subset is a no-op - return self unchanged
+ if not subset:
+ return self
+
+ conditions = [table_col(col_name).is_not_null for col_name in subset]
+ condition = and_(*conditions) if len(conditions) > 1 else conditions[0]
+ return DataFrame(self._table.filter(condition))
+
+ @PublicEvolving()
+ def drop_nan(self, subset: Optional[List[str]] = None) -> "DataFrame":
+ """
+ Remove rows containing NaN values (for float/double columns).
+
+ This method uses three-valued logic: NaN values in the specified
columns
+ will cause the row to be filtered out. NULL values are preserved (not
+ treated as NaN). Only applies to floating-point numeric types.
+
+ :param subset: Column names to check. If None, checks all columns.
+ :return: A new DataFrame with rows containing NaN values removed.
+ :raises ValueError: If subset is empty or contains invalid column
names.
+ :raises TypeError: If subset is not a list of strings.
+
+ Example::
+
+ >>> import pyflink.dataframe as pf
+ >>> df = pf.from_records([
+ ... {"id": 1, "score": 0.95},
+ ... {"id": 2, "score": float('nan')},
+ ... ])
+ >>> df.drop_nan() # Drop rows with any NaN
+ >>> df.drop_nan(subset=["score"]) # Drop rows where "score" is NaN
+
+ .. versionadded:: 2.4.0
+ """
+ schema = self._table.get_schema()
+
+ if subset is None:
+ # Auto-filter to only floating-point columns
+ subset = [
+ col_name for col_name in schema.get_field_names()
+ if isinstance(schema.get_field_data_type(col_name),
(FloatType, DoubleType))
+ ]
+
+ # If no floating-point columns, return unchanged DataFrame
+ if not subset:
+ return self
+ else:
+ subset = self._validate_subset(subset)
+
+ # Empty subset is a no-op - return self unchanged
+ if not subset:
+ return self
+
+ # Preserve NULL values: (is_not_nan(col) OR col IS NULL)
+ conditions = [
+ or_(is_not_nan(table_col(col_name)), table_col(col_name).is_null)
+ for col_name in subset
+ ]
+ condition = and_(*conditions) if len(conditions) > 1 else conditions[0]
+ return DataFrame(self._table.filter(condition))
+
+ @PublicEvolving()
+ def fill_null(self, value: Any, subset: Optional[List[str]] = None) ->
"DataFrame":
+ """
+ Replace NULL values with a specified value.
+
+ This method uses three-valued logic: NULL values in the specified
columns
+ are replaced with the provided value, while non-NULL values are
preserved.
+ The replacement value is automatically cast to match each column's
data type.
Review Comment:
Could document that type inconsistent columns will be ignored.
##########
flink-python/pyflink/dataframe/dataframe.py:
##########
@@ -40,9 +40,25 @@
and_,
call_sql,
col as table_col,
+ if_then_else,
+ is_nan,
+ is_not_nan,
lit as table_lit,
+ or_,
)
from pyflink.table.table import Table
+from pyflink.table.types import (
+ DataTypes as TableDataTypes,
Review Comment:
unused import. Besides, there are the following checkstyle issues which
break the tests:
```
Aug 25 21:34:21 ./pyflink/dataframe/dataframe.py:50:1: F401
'pyflink.table.types.DataTypes as TableDataTypes' imported but unused
Aug 25 21:34:21 ./pyflink/dataframe/dataframe.py:830:1: W293 blank line
contains whitespace
Aug 25 21:34:21 ./pyflink/dataframe/dataframe.py:837:1: W293 blank line
contains whitespace
Aug 25 21:34:21 ./pyflink/dataframe/dataframe.py:847:1: W293 blank line
contains whitespace
Aug 25 21:34:21 ./pyflink/dataframe/dataframe.py:916:1: W293 blank line
contains whitespace
Aug 25 21:34:21 ./pyflink/dataframe/dataframe.py:923:1: W293 blank line
contains whitespace
Aug 25 21:34:21 ./pyflink/dataframe/dataframe.py:927:1: W293 blank line
contains whitespace
Aug 25 21:34:21 ./pyflink/dataframe/tests/test_dataframe.py:1707:5: E303 too
many blank lines (2)
Aug 25 21:34:21 ./pyflink/dataframe/tests/test_dataframe.py:1746:5: E303 too
many blank lines (4)
Aug 25 21:34:21 ./pyflink/dataframe/tests/test_dataframe.py:1824:5: E303 too
many blank lines (2)
```
--
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]
