[
https://issues.apache.org/jira/browse/FLINK-40197?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=18109156#comment-18109156
]
Federico Dolce commented on FLINK-40197:
----------------------------------------
Ok, I'll keep it as described in this issue then.
Just a couple of points:
The last solution I proposed was a bit different than what table api currently
supports. Looking into it, Table API allows you to compose a string like
`t_env.sqlQuery("SELECT * FROM %s" % table)`. This works because
`Table.__str__` goes to the java side that registers the table as temporary
view, and also never drops it. I checked, and we don't do that in DataFrame
currently.
My idea was to replace auto bindings with only explicit ones (even using a
different character instead of {}): `pf.sql("SELECT * FROM ?table")`. In that
case, we only look for a variable named "table" and register it, instead of
doing it for all of them, and it's only slightly less magic then
`pf.sql("SELECT * FROM table")`. But we still need to use a regexp or something
to extract the names, which I don't like too much.
Anyway, mixing auto bindings and the way Table API does it, we register views
twice, but I'm not sure we can drop the ones defined from
`DataFrame.__str__->self._j_table.toString()`. We said registering is not too
much of a problem, but not dropping is not ideal. Anyway I think this might be
worth looking at in a separate issue, or discuss about it in the PR with the
implementation under our eyes. I'll open the PR without this at first
> 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
> Assignee: Federico Dolce
> 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)
