[FLINK-39261][table] add from_changelog to python api

Author: raminqaf

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)
+
+
 @PublicEvolving()
 class GroupedTable(object):
     """

flink-python/pyflink/table/tests/test_table_completeness.py

@@ -44,7 +44,6 @@ def excluded_methods(cls):
             'process',
             'partitionBy',
             'toChangelog',
-            'fromChangelog',
         }
 
     @classmethod