You are viewing a plain text version of this content. The canonical link for it is here.

Posted to issues@flink.apache.org by "auroflow (via GitHub)" <gi...@apache.org> on 2026/09/16 13:07:34 UTC

[PR] [FLINK-40472][python] Support Arrow vectorized UDFs in DataFrame API [flink]

auroflow opened a new pull request, #29210:
URL: https://github.com/apache/flink/pull/29210

## What is the purpose of the change

Implement [FLINK-40472](https://issues.apache.org/jira/browse/FLINK-40472) by adding synchronous Arrow vectorized scalar UDFs to the PyFlink DataFrame API. UDFs receive Arrow arrays and can apply `pyarrow.compute` operations directly to column batches.

Arrow mode can be inferred from Arrow container annotations or selected with `func_type="arrow"`. The Table API supports explicit Arrow mode.

## Brief change log

- Add Arrow UDF declaration and annotation inference, and document the new execution mode.
- Distinguish general, pandas and Arrow functions in Python calculation splitting and map merging.
- Extend the existing Arrow operator, worker and coder protocol to carry native Arrow batches while retaining pandas defaults.
- Support nested and whole-row arguments, normalize chunked results, and validate result containers, row counts and terminal schemas.
- Share Arrow schema mapping, preserve nested nullability and PyArrow 5 compatibility, reduce temporary validation allocations, and reset map child writers between batches.

## Verifying this change

This change extends existing declaration, worker, coder, planner and Arrow reader/writer tests, plus the combined DataFrame UDF integration test. Tests cover observable values, schemas, execution plans and error behavior.

- Full Python suite: **1,284 passed, 61 skipped, 528 subtests passed**.
- Focused pure-Python and compiled coder checks passed on PyArrow 5, 18 and 23.
- Targeted Java planner, protocol, Arrow operator and reader/writer tests passed, including `PythonCalcSplitFunctionKindRuleTest`, `ProtoUtilsTest` and `ArrowReaderWriterTest`.
- Java 17 reactor build passed with `./mvnw clean install -DskipTests -Dfast -Pskip-webui-build -T4`. The full Maven test suite was not run.
- Spotless, flake8 and mypy checks passed, including the pre-push checks.

## Does this pull request potentially affect one of the following parts:

- Dependencies (does it add or upgrade a dependency): **no**
- The public API, i.e., is any changed class annotated with `@Public(Evolving)`: **yes** — extends the Python DataFrame and Table UDF APIs.
- The serializers: **yes** — Python Arrow coders, Arrow writers and additive worker-protocol metadata; managed-state serializers are unchanged.
- The runtime per-record code paths (performance sensitive): **yes** — Python scalar execution and Arrow batch serialization.
- Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Kubernetes/Yarn, ZooKeeper: **no**
- The S3 file system connector: **no**

## Documentation

- Does this pull request introduce a new feature? **yes**
- If yes, how is the feature documented? **docs** — DataFrame UDF reference documentation and Python API docstrings.

---

##### Was generative AI tooling used to co-author this PR?

- [X] Yes (please specify the tool below)

Generated-by: Codex (GPT-6)

--
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: issues-unsubscribe@flink.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] [FLINK-40472][python] Support Arrow vectorized UDFs in DataFrame API [flink]

Posted by "auroflow (via GitHub)" <gi...@apache.org>.

auroflow commented on code in PR #29210:
URL: https://github.com/apache/flink/pull/29210#discussion_r4043947251


##########
flink-python/src/main/java/org/apache/flink/table/runtime/arrow/writers/MapWriter.java:
##########
@@ -84,6 +84,13 @@ public void doWrite(T in, int ordinal) {
         }
     }
 
+    @Override
+    public void reset() {

Review Comment:
   Yes, this fixes an existing bug affecting pandas UDFs too. Between Arrow batches, the map vector resets, but its key/value writers retain their counters. Subsequent batches then write children at incorrect positions.
   
   I have taken this fix out of this PR and created [FLINK-40705](https://issues.apache.org/jira/browse/FLINK-40705) to address this 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: issues-unsubscribe@flink.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] [FLINK-40472][python] Support Arrow vectorized UDFs in DataFrame API [flink]

Posted by "auroflow (via GitHub)" <gi...@apache.org>.

auroflow commented on code in PR #29210:
URL: https://github.com/apache/flink/pull/29210#discussion_r4043947251


##########
flink-python/src/main/java/org/apache/flink/table/runtime/arrow/writers/MapWriter.java:
##########
@@ -84,6 +84,13 @@ public void doWrite(T in, int ordinal) {
         }
     }
 
+    @Override
+    public void reset() {

Review Comment:
   Yes, this fixes an existing bug affecting pandas UDFs too. Between Arrow batches, the map vector resets, but its key/value writers retain their counters. Subsequent batches then write children at incorrect positions.
   
   I have taken this fix out of this PR and created [LINK-40705](https://issues.apache.org/jira/browse/FLINK-40705) to address this 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: issues-unsubscribe@flink.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] [FLINK-40472][python] Support Arrow vectorized UDFs in DataFrame API [flink]

Posted by "flinkbot (via GitHub)" <gi...@apache.org>.

flinkbot commented on PR #29210:
URL: https://github.com/apache/flink/pull/29210#issuecomment-5698256564

   <!--
   Meta data
   {
     "version" : 1,
     "metaDataEntries" : [ {
       "hash" : "c3b03f999a548d455e6bd82e6835a744e5716a3c",
       "status" : "UNKNOWN",
       "url" : "TBD",
       "triggerID" : "c3b03f999a548d455e6bd82e6835a744e5716a3c",
       "triggerType" : "PUSH"
     } ]
   }-->
   ## CI report:
   
   * c3b03f999a548d455e6bd82e6835a744e5716a3c UNKNOWN
   
   <details>
   <summary>Bot commands</summary>
     The @flinkbot bot supports the following commands:
   
    - `@flinkbot run azure` re-run the last Azure build
   </details>


-- 
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: issues-unsubscribe@flink.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] [FLINK-40472][python] Support Arrow vectorized UDFs in DataFrame API [flink]

Posted by "dianfu (via GitHub)" <gi...@apache.org>.

dianfu merged PR #29210:
URL: https://github.com/apache/flink/pull/29210


-- 
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: issues-unsubscribe@flink.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] [FLINK-40472][python] Support Arrow vectorized UDFs in DataFrame API [flink]

Posted by "auroflow (via GitHub)" <gi...@apache.org>.

auroflow commented on code in PR #29210:
URL: https://github.com/apache/flink/pull/29210#discussion_r4043952842


##########
flink-python/pyflink/table/udf.py:
##########
@@ -760,10 +762,19 @@ def _get_python_env():
 
 
 def _create_udf(f, input_types, result_type, func_type, deterministic, name):
+    if func_type == 'arrow':

Review Comment:
   Yes, this should apply to pandas UDFs as well, until we support async functions for both modes. I have broaden the validation here.



-- 
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: issues-unsubscribe@flink.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] [FLINK-40472][python] Support Arrow vectorized UDFs in DataFrame API [flink]

Posted by "dianfu (via GitHub)" <gi...@apache.org>.

dianfu commented on code in PR #29210:
URL: https://github.com/apache/flink/pull/29210#discussion_r4043651939


##########
flink-python/pyflink/table/udf.py:
##########
@@ -760,10 +762,19 @@ def _get_python_env():
 
 
 def _create_udf(f, input_types, result_type, func_type, deterministic, name):
+    if func_type == 'arrow':

Review Comment:
   I guess we can also apply this validation for pandas UDF?



##########
flink-python/src/main/java/org/apache/flink/table/runtime/arrow/writers/MapWriter.java:
##########
@@ -84,6 +84,13 @@ public void doWrite(T in, int ordinal) {
         }
     }
 
+    @Override
+    public void reset() {

Review Comment:
   What's the purpose of this change? Does it apply for existing Pandas UDF? If so, I guess we need split it into a separate PR to make it possible to backport to existing releases.



-- 
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: issues-unsubscribe@flink.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org