Skip to content

[FLINK-40666][table] Simplify SNAPSHOT function by removing load_completed_condition argument #29195

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
fhueske merged 2 commits into from
Sep 16, 2026
Merged

[FLINK-40666][table] Simplify SNAPSHOT function by removing load_completed_condition argument #29195

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
4750060
[FLINK-40666][table] Simplify SNAPSHOT function by removing load_comp…
fhueske Sep 15, 2026
60127ee
[FLINK-40666][table] Derive default SNAPSHOT load_completed_time from…
fhueske Sep 16, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
10 changes: 4 additions & 6 deletions docs/content.zh/docs/sql/reference/queries/joins.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -349,7 +349,7 @@ A `LATERAL SNAPSHOT` join operates in two phases to avoid joining probe-side row

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. 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
- the build-side watermark reaches a configured `load_completed_time`. This time is either explicitly set by the user, or, if `load_completed_time` is not provided, automatically set to the wall-clock time when the query is compiled, 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.
Expand All @@ -369,7 +369,6 @@ FROM probe_table
[LEFT] JOIN LATERAL SNAPSHOT(
input => TABLE build_table,
[ on_time => DESCRIPTOR(<rowtime_column>), ]
[ load_completed_condition => <'compile_time' | 'user_time'>, ]
[ load_completed_time => <timestamp_ltz>, ]
[ load_completed_idle_timeout => <interval>, ]
[ state_ttl => <interval> ]) AS s
Expand All @@ -382,12 +381,11 @@ The `SNAPSHOT` function accepts the following arguments:
| --- | --- | --- |-------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|
| `input` | TABLE | yes | The build-side table. It may use any changelog mode (inserts, updates, and deletes). |
| `on_time` | DESCRIPTOR | no | Declares a build-side rowtime column that defines the order in which the build-side changes are applied. The referenced column must exist in `input` and be a `TIMESTAMP` or `TIMESTAMP_LTZ` column (up to precision 3) that is declared as a [watermarked rowtime attribute]({{< ref "docs/concepts/sql-table-concepts/time_attributes" >}}#event-time). The argument is **required for streaming queries**. |
| `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_time` | TIMESTAMP_LTZ(3) | no | The build-side event time that completes the load phase. If omitted, the load phase completes once the build-side watermark reaches the wall-clock time at which the query was compiled. |
| `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). |

`on_time`, `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).
`on_time`, `load_completed_time`, `load_completed_idle_timeout`, and `state_ttl` only affect streaming execution and are ignored in batch mode (see **Batch mode** below).

**Result and state characteristics**

Expand All @@ -399,7 +397,7 @@ The build-side state grows with the number of distinct build-side keys, and duri

**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 (`on_time`, `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 or provide a `on_time`.
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 (`on_time`, `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 or provide a `on_time`.

Because every probe-side row is joined against the final, complete build side, the batch result is **deterministic**.

Expand Down
10 changes: 4 additions & 6 deletions docs/content/docs/sql/reference/queries/joins.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -356,7 +356,7 @@ it first loads the build side up to a well-defined point in time (the *load phas
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.
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
- the build-side watermark reaches a configured `load_completed_time`. This time is either explicitly set by the user, or, if `load_completed_time` is not provided, automatically set to the wall-clock time when the query is compiled, 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.
Expand All @@ -378,7 +378,6 @@ FROM probe_table
[LEFT] JOIN LATERAL SNAPSHOT(
input => TABLE build_table,
[ on_time => DESCRIPTOR(<rowtime_column>), ]
[ load_completed_condition => <'compile_time' | 'user_time'>, ]
[ load_completed_time => <timestamp_ltz>, ]
[ load_completed_idle_timeout => <interval>, ]
[ state_ttl => <interval> ]) AS s
Expand All @@ -391,12 +390,11 @@ The `SNAPSHOT` function accepts the following arguments:
| --- | --- | --- |-----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|
| `input` | TABLE | yes | The build-side table. It may use any [changelog mode]({{< ref "docs/sql/reference/queries/changelog" >}}) (inserts, updates, and deletes). |
| `on_time` | DESCRIPTOR | no | Declares a build-side rowtime column that defines the order in which the build-side changes are applied. The referenced column must exist in `input` and be a `TIMESTAMP` or `TIMESTAMP_LTZ` column (up to precision 3) that is declared as a [watermarked rowtime attribute]({{< ref "docs/concepts/sql-table-concepts/time_attributes" >}}#event-time). The argument is **required for streaming queries**. |
| `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_time` | TIMESTAMP_LTZ(3) | no | The build-side event time that completes the load phase. If omitted, the load phase completes once the build-side watermark reaches the wall-clock time at which the query was compiled. |
| `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). |

`on_time`, `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).
`on_time`, `load_completed_time`, `load_completed_idle_timeout`, and `state_ttl` only affect streaming execution and are ignored in batch mode (see **Batch mode** below).

**Result and state characteristics**

Expand All @@ -408,7 +406,7 @@ The build-side state grows with the number of distinct build-side keys, and duri

**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 (`on_time`, `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 or provide a `on_time`.
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 (`on_time`, `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 or provide a `on_time`.

Because every probe-side row is joined against the final, complete build side, the batch result is **deterministic**.

Expand Down
4 changes: 2 additions & 2 deletions ...le/flink-sql-parser/src/test/java/org/apache/flink/sql/parser/FlinkSqlParserImplTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -4167,13 +4167,13 @@ void testLateralImplicitTableFunction() {
// Named arguments and TABLE-typed arg passed to the function.
sql("select * from t, lateral snapshot("
+ "input => table s, "
+ "load_completed_condition => 'on_time')")
+ "load_completed_idle_timeout => interval '10' second)")
.ok(
"SELECT *\n"
+ "FROM `T`,\n"
+ "LATERAL TABLE(`SNAPSHOT`("
+ "`INPUT` => (TABLE `S`), "
+ "`LOAD_COMPLETED_CONDITION` => 'on_time'))");
+ "`LOAD_COMPLETED_IDLE_TIMEOUT` => INTERVAL '10' SECOND))");

// LATERAL fn(...) as the first FROM entry (no preceding table).
sql("select * from lateral ramp(3)").ok("SELECT *\n" + "FROM LATERAL TABLE(`RAMP`(3))");
Expand Down
2 changes: 0 additions & 2 deletions ...ble-common/src/main/java/org/apache/flink/table/functions/BuiltInFunctionDefinitions.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -947,8 +947,6 @@ ANY, and(logical(LogicalTypeRoot.BOOLEAN), LITERAL)
StaticArgumentTrait.REQUIRE_UPDATE_BEFORE,
StaticArgumentTrait.REQUIRE_FULL_DELETE)),
StaticArgument.scalar("on_time", DataTypes.DESCRIPTOR(), true),
StaticArgument.scalar(
"load_completed_condition", DataTypes.STRING(), true),
StaticArgument.scalar(
"load_completed_time", DataTypes.TIMESTAMP_LTZ(3), true),
StaticArgument.scalar(
Expand Down
Loading