[
https://issues.apache.org/jira/browse/FLINK-39430?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=18073255#comment-18073255
]
Shekhar Prasad Rajak commented on FLINK-39430:
----------------------------------------------
Hi,
Please assign to me.
> 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 PARTITION_BY id
> 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)
