[
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:
```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:
{code:java}
IntStream.range(pos, pos + partitionKeyCount) {code}
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.
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:
```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.
> 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:
> ```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:
> {code:java}
> IntStream.range(pos, pos + partitionKeyCount) {code}
> 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)
