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 b810155feca [FLINK-40300][python] Fix literal expression in Python
Table API (#28924)
b810155feca is described below
commit b810155feca84317d9cbfa1c1db8335b70d3c0d1
Author: Liu Liu <[email protected]>
AuthorDate: Tue Aug 18 19:16:56 2026 +0800
[FLINK-40300][python] Fix literal expression in Python Table API (#28924)
---
flink-python/pyflink/dataframe/dataframe.py | 11 -
.../pyflink/dataframe/tests/test_dataframe.py | 235 ++++++++-
flink-python/pyflink/table/expressions.py | 13 +-
flink-python/pyflink/table/literal.py | 234 +++++++++
.../pyflink/table/tests/test_expression.py | 77 +++
flink-python/pyflink/table/tests/test_literal.py | 536 +++++++++++++++++++++
.../flink/table/utils/python/PythonTableUtils.java | 494 ++++++++++++++++++-
.../table/utils/python/PythonTableUtilsTest.java | 325 +++++++++++++
8 files changed, 1895 insertions(+), 30 deletions(-)
diff --git a/flink-python/pyflink/dataframe/dataframe.py
b/flink-python/pyflink/dataframe/dataframe.py
index 35f03af35f8..793f074f8da 100644
--- a/flink-python/pyflink/dataframe/dataframe.py
+++ b/flink-python/pyflink/dataframe/dataframe.py
@@ -31,7 +31,6 @@ from pyflink.table.expressions import (
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", "GroupedDataFrame", "col", "lit"]
@@ -84,16 +83,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 a3d095f2ab8..0e6651e44d7 100644
--- a/flink-python/pyflink/dataframe/tests/test_dataframe.py
+++ b/flink-python/pyflink/dataframe/tests/test_dataframe.py
@@ -16,8 +16,10 @@
# limitations under the License.
################################################################################
+import array
+import decimal
import unittest
-from datetime import datetime, timezone
+from datetime import date, datetime, time, timedelta, timezone
from typing import NamedTuple
import pandas as pd
@@ -630,30 +632,142 @@ class DataFrameLiteralTests(PyFlinkDataFrameUTTestCase):
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": date(2026, 8, 3),
+ "inferred_time": time(1, 2, 3),
+ "inferred_timestamp": datetime(2026, 8, 3, 1, 2, 3),
+ "inferred_aware_timestamp": datetime(
+ 2026, 8, 3, 1, 2, 3, tzinfo=timezone.utc
+ ),
+ "inferred_timedelta": timedelta(days=1, seconds=2,
microseconds=3000),
+ "inferred_list": ["abc"],
+ "inferred_nested_list": [[date(2026, 8, 3)]],
+ "inferred_tuple": (1, 2),
+ "inferred_array": array.array("h", [1, 2]),
+ }
+ result = self.dataframe.select(
+ **{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(
- 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()),
+ 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(date(2026, 8, 3), pf.DataType.date()),
+ explicit_time=pf.lit(time(1, 2, 3, 4000), pf.DataType.time(6)),
+ explicit_timestamp=pf.lit(
+ datetime(2026, 8, 3, 1, 2, 3, 4000),
+ pf.DataType.timestamp(6),
+ ),
+ explicit_timestamp_ltz=pf.lit(
+ datetime(
+ 2026, 8, 3, 1, 2, 3, 4000, tzinfo=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(),
],
)
@@ -661,12 +775,33 @@ class DataFrameLiteralTests(PyFlinkDataFrameUTTestCase):
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):
@@ -684,6 +819,7 @@ class DataFrameLiteralTests(PyFlinkDataFrameUTTestCase):
(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):
@@ -801,6 +937,39 @@ class DataFrameITTests(PyFlinkStreamDataFrameTestCase):
[Row(1, "Alice"), Row(2, "Bob")],
)
+ def
test_watermark_precision_normalization_floors_pre_epoch_timestamps(self):
+ original_timezone = self.t_env.get_config().get_local_timezone()
+ self.t_env.get_config().set_local_timezone("UTC")
+ try:
+ timestamp = datetime(1969, 12, 31, 23, 59, 59, 999999)
+ creators = [
+ (
+ "from_dict",
+ lambda: pf.from_dict(
+ {"ts": [timestamp]},
+ watermark=("ts", "ts - INTERVAL '1' SECOND"),
+ ),
+ ),
+ (
+ "from_records",
+ lambda: pf.from_records(
+ [{"ts": timestamp}],
+ watermark=("ts", "ts - INTERVAL '1' SECOND"),
+ ),
+ ),
+ ]
+
+ for name, creator in creators:
+ with self.subTest(creator=name):
+ result = creator().select(
+ ts=pf.col("ts").cast(TableDataTypes.STRING())
+ )
+ self.assertEqual(
+ result.collect(), [Row("1969-12-31 23:59:59.999")]
+ )
+ finally:
+ self.t_env.get_config().set_local_timezone(original_timezone)
+
def test_pandas_to_pandas_round_trip(self):
original_timezone = self.t_env.get_config().get_local_timezone()
self.t_env.get_config().set_local_timezone("America/New_York")
@@ -839,6 +1008,44 @@ class DataFrameITTests(PyFlinkStreamDataFrameTestCase):
finally:
self.t_env.get_config().set_local_timezone(original_timezone)
+ 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(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(
+ 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 b7c48954ba7..42f3d587499 100644
--- a/flink-python/pyflink/table/expressions.py
+++ b/flink-python/pyflink/table/expressions.py
@@ -20,6 +20,7 @@ from typing import Union
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))
+ gateway = get_gateway()
+ j_data_type = _to_java_data_type(data_type) if data_type is not None else
None
+ _j_literal_value = _to_java_literal_value(v, data_type)
+ 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 00000000000..87b88b7465e
--- /dev/null
+++ b/flink-python/pyflink/table/literal.py
@@ -0,0 +1,234 @@
+################################################################################
+# 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 datetime
+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,
+)
+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
+ if not value:
+ array_data_type =
ArrayType(_array_type_mappings[value.typecode]).not_null()
+ return (
+ jvm.org.apache.flink.table.utils.python.PythonTableUtils
+ .createInferredArrayValue(j_array,
_to_java_data_type(array_data_type))
+ )
+ 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_instant(value, jvm):
+ utc_value = value.astimezone(datetime.timezone.utc)
+ epoch = datetime.datetime(1970, 1, 1, tzinfo=datetime.timezone.utc)
+ delta = utc_value - epoch
+ # Avoid floating-point timestamp conversion for pre-epoch and subsecond
values.
+ seconds = delta.days * 86400 + delta.seconds
+ return jvm.java.time.Instant.ofEpochSecond(seconds, utc_value.microsecond
* 1000)
+
+
+def _to_java_local_datetime(value, jvm):
+ return jvm.java.time.LocalDateTime.ofInstant(
+ _to_java_instant(value, jvm), jvm.java.time.ZoneId.systemDefault()
+ )
+
+
+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):
+ if value.utcoffset() is not None:
+ # TIME has no date. Use the Unix epoch date to match PyFlink's
existing transport
+ # conversion when rendering an offset time in the client JVM's
local time zone.
+ date_time = datetime.datetime.combine(datetime.date(1970, 1, 1),
value)
+ return _to_java_local_datetime(date_time, jvm).toLocalTime()
+ 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):
+ if value.utcoffset() is not None:
+ # Match PyFlink's TIMESTAMP transport conversion by rendering the
instant in the
+ # client JVM's local time zone before dropping the zone
information.
+ return _to_java_local_datetime(value, jvm)
+ 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
+ ):
+ if value.utcoffset() is None:
+ value = value.astimezone()
+ return _to_java_instant(value, jvm)
+ 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)
+ if data_type is not None:
+ # Keep undeclared fields so Java validates the original Row arity.
+ for field_name in value._fields:
+ if field_name not in field_names:
+ j_row.setField(
+ field_name,
+ _to_java_inferred_literal_value(value[field_name]),
+ )
+ 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 2b957e7aea6..f50fb6309ce 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,80 @@ class PyFlinkBatchExpressionTests(PyFlinkTestCase):
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.datetime(2026, 8, 3, 1, 2, 3, 4000,
datetime.timezone.utc),
+ "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_unsupported_data_type_before_value_conversion(self):
+ with self.assertRaisesRegex(TypeError, "Unsupported data type: INT"):
+ lit(1, "INT")
+
+ def test_lit_rejects_binary_scalar_as_constructed_value(self):
+ array_type = DataTypes.ARRAY(DataTypes.TINYINT()).not_null()
+ row_type = DataTypes.ROW(
+ [
+ DataTypes.FIELD("a", DataTypes.TINYINT()),
+ DataTypes.FIELD("b", DataTypes.TINYINT()),
+ ]
+ ).not_null()
+ for value in [b"a", bytearray(b"a")]:
+ with self.subTest(value=value):
+ with self.assertRaises(Py4JJavaError):
+ lit([value, []])
+ with self.assertRaises(Py4JJavaError):
+ lit([[value], [[]]])
+ with self.assertRaises(Py4JJavaError):
+ lit(value, array_type)
+ with self.assertRaises(Py4JJavaError):
+ lit(value, row_type)
+
+ 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 00000000000..ccfea78df57
--- /dev/null
+++ b/flink-python/pyflink/table/tests/test_literal.py
@@ -0,0 +1,536 @@
+################################################################################
+# 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, RowKind
+from pyflink.java_gateway import get_gateway
+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
+
+
+# Keep the fold test independent of the system's IANA time zone database.
+class _FoldAwareTimezone(datetime.tzinfo):
+ def utcoffset(self, value):
+ return datetime.timedelta(hours=-4 if value.fold == 0 else -5)
+
+ def dst(self, value):
+ return datetime.timedelta(hours=1 if value.fold == 0 else 0)
+
+
+class LiteralITCase(PyFlinkBatchTableTestCase):
+ def test_timezone_aware_datetime_literals(self):
+ source = self.t_env.from_elements([(1,)], ["id"])
+ same_instant_with_different_offsets = (
+ datetime.datetime(
+ 2026, 8, 3, 12,
tzinfo=datetime.timezone(datetime.timedelta(hours=8))
+ ),
+ datetime.datetime(2026, 8, 3, 4, tzinfo=datetime.timezone.utc),
+ )
+ fractional_offset_instant = (
+ datetime.datetime(
+ 1970,
+ 1,
+ 1,
+ 0,
+ 0,
+ 0,
+ 100000,
+
tzinfo=datetime.timezone(datetime.timedelta(microseconds=500000)),
+ ),
+ datetime.datetime(
+ 1969, 12, 31, 23, 59, 59, 600000, tzinfo=datetime.timezone.utc
+ ),
+ )
+ ambiguous_local_time = datetime.datetime(
+ 2026, 11, 1, 1, 30, tzinfo=_FoldAwareTimezone()
+ )
+
+ result = source.select(
+ lit(same_instant_with_different_offsets[0])
+ == lit(same_instant_with_different_offsets[1]),
+ lit(
+ same_instant_with_different_offsets[0],
+ DataTypes.TIMESTAMP(6).not_null(),
+ )
+ == lit(
+ same_instant_with_different_offsets[1],
+ DataTypes.TIMESTAMP(6).not_null(),
+ ),
+ lit(fractional_offset_instant[0]) ==
lit(fractional_offset_instant[1]),
+ lit(
+ fractional_offset_instant[0],
+ DataTypes.TIMESTAMP_LTZ(6).not_null(),
+ )
+ == lit(
+ fractional_offset_instant[1],
+ DataTypes.TIMESTAMP_LTZ(6).not_null(),
+ ),
+ lit(
+ ambiguous_local_time.replace(fold=0),
+ DataTypes.TIMESTAMP_LTZ(6).not_null(),
+ )
+ == lit(
+ datetime.datetime(2026, 11, 1, 5, 30,
tzinfo=datetime.timezone.utc),
+ DataTypes.TIMESTAMP_LTZ(6).not_null(),
+ ),
+ lit(
+ ambiguous_local_time.replace(fold=1),
+ DataTypes.TIMESTAMP_LTZ(6).not_null(),
+ )
+ == lit(
+ datetime.datetime(2026, 11, 1, 6, 30,
tzinfo=datetime.timezone.utc),
+ DataTypes.TIMESTAMP_LTZ(6).not_null(),
+ ),
+ lit(
+ ambiguous_local_time.replace(fold=0),
+ DataTypes.TIMESTAMP_LTZ(6).not_null(),
+ )
+ != lit(
+ ambiguous_local_time.replace(fold=1),
+ DataTypes.TIMESTAMP_LTZ(6).not_null(),
+ ),
+ )
+
+ self.assertEqual(
+ list(result.execute().collect()),
+ [Row(True, True, True, True, True, True, True)],
+ )
+
+ def test_aware_temporal_literals_use_client_timezone(self):
+ jvm = get_gateway().jvm
+ original_timezone = jvm.java.util.TimeZone.getDefault()
+ try:
+ jvm.java.util.TimeZone.setDefault(
+ jvm.java.util.TimeZone.getTimeZone("Asia/Shanghai")
+ )
+ source = self.t_env.from_elements([(1,)], ["id"])
+ aware_time = datetime.time(4, tzinfo=datetime.timezone.utc)
+ aware_timestamp = datetime.datetime(
+ 2026, 8, 3, 4, tzinfo=datetime.timezone.utc
+ )
+ inferred_time = lit(aware_time)
+ explicit_time = lit(aware_time, DataTypes.TIME().not_null())
+ inferred_timestamp = lit(aware_timestamp)
+ explicit_timestamp = lit(
+ aware_timestamp, DataTypes.TIMESTAMP().not_null()
+ )
+
+ local_time_class =
jvm.java.lang.Class.forName("java.time.LocalTime")
+ local_datetime_class = jvm.java.lang.Class.forName(
+ "java.time.LocalDateTime"
+ )
+ expected_time = jvm.java.time.LocalTime.of(12, 0)
+ expected_timestamp = jvm.java.time.LocalDateTime.of(2026, 8, 3,
12, 0)
+ for expression in (inferred_time, explicit_time):
+ literal = expression._j_expr.toExpr()
+ self.assertEqual(expected_time,
literal.getValueAs(local_time_class).get())
+ for expression in (inferred_timestamp, explicit_timestamp):
+ literal = expression._j_expr.toExpr()
+ self.assertEqual(
+ expected_timestamp,
+ literal.getValueAs(local_datetime_class).get(),
+ )
+
+ result = source.select(
+ inferred_time == explicit_time,
+ inferred_timestamp == explicit_timestamp,
+ )
+ self.assertEqual(
+ list(result.execute().collect()),
+ [Row(True, True)],
+ )
+ finally:
+ jvm.java.util.TimeZone.setDefault(original_timezone)
+
+ 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(b"x", DataTypes.BINARY(1).not_null()),
+ lit(bytearray(b"y"), DataTypes.BINARY(1).not_null()),
+ 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",
+ 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(
+ [b"x"],
+ DataTypes.ARRAY(DataTypes.BINARY(1)).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],
+ [b"x"],
+ 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_inferred_nested_arrays_use_sibling_types(self):
+ source = self.t_env.from_elements([(1,)], ["id"])
+ empty_inner_array = lit([[1], []])
+ null_only_inner_array = lit([[1], [None]])
+ char_empty_inner_array = lit([["a"], []])
+ char_null_only_inner_array = lit([["a"], [None]])
+ decimal_empty_inner_array = lit([[decimal.Decimal("1.20")], []])
+ time_empty_inner_array = lit([[datetime.time(12, 0, 0, 123000)], []])
+ binary_empty_inner_array = lit([[b"a"], []])
+ nested_array_type =
DataTypes.ARRAY(DataTypes.ARRAY(DataTypes.INT())).not_null()
+
+ literal_table = source.select(
+ empty_inner_array,
+ null_only_inner_array,
+ char_empty_inner_array,
+ char_null_only_inner_array,
+ decimal_empty_inner_array,
+ time_empty_inner_array,
+ binary_empty_inner_array,
+ )
+ self.assertEqual(
+ literal_table.get_resolved_schema().get_column_data_types(),
+ [
+ nested_array_type,
+ nested_array_type,
+ DataTypes.ARRAY(DataTypes.ARRAY(DataTypes.CHAR(1))).not_null(),
+ DataTypes.ARRAY(DataTypes.ARRAY(DataTypes.CHAR(1))).not_null(),
+ DataTypes.ARRAY(DataTypes.ARRAY(DataTypes.DECIMAL(3,
2))).not_null(),
+ DataTypes.ARRAY(DataTypes.ARRAY(DataTypes.TIME(3))).not_null(),
+
DataTypes.ARRAY(DataTypes.ARRAY(DataTypes.BINARY(1))).not_null(),
+ ],
+ )
+ self.assertIsInstance(literal_table.explain(), str)
+
+ result = source.select(
+ empty_inner_array.cardinality,
+ empty_inner_array.at(1).cardinality,
+ empty_inner_array.at(2).cardinality,
+ empty_inner_array.at(1).at(1),
+ null_only_inner_array.at(2).cardinality,
+ null_only_inner_array.at(2).at(1).is_null,
+ char_empty_inner_array.at(1).at(1),
+ char_empty_inner_array.at(2).cardinality,
+ char_null_only_inner_array.at(2).at(1).is_null,
+ decimal_empty_inner_array.at(1).at(1),
+ decimal_empty_inner_array.at(2).cardinality,
+ time_empty_inner_array.at(1).at(1),
+ time_empty_inner_array.at(2).cardinality,
+ binary_empty_inner_array.at(1).at(1),
+ binary_empty_inner_array.at(2).cardinality,
+ )
+ self.assertEqual(
+ list(result.execute().collect()),
+ [
+ Row(
+ 2,
+ 1,
+ 0,
+ 1,
+ 1,
+ True,
+ "a",
+ 0,
+ True,
+ decimal.Decimal("1.20"),
+ 0,
+ datetime.time(12, 0, 0, 123000),
+ 0,
+ b"a",
+ 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(),
+ )
+
+ with self.assertRaisesRegex(Py4JJavaError, "ROW literal has arity 2"):
+ lit(
+ Row(a=1, b=2),
+ DataTypes.ROW([DataTypes.FIELD("a",
DataTypes.INT())]).not_null(),
+ )
+
+ with self.assertRaisesRegex(Py4JJavaError, "Unsupported kind
'DELETE'"):
+ lit(
+ Row.of_kind(RowKind.DELETE, 1),
+ DataTypes.ROW([DataTypes.FIELD("value",
DataTypes.INT())]).not_null(),
+ )
+
+ def test_primitive_java_arrays_can_be_executed(self):
+ gateway = get_gateway()
+ jvm = gateway.jvm
+ primitive_values = [
+ (jvm.boolean, True, DataTypes.BOOLEAN().not_null()),
+ (jvm.short, 1, DataTypes.SMALLINT().not_null()),
+ (jvm.int, 1, DataTypes.INT().not_null()),
+ (jvm.long, 1, DataTypes.BIGINT().not_null()),
+ (jvm.float, 1.0, DataTypes.FLOAT().not_null()),
+ (jvm.double, 1.0, DataTypes.DOUBLE().not_null()),
+ ]
+ expressions = []
+ for primitive_class, value, _ in primitive_values:
+ j_array = gateway.new_array(primitive_class, 1)
+ j_array[0] = value
+ expressions.append(lit(j_array))
+
+ source = self.t_env.from_elements([(1,)], ["id"])
+ result = source.select(*expressions)
+ self.assertEqual(
+ result.get_resolved_schema().get_column_data_types(),
+ [
+ DataTypes.ARRAY(element_data_type).not_null()
+ for _, _, element_data_type in primitive_values
+ ],
+ )
+ self.assertIsInstance(result.explain(), str)
+ self.assertEqual(
+ list(result.execute().collect()),
+ [Row([True], [1], [1], [1], [1.0], [1.0])],
+ )
+
+ j_int_array = gateway.new_array(jvm.int, 1)
+ j_int_array[0] = 1
+ nested_expression = lit([j_int_array, []])
+ nested_result = source.select(
+ nested_expression.cardinality,
+ nested_expression.at(1).cardinality,
+ nested_expression.at(2).cardinality,
+ nested_expression.at(1).at(1),
+ )
+ self.assertIsInstance(nested_result.explain(), str)
+ self.assertEqual(
+ list(nested_result.execute().collect()),
+ [Row(2, 1, 0, 1)],
+ )
+
+ def test_empty_python_arrays_preserve_typecodes(self):
+ typecodes = sorted(_array_type_mappings)
+ 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_empty_unicode_array_typecode_propagates_to_sibling(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"])
+ expression = lit([array.array("u"), []])
+ result = source.select(expression)
+
+ self.assertEqual(
+ result.get_resolved_schema().get_column_data_types(),
+ [
+ DataTypes.ARRAY(
+ DataTypes.ARRAY(_array_type_mappings["u"])
+ ).not_null()
+ ],
+ )
+ self.assertIsInstance(result.explain(), str)
+ self.assertEqual(
+ list(
+ source.select(
+ expression.at(1).cardinality,
+ expression.at(2).cardinality,
+ )
+ .execute()
+ .collect()
+ ),
+ [Row(0, 0)],
+ )
+
+ 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 3681d0664ae..aa1121e737f 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 @@ package org.apache.flink.table.utils.python;
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,8 +38,10 @@ import org.apache.flink.table.data.GenericRowData;
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.extraction.ExtractionUtils;
import org.apache.flink.table.types.logical.ArrayType;
import org.apache.flink.table.types.logical.BigIntType;
import org.apache.flink.table.types.logical.BinaryType;
@@ -62,8 +68,12 @@ import org.apache.flink.table.types.logical.VarBinaryType;
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.table.types.utils.ValueDataTypeConverter;
+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 +81,7 @@ import java.time.Instant;
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;
@@ -143,6 +154,405 @@ public final class PythonTableUtils {
dataCollection,
InternalSerializers.create(dataType.getLogicalType()));
}
+ /** Retains an empty Python array's typecode-derived data type across the
Py4J boundary. */
+ public static Object createInferredArrayValue(final Object array, final
DataType dataType) {
+ if (!array.getClass().isArray()
+ || !(dataType.getLogicalType() instanceof ArrayType)
+ || Array.getLength(array) != 0
+ || array.getClass() != dataType.getConversionClass()) {
+ throw new IllegalArgumentException(
+ "An inferred array value requires an empty array matching
the ARRAY conversion class.");
+ }
+ return new InferredArrayValue(array, dataType);
+ }
+
+ /**
+ * Creates a literal from a value received through Py4J.
+ *
+ * <p>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 such as
+ * arrays, maps, and rows are represented as constructor calls because
Calcite cannot plan them
+ * as single value literals.
+ *
+ * @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 (value instanceof InferredArrayValue) {
+ final InferredArrayValue inferredArrayValue = (InferredArrayValue)
value;
+ return createTypedLiteral(
+ inferredArrayValue.getArray(),
+ dataType == null ? inferredArrayValue.getDataType() :
dataType);
+ }
+ if (dataType != null) {
+ return createTypedLiteral(value, dataType);
+ }
+
+ if (value instanceof List) {
+ final InferredArrayMaterialization materializedArray =
+ materializeInferredArray((List<?>) value);
+ if (materializedArray.getState() == InferredArrayState.CONCRETE) {
+ return createTypedLiteral(
+ materializedArray.getArray(),
materializedArray.getDataType());
+ }
+ return Expressions.lit(materializedArray.getArray());
+ }
+
+ final ApiExpression literal = Expressions.lit(value);
+ 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(value, 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) && !isLiteralJavaArray(value)) {
+ // 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);
+ }
+ if (value instanceof Row && ((Row) value).getKind() !=
RowKind.INSERT) {
+ final Row row = (Row) value;
+ throw new ValidationException(
+ String.format(
+ "Unsupported kind '%s' of a row [%s]. Only
rows with 'INSERT' kind are supported when"
+ + " converting to an expression.",
+ row.getKind(), row));
+ }
+ final RowType rowType = (RowType) dataType.getLogicalType();
+ final List<DataType> 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<String> 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.
+ DataType literalDataType = value == null ? declaredDataType :
declaredDataType.notNull();
+ if (value != null &&
literalDataType.getConversionClass().isPrimitive()) {
+ // Reflection boxes primitive array elements before they reach
literal validation.
+ literalDataType =
+ literalDataType.bridgedTo(
+ ExtractionUtils.primitiveToWrapper(
+ literalDataType.getConversionClass()));
+ }
+ return createTypedLiteral(value, literalDataType);
+ }
+
+ private static InferredArrayMaterialization materializeInferredArray(final
List<?> values) {
+ final Object[] convertedValues = new Object[values.size()];
+ final InferredArrayMaterialization[] nestedArrays =
+ new InferredArrayMaterialization[values.size()];
+ DataType elementDataType = null;
+ boolean compatible = true;
+
+ for (int pos = 0; pos < values.size(); pos++) {
+ final Object value = values.get(pos);
+ final Optional<DataType> possibleValueDataType;
+ if (value instanceof InferredArrayValue) {
+ final InferredArrayValue inferredArrayValue =
(InferredArrayValue) value;
+ convertedValues[pos] = inferredArrayValue.getArray();
+ possibleValueDataType =
Optional.of(inferredArrayValue.getDataType());
+ } else if (value instanceof List) {
+ final InferredArrayMaterialization nestedArray =
+ materializeInferredArray((List<?>) value);
+ nestedArrays[pos] = nestedArray;
+ convertedValues[pos] = nestedArray.getArray();
+ if (nestedArray.getState() == InferredArrayState.INCOMPATIBLE)
{
+ compatible = false;
+ continue;
+ } else if (nestedArray.getState() ==
InferredArrayState.WILDCARD) {
+ continue;
+ }
+ possibleValueDataType = Optional.of(nestedArray.getDataType());
+ } else {
+ convertedValues[pos] = value;
+ if (value == null) {
+ continue;
+ }
+ possibleValueDataType =
ValueDataTypeConverter.extractDataType(value);
+ }
+
+ if (!possibleValueDataType.isPresent()) {
+ compatible = false;
+ continue;
+ }
+ final DataType valueDataType =
possibleValueDataType.get().nullable();
+ if (elementDataType == null) {
+ elementDataType = valueDataType;
+ } else if (!elementDataType.equals(valueDataType)) {
+ compatible = false;
+ }
+ }
+
+ if (!compatible) {
+ return InferredArrayMaterialization.incompatible(convertedValues);
+ }
+ if (elementDataType == null) {
+ return InferredArrayMaterialization.wildcard(convertedValues);
+ }
+
+ // The sibling's inferred data type preserves the shape and
value-dependent logical details
+ // of empty and null-only nested lists.
+ for (int pos = 0; pos < nestedArrays.length; pos++) {
+ final InferredArrayMaterialization nestedArray = nestedArrays[pos];
+ if (nestedArray == null || nestedArray.getState() !=
InferredArrayState.WILDCARD) {
+ continue;
+ }
+ final Object convertedNestedArray =
+ materializeWildcardArray((Object[])
nestedArray.getArray(), elementDataType);
+ if (convertedNestedArray == null) {
+ return
InferredArrayMaterialization.incompatible(convertedValues);
+ }
+ convertedValues[pos] = convertedNestedArray;
+ }
+
+ final Class<?> componentClass = elementDataType.getConversionClass();
+ final Object convertedArray = Array.newInstance(componentClass,
convertedValues.length);
+ for (int pos = 0; pos < convertedValues.length; pos++) {
+ Array.set(convertedArray, pos, convertedValues[pos]);
+ }
+
+ final DataType inferredDataType =
DataTypes.ARRAY(elementDataType).notNull();
+ return InferredArrayMaterialization.concrete(convertedArray,
inferredDataType);
+ }
+
+ private static @Nullable Object materializeWildcardArray(
+ final Object[] values, final DataType expectedArrayDataType) {
+ final Class<?> expectedArrayClass =
expectedArrayDataType.getConversionClass();
+ // An array conversion class can also represent a scalar type such as
BINARY.
+ if (!(expectedArrayDataType.getLogicalType() instanceof ArrayType)
+ || !expectedArrayClass.isArray()) {
+ return null;
+ }
+
+ final Class<?> componentClass = expectedArrayClass.getComponentType();
+ final DataType elementDataType =
expectedArrayDataType.getChildren().get(0);
+ final Object convertedArray = Array.newInstance(componentClass,
values.length);
+ for (int pos = 0; pos < values.length; pos++) {
+ final Object value = values[pos];
+ final Object convertedValue;
+ if (value instanceof Object[]) {
+ convertedValue = materializeWildcardArray((Object[]) value,
elementDataType);
+ if (convertedValue == null) {
+ return null;
+ }
+ } else if (value == null && !componentClass.isPrimitive()) {
+ convertedValue = null;
+ } else {
+ return null;
+ }
+ Array.set(convertedArray, pos, convertedValue);
+ }
+ return convertedArray;
+ }
+
+ private enum InferredArrayState {
+ CONCRETE,
+ WILDCARD,
+ INCOMPATIBLE
+ }
+
+ private static final class InferredArrayValue {
+
+ private final Object array;
+ private final DataType dataType;
+
+ private InferredArrayValue(final Object array, final DataType
dataType) {
+ this.array = array;
+ this.dataType = dataType;
+ }
+
+ private Object getArray() {
+ return array;
+ }
+
+ private DataType getDataType() {
+ return dataType;
+ }
+ }
+
+ private static final class InferredArrayMaterialization {
+
+ private final InferredArrayState state;
+ private final Object array;
+ private final @Nullable DataType dataType;
+
+ private InferredArrayMaterialization(
+ final InferredArrayState state,
+ final Object array,
+ @Nullable final DataType dataType) {
+ this.state = state;
+ this.array = array;
+ this.dataType = dataType;
+ }
+
+ private static InferredArrayMaterialization concrete(
+ final Object array, final DataType dataType) {
+ return new
InferredArrayMaterialization(InferredArrayState.CONCRETE, array, dataType);
+ }
+
+ private static InferredArrayMaterialization wildcard(final Object
array) {
+ return new
InferredArrayMaterialization(InferredArrayState.WILDCARD, array, null);
+ }
+
+ private static InferredArrayMaterialization incompatible(final Object
array) {
+ return new
InferredArrayMaterialization(InferredArrayState.INCOMPATIBLE, array, null);
+ }
+
+ private InferredArrayState getState() {
+ return state;
+ }
+
+ private Object getArray() {
+ return array;
+ }
+
+ private DataType getDataType() {
+ if (dataType == null) {
+ throw new IllegalStateException("Only concrete arrays retain
their data type.");
+ }
+ return dataType;
+ }
+ }
+
+ 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
+ || isLiteralJavaArray(value);
+ }
+
+ private static boolean isLiteralJavaArray(final Object value) {
+ // ValueDataTypeConverter treats byte[] as scalar BINARY rather than
ARRAY.
+ return value.getClass().isArray() && !(value instanceof byte[]);
+ }
+
+ 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<Integer, Function<Integer, Object>, Object>
arrayConstructor(
final LogicalType elementType) {
if (elementType instanceof BooleanType) {
@@ -375,7 +785,8 @@ public final class PythonTableUtils {
return c ->
c instanceof Integer || c instanceof Long
? TimestampData.fromInstant(
- Instant.ofEpochMilli(((Number)
c).longValue() / 1000))
+ Instant.ofEpochMilli(
+ Math.floorDiv(((Number)
c).longValue(), 1000)))
: null;
}
@@ -503,6 +914,87 @@ public final class PythonTableUtils {
throw new IllegalStateException("Failed to get converter for
LogicalType: " + logicalType);
}
+ private static Function<Object, Object> 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 && isLiteralJavaArray(value)) {
+ 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 00000000000..fe88205ba2a
--- /dev/null
+++
b/flink-python/src/test/java/org/apache/flink/table/utils/python/PythonTableUtilsTest.java
@@ -0,0 +1,325 @@
+/*
+ * 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.apache.flink.types.Row;
+import org.apache.flink.types.RowKind;
+
+import org.junit.jupiter.api.Test;
+
+import java.math.BigDecimal;
+import java.time.LocalTime;
+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 testCreateLiteralInfersNestedArraysFromSiblings() {
+ assertThat(
+ PythonTableUtils.createLiteral(
+ Arrays.asList(Arrays.asList(1),
Collections.emptyList()), null))
+ .isNotNull();
+ assertThat(
+ PythonTableUtils.createLiteral(
+ Arrays.asList(Collections.emptyList(),
Arrays.asList(1)), null))
+ .isNotNull();
+ assertThat(
+ PythonTableUtils.createLiteral(
+ Arrays.asList(Arrays.asList(1),
Collections.singletonList(null)),
+ null))
+ .isNotNull();
+ assertThat(
+ PythonTableUtils.createLiteral(
+ Arrays.asList(
+
Collections.singletonList(Collections.singletonList(1)),
+
Collections.singletonList(Collections.emptyList())),
+ null))
+ .isNotNull();
+ assertThat(
+ PythonTableUtils.createLiteral(
+ Arrays.<Object>asList(new Integer[] {1},
Collections.emptyList()),
+ null))
+ .isNotNull();
+ }
+
+ @Test
+ void testCreateLiteralInfersValueDependentNestedArraysFromSiblings() {
+ assertThat(
+ PythonTableUtils.createLiteral(
+ Arrays.asList(
+ Collections.singletonList("a"),
Collections.emptyList()),
+ null))
+ .isNotNull();
+ assertThat(
+ PythonTableUtils.createLiteral(
+ Arrays.asList(
+ Collections.singletonList("a"),
+ Collections.singletonList(null)),
+ null))
+ .isNotNull();
+ assertThat(
+ PythonTableUtils.createLiteral(
+ Arrays.asList(
+ Collections.singletonList(new
BigDecimal("1.20")),
+ Collections.emptyList()),
+ null))
+ .isNotNull();
+ assertThat(
+ PythonTableUtils.createLiteral(
+ Arrays.asList(
+ Collections.singletonList(
+ LocalTime.of(12, 0, 0,
123_000_000)),
+ Collections.emptyList()),
+ null))
+ .isNotNull();
+ assertThat(
+ PythonTableUtils.createLiteral(
+ Arrays.asList(
+ Collections.singletonList(new byte[]
{1}),
+ Collections.emptyList()),
+ null))
+ .isNotNull();
+ }
+
+ @Test
+ void testCreateLiteralSupportsPrimitiveArrays() {
+ for (final Object value :
+ Arrays.<Object>asList(
+ new boolean[] {true},
+ new short[] {1},
+ new int[] {1},
+ new long[] {1},
+ new float[] {1},
+ new double[] {1})) {
+ assertThat(PythonTableUtils.createLiteral(value,
null)).isNotNull();
+ }
+ assertThat(
+ PythonTableUtils.createLiteral(
+ Arrays.asList(new int[] {1},
Collections.emptyList()), null))
+ .isNotNull();
+ }
+
+ @Test
+ void testCreateLiteralUsesInferredArrayTypeHint() {
+ final Object inferredArrayValue =
+ PythonTableUtils.createInferredArrayValue(
+ new String[0],
DataTypes.ARRAY(DataTypes.CHAR(4)).notNull());
+
+ assertThat(PythonTableUtils.createLiteral(inferredArrayValue,
null)).isNotNull();
+ assertThat(
+ PythonTableUtils.createLiteral(
+ Arrays.asList(inferredArrayValue,
Collections.emptyList()), null))
+ .isNotNull();
+ }
+
+ @Test
+ void testCreateInferredArrayValueValidatesCarrier() {
+ assertThatThrownBy(
+ () ->
+ PythonTableUtils.createInferredArrayValue(
+ new String[] {"a"},
+
DataTypes.ARRAY(DataTypes.CHAR(1)).notNull()))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessage(
+ "An inferred array value requires an empty array
matching the ARRAY "
+ + "conversion class.");
+ assertThatThrownBy(
+ () ->
+ PythonTableUtils.createInferredArrayValue(
+ new String[0],
DataTypes.ARRAY(DataTypes.INT()).notNull()))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessage(
+ "An inferred array value requires an empty array
matching the ARRAY "
+ + "conversion class.");
+ }
+
+ @Test
+ void testCreateLiteralRejectsArraysWithoutCommonSiblingType() {
+ assertThatThrownBy(
+ () ->
PythonTableUtils.createLiteral(Collections.singletonList(null), null))
+ .isInstanceOf(ValidationException.class);
+ assertThatThrownBy(
+ () ->
+ PythonTableUtils.createLiteral(
+ Arrays.asList(
+ Collections.emptyList(),
+
Collections.singletonList(null)),
+ null))
+ .isInstanceOf(ValidationException.class);
+ assertThatThrownBy(
+ () ->
+ PythonTableUtils.createLiteral(
+ Arrays.asList(
+ Collections.singletonList(1),
+ Collections.singletonList(1L)),
+ null))
+ .isInstanceOf(ValidationException.class);
+ assertThatThrownBy(
+ () ->
+ PythonTableUtils.createLiteral(
+ Arrays.asList(
+ Collections.singletonList("a"),
+
Collections.singletonList("bb")),
+ null))
+ .isInstanceOf(ValidationException.class);
+ assertThatThrownBy(
+ () ->
+ PythonTableUtils.createLiteral(
+ Arrays.asList(1,
Collections.emptyList()), null))
+ .isInstanceOf(ValidationException.class);
+ assertThatThrownBy(
+ () ->
+ PythonTableUtils.createLiteral(
+ Arrays.<Object>asList(
+ new byte[] {1},
Collections.emptyList()),
+ null))
+ .isInstanceOf(ValidationException.class);
+ assertThatThrownBy(
+ () ->
+ PythonTableUtils.createLiteral(
+ Arrays.asList(
+ Collections.singletonList(new
byte[] {1}),
+
Collections.singletonList(Collections.emptyList())),
+ null))
+ .isInstanceOf(ValidationException.class);
+ }
+
+ @Test
+ void testCreateLiteralRejectsBinaryAsConstructedValue() {
+ assertThatThrownBy(
+ () ->
+ PythonTableUtils.createLiteral(
+ new byte[] {1},
+
DataTypes.ARRAY(DataTypes.TINYINT()).notNull()))
+ .isInstanceOf(ValidationException.class);
+ assertThatThrownBy(
+ () ->
+ PythonTableUtils.createLiteral(
+ new byte[] {1, 2},
+ DataTypes.ROW(
+ DataTypes.FIELD("a",
DataTypes.TINYINT()),
+ DataTypes.FIELD("b",
DataTypes.TINYINT()))
+ .notNull()))
+ .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.");
+ }
+
+ @Test
+ void testCreateLiteralRejectsNonInsertRow() {
+ assertThatThrownBy(
+ () ->
+ PythonTableUtils.createLiteral(
+ Row.ofKind(RowKind.DELETE, 1),
+ DataTypes.ROW(DataTypes.FIELD("value",
DataTypes.INT()))
+ .notNull()))
+ .isInstanceOf(ValidationException.class)
+ .hasMessage(
+ "Unsupported kind 'DELETE' of a row [-D[1]]. "
+ + "Only rows with 'INSERT' kind are supported
when converting "
+ + "to an expression.");
+ }
+
+ 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);
+ }
+}
