[
https://issues.apache.org/jira/browse/FLINK-40304?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Qilong Wang updated FLINK-40304:
--------------------------------
Description:
*Description*
For a process table function (PTF) with multiple table arguments, the planner
can derive incorrect output positions for the partition columns after the first
argument.
`StreamPhysicalProcessTableFunction.toPartitionColumns()` tracks the cumulative
output offset in `pos`, but uses `partitionKeyCount` directly as the exclusive
upper bound:
{code:java}
IntStream.range(pos, partitionKeyCount){code}
However, `partitionKeyCount` is a count rather than an absolute output position.
For example, when two table arguments each have one partition key, the first
argument correctly produces `\{0}`, while the second argument incorrectly
produces an empty key instead of `\{1}`.
Upsert PTFs use these partition columns as upsert key candidates. Consequently,
when a downstream upsert sink uses a partition column from the second table
argument as its primary key, the planner cannot satisfy the required upsert key
and rejects an otherwise valid query with:
{code:java}
Can't generate a valid execution plan for the given query.{code}
*Expected Behavior*
The planner should account for the cumulative output offset when deriving
partition columns.
When two table arguments each have one partition key, the PTF should expose
`\{0}` and `\{1}` as upsert key candidates. A downstream sink keyed by either
partition column should therefore be able to consume the PTF's upsert changelog.
was:
*Description*
For a process table function (PTF) with multiple table arguments, the planner
can derive incorrect output positions for the partition columns after the first
argument.
`StreamPhysicalProcessTableFunction.toPartitionColumns()` tracks the cumulative
output offset in `pos`, but uses `partitionKeyCount` directly as the exclusive
upper bound:
{code:java}
IntStream.range(pos, partitionKeyCount){code}
However, `partitionKeyCount` is a count rather than an absolute output position.
For example, when two table arguments each have one partition key, the first
argument correctly produces `\{0}`, while the second argument incorrectly
produces an empty key instead of `\{1}`.
Upsert PTFs use these partition columns as upsert key candidates. Consequently,
when a downstream upsert sink uses a partition column from the second table
argument as its primary key, the planner cannot satisfy the required upsert key
and rejects an otherwise valid query with:
{code:java}
Can't generate a valid execution plan for the given query.{code}
*Expected Behavior*
The planner should account for the cumulative output offset when deriving
partition columns.
When two table arguments each have one partition key, the PTF should expose
`\{0}` and `\{1}` as upsert key candidates. A downstream sink keyed by either
partition column should therefore be able to consume the PTF's upsert changelog.
*Actual Behavior*
Partition columns belonging to the second and subsequent table arguments can be
empty or truncated. The corresponding upsert key candidate is missing, causing
changelog mode inference and physical plan generation to fail.
> Incorrect upsert key inference for PTFs with multiple table arguments
> ---------------------------------------------------------------------
>
> Key: FLINK-40304
> URL: https://issues.apache.org/jira/browse/FLINK-40304
> Project: Flink
> Issue Type: Bug
> Components: Table SQL / Planner
> Affects Versions: 2.4.0
> Reporter: Qilong Wang
> Priority: Major
> Fix For: 2.4.0
>
>
> *Description*
> For a process table function (PTF) with multiple table arguments, the planner
> can derive incorrect output positions for the partition columns after the
> first argument.
> `StreamPhysicalProcessTableFunction.toPartitionColumns()` tracks the
> cumulative output offset in `pos`, but uses `partitionKeyCount` directly as
> the exclusive upper bound:
> {code:java}
> IntStream.range(pos, partitionKeyCount){code}
> However, `partitionKeyCount` is a count rather than an absolute output
> position.
> For example, when two table arguments each have one partition key, the first
> argument correctly produces `\{0}`, while the second argument incorrectly
> produces an empty key instead of `\{1}`.
> Upsert PTFs use these partition columns as upsert key candidates.
> Consequently, when a downstream upsert sink uses a partition column from the
> second table argument as its primary key, the planner cannot satisfy the
> required upsert key and rejects an otherwise valid query with:
> {code:java}
> Can't generate a valid execution plan for the given query.{code}
> *Expected Behavior*
> The planner should account for the cumulative output offset when deriving
> partition columns.
> When two table arguments each have one partition key, the PTF should expose
> `\{0}` and `\{1}` as upsert key candidates. A downstream sink keyed by either
> partition column should therefore be able to consume the PTF's upsert
> changelog.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
