You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by "Ramin Gharib (Jira)" <ji...@apache.org> on 2026/04/13 12:21:00 UTC
[jira] [Updated] (FLINK-39430) Add `invalid_op_handling` parameter to FROM_CHANGELOG PTF
[ https://issues.apache.org/jira/browse/FLINK-39430?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ]
Ramin Gharib updated FLINK-39430:
---------------------------------
Description:
Add an `invalid_op_handling` parameter to the `FROM_CHANGELOG` built-in process table function as specified in FLIP-564.
Currently, when FROM_CHANGELOG encounters an operation code in the input that is not present in the `op_mapping`, the row is silently dropped. The `invalid_op_handling` parameter would allow users to control this behavior with three modes:
- {*}{*}`FAIL` — throw an exception when an unmapped operation code is encountered (strict mode)
- `LOG` — log a warning and drop the row
- `SKIP` — silently drop the row (current default behavior)
h3. Example
{code:java}
SELECT * FROM FROM_CHANGELOG(
input => TABLE cdc_stream,
op => DESCRIPTOR(__op),
op_mapping => MAP['c, r', 'INSERT', 'u', 'UPDATE_AFTER', 'd', 'DELETE'],
invalid_op_handling => 'FAIL'
){code}
h3. Scope
Add `invalid_op_handling` as a new optional scalar argument (`StaticArgument.scalar`) of type `STRING` to the FROM_CHANGELOG definition in `BuiltInFunctionDefinitions`
Add input validation in `FromChangelogTypeStrategy` to check that the value is one of `FAIL`, `LOG`, `SKIP`
Implement the three modes in `FromChangelogFunction.eval()`: * `FAIL`: throw a runtime exception with details about the unmapped code
* `LOG`: log a warning via `FunctionContext` and skip the row
* `SKIP`: current behavior (return without collecting)
* Update documentation in `changelog.md`
* Add semantic tests for each mode
was:
Add an `invalid_op_handling` parameter to the `FROM_CHANGELOG` built-in process table function as specified in FLIP-564.
Currently, when FROM_CHANGELOG encounters an operation code in the input that is not present in the `op_mapping`, the row is silently dropped. The `invalid_op_handling` parameter would allow users to control this behavior with three modes:
- *{*}`FAIL`{*}* — throw an exception when an unmapped operation code is encountered (strict mode)
- *{*}`LOG`{*}* — log a warning and drop the row
- *{*}`SKIP`{*}* — silently drop the row (current default behavior)
h3. Example
{code:java}
SELECT * FROM FROM_CHANGELOG(
input => TABLE cdc_stream,
op => DESCRIPTOR(__op),
op_mapping => MAP['c, r', 'INSERT', 'u', 'UPDATE_AFTER', 'd', 'DELETE'],
invalid_op_handling => 'FAIL'
){code}
h3. Scope
Add `invalid_op_handling` as a new optional scalar argument (`StaticArgument.scalar`) of type `STRING` to the FROM_CHANGELOG definition in `BuiltInFunctionDefinitions`
Add input validation in `FromChangelogTypeStrategy` to check that the value is one of `FAIL`, `LOG`, `SKIP`
Implement the three modes in `FromChangelogFunction.eval()`: * `FAIL`: throw a runtime exception with details about the unmapped code
* `LOG`: log a warning via `FunctionContext` and skip the row
* `SKIP`: current behavior (return without collecting)
* Update documentation in `changelog.md`
* Add semantic tests for each mode
> Add `invalid_op_handling` parameter to FROM_CHANGELOG PTF
> ----------------------------------------------------------
>
> Key: FLINK-39430
> URL: https://issues.apache.org/jira/browse/FLINK-39430
> Project: Flink
> Issue Type: Sub-task
> Components: Table SQL / API
> Reporter: Ramin Gharib
> Priority: Major
>
> Add an `invalid_op_handling` parameter to the `FROM_CHANGELOG` built-in process table function as specified in FLIP-564.
> Currently, when FROM_CHANGELOG encounters an operation code in the input that is not present in the `op_mapping`, the row is silently dropped. The `invalid_op_handling` parameter would allow users to control this behavior with three modes:
> - {*}{*}`FAIL` — throw an exception when an unmapped operation code is encountered (strict mode)
> - `LOG` — log a warning and drop the row
> - `SKIP` — silently drop the row (current default behavior)
> h3. Example
> {code:java}
> SELECT * FROM FROM_CHANGELOG(
> input => TABLE cdc_stream,
> op => DESCRIPTOR(__op),
> op_mapping => MAP['c, r', 'INSERT', 'u', 'UPDATE_AFTER', 'd', 'DELETE'],
> invalid_op_handling => 'FAIL'
> ){code}
>
> h3. Scope
> Add `invalid_op_handling` as a new optional scalar argument (`StaticArgument.scalar`) of type `STRING` to the FROM_CHANGELOG definition in `BuiltInFunctionDefinitions`
> Add input validation in `FromChangelogTypeStrategy` to check that the value is one of `FAIL`, `LOG`, `SKIP`
> Implement the three modes in `FromChangelogFunction.eval()`: * `FAIL`: throw a runtime exception with details about the unmapped code
> * `LOG`: log a warning via `FunctionContext` and skip the row
> * `SKIP`: current behavior (return without collecting)
> * Update documentation in `changelog.md`
> * Add semantic tests for each mode
--
This message was sent by Atlassian Jira
(v8.20.10#820010)