raminqaf commented on code in PR #27901:
URL: https://github.com/apache/flink/pull/27901#discussion_r3099067753
##########
flink-python/pyflink/table/table.py:
##########
@@ -1188,6 +1188,42 @@ def insert_into(
)
+ def from_changelog(self, *arguments: Expression) -> 'Table':
+ """
+ Converts this append-only table with an explicit operation code column
into a
+ (potentially updating) dynamic table using the built-in
``FROM_CHANGELOG``
+ process table function.
+
+ Each input row is expected to have a string operation code column
(default: ``op``)
+ that indicates the change operation. The operation column is
interpreted by the
+ engine and removed from the output.
+
+ Example:
+ ::
+
+ >>> from pyflink.table.expressions import descriptor, map_
+ >>> # Default: reads 'op' column with standard change operation
names
+ >>> table.from_changelog()
+ >>> # With custom op column name
+ >>> table.from_changelog(
+ ... descriptor("operation").as_argument("op")
+ ... )
+ >>> # With custom op_mapping
+ >>> table.from_changelog(
+ ... descriptor("op").as_argument("op"),
+ ... map_("c, r", "INSERT",
+ ... "ub", "UPDATE_BEFORE",
+ ... "ua", "UPDATE_AFTER",
+ ... "d", "DELETE").as_argument("op_mapping")
+ ... )
+
+ :param arguments: Optional named arguments for ``op`` and
``op_mapping``.
+ :return: A (potentially updating) :class:`~pyflink.table.Table` with
the op column
+ removed and proper change operation semantics.
+ """
+ return
Table(self._j_table.fromChangelog(to_expression_jarray(arguments)), self._t_env)
+
+
Review Comment:
Reverted the comment and will do a followup PR on this:
https://issues.apache.org/jira/browse/FLINK-39479
--
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]
