[
https://issues.apache.org/jira/browse/FLINK-39532?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Dian Fu closed FLINK-39532.
---------------------------
Fix Version/s: 2.3.1
2.4.0
Resolution: Fixed
Fixed in:
- master via 11c6b478ac35faf42c2cb11a7342e996044911ce
- release-2.3 via 3024cb24f917f668be98a35605f197d1b7670dbb
> Python AsyncScalarFunctionOperation has race condition when using Cython coder
> ------------------------------------------------------------------------------
>
> Key: FLINK-39532
> URL: https://issues.apache.org/jira/browse/FLINK-39532
> Project: Flink
> Issue Type: Bug
> Components: API / Python
> Affects Versions: 2.3.0
> Reporter: Liu Liu
> Assignee: Liu Liu
> Priority: Major
> Labels: pull-request-available
> Fix For: 2.3.1, 2.4.0
>
>
> Currently in AsyncScalarFunctionOperation, the process_element method submits
> the input {{value}} to an asyncio event loop and returns immediately. For
> flatten row, the input comes from FlattenRowCoderImpl.decode_from_stream. In
> the Cython implementation in coder_impl_fast.pyx, this method returns a
> reused list; by the time the async function runs, this reused list has
> already been overwritten by the next decoded row. This will make Python async
> UDFs receive wrong input values, causing corrupted results.
> This problem has been reproduced by the following SQL job using Python async
> UDFs:
> {{@udf(input_types=[DataTypes.BIGINT()], result_type=DataTypes.BIGINT())}}
> {{async def async_add_one:}}
> {{ await asyncio.sleep(0.001)}}
> {{ return i + 1}}
> {{@udf(input_types=[DataTypes.BIGINT()], result_type=DataTypes.STRING())}}
> {{async def async_wrap:}}
> {{ await asyncio.sleep(0.001)}}
> {{ return f"v=\{i}"}}
> {{INSERT INTO sink}}
> {{SELECT}}
> {{a AS a,}}
> {{async_add_one(a) AS plus_one,}}
> {{async_wrap(async_add_one(a)) AS wrapped}}
> {{FROM source;}}
> Expected result:
> {{+I[22451, 22452, v=22452]}}
> {{+I[67909, 67910, v=67910]}}
> {{+I[3566, 3567, v=3567]}}
> {{+I[91654, 91655, v=91655]}}
> {{+I[2134, 2135, v=2135]}}
> {{...}}
> Actual result:
> {{+I[92581, 97806, v=52384]}}
> {{+I[97805, 52384, v=86151]}}
> {{+I[52383, 86151, v=72423]}}
> {{+I[86150, 72423, v=77937]}}
> {{+I[72422, 77937, v=10947]}}
> {{...}}
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
