[
https://issues.apache.org/jira/browse/FLINK-40197?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=18105921#comment-18105921
]
Federico Dolce edited comment on FLINK-40197 at 8/19/26 2:13 PM:
-----------------------------------------------------------------
Hi [~dianfu] . I'd like to start working on this issue if it's ok for you,
could you assign me to it?
was (Author: JIRAUSER310156):
Hi [~dianfu] . I'd like to start working on this is issue if it's ok for you,
could you assign me to it?
> Add sql bindings to DataFrame API
> ---------------------------------
>
> Key: FLINK-40197
> URL: https://issues.apache.org/jira/browse/FLINK-40197
> Project: Flink
> Issue Type: Sub-task
> Components: API / Python
> Reporter: Dian Fu
> Priority: Major
> Fix For: 2.4.0
>
>
> def sql(
> query: str,
> *,
> auto_bind: bool = True,
> **bindings,
> ) -> DataFrame
> Run a SQL SELECT query and return its result as a DataFrame, so SQL and the
> DataFrame API mix freely. DataFrames referenced in the query are exposed as
> temporary views, and @pf.udf / @pf.udtf functions as temporary functions;
> both are registered only for the duration of the call and dropped afterwards.
> There are two ways to make Python objects visible to the query:
> * Auto-bind (default, auto_bind=True): sql scans the caller's local and
> global variables for DataFrame and UDF/UDTF objects and registers each under
> its Python variable name. Auto-bind is best-effort — it warns and skips names
> that are not valid SQL identifiers or that collide with an existing
> table/view/function, and it never shadows permanent catalog objects or
> built-in functions.
> * Explicit bindings: pass keyword arguments to choose the SQL name yourself,
> e.g. pf.sql("SELECT * FROM s", s=df). Explicit bindings are strict (conflicts
> raise ValueError), take precedence over auto-bind on name collisions, and are
> required to intentionally shadow a catalog table/view or a built-in function.
> Only SELECT queries are supported (no INSERT / DDL). The returned DataFrame
> can be further transformed with the DataFrame API.
> Example:
> {code:python}
> 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")
> # UDFs are auto-bound too
> @pf.udf
> def add_one(x: int) -> int:
> return x + 1
> incremented = pf.sql("SELECT add_one(a) AS a1 FROM df1")
> # 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()
> {code}
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
