Skip to content

[FLINK-40300][python] Fix literal expression in Python Table API #28924

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
auroflow wants to merge 5 commits into
base: master
Choose a base branch
from
Open

[FLINK-40300][python] Fix literal expression in Python Table API #28924

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
26740da
[FLINK-40300][python] Fix typed literal conversion in PyFlink
auroflow Aug 3, 2026
13f1ee1
[FLINK-40300][python] Refactor typed literal converter
auroflow Aug 5, 2026
92f2ef2
[FLINK-40300][python] Support multiset literal conversion
auroflow Aug 5, 2026
56bc4c5
[FLINK-40300][python] Support inferred literal conversion in PyFlink
auroflow Aug 5, 2026
7256a9c
[FLINK-40300][python] Align PyFlink literals with Java Table API
auroflow Aug 6, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
11 changes: 0 additions & 11 deletions flink-python/pyflink/dataframe/dataframe.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -28,7 +28,6 @@
lit as table_lit,
)
from pyflink.table.table import Table
from pyflink.table.types import DataTypes as TableDataTypes
from pyflink.util.api_stability_decorators import PublicEvolving

__all__ = ["DataFrame", "col", "lit"]
Expand Down Expand Up @@ -81,16 +80,6 @@ def lit(value: Any, data_type: Optional[DataType] = None) -> Expression:
table_data_type = data_type._to_table_data_type()
if value is None:
return table_lit(value, table_data_type)
if (
table_data_type.nullable() == TableDataTypes.BIGINT()
and isinstance(value, int)
and not isinstance(value, bool)
and -(1 << 31) <= value < (1 << 31)
):
# Py4J sends Python integers in this range as java.lang.Integer, but a typed BIGINT
# literal requires java.lang.Long. Match BIGINT independently of its nullability, then
# cast a typed INT literal to the originally declared BIGINT type.
return table_lit(value, TableDataTypes.INT().not_null()).cast(table_data_type)
return table_lit(value, table_data_type.not_null())


Expand Down
201 changes: 188 additions & 13 deletions flink-python/pyflink/dataframe/tests/test_dataframe.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -16,6 +16,9 @@
# limitations under the License.
################################################################################

import array
import datetime
import decimal
import unittest
from typing import NamedTuple

Expand Down Expand Up @@ -401,43 +404,176 @@ def setUp(self):
super().setUp()
self.dataframe = pf.from_records([(1,)], schema=["id"])

def test_lit_supports_inferred_and_explicit_types(self):
def test_lit_infers_supported_python_types(self):
literal_values = {
"inferred_bool": True,
"inferred_int": 2,
"inferred_bigint": 1 << 40,
"inferred_float": 1.25,
"inferred_string": "x",
"inferred_bytes": b"x",
"inferred_bytearray": bytearray(b"x"),
"inferred_decimal": decimal.Decimal("1.25"),
"inferred_date": datetime.date(2026, 8, 3),
"inferred_time": datetime.time(1, 2, 3),
"inferred_timestamp": datetime.datetime(2026, 8, 3, 1, 2, 3),
"inferred_aware_timestamp": datetime.datetime(
2026, 8, 3, 1, 2, 3, tzinfo=datetime.timezone.utc
),
"inferred_timedelta": datetime.timedelta(days=1, seconds=2, microseconds=3000),
"inferred_list": ["abc"],
"inferred_nested_list": [[datetime.date(2026, 8, 3)]],
"inferred_tuple": (1, 2),
"inferred_array": array.array("h", [1, 2]),
}
result = self.dataframe.select(
inferred_int=pf.lit(2),
inferred_string=pf.lit("x"),
explicit_int=pf.lit(3, pf.DataType.int64()),
explicit_large_int=pf.lit(1 << 40, pf.DataType.int64()),
**{name: pf.lit(value) for name, value in literal_values.items()}
)

self.assert_dataframe_schema(
result,
list(literal_values),
[
TableDataTypes.BOOLEAN().not_null(),
TableDataTypes.INT().not_null(),
TableDataTypes.BIGINT().not_null(),
TableDataTypes.DOUBLE().not_null(),
TableDataTypes.CHAR(1).not_null(),
TableDataTypes.BINARY(1).not_null(),
TableDataTypes.BINARY(1).not_null(),
TableDataTypes.DECIMAL(3, 2).not_null(),
TableDataTypes.DATE().not_null(),
TableDataTypes.TIME().not_null(),
TableDataTypes.TIMESTAMP(0).not_null(),
TableDataTypes.TIMESTAMP(0).not_null(),
TableDataTypes.INTERVAL(
TableDataTypes.DAY(1), TableDataTypes.SECOND(3)
),
TableDataTypes.ARRAY(TableDataTypes.CHAR(3)).not_null(),
TableDataTypes.ARRAY(
TableDataTypes.ARRAY(TableDataTypes.DATE())
).not_null(),
TableDataTypes.ARRAY(TableDataTypes.INT()).not_null(),
TableDataTypes.ARRAY(TableDataTypes.SMALLINT()).not_null(),
],
)

def test_lit_supports_explicit_types(self):
list_type = pf.DataType.list(pf.DataType.int16())
map_type = pf.DataType.map(pf.DataType.int16(), pf.DataType.float32())
struct_type = pf.DataType.struct(
{
"small_value": pf.DataType.int16(),
"float_value": pf.DataType.float32(),
}
)
result = self.dataframe.select(
explicit_int8=pf.lit(3, pf.DataType.int8()),
explicit_int16=pf.lit(3, pf.DataType.int16()),
explicit_int32=pf.lit(3, pf.DataType.int32()),
explicit_int64=pf.lit(3, pf.DataType.int64()),
explicit_float32=pf.lit(1.25, pf.DataType.float32()),
explicit_float64=pf.lit(1.25, pf.DataType.float64()),
explicit_decimal=pf.lit(decimal.Decimal("1.25"), pf.DataType.decimal(3, 2)),
explicit_bool=pf.lit(True, pf.DataType.bool()),
explicit_string=pf.lit("y", pf.DataType.string()),
explicit_fixed_string=pf.lit("y", pf.DataType.fixed_size_string(1)),
explicit_binary=pf.lit(b"y", pf.DataType.binary()),
explicit_fixed_binary=pf.lit(b"y", pf.DataType.fixed_size_binary(1)),
explicit_date=pf.lit(datetime.date(2026, 8, 3), pf.DataType.date()),
explicit_time=pf.lit(datetime.time(1, 2, 3, 4000), pf.DataType.time(6)),
explicit_timestamp=pf.lit(
datetime.datetime(2026, 8, 3, 1, 2, 3, 4000),
pf.DataType.timestamp(6),
),
explicit_timestamp_ltz=pf.lit(
datetime.datetime(
2026, 8, 3, 1, 2, 3, 4000, tzinfo=datetime.timezone.utc
),
pf.DataType.timestamp_ltz(6),
),
explicit_list=pf.lit([1, 2], list_type),
explicit_map=pf.lit({1: 1.25}, map_type),
explicit_struct=pf.lit((1, 1.25), struct_type),
)

self.assert_dataframe_schema(
result,
[
"inferred_int",
"inferred_string",
"explicit_int",
"explicit_large_int",
"explicit_int8",
"explicit_int16",
"explicit_int32",
"explicit_int64",
"explicit_float32",
"explicit_float64",
"explicit_decimal",
"explicit_bool",
"explicit_string",
"explicit_fixed_string",
"explicit_binary",
"explicit_fixed_binary",
"explicit_date",
"explicit_time",
"explicit_timestamp",
"explicit_timestamp_ltz",
"explicit_list",
"explicit_map",
"explicit_struct",
],
[
TableDataTypes.TINYINT().not_null(),
TableDataTypes.SMALLINT().not_null(),
TableDataTypes.INT().not_null(),
TableDataTypes.CHAR(1).not_null(),
TableDataTypes.BIGINT().not_null(),
TableDataTypes.BIGINT().not_null(),
TableDataTypes.FLOAT().not_null(),
TableDataTypes.DOUBLE().not_null(),
TableDataTypes.DECIMAL(3, 2).not_null(),
TableDataTypes.BOOLEAN().not_null(),
TableDataTypes.STRING().not_null(),
TableDataTypes.CHAR(1).not_null(),
TableDataTypes.BYTES().not_null(),
TableDataTypes.BINARY(1).not_null(),
TableDataTypes.DATE().not_null(),
TableDataTypes.TIME(6).not_null(),
TableDataTypes.TIMESTAMP(6).not_null(),
TableDataTypes.TIMESTAMP_LTZ(6).not_null(),
list_type._to_table_data_type().not_null(),
map_type._to_table_data_type().not_null(),
struct_type._to_table_data_type().not_null(),
],
)

def test_lit_supports_explicitly_typed_nulls(self):
result = self.dataframe.select(
null_int=pf.lit(None, pf.DataType.int64()),
null_string=pf.lit(None, pf.DataType.string()),
null_list=pf.lit(None, pf.DataType.list(pf.DataType.int16())),
null_map=pf.lit(
None, pf.DataType.map(pf.DataType.int16(), pf.DataType.float32())
),
null_struct=pf.lit(
None, pf.DataType.struct({"value": pf.DataType.int16()})
),
)

self.assert_dataframe_schema(
result,
["null_int", "null_string"],
[TableDataTypes.BIGINT(), TableDataTypes.STRING()],
[
"null_int",
"null_string",
"null_list",
"null_map",
"null_struct",
],
[
TableDataTypes.BIGINT(),
TableDataTypes.STRING(),
TableDataTypes.ARRAY(TableDataTypes.SMALLINT()),
TableDataTypes.MAP(TableDataTypes.SMALLINT(), TableDataTypes.FLOAT()),
TableDataTypes.ROW(
[TableDataTypes.FIELD("value", TableDataTypes.SMALLINT())]
),
],
)

def test_lit_supports_small_int_for_non_nullable_bigint(self):
Expand All @@ -455,6 +591,7 @@ def test_lit_rejects_values_incompatible_with_explicit_type(self):
(3.14, pf.DataType.int64()),
("abc", pf.DataType.int64()),
(42, pf.DataType.string()),
([1.25], pf.DataType.list(pf.DataType.int16())),
]
for value, data_type in incompatible_values:
with self.subTest(value=value, data_type=data_type):
Expand All @@ -480,6 +617,44 @@ def test_from_records(self):
[Row(1, "Alice"), Row(2, "Bob")],
)

def test_lit_supports_inferred_and_explicit_types(self):
dataframe = pf.from_records([(1,)], schema=["id"])
map_type = pf.DataType.map(pf.DataType.int16(), pf.DataType.float32())
struct_type = pf.DataType.struct(
{
"small_value": pf.DataType.int16(),
"float_value": pf.DataType.float32(),
}
)

result = dataframe.select(
inferred_date=pf.lit(datetime.date(2026, 8, 3)),
inferred_list=pf.lit(["abc"]),
explicit_small_int=pf.lit(1, pf.DataType.int16()),
explicit_float=pf.lit(1.25, pf.DataType.float32()),
explicit_list=pf.lit(
[1, 2],
pf.DataType.list(pf.DataType.int16()),
),
explicit_map=pf.lit({1: 1.25}, map_type),
explicit_struct=pf.lit((1, 1.25), struct_type),
)

self.assertEqual(
result.collect(),
[
Row(
datetime.date(2026, 8, 3),
["abc"],
1,
1.25,
[1, 2],
{1: 1.25},
Row(1, 1.25),
)
],
)

def test_basic_functionality(self):
df = pf.from_dict(
{
Expand Down
13 changes: 9 additions & 4 deletions flink-python/pyflink/table/expressions.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -20,6 +20,7 @@
from pyflink import add_version_doc
from pyflink.java_gateway import get_gateway
from pyflink.table.expression import Expression, _get_java_expression, TimePointUnit, JsonOnNull
from pyflink.table.literal import _to_java_literal_value
from pyflink.table.types import _to_java_data_type, DataType
from pyflink.table.udf import UserDefinedFunctionWrapper
from pyflink.util.api_stability_decorators import PublicEvolving
Expand Down Expand Up @@ -115,10 +116,14 @@ def lit(v, data_type: DataType = None) -> Expression:

>>> tab.select(col("key"), lit("abc"))
"""
if data_type is None:
return _unary_op("lit", v)
else:
return _binary_op("lit", v, _to_java_data_type(data_type))
_j_literal_value = _to_java_literal_value(v, data_type)
gateway = get_gateway()
j_data_type = _to_java_data_type(data_type) if data_type is not None else None
return Expression(
gateway.jvm.org.apache.flink.table.utils.python.PythonTableUtils.createLiteral(
_j_literal_value, j_data_type
)
)


@PublicEvolving()
Expand Down
Loading