[FLINK-40422][python] Add slicing APIs to DataFrame API

[Milesian111]

What is the purpose of the change

This pull request implements FLINK-40422 by adding row-slicing transformations to the PyFlink DataFrame API.

The new APIs reuse the existing Table API operations and remain lazy:

• DataFrame.limit(n) delegates to Table.fetch(n).
• DataFrame.offset(n) delegates to Table.offset(n).
• DataFrame.head(n) delegates to DataFrame.limit(n).

offset and limit can be composed for pagination. The documentation clarifies that row order is deterministic only when the underlying table has an explicit ordering.

Brief change log

• Added the public DataFrame.limit, DataFrame.offset, and DataFrame.head APIs.
• Added validation requiring n to be a non-negative integer.
• Kept head as a delegating wrapper around limit.
• Added tests covering:
  • delegation to the underlying Table API
  • lazy execution and preservation of the original DataFrame
  • pagination through offset(a).limit(b)
  • equivalence between head and limit
  • zero-length slices
  • negative values and unsupported argument types
  • deterministic batch results with explicit ordering
• Added the new methods to the DataFrame API reference documentation.

Verifying this change

This change added tests and can be verified as follows:

• Added unit tests covering delegation to the existing Table API, lazy transformations, preservation of the original DataFrame, head delegation, pagination composition, zero-length slices, and invalid arguments.
• Added deterministic batch integration tests covering limit, offset, offset(a).limit(b), equivalence between head and limit, and n == 0.

Local execution of the batch integration tests was blocked by pre-existing invalid Java planner artifacts in the local build environment. The added batch tests are included for verification in CI.

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 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? docs and Python API 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:

• 10fc884 Azure: SUCCESS

Bot commands

The @flinkbot bot supports the following commands:

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

[davidradl]

I suggest explaining or referencing what lazy transformation means here.
Can we expose orderBy as well?
nit: I assume that changes to the underlying table content could make the row order not deterministic.

[Milesian111]

Thanks for the suggestions.

I will clarify that a lazy transformation only builds and returns a new
DataFrame plan without starting a Flink job. Execution is triggered later by
an action such as collect() or to_pandas().

Ordering support is already tracked separately by FLINK-40421. FLIP-591
specifies DataFrame.sort(), which will delegate to the existing
Table.order_by() API. To avoid introducing two overlapping public
DataFrame APIs and to keep this PR scoped to FLINK-40422, I suggest leaving
sorting support to FLINK-40421. Once implemented, users will be able to write:

df.sort("ts", descending=True).limit(100)

I will also reword the ordering note to clarify that, without explicit
sorting, the selected rows and their order are unspecified. Changes to the
underlying table content may naturally change the result, and fully stable
ordering also requires ordering keys that uniquely determine row order.

Would clarifying the lazy semantics and referencing FLINK-40421 address the
concern?

[Milesian111]

I have pushed a follow-up commit that clarifies the lazy transformation
semantics and the ordering behavior for limit, offset, and head.

Sorting support remains tracked separately by FLINK-40421, so I did not add a
separate DataFrame.order_by() API to this PR.