sangkyoonnam commented on code in PR #29287:
URL: https://github.com/apache/flink/pull/29287#discussion_r4296704991
##########
flink-python/src/main/java/org/apache/flink/table/runtime/operators/python/scalar/EmbeddedPythonScalarFunctionOperator.java:
##########
@@ -104,16 +113,24 @@ public EmbeddedPythonScalarFunctionOperator(
@SuppressWarnings("unchecked")
@Override
public void open() throws Exception {
- hasOnlyOneInputArgument = udfInputOffsets.length == 1;
+ takesRowAsInput =
+ Arrays.stream(scalarFunctions)
+ .anyMatch(
+ pythonFunctionInfo ->
+
pythonFunctionInfo.getPythonFunction().takesRowAsInput());
+ hasOnlyOneInputArgument = takesRowAsInput || udfInputOffsets.length ==
1;
Review Comment:
Forcing this to `true` changes more than the input converter. It's also
passed to Python as `one_arg_optimization`, and `extract_user_defined_function`
then renders every `inputOffset` argument as `value`. For the merged chain
`t.map(plain(t.a, t.b)).map(f)`, that gives `f(plain(value, value))`, so
`plain` returns `Row(value, value)` and the second map raises `TypeError` at
`r[0] * 10` because `r[0]` is a `Row`.
Going back to `udfInputOffsets.length == 1` alone wouldn't cover the
one-column chain, since the flag would be on again with a `Row` input:
`f(plain(value))` passes the whole `Row` to `lambda a: Row(a * 10)`, which
raises `TypeError` before the second map runs. A single row-based `map`
generates `f(value)` with or without the flag, and `from_row_schema_proto`
ignores it. Could this be `!takesRowAsInput && udfInputOffsets.length == 1`,
with thread-mode IT cases for a merged chain over one and two input columns?
##########
flink-python/pyflink/fn_execution/embedded/operation_utils.py:
##########
@@ -81,8 +90,7 @@ def create_table_operation_from_proto(proto,
input_coder_info, output_coder_into
serialized_fn = parse_function_proto(proto)
- input_data_converter = (
-
from_schema_proto(parse_coder_proto(input_coder_info).flatten_row_type.schema))
+ input_data_converter =
row_data_converter_from_coder_proto(input_coder_info)
Review Comment:
`EmbeddedPythonTableFunctionOperator` still sends
`createFlattenRowTypeCoderInfoDescriptorProto` and the raw field array, so
`row_type` is never set here and `flat_map` keeps the tuple input (last row
above). Covering it would take both the row coder and the `(rowKind, fields)`
envelope in that operator, plus a named-field test. Would you include that
here, or keep this helper call unchanged and handle `flat_map` separately?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
