Qilong Wang created FLINK-40304:
-----------------------------------
Summary: 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
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:
```java
IntStream.range(pos, partitionKeyCount)
```
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:
```text
Can't generate a valid execution plan for the given query.
The conflict is at:
Sink(expectedChangelogMode=[I,UA,PD])
+- ProcessTableFunction(changelogMode=[I,UA,D])
```
*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.
*Proposed Fix*
Use the cumulative offset when calculating the exclusive upper bound:
```java
IntStream.range(pos, pos + partitionKeyCount)
```
Add a regression test with two partitioned table arguments and an upsert sink
whose primary key corresponds to the partition column from the second argument.
*Verification*
`ProcessTableFunctionTest#testUpsertKeyWithMultipleTableArgs` reproduces the
planning failure with the old implementation and succeeds after the fix.
The complete `ProcessTableFunctionTest` suite passes with 69 tests.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
