[FLINK-40196][python] Add missing-value handling to DataFrame API by MattBelle · Pull Request #28979 · apache/flink

MattBelle · 2026-08-14T23:03:47Z

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

Added drop_null() method to remove rows containing NULL values
Added drop_nan() method to remove rows containing NaN values
Added fill_null() method to replace NULL values with specified values
Added fill_nan() method to replace NaN values with specified values
Implemented IS_NAN and IS_NOT_NAN built-in functions in Flink Table API (required infrastructure for NaN detection)
Added is_nan and is_not_nan helper functions to Expression API (mirrors is_null/is_not_null pattern)
Added helper methods _validate_subset() and _fill_values() for DRY code organization
Added comprehensive unit and integration tests for all new methods

Verifying this change

This change added tests and can be verified as follows:

Added unit tests (DataFrameDropNullTests, DataFrameDropNanTests, DataFrameFillNullTests, DataFrameFillNanTests) that verify:
- Schema preservation for all operations
- Parameter validation with comprehensive error case coverage:
  - Empty subset lists raise ValueError
  - Invalid column names raise ValueError with clear error messages
  - Non-list subset parameters raise TypeError
Added integration tests (DataFrameNullNanITTests) that verify correct behavior with real data:
- drop_null() correctly removes rows with NULL values
- drop_nan() correctly removes rows with NaN values
- fill_null() correctly replaces NULL with specified values (numeric and string)
- fill_nan() correctly replaces NaN with specified values
- All methods work correctly with subset parameter to target specific columns
Tests cover success cases, error conditions, and different data types

Does this pull request potentially affect one of the following parts:

Dependencies (does it add or upgrade a dependency): no
The public API, i.e., is any changed class annotated with @Public(Evolving): yes (new @PublicEvolving DataFrame methods)
The serializers: no
The runtime per-record code paths (performance sensitive): no
Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Kubernetes/Yarn, ZooKeeper: no
The S3 file system connector: no

Documentation

Does this pull request introduce a new feature? yes
If yes, how is the feature documented? JavaDocs (comprehensive Python docstrings with examples and version tags)

Was generative AI tooling used to co-author this PR?

Yes (please specify the tool below)

Generated-by: Bob Shell 1.0.6

flinkbot · 2026-08-14T23:14:25Z

CI report:

b90cad0 Azure: SUCCESS
a21e01c Azure: PENDING

Bot commands

The @flinkbot bot supports the following commands:

@flinkbot run azure re-run the last Azure build

dianfu · 2026-08-17T05:40:02Z

+        .. versionadded:: 2.4.0
+        """
+        subset = self._validate_subset(subset)
+        conditions = [table_col(col_name).is_not_nan for col_name in subset]


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.

Good catch! Wrote a new test for this scenario (test_drop_nan_preserves_null_values). Added OR col IS NULL to the filter condition in drop_nan() to explicitly preserve NULL values. All DataFrameNullNanITTests are passing.

dianfu · 2026-08-17T05:43:39Z

+        .. versionadded:: 2.4.0
+        """
+        subset = self._validate_subset(subset)
+        conditions = [table_col(col_name).is_not_nan for col_name in subset]


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.

Fixed! Added auto-filtering to FLOAT/DOUBLE columns when subset=None for both drop_nan() and fill_nan(). This prevents validation errors on mixed schemas. Added 4 new tests covering mixed schemas and edge cases. All DataFrameNullNanITTests are passing. This matches the behavior of Polars and Daft.

dianfu · 2026-08-17T05:47:28Z

+        return self._fill_values(value, subset, lambda col: col.is_null)
+
+    @PublicEvolving()
+    def fill_nan(self, value: Any, subset: Optional[List[str]] = None) -> "DataFrame":


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.

Addressed. I updated drop_nan() and fill_nan() so that when subset=None, they only operate on floating-point columns instead of attempting to apply IS_NAN/IS_NOT_NAN across the full schema. That resolves the undefined behavior for non-numeric columns in the default path and aligns the behavior noted in the review for Polars and Daft.

dianfu · 2026-08-17T05:51:08Z

+            raise TypeError("subset must be a list of strings")
+
+        if not subset:
+            raise ValueError("subset cannot be empty")


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.

Addressed. I updated drop_null() to treat empty subset as a no-op (returns DataFrame unchanged), aligning with Pandas/Polars behavior and matching the existing behavior in fill_null() and fill_nan().

dianfu · 2026-08-17T06:36:23Z

+            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)


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.

Addressed. I moved the execution-based fill_null compatibility coverage into DataFrameNullNanITTests, since these assertions depend on collecting real results rather than just validating schema. The incompatible-type behavior is covered by:

test_fill_null_type_compatibility

test_fill_null_skips_incompatible_array_column

test_fill_null_skips_incompatible_string_column

These tests are now passing.

I also decided to treat STRING columns as incompatible with numeric fill values and not coerce values like 0 to "0", since that matches Spark behavior.

dianfu · 2026-08-17T06:43:38Z

+    @PublicEvolving()
+    def fill_nan(self, value: Any, subset: Optional[List[str]] = None) -> "DataFrame":
+        """
+        Replace NaN values with a specified value (for float/double columns).


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?

Addressed. I updated the fill_nan() docstring to document the fill_nan(None) behavior explicitly, including that None converts NaN values to NULL.

dianfu · 2026-08-17T07:08:14Z

+
+/** Implementation of {@link BuiltInFunctionDefinitions#IS_NOT_NAN}. */
+@Internal
+public final class IsNotNanFunction extends BuiltInScalarFunction {


Do we really need IsNotNan?

I originally did not add it, but I noticed IS_NULL had a corresponding IS_NOT_NULL, so I mirrored that behavior for NaN for consistency. If you prefer, I can revert it back to just IS_NAN.

…handling - Fix NULL preservation in drop_nan() by adding OR col IS NULL condition - Add auto-filtering to FLOAT/DOUBLE columns for drop_nan/fill_nan with subset=None - Add type compatibility checking in fill_null() to skip incompatible columns - Change empty subset behavior to no-op (align with Pandas/Polars) - Update fill_nan() docstring to document fill_nan(None) behavior - Add 8 new integration tests for edge cases and type compatibility Generated-by: Bob Shell 1.0.6

[FLINK-40196][python] Add missing-value handling to DataFrame API

b90cad0

dianfu reviewed Aug 17, 2026

View reviewed changes

Uh oh!

Conversation

MattBelle commented Aug 14, 2026

What is the purpose of the change

Brief change log

Verifying this change

Does this pull request potentially affect one of the following parts:

Documentation

Was generative AI tooling used to co-author this PR?

Uh oh!

flinkbot commented Aug 14, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

CI report:

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

dianfu Aug 17, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

flinkbot commented Aug 14, 2026 •

edited

Loading

dianfu Aug 17, 2026 •

edited

Loading