fdolce opened a new pull request, #29262:
URL: https://github.com/apache/flink/pull/29262
## What is the purpose of the change
This pull request adds catalog support to the PyFlink DataFrame API. Until
now a DataFrame could only be created from in-memory data or via
`read_generic`, and written via `write_generic`. Users could not create or
switch catalogs, nor read from or write to tables registered in a catalog,
without dropping down to the Table API or SQL. This change exposes catalog and
database management as module-level functions and adds `read_catalog_table` /
`DataFrame.write_catalog_table` for catalog table I/O, mirroring `CREATE
CATALOG`, `USE CATALOG`, `USE`, and `INSERT [OVERWRITE] INTO`.
It also makes Flink-side user errors friendlier: `ValidationException` and
`SqlParserException` raised from the JVM (unknown table, malformed path,
invalid connector options, bad watermark expression, ...) are now surfaced as a
Python `ValueError` carrying Flink's message and chained to the original
`Py4JJavaError`, instead of a raw Py4J stack trace. This applies to the new
functions and to the existing `read_generic` / `write_generic`.
## Brief change log
- New module `pyflink.dataframe.catalog` with `create_catalog`,
`get_catalog`, `use_catalog`, `get_current_catalog`, `list_catalogs`,
`use_database`, `get_current_database`, `list_databases`, all operating on the
shared DataFrame `TableEnvironment`
- New `pyflink.dataframe.read_catalog_table(path)` that resolves `table`,
`db.table`, or `catalog.db.table` against the current catalog/database
- New `DataFrame.write_catalog_table(path, *, overwrite=False)`; the
insert is shared with `write_generic` through a private `_execute_insert`
helper that keeps the wait-for-completion behaviour on local/MiniCluster
execution
- New module `pyflink.dataframe.errors` with `_raise_as_value_error`,
which translates Flink `ValidationException` / `SqlParserException` (including
subclasses) into `ValueError` and re-raises everything else unchanged, so
exceptions PyFlink already maps (e.g. `CatalogException`) keep their type
- `read_generic` and `write_generic` now route Flink errors through the
same translation
- Python-side argument validation (`TypeError` for wrong types,
`ValueError` for empty names/keys, duplicate catalog names, unknown
catalog/database names)
- All new public functions re-exported from `pyflink.dataframe` and marked
`@PublicEvolving`
- Sphinx docs: new `catalog.rst` reference page, `read_catalog_table` and
`DataFrame.write_catalog_table` added to `io.rst`
## Verifying this change
This change added tests and can be verified as follows:
- `pyflink/dataframe/tests/test_catalog.py` (`CatalogTests`): descriptor
construction for `create_catalog`, argument validation, duplicate catalog
rejection, catalog/database navigation, unknown-name rejection, translation of
Flink `ValidationException` into `ValueError`, non-user errors passing through
untranslated, and use of the DataFrame `TableEnvironment
- `pyflink/dataframe/tests/test_io.py` (`CatalogTableIOTests`): path
resolution for `read_catalog_table` (short, qualified, and fully qualified
paths), invalid path handling, `write_catalog_table` argument passing incl.
`overwrite`, wait-for-completion on local and MiniCluster execution targets,
error translation for both read and write, and an end-to-end round trmemory`
catalog
- `test_io.py` also adds `test_read_geners` and
`test_write_generic_translates_flink_erroric I/O
## Does this pull request potentially affearts:
- Dependencies (does it add or upgrade a
- The public API, i.e., is any changed class annotated with
`@Public(Evolving)`: yes (new `@PublicEvolving` functions in
`pyflink.dataframe` and a new on `DataFrame`)
- The serializers: no
- The runtime per-record code paths (performance sensitive): no
- Anything that affects deployment or rets 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?ce: new `catalog.rst`, updated
`io.rst`)
and docstrings with examples on every new
---
##### 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]
