You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by "fdolce (via GitHub)" <gi...@apache.org> on 2026/08/28 12:57:01 UTC
[PR] [FLINK-40197][python] Add sql() to the DataFrame API [flink]
fdolce opened a new pull request, #29036:
URL: https://github.com/apache/flink/pull/29036
## What is the purpose of the change
Adds `pyflink.dataframe.sql()` (FLINK-40197), which executes a SQL SELECT query and returns the result as a `DataFrame`, 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
- Added `pyflink/dataframe/sql.py` implementing `sql(query, *, auto_bind=True, **bindings)`, exported as `pyflink.dataframe.sql`
- Auto-bind is best-effort: it warns and skips names that are not valid SQL identifiers, collide with existing tables/views, or belong to a different TableEnvironment; it never shadows permanent catalog objects
- Explicit bindings are strict (`ValueError` on temporary-view collisions, `TypeError` for non-DataFrame values), take precedence over auto-bind, and are the intentional way to shadow a permanent catalog table
- Only SELECT-style queries are accepted; other statements raise `ValueError` pointing to `TableEnvironment.execute_sql()`
- Registered views are always dropped after the call, including on failure and on partially-completed registration
- Added the Sphinx API reference page (`docs/reference/pyflink.dataframe/sql.rst`)
## Verifying this change
This change added tests and can be verified as follows:
- Added `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 API
- Run with: `cd flink-python && python -m pytest pyflink/dataframe/tests/test_sql.py`
## 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 new `sql()` function is annotated `@PublicEvolving`
- The serializers: no
- The runtime per-record code paths (performance sensitive): no (view registration happens once per call at query compilation time)
- Anything that affects deployment or recovery: 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 (Sphinx API reference page) and the `sql()` docstring with examples
---
##### Was generative AI tooling used to co-author this PR?
- [X] Yes (please specify the tool below)
Generated-by: Claude Code 2.1.250 (Claude Fable 5)
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-40197][python] Add sql() to the DataFrame API [flink]
Posted by "fdolce (via GitHub)" <gi...@apache.org>.
fdolce commented on code in PR #29036:
URL: https://github.com/apache/flink/pull/29036#discussion_r3901707430
##########
flink-python/pyflink/dataframe/sql.py:
##########
@@ -0,0 +1,215 @@
+################################################################################
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+################################################################################
+
+import inspect
+import warnings
+from typing import Any, Dict, List
+
+from py4j.protocol import Py4JJavaError
+
+from pyflink.dataframe.context import get_or_create_table_environment
+from pyflink.dataframe.dataframe import DataFrame
+from pyflink.table import Table, TableEnvironment
+from pyflink.util.api_stability_decorators import PublicEvolving
+from pyflink.util.java_utils import is_instance_of
+
+__all__ = ["sql"]
+
+
+@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`.
Review Comment:
Mh right. If it's ok to allow those too, then yes I'll make the comment less specific.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-40197][python] Add sql() to the DataFrame API [flink]
Posted by "fdolce (via GitHub)" <gi...@apache.org>.
fdolce commented on code in PR #29036:
URL: https://github.com/apache/flink/pull/29036#discussion_r3904530352
##########
flink-python/pyflink/dataframe/sql.py:
##########
@@ -0,0 +1,215 @@
+################################################################################
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+################################################################################
+
+import inspect
+import warnings
+from typing import Any, Dict, List
+
+from py4j.protocol import Py4JJavaError
+
+from pyflink.dataframe.context import get_or_create_table_environment
+from pyflink.dataframe.dataframe import DataFrame
+from pyflink.table import Table, TableEnvironment
+from pyflink.util.api_stability_decorators import PublicEvolving
+from pyflink.util.java_utils import is_instance_of
+
+__all__ = ["sql"]
+
+
+@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`.
+
+ Only SELECT queries are supported (no INSERT / DDL). The referenced DataFrames are
+ registered as temporary views for the duration of the call and dropped afterwards.
+ The result can be further transformed with the DataFrame API.
+
+ When ``auto_bind`` is ``True`` (the default), the caller's local and global variables
+ are scanned for :class:`DataFrame` objects and each is registered under its Python
+ variable name. Auto-binding is best-effort: it warns and skips names that are not
+ valid SQL identifiers or that collide with an existing table or view, and it never
+ shadows permanent catalog objects.
+
+ Explicit keyword ``bindings`` define the SQL names directly. They are strict
+ (conflicts with existing temporary views raise :class:`ValueError`), take
+ precedence over auto-bind on name collisions, and are required to intentionally
+ shadow a permanent catalog table or view.
+
+ :param query: The SELECT query to execute.
+ :param auto_bind: Whether to scan the caller's variables for DataFrames.
+ :param bindings: Explicit name to :class:`DataFrame` bindings.
+ :return: The query result.
+ :raises ValueError: If the query is not a SELECT query, or an explicit binding
+ conflicts with an existing temporary view.
+ :raises TypeError: If an explicit binding is not a :class:`DataFrame`.
+
+ Example::
+
+ >>> import pyflink.dataframe as pf
+ >>> df1 = pf.from_dict({"a": [1, 2, 3], "b": ["x", "y", "z"]})
+ >>> df2 = pf.from_dict({"a": [1, 2, 3], "c": ["p", "q", "r"]})
+ >>> # Auto-bind: df1 / df2 are registered under their variable names
+ >>> joined = pf.sql("SELECT df1.a, b, c FROM df1 JOIN df2 ON df1.a = df2.a")
+ >>> # Explicit bindings: pick the SQL names, turn off scanning
+ >>> result = pf.sql(
+ ... "SELECT * FROM src WHERE a > 1",
+ ... auto_bind=False,
+ ... src=df1,
+ ... )
+ >>> # Mix SQL and the DataFrame API
+ >>> pf.sql("SELECT a, b FROM df1").filter(pf.col("a") > 1).to_pandas()
+
+ .. versionadded:: 2.4.0
+ """
+ if not isinstance(query, str):
+ raise TypeError("query must be a string")
+ auto_bindings: Dict[str, DataFrame] = {}
+ if auto_bind:
+ frame = inspect.currentframe()
+ caller = frame.f_back if frame is not None else None
+ try:
+ if caller is not None:
+ # Locals take precedence over globals.
+ namespace = {**caller.f_globals, **caller.f_locals}
+ auto_bindings = {
+ name: value
+ for name, value in namespace.items()
+ if isinstance(value, DataFrame)
+ }
+ finally:
+ del frame, caller
+ t_env = get_or_create_table_environment()
+ registered: List[str] = []
+ try:
+ _register_bindings(t_env, bindings, auto_bindings, registered)
+ return DataFrame(_execute_query(t_env, query))
+ finally:
+ for name in registered:
+ # Best-effort cleanup: dropping must not mask an exception raised by the
+ # query itself, but only names registered by this call are dropped, so a
+ # failure is an anomaly the user should hear about.
+ try:
+ t_env.drop_temporary_view(name)
+ except Exception as e:
+ warnings.warn(
+ f"sql() failed to drop temporary view '{name}': {e}",
+ UserWarning,
+ )
+
+
+def _execute_query(t_env: TableEnvironment, query: str) -> Table:
+ """
+ Run ``query`` through :meth:`TableEnvironment.sql_query`, which parses the statement
+ and rejects anything that is not a single SELECT-style query. Translate that
+ rejection into a plain :class:`ValueError`.
+ """
+ try:
+ return t_env.sql_query(query)
+ except Py4JJavaError as e:
+ if "Unsupported SQL query!" in str(e.java_exception):
+ raise ValueError(
+ "sql() only supports SELECT queries (no INSERT / DDL); use "
+ "TableEnvironment.execute_sql() for other statements."
+ ) from e
+ raise
+
+
+def _is_simple_sql_identifier(t_env: TableEnvironment, name: str) -> bool:
+ """
+ Whether ``name`` is accepted verbatim as a single-part identifier by the SQL parser,
+ i.e. whether registering a temporary view under it can succeed. This is the same
+ validation :meth:`TableEnvironment.create_temporary_view` applies to its path, so
+ keywords like ``order`` pass (queries reference them with backticks) while names
+ that would need quoting or resolve to a different or multi-part path do not.
+ """
+ try:
+ identifier = t_env._j_tenv.getParser().parseIdentifier(name)
+ except Py4JJavaError as e:
+ if not is_instance_of(
+ e.java_exception, "org.apache.flink.table.api.SqlParserException"
+ ):
+ raise
+ return False
+ return (
+ not identifier.getCatalogName().isPresent()
+ and not identifier.getDatabaseName().isPresent()
+ and identifier.getObjectName() == name
+ )
+
+
+def _register_bindings(
+ t_env: TableEnvironment,
+ explicit: Dict[str, Any],
+ auto: Dict[str, DataFrame],
+ registered: List[str],
+) -> None:
+ """
+ Register explicit and auto-collected bindings as temporary views, appending each
+ successful registration to ``registered``.
+ """
+ temporary_tables = set(t_env.list_temporary_tables())
+ # list_tables() covers both permanent and temporary tables and views.
+ all_tables = set(t_env.list_tables())
+
+ for name, value in explicit.items():
+ if not isinstance(value, DataFrame):
+ raise TypeError(
+ f"sql() binding '{name}' must be a DataFrame, got {type(value).__name__}"
+ )
+ if value._table._t_env is not t_env:
+ raise ValueError(
+ f"DataFrame bound as '{name}' belongs to a different TableEnvironment "
+ "than the one used by pyflink.dataframe; use set_table_environment() "
+ "to align them"
+ )
+ if name in temporary_tables:
+ raise ValueError(
+ f"cannot bind '{name}': a temporary table or view with this name "
+ "already exists"
+ )
+ t_env.create_temporary_view(name, value.to_table())
Review Comment:
Fixed
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-40197][python] Add sql() to the DataFrame API [flink]
Posted by "fdolce (via GitHub)" <gi...@apache.org>.
fdolce commented on code in PR #29036:
URL: https://github.com/apache/flink/pull/29036#discussion_r3901686104
##########
flink-python/pyflink/dataframe/sql.py:
##########
@@ -0,0 +1,215 @@
+################################################################################
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+################################################################################
+
+import inspect
+import warnings
+from typing import Any, Dict, List
+
+from py4j.protocol import Py4JJavaError
+
+from pyflink.dataframe.context import get_or_create_table_environment
+from pyflink.dataframe.dataframe import DataFrame
+from pyflink.table import Table, TableEnvironment
+from pyflink.util.api_stability_decorators import PublicEvolving
+from pyflink.util.java_utils import is_instance_of
+
+__all__ = ["sql"]
+
+
+@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`.
+
+ Only SELECT queries are supported (no INSERT / DDL). The referenced DataFrames are
+ registered as temporary views for the duration of the call and dropped afterwards.
+ The result can be further transformed with the DataFrame API.
+
+ When ``auto_bind`` is ``True`` (the default), the caller's local and global variables
+ are scanned for :class:`DataFrame` objects and each is registered under its Python
+ variable name. Auto-binding is best-effort: it warns and skips names that are not
+ valid SQL identifiers or that collide with an existing table or view, and it never
+ shadows permanent catalog objects.
+
+ Explicit keyword ``bindings`` define the SQL names directly. They are strict
+ (conflicts with existing temporary views raise :class:`ValueError`), take
+ precedence over auto-bind on name collisions, and are required to intentionally
+ shadow a permanent catalog table or view.
+
+ :param query: The SELECT query to execute.
+ :param auto_bind: Whether to scan the caller's variables for DataFrames.
+ :param bindings: Explicit name to :class:`DataFrame` bindings.
+ :return: The query result.
+ :raises ValueError: If the query is not a SELECT query, or an explicit binding
+ conflicts with an existing temporary view.
+ :raises TypeError: If an explicit binding is not a :class:`DataFrame`.
+
+ Example::
+
+ >>> import pyflink.dataframe as pf
+ >>> df1 = pf.from_dict({"a": [1, 2, 3], "b": ["x", "y", "z"]})
+ >>> df2 = pf.from_dict({"a": [1, 2, 3], "c": ["p", "q", "r"]})
+ >>> # Auto-bind: df1 / df2 are registered under their variable names
+ >>> joined = pf.sql("SELECT df1.a, b, c FROM df1 JOIN df2 ON df1.a = df2.a")
+ >>> # Explicit bindings: pick the SQL names, turn off scanning
+ >>> result = pf.sql(
+ ... "SELECT * FROM src WHERE a > 1",
+ ... auto_bind=False,
+ ... src=df1,
+ ... )
+ >>> # Mix SQL and the DataFrame API
+ >>> pf.sql("SELECT a, b FROM df1").filter(pf.col("a") > 1).to_pandas()
+
+ .. versionadded:: 2.4.0
+ """
+ if not isinstance(query, str):
+ raise TypeError("query must be a string")
+ auto_bindings: Dict[str, DataFrame] = {}
+ if auto_bind:
+ frame = inspect.currentframe()
+ caller = frame.f_back if frame is not None else None
+ try:
+ if caller is not None:
+ # Locals take precedence over globals.
+ namespace = {**caller.f_globals, **caller.f_locals}
+ auto_bindings = {
+ name: value
+ for name, value in namespace.items()
+ if isinstance(value, DataFrame)
+ }
+ finally:
+ del frame, caller
+ t_env = get_or_create_table_environment()
Review 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?
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-40197][python] Add sql() to the DataFrame API [flink]
Posted by "dianfu (via GitHub)" <gi...@apache.org>.
dianfu closed pull request #29036: [FLINK-40197][python] Add sql() to the DataFrame API
URL: https://github.com/apache/flink/pull/29036
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-40197][python] Add sql() to the DataFrame API [flink]
Posted by "dianfu (via GitHub)" <gi...@apache.org>.
dianfu commented on code in PR #29036:
URL: https://github.com/apache/flink/pull/29036#discussion_r3893986419
##########
flink-python/pyflink/dataframe/sql.py:
##########
@@ -0,0 +1,215 @@
+################################################################################
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+################################################################################
+
+import inspect
+import warnings
+from typing import Any, Dict, List
+
+from py4j.protocol import Py4JJavaError
+
+from pyflink.dataframe.context import get_or_create_table_environment
+from pyflink.dataframe.dataframe import DataFrame
+from pyflink.table import Table, TableEnvironment
+from pyflink.util.api_stability_decorators import PublicEvolving
+from pyflink.util.java_utils import is_instance_of
+
+__all__ = ["sql"]
+
+
+@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`.
Review 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`
##########
flink-python/pyflink/dataframe/sql.py:
##########
@@ -0,0 +1,215 @@
+################################################################################
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+################################################################################
+
+import inspect
+import warnings
+from typing import Any, Dict, List
+
+from py4j.protocol import Py4JJavaError
+
+from pyflink.dataframe.context import get_or_create_table_environment
+from pyflink.dataframe.dataframe import DataFrame
+from pyflink.table import Table, TableEnvironment
+from pyflink.util.api_stability_decorators import PublicEvolving
+from pyflink.util.java_utils import is_instance_of
+
+__all__ = ["sql"]
+
+
+@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`.
+
+ Only SELECT queries are supported (no INSERT / DDL). The referenced DataFrames are
+ registered as temporary views for the duration of the call and dropped afterwards.
+ The result can be further transformed with the DataFrame API.
+
+ When ``auto_bind`` is ``True`` (the default), the caller's local and global variables
+ are scanned for :class:`DataFrame` objects and each is registered under its Python
+ variable name. Auto-binding is best-effort: it warns and skips names that are not
+ valid SQL identifiers or that collide with an existing table or view, and it never
+ shadows permanent catalog objects.
+
+ Explicit keyword ``bindings`` define the SQL names directly. They are strict
+ (conflicts with existing temporary views raise :class:`ValueError`), take
+ precedence over auto-bind on name collisions, and are required to intentionally
+ shadow a permanent catalog table or view.
+
+ :param query: The SELECT query to execute.
+ :param auto_bind: Whether to scan the caller's variables for DataFrames.
+ :param bindings: Explicit name to :class:`DataFrame` bindings.
+ :return: The query result.
+ :raises ValueError: If the query is not a SELECT query, or an explicit binding
+ conflicts with an existing temporary view.
+ :raises TypeError: If an explicit binding is not a :class:`DataFrame`.
+
+ Example::
+
+ >>> import pyflink.dataframe as pf
+ >>> df1 = pf.from_dict({"a": [1, 2, 3], "b": ["x", "y", "z"]})
+ >>> df2 = pf.from_dict({"a": [1, 2, 3], "c": ["p", "q", "r"]})
+ >>> # Auto-bind: df1 / df2 are registered under their variable names
+ >>> joined = pf.sql("SELECT df1.a, b, c FROM df1 JOIN df2 ON df1.a = df2.a")
+ >>> # Explicit bindings: pick the SQL names, turn off scanning
+ >>> result = pf.sql(
+ ... "SELECT * FROM src WHERE a > 1",
+ ... auto_bind=False,
+ ... src=df1,
+ ... )
+ >>> # Mix SQL and the DataFrame API
+ >>> pf.sql("SELECT a, b FROM df1").filter(pf.col("a") > 1).to_pandas()
+
+ .. versionadded:: 2.4.0
+ """
+ if not isinstance(query, str):
+ raise TypeError("query must be a string")
+ auto_bindings: Dict[str, DataFrame] = {}
+ if auto_bind:
+ frame = inspect.currentframe()
+ caller = frame.f_back if frame is not None else None
+ try:
+ if caller is not None:
+ # Locals take precedence over globals.
+ namespace = {**caller.f_globals, **caller.f_locals}
+ auto_bindings = {
+ name: value
+ for name, value in namespace.items()
+ if isinstance(value, DataFrame)
+ }
+ finally:
+ del frame, caller
+ t_env = get_or_create_table_environment()
Review 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.
##########
flink-python/pyflink/dataframe/sql.py:
##########
@@ -0,0 +1,215 @@
+################################################################################
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+################################################################################
+
+import inspect
+import warnings
+from typing import Any, Dict, List
+
+from py4j.protocol import Py4JJavaError
+
+from pyflink.dataframe.context import get_or_create_table_environment
+from pyflink.dataframe.dataframe import DataFrame
+from pyflink.table import Table, TableEnvironment
+from pyflink.util.api_stability_decorators import PublicEvolving
+from pyflink.util.java_utils import is_instance_of
+
+__all__ = ["sql"]
+
+
+@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`.
+
+ Only SELECT queries are supported (no INSERT / DDL). The referenced DataFrames are
+ registered as temporary views for the duration of the call and dropped afterwards.
+ The result can be further transformed with the DataFrame API.
+
+ When ``auto_bind`` is ``True`` (the default), the caller's local and global variables
+ are scanned for :class:`DataFrame` objects and each is registered under its Python
+ variable name. Auto-binding is best-effort: it warns and skips names that are not
+ valid SQL identifiers or that collide with an existing table or view, and it never
+ shadows permanent catalog objects.
+
+ Explicit keyword ``bindings`` define the SQL names directly. They are strict
+ (conflicts with existing temporary views raise :class:`ValueError`), take
+ precedence over auto-bind on name collisions, and are required to intentionally
+ shadow a permanent catalog table or view.
+
+ :param query: The SELECT query to execute.
+ :param auto_bind: Whether to scan the caller's variables for DataFrames.
+ :param bindings: Explicit name to :class:`DataFrame` bindings.
+ :return: The query result.
+ :raises ValueError: If the query is not a SELECT query, or an explicit binding
+ conflicts with an existing temporary view.
+ :raises TypeError: If an explicit binding is not a :class:`DataFrame`.
+
+ Example::
+
+ >>> import pyflink.dataframe as pf
+ >>> df1 = pf.from_dict({"a": [1, 2, 3], "b": ["x", "y", "z"]})
+ >>> df2 = pf.from_dict({"a": [1, 2, 3], "c": ["p", "q", "r"]})
+ >>> # Auto-bind: df1 / df2 are registered under their variable names
+ >>> joined = pf.sql("SELECT df1.a, b, c FROM df1 JOIN df2 ON df1.a = df2.a")
+ >>> # Explicit bindings: pick the SQL names, turn off scanning
+ >>> result = pf.sql(
+ ... "SELECT * FROM src WHERE a > 1",
+ ... auto_bind=False,
+ ... src=df1,
+ ... )
+ >>> # Mix SQL and the DataFrame API
+ >>> pf.sql("SELECT a, b FROM df1").filter(pf.col("a") > 1).to_pandas()
+
+ .. versionadded:: 2.4.0
+ """
+ if not isinstance(query, str):
+ raise TypeError("query must be a string")
+ auto_bindings: Dict[str, DataFrame] = {}
+ if auto_bind:
+ frame = inspect.currentframe()
+ caller = frame.f_back if frame is not None else None
+ try:
+ if caller is not None:
+ # Locals take precedence over globals.
+ namespace = {**caller.f_globals, **caller.f_locals}
+ auto_bindings = {
+ name: value
+ for name, value in namespace.items()
+ if isinstance(value, DataFrame)
+ }
+ finally:
+ del frame, caller
+ t_env = get_or_create_table_environment()
+ registered: List[str] = []
+ try:
+ _register_bindings(t_env, bindings, auto_bindings, registered)
+ return DataFrame(_execute_query(t_env, query))
+ finally:
+ for name in registered:
+ # Best-effort cleanup: dropping must not mask an exception raised by the
+ # query itself, but only names registered by this call are dropped, so a
+ # failure is an anomaly the user should hear about.
+ try:
+ t_env.drop_temporary_view(name)
+ except Exception as e:
+ warnings.warn(
+ f"sql() failed to drop temporary view '{name}': {e}",
+ UserWarning,
+ )
+
+
+def _execute_query(t_env: TableEnvironment, query: str) -> Table:
+ """
+ Run ``query`` through :meth:`TableEnvironment.sql_query`, which parses the statement
+ and rejects anything that is not a single SELECT-style query. Translate that
+ rejection into a plain :class:`ValueError`.
+ """
+ try:
+ return t_env.sql_query(query)
+ except Py4JJavaError as e:
+ if "Unsupported SQL query!" in str(e.java_exception):
+ raise ValueError(
+ "sql() only supports SELECT queries (no INSERT / DDL); use "
+ "TableEnvironment.execute_sql() for other statements."
+ ) from e
+ raise
+
+
+def _is_simple_sql_identifier(t_env: TableEnvironment, name: str) -> bool:
+ """
+ Whether ``name`` is accepted verbatim as a single-part identifier by the SQL parser,
+ i.e. whether registering a temporary view under it can succeed. This is the same
+ validation :meth:`TableEnvironment.create_temporary_view` applies to its path, so
+ keywords like ``order`` pass (queries reference them with backticks) while names
+ that would need quoting or resolve to a different or multi-part path do not.
+ """
+ try:
+ identifier = t_env._j_tenv.getParser().parseIdentifier(name)
+ except Py4JJavaError as e:
+ if not is_instance_of(
+ e.java_exception, "org.apache.flink.table.api.SqlParserException"
+ ):
+ raise
+ return False
+ return (
+ not identifier.getCatalogName().isPresent()
+ and not identifier.getDatabaseName().isPresent()
+ and identifier.getObjectName() == name
+ )
+
+
+def _register_bindings(
+ t_env: TableEnvironment,
+ explicit: Dict[str, Any],
+ auto: Dict[str, DataFrame],
+ registered: List[str],
+) -> None:
+ """
+ Register explicit and auto-collected bindings as temporary views, appending each
+ successful registration to ``registered``.
+ """
+ temporary_tables = set(t_env.list_temporary_tables())
+ # list_tables() covers both permanent and temporary tables and views.
+ all_tables = set(t_env.list_tables())
+
+ for name, value in explicit.items():
+ if not isinstance(value, DataFrame):
+ raise TypeError(
+ f"sql() binding '{name}' must be a DataFrame, got {type(value).__name__}"
+ )
+ if value._table._t_env is not t_env:
+ raise ValueError(
+ f"DataFrame bound as '{name}' belongs to a different TableEnvironment "
+ "than the one used by pyflink.dataframe; use set_table_environment() "
+ "to align them"
+ )
+ if name in temporary_tables:
+ raise ValueError(
+ f"cannot bind '{name}': a temporary table or view with this name "
+ "already exists"
+ )
+ t_env.create_temporary_view(name, value.to_table())
Review Comment:
Auto-bound names are validated before registration, while explicit names are passed directly to create_temporary_view().
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-40197][python] Add sql() to the DataFrame API [flink]
Posted by "flinkbot (via GitHub)" <gi...@apache.org>.
flinkbot commented on PR #29036:
URL: https://github.com/apache/flink/pull/29036#issuecomment-5452838461
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "de67f336fb86c3d4e235417112be19a7720fdd3a",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "de67f336fb86c3d4e235417112be19a7720fdd3a",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* de67f336fb86c3d4e235417112be19a7720fdd3a UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run azure` re-run the last Azure build
</details>
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org