[FLINK-40431][python] Support per-UDF concurrency and batch size in PyFlink DataFrame API

[auroflow]

What is the purpose of the change

This pull request adds per-UDF concurrency and batch-size settings to the PyFlink
DataFrame API.

Users can configure these settings when declaring a UDF:

@udf(concurrency=4)
def add_one(value: int) -> int:
    return value + 1


@udf(
    return_dtype=DataType.int64(),
    func_type="pandas",
    batch_size=256,
)
def pandas_add_one(values):
    return values + 1

The planner applies an explicitly configured concurrency as the Python operator
parallelism. UDFs with incompatible explicit concurrency values are placed in
separate operators.

When compatible pandas UDFs are fused into one operator, the smallest explicitly
configured batch size is used. If no UDF specifies a batch size, the existing
global Arrow batch-size configuration remains effective.

This change builds on the PyFlink DataFrame UDF support currently under review
and is intended to be rebased onto master after that change is merged.

Brief change log

• Added optional concurrency and batch_size arguments to DataFrame and Table
  scalar UDF declarations.
• Added validation for positive integer values and restricted batch_size to
  pandas UDFs.
• Propagated the settings through Python function metadata while retaining
  backward-compatible Java constructors and defaults.
• Added planner rules that separate Python calculations with incompatible
  explicit concurrency values, including nested UDF calls.
• Applied explicit concurrency as configured operator parallelism for synchronous
  and asynchronous Python operators.
• Applied the minimum explicitly requested batch size to fused pandas UDFs while
  retaining the global configuration when no per-UDF value is set.
• Prevented Table map operators with conflicting explicit execution settings from
  being merged.
• Updated the DataFrame UDF API documentation and examples.

Verifying this change

This change added and extended tests that can be verified as follows:

• Extended DataFrame UDF tests to cover argument validation, metadata propagation,
  backward compatibility, and execution.
• Added planner tests for recursive concurrency grouping, operator splitting, and
  map-merge compatibility.
• Added translation tests that verify configured operator parallelism, inherited
  parallelism, fused minimum batch size, global batch-size fallback, and
  asynchronous UDF concurrency without starting a MiniCluster.
• Ran pyflink.dataframe.tests.test_udf: 15 tests passed.
• Ran the focused 66-module Maven reactor build: 41 planner tests and 3 translation
  tests passed.
• Ran flake8 and targeted mypy checks for the affected Python API files.

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,
  the public PyFlink DataFrame and Table UDF creation APIs gain optional arguments
• The serializers: no
• The runtime per-record code paths (performance sensitive): no
• 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? API documentation and docstrings

---

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

• Yes (please specify the tool below)

Generated-by: Codex (GPT-5)

[flinkbot]

CI report:

• a98fcff Azure: SUCCESS

Bot commands

The @flinkbot bot supports the following commands:

• @flinkbot run azure re-run the last Azure build