[
https://issues.apache.org/jira/browse/FLINK-40705?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Liu Liu updated FLINK-40705:
----------------------------
Description:
Pandas UDFs consuming MAP columns can fail when input records span multiple
Arrow batches.
{{MapWriter}} does not override {{reset()}} to reset its key and value writers.
Consequently, the map vector restarts at the beginning of each batch, while its
child writers retain their previous counters. Subsequent batches write map
entries at incorrect positions.
This can produce incorrect decoded values or fail with:
{{Map array child array should have no nulls}}
To reproduce, pass two records containing
{{{1: 11}}}
and
{{{2: 22}}}
in a {{MAP<INT NOT NULL, INT>}} column to a pandas UDF, with:
{code:java}
t_env.get_config().set("python.fn-execution.arrow.batch.size", "1")
t_env.get_config().set("python.fn-execution.bundle.size", "100") {code}
The second single-record batch fails; setting the Arrow batch size to {{2}}
allows both records to succeed in one batch.
The defect was introduced with MAP support in FLINK-30607.
was:
Pandas UDFs consuming MAP columns can fail when input records span multiple
Arrow batches.
{{MapWriter}} does not override {{reset()}} to reset its key and value writers.
Consequently, the map vector restarts at the beginning of each batch, while its
child writers retain their previous counters. Subsequent batches write map
entries at incorrect positions.
This can produce incorrect decoded values or fail with:
{{Map array child array should have no nulls}}
To reproduce, pass two records containing {{{1: 11}}} and {{{2: 22}}} in a
{{MAP<INT NOT NULL, INT>}} column to a pandas UDF, with:
{{t_env.get_config().set("python.fn-execution.arrow.batch.size", "1")
t_env.get_config().set("python.fn-execution.bundle.size", "100")}}
The second single-record batch fails; setting the Arrow batch size to {{2}}
allows both records to succeed in one batch.
The defect was introduced with MAP support in FLINK-30607.
> PyFlink pandas UDFs fail on MAP inputs across Arrow batches
> -----------------------------------------------------------
>
> Key: FLINK-40705
> URL: https://issues.apache.org/jira/browse/FLINK-40705
> Project: Flink
> Issue Type: Bug
> Components: API / Python
> Affects Versions: 1.17.0
> Reporter: Liu Liu
> Priority: Major
>
> Pandas UDFs consuming MAP columns can fail when input records span multiple
> Arrow batches.
> {{MapWriter}} does not override {{reset()}} to reset its key and value
> writers. Consequently, the map vector restarts at the beginning of each
> batch, while its child writers retain their previous counters. Subsequent
> batches write map entries at incorrect positions.
> This can produce incorrect decoded values or fail with:
> {{Map array child array should have no nulls}}
> To reproduce, pass two records containing
> {{{1: 11}}}
> and
> {{{2: 22}}}
> in a {{MAP<INT NOT NULL, INT>}} column to a pandas UDF, with:
> {code:java}
> t_env.get_config().set("python.fn-execution.arrow.batch.size", "1")
> t_env.get_config().set("python.fn-execution.bundle.size", "100") {code}
> The second single-record batch fails; setting the Arrow batch size to {{2}}
> allows both records to succeed in one batch.
> The defect was introduced with MAP support in FLINK-30607.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
