[
https://issues.apache.org/jira/browse/FLINK-39532?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Liu Liu updated FLINK-39532:
----------------------------
Description:
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(i):}}
{{ await asyncio.sleep(0.001)}}
{{ return i + 1}}
{{@udf(input_types=[DataTypes.BIGINT()], result_type=DataTypes.STRING())}}
{{async def async_wrap(i):}}
{{ 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]}}
{{...}}
was:
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 corrupt the result of Python
async functions.
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(i):{}}}{{
await asyncio.sleep(0.001)}}{{ return i + 1}}
{{{}@udf(input_types=[DataTypes.BIGINT()],
result_type=DataTypes.STRING()){}}}{{{}async def async_wrap(i):{}}}{{ 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;{}}}
> 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
> Priority: Major
>
> 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(i):}}
> {{ await asyncio.sleep(0.001)}}
> {{ return i + 1}}
> {{@udf(input_types=[DataTypes.BIGINT()], result_type=DataTypes.STRING())}}
> {{async def async_wrap(i):}}
> {{ 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)
