[FLINK-39442][python] Add descriptor() and to_changelog() to Python Table API

This closes #27934.

Author: gustavodemorais

flink-python/pyflink/table/expressions.py

@@ -34,7 +34,7 @@
            'concat_ws', 'uuid', 'null_of', 'log', 'with_columns', 'without_columns', 'json',
            'json_string', 'json_object', 'json_object_agg', 'json_array', 'json_array_agg',
            'call', 'call_sql', 'source_watermark', 'to_timestamp_ltz', 'from_unixtime', 'to_date',
-           'to_timestamp', 'convert_tz', 'unix_timestamp']
+           'to_timestamp', 'convert_tz', 'unix_timestamp', 'descriptor']
 
 
 def _leaf_op(op_name: str) -> Expression:
@@ -578,6 +578,26 @@ def map_from_arrays(key, value) -> Expression:
     return _binary_op("mapFromArrays", key, value)
 
 
+@PublicEvolving()
+def descriptor(*column_names: str) -> Expression:
+    """
+    Creates a literal describing an arbitrary, unvalidated list of column names.
+
+    Passing a list of columns can be useful for parameterizing a function. In particular,
+    it enables declaring the ``on_time`` argument for process table functions.
+
+    Example:
+    ::
+
+        >>> descriptor("ts_column")
+        >>> descriptor("col1", "col2")
+
+    :param column_names: One or more column names.
+    :return: A descriptor expression.
+    """
+    return _varargs_op("descriptor", *column_names)
+
+
 @PublicEvolving()
 def object_of(class_name: str, *args) -> Expression:
     """

flink-python/pyflink/table/table.py

@@ -1187,6 +1187,39 @@ def insert_into(
                 t_env=self._t_env,
             )
 
+    def to_changelog(self, *arguments: Expression) -> 'Table':
+        """
+        Converts this table into an append-only table with an explicit operation code
+        column using the built-in ``TO_CHANGELOG`` process table function.
+
+        Each input row - regardless of its original change operation - is emitted as an
+        INSERT-only row with a string ``op`` column indicating the original operation
+        (INSERT, UPDATE_BEFORE, UPDATE_AFTER, DELETE).
+
+        Example:
+        ::
+
+            >>> from pyflink.table.expressions import descriptor, map_
+            >>> # Default: adds 'op' column with standard change operation names
+            >>> table.to_changelog()
+            >>> # Custom op column name and mapping
+            >>> table.to_changelog(
+            ...     descriptor("op_code").as_argument("op"),
+            ...     map_("INSERT", "I", "UPDATE_AFTER", "U").as_argument("op_mapping")
+            ... )
+            >>> # Deletion flag pattern
+            >>> table.to_changelog(
+            ...     descriptor("deleted").as_argument("op"),
+            ...     map_("INSERT, UPDATE_AFTER", "false",
+            ...          "DELETE", "true").as_argument("op_mapping")
+            ... )
+
+        :param arguments: Optional named arguments for ``op`` and ``op_mapping``.
+        :return: An append-only :class:`~pyflink.table.Table` with an ``op`` column prepended
+                 to the input columns.
+        """
+        return Table(self._j_table.toChangelog(to_expression_jarray(arguments)), self._t_env)
+
 
 @PublicEvolving()
 class GroupedTable(object):

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

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