[
https://issues.apache.org/jira/browse/FLINK-39735?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Ramin Gharib updated FLINK-39735:
---------------------------------
Description:
## Problem
`ProcessTableFunction`s receive a `TableSemantics` for each table-typed
argument. Semantics surface today exposes:
- `dataType()`
- `partitionByColumns()` — only populated when caller wrote `PARTITION BY` in
SQL.
- `orderByColumns()` / `orderByDirections()`
- `timeColumn()`
- `changelogMode()` — planner-derived, lifecycle-aware (empty during type
inference).
Does **not** expose input table's upsert key. Planner already derives it via
`FlinkRelMetadataQuery.getUpsertKeys(input)` (metadata handler chain in
`FlinkRelMdUpsertKeys`), but result is invisible to PTF.
Forces PTF authors to either:
- Require caller to repeat key via `PARTITION BY` even when planner already
knows it from PK constraints, or
- Re-derive upsert keys inside function (impossible — at constructor time,
function has only `TableSemantics`, not a `RelNode` or `RelMetadataQuery`).
Concrete impact for `TO_CHANGELOG` (FLINK-39636): without access to input
upsert key, function cannot emit partial DELETE rows that preserve identity
columns in row semantics. Current workaround is "add `PARTITION BY <pk>`",
unergonomic for users whose input table already declares a primary key.
## Proposal
Add `int[] upsertKeyColumns()` to `TableSemantics`, populated by planner via
`FlinkRelMetadataQuery.getUpsertKeys(input)` collapsed to one candidate via
`UpsertKeyUtil.smallestKey(...)`. Returns empty array when no upsert key is
derivable (pure append-only sources, post-Sort streams that destroyed the key,
etc.) or during type inference (metadata not yet computed).
Plumbing:
1. **`TableSemantics`** (`flink-table-common`): add `default int[]
upsertKeyColumns()` method returning empty array. Default avoids breaking
source-compatibility for rare external implementor.
2. **`RuntimeTableSemantics`** (`flink-table-runtime`): add serializable `int[]
upsertKeyColumns` field, constructor parameter, accessor.
3. **`StreamExecProcessTableFunction`** (`flink-table-planner`): persist
`List<int[]> inputUpsertKeys` as new `@JsonProperty` field (one entry per table
input). Default to per-input empty arrays for older compiled plans
(back-compat).
4. **`StreamPhysicalProcessTableFunction.translateToExecNode`**: derive upsert
keys for each input via `FlinkRelMetadataQuery.getUpsertKeys(input)` +
`UpsertKeyUtil.smallestKey(...).orElse(new int[0])`. Pass list to new ExecNode
field.
5. **`OperatorBindingCallContext`** / **`OperatorBindingTableSemantics`**:
extend to accept `inputUpsertKeys` so value is visible at specialization time
(function constructor sees operator-binding context, not runtime context).
6. **`BridgingSqlFunction.toCallContext`** /
**`ProcessTableRunnerGenerator.generate`**: thread `inputUpsertKeys` through
codegen path.
7. **`TableSemanticsMock`** and **`TestHarnessTableSemantics`**: accept
optional setter for unit testing.
8. **`TO_CHANGELOG` runtime** (`ToChangelogFunction`): consume
`tableSemantics.upsertKeyColumns()` to decide which columns to preserve on
DELETE when `produces_full_deletes=false`. Solves row-semantics case from
FLINK-39636 without requiring `PARTITION BY`.
was:
h3. Problem
{\{ProcessTableFunction}}s receive a \{{TableSemantics}} for each table-typed
argument. Semantics surface today exposes:
{\{dataType()}}
{\{partitionByColumns()}} — only populated when caller wrote \{{PARTITION BY}}
in SQL.
{\{orderByColumns()}} / \{{orderByDirections()}}
{\{timeColumn()}}
{\{changelogMode()}} — planner-derived, lifecycle-aware (empty during type
inference).
Does not expose input table's upsert key. Planner already derives it via
\{{FlinkRelMetadataQuery.getUpsertKeys(input)}} (metadata handler chain in
\{{FlinkRelMdUpsertKeys}}), but result is invisible to PTF.
Forces PTF authors to either:
Require caller to repeat key via \{{PARTITION BY}} even when planner already
knows it from PK constraints, or
Re-derive upsert keys inside function (impossible — at constructor time,
function has only \{{TableSemantics}}, not a \{{RelNode}} or
\{{RelMetadataQuery}}).
Concrete impact for \{{TO_CHANGELOG}} (FLINK-39636): without access to input
upsert key, function cannot emit partial DELETE rows that preserve identity
columns in row semantics. Current workaround is "add \{{PARTITION BY <pk>}}",
unergonomic for users whose input table already declares a primary key.
h3. Proposal
Add \{{int[] upsertKeyColumns()}} to \{{TableSemantics}}, populated by planner
via \{{FlinkRelMetadataQuery.getUpsertKeys(input)}} collapsed to one candidate
via \{{UpsertKeyUtil.smallestKey(...)}}. Returns empty array when no upsert key
is derivable (pure append-only sources, post-Sort streams that destroyed the
key, etc.) or during type inference (metadata not yet computed).
Plumbing:
{\{TableSemantics}} (\{{flink-table-common}}): add \{{default int[]
upsertKeyColumns() { return new int[0]; }}} method. Default avoids breaking
source-compatibility for rare external implementor.
{\{RuntimeTableSemantics}} (\{{flink-table-runtime}}): add serializable
\{{int[] upsertKeyColumns}} field, constructor parameter, accessor.
{\{StreamExecProcessTableFunction}} (\{{flink-table-planner}}): persist
\{{List<int[]> inputUpsertKeys}} as new \{{@JsonProperty}} field (one entry per
table input). Default to per-input empty arrays for older compiled plans
(back-compat).
{\{StreamPhysicalProcessTableFunction.translateToExecNode}}: derive upsert keys
for each input via \{{FlinkRelMetadataQuery.getUpsertKeys(input)}} +
\{{UpsertKeyUtil.smallestKey(...).orElse(new int[0])}}. Pass list to new
ExecNode field.
{\{OperatorBindingCallContext}} / \{{OperatorBindingTableSemantics}}: extend to
accept \{{inputUpsertKeys}} so value is visible at specialization time
(function constructor sees operator-binding context, not runtime context).
{\{BridgingSqlFunction.toCallContext}} /
\{{ProcessTableRunnerGenerator.generate}}: thread \{{inputUpsertKeys}} through
codegen path.
{\{TableSemanticsMock}} and \{{TestHarnessTableSemantics}}: accept optional
setter for unit testing.
{\{TO_CHANGELOG}} runtime (\{{ToChangelogFunction}}): consume
\{{tableSemantics.upsertKeyColumns()}} to decide which columns to preserve on
DELETE when \{{produces_full_deletes=false}}. Solves row-semantics case from
FLINK-39636 without requiring \{{PARTITION BY}}.
> Expose input upsert key on TableSemantics for ProcessTableFunctions
> -------------------------------------------------------------------
>
> Key: FLINK-39735
> URL: https://issues.apache.org/jira/browse/FLINK-39735
> Project: Flink
> Issue Type: Sub-task
> Components: Table SQL / API
> Reporter: Ramin Gharib
> Priority: Major
>
> ## Problem
> `ProcessTableFunction`s receive a `TableSemantics` for each table-typed
> argument. Semantics surface today exposes:
> - `dataType()`
> - `partitionByColumns()` — only populated when caller wrote `PARTITION BY` in
> SQL.
> - `orderByColumns()` / `orderByDirections()`
> - `timeColumn()`
> - `changelogMode()` — planner-derived, lifecycle-aware (empty during type
> inference).
> Does **not** expose input table's upsert key. Planner already derives it via
> `FlinkRelMetadataQuery.getUpsertKeys(input)` (metadata handler chain in
> `FlinkRelMdUpsertKeys`), but result is invisible to PTF.
> Forces PTF authors to either:
> - Require caller to repeat key via `PARTITION BY` even when planner already
> knows it from PK constraints, or
> - Re-derive upsert keys inside function (impossible — at constructor time,
> function has only `TableSemantics`, not a `RelNode` or `RelMetadataQuery`).
> Concrete impact for `TO_CHANGELOG` (FLINK-39636): without access to input
> upsert key, function cannot emit partial DELETE rows that preserve identity
> columns in row semantics. Current workaround is "add `PARTITION BY <pk>`",
> unergonomic for users whose input table already declares a primary key.
> ## Proposal
> Add `int[] upsertKeyColumns()` to `TableSemantics`, populated by planner via
> `FlinkRelMetadataQuery.getUpsertKeys(input)` collapsed to one candidate via
> `UpsertKeyUtil.smallestKey(...)`. Returns empty array when no upsert key is
> derivable (pure append-only sources, post-Sort streams that destroyed the
> key, etc.) or during type inference (metadata not yet computed).
> Plumbing:
> 1. **`TableSemantics`** (`flink-table-common`): add `default int[]
> upsertKeyColumns()` method returning empty array. Default avoids breaking
> source-compatibility for rare external implementor.
> 2. **`RuntimeTableSemantics`** (`flink-table-runtime`): add serializable
> `int[] upsertKeyColumns` field, constructor parameter, accessor.
> 3. **`StreamExecProcessTableFunction`** (`flink-table-planner`): persist
> `List<int[]> inputUpsertKeys` as new `@JsonProperty` field (one entry per
> table input). Default to per-input empty arrays for older compiled plans
> (back-compat).
> 4. **`StreamPhysicalProcessTableFunction.translateToExecNode`**: derive
> upsert keys for each input via `FlinkRelMetadataQuery.getUpsertKeys(input)` +
> `UpsertKeyUtil.smallestKey(...).orElse(new int[0])`. Pass list to new
> ExecNode field.
> 5. **`OperatorBindingCallContext`** / **`OperatorBindingTableSemantics`**:
> extend to accept `inputUpsertKeys` so value is visible at specialization time
> (function constructor sees operator-binding context, not runtime context).
> 6. **`BridgingSqlFunction.toCallContext`** /
> **`ProcessTableRunnerGenerator.generate`**: thread `inputUpsertKeys` through
> codegen path.
> 7. **`TableSemanticsMock`** and **`TestHarnessTableSemantics`**: accept
> optional setter for unit testing.
> 8. **`TO_CHANGELOG` runtime** (`ToChangelogFunction`): consume
> `tableSemantics.upsertKeyColumns()` to decide which columns to preserve on
> DELETE when `produces_full_deletes=false`. Solves row-semantics case from
> FLINK-39636 without requiring `PARTITION BY`.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
