[
https://issues.apache.org/jira/browse/FLINK-40197?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=18108250#comment-18108250
]
Federico Dolce commented on FLINK-40197:
----------------------------------------
Hi [~dianfu] . I have this implemented, but I've spent a couple of days looking
for other ways of doing the auto bindings.
I don't have a definitively better solution than what's described, but I want
to highlight a few issues with the implementation as it is.
{quote}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.
{quote}
So, this implies that every DataFrame object in the caller's scope is
registered and dropped at every pf.sql call.
I know this is intended, but it means that, for example, in a notebook/script
with 10 DataFrames declared, a pf.sql("SELECT 1") creates and drops 10
temporary views that are never used. The drops can also fail, which adds noise,
and if a DataFrame in scope is assigned to a variable whose name is not a valid
SQL identifier, users get warnings about it even if it's never used in a query.
All this to say that it would be nice to limit which DataFrames are registered
as temporary views instead of registering everything every time, especially
considering that passing explicit bindings doesn't skip the auto-binding scan.
I've tried a few things, none of which I found optimal, so I'd like your input
on whether you think these issues are relevant. I'm fine with the original
approach if you've already gone through this and think the trade-offs are
acceptable.
The first idea is to find a way to extract which table names are referenced in
the SQL string.
* Using regexps kind of works, but I think it's a bad idea because you have to
handle SQL syntax properly (string literals, comments, quoting) or you keep
hitting edge cases where things don't work.
* I tried hooking into the Java-side parser to extract references, but there's
no easy way to extract references from an unvalidated statement, and we'd need
to add methods to the Java API (the Parser interface is @PublicEvolving) that
only serve this Python implementation, which I'm not sure is OK in the context
of this FLIP.
* At that point, I tried to instead validate the statement and read back the
errors. This too works, but we end up relying on Calcite's error message
format, and since only the first error is reported, we have to resubmit the
statement in a loop until there are no more "Object 'x' not found" errors,
which is quite ugly. The situation might be improved by making changes in the
java side, but again might go out of scope.
Another possible approach would be to use explicit placeholders, so that we
only look at variable names the user actually mentions. A middle way between
explicit bindings and auto-binding: pf.sql("SELECT * FROM \{df}") or
pf.sql("SELECT * FROM \{table}", table=df). But that's not the auto-binding
this issue asks for, so I'm here to ask what you think about this.
> 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)
