fdolce opened a new pull request, #29088:
URL: https://github.com/apache/flink/pull/29088
## What is the purpose of the change
This pull request adds configuration support to the PyFlink DataFrame API
(FLIP-591, FLINK-40430). It introduces a unified configuration entry point,
`pf.config`, a singleton `DataFrameConfig` object that accepts any Flink
configuration key. Values can be set at any time, even before a
`TableEnvironment` exists, because they are buffered and applied automatically
once the underlying environment is lazily created or injected via
`set_table_environment`. This means users neither manage environment-creation
order nor choose between environment objects (`StreamExecutionEnvironment` vs.
`TableConfig`) for the common case.
```python
import pyflink.dataframe as pf
pf.config.set("parallelism.default", "4")
pf.config.set("execution.runtime-mode", "batch")
```
Semantics, where FLIP-591 leaves them open:
- The lazily created environment is built *from* the buffered values (they
are passed as the `Configuration` of the `StreamExecutionEnvironment`), so
options that can only be chosen at creation time (`execution.runtime-mode`)
work. Applying them to the `TableConfig` after creation would make the planner
reject the first `execute()` with "Mismatch between configured runtime mode and
actual runtime mode".
- An injected environment receives the buffered values only for keys it
does not already set explicitly in its own `TableConfig`, so an environment the
user configured themselves is not silently overridden.
- Buffered values persist across environments: they are re-applied to
every environment created or injected later, not consumed on first use.
- Applying the buffer happens before the environment is stored as the
active one. If it fails, the previously active environment is kept and a retry
re-applies the buffer, rather than retaining a half-configured environment.
- A value the active environment rejects (e.g. an invalid `pipeline.jars`
URL) is not buffered, so one bad `set` cannot make every later
`set_table_environment` fail.
## Brief change log
- Added `pyflink/dataframe/_config.py` with the `DataFrameConfig` class
(chainable `set(key, value)` and `get(key, default)`) and the module-level
`config` singleton. The module is private (`_config`) because a public module
named `config` would be shadowed by the `pf.config` attribute (`import
pyflink.dataframe.config as m` would yield the singleton instead of the module).
- `set` writes the value through to the active environment's `TableConfig`
when one exists and buffers it only on success; `get` reads from the active
environment (falling back to the root configuration via `TableConfig.get`) or
from the buffer when no environment exists
- `get_or_create_table_environment` creates the
`StreamExecutionEnvironment` from a `Configuration` built from the buffered
values, applies the buffer to the resulting `TableConfig`, and only then stores
the environment
- `set_table_environment` applies the buffered values that the injected
environment does not set explicitly, and only then stores the environment
- Exported `DataFrameConfig` and `config` from `pyflink.dataframe`
- Added a Configuration page to the PyFlink DataFrame API reference docs,
listing both `config` and `DataFrameConfig`
## Verifying this change
This change added tests and can be verified as follows:
- Added `pyflink/dataframe/tests/test_config.py` covering: type validation
of `set`/`get` arguments, chaining, buffering before an environment exists,
returning defaults for unset keys, applying buffered values on environment
injection and on lazy environment creation, writing through to the active
environment, and reading values set directly on the active environment
- Added tests for the semantics above: `execution.runtime-mode` buffered
before creation is honored by executing a query on the created environment; a
failure while applying the buffer leaves `get_table_environment()` unchanged
for both `set_table_environment` and `get_or_create_table_environment`, and a
subsequent `get_or_create_table_environment` re-applies the buffer; a value
rejected by the active environment is not buffered and not replayed onto a
later environment; an injected environment keeps a value it set explicitly; no
`pyflink.dataframe` submodule is shadowed by a package attribute
- Existing tests in `pyflink/dataframe/tests/` all pass
## 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 `PublicEvolving` Python API `DataFrameConfig` /
`pf.config`)
- 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 (new Configuration page in
the PyFlink DataFrame API reference,
`flink-python/docs/reference/pyflink.dataframe/config.rst`) and Python
docstrings
---
##### Was generative AI tooling used to co-author this PR?
- [X] Yes (please specify the tool below)
Generated-by: Claude Code
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
