[FLINK-40260][python] Support the Python PTF definition API and the stateless runtime path

[hsrain3]

What is the purpose of the change

Java Process Table Functions are not currently accessible through the PyFlink Table API, and
PyFlink does not provide APIs or a Python runtime contract for defining Python Process Table
Functions.

This pull request introduces the PyFlink Process Table Function public APIs and the corresponding
Python and Java runtime foundations. The implementation reuses Flink's existing Java PTF type
inference, Beam Python function infrastructure, keyed state backend, and internal timer service.

This is a foundational subtask. Planner translation to the Python PTF operator and end-to-end
execution tests will be added in follow-up subtasks.

The first Python PTF version is designed for a single table argument, append-only input and output,
ROW value state with TTL, and event-time timers. Multi-input Python PTFs, updating inputs,
ORDER BY, inline invocation, and additional state types are outside the scope of this change.

Brief change log

• Added the public PyFlink APIs ProcessTableFunction, udptf(),
  ProcessTableFunctionArgument, ProcessTableFunctionState, and the supported argument traits.
• Added callback signature and metadata validation for Python PTF arguments, states, result
  types, table semantics, and timer usage.
• Added Table.process(), Table.partition_by(), Table.as_argument(),
  PartitionedTable, and TableEnvironment.from_call().
• Enabled PyFlink Table API calls to registered Java Process Table Functions, including
  multi-input Java PTF calls through from_call().
• Added the internal PythonProcessTableFunction planner placeholder with explicit static
  arguments, state type strategies, state TTL, and result type inference.
• Extended the Python Fn Execution protocol with the Python PTF function, argument, state,
  key, TTL, runtime context, metrics, and state-cache metadata.
• Added the Python worker operation for 0-N result emission, mutable ROW value state,
  automatic state write-back, explicit state clearing, and timer callbacks.
• Added a Beam Python function runner adapter that declares the PTF transform and event-time
  timer family.
• Added the Java Python PTF operator and keyed timer registration for named and anonymous
  event-time timers, timer deletion, clear-all operations, and checkpoint recovery.
• Extracted a shared timer registration handler so PTF timer commands can reuse the existing
  Python timer command channel.

Verifying this change

This change added tests and can be verified as follows:

• Added Python API tests for PTF definition, ordered argument/state metadata, callback
  signatures, traits, result types, TTL, and unsupported first-version combinations.
• Added PyFlink Table API tests for process(), partition_by(), as_argument(), and
  multi-input Java PTF calls through from_call().
• Added Python worker tests for empty and multiple states, automatic write-back, explicit
  clearing, exception handling, TTL forwarding, 0-N output, time conversions, and timer
  callbacks.
• Added Java tests for Python PTF type inference and metadata validation.
• Added Beam runner tests for stateless transforms and event-time timer-family declarations.
• Added timer registration tests for named and anonymous timers, key isolation, deletion,
  clear-all operations, invalid commands, and checkpoint restore.
• Verified the protobuf generated files with the protobuf synchronization test.
• Ran the flink-python, flink-table-common, and flink-table-runtime module test suites.
• Ran the PyFlink PTF, API completeness, UDF/UDTF regression tests, and Python lint 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, new
  PyFlink public APIs are added; no existing public API is changed incompatibly
• The serializers: no; the Python Fn Execution protobuf is extended, but no Flink state
  serializer is changed
• The runtime per-record code paths (performance sensitive): yes, for the new Python PTF
  execution path
• Anything that affects deployment or recovery: yes, the new keyed PTF timer and state path
  participates in checkpoint and recovery; no JobManager or deployment mechanism is changed
• The S3 file system connector: no

Documentation

• Does this pull request introduce a new feature? yes
• If yes, how is the feature documented? Python API docstrings and JavaDocs. End-user
  documentation and examples will be added with the planner integration subtask.

---

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

• Yes (Codex)

Generated-by: Codex (GPT-5.6)

[flinkbot]

CI report:

• 2081555 Azure: SUCCESS

Bot commands

The @flinkbot bot supports the following commands:

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