Dian Fu created FLINK-41426:
-------------------------------
Summary: Python Table API thread mode fails to convert null ROW
fields
Key: FLINK-41426
URL: https://issues.apache.org/jira/browse/FLINK-41426
Project: Flink
Issue Type: Bug
Components: API / Python
Reporter: Dian Fu
The Table API conversion path in Python thread mode does not correctly handle
null ROW fields on master.
When a Table.map() input contains a null nested ROW, the Table runtime's
RowDataConverter dereferences the null value during Java-side input conversion.
This causes a NullPointerException before the Python UDF is invoked, even if
the UDF explicitly handles None.
Reproducer:
{code:python}
import sys
from pyflink.common import Row
from pyflink.table import DataTypes, EnvironmentSettings, TableEnvironment
from pyflink.table.udf import udf
table_env = TableEnvironment.create(EnvironmentSettings.in_batch_mode())
config = table_env.get_config()
config.set("python.execution-mode", "thread")
config.set("python.executable", sys.executable)
config.set("python.client.executable", sys.executable)
input_type = DataTypes.ROW([
DataTypes.FIELD("a", DataTypes.BIGINT()),
DataTypes.FIELD(
"child",
DataTypes.ROW([DataTypes.FIELD("x", DataTypes.BIGINT())]),
),
])
result_type = DataTypes.ROW([
DataTypes.FIELD("a", DataTypes.BIGINT()),
DataTypes.FIELD("b", DataTypes.BIGINT()),
])
table = table_env.from_elements([(7, None)], input_type)
mapper = udf(
lambda row: Row(row[0], None if row[1] is None else row[1][0]),
result_type=result_type,
)
with table.map(mapper).execute().collect() as rows:
print([tuple(row) for row in rows])
{code}
Expected behavior:
The null nested ROW should be passed to the Python UDF as None, producing:
{code:java}
[(7, None)]
{code}
Failure mechanism on master:
EmbeddedPythonScalarFunctionOperator converts input fields using
org.apache.flink.table.runtime.typeutils.PythonTypeUtils.
For a null field, DataFormatConverters.DataFormatConverter.toExternal(RowData,
int) returns Java null. The Table runtime's DataConverter.toExternal() forwards
this value unconditionally:
{code:java}
public final OUT toExternal(RowData row, int column) {
return toExternalImpl(dataFormatConverter.toExternal(row, column));
}
{code}
RowDataConverter.toExternalImpl() then dereferences it without checking for
null:
{code:java}
Object[] toExternalImpl(Row value) {
reuseExternalRow[0] = (long) value.getKind().toByteValue();
// ...
}
{code}
This results in a NullPointerException before execution reaches Python.
There are two separate Java classes named PythonTypeUtils. The DataStream
implementation,
org.apache.flink.streaming.api.utils.PythonTypeUtils.RowDataConverter, already
handles null values. However, Table API operators use
org.apache.flink.table.runtime.typeutils.PythonTypeUtils, whose corresponding
conversion lacks this handling.
--
This message was sent by Atlassian Jira
(v10.3.26#10030026)
