[
https://issues.apache.org/jira/browse/FLINK-40193?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Dian Fu reassigned FLINK-40193:
-------------------------------
Assignee: Liu Liu
> 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
> Assignee: Liu Liu
> Priority: Major
> 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)
