This is an automated email from the ASF dual-hosted git repository.
dianfu pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/master by this push:
new 6359419dbd2 [FLINK-40196][python] Add missing-value handling to
DataFrame API
6359419dbd2 is described below
commit 6359419dbd209fc1e82f5fe6adb0578ec4099031
Author: Matt Belle <[email protected]>
AuthorDate: Fri Aug 14 12:57:39 2026 -0400
[FLINK-40196][python] Add missing-value handling to DataFrame API
This closes #28979.
---
docs/data/sql_functions.yml | 6 +
docs/data/sql_functions_zh.yml | 6 +
.../docs/reference/pyflink.dataframe/dataframe.rst | 13 +
.../docs/reference/pyflink.table/expressions.rst | 4 +
flink-python/pyflink/dataframe/dataframe.py | 328 ++++++++++-
.../pyflink/dataframe/tests/test_dataframe.py | 646 +++++++++++++++++++++
flink-python/pyflink/table/expression.py | 18 +
flink-python/pyflink/table/expressions.py | 26 +-
.../org/apache/flink/table/api/Expressions.java | 20 +
.../flink/table/api/internal/BaseExpressions.java | 12 +
.../functions/BuiltInFunctionDefinitions.java | 22 +
.../runtime/functions/scalar/IsNanFunction.java | 63 ++
.../runtime/functions/scalar/IsNotNanFunction.java | 63 ++
tools/maven/suppressions.xml | 1 +
14 files changed, 1225 insertions(+), 3 deletions(-)
diff --git a/docs/data/sql_functions.yml b/docs/data/sql_functions.yml
index c53b6767ade..a474962d73a 100644
--- a/docs/data/sql_functions.yml
+++ b/docs/data/sql_functions.yml
@@ -40,6 +40,12 @@ comparison:
- sql: value IS NOT NULL
table: value.isNotNull
description: Returns TRUE if value is not NULL.
+ - sql: IS_NAN(value)
+ table: value.isNan
+ description: Returns TRUE if value is NaN; returns FALSE if value is a
non-NaN numeric value; returns NULL if value is NULL.
+ - sql: IS_NOT_NAN(value)
+ table: value.isNotNan
+ description: Returns TRUE if value is a non-NaN numeric value; returns
FALSE if value is NaN; returns NULL if value is NULL.
- sql: value1 IS DISTINCT FROM value2
description: Returns TRUE if two values are different. NULL values are
treated as identical here. E.g., 1 IS DISTINCT FROM NULL returns TRUE; NULL IS
DISTINCT FROM NULL returns FALSE.
- sql: value1 IS NOT DISTINCT FROM value2
diff --git a/docs/data/sql_functions_zh.yml b/docs/data/sql_functions_zh.yml
index 7c0a4e95d15..3225ce7ab40 100644
--- a/docs/data/sql_functions_zh.yml
+++ b/docs/data/sql_functions_zh.yml
@@ -40,6 +40,12 @@ comparison:
- sql: value IS NOT NULL
table: value.isNotNull
description: 如果值不为 `NULL` 返回 `TRUE`。
+ - sql: IS_NAN(value)
+ table: value.isNan
+ description: 如果值为 `NaN` 返回 `TRUE`；如果值是非 `NaN` 的数值返回 `FALSE`；如果值为 `NULL` 返回
`NULL`。
+ - sql: IS_NOT_NAN(value)
+ table: value.isNotNan
+ description: 如果值是非 `NaN` 的数值返回 `TRUE`；如果值为 `NaN` 返回 `FALSE`；如果值为 `NULL` 返回
`NULL`。
- sql: value1 IS DISTINCT FROM value2
description: |
A 和 B 的数据类型和值不完全相同返回 `TRUE`。A 和 B 的数据类型和值都相同返回 `FALSE`。`NULL` 视为与任何其他值相同。
diff --git a/flink-python/docs/reference/pyflink.dataframe/dataframe.rst
b/flink-python/docs/reference/pyflink.dataframe/dataframe.rst
index 46286497076..9f0f7764dcc 100644
--- a/flink-python/docs/reference/pyflink.dataframe/dataframe.rst
+++ b/flink-python/docs/reference/pyflink.dataframe/dataframe.rst
@@ -104,6 +104,19 @@ Joins
DataFrame.join
+Missing-Value Handling
+----------------------
+
+.. currentmodule:: pyflink.dataframe
+
+.. autosummary::
+ :toctree: api/
+
+ DataFrame.drop_null
+ DataFrame.drop_nan
+ DataFrame.fill_null
+ DataFrame.fill_nan
+
Aggregations
------------
diff --git a/flink-python/docs/reference/pyflink.table/expressions.rst
b/flink-python/docs/reference/pyflink.table/expressions.rst
index a76320245a2..e2ee9777c58 100644
--- a/flink-python/docs/reference/pyflink.table/expressions.rst
+++ b/flink-python/docs/reference/pyflink.table/expressions.rst
@@ -35,6 +35,8 @@ Expressions
and_
or_
not_
+ is_nan
+ is_not_nan
current_database
current_date
current_time
@@ -120,6 +122,8 @@ arithmetic functions
Expression.if_null
Expression.is_null
Expression.is_not_null
+ Expression.is_nan
+ Expression.is_not_nan
Expression.is_true
Expression.is_false
Expression.is_not_true
diff --git a/flink-python/pyflink/dataframe/dataframe.py
b/flink-python/pyflink/dataframe/dataframe.py
index 8698dda1367..8418a90dad1 100644
--- a/flink-python/pyflink/dataframe/dataframe.py
+++ b/flink-python/pyflink/dataframe/dataframe.py
@@ -19,6 +19,7 @@
import datetime
import keyword
from contextlib import ExitStack
+from decimal import Decimal
from typing import (
TYPE_CHECKING,
Any,
@@ -63,12 +64,38 @@ from pyflink.table.expressions import (
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.statement_set import StatementSet
-from pyflink.table.types import ArrayType, MapType, MultisetType, RowType
from pyflink.table.table_descriptor import TableDescriptor
+from pyflink.table.types import (
+ ArrayType,
+ MapType,
+ MultisetType,
+ RowType,
+ FloatType,
+ DoubleType,
+ AtomicType,
+ IntegralType,
+ FractionalType,
+ DecimalType,
+ CharType,
+ VarCharType,
+ BooleanType,
+ BinaryType,
+ VarBinaryType,
+ DateType,
+ DayTimeIntervalType,
+ TimeType,
+ TimestampType,
+ LocalZonedTimestampType,
+ DataType as TableDataType,
+)
from pyflink.util.api_stability_decorators import PublicEvolving
from pyflink.util.java_utils import to_jarray
@@ -1709,6 +1736,305 @@ class DataFrame:
"""
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) or not all(
+ isinstance(column, str) for column in subset
+ ):
+ 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: TableDataType) -> 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
+
+ # None is compatible with any atomic column type (assigning NULL)
+ if value is None:
+ return True
+
+ # Map Python types to compatible Flink types
+ type_map = {
+ bool: (BooleanType,),
+ int: (IntegralType, FractionalType, DecimalType),
+ float: (FractionalType, DecimalType),
+ str: (CharType, VarCharType),
+ Decimal: (DecimalType,),
+ bytes: (BinaryType, VarBinaryType),
+ bytearray: (BinaryType, VarBinaryType),
+ datetime.date: (DateType,),
+ datetime.time: (TimeType,),
+ datetime.datetime: (TimestampType, LocalZonedTimestampType),
+ datetime.timedelta: (DayTimeIntervalType,),
+ }
+
+ compatible_types = type_map.get(type(value))
+ if compatible_types is None:
+ return False
+
+ return isinstance(col_type, compatible_types)
+
+ def _filter_to_float_columns(self, subset: Optional[List[str]]) ->
List[str]:
+ """
+ Resolve the floating-point columns to operate on for NaN handling.
+
+ :param subset: Column names to check, or None for all columns.
+ :return: The subset's column names that are FLOAT or DOUBLE typed.
+ :raises ValueError: If subset contains invalid column names.
+ :raises TypeError: If subset is not a list of strings.
+ """
+ schema = self._table.get_schema()
+ subset = schema.get_field_names() if subset is None else
self._validate_subset(subset)
+
+ return [
+ col_name for col_name in subset
+ if isinstance(schema.get_field_data_type(col_name), (FloatType,
DoubleType))
+ ]
+
+ 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):
+ if value is None:
+ # A NULL literal cannot have its type inferred; it
must be
+ # given explicitly, and must be nullable.
+ typed_value = table_lit(None, col_type.nullable())
+ else:
+ 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 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 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
+ """
+ subset = self._filter_to_float_columns(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.
+ Columns whose data type is not compatible with ``value`` are left
unchanged.
+ If ``value`` is ``None``, the DataFrame is returned unchanged.
+
+ :param value: The value to replace NULL with.
+ :param subset: Column names to fill. If None, fills all columns.
+ :return: A DataFrame with NULL values replaced, or this DataFrame
unchanged if ``value``
+ is ``None``.
+ :raises ValueError: If subset 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", "quantity": 10},
+ ... {"id": 2, "name": None, "quantity": None},
+ ... ])
+ >>> df.fill_null(0, subset=["quantity"])
+ >>> df.fill_null("unknown", subset=["name"])
+
+ .. versionadded:: 2.4.0
+ """
+ if value is None:
+ self._validate_subset(subset)
+ return self
+
+ return self._fill_values(value, subset, lambda col: col.is_null)
+
+ @PublicEvolving()
+ def fill_nan(self, value: Any, subset: Optional[List[str]] = None) ->
"DataFrame":
+ """
+ Replace NaN values with a specified value (for float/double columns).
+
+ This method uses three-valued logic: NaN values in the specified
columns
+ are replaced with the provided value, while non-NaN values are
preserved.
+ NULL values are preserved (not treated as NaN). The replacement value
is
+ automatically cast to match each column's data type. If ``value`` is
+ ``None``, NaN values are converted to NULL.
+
+ Non-floating-point columns in ``subset`` are ignored, since only
+ floating-point types support NaN. Columns whose data type is not
+ compatible with ``value`` are also left unchanged.
+
+ :param value: The value to replace NaN with. Can be ``None`` to convert
+ NaN values to NULL.
+ :param subset: Column names to fill. If None, fills all floating-point
+ columns.
+ :return: A new DataFrame with NaN values replaced.
+ :raises ValueError: If subset 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.fill_nan(0.0, subset=["score"])
+
+ .. versionadded:: 2.4.0
+ """
+ subset = self._filter_to_float_columns(subset)
+
+ # Empty subset is a no-op - return self unchanged
+ if not subset:
+ return self
+
+ return self._fill_values(value, subset, lambda col: is_nan(col))
+
# ======================== Conversion ========================
@PublicEvolving()
diff --git a/flink-python/pyflink/dataframe/tests/test_dataframe.py
b/flink-python/pyflink/dataframe/tests/test_dataframe.py
index b2833cfd8b5..5305802a406 100644
--- a/flink-python/pyflink/dataframe/tests/test_dataframe.py
+++ b/flink-python/pyflink/dataframe/tests/test_dataframe.py
@@ -19,6 +19,7 @@
import array
import decimal
import inspect
+import math
import os
import pandas as pd
import pyarrow as pa
@@ -43,6 +44,7 @@ from pyflink.table import (
TableSchema,
)
from pyflink.table.expression import Expression
+from pyflink.table.expressions import is_nan, is_not_nan
from pyflink.table.types import LocalZonedTimestampType, TimestampType
from pyflink.testing.test_case_utils import (
PyFlinkDataFrameUTTestCase,
@@ -2969,6 +2971,183 @@ class DataFrameITTests(PyFlinkStreamDataFrameTestCase):
)
+class DataFrameDropNullTests(PyFlinkDataFrameUTTestCase):
+ def setUp(self):
+ super().setUp()
+ self.dataframe = pf.from_records(
+ [
+ {"id": 1, "name": "Alice", "age": 30},
+ {"id": 2, "name": None, "age": 25},
+ {"id": 3, "name": "Bob", "age": None},
+ {"id": 4, "name": None, "age": None},
+ ],
+ schema=["id", "name", "age"],
+ )
+ self.expected_schema = [
+ TableDataTypes.BIGINT(),
+ TableDataTypes.STRING(),
+ TableDataTypes.BIGINT(),
+ ]
+
+ def test_drop_null_without_subset_checks_all_columns(self):
+ result = self.dataframe.drop_null()
+ self.assert_dataframe_schema(result, ["id", "name", "age"],
self.expected_schema)
+
+ def test_drop_null_with_single_column_subset(self):
+ result = self.dataframe.drop_null(subset=["age"])
+ self.assert_dataframe_schema(result, ["id", "name", "age"],
self.expected_schema)
+
+ def test_drop_null_with_multiple_column_subset(self):
+ result = self.dataframe.drop_null(subset=["name", "age"])
+ self.assert_dataframe_schema(result, ["id", "name", "age"],
self.expected_schema)
+
+ def test_drop_null_with_empty_subset_returns_unchanged(self):
+ result = self.dataframe.drop_null(subset=[])
+ self.assert_dataframe_schema(result, ["id", "name", "age"],
self.expected_schema)
+ # Verify it's a no-op by checking the result is the same DataFrame
+ self.assertEqual(result._table, self.dataframe._table)
+
+ def test_drop_null_with_invalid_column_raises_error(self):
+ with self.assertRaises(ValueError) as context:
+ self.dataframe.drop_null(subset=["invalid_column"])
+ self.assertIn("Columns not found in DataFrame", str(context.exception))
+
+ def test_drop_null_with_non_list_subset_raises_error(self):
+ with self.assertRaises(TypeError) as context:
+ self.dataframe.drop_null(subset="age")
+ self.assertIn("subset must be a list of strings",
str(context.exception))
+
+
+class DataFrameDropNanTests(PyFlinkDataFrameUTTestCase):
+ def setUp(self):
+ super().setUp()
+ self.dataframe = pf.from_records(
+ [
+ {"id": 1, "score": 0.95},
+ {"id": 2, "score": float('nan')},
+ {"id": 3, "score": 0.85},
+ ],
+ schema=["id", "score"],
+ )
+ self.expected_schema = [TableDataTypes.BIGINT(),
TableDataTypes.DOUBLE()]
+
+ def test_drop_nan_without_subset_checks_all_columns(self):
+ result = self.dataframe.drop_nan()
+ self.assert_dataframe_schema(result, ["id", "score"],
self.expected_schema)
+
+ def test_drop_nan_with_subset(self):
+ result = self.dataframe.drop_nan(subset=["score"])
+ self.assert_dataframe_schema(result, ["id", "score"],
self.expected_schema)
+
+ def test_drop_nan_with_multiple_column_subset(self):
+ result = self.dataframe.drop_nan(subset=["id", "score"])
+ self.assert_dataframe_schema(result, ["id", "score"],
self.expected_schema)
+
+ def test_drop_nan_with_empty_subset_returns_unchanged(self):
+ result = self.dataframe.drop_nan(subset=[])
+ self.assert_dataframe_schema(result, ["id", "score"],
self.expected_schema)
+ # Verify it's a no-op by checking the result is the same DataFrame
+ self.assertEqual(result._table, self.dataframe._table)
+
+ def test_drop_nan_with_invalid_column_raises_error(self):
+ with self.assertRaises(ValueError) as context:
+ self.dataframe.drop_nan(subset=["invalid_column"])
+ self.assertIn("Columns not found in DataFrame", str(context.exception))
+
+ def test_drop_nan_with_non_list_subset_raises_error(self):
+ with self.assertRaises(TypeError) as context:
+ self.dataframe.drop_nan(subset="score")
+ self.assertIn("subset must be a list of strings",
str(context.exception))
+
+
+class DataFrameFillNullTests(PyFlinkDataFrameUTTestCase):
+ def setUp(self):
+ super().setUp()
+ self.dataframe = pf.from_records(
+ [
+ {"id": 1, "name": "Alice", "quantity": 10},
+ {"id": 2, "name": None, "quantity": None},
+ {"id": 3, "name": "Bob", "quantity": 5},
+ ],
+ schema=["id", "name", "quantity"],
+ )
+ self.expected_schema = [
+ TableDataTypes.BIGINT(),
+ TableDataTypes.STRING(),
+ TableDataTypes.BIGINT(),
+ ]
+
+ def test_fill_null_with_numeric_value(self):
+ result = self.dataframe.fill_null(0, subset=["quantity"])
+ self.assert_dataframe_schema(result, ["id", "name", "quantity"],
self.expected_schema)
+
+ def test_fill_null_with_string_value(self):
+ result = self.dataframe.fill_null("unknown", subset=["name"])
+ self.assert_dataframe_schema(result, ["id", "name", "quantity"],
self.expected_schema)
+
+ def test_fill_null_without_subset_fills_all_columns(self):
+ result = self.dataframe.fill_null(0)
+ self.assert_dataframe_schema(result, ["id", "name", "quantity"],
self.expected_schema)
+
+ def test_fill_null_with_empty_subset_returns_unchanged(self):
+ result = self.dataframe.fill_null(0, subset=[])
+ self.assert_dataframe_schema(result, ["id", "name", "quantity"],
self.expected_schema)
+ # Verify it's a no-op by checking the result is the same DataFrame
+ self.assertEqual(result._table, self.dataframe._table)
+
+ def test_fill_null_with_invalid_column_raises_error(self):
+ with self.assertRaises(ValueError) as context:
+ self.dataframe.fill_null(0, subset=["invalid_column"])
+ self.assertIn("Columns not found in DataFrame", str(context.exception))
+
+ def test_fill_null_with_non_list_subset_raises_error(self):
+ with self.assertRaises(TypeError) as context:
+ self.dataframe.fill_null(0, subset="quantity")
+ self.assertIn("subset must be a list of strings",
str(context.exception))
+
+
+class DataFrameFillNanTests(PyFlinkDataFrameUTTestCase):
+ def setUp(self):
+ super().setUp()
+ self.dataframe = pf.from_records(
+ [
+ {"id": 1, "score": 0.95},
+ {"id": 2, "score": float('nan')},
+ {"id": 3, "score": 0.85},
+ ],
+ schema=["id", "score"],
+ )
+ self.expected_schema = [TableDataTypes.BIGINT(),
TableDataTypes.DOUBLE()]
+
+ def test_fill_nan_with_numeric_value(self):
+ result = self.dataframe.fill_nan(0.0, subset=["score"])
+ self.assert_dataframe_schema(result, ["id", "score"],
self.expected_schema)
+
+ def test_fill_nan_without_subset_fills_all_columns(self):
+ result = self.dataframe.fill_nan(0.0)
+ self.assert_dataframe_schema(result, ["id", "score"],
self.expected_schema)
+
+ def test_fill_nan_with_multiple_column_subset(self):
+ result = self.dataframe.fill_nan(0.0, subset=["id", "score"])
+ self.assert_dataframe_schema(result, ["id", "score"],
self.expected_schema)
+
+ def test_fill_nan_with_empty_subset_returns_unchanged(self):
+ result = self.dataframe.fill_nan(0.0, subset=[])
+ self.assert_dataframe_schema(result, ["id", "score"],
self.expected_schema)
+ # Verify it's a no-op by checking the result is the same DataFrame
+ self.assertEqual(result._table, self.dataframe._table)
+
+ def test_fill_nan_with_invalid_column_raises_error(self):
+ with self.assertRaises(ValueError) as context:
+ self.dataframe.fill_nan(0.0, subset=["invalid_column"])
+ self.assertIn("Columns not found in DataFrame", str(context.exception))
+
+ def test_fill_nan_with_non_list_subset_raises_error(self):
+ with self.assertRaises(TypeError) as context:
+ self.dataframe.fill_nan(0.0, subset="score")
+ self.assertIn("subset must be a list of strings",
str(context.exception))
+
+
class DataFrameBatchITTests(PyFlinkITTestCase):
def setUp(self):
previous_environment = pf.get_table_environment()
@@ -3764,5 +3943,472 @@ class
DataFrameWindowITTests(PyFlinkStreamDataFrameTestCase):
)
+class DataFrameNullNanITTests(PyFlinkStreamDataFrameTestCase):
+ def test_fill_null_type_compatibility(self):
+ # Create DataFrame with one column per type, each starting null in row
2
+ df = pf.from_records(
+ [
+ {
+ "id": 1,
+ "name": "Alice",
+ "score": 0.95,
+ "active": True,
+ "data": bytearray(b"payload"),
+ "joined": date(2024, 1, 1),
+ "checkin": time(9, 0),
+ },
+ {
+ "id": None,
+ "name": None,
+ "score": None,
+ "active": None,
+ "data": None,
+ "joined": None,
+ "checkin": None,
+ },
+ ],
+ schema=["id", "name", "score", "active", "data", "joined",
"checkin"],
+ )
+
+ # Each value should only fill the column(s) of a compatible type,
+ # leaving type-mismatched columns null.
+ fill_value_to_expected_columns = [
+ (0, {"id", "score"}),
+ ("unknown", {"name"}),
+ (False, {"active"}),
+ (bytearray(b"default"), {"data"}),
+ (date(2000, 1, 1), {"joined"}),
+ (time(0, 0), {"checkin"}),
+ ]
+
+ for value, filled_columns in fill_value_to_expected_columns:
+ rows = df.fill_null(value).collect()
+ for column in df.columns:
+ expected = value if column in filled_columns else None
+ self.assertEqual(rows[1][df.columns.index(column)], expected)
+
+ def test_fill_null_with_timestamp_column(self):
+ # This is separate from test_fill_null_type_compatibility because a
plain
+ # TIMESTAMP column can't be obtained directly from from_records(): it
always
+ # infers datetime.datetime values as TIMESTAMP_LTZ, and collect() on a
+ # TIMESTAMP_LTZ column hits an unrelated, pre-existing bug in the
Java/Python
+ # bridge (FLINK-38418). So we build the column as TIMESTAMP_LTZ and
then cast
+ # it to TIMESTAMP to exercise the TimestampType branch of
+ # _is_type_compatible's type map. Merge back into
+ # test_fill_null_type_compatibility once FLINK-38418 is fixed.
+ # Pin the local timezone so the LTZ -> TIMESTAMP cast doesn't shift the
+ # wall-clock value depending on the machine's system timezone.
+ original_timezone = self.t_env.get_config().get_local_timezone()
+ self.t_env.get_config().set_local_timezone("UTC")
+ try:
+ df = pf.from_records(
+ [
+ {"id": 1, "last_seen": datetime(2024, 1, 1, 9, 0)},
+ {"id": 2, "last_seen": None},
+ ],
+ schema=["id", "last_seen"],
+ )
+ df = df.with_column(
+ "last_seen",
pf.col("last_seen").cast(TableDataTypes.TIMESTAMP(3))
+ )
+
+ result = df.fill_null(datetime(2000, 1, 1)).collect()
+ self.assertEqual(result[0], Row(1, datetime(2024, 1, 1, 9, 0)))
+ self.assertEqual(result[1], Row(2, datetime(2000, 1, 1)))
+ finally:
+ self.t_env.get_config().set_local_timezone(original_timezone)
+
+ def test_drop_null_removes_rows_with_null_values(self):
+ df = pf.from_records(
+ [
+ {"id": 1, "name": "Alice", "age": 30},
+ {"id": 2, "name": None, "age": 25},
+ {"id": 3, "name": "Bob", "age": None},
+ {"id": 4, "name": None, "age": None},
+ ],
+ schema=["id", "name", "age"],
+ )
+
+ result = df.drop_null().collect()
+ self.assertEqual(result, [Row(1, "Alice", 30)])
+
+ def
test_drop_null_with_subset_removes_rows_with_null_in_specified_columns(self):
+ df = pf.from_records(
+ [
+ {"id": 1, "name": "Alice", "age": 30},
+ {"id": 2, "name": None, "age": 25},
+ {"id": 3, "name": "Bob", "age": None},
+ ],
+ schema=["id", "name", "age"],
+ )
+
+ result = df.drop_null(subset=["age"]).collect()
+ self.assertEqual(result, [Row(1, "Alice", 30), Row(2, None, 25)])
+
+ def test_drop_nan_removes_rows_with_nan_values(self):
+ df = pf.from_records(
+ [
+ {"id": 1, "score": 0.95},
+ {"id": 2, "score": float('nan')},
+ {"id": 3, "score": 0.85},
+ ],
+ schema=["id", "score"],
+ )
+
+ result = df.drop_nan().collect()
+ self.assertEqual(len(result), 2)
+ self.assertEqual(result[0][0], 1)
+ self.assertAlmostEqual(result[0][1], 0.95)
+ self.assertEqual(result[1][0], 3)
+ self.assertAlmostEqual(result[1][1], 0.85)
+
+ def
test_drop_nan_with_subset_removes_rows_with_nan_in_specified_columns(self):
+ df = pf.from_records(
+ [
+ {"id": 1, "score": 0.95, "rating": 4.5},
+ {"id": 2, "score": float('nan'), "rating": 3.0},
+ {"id": 3, "score": 0.85, "rating": float('nan')},
+ ],
+ schema=["id", "score", "rating"],
+ )
+
+ result = df.drop_nan(subset=["score"]).collect()
+ self.assertEqual(len(result), 2)
+ self.assertEqual(result[0][0], 1)
+ self.assertAlmostEqual(result[0][1], 0.95)
+ self.assertEqual(result[1][0], 3)
+ self.assertAlmostEqual(result[1][1], 0.85)
+
+ def test_drop_nan_preserves_null_values(self):
+ """Test that drop_nan preserves NULL values (critical bug fix)."""
+ df = pf.from_records(
+ [
+ {"id": 1, "score": 0.95},
+ {"id": 2, "score": None}, # NULL should be preserved
+ {"id": 3, "score": float('nan')}, # NaN should be dropped
+ {"id": 4, "score": 0.85},
+ ],
+ schema=["id", "score"],
+ )
+
+ result = df.drop_nan(subset=["score"]).collect()
+ # Should have 3 rows: id=1, id=2 (NULL preserved), id=4
+ self.assertEqual(len(result), 3)
+ self.assertEqual(result[0][0], 1)
+ self.assertAlmostEqual(result[0][1], 0.95)
+ self.assertEqual(result[1][0], 2)
+ self.assertIsNone(result[1][1]) # NULL preserved
+ self.assertEqual(result[2][0], 4)
+ self.assertAlmostEqual(result[2][1], 0.85)
+
+ def test_drop_nan_with_mixed_schema_auto_filters_to_float_columns(self):
+ """Test that drop_nan with subset=None only checks FLOAT/DOUBLE
columns."""
+ df = pf.from_records(
+ [
+ {"id": 1, "name": "Alice", "score": 0.95, "rating": 4.5},
+ {"id": 2, "name": "Bob", "score": float('nan'), "rating": 3.0},
+ {"id": 3, "name": "Charlie", "score": 0.85, "rating":
float('nan')},
+ ],
+ schema=["id", "name", "score", "rating"],
+ )
+
+ # Should only check score and rating (FLOAT/DOUBLE), not id (INT) or
name (STRING)
+ result = df.drop_nan().collect()
+ # Only row with id=1 has no NaN in float columns
+ self.assertEqual(len(result), 1)
+ self.assertEqual(result[0][0], 1)
+ self.assertEqual(result[0][1], "Alice")
+ self.assertAlmostEqual(result[0][2], 0.95)
+ self.assertAlmostEqual(result[0][3], 4.5)
+
+ def test_drop_nan_with_no_float_columns_returns_unchanged(self):
+ """Test that drop_nan with no FLOAT/DOUBLE columns returns unchanged
DataFrame."""
+ df = pf.from_records(
+ [
+ {"id": 1, "name": "Alice", "active": True},
+ {"id": 2, "name": "Bob", "active": False},
+ ],
+ schema=["id", "name", "active"],
+ )
+
+ # No float columns, should return all rows unchanged
+ result = df.drop_nan().collect()
+ self.assertEqual(len(result), 2)
+ self.assertEqual(result[0], Row(1, "Alice", True))
+ self.assertEqual(result[1], Row(2, "Bob", False))
+
+ def test_drop_nan_with_explicit_subset_ignores_non_float_columns(self):
+ """Test that drop_nan with an explicit subset still only checks
FLOAT/DOUBLE columns."""
+ df = pf.from_records(
+ [
+ {"id": 1, "name": "Alice", "score": 0.95},
+ {"id": 2, "name": "Bob", "score": float('nan')},
+ ],
+ schema=["id", "name", "score"],
+ )
+
+ # "name" is a STRING column and should be silently ignored, not raise
+ # or attempt to apply is_not_nan to it.
+ result = df.drop_nan(subset=["name", "score"]).collect()
+ self.assertEqual(len(result), 1)
+ self.assertEqual(result[0][0], 1)
+ self.assertEqual(result[0][1], "Alice")
+ self.assertAlmostEqual(result[0][2], 0.95)
+
+ def
test_drop_nan_with_explicit_subset_of_only_non_float_columns_returns_unchanged(self):
+ """Test that drop_nan with an explicit subset of only non-float
columns is a no-op."""
+ df = pf.from_records(
+ [
+ {"id": 1, "name": "Alice", "score": float('nan')},
+ {"id": 2, "name": "Bob", "score": 0.85},
+ ],
+ schema=["id", "name", "score"],
+ )
+
+ result = df.drop_nan(subset=["id", "name"]).collect()
+ self.assertEqual(len(result), 2)
+ self.assertEqual(result[0][0], 1)
+ self.assertEqual(result[1][0], 2)
+
+ def test_fill_nan_with_mixed_schema_auto_filters_to_float_columns(self):
+ """Test that fill_nan with subset=None only fills FLOAT/DOUBLE
columns."""
+ df = pf.from_records(
+ [
+ {"id": 1, "name": "Alice", "score": 0.95, "rating": 4.5},
+ {"id": 2, "name": "Bob", "score": float('nan'), "rating": 3.0},
+ {"id": 3, "name": "Charlie", "score": 0.85, "rating":
float('nan')},
+ ],
+ schema=["id", "name", "score", "rating"],
+ )
+
+ # Should only fill score and rating (FLOAT/DOUBLE), not id (INT) or
name (STRING)
+ result = df.fill_nan(0.0).collect()
+ self.assertEqual(len(result), 3)
+ self.assertEqual(result[0][0], 1)
+ self.assertEqual(result[0][1], "Alice")
+ self.assertAlmostEqual(result[0][2], 0.95)
+ self.assertAlmostEqual(result[0][3], 4.5)
+ self.assertEqual(result[1][0], 2)
+ self.assertEqual(result[1][1], "Bob")
+ self.assertAlmostEqual(result[1][2], 0.0) # NaN filled
+ self.assertAlmostEqual(result[1][3], 3.0)
+ self.assertEqual(result[2][0], 3)
+ self.assertEqual(result[2][1], "Charlie")
+ self.assertAlmostEqual(result[2][2], 0.85)
+ self.assertAlmostEqual(result[2][3], 0.0) # NaN filled
+
+ def test_fill_nan_with_no_float_columns_returns_unchanged(self):
+ """Test that fill_nan with no FLOAT/DOUBLE columns returns unchanged
DataFrame."""
+ df = pf.from_records(
+ [
+ {"id": 1, "name": "Alice", "active": True},
+ {"id": 2, "name": "Bob", "active": False},
+ ],
+ schema=["id", "name", "active"],
+ )
+
+ # No float columns, should return all rows unchanged
+ result = df.fill_nan(0.0).collect()
+ self.assertEqual(len(result), 2)
+ self.assertEqual(result[0], Row(1, "Alice", True))
+ self.assertEqual(result[1], Row(2, "Bob", False))
+
+ def test_fill_nan_with_explicit_subset_ignores_non_float_columns(self):
+ """Test that fill_nan with an explicit subset still only fills
FLOAT/DOUBLE columns."""
+ df = pf.from_records(
+ [
+ {"id": 1, "name": "Alice", "score": 0.95},
+ {"id": 2, "name": "Bob", "score": float('nan')},
+ ],
+ schema=["id", "name", "score"],
+ )
+
+ # "name" is a STRING column and should be silently ignored, not raise
+ # or attempt to apply is_nan to it.
+ result = df.fill_nan(0.0, subset=["name", "score"]).collect()
+ self.assertEqual(len(result), 2)
+ self.assertEqual(result[0], Row(1, "Alice", 0.95))
+ self.assertEqual(result[1][0], 2)
+ self.assertEqual(result[1][1], "Bob")
+ self.assertAlmostEqual(result[1][2], 0.0)
+
+ def
test_fill_nan_with_explicit_subset_of_only_non_float_columns_returns_unchanged(self):
+ """Test that fill_nan with an explicit subset of only non-float
columns is a no-op."""
+ df = pf.from_records(
+ [
+ {"id": 1, "name": "Alice", "score": float('nan')},
+ {"id": 2, "name": "Bob", "score": 0.85},
+ ],
+ schema=["id", "name", "score"],
+ )
+
+ result = df.fill_nan(0.0, subset=["id", "name"]).collect()
+ self.assertEqual(len(result), 2)
+ self.assertEqual(result[0][0], 1)
+ self.assertTrue(math.isnan(result[0][2]))
+ self.assertEqual(result[1][0], 2)
+ self.assertAlmostEqual(result[1][2], 0.85)
+
+ def test_fill_null_skips_incompatible_array_column(self):
+ """Test that fill_null with incompatible type (ARRAY) skips the
column."""
+ df = pf.from_records(
+ [
+ {"id": 1, "tags": ["python", "flink"]},
+ {"id": 2, "tags": None},
+ ],
+ schema=["id", "tags"],
+ )
+
+ # Should skip ARRAY column when value is INT
+ result = df.fill_null(0).collect()
+ self.assertEqual(len(result), 2)
+ self.assertEqual(result[0][0], 1)
+ self.assertEqual(result[0][1], ["python", "flink"])
+ self.assertEqual(result[1][0], 2)
+ self.assertIsNone(result[1][1]) # Should remain NULL
+
+ def test_fill_null_skips_incompatible_string_column(self):
+ """Test that fill_null with numeric value skips STRING columns."""
+ df = pf.from_records(
+ [
+ {"id": 1, "name": "Alice"},
+ {"id": 2, "name": None},
+ ],
+ schema=["id", "name"],
+ )
+
+ # Numeric value should skip STRING column
+ result = df.fill_null(0, subset=["name"]).collect()
+ self.assertEqual(len(result), 2)
+ self.assertEqual(result[0], Row(1, "Alice"))
+ self.assertIsNone(result[1][1]) # Should remain NULL
+
+ def test_fill_null_replaces_null_with_specified_value(self):
+ df = pf.from_records(
+ [
+ {"id": 1, "name": "Alice", "quantity": 10},
+ {"id": 2, "name": None, "quantity": None},
+ {"id": 3, "name": "Bob", "quantity": 5},
+ ],
+ schema=["id", "name", "quantity"],
+ )
+
+ result = df.fill_null(0, subset=["quantity"]).collect()
+ self.assertEqual(result, [Row(1, "Alice", 10), Row(2, None, 0), Row(3,
"Bob", 5)])
+
+ def test_fill_null_with_decimal_value(self):
+ """Regression test: the type-compatibility map must recognize Decimal
values,
+ since literals also support Decimal, binary, and temporal types beyond
+ bool/int/float/str."""
+ df = pf.from_records(
+ [
+ {"id": 1, "amount": decimal.Decimal("1.25")},
+ {"id": 2, "amount": None},
+ ],
+ schema=["id", "amount"],
+ )
+
+ result = df.fill_null(decimal.Decimal("0.00"),
subset=["amount"]).collect()
+ self.assertEqual(result[0], Row(1, decimal.Decimal("1.25")))
+ self.assertEqual(result[1], Row(2, decimal.Decimal("0.00")))
+
+ def test_fill_null_with_day_time_interval_value(self):
+ df = pf.from_table(
+ self.t_env.sql_query(
+ "SELECT * FROM (VALUES "
+ "(1, CAST(INTERVAL '1' DAY AS INTERVAL DAY TO SECOND)), "
+ "(2, CAST(NULL AS INTERVAL DAY TO SECOND))) "
+ "AS T(id, duration)"
+ )
+ )
+
+ result = (
+ df.fill_null(timedelta(0), subset=["duration"])
+ .select(
+ "id",
+ is_zero=pf.col("duration") == pf.lit(timedelta(0)),
+ )
+ .collect()
+ )
+ self.assertEqual(
+ result,
+ [Row(1, False), Row(2, True)],
+ )
+
+ def test_fill_null_with_string_value(self):
+ df = pf.from_records(
+ [
+ {"id": 1, "name": "Alice"},
+ {"id": 2, "name": None},
+ ],
+ schema=["id", "name"],
+ )
+
+ result = df.fill_null("unknown", subset=["name"]).collect()
+ self.assertEqual(result, [Row(1, "Alice"), Row(2, "unknown")])
+
+ def test_fill_nan_replaces_nan_with_specified_value(self):
+ df = pf.from_records(
+ [
+ {"id": 1, "score": 0.95},
+ {"id": 2, "score": float('nan')},
+ {"id": 3, "score": 0.85},
+ ],
+ schema=["id", "score"],
+ )
+
+ result = df.fill_nan(0.0, subset=["score"]).collect()
+ self.assertEqual(len(result), 3)
+ self.assertEqual(result[0][0], 1)
+ self.assertAlmostEqual(result[0][1], 0.95)
+ self.assertEqual(result[1][0], 2)
+ self.assertAlmostEqual(result[1][1], 0.0)
+ self.assertEqual(result[2][0], 3)
+ self.assertAlmostEqual(result[2][1], 0.85)
+
+ def test_fill_nan_with_none_value_converts_nan_to_null(self):
+ """Regression test: fill_nan(None) must convert NaN to NULL, per the
fill_nan
+ docstring, rather than raising or leaving the NaN in place."""
+ df = pf.from_records(
+ [
+ {"id": 1, "score": 0.95},
+ {"id": 2, "score": float('nan')},
+ ],
+ schema=["id", "score"],
+ )
+
+ result = df.fill_nan(None, subset=["score"]).collect()
+ self.assertEqual(len(result), 2)
+ self.assertEqual(result[0][0], 1)
+ self.assertAlmostEqual(result[0][1], 0.95)
+ self.assertEqual(result[1][0], 2)
+ self.assertIsNone(result[1][1])
+
+ def test_is_nan_and_is_not_nan_support_decimal_columns(self):
+ """Regression test: IsNanFunction/IsNotNanFunction must dispatch on
DecimalData,
+ not BigDecimal, since BuiltInScalarFunction receives internal
representations
+ and DECIMAL columns never matched an eval(BigDecimal) overload."""
+ df = pf.from_records(
+ [
+ {"id": 1, "amount": decimal.Decimal("1.25")},
+ {"id": 2, "amount": None},
+ ],
+ schema=["id", "amount"],
+ )
+
+ result = df.select(
+ "id",
+ is_nan_result=is_nan(pf.col("amount")),
+ is_not_nan_result=is_not_nan(pf.col("amount")),
+ ).collect()
+
+ self.assertEqual(result[0][0], 1)
+ self.assertEqual(result[0][1], False)
+ self.assertEqual(result[0][2], True)
+ self.assertEqual(result[1][0], 2)
+ self.assertIsNone(result[1][1])
+ self.assertIsNone(result[1][2])
+
+
if __name__ == "__main__":
unittest.main()
diff --git a/flink-python/pyflink/table/expression.py
b/flink-python/pyflink/table/expression.py
index 9c3fc1069dc..2b2ff232467 100644
--- a/flink-python/pyflink/table/expression.py
+++ b/flink-python/pyflink/table/expression.py
@@ -766,6 +766,24 @@ class Expression(Generic[T]):
"""
return _unary_op("isNotFalse")(self)
+ @property
+ def is_nan(self) -> 'Expression[bool]':
+ """
+ Returns true if the given numeric expression is NaN (Not-a-Number).
+
+ .. seealso:: :py:attr:`~Expression.is_not_nan`
+ """
+ return _unary_op("isNan")(self)
+
+ @property
+ def is_not_nan(self) -> 'Expression[bool]':
+ """
+ Returns true if the given numeric expression is not NaN (Not-a-Number).
+
+ .. seealso:: :py:attr:`~Expression.is_nan`
+ """
+ return _unary_op("isNotNan")(self)
+
@property
def distinct(self) -> 'Expression':
"""
diff --git a/flink-python/pyflink/table/expressions.py
b/flink-python/pyflink/table/expressions.py
index 9a6ffc600f2..560d1251596 100644
--- a/flink-python/pyflink/table/expressions.py
+++ b/flink-python/pyflink/table/expressions.py
@@ -26,8 +26,8 @@ from pyflink.table.udf import UserDefinedFunctionWrapper
from pyflink.util.api_stability_decorators import PublicEvolving
from pyflink.util.java_utils import to_jarray, load_java_class
-__all__ = ['if_then_else', 'lit', 'col', 'range_', 'and_', 'or_', 'not_',
'UNBOUNDED_ROW',
- 'UNBOUNDED_RANGE', 'CURRENT_ROW', 'CURRENT_RANGE',
'current_database',
+__all__ = ['if_then_else', 'lit', 'col', 'range_', 'and_', 'or_', 'not_',
'is_nan', 'is_not_nan',
+ 'UNBOUNDED_ROW', 'UNBOUNDED_RANGE', 'CURRENT_ROW', 'CURRENT_RANGE',
'current_database',
'current_date', 'current_time', 'current_timestamp',
'current_watermark', 'local_time', 'local_timestamp',
'temporal_overlaps', 'date_format', 'timestamp_diff', 'array',
'row', 'map_',
@@ -186,6 +186,28 @@ def not_(expression: Expression[bool]) -> Expression[bool]:
return _unary_op("not", expression)
+@PublicEvolving()
+def is_nan(expr) -> Expression[bool]:
+ """
+ Returns true if the given numeric expression is NaN (Not-a-Number).
+
+ This method supports a three-valued logic by preserving NULL. This means
if the
+ input expression is NULL, the result will also be NULL.
+ """
+ return _unary_op("isNan", expr)
+
+
+@PublicEvolving()
+def is_not_nan(expr) -> Expression[bool]:
+ """
+ Returns true if the given numeric expression is not NaN (Not-a-Number).
+
+ This method supports a three-valued logic by preserving NULL. This means
if the
+ input expression is NULL, the result will also be NULL.
+ """
+ return _unary_op("isNotNan", expr)
+
+
"""
Offset constant to be used in the `preceding` clause of unbounded
:class:`~pyflink.table.window.Over`. Use this constant for a time interval.
diff --git
a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/Expressions.java
b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/Expressions.java
index c91441869f5..54bd18b8e71 100644
---
a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/Expressions.java
+++
b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/Expressions.java
@@ -223,6 +223,26 @@ public final class Expressions {
return apiCall(BuiltInFunctionDefinitions.NOT, expression);
}
+ /**
+ * Returns true if the given numeric expression is NaN (Not-a-Number).
+ *
+ * <p>This method supports a three-valued logic by preserving {@code
NULL}. This means if the
+ * input expression is {@code NULL}, the result will also be {@code NULL}.
+ */
+ public static ApiExpression isNan(Object expression) {
+ return apiCall(BuiltInFunctionDefinitions.IS_NAN, expression);
+ }
+
+ /**
+ * Returns true if the given numeric expression is not NaN (Not-a-Number).
+ *
+ * <p>This method supports a three-valued logic by preserving {@code
NULL}. This means if the
+ * input expression is {@code NULL}, the result will also be {@code NULL}.
+ */
+ public static ApiExpression isNotNan(Object expression) {
+ return apiCall(BuiltInFunctionDefinitions.IS_NOT_NAN, expression);
+ }
+
/**
* Offset constant to be used in the {@code preceding} clause of unbounded
{@code Over} windows.
* Use this constant for a time interval. Unbounded over windows start
with the first row of a
diff --git
a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/internal/BaseExpressions.java
b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/internal/BaseExpressions.java
index d4ba8815385..52917c65638 100644
---
a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/internal/BaseExpressions.java
+++
b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/internal/BaseExpressions.java
@@ -133,7 +133,9 @@ import static
org.apache.flink.table.functions.BuiltInFunctionDefinitions.INIT_C
import static
org.apache.flink.table.functions.BuiltInFunctionDefinitions.INSTR;
import static
org.apache.flink.table.functions.BuiltInFunctionDefinitions.IS_FALSE;
import static
org.apache.flink.table.functions.BuiltInFunctionDefinitions.IS_JSON;
+import static
org.apache.flink.table.functions.BuiltInFunctionDefinitions.IS_NAN;
import static
org.apache.flink.table.functions.BuiltInFunctionDefinitions.IS_NOT_FALSE;
+import static
org.apache.flink.table.functions.BuiltInFunctionDefinitions.IS_NOT_NAN;
import static
org.apache.flink.table.functions.BuiltInFunctionDefinitions.IS_NOT_NULL;
import static
org.apache.flink.table.functions.BuiltInFunctionDefinitions.IS_NOT_TRUE;
import static
org.apache.flink.table.functions.BuiltInFunctionDefinitions.IS_NULL;
@@ -536,6 +538,16 @@ public abstract class BaseExpressions<InType, OutType> {
return toApiSpecificExpression(unresolvedCall(IS_NOT_FALSE, toExpr()));
}
+ /** Returns true if the given numeric expression is NaN (Not-a-Number).
Null if null. */
+ public OutType isNan() {
+ return toApiSpecificExpression(unresolvedCall(IS_NAN, toExpr()));
+ }
+
+ /** Returns true if the given numeric expression is not NaN
(Not-a-Number). Null if null. */
+ public OutType isNotNan() {
+ return toApiSpecificExpression(unresolvedCall(IS_NOT_NAN, toExpr()));
+ }
+
/**
* Similar to a SQL distinct aggregation clause such as COUNT(DISTINCT a),
declares that an
* aggregation function is only applied on distinct input values.
diff --git
a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/functions/BuiltInFunctionDefinitions.java
b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/functions/BuiltInFunctionDefinitions.java
index 0bcb4c56649..e2f727092af 100644
---
a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/functions/BuiltInFunctionDefinitions.java
+++
b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/functions/BuiltInFunctionDefinitions.java
@@ -2426,6 +2426,28 @@ public final class BuiltInFunctionDefinitions {
.build();
//
--------------------------------------------------------------------------------------------
+
+ public static final BuiltInFunctionDefinition IS_NAN =
+ BuiltInFunctionDefinition.newBuilder()
+ .name("isNan")
+ .sqlName("IS_NAN")
+ .kind(SCALAR)
+
.inputTypeStrategy(sequence(logical(LogicalTypeFamily.NUMERIC)))
+
.outputTypeStrategy(nullableIfArgs(explicit(DataTypes.BOOLEAN())))
+
.runtimeClass("org.apache.flink.table.runtime.functions.scalar.IsNanFunction")
+ .build();
+
+ public static final BuiltInFunctionDefinition IS_NOT_NAN =
+ BuiltInFunctionDefinition.newBuilder()
+ .name("isNotNan")
+ .sqlName("IS_NOT_NAN")
+ .kind(SCALAR)
+
.inputTypeStrategy(sequence(logical(LogicalTypeFamily.NUMERIC)))
+
.outputTypeStrategy(nullableIfArgs(explicit(DataTypes.BOOLEAN())))
+ .runtimeClass(
+
"org.apache.flink.table.runtime.functions.scalar.IsNotNanFunction")
+ .build();
+
// Catalog functions
//
--------------------------------------------------------------------------------------------
diff --git
a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/scalar/IsNanFunction.java
b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/scalar/IsNanFunction.java
new file mode 100644
index 00000000000..7c4bda3ba61
--- /dev/null
+++
b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/scalar/IsNanFunction.java
@@ -0,0 +1,63 @@
+/*
+ * 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.data.DecimalData;
+import org.apache.flink.table.functions.BuiltInFunctionDefinitions;
+import org.apache.flink.table.functions.SpecializedFunction.SpecializedContext;
+
+import javax.annotation.Nullable;
+
+/** 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 DecimalData value) {
+ return value == null ? null : false;
+ }
+}
diff --git
a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/scalar/IsNotNanFunction.java
b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/scalar/IsNotNanFunction.java
new file mode 100644
index 00000000000..5514c5cecf3
--- /dev/null
+++
b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/scalar/IsNotNanFunction.java
@@ -0,0 +1,63 @@
+/*
+ * 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.data.DecimalData;
+import org.apache.flink.table.functions.BuiltInFunctionDefinitions;
+import org.apache.flink.table.functions.SpecializedFunction.SpecializedContext;
+
+import javax.annotation.Nullable;
+
+/** Implementation of {@link BuiltInFunctionDefinitions#IS_NOT_NAN}. */
+@Internal
+public final class IsNotNanFunction extends BuiltInScalarFunction {
+
+ public IsNotNanFunction(SpecializedContext context) {
+ super(BuiltInFunctionDefinitions.IS_NOT_NAN, context);
+ }
+
+ public @Nullable Boolean eval(final @Nullable Byte value) {
+ return value == null ? null : true;
+ }
+
+ public @Nullable Boolean eval(final @Nullable Short value) {
+ return value == null ? null : true;
+ }
+
+ public @Nullable Boolean eval(final @Nullable Integer value) {
+ return value == null ? null : true;
+ }
+
+ public @Nullable Boolean eval(final @Nullable Long value) {
+ return value == null ? null : true;
+ }
+
+ 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 DecimalData value) {
+ return value == null ? null : true;
+ }
+}
diff --git a/tools/maven/suppressions.xml b/tools/maven/suppressions.xml
index 957d80bf068..fb8cf9dc83e 100644
--- a/tools/maven/suppressions.xml
+++ b/tools/maven/suppressions.xml
@@ -41,6 +41,7 @@ under the License.
<suppress files="NFAITCase.java" checks="FileLength"/>
<suppress files="HyperLogLogPlusPlus.java" checks="FileLength"/>
<suppress files="BuiltInFunctionDefinitions.java"
checks="FileLength"/>
+ <suppress files="BaseExpressions.java" checks="FileLength"/>
<suppress files="ProcessTableFunctionTestHarnessTest.java"
checks="FileLength"/>
<!-- Legacy mockito usages -->
