dianfu commented on code in PR #29146:
URL: https://github.com/apache/flink/pull/29146#discussion_r3976445004
##########
flink-python/pyflink/dataframe/sql.py:
##########
@@ -257,6 +315,97 @@ def _register_bindings(
_warn_skipped(name, f"registration failed: {e}")
continue
registered.append(name)
+ return registered
+
+
+def _register_functions(
+ t_env: TableEnvironment,
+ explicit: Dict[str, _DataFrameUDFWrapper],
+ auto: Dict[str, _DataFrameUDFWrapper],
+) -> List[str]:
+ """
+ Register explicit and auto-collected UDFs as temporary system functions
and return
+ the registered names. Registration is all-or-nothing: if an explicit
binding is
+ rejected, the functions registered before it are dropped again before
raising.
+
+ System functions are looked up by bare name independently of the current
catalog
+ and database, which matches how a Python name is referenced in the query.
Function
+ names are case-insensitive: the catalog normalizes them to lower case, so
+ collisions are checked case-insensitively. Mirroring views, explicit
bindings may
+ shadow built-in and permanent catalog functions but never an existing
temporary
+ function; auto-bind shadows nothing.
+ """
+ if not explicit and not auto:
+ return []
+ # list_user_defined_functions() covers temporary and permanent functions
alike;
+ # the permanent ones are those the current catalog lists for the current
database.
+ user_defined = {f.lower() for f in t_env.list_user_defined_functions()}
+ temporary_functions = user_defined - _permanent_functions(t_env)
Review Comment:
The temporary functions cannot be derived by subtracting these two name
sets. `list_user_defined_functions()` merges temporary system, temporary
catalog, and permanent catalog functions, so the source information has already
been lost.
For example, if both a permanent function `f` and a temporary catalog
function `f` exist:
- `user_defined == {"f"}`
- `_permanent_functions(t_env) == {"f"}`
- `temporary_functions == set()`
An explicit `f=my_udf` binding is then allowed and registered as a temporary
system function, which shadows the existing temporary function instead of
raising the documented `ValueError`.
##########
flink-python/pyflink/dataframe/sql.py:
##########
@@ -257,6 +315,97 @@ def _register_bindings(
_warn_skipped(name, f"registration failed: {e}")
continue
registered.append(name)
+ return registered
+
+
+def _register_functions(
+ t_env: TableEnvironment,
+ explicit: Dict[str, _DataFrameUDFWrapper],
+ auto: Dict[str, _DataFrameUDFWrapper],
+) -> List[str]:
+ """
+ Register explicit and auto-collected UDFs as temporary system functions
and return
+ the registered names. Registration is all-or-nothing: if an explicit
binding is
+ rejected, the functions registered before it are dropped again before
raising.
+
+ System functions are looked up by bare name independently of the current
catalog
+ and database, which matches how a Python name is referenced in the query.
Function
+ names are case-insensitive: the catalog normalizes them to lower case, so
+ collisions are checked case-insensitively. Mirroring views, explicit
bindings may
+ shadow built-in and permanent catalog functions but never an existing
temporary
+ function; auto-bind shadows nothing.
+ """
+ if not explicit and not auto:
+ return []
+ # list_user_defined_functions() covers temporary and permanent functions
alike;
+ # the permanent ones are those the current catalog lists for the current
database.
+ user_defined = {f.lower() for f in t_env.list_user_defined_functions()}
+ temporary_functions = user_defined - _permanent_functions(t_env)
+ # Explicit bindings take precedence on name collisions, so only the
remaining
+ # auto-bound candidates need the built-in function names. list_functions()
covers
+ # those as well, but is comparatively expensive, so skip it when nothing
needs it.
+ auto_candidates = {name: value for name, value in auto.items() if name not
in explicit}
Review Comment:
Should we normalize the explicit names first and filter auto candidates
using the normalized name set?
For the following example:
add_one = pf.udf(...)
pf.sql("SELECT ADD_ONE(a)", ADD_ONE=override)
`add_one` is not removed from `auto_candidates` because this comparison uses
the original Python keys. The explicit function is registered first, and the
auto-bound `add_one` then produces a misleading "function already exists"
warning.
##########
flink-python/pyflink/dataframe/sql.py:
##########
@@ -90,48 +103,84 @@ def sql(query: str, *, auto_bind: bool = True, **bindings:
DataFrame) -> DataFra
... auto_bind=False,
... src=df1,
... )
+ >>> # UDFs are bound the same way, under their variable or keyword name
+ >>> @pf.udf
+ ... def add_one(value: int) -> int:
+ ... return value + 1
+ >>> pf.sql("SELECT add_one(a) AS a1 FROM df1")
+ >>> pf.sql("SELECT inc(a) FROM src", auto_bind=False, src=df1,
inc=add_one)
>>> # 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] = {}
+
+ # Collect auto bindings first.
+ variables = {}
+ # Gather the variables in the namespace
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 = _resolve_table_environment(bindings, auto_bindings)
- registered: List[str] = []
+ if frame := inspect.currentframe():
+ if outer_frame := frame.f_back:
+ variables = {**outer_frame.f_globals, **outer_frame.f_locals}
+ # Suggested by python docs
+ del outer_frame
+ del frame
+ auto_frames = _get_dataframes(variables)
+ auto_udfs = _get_udfs(variables)
+
+ # Check that explicit bindings are the correct type first
+ for name, value in bindings.items():
+ if not isinstance(value, _BINDABLE_TYPES):
+ raise TypeError(
+ f"sql() binding '{name}' must be a DataFrame or a UDF created
with "
+ f"pyflink.dataframe.udf, got {type(value).__name__}"
+ )
+ explicit_frames = _get_dataframes(bindings)
+ explicit_udfs = _get_udfs(bindings)
+
+ t_env = _resolve_table_environment(explicit_frames, auto_frames)
+ # Each registration step is all-or-nothing: it rolls back its own partial
work on
+ # failure and only returns names on success, so a step that raises leaves
nothing
+ # of its own behind and the finally block only drops what earlier steps
returned.
+ views: List[str] = []
+ functions: List[str] = []
try:
- _register_bindings(t_env, bindings, auto_bindings, registered)
+ views = _register_views(t_env, explicit_frames, auto_frames)
Review Comment:
The registered names are assigned to `views` / `functions` only after each
helper returns successfully. If an exception occurs after a partial auto-bind,
the outer `finally` cannot see or clean up the objects that were already
registered.
##########
flink-python/pyflink/dataframe/sql.py:
##########
@@ -33,49 +34,61 @@
_LOG = logging.getLogger(__name__)
+_Binding = Union[DataFrame, _DataFrameUDFWrapper]
Review Comment:
The public type contract here does not match the return type of `pf.udf()`.
`sql()` accepts `_DataFrameUDFWrapper`, while the `pf.udf()` overloads are
declared to return `Callable[..., Expression]`.
The documented usage works at runtime, but it will be rejected by static
type checkers:
df = pf.from_dict({"a": [1]})
@pf.udf
def inc(value: int) -> int:
return value + 1
result = pf.sql(
"SELECT inc(a) FROM df",
auto_bind=False,
df=df,
inc=inc,
)
At runtime, `inc` is an `_DataFrameUDFWrapper`. However, the `pf.udf()`
overload declares it as `Callable[..., Expression]`, which is incompatible with
the `_Binding` annotation expected by `pf.sql()`.
Do you think it makes sense to update the return type of py.udf from
`Callable[..., Expression]` to `_DataFrameUDFWrapper` though it's not directly
introduced by this PR?
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
