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 11c6b478ac3 [FLINK-39532][python] Fix race condition in Python
AsyncScalarFunctionOperation
11c6b478ac3 is described below
commit 11c6b478ac35faf42c2cb11a7342e996044911ce
Author: Liu Liu <[email protected]>
AuthorDate: Tue Apr 28 13:59:33 2026 +0800
[FLINK-39532][python] Fix race condition in Python
AsyncScalarFunctionOperation
This closes #28036.
---
.../pyflink/fn_execution/table/async_function/operations.py | 11 +++++++++++
1 file changed, 11 insertions(+)
diff --git
a/flink-python/pyflink/fn_execution/table/async_function/operations.py
b/flink-python/pyflink/fn_execution/table/async_function/operations.py
index 5d536c47496..da325ab9aa3 100644
--- a/flink-python/pyflink/fn_execution/table/async_function/operations.py
+++ b/flink-python/pyflink/fn_execution/table/async_function/operations.py
@@ -60,6 +60,11 @@ class AsyncScalarFunctionOperation(Operation,
AsyncOperationMixin):
operation_utils.extract_user_defined_function(
serialized_fn.udfs[0], one_arg_optimization=False)
+ # Mirror PythonScalarFunctionOperator.createInputCoderInfoDescriptor:
+ # Java picks FlattenRowCoder unless some UDF takes a row as input.
+ self._input_is_flatten_row = not any(
+ udf.takes_row_as_input for udf in serialized_fn.udfs)
+
# Create the eval function
self._eval_func = eval('lambda value: %s' % scalar_function,
variable_dict)
@@ -140,6 +145,12 @@ class AsyncScalarFunctionOperation(Operation,
AsyncOperationMixin):
"""
self._raise_exception_if_exists()
+ # The Cython FlattenRowCoderImpl returns a reused list whose slots are
+ # overwritten by the next decode, so we must snapshot the row before
+ # the async closure can capture it.
+ if self._input_is_flatten_row:
+ value = list(value)
+
entry = self._queue.put(None, 0, 0, value)
async def execute_async(rh):
