[jira] [Commented] (FLINK-39735) Expose input upsert key on TableSemantics for ProcessTableFunctions

["Jubin Soni (Jira)" <ji...@apache.org>]

    [ https://issues.apache.org/jira/browse/FLINK-39735?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=18083026#comment-18083026 ] 

Jubin Soni commented on FLINK-39735:
------------------------------------

For the approach, I’ll expose upsert keys end-to-end for PTFs by adding {{TableSemantics#upsertKeyColumns()}} with a default empty array for compatibility. I’ll wire it through {{{}RuntimeTableSemantics{}}}, derive it in the planner via {{{}FlinkRelMetadataQuery.getUpsertKeys(){}}}, and collapse per input using {{{}UpsertKeyUtil.smallestKey(...){}}}.

I’ll pass it through {{StreamPhysicalProcessTableFunction}} and {{StreamExecProcessTableFunction}} (with JSON back-compat), and extend the specialization/type-inference path so it’s available at construction time (before metadata is available, it returns {{{}new int[0]{}}}).

Finally, I’ll update codegen and {{TO_CHANGELOG}} to consume it, with fallback to existing {{PARTITION BY}} behavior when the upsert key is absent.

Can someone assign this ticket to me?

> 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
>              Labels: pull-request-available
>
> h2. 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()}}
>  * 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.
> h2. 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 a default {{upsertKeyColumns()}} method returning an empty array. 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 a 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(...){}}}. 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{}}}.



--
This message was sent by Atlassian Jira
(v8.20.10#820010)