diff --git a/flink-python/pyflink/dataframe/dataframe.py b/flink-python/pyflink/dataframe/dataframe.py index 66c2020ee4db7..2bf6f9a9423dc 100644 --- a/flink-python/pyflink/dataframe/dataframe.py +++ b/flink-python/pyflink/dataframe/dataframe.py @@ -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"] @@ -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()) diff --git a/flink-python/pyflink/dataframe/tests/test_dataframe.py b/flink-python/pyflink/dataframe/tests/test_dataframe.py index 98714e9fbb6a4..e294038c59dff 100644 --- a/flink-python/pyflink/dataframe/tests/test_dataframe.py +++ b/flink-python/pyflink/dataframe/tests/test_dataframe.py @@ -16,6 +16,9 @@ # limitations under the License. ################################################################################ +import array +import datetime +import decimal import unittest from typing import NamedTuple @@ -401,30 +404,142 @@ 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(), ], ) @@ -432,12 +547,33 @@ 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): @@ -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): @@ -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( { diff --git a/flink-python/pyflink/table/expressions.py b/flink-python/pyflink/table/expressions.py index b7c48954ba78d..ea00237696e04 100644 --- a/flink-python/pyflink/table/expressions.py +++ b/flink-python/pyflink/table/expressions.py @@ -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 @@ -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() diff --git a/flink-python/pyflink/table/literal.py b/flink-python/pyflink/table/literal.py new file mode 100644 index 0000000000000..e3dba8545470e --- /dev/null +++ b/flink-python/pyflink/table/literal.py @@ -0,0 +1,219 @@ +################################################################################ +# 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. +################################################################################ + +import calendar +import datetime +import time +from array import array + +from pyflink.common import Row +from pyflink.java_gateway import get_gateway +from pyflink.table.types import ( + _array_type_mappings, + _to_java_data_type, + ArrayType, + DataType, + DateType, + DayTimeIntervalType, + LocalZonedTimestampType, + MapType, + MultisetType, + RowType, + TimeType, + TimestampType, + ZonedTimestampType, +) +from pyflink.util.api_stability_decorators import Internal + + +@Internal() +def _to_java_literal_value(value, data_type: DataType = None): + """Converts Python-only literal values into objects accepted by Py4J.""" + if data_type is None: + return _to_java_inferred_literal_value(value) + return _to_java_typed_literal_value(value, data_type) + + +def _to_java_inferred_literal_value(value): + if value is None: + return value + + gateway = get_gateway() + jvm = gateway.jvm + if isinstance(value, datetime.datetime): + return _to_java_typed_literal_value(value, TimestampType()) + elif isinstance(value, datetime.date): + return _to_java_typed_literal_value(value, DateType()) + elif isinstance(value, datetime.time): + return _to_java_typed_literal_value(value, TimeType()) + elif isinstance(value, datetime.timedelta): + return _to_java_typed_literal_value( + value, + DayTimeIntervalType(DayTimeIntervalType.DayTimeResolution.DAY_TO_SECOND), + ) + elif isinstance(value, array): + if value.typecode not in _array_type_mappings: + raise TypeError(f"not supported type: array({value.typecode})") + element_data_type = _to_java_data_type(_array_type_mappings[value.typecode]) + j_array = jvm.java.lang.reflect.Array.newInstance( + element_data_type.getConversionClass(), len(value) + ) + for pos, element in enumerate(value): + j_array[pos] = element + return j_array + elif isinstance(value, (list, tuple)): + j_values = jvm.java.util.ArrayList() + for element in value: + j_values.add(_to_java_inferred_literal_value(element)) + return j_values + elif isinstance(value, Row): + return _to_java_row(value) + return value + + +def _to_java_typed_literal_value(value, data_type: DataType): + if value is None or data_type._conversion_cls: + return value + + jvm = get_gateway().jvm + if isinstance(data_type, DateType) and isinstance(value, datetime.datetime): + value = value.date() + if isinstance(data_type, DateType) and isinstance(value, datetime.date): + return jvm.java.time.LocalDate.of(value.year, value.month, value.day) + elif isinstance(data_type, TimeType) and isinstance(value, datetime.time): + return jvm.java.time.LocalTime.of( + value.hour, value.minute, value.second, value.microsecond * 1000 + ) + elif isinstance(data_type, TimestampType) and isinstance(value, datetime.datetime): + return jvm.java.time.LocalDateTime.of( + value.year, + value.month, + value.day, + value.hour, + value.minute, + value.second, + value.microsecond * 1000, + ) + elif isinstance(data_type, LocalZonedTimestampType) and isinstance( + value, datetime.datetime + ): + seconds = ( + calendar.timegm(value.utctimetuple()) + if value.tzinfo + else int(time.mktime(value.timetuple())) + ) + return jvm.java.time.Instant.ofEpochSecond(seconds, value.microsecond * 1000) + elif isinstance(data_type, ZonedTimestampType) and isinstance(value, datetime.datetime): + if value.tzinfo is None: + value = value.astimezone() + offset = value.utcoffset() + if offset is None or offset.microseconds != 0: + return value + j_offset = jvm.java.time.ZoneOffset.ofTotalSeconds(int(offset.total_seconds())) + return jvm.java.time.OffsetDateTime.of( + value.year, + value.month, + value.day, + value.hour, + value.minute, + value.second, + value.microsecond * 1000, + j_offset, + ) + elif isinstance(data_type, DayTimeIntervalType) and isinstance( + value, datetime.timedelta + ): + seconds = value.days * 86400 + value.seconds + return jvm.java.time.Duration.ofSeconds(seconds, value.microseconds * 1000) + elif isinstance(data_type, ArrayType) and isinstance(value, (list, tuple, array)): + j_values = jvm.java.util.ArrayList() + for element in value: + j_values.add(_to_java_typed_literal_value(element, data_type.element_type)) + return j_values + elif isinstance(data_type, MultisetType) and isinstance(value, dict): + j_values = jvm.java.util.HashMap() + for element, count in value.items(): + j_values.put( + _to_java_typed_literal_value(element, data_type.element_type), count + ) + return j_values + elif isinstance(data_type, MapType) and isinstance(value, dict): + j_values = jvm.java.util.HashMap() + for key, map_value in value.items(): + j_values.put( + _to_java_typed_literal_value(key, data_type.key_type), + _to_java_typed_literal_value(map_value, data_type.value_type), + ) + return j_values + elif isinstance(data_type, RowType): + if isinstance(value, Row): + return _to_java_row(value, data_type) + elif isinstance(value, dict): + j_values = jvm.java.util.HashMap() + for field in data_type.fields: + j_values.put( + field.name, + _to_java_typed_literal_value(value.get(field.name), field.data_type), + ) + return j_values + elif isinstance(value, (list, tuple)): + j_values = jvm.java.util.ArrayList() + for pos, field_value in enumerate(value): + if pos < len(data_type.fields): + field_value = _to_java_typed_literal_value( + field_value, data_type.fields[pos].data_type + ) + else: + field_value = _to_java_inferred_literal_value(field_value) + j_values.add(field_value) + return j_values + return value + + +def _to_java_row(value: Row, data_type: RowType = None): + jvm = get_gateway().jvm + if hasattr(value, "_fields"): + j_row = jvm.org.apache.flink.types.Row.withNames(value.get_row_kind().to_j_row_kind()) + field_names = ( + value._fields + if data_type is None + else [field.name for field in data_type.fields] + ) + for pos, field_name in enumerate(field_names): + field_value = value[field_name] + if data_type is not None: + field_value = _to_java_typed_literal_value( + field_value, data_type.fields[pos].data_type + ) + else: + field_value = _to_java_inferred_literal_value(field_value) + j_row.setField(field_name, field_value) + return j_row + + j_row = jvm.org.apache.flink.types.Row.withPositions( + value.get_row_kind().to_j_row_kind(), len(value) + ) + for pos, field_value in enumerate(value): + if data_type is not None and pos < len(data_type.fields): + field_value = _to_java_typed_literal_value( + field_value, data_type.fields[pos].data_type + ) + else: + field_value = _to_java_inferred_literal_value(field_value) + j_row.setField(pos, field_value) + return j_row diff --git a/flink-python/pyflink/table/tests/test_expression.py b/flink-python/pyflink/table/tests/test_expression.py index 2b957e7aea662..f56484d92cc8d 100644 --- a/flink-python/pyflink/table/tests/test_expression.py +++ b/flink-python/pyflink/table/tests/test_expression.py @@ -15,8 +15,11 @@ # See the License for the specific language governing permissions and # limitations under the License. ################################################################################ +import datetime import unittest +from py4j.protocol import Py4JJavaError + from pyflink.table import DataTypes from pyflink.table.expression import TimeIntervalUnit, TimePointUnit, JsonExistsOnError, \ JsonValueOnEmptyOrError, JsonType, JsonQueryWrapper, JsonQueryOnEmptyOrError @@ -378,6 +381,55 @@ def test_expressions(self): self.assertEqual('withColumns(a, b, c)', str(with_columns(expr1, expr2, expr3))) self.assertEqual('a.b.c(a)', str(call('a.b.c', expr1))) + def test_lit_converts_python_values_to_declared_data_types(self): + test_cases = [ + (1, DataTypes.TINYINT().not_null(), "TINYINT"), + (1, DataTypes.SMALLINT().not_null(), "SMALLINT"), + (1, DataTypes.BIGINT().not_null(), "BIGINT"), + (1.25, DataTypes.FLOAT().not_null(), "FLOAT"), + (datetime.date(2026, 8, 3), DataTypes.DATE().not_null(), "DATE"), + (datetime.time(1, 2, 3, 4000), DataTypes.TIME(6).not_null(), + "TIME_WITHOUT_TIME_ZONE"), + (datetime.datetime(2026, 8, 3, 1, 2, 3, 4000), + DataTypes.TIMESTAMP(6).not_null(), "TIMESTAMP_WITHOUT_TIME_ZONE"), + (datetime.datetime(2026, 8, 3, 1, 2, 3, 4000, datetime.timezone.utc), + DataTypes.TIMESTAMP_LTZ(6).not_null(), "TIMESTAMP_WITH_LOCAL_TIME_ZONE"), + (datetime.timedelta(days=1, seconds=2, microseconds=3000), + DataTypes.INTERVAL(DataTypes.DAY(), DataTypes.SECOND(6)).not_null(), + "INTERVAL_DAY_TIME"), + ] + + for value, data_type, expected_type_root in test_cases: + with self.subTest(value=value, data_type=data_type): + literal = lit(value, data_type)._j_expr.toExpr() + actual_type_root = literal.getOutputDataType() \ + .getLogicalType().getTypeRoot().name() + self.assertEqual(expected_type_root, actual_type_root) + + def test_lit_converts_python_values_for_inferred_data_types(self): + test_cases = [ + (datetime.date(2026, 8, 3), "DATE NOT NULL"), + (datetime.time(1, 2, 3, 4000), "TIME(3) NOT NULL"), + (datetime.datetime(2026, 8, 3, 1, 2, 3, 4000), + "TIMESTAMP(3) NOT NULL"), + (datetime.timedelta(days=1, seconds=2, microseconds=3000), + "INTERVAL DAY(1) TO SECOND(3) NOT NULL"), + ] + + for value, expected_data_type in test_cases: + with self.subTest(value=value): + literal = lit(value)._j_expr.toExpr() + self.assertEqual(expected_data_type, str(literal.getOutputDataType())) + + def test_lit_rejects_out_of_range_integer_values(self): + for value, data_type in [ + (128, DataTypes.TINYINT().not_null()), + (32768, DataTypes.SMALLINT().not_null()), + ]: + with self.subTest(value=value, data_type=data_type): + with self.assertRaises(Py4JJavaError): + lit(value, data_type) + if __name__ == "__main__": try: diff --git a/flink-python/pyflink/table/tests/test_literal.py b/flink-python/pyflink/table/tests/test_literal.py new file mode 100644 index 0000000000000..04145e1f5ddf2 --- /dev/null +++ b/flink-python/pyflink/table/tests/test_literal.py @@ -0,0 +1,231 @@ +################################################################################ +# 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. +################################################################################ + +import array +import datetime +import decimal + +from py4j.protocol import Py4JJavaError + +from pyflink.common import Row +from pyflink.table import DataTypes +from pyflink.table.expressions import lit +from pyflink.table.types import _array_type_mappings +from pyflink.testing.test_case_utils import PyFlinkBatchTableTestCase + + +class LiteralITTests(PyFlinkBatchTableTestCase): + def test_scalar_literals_can_be_executed(self): + source = self.t_env.from_elements([(1,)], ["id"]) + + result = source.select( + lit(True), + lit(2), + lit(1.25), + lit("x"), + lit(b"x"), + lit(bytearray(b"y")), + lit(decimal.Decimal("1.25")), + lit(datetime.date(2026, 8, 3)), + lit(datetime.time(1, 2, 3, 4000)), + lit(datetime.datetime(2026, 8, 3, 1, 2, 3, 4000)), + lit(datetime.timedelta(days=1, seconds=2, microseconds=3000)).is_not_null, + lit(1, DataTypes.TINYINT().not_null()), + lit(1, DataTypes.SMALLINT().not_null()), + lit(1, DataTypes.BIGINT().not_null()), + lit(1.25, DataTypes.FLOAT().not_null()), + lit( + datetime.datetime(2026, 8, 3, 1, 2, 3, 4000), + DataTypes.DATE().not_null(), + ), + lit( + datetime.time(1, 2, 3, 4000), + DataTypes.TIME(6).not_null(), + ), + lit( + datetime.datetime(2026, 8, 3, 1, 2, 3, 4000), + DataTypes.TIMESTAMP(6).not_null(), + ), + lit( + datetime.datetime( + 2026, + 8, + 3, + 1, + 2, + 3, + 4000, + datetime.timezone.utc, + ), + DataTypes.TIMESTAMP_LTZ(6).not_null(), + ).is_not_null, + lit( + datetime.timedelta(days=1, seconds=2, microseconds=3000), + DataTypes.INTERVAL(DataTypes.DAY(), DataTypes.SECOND(6)).not_null(), + ).is_not_null, + lit( + 14, + DataTypes.INTERVAL(DataTypes.YEAR(), DataTypes.MONTH()).not_null(), + ).is_not_null, + lit(None, DataTypes.ARRAY(DataTypes.SMALLINT())), + lit(None, DataTypes.MAP(DataTypes.SMALLINT(), DataTypes.FLOAT())), + lit( + None, + DataTypes.ROW( + [DataTypes.FIELD("small_value", DataTypes.SMALLINT())] + ), + ), + lit(None, DataTypes.MULTISET(DataTypes.SMALLINT())), + ) + + self.assertEqual( + list(result.execute().collect()), + [ + Row( + True, + 2, + 1.25, + "x", + b"x", + b"y", + decimal.Decimal("1.25"), + datetime.date(2026, 8, 3), + datetime.time(1, 2, 3, 4000), + datetime.datetime(2026, 8, 3, 1, 2, 3, 4000), + True, + 1, + 1, + 1, + 1.25, + datetime.date(2026, 8, 3), + datetime.time(1, 2, 3, 4000), + datetime.datetime(2026, 8, 3, 1, 2, 3, 4000), + True, + True, + True, + None, + None, + None, + None, + ) + ], + ) + + def test_constructed_literals_can_be_executed(self): + source = self.t_env.from_elements([(1,)], ["id"]) + row_type = DataTypes.ROW( + [ + DataTypes.FIELD("small_value", DataTypes.SMALLINT()), + DataTypes.FIELD("float_value", DataTypes.FLOAT()), + ] + ).not_null() + map_type = DataTypes.MAP( + DataTypes.SMALLINT(), + DataTypes.FLOAT(), + ).not_null() + nested_type = DataTypes.ARRAY( + DataTypes.ROW( + [ + DataTypes.FIELD("values", DataTypes.ARRAY(DataTypes.SMALLINT())), + DataTypes.FIELD("mapping", map_type), + ] + ) + ).not_null() + + result = source.select( + lit(["abc"]), + lit([[datetime.date(2026, 8, 3)]]), + lit((1, 2)), + lit([1, 2], DataTypes.ARRAY(DataTypes.SMALLINT()).not_null()), + lit((1, 1.25), row_type), + lit({1: 1.25}, map_type), + lit([([1, 2], {3: 1.25})], nested_type), + lit( + [], + DataTypes.ARRAY(DataTypes.SMALLINT().not_null()).not_null(), + ), + lit({}, map_type), + lit(array.array("h")), + *(lit(array.array(typecode, [1, 2])) for typecode in "bhilfd"), + ) + + self.assertIsInstance(result.explain(), str) + self.assertEqual( + list(result.execute().collect()), + [ + Row( + ["abc"], + [[datetime.date(2026, 8, 3)]], + [1, 2], + [1, 2], + Row(1, 1.25), + {1: 1.25}, + [Row([1, 2], {3: 1.25})], + [], + {}, + [], + [1, 2], + [1, 2], + [1, 2], + [1, 2], + [1.0, 2.0], + [1.0, 2.0], + ) + ], + ) + + def test_unsupported_constructed_literals_are_rejected(self): + with self.assertRaisesRegex(Py4JJavaError, "Non-null MULTISET literals are not supported"): + lit({1: 2}, DataTypes.MULTISET(DataTypes.SMALLINT()).not_null()) + + with self.assertRaisesRegex(Py4JJavaError, "Non-null empty ROW literals are not supported"): + lit((), DataTypes.ROW([]).not_null()) + + with self.assertRaises(Py4JJavaError): + lit([1.25], DataTypes.ARRAY(DataTypes.SMALLINT()).not_null()) + + with self.assertRaisesRegex(Py4JJavaError, "ROW literal has arity 2"): + lit( + (1, 2), + DataTypes.ROW([DataTypes.FIELD("value", DataTypes.INT())]).not_null(), + ) + + def test_empty_python_arrays_preserve_numeric_typecodes(self): + typecodes = sorted(set(_array_type_mappings) - {"u"}) + source = self.t_env.from_elements([(1,)], ["id"]) + result = source.select(*(lit(array.array(typecode)) for typecode in typecodes)) + + expected_types = [ + DataTypes.ARRAY(_array_type_mappings[typecode]).not_null() + for typecode in typecodes + ] + self.assertEqual(result.get_resolved_schema().get_column_data_types(), expected_types) + self.assertEqual(list(result.execute().collect()), [Row(*([[]] * len(typecodes)))]) + + def test_unicode_python_array_can_be_executed(self): + if "u" not in _array_type_mappings: + self.skipTest("Unicode arrays are not supported on this Python version") + + source = self.t_env.from_elements([(1,)], ["id"]) + result = source.select(lit(array.array("u", "ab"))) + + self.assertEqual( + result.get_resolved_schema().get_column_data_types(), + [DataTypes.ARRAY(DataTypes.CHAR(1)).not_null()], + ) + self.assertEqual(list(result.execute().collect()), [Row(["a", "b"])]) diff --git a/flink-python/src/main/java/org/apache/flink/table/utils/python/PythonTableUtils.java b/flink-python/src/main/java/org/apache/flink/table/utils/python/PythonTableUtils.java index 01dfab186dddc..512125e41f5c4 100644 --- a/flink-python/src/main/java/org/apache/flink/table/utils/python/PythonTableUtils.java +++ b/flink-python/src/main/java/org/apache/flink/table/utils/python/PythonTableUtils.java @@ -21,10 +21,14 @@ import org.apache.flink.annotation.Internal; import org.apache.flink.api.common.io.InputFormat; import org.apache.flink.streaming.api.legacy.io.CollectionInputFormat; +import org.apache.flink.table.api.ApiExpression; +import org.apache.flink.table.api.DataTypes; +import org.apache.flink.table.api.Expressions; import org.apache.flink.table.api.Schema; import org.apache.flink.table.api.Table; import org.apache.flink.table.api.TableDescriptor; import org.apache.flink.table.api.TableEnvironment; +import org.apache.flink.table.api.ValidationException; import org.apache.flink.table.api.dataview.ListView; import org.apache.flink.table.api.dataview.MapView; import org.apache.flink.table.data.DecimalData; @@ -34,6 +38,7 @@ import org.apache.flink.table.data.RowData; import org.apache.flink.table.data.StringData; import org.apache.flink.table.data.TimestampData; +import org.apache.flink.table.expressions.ValueLiteralExpression; import org.apache.flink.table.runtime.typeutils.InternalSerializers; import org.apache.flink.table.types.DataType; import org.apache.flink.table.types.logical.ArrayType; @@ -62,8 +67,11 @@ import org.apache.flink.table.types.logical.VarCharType; import org.apache.flink.table.types.logical.YearMonthIntervalType; import org.apache.flink.table.types.logical.ZonedTimestampType; +import org.apache.flink.types.Row; import org.apache.flink.types.RowKind; +import javax.annotation.Nullable; + import java.lang.reflect.Array; import java.math.BigDecimal; import java.nio.charset.StandardCharsets; @@ -71,6 +79,7 @@ import java.time.LocalDate; import java.time.LocalDateTime; import java.time.LocalTime; +import java.time.Period; import java.time.ZoneId; import java.util.Arrays; import java.util.Collection; @@ -127,6 +136,214 @@ public static Table createTableFromElement( dataCollection, InternalSerializers.create(dataType.getLogicalType())); } + /** + * Creates a literal from a value received through Py4J. + * + *

Py4J represents Python numeric values as {@link Integer}, {@link Long}, or {@link Double}, + * which does not preserve the boxed Java classes required by some {@link DataType}s. This + * method adapts the value to the data type's external representation and creates the literal in + * the same JVM call so that the adapted value is not converted by Py4J again. If {@code + * dataType} is absent, Java literal inference remains authoritative. Constructed values are + * represented by constructor expressions because raw constructed value literals cannot be + * planned. + * + * @param value the literal value received through Py4J + * @param dataType the declared data type, or {@code null} for type inference + * @return the literal expression + * @throws ValidationException if the constructed value has no plannable literal expression + */ + public static ApiExpression createLiteral( + final Object value, @Nullable final DataType dataType) { + if (dataType != null) { + return createTypedLiteral(value, dataType); + } + + final Object inferredValue = materializeInferredArrays(value); + final ApiExpression literal = Expressions.lit(inferredValue); + final DataType inferredDataType = + ((ValueLiteralExpression) literal.toExpr()).getOutputDataType(); + // Raw array literals can be inferred but not planned. Rebuild the array as a constructor + // expression while preserving the data type inferred by Java. + if (inferredDataType.getLogicalType() instanceof ArrayType) { + return createTypedLiteral(inferredValue, inferredDataType); + } + return literal; + } + + private static ApiExpression createTypedLiteral(final Object value, final DataType dataType) { + if (value == null) { + // A typed null carries no composite payload and is directly plannable. + return Expressions.lit(value, dataType); + } + if (dataType.getLogicalType().isNullable()) { + // Delegate the invalid non-null value/nullable type combination to Java validation. + return Expressions.lit(value, dataType); + } + if (!usesDefaultLiteralConversion(dataType)) { + // Custom conversion classes are opaque to this bridge; use native literal handling. + return Expressions.lit(value, dataType); + } + + if (dataType.getLogicalType() instanceof ArrayType) { + if (!(value instanceof List) && !value.getClass().isArray()) { + // Delegate incompatible ARRAY representations to standard literal validation. + return Expressions.lit(value, dataType); + } + final int length = getLiteralArrayLength(value); + if (length == 0) { + return createEmptyArray(dataType); + } + final DataType elementDataType = dataType.getChildren().get(0); + final Object[] tail = new Object[length - 1]; + for (int pos = 1; pos < length; pos++) { + tail[pos - 1] = + createNestedLiteral(getLiteralArrayElement(value, pos), elementDataType); + } + return Expressions.array( + createNestedLiteral(getLiteralArrayElement(value, 0), elementDataType), + tail) + .cast(dataType); + } + if (dataType.getLogicalType() instanceof RowType) { + if (!isLiteralRow(value)) { + // Delegate incompatible ROW representations to standard literal validation. + return Expressions.lit(value, dataType); + } + final RowType rowType = (RowType) dataType.getLogicalType(); + final List fieldDataTypes = dataType.getChildren(); + if (fieldDataTypes.isEmpty()) { + throw new ValidationException("Non-null empty ROW literals are not supported."); + } + if (!(value instanceof Map)) { + final int valueArity = getLiteralRowArity(value); + if (valueArity != fieldDataTypes.size()) { + throw new ValidationException( + String.format( + "ROW literal has arity %d but the data type has arity %d.", + valueArity, fieldDataTypes.size())); + } + } + final List fieldNames = rowType.getFieldNames(); + final Object[] tail = new Object[fieldDataTypes.size() - 1]; + for (int pos = 1; pos < fieldDataTypes.size(); pos++) { + tail[pos - 1] = + createNestedLiteral( + getLiteralRowField(value, pos, fieldNames.get(pos)), + fieldDataTypes.get(pos)); + } + return Expressions.row( + createNestedLiteral( + getLiteralRowField(value, 0, fieldNames.get(0)), + fieldDataTypes.get(0)), + tail) + .cast(dataType); + } + if (dataType.getLogicalType() instanceof MultisetType) { + throw new ValidationException("Non-null MULTISET literals are not supported."); + } + if (dataType.getLogicalType() instanceof MapType) { + if (!(value instanceof Map)) { + // Delegate incompatible MAP representations to standard literal validation. + return Expressions.lit(value, dataType); + } + final Map map = (Map) value; + final DataType keyDataType = dataType.getChildren().get(0); + final DataType valueDataType = dataType.getChildren().get(1); + if (map.isEmpty()) { + return Expressions.mapFromArrays( + createEmptyArray(DataTypes.ARRAY(keyDataType).notNull()), + createEmptyArray(DataTypes.ARRAY(valueDataType).notNull())) + .cast(dataType); + } + final Object[] arguments = new Object[map.size() * 2]; + int pos = 0; + for (final Map.Entry entry : map.entrySet()) { + arguments[pos++] = createNestedLiteral(entry.getKey(), keyDataType); + arguments[pos++] = createNestedLiteral(entry.getValue(), valueDataType); + } + return Expressions.map( + arguments[0], + arguments[1], + Arrays.copyOfRange(arguments, 2, arguments.length)) + .cast(dataType); + } + // After normalizing Py4J numerics, let Java perform final scalar validation. + return Expressions.lit(scalarLiteralConverter(dataType).apply(value), dataType); + } + + private static ApiExpression createEmptyArray(final DataType dataType) { + final DataType elementDataType = dataType.getChildren().get(0); + // The ARRAY constructor requires an argument, so slice a typed one-element array to empty. + return Expressions.array(Expressions.lit(null, elementDataType.nullable())) + .arraySlice(2, 1) + .cast(dataType); + } + + private static ApiExpression createNestedLiteral( + final Object value, final DataType declaredDataType) { + // Java requires non-null literals to use a NOT NULL type. Preserve declared nullability for + // null values so that invalid nulls remain rejected. + final DataType literalDataType = + value == null ? declaredDataType : declaredDataType.notNull(); + return createTypedLiteral(value, literalDataType); + } + + private static Object materializeInferredArrays(final Object value) { + if (!(value instanceof List)) { + return value; + } + + final List values = (List) value; + final Object[] convertedValues = new Object[values.size()]; + Class componentClass = null; + boolean hasCommonComponentClass = true; + for (int pos = 0; pos < values.size(); pos++) { + final Object convertedValue = materializeInferredArrays(values.get(pos)); + convertedValues[pos] = convertedValue; + if (convertedValue == null) { + continue; + } + if (componentClass == null) { + componentClass = convertedValue.getClass(); + } else if (componentClass != convertedValue.getClass()) { + hasCommonComponentClass = false; + } + } + + if (!hasCommonComponentClass || componentClass == null) { + return convertedValues; + } + final Object convertedArray = Array.newInstance(componentClass, convertedValues.length); + for (int pos = 0; pos < convertedValues.length; pos++) { + Array.set(convertedArray, pos, convertedValues[pos]); + } + return convertedArray; + } + + private static int getLiteralArrayLength(final Object value) { + return value instanceof List ? ((List) value).size() : Array.getLength(value); + } + + private static Object getLiteralArrayElement(final Object value, final int pos) { + return value instanceof List ? ((List) value).get(pos) : Array.get(value, pos); + } + + private static boolean isLiteralRow(final Object value) { + return value instanceof Row + || value instanceof List + || value instanceof Map + || value.getClass().isArray(); + } + + private static int getLiteralRowArity(final Object value) { + if (value instanceof Row) { + return ((Row) value).getArity(); + } else if (value instanceof List) { + return ((List) value).size(); + } + return Array.getLength(value); + } + private static BiFunction, Object> arrayConstructor( final LogicalType elementType) { if (elementType instanceof BooleanType) { @@ -487,6 +704,87 @@ private static Function converter(LogicalType logicalType) { throw new IllegalStateException("Failed to get converter for LogicalType: " + logicalType); } + private static Function scalarLiteralConverter(final DataType dataType) { + if (!usesDefaultLiteralConversion(dataType)) { + return Function.identity(); + } + + final LogicalType logicalType = dataType.getLogicalType(); + if (logicalType instanceof TinyIntType) { + return value -> { + if (!isIntegral(value)) { + return value; + } + final long number = ((Number) value).longValue(); + return number >= Byte.MIN_VALUE && number <= Byte.MAX_VALUE ? (byte) number : value; + }; + } + if (logicalType instanceof SmallIntType) { + return value -> { + if (!isIntegral(value)) { + return value; + } + final long number = ((Number) value).longValue(); + return number >= Short.MIN_VALUE && number <= Short.MAX_VALUE + ? (short) number + : value; + }; + } + if (logicalType instanceof BigIntType) { + return value -> isIntegral(value) ? ((Number) value).longValue() : value; + } + if (logicalType instanceof FloatType) { + return value -> value instanceof Double ? ((Double) value).floatValue() : value; + } + if (logicalType instanceof YearMonthIntervalType) { + return value -> { + if (!isIntegral(value)) { + return value; + } + final long number = ((Number) value).longValue(); + return number >= Integer.MIN_VALUE && number <= Integer.MAX_VALUE + ? Period.ofMonths((int) number) + : value; + }; + } + return Function.identity(); + } + + private static boolean usesDefaultLiteralConversion(final DataType dataType) { + final LogicalType logicalType = dataType.getLogicalType(); + final Class conversionClass = dataType.getConversionClass(); + if (logicalType instanceof ArrayType) { + return conversionClass.isArray(); + } else if (logicalType instanceof MapType || logicalType instanceof MultisetType) { + return Map.class.isAssignableFrom(conversionClass); + } else if (logicalType instanceof RowType) { + return conversionClass == Row.class; + } + return conversionClass == logicalType.getDefaultConversion(); + } + + private static boolean isIntegral(final Object value) { + return value instanceof Byte + || value instanceof Short + || value instanceof Integer + || value instanceof Long; + } + + private static Object getLiteralRowField( + final Object value, final int pos, final String fieldName) { + if (value instanceof Row) { + final Row row = (Row) value; + return row.getFieldNames(false) == null ? row.getField(pos) : row.getField(fieldName); + } else if (value instanceof List) { + return ((List) value).get(pos); + } else if (value instanceof Map) { + return ((Map) value).get(fieldName); + } else if (value != null && value.getClass().isArray()) { + return Array.get(value, pos); + } + return value; + } + private static int getOffsetFromLocalMillis(final long millisLocal) { TimeZone localZone = TimeZone.getDefault(); int result = localZone.getRawOffset(); diff --git a/flink-python/src/test/java/org/apache/flink/table/utils/python/PythonTableUtilsTest.java b/flink-python/src/test/java/org/apache/flink/table/utils/python/PythonTableUtilsTest.java new file mode 100644 index 0000000000000..29af59bcdb247 --- /dev/null +++ b/flink-python/src/test/java/org/apache/flink/table/utils/python/PythonTableUtilsTest.java @@ -0,0 +1,116 @@ +/* + * 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.utils.python; + +import org.apache.flink.table.api.DataTypes; +import org.apache.flink.table.api.ValidationException; +import org.apache.flink.table.expressions.ValueLiteralExpression; +import org.apache.flink.table.types.DataType; + +import org.junit.jupiter.api.Test; + +import java.time.Period; +import java.util.Arrays; +import java.util.Collections; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; + +class PythonTableUtilsTest { + + @Test + void testCreateLiteralConvertsPy4JNumbers() { + assertThat(literalValue(1, DataTypes.TINYINT().notNull())).isInstanceOf(Byte.class); + assertThat(literalValue(1, DataTypes.SMALLINT().notNull())).isInstanceOf(Short.class); + assertThat(literalValue(1, DataTypes.BIGINT().notNull())).isInstanceOf(Long.class); + assertThat(literalValue(1.25, DataTypes.FLOAT().notNull())).isInstanceOf(Float.class); + assertThat( + literalValue( + 14, + DataTypes.INTERVAL(DataTypes.YEAR(), DataTypes.MONTH()).notNull())) + .isEqualTo(Period.ofMonths(14)); + } + + @Test + void testCreateLiteralRejectsIncompatibleNestedValue() { + assertThatThrownBy( + () -> + PythonTableUtils.createLiteral( + Collections.singletonList(1.25), + DataTypes.ARRAY(DataTypes.SMALLINT()).notNull())) + .isInstanceOf(ValidationException.class); + } + + @Test + void testCreateLiteralRejectsNonNullMultiset() { + assertThatThrownBy( + () -> + PythonTableUtils.createLiteral( + Collections.singletonMap(1, 2), + DataTypes.MULTISET(DataTypes.SMALLINT()).notNull())) + .isInstanceOf(ValidationException.class) + .hasMessage("Non-null MULTISET literals are not supported."); + } + + @Test + void testCreateLiteralSupportsTypedNull() { + final DataType dataType = DataTypes.ARRAY(DataTypes.SMALLINT()); + final ValueLiteralExpression literal = + (ValueLiteralExpression) PythonTableUtils.createLiteral(null, dataType).toExpr(); + + assertThat(literal.getOutputDataType()).isEqualTo(dataType); + assertThat(literal.getValueAs(Object.class)).isEmpty(); + } + + @Test + void testCreateLiteralUsesJavaArrayInferenceRules() { + assertThatThrownBy(() -> PythonTableUtils.createLiteral(Collections.emptyList(), null)) + .isInstanceOf(ValidationException.class); + assertThatThrownBy(() -> PythonTableUtils.createLiteral(Arrays.asList("a", "bb"), null)) + .isInstanceOf(ValidationException.class); + } + + @Test + void testCreateLiteralRejectsEmptyRow() { + assertThatThrownBy( + () -> + PythonTableUtils.createLiteral( + Collections.emptyList(), DataTypes.ROW().notNull())) + .isInstanceOf(ValidationException.class) + .hasMessage("Non-null empty ROW literals are not supported."); + } + + @Test + void testCreateLiteralRejectsWrongRowArity() { + assertThatThrownBy( + () -> + PythonTableUtils.createLiteral( + Arrays.asList(1, 2), + DataTypes.ROW(DataTypes.FIELD("value", DataTypes.INT())) + .notNull())) + .isInstanceOf(ValidationException.class) + .hasMessage("ROW literal has arity 2 but the data type has arity 1."); + } + + private static Object literalValue(final Object value, final DataType dataType) { + final ValueLiteralExpression literal = + (ValueLiteralExpression) PythonTableUtils.createLiteral(value, dataType).toExpr(); + return literal.getValueAs(Object.class).orElseThrow(AssertionError::new); + } +}