From bdde8890e2e7575ef5e457e84ce1e3012d25b1cd Mon Sep 17 00:00:00 2001 From: Fabian Hueske Date: Mon, 13 Jul 2026 12:38:01 +0200 Subject: [PATCH 1/2] [FLINK-40131][docs] Document LATERAL SNAPSHOT join Co-Authored-By: Claude Opus 4.8 (1M context) --- .../docs/sql/reference/queries/joins.md | 68 +++++++++++++++++++ .../docs/sql/reference/queries/joins.md | 68 +++++++++++++++++++ 2 files changed, 136 insertions(+) diff --git a/docs/content.zh/docs/sql/reference/queries/joins.md b/docs/content.zh/docs/sql/reference/queries/joins.md index a1e1db2caab8be..d3429c2571d49b 100644 --- a/docs/content.zh/docs/sql/reference/queries/joins.md +++ b/docs/content.zh/docs/sql/reference/queries/joins.md @@ -295,6 +295,74 @@ WHERE - SQL 中可以定义 temporal table DDL，但不能定义 temporal table 函数; - temporal table DDL 和 temporal table function 都支持 temporal join 版本表，但只有 temporal table function 可以 temporal join 任何表/视图的最新版本（即"处理时间 Temporal Join"）。 +LATERAL SNAPSHOT Join +-------------- + +{{< label Streaming >}} + +A `LATERAL SNAPSHOT` join is a *stream enrichment* join that augments an append-only table (the *probe* side) with the current state of an updating table (the *build* side). +Unlike an [event-time temporal join](#event-time-temporal-join), it does not correlate each probe-side row with a specific historical version of the build side. +Instead, every probe-side row is joined with the build-side state that is current at the time the row is processed, similar to a [processing-time temporal join](#processing-time-temporal-join), but with well-defined behavior at query start-up. + +The `LATERAL SNAPSHOT` join is designed for scenarios where the other temporal joins are a poor fit: + +- The build side updates **infrequently**. An event-time temporal join relies on continuous build-side watermarks to emit results, so a build side that rarely advances its watermark stalls the join and lets probe-side state accumulate. A `LATERAL SNAPSHOT` join does not stall when the build side goes idle. +- The application has **low-latency** requirements that are incompatible with the watermark-induced delay of an event-time temporal join. +- The build side has **no primary key**. Event-time and processing-time temporal joins require the build-side primary key to appear in the equi-join condition; a `LATERAL SNAPSHOT` join does not. + +The build side is wrapped in the `SNAPSHOT` table function inside a `LATERAL TABLE` clause. Both `INNER JOIN` and `LEFT [OUTER] JOIN` are supported. The join requires at least one conjunctive equality predicate; additional non-equi predicates are allowed in the `ON` clause. + +```sql +SELECT [column_list] +FROM probe_table +[LEFT] JOIN LATERAL TABLE( + SNAPSHOT( + input => TABLE build_table, + [ load_completed_condition => <'compile_time' | 'user_time'>, ] + [ load_completed_time => , ] + [ load_completed_idle_timeout => , ] + [ state_ttl => ])) AS s +ON probe_table.key = s.key +``` + +The `SNAPSHOT` function accepts the following arguments: + +| Argument | Type | Required | Description | +| --- | --- | --- | --- | +| `input` | TABLE | yes | The build-side table. It may use any [changelog mode]({{< ref "docs/sql/reference/queries/changelog" >}}) (inserts, updates, and deletes) and must declare a [watermark]({{< ref "docs/concepts/sql-table-concepts/time_attributes" >}}#event-time). | +| `load_completed_condition` | STRING | no | Determines when the initial load phase completes. One of `'compile_time'` (default) or `'user_time'`. With `'compile_time'`, the load phase completes once the build-side watermark reaches the wall-clock time at which the query was compiled. With `'user_time'`, it completes once the build-side watermark reaches the explicit `load_completed_time`. | +| `load_completed_time` | TIMESTAMP_LTZ(3) | no | The build-side event time that completes the load phase. Required when `load_completed_condition` is `'user_time'` and must not be set otherwise. | +| `load_completed_idle_timeout` | INTERVAL | no | A processing-time fallback that completes the load phase if the build-side watermark stalls before reaching the configured time. | +| `state_ttl` | INTERVAL | no | Retention time for build-side state. Keys that are not accessed within this duration become eligible for eviction. Only applied during the join phase. Defaults to the pipeline's [state TTL]({{< ref "docs/dev/table/config" >}}#table-exec-state-ttl). | + +The operator processes its inputs in two sequential phases. During the **load phase**, the operator builds up the build-side table by accumulating its changes into state until the load completion condition is met. This allows the build side to load its initial state (for example the current contents of a compacted topic or a change-data-capture snapshot) before any probe-side row is joined, ensuring that early probe-side rows are not joined against an incomplete build side. While the build side is not yet ready, incoming probe-side rows are buffered. The load phase completes when the build-side watermark reaches the wall-clock time at which the query was compiled (`compile_time`, the default), when the build-side watermark reaches `load_completed_time` (`user_time`), or, as a fallback, when the `load_completed_idle_timeout` elapses in processing time without the build-side watermark advancing (which handles build sides that become idle during start-up). + +When transitioning from load to **join phase**, all buffered probe-side rows are joined against the current build-side state and emitted. From this point onwards, each probe-side row is joined and emitted immediately against the build-side state that is current at that moment. Build-side updates continue to be applied to the state and are made visible to later probe-side rows. + +Reusing the tables from the [event-time temporal join](#event-time-temporal-join) example, the following query enriches each order with the conversion rate that is current when the order is processed. In contrast to the temporal join, the join keeps making progress even if the rates are updated only rarely. + +```sql +SELECT + order_id, + price, + orders.currency, + conversion_rate, + order_time +FROM orders +JOIN LATERAL TABLE( + SNAPSHOT(input => TABLE currency_rates)) AS rates +ON orders.currency = rates.currency; +``` + +The probe side must be an append-only table, and its time attributes are preserved in the output. The build side may be an updating table and must declare a watermark. The result is append-only. A build-side rowtime attribute that is projected into the output is materialized as a regular `TIMESTAMP` and is no longer a time attribute. Probe-side watermarks are forwarded downstream in join phase; build-side watermarks are consumed internally and are not propagated. + +Because rows are joined against the build-side state that is current at processing time, the result is **not deterministic**. A given probe-side row may be joined with different build-side versions across different runs, depending on the relative timing of the two inputs. Probe and build-side inputs can be configured with watermark alignment to keep the two inputs roughly aligned on event time, so that a probe-side row tends to be joined with build-side changes of a similar event time. This is a best-effort alignment and does not make the result deterministic. + +The build-side state grows with the number of distinct build-side keys, and during the load phase the buffered probe-side rows add to the state footprint until the operator transitions to join phase. Use `state_ttl` to bound the build-side state for keys that are no longer updated or joined. You can reduce the amount of data that is buffered and processed during the load phase by configuring scan start offsets on the build and probe-side inputs, for example with a `scan.startup.*` [dynamic table option hint]({{< ref "docs/sql/reference/queries/hints" >}}#dynamic-table-options). + +In contrast to [regular joins](#regular-joins), changes on the build side do not retroactively update results that were already emitted. +Compared to [interval joins](#interval-joins), a `LATERAL SNAPSHOT` join does not define a time window within which rows are joined. + Lookup Join -------------- diff --git a/docs/content/docs/sql/reference/queries/joins.md b/docs/content/docs/sql/reference/queries/joins.md index d000a46b822ceb..c8770766e3541f 100644 --- a/docs/content/docs/sql/reference/queries/joins.md +++ b/docs/content/docs/sql/reference/queries/joins.md @@ -300,6 +300,74 @@ The main difference between above Temporal Table DDL and Temporal Table Function - The temporal table DDL can be defined in SQL but temporal table function can not; - Both temporal table DDL and temporal table function support temporal join versioned table, but only temporal table function can temporal join the latest version of any table/view. +LATERAL SNAPSHOT Join +-------------- + +{{< label Streaming >}} + +A `LATERAL SNAPSHOT` join is a *stream enrichment* join that augments an append-only table (the *probe* side) with the current state of an updating table (the *build* side). +Unlike an [event-time temporal join](#event-time-temporal-join), it does not correlate each probe-side row with a specific historical version of the build side. +Instead, every probe-side row is joined with the build-side state that is current at the time the row is processed, similar to a [processing-time temporal join](#processing-time-temporal-join), but with well-defined behavior at query start-up. + +The `LATERAL SNAPSHOT` join is designed for scenarios where the other temporal joins are a poor fit: + +- The build side updates **infrequently**. An event-time temporal join relies on continuous build-side watermarks to emit results, so a build side that rarely advances its watermark stalls the join and lets probe-side state accumulate. A `LATERAL SNAPSHOT` join does not stall when the build side goes idle. +- The application has **low-latency** requirements that are incompatible with the watermark-induced delay of an event-time temporal join. +- The build side has **no primary key**. Event-time and processing-time temporal joins require the build-side primary key to appear in the equi-join condition; a `LATERAL SNAPSHOT` join does not. + +The build side is wrapped in the `SNAPSHOT` table function inside a `LATERAL TABLE` clause. Both `INNER JOIN` and `LEFT [OUTER] JOIN` are supported. The join requires at least one conjunctive equality predicate; additional non-equi predicates are allowed in the `ON` clause. + +```sql +SELECT [column_list] +FROM probe_table +[LEFT] JOIN LATERAL TABLE( + SNAPSHOT( + input => TABLE build_table, + [ load_completed_condition => <'compile_time' | 'user_time'>, ] + [ load_completed_time => , ] + [ load_completed_idle_timeout => , ] + [ state_ttl => ])) AS s +ON probe_table.key = s.key +``` + +The `SNAPSHOT` function accepts the following arguments: + +| Argument | Type | Required | Description | +| --- | --- | --- | --- | +| `input` | TABLE | yes | The build-side table. It may use any [changelog mode]({{< ref "docs/sql/reference/queries/changelog" >}}) (inserts, updates, and deletes) and must declare a [watermark]({{< ref "docs/concepts/sql-table-concepts/time_attributes" >}}#event-time). | +| `load_completed_condition` | STRING | no | Determines when the initial load phase completes. One of `'compile_time'` (default) or `'user_time'`. With `'compile_time'`, the load phase completes once the build-side watermark reaches the wall-clock time at which the query was compiled. With `'user_time'`, it completes once the build-side watermark reaches the explicit `load_completed_time`. | +| `load_completed_time` | TIMESTAMP_LTZ(3) | no | The build-side event time that completes the load phase. Required when `load_completed_condition` is `'user_time'` and must not be set otherwise. | +| `load_completed_idle_timeout` | INTERVAL | no | A processing-time fallback that completes the load phase if the build-side watermark stalls before reaching the configured time. | +| `state_ttl` | INTERVAL | no | Retention time for build-side state. Keys that are not accessed within this duration become eligible for eviction. Only applied during the join phase. Defaults to the pipeline's [state TTL]({{< ref "docs/dev/table/config" >}}#table-exec-state-ttl). | + +The operator processes its inputs in two sequential phases. During the **load phase**, the operator builds up the build-side table by accumulating its changes into state until the load completion condition is met. This allows the build side to load its initial state (for example the current contents of a compacted topic or a change-data-capture snapshot) before any probe-side row is joined, ensuring that early probe-side rows are not joined against an incomplete build side. While the build side is not yet ready, incoming probe-side rows are buffered. The load phase completes when the build-side watermark reaches the wall-clock time at which the query was compiled (`compile_time`, the default), when the build-side watermark reaches `load_completed_time` (`user_time`), or, as a fallback, when the `load_completed_idle_timeout` elapses in processing time without the build-side watermark advancing (which handles build sides that become idle during start-up). + +When transitioning from load to **join phase**, all buffered probe-side rows are joined against the current build-side state and emitted. From this point onwards, each probe-side row is joined and emitted immediately against the build-side state that is current at that moment. Build-side updates continue to be applied to the state and are made visible to later probe-side rows. + +Reusing the tables from the [event-time temporal join](#event-time-temporal-join) example, the following query enriches each order with the conversion rate that is current when the order is processed. In contrast to the temporal join, the join keeps making progress even if the rates are updated only rarely. + +```sql +SELECT + order_id, + price, + orders.currency, + conversion_rate, + order_time +FROM orders +JOIN LATERAL TABLE( + SNAPSHOT(input => TABLE currency_rates)) AS rates +ON orders.currency = rates.currency; +``` + +The probe side must be an append-only table, and its time attributes are preserved in the output. The build side may be an updating table and must declare a watermark. The result is append-only. A build-side rowtime attribute that is projected into the output is materialized as a regular `TIMESTAMP` and is no longer a time attribute. Probe-side watermarks are forwarded downstream in join phase; build-side watermarks are consumed internally and are not propagated. + +Because rows are joined against the build-side state that is current at processing time, the result is **not deterministic**. A given probe-side row may be joined with different build-side versions across different runs, depending on the relative timing of the two inputs. Probe and build-side inputs can be configured with watermark alignment to keep the two inputs roughly aligned on event time, so that a probe-side row tends to be joined with build-side changes of a similar event time. This is a best-effort alignment and does not make the result deterministic. + +The build-side state grows with the number of distinct build-side keys, and during the load phase the buffered probe-side rows add to the state footprint until the operator transitions to join phase. Use `state_ttl` to bound the build-side state for keys that are no longer updated or joined. You can reduce the amount of data that is buffered and processed during the load phase by configuring scan start offsets on the build and probe-side inputs, for example with a `scan.startup.*` [dynamic table option hint]({{< ref "docs/sql/reference/queries/hints" >}}#dynamic-table-options). + +In contrast to [regular joins](#regular-joins), changes on the build side do not retroactively update results that were already emitted. +Compared to [interval joins](#interval-joins), a `LATERAL SNAPSHOT` join does not define a time window within which rows are joined. + Lookup Join -------------- From 3549506c4ccb1fd7c443be6999bfabb511c2e038 Mon Sep 17 00:00:00 2001 From: Fabian Hueske Date: Fri, 17 Jul 2026 14:43:24 +0200 Subject: [PATCH 2/2] [FLINK-40131][docs] Address reviewer feedback --- .../docs/sql/functions/built-in-functions.md | 18 +++++ .../docs/sql/reference/queries/joins.md | 73 +++++++++++++------ .../docs/sql/functions/built-in-functions.md | 18 +++++ .../docs/sql/reference/queries/joins.md | 73 +++++++++++++------ 4 files changed, 136 insertions(+), 46 deletions(-) diff --git a/docs/content.zh/docs/sql/functions/built-in-functions.md b/docs/content.zh/docs/sql/functions/built-in-functions.md index 4aaf786e860c97..6165b239fcbbe3 100644 --- a/docs/content.zh/docs/sql/functions/built-in-functions.md +++ b/docs/content.zh/docs/sql/functions/built-in-functions.md @@ -132,6 +132,24 @@ JSON 函数使用符合 ISO/IEC TR 19075-6 SQL标准的 JSON 路径表达式。 {{< sql_functions_zh "bitmapagg" >}} +Table Functions +--------------- + +Table functions take zero, one, or more values as input and return multiple rows (a table) as the result. Most built-in table functions take a table as an input argument. +Table functions can be used in two ways: as stand-alone inputs, where they are invoked just once, or in a `LATERAL` context, where they are invoked for each row of an outer table. + +| Function | Description | +|--------------------------------------------|-----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------| +| `TUMBLE(data => TABLE t, ...)` | Assigns each row of the `data` table to a tumbling window specified by additional window columns (`window_start`, `window_end`, `window_time`). See [Window TVF]({{< ref "docs/sql/reference/queries/window-tvf" >}}#tumble) for the full list of arguments, semantics, and usage. | +| `HOP(data => TABLE t, ...)` | Assigns each row of the `data` table to a hopping window specified by additional window columns (`window_start`, `window_end`, `window_time`). See [Window TVF]({{< ref "docs/sql/reference/queries/window-tvf" >}}#hop) for the full list of arguments, semantics, and usage. | +| `CUMULATE(data => TABLE t, ...)` | Assigns each row of the `data` table to a cumulating window specified by additional window columns (`window_start`, `window_end`, `window_time`). See [Window TVF]({{< ref "docs/sql/reference/queries/window-tvf" >}}#cumulate) for the full list of arguments, semantics, and usage. | +| `SESSION(data => TABLE t, ...)` | Assigns each row of the `data` table to a session window specified by additional window columns (`window_start`, `window_end`, `window_time`). See [Window TVF]({{< ref "docs/sql/reference/queries/window-tvf" >}}#session) for the full list of arguments, semantics, and usage. | +| `FROM_CHANGELOG(input => TABLE t [, ...])` | Converts an append-only table with an explicit operation column into a dynamic table. See Changelog Conversion for the full list of arguments, semantics, and usage. | +| `TO_CHANGELOG(input => TABLE t [, ...])` | Converts a dynamic table into an append-only table with an explicit operation column. See Changelog Conversion for the full list of arguments, semantics, and usage. | +| `SNAPSHOT(input => TABLE t [, ...])` | Returns the current state of a dynamic table `t`. `SNAPSHOT` can only be used in a `LATERAL` context and not as a stand-alone table function. See [LATERAL SNAPSHOT join]({{< ref "docs/sql/reference/queries/joins" >}}#lateral-snapshot-join) for the full list of arguments, the join semantics, and usage. | + +To implement your own table functions, see [user-defined table functions]({{< ref "docs/dev/table/functions/udfs" >}}#table-functions). + 时间间隔单位和时间点单位标识符 --------------------------------------- diff --git a/docs/content.zh/docs/sql/reference/queries/joins.md b/docs/content.zh/docs/sql/reference/queries/joins.md index d3429c2571d49b..14fbbda01b096c 100644 --- a/docs/content.zh/docs/sql/reference/queries/joins.md +++ b/docs/content.zh/docs/sql/reference/queries/joins.md @@ -298,19 +298,41 @@ WHERE LATERAL SNAPSHOT Join -------------- -{{< label Streaming >}} +{{< label Streaming >}} {{< label Batch >}} -A `LATERAL SNAPSHOT` join is a *stream enrichment* join that augments an append-only table (the *probe* side) with the current state of an updating table (the *build* side). -Unlike an [event-time temporal join](#event-time-temporal-join), it does not correlate each probe-side row with a specific historical version of the build side. -Instead, every probe-side row is joined with the build-side state that is current at the time the row is processed, similar to a [processing-time temporal join](#processing-time-temporal-join), but with well-defined behavior at query start-up. +A `LATERAL SNAPSHOT` join is a *stream enrichment* join that augments an append-only table with the current state of an updating table. +As in the [temporal joins](#temporal-joins), the enriched (left) input is called the *probe side* and the enriching (right) input is called the *build side*. +Every probe-side row is joined with the build-side state that is current at the time the row is processed. -The `LATERAL SNAPSHOT` join is designed for scenarios where the other temporal joins are a poor fit: +This is the same mode of operation as the [processing-time temporal join](#processing-time-temporal-join), which also joins each probe-side row against the latest build-side state. +However, the processing-time temporal join has been disabled for Flink SQL due to a fundamental limitation. +The processing-time temporal join starts joining immediately when the query is started: early probe-side rows are joined against whatever build-side rows happen to have been loaded so far. +Right after query start-up, the build side is typically still incomplete, so early probe-side rows may be enriched with missing or stale data, and the result depends on the order in which the two inputs are read. +A `LATERAL SNAPSHOT` join addresses this problem by first loading the build side up to a well-defined point in time (the *load phase*) before it starts joining (the *join phase*). -- The build side updates **infrequently**. An event-time temporal join relies on continuous build-side watermarks to emit results, so a build side that rarely advances its watermark stalls the join and lets probe-side state accumulate. A `LATERAL SNAPSHOT` join does not stall when the build side goes idle. -- The application has **low-latency** requirements that are incompatible with the watermark-induced delay of an event-time temporal join. -- The build side has **no primary key**. Event-time and processing-time temporal joins require the build-side primary key to appear in the equi-join condition; a `LATERAL SNAPSHOT` join does not. +**When to use it** -The build side is wrapped in the `SNAPSHOT` table function inside a `LATERAL TABLE` clause. Both `INNER JOIN` and `LEFT [OUTER] JOIN` are supported. The join requires at least one conjunctive equality predicate; additional non-equi predicates are allowed in the `ON` clause. +The `LATERAL SNAPSHOT` join is designed for enrichment scenarios where the other temporal joins are a poor fit: + +- **The build side does not receive continuous updates.** An event-time temporal join only emits a joined row once the combined watermark of both inputs has passed the event time of the probe-side row. A build side that does not continuously produce records, and therefore does not advance its watermark, stalls the join and lets probe-side state accumulate. A `LATERAL SNAPSHOT` join keeps making progress even when the build side is idle. +- **Low latency is required.** An event-time temporal join holds back probe-side rows until the watermark catches up, which adds latency. After the initial load phase, the `LATERAL SNAPSHOT` join immediately joins a probe-side row when it arrives. +- **The build side has no primary key.** Event-time and processing-time temporal joins require the build side to have a primary key that appears in the equi-join condition. A `LATERAL SNAPSHOT` join has no such requirement; neither the probe side nor the build side needs a primary key. + +**How the join works** + +During the *load phase*, the operator accumulates the build-side changes into state until the load-completion condition is met, without emitting any results yet. This lets the build side load its initial state so that early probe-side rows are not joined against an incomplete build side. Probe-side rows that arrive during the load phase are buffered. The load phase completes when one of the following occurs: + +- the build-side watermark reaches a configured `load_completed_time`. This time is either explicitly set by the user (`load_completed_condition => 'user_time'`) or automatically set to the wall-clock time when the query is compiled (`load_completed_condition => 'compile_time'`), or +- as a fallback, the `load_completed_idle_timeout` elapses in processing time without the build-side watermark advancing (which handles build sides that become idle during start-up). + +When the load phase completes, the operator transitions to the *join phase*: all buffered probe-side rows are joined against the current build-side state and emitted. +From then on, each probe-side row is joined and emitted immediately against the build-side state that is current at that moment. +Build-side updates continue to be applied to the state and become visible to subsequent probe-side rows. + +**Syntax** + +The build side is wrapped in the `SNAPSHOT` table function inside a `LATERAL TABLE` clause. The outer (probe-side) table must be an append-only table. +Both `INNER JOIN` and `LEFT [OUTER] JOIN` are supported. The join requires at least one conjunctive equality predicate; additional non-equi predicates are allowed in the `ON` clause. ```sql SELECT [column_list] @@ -322,24 +344,24 @@ FROM probe_table [ load_completed_time => , ] [ load_completed_idle_timeout => , ] [ state_ttl => ])) AS s -ON probe_table.key = s.key +ON probe_table.col = s.col ``` The `SNAPSHOT` function accepts the following arguments: -| Argument | Type | Required | Description | -| --- | --- | --- | --- | -| `input` | TABLE | yes | The build-side table. It may use any [changelog mode]({{< ref "docs/sql/reference/queries/changelog" >}}) (inserts, updates, and deletes) and must declare a [watermark]({{< ref "docs/concepts/sql-table-concepts/time_attributes" >}}#event-time). | +| Argument | Type | Required | Description | +| --- | --- | --- |-------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------| +| `input` | TABLE | yes | The build-side table. It may use any changelog mode (inserts, updates, and deletes). In streaming mode it must declare a [watermark]({{< ref "docs/concepts/sql-table-concepts/time_attributes" >}}#event-time). | | `load_completed_condition` | STRING | no | Determines when the initial load phase completes. One of `'compile_time'` (default) or `'user_time'`. With `'compile_time'`, the load phase completes once the build-side watermark reaches the wall-clock time at which the query was compiled. With `'user_time'`, it completes once the build-side watermark reaches the explicit `load_completed_time`. | -| `load_completed_time` | TIMESTAMP_LTZ(3) | no | The build-side event time that completes the load phase. Required when `load_completed_condition` is `'user_time'` and must not be set otherwise. | -| `load_completed_idle_timeout` | INTERVAL | no | A processing-time fallback that completes the load phase if the build-side watermark stalls before reaching the configured time. | -| `state_ttl` | INTERVAL | no | Retention time for build-side state. Keys that are not accessed within this duration become eligible for eviction. Only applied during the join phase. Defaults to the pipeline's [state TTL]({{< ref "docs/dev/table/config" >}}#table-exec-state-ttl). | +| `load_completed_time` | TIMESTAMP_LTZ(3) | no | The build-side event time that completes the load phase. Required when `load_completed_condition` is `'user_time'` and must not be set otherwise. | +| `load_completed_idle_timeout` | INTERVAL | no | A processing-time fallback to complete the load phase. The transition to the join phase happens when the build-side watermark does not advance for more than the configured interval. | +| `state_ttl` | INTERVAL | no | Retention time for build-side state. Join keys that are not accessed within this duration become eligible for eviction. Only applied during the join phase. Defaults to the pipeline's [state TTL]({{< ref "docs/dev/table/config" >}}#table-exec-state-ttl). | -The operator processes its inputs in two sequential phases. During the **load phase**, the operator builds up the build-side table by accumulating its changes into state until the load completion condition is met. This allows the build side to load its initial state (for example the current contents of a compacted topic or a change-data-capture snapshot) before any probe-side row is joined, ensuring that early probe-side rows are not joined against an incomplete build side. While the build side is not yet ready, incoming probe-side rows are buffered. The load phase completes when the build-side watermark reaches the wall-clock time at which the query was compiled (`compile_time`, the default), when the build-side watermark reaches `load_completed_time` (`user_time`), or, as a fallback, when the `load_completed_idle_timeout` elapses in processing time without the build-side watermark advancing (which handles build sides that become idle during start-up). +`load_completed_condition`, `load_completed_time`, `load_completed_idle_timeout`, and `state_ttl` only affect streaming execution and are ignored in batch mode (see **Batch mode** below). -When transitioning from load to **join phase**, all buffered probe-side rows are joined against the current build-side state and emitted. From this point onwards, each probe-side row is joined and emitted immediately against the build-side state that is current at that moment. Build-side updates continue to be applied to the state and are made visible to later probe-side rows. +**Example** -Reusing the tables from the [event-time temporal join](#event-time-temporal-join) example, the following query enriches each order with the conversion rate that is current when the order is processed. In contrast to the temporal join, the join keeps making progress even if the rates are updated only rarely. +Reusing the `orders` and `currency_rates` tables from the [event-time temporal join](#event-time-temporal-join) example, the following query enriches each order with the conversion rate that is current when the order is processed. In contrast to the event-time temporal join, the join keeps making progress even if the rates are updated only rarely, e.g., once a day. ```sql SELECT @@ -354,14 +376,19 @@ JOIN LATERAL TABLE( ON orders.currency = rates.currency; ``` -The probe side must be an append-only table, and its time attributes are preserved in the output. The build side may be an updating table and must declare a watermark. The result is append-only. A build-side rowtime attribute that is projected into the output is materialized as a regular `TIMESTAMP` and is no longer a time attribute. Probe-side watermarks are forwarded downstream in join phase; build-side watermarks are consumed internally and are not propagated. +**Result and state characteristics** + +The result is append-only and preserves the probe-side time attributes. A build-side rowtime attribute that is projected into the output is materialized as a regular `TIMESTAMP` and is no longer a time attribute. Probe-side watermarks are forwarded downstream during the join phase; build-side watermarks are consumed internally and are not propagated. Because rows are joined against the build-side state that is current at processing time, the result is **not deterministic**. A given probe-side row may be joined with different build-side versions across different runs, depending on the relative timing of the two inputs. Probe and build-side inputs can be configured with watermark alignment to keep the two inputs roughly aligned on event time, so that a probe-side row tends to be joined with build-side changes of a similar event time. This is a best-effort alignment and does not make the result deterministic. -The build-side state grows with the number of distinct build-side keys, and during the load phase the buffered probe-side rows add to the state footprint until the operator transitions to join phase. Use `state_ttl` to bound the build-side state for keys that are no longer updated or joined. You can reduce the amount of data that is buffered and processed during the load phase by configuring scan start offsets on the build and probe-side inputs, for example with a `scan.startup.*` [dynamic table option hint]({{< ref "docs/sql/reference/queries/hints" >}}#dynamic-table-options). +The build-side state grows with the number of distinct build-side keys, and during the load phase the buffered probe-side rows add to the state footprint until the operator transitions to the join phase. Use `state_ttl` to bound the build-side state for keys that are no longer updated or joined. You can reduce the amount of data that is buffered and processed during the load phase by configuring scan start offsets on the build and probe-side inputs, for example with a `scan.startup.*` [dynamic table option hint]({{< ref "docs/sql/reference/queries/hints" >}}#dynamic-table-options). + +**Batch mode** + +In batch mode, a `LATERAL SNAPSHOT` join is executed as a regular (`INNER` or `LEFT`) join between the probe side and the complete build side. Batch execution reads the entire build side before joining, so there is no load phase and no incremental state build-up. The streaming-specific arguments (`load_completed_condition`, `load_completed_time`, `load_completed_idle_timeout`, and `state_ttl`) are accepted but have no effect, and the build side does not need to declare a watermark. -In contrast to [regular joins](#regular-joins), changes on the build side do not retroactively update results that were already emitted. -Compared to [interval joins](#interval-joins), a `LATERAL SNAPSHOT` join does not define a time window within which rows are joined. +Because every probe-side row is joined against the final, complete build side, the batch result is **deterministic**. Lookup Join -------------- diff --git a/docs/content/docs/sql/functions/built-in-functions.md b/docs/content/docs/sql/functions/built-in-functions.md index 42808103674bb9..3946cd225abee5 100644 --- a/docs/content/docs/sql/functions/built-in-functions.md +++ b/docs/content/docs/sql/functions/built-in-functions.md @@ -135,6 +135,24 @@ The aggregate functions take an expression across all the rows as the input and {{< sql_functions "bitmapagg" >}} +Table Functions +--------------- + +Table functions take zero, one, or more values as input and return multiple rows (a table) as the result. Most built-in table functions take a table as an input argument. +Table functions can be used in two ways: as stand-alone inputs, where they are invoked just once, or in a `LATERAL` context, where they are invoked for each row of an outer table. + +| Function | Description | +|--------------------------------------------|-----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------| +| `TUMBLE(data => TABLE t, ...)` | Assigns each row of the `data` table to a tumbling window specified by additional window columns (`window_start`, `window_end`, `window_time`). See [Window TVF]({{< ref "docs/sql/reference/queries/window-tvf" >}}#tumble) for the full list of arguments, semantics, and usage. | +| `HOP(data => TABLE t, ...)` | Assigns each row of the `data` table to a hopping window specified by additional window columns (`window_start`, `window_end`, `window_time`). See [Window TVF]({{< ref "docs/sql/reference/queries/window-tvf" >}}#hop) for the full list of arguments, semantics, and usage. | +| `CUMULATE(data => TABLE t, ...)` | Assigns each row of the `data` table to a cumulating window specified by additional window columns (`window_start`, `window_end`, `window_time`). See [Window TVF]({{< ref "docs/sql/reference/queries/window-tvf" >}}#cumulate) for the full list of arguments, semantics, and usage. | +| `SESSION(data => TABLE t, ...)` | Assigns each row of the `data` table to a session window specified by additional window columns (`window_start`, `window_end`, `window_time`). See [Window TVF]({{< ref "docs/sql/reference/queries/window-tvf" >}}#session) for the full list of arguments, semantics, and usage. | +| `FROM_CHANGELOG(input => TABLE t [, ...])` | Converts an append-only table with an explicit operation column into a dynamic table. See [Changelog Conversion]({{< ref "docs/sql/reference/queries/changelog" >}}#from_changelog) for the full list of arguments, semantics, and usage. | +| `TO_CHANGELOG(input => TABLE t [, ...])` | Converts a dynamic table into an append-only table with an explicit operation column. See [Changelog Conversion]({{< ref "docs/sql/reference/queries/changelog" >}}#to_changelog) for the full list of arguments, semantics, and usage. | +| `SNAPSHOT(input => TABLE t [, ...])` | Returns the current state of a dynamic table `t`. `SNAPSHOT` can only be used in a `LATERAL` context and not as a stand-alone table function. See [LATERAL SNAPSHOT join]({{< ref "docs/sql/reference/queries/joins" >}}#lateral-snapshot-join) for the full list of arguments, the join semantics, and usage. | + +To implement your own table functions, see [user-defined table functions]({{< ref "docs/dev/table/functions/udfs" >}}#table-functions). + Time Interval and Point Unit Specifiers --------------------------------------- diff --git a/docs/content/docs/sql/reference/queries/joins.md b/docs/content/docs/sql/reference/queries/joins.md index c8770766e3541f..82ab96c5bea8ed 100644 --- a/docs/content/docs/sql/reference/queries/joins.md +++ b/docs/content/docs/sql/reference/queries/joins.md @@ -303,19 +303,41 @@ The main difference between above Temporal Table DDL and Temporal Table Function LATERAL SNAPSHOT Join -------------- -{{< label Streaming >}} +{{< label Streaming >}} {{< label Batch >}} -A `LATERAL SNAPSHOT` join is a *stream enrichment* join that augments an append-only table (the *probe* side) with the current state of an updating table (the *build* side). -Unlike an [event-time temporal join](#event-time-temporal-join), it does not correlate each probe-side row with a specific historical version of the build side. -Instead, every probe-side row is joined with the build-side state that is current at the time the row is processed, similar to a [processing-time temporal join](#processing-time-temporal-join), but with well-defined behavior at query start-up. +A `LATERAL SNAPSHOT` join is a *stream enrichment* join that augments an append-only table with the current state of an updating table. +As in the [temporal joins](#temporal-joins), the enriched (left) input is called the *probe side* and the enriching (right) input is called the *build side*. +Every probe-side row is joined with the build-side state that is current at the time the row is processed. -The `LATERAL SNAPSHOT` join is designed for scenarios where the other temporal joins are a poor fit: +This is the same mode of operation as the [processing-time temporal join](#processing-time-temporal-join), which also joins each probe-side row against the latest build-side state. +However, the processing-time temporal join has been disabled for Flink SQL due to a fundamental limitation. +The processing-time temporal join starts joining immediately when the query is started: early probe-side rows are joined against whatever build-side rows happen to have been loaded so far. +Right after query start-up, the build side is typically still incomplete, so early probe-side rows may be enriched with missing or stale data, and the result depends on the order in which the two inputs are read. +A `LATERAL SNAPSHOT` join addresses this problem by first loading the build side up to a well-defined point in time (the *load phase*) before it starts joining (the *join phase*). -- The build side updates **infrequently**. An event-time temporal join relies on continuous build-side watermarks to emit results, so a build side that rarely advances its watermark stalls the join and lets probe-side state accumulate. A `LATERAL SNAPSHOT` join does not stall when the build side goes idle. -- The application has **low-latency** requirements that are incompatible with the watermark-induced delay of an event-time temporal join. -- The build side has **no primary key**. Event-time and processing-time temporal joins require the build-side primary key to appear in the equi-join condition; a `LATERAL SNAPSHOT` join does not. +**When to use it** -The build side is wrapped in the `SNAPSHOT` table function inside a `LATERAL TABLE` clause. Both `INNER JOIN` and `LEFT [OUTER] JOIN` are supported. The join requires at least one conjunctive equality predicate; additional non-equi predicates are allowed in the `ON` clause. +The `LATERAL SNAPSHOT` join is designed for enrichment scenarios where the other temporal joins are a poor fit: + +- **The build side does not receive continuous updates.** An event-time temporal join only emits a joined row once the combined watermark of both inputs has passed the event time of the probe-side row. A build side that does not continuously produce records, and therefore does not advance its watermark, stalls the join and lets probe-side state accumulate. A `LATERAL SNAPSHOT` join keeps making progress even when the build side is idle. +- **Low latency is required.** An event-time temporal join holds back probe-side rows until the watermark catches up, which adds latency. After the initial load phase, the `LATERAL SNAPSHOT` join immediately joins a probe-side row when it arrives. +- **The build side has no primary key.** Event-time and processing-time temporal joins require the build side to have a primary key that appears in the equi-join condition. A `LATERAL SNAPSHOT` join has no such requirement; neither the probe side nor the build side needs a primary key. + +**How the join works** + +During the *load phase*, the operator accumulates the build-side changes into state until the load-completion condition is met, without emitting any results yet. This lets the build side load its initial state so that early probe-side rows are not joined against an incomplete build side. Probe-side rows that arrive during the load phase are buffered. The load phase completes when one of the following occurs: + +- the build-side watermark reaches a configured `load_completed_time`. This time is either explicitly set by the user (`load_completed_condition => 'user_time'`) or automatically set to the wall-clock time when the query is compiled (`load_completed_condition => 'compile_time'`), or +- as a fallback, the `load_completed_idle_timeout` elapses in processing time without the build-side watermark advancing (which handles build sides that become idle during start-up). + +When the load phase completes, the operator transitions to the *join phase*: all buffered probe-side rows are joined against the current build-side state and emitted. +From then on, each probe-side row is joined and emitted immediately against the build-side state that is current at that moment. +Build-side updates continue to be applied to the state and become visible to subsequent probe-side rows. + +**Syntax** + +The build side is wrapped in the `SNAPSHOT` table function inside a `LATERAL TABLE` clause. The outer (probe-side) table must be an append-only table. +Both `INNER JOIN` and `LEFT [OUTER] JOIN` are supported. The join requires at least one conjunctive equality predicate; additional non-equi predicates are allowed in the `ON` clause. ```sql SELECT [column_list] @@ -327,24 +349,24 @@ FROM probe_table [ load_completed_time => , ] [ load_completed_idle_timeout => , ] [ state_ttl => ])) AS s -ON probe_table.key = s.key +ON probe_table.col = s.col ``` The `SNAPSHOT` function accepts the following arguments: -| Argument | Type | Required | Description | -| --- | --- | --- | --- | -| `input` | TABLE | yes | The build-side table. It may use any [changelog mode]({{< ref "docs/sql/reference/queries/changelog" >}}) (inserts, updates, and deletes) and must declare a [watermark]({{< ref "docs/concepts/sql-table-concepts/time_attributes" >}}#event-time). | +| Argument | Type | Required | Description | +| --- | --- | --- |-------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------| +| `input` | TABLE | yes | The build-side table. It may use any [changelog mode]({{< ref "docs/sql/reference/queries/changelog" >}}) (inserts, updates, and deletes). In streaming mode it must declare a [watermark]({{< ref "docs/concepts/sql-table-concepts/time_attributes" >}}#event-time). | | `load_completed_condition` | STRING | no | Determines when the initial load phase completes. One of `'compile_time'` (default) or `'user_time'`. With `'compile_time'`, the load phase completes once the build-side watermark reaches the wall-clock time at which the query was compiled. With `'user_time'`, it completes once the build-side watermark reaches the explicit `load_completed_time`. | -| `load_completed_time` | TIMESTAMP_LTZ(3) | no | The build-side event time that completes the load phase. Required when `load_completed_condition` is `'user_time'` and must not be set otherwise. | -| `load_completed_idle_timeout` | INTERVAL | no | A processing-time fallback that completes the load phase if the build-side watermark stalls before reaching the configured time. | -| `state_ttl` | INTERVAL | no | Retention time for build-side state. Keys that are not accessed within this duration become eligible for eviction. Only applied during the join phase. Defaults to the pipeline's [state TTL]({{< ref "docs/dev/table/config" >}}#table-exec-state-ttl). | +| `load_completed_time` | TIMESTAMP_LTZ(3) | no | The build-side event time that completes the load phase. Required when `load_completed_condition` is `'user_time'` and must not be set otherwise. | +| `load_completed_idle_timeout` | INTERVAL | no | A processing-time fallback to complete the load phase. The transition to the join phase happens when the build-side watermark does not advance for more than the configured interval. | +| `state_ttl` | INTERVAL | no | Retention time for build-side state. Join keys that are not accessed within this duration become eligible for eviction. Only applied during the join phase. Defaults to the pipeline's [state TTL]({{< ref "docs/dev/table/config" >}}#table-exec-state-ttl). | -The operator processes its inputs in two sequential phases. During the **load phase**, the operator builds up the build-side table by accumulating its changes into state until the load completion condition is met. This allows the build side to load its initial state (for example the current contents of a compacted topic or a change-data-capture snapshot) before any probe-side row is joined, ensuring that early probe-side rows are not joined against an incomplete build side. While the build side is not yet ready, incoming probe-side rows are buffered. The load phase completes when the build-side watermark reaches the wall-clock time at which the query was compiled (`compile_time`, the default), when the build-side watermark reaches `load_completed_time` (`user_time`), or, as a fallback, when the `load_completed_idle_timeout` elapses in processing time without the build-side watermark advancing (which handles build sides that become idle during start-up). +`load_completed_condition`, `load_completed_time`, `load_completed_idle_timeout`, and `state_ttl` only affect streaming execution and are ignored in batch mode (see **Batch mode** below). -When transitioning from load to **join phase**, all buffered probe-side rows are joined against the current build-side state and emitted. From this point onwards, each probe-side row is joined and emitted immediately against the build-side state that is current at that moment. Build-side updates continue to be applied to the state and are made visible to later probe-side rows. +**Example** -Reusing the tables from the [event-time temporal join](#event-time-temporal-join) example, the following query enriches each order with the conversion rate that is current when the order is processed. In contrast to the temporal join, the join keeps making progress even if the rates are updated only rarely. +Reusing the `orders` and `currency_rates` tables from the [event-time temporal join](#event-time-temporal-join) example, the following query enriches each order with the conversion rate that is current when the order is processed. In contrast to the event-time temporal join, the join keeps making progress even if the rates are updated only rarely, e.g., once a day. ```sql SELECT @@ -359,14 +381,19 @@ JOIN LATERAL TABLE( ON orders.currency = rates.currency; ``` -The probe side must be an append-only table, and its time attributes are preserved in the output. The build side may be an updating table and must declare a watermark. The result is append-only. A build-side rowtime attribute that is projected into the output is materialized as a regular `TIMESTAMP` and is no longer a time attribute. Probe-side watermarks are forwarded downstream in join phase; build-side watermarks are consumed internally and are not propagated. +**Result and state characteristics** + +The result is append-only and preserves the probe-side time attributes. A build-side rowtime attribute that is projected into the output is materialized as a regular `TIMESTAMP` and is no longer a time attribute. Probe-side watermarks are forwarded downstream during the join phase; build-side watermarks are consumed internally and are not propagated. Because rows are joined against the build-side state that is current at processing time, the result is **not deterministic**. A given probe-side row may be joined with different build-side versions across different runs, depending on the relative timing of the two inputs. Probe and build-side inputs can be configured with watermark alignment to keep the two inputs roughly aligned on event time, so that a probe-side row tends to be joined with build-side changes of a similar event time. This is a best-effort alignment and does not make the result deterministic. -The build-side state grows with the number of distinct build-side keys, and during the load phase the buffered probe-side rows add to the state footprint until the operator transitions to join phase. Use `state_ttl` to bound the build-side state for keys that are no longer updated or joined. You can reduce the amount of data that is buffered and processed during the load phase by configuring scan start offsets on the build and probe-side inputs, for example with a `scan.startup.*` [dynamic table option hint]({{< ref "docs/sql/reference/queries/hints" >}}#dynamic-table-options). +The build-side state grows with the number of distinct build-side keys, and during the load phase the buffered probe-side rows add to the state footprint until the operator transitions to the join phase. Use `state_ttl` to bound the build-side state for keys that are no longer updated or joined. You can reduce the amount of data that is buffered and processed during the load phase by configuring scan start offsets on the build and probe-side inputs, for example with a `scan.startup.*` [dynamic table option hint]({{< ref "docs/sql/reference/queries/hints" >}}#dynamic-table-options). + +**Batch mode** + +In batch mode, a `LATERAL SNAPSHOT` join is executed as a regular (`INNER` or `LEFT`) join between the probe side and the complete build side. Batch execution reads the entire build side before joining, so there is no load phase and no incremental state build-up. The streaming-specific arguments (`load_completed_condition`, `load_completed_time`, `load_completed_idle_timeout`, and `state_ttl`) are accepted but have no effect, and the build side does not need to declare a watermark. -In contrast to [regular joins](#regular-joins), changes on the build side do not retroactively update results that were already emitted. -Compared to [interval joins](#interval-joins), a `LATERAL SNAPSHOT` join does not define a time window within which rows are joined. +Because every probe-side row is joined against the final, complete build side, the batch result is **deterministic**. Lookup Join --------------