[FLINK-40197][python] Add sql() to the DataFrame API - #29036
Conversation
| } | ||
| finally: | ||
| del frame, caller | ||
| t_env = get_or_create_table_environment() |
There was a problem hiding this comment.
It always obtains the global TableEnvironment. This prevents a DataFrame created through the public from_table() API from being used unless the caller also mutates the global DataFrame context. This is a simple example to reproduce it:
custom_env = TableEnvironment.create(EnvironmentSettings.in_batch_mode())
source = pf.from_table(
custom_env.from_elements([(1,), (2,)], ["a"])
)
pf.sql(
"SELECT * FROM source",
auto_bind=False,
source=source,
)
We may need to resolve the environment from explicit bindings first, then auto-bound candidates, and fall back to the global environment if none is available.
There was a problem hiding this comment.
Makes sense. I guess we need to check all the explicit bindings, and raise an error if they come from different environments too right?
| f"cannot bind '{name}': a temporary table or view with this name " | ||
| "already exists" | ||
| ) | ||
| t_env.create_temporary_view(name, value.to_table()) |
There was a problem hiding this comment.
Auto-bound names are validated before registration, while explicit names are passed directly to create_temporary_view().
| @PublicEvolving() | ||
| def sql(query: str, *, auto_bind: bool = True, **bindings: DataFrame) -> DataFrame: | ||
| """ | ||
| Execute a SQL SELECT query and return the result as a :class:`DataFrame`. |
There was a problem hiding this comment.
The documentation and error message say that only SELECT queries are supported, but actually it also accepts query operations such as VALUES(1), UNION, INTERSECT, EXCEPT, and ORDER BY, etc. I guess we could just describe it simply as Execute a SQL query
There was a problem hiding this comment.
Mh right. If it's ok to allow those too, then yes I'll make the comment less specific.
Add pyflink.dataframe.sql(), which executes a SQL SELECT query and returns the result as a DataFrame. Referenced DataFrames are registered as temporary views for the duration of the call and dropped afterwards: auto-binding registers DataFrames from the caller's scope under their variable names (best-effort, with warnings on invalid identifiers and collisions, never shadowing permanent catalog objects), while explicit keyword bindings are strict, take precedence, and may intentionally shadow permanent catalog tables. UDF/UDTF bindings are intentionally out of scope and will be added in a separate PR once the DataFrame API gains UDF support in general.
c71b1e2 to
c948c3c
Compare
3 participants
What is the purpose of the change
Adds
pyflink.dataframe.sql()(FLINK-40197), which executes a SQL SELECT query and returns the result as aDataFrame, so SQL and DataFrame operations can be mixed freely. Referenced DataFrames are registered as temporary views only for the duration of the call and dropped afterwards. By default the caller's variables are scanned and DataFrames are registered under their Python variable names (auto-bind); explicit keyword bindings choose the SQL names directly.UDF/UDTF bindings are intentionally out of scope and will be added in a separate PR once the DataFrame API gains UDF support in general.
Brief change log
pyflink/dataframe/sql.pyimplementingsql(query, *, auto_bind=True, **bindings), exported aspyflink.dataframe.sqlValueErroron temporary-view collisions,TypeErrorfor non-DataFrame values), take precedence over auto-bind, and are the intentional way to shadow a permanent catalog tableValueErrorpointing toTableEnvironment.execute_sql()docs/reference/pyflink.dataframe/sql.rst)Verifying this change
This change added tests and can be verified as follows:
pyflink/dataframe/tests/test_sql.py(27 tests) covering: SELECT-only enforcement and error translation, auto-bind over locals/globals (including locals-over-globals precedence), joins across auto-bound DataFrames, explicit bindings with and without auto-bind, precedence on name collisions, strict collision errors for explicit bindings, warn-and-skip on collisions with existing temporary views and permanent tables, identifier validation (invalid names, unicode names, SQL keywords via backtick quoting), cleanup after success / query failure / partial registration failure, rejection of DataFrames from a foreign TableEnvironment, rejection of UDF bindings, and composition of the result with the DataFrame APIcd flink-python && python -m pytest pyflink/dataframe/tests/test_sql.pyDoes this pull request potentially affect one of the following parts:
@Public(Evolving): yes, the newsql()function is annotated@PublicEvolvingDocumentation
sql()docstring with examplesWas generative AI tooling used to co-author this PR?
Generated-by: Claude Code 2.1.250 (Claude Fable 5)