raminqaf commented on code in PR #27901: URL: https://github.com/apache/flink/pull/27901#discussion_r3080036547
########## docs/content/docs/sql/reference/queries/changelog.md: ########## @@ -30,9 +30,117 @@ Flink SQL provides built-in process table functions (PTFs) for working with chan | Function | Description | |:---------|:------------| +| [FROM_CHANGELOG](#from_changelog) | Converts an append-only table with operation codes into a dynamic table | | [TO_CHANGELOG](#to_changelog) | Converts a dynamic table into an append-only table with explicit operation codes | -<!-- Placeholder for future FROM_CHANGELOG function --> +## FROM_CHANGELOG + +The `FROM_CHANGELOG` PTF converts an append-only table with an explicit operation code column into a dynamic table (i.e. an updating table). Each input row is expected to have a string column that indicates the change operation. The op column is removed from the output and the row is emitted with the corresponding change operation. + +This is useful when consuming Change Data Capture (CDC) streams from systems like Debezium, Maxwell, or Canal, where events arrive as flat append-only records with an explicit operation field. It's also useful to be used in combination with the TO_CHANGELOG function, when the user wants to turn the append-only table back into an updating table after doing some specific transformation to the events. + +### Syntax + +```sql +SELECT * FROM FROM_CHANGELOG( + input => TABLE source_table [PARTITION BY key_col], + [op => DESCRIPTOR(op_column_name),] + [op_mapping => MAP['c, r', 'INSERT', 'u', 'UPDATE_AFTER', 'd', 'DELETE']] +) +``` + +`PARTITION BY` is optional when the mapping includes `UPDATE_BEFORE` (retract mode). It is required when the mapping produces upsert mode (no `UPDATE_BEFORE`), because downstream operators need a key for updates and deletes. When provided, records are distributed by the partition key for parallel processing. + +### Parameters + +| Parameter | Required | Description | +|:-------------|:---------|:------------| +| `input` | Yes | The input table. Must be append-only. `PARTITION BY` is optional for retract mode (with `UPDATE_BEFORE`) and required for upsert mode (without `UPDATE_BEFORE`). | +| `op` | No | A `DESCRIPTOR` with a single column name for the operation code column. Defaults to `op`. The column must exist in the input table and be of type STRING. | +| `op_mapping` | No | A `MAP<STRING, STRING>` mapping user-defined operation codes to change operation names. Keys are user codes (e.g., `'c'`, `'u'`, `'d'`), values are change operation names (`INSERT`, `UPDATE_BEFORE`, `UPDATE_AFTER`, `DELETE`). Keys can contain comma-separated codes to map multiple codes to the same operation (e.g., `'c, r'`). When provided, only mapped codes are forwarded - unmapped codes are dropped. Each change operation may appear at most once across all entries. | Review Comment: Addressed -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
