[FLINK-39913][table] Improve the error message when a foreground query emits an upsert changelog

[raminqaf]

What is the purpose of the change

A foreground query (a bare SELECT collected through the anonymous result sink, which has no primary key) over a relation that emits an upsert changelog fails with an unhelpful error: Can't generate a valid execution plan for the given query followed by the optimized plan. A common trigger is FROM_CHANGELOG with PARTITION BY and an op_mapping without UPDATE_BEFORE. This change appends actionable guidance to that error. Behavior is unchanged; only the message is improved.

Brief change log

• FlinkChangelogModeInferenceProgram: when the rejected plan involves updates, append guidance explaining the upsert-vs-retract mismatch and the two fixes (declare a PRIMARY KEY on the sink, or make the upstream emit UPDATE_BEFORE). Gated on ModifyKind.UPDATE so unrelated planning failures keep the original message.
• If there is not mismatch in the foreground sink operator and the other operators changelog mode, the error is produced as before and includes the changelog mode.

Verifiying the output

Locally setup a test and verified the exception message. The plan now includes the changelog mode of the operators. Before, it was all set to NONE.

Can't generate a valid execution plan for the given query.

There is a changelog mismatch between two operators. One produces an upsert changelog (UPDATE_AFTER without UPDATE_BEFORE). The other requires a retract changelog (UPDATE_BEFORE and UPDATE_AFTER), for example a sink without a primary key. To resolve it, declare a PRIMARY KEY on the sink, or make the input produce UPDATE_BEFORE.

The conflict is at:
Sink(changelogMode=[NONE])
  +- ProcessTableFunction(changelogMode=[I,UA,D])

Does this pull request potentially affect one of the following parts:

• Dependencies: no
• @Public(Evolving) API: no
• Serializers: no
• Runtime per-record code paths: no
• Deployment / recovery: no
• S3 connector: no

Documentation

• New feature: no

[flinkbot]

CI report:

• 1ee1ea0 Azure: SUCCESS

Bot commands

The @flinkbot bot supports the following commands:

• @flinkbot run azure re-run the last Azure build