Dian Fu created FLINK-40193:
-------------------------------
Summary: Add join_lateral support to DataFrame API
Key: FLINK-40193
URL: https://issues.apache.org/jira/browse/FLINK-40193
Project: Flink
Issue Type: Sub-task
Components: API / Python
Reporter: Dian Fu
Fix For: 2.4.0
def join_lateral(
self,
table_function_call: Union[DataFrameUDTFCall, Expression],
*,
on: Expression = None,
ignore_empty: bool = True,
) -> "DataFrame"
Join each input row with the rows a table function (UDTF) emits for it — a
lateral (correlated) join. Where explode unnests a built-in collection column
and flat_map applies a whole-row UDTF, join_lateral joins against a UDTF
invoked with explicit column arguments, keeping the original columns alongside
the UDTF output.
* table_function_call — a UDTF call built by invoking a @udtf function with
column expressions, e.g. chars(pf.col("text")). Name the output field(s) with
.alias(...); names may instead be inferred from a named TypedDict / struct
return_dtype. A raw Expression must name its outputs with .alias.
* on — optional boolean join predicate, matching the Table API lateral-join
predicate.
* ignore_empty — True (default) is inner lateral-join semantics: input rows
whose UDTF emits no rows are dropped. False is left-outer lateral-join
semantics: such rows are kept with NULL UDTF output.
Example:
{code:python}
from typing import Iterator, TypedDict
import pyflink.dataframe as pf
@udtf
def chars(text: str) -> Iterator[str]:
for ch in text:
yield ch
# Inner lateral join: rows with empty text are dropped
df.join_lateral(chars(pf.col("text")).alias("ch"))
# Left-outer lateral join: rows with empty text are kept with NULL ch
df.join_lateral(chars(pf.col("text")).alias("ch"), ignore_empty=False)
# With an extra join predicate
df.join_lateral(chars(pf.col("text")).alias("ch"), on=pf.col("id") > 0)
# Multi-field output via a TypedDict return type (names inferred)
class Token(TypedDict):
word: str
length: int
@udtf
def tokenize(text: str) -> Iterator[Token]:
for word in text.split():
yield {"word": word, "length": len(word)}
df.join_lateral(tokenize(pf.col("text")))
{code}
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
