[FLINK-25986][python] Add remaining FLIP-190 API methods to Python Table API

This closes #26281.

Author: autophagy

flink-python/docs/reference/pyflink.table/compiled_plans.rst

@@ -0,0 +1,98 @@
+.. ################################################################################
+     Licensed to the Apache Software Foundation (ASF) under one
+     or more contributor license agreements.  See the NOTICE file
+     distributed with this work for additional information
+     regarding copyright ownership.  The ASF licenses this file
+     to you under the Apache License, Version 2.0 (the
+     "License"); you may not use this file except in compliance
+     with the License.  You may obtain a copy of the License at
+
+         http://www.apache.org/licenses/LICENSE-2.0
+
+     Unless required by applicable law or agreed to in writing, software
+     distributed under the License is distributed on an "AS IS" BASIS,
+     WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+     See the License for the specific language governing permissions and
+    limitations under the License.
+   ################################################################################
+
+
+==============
+Compiled Plans
+==============
+
+CompiledPlan
+------------
+
+Represents an immutable, fully optimized, and executable entity that has been compiled from a
+Table & SQL API pipeline definition. It encodes operators, expressions, functions, data types,
+and table connectors.
+
+Every new Flink version might introduce improved optimizer rules, more efficient operators,
+and other changes that impact the behavior of previously defined pipelines. In order to ensure
+backwards compatibility and enable stateful streaming job upgrades, compiled plans can be
+persisted and reloaded across Flink versions. See the
+`website documentation <https://flink.apache.org/documentation/>`_ for more information about
+provided guarantees during stateful pipeline upgrades.
+
+A plan can be compiled from a SQL query using
+:func:`~pyflink.table.TableEnvironment.compile_plan_sql`.
+It can be persisted using :func:`~pyflink.table.CompiledPlan.write_to_file` or by manually
+extracting the JSON representation with func:`~pyflink.table.CompiledPlan.as_json_string`.
+A plan can be loaded back from a file or a string using
+:func:`~pyflink.table.TableEnvironment.load_plan` with a :class:`~pyflink.table.PlanReference`.
+Instances can be executed using :func:`~pyflink.table.CompiledPlan.execute`.
+
+Depending on the configuration, permanent catalog metadata (such as information about tables
+and functions) will be persisted in the plan as well. Anonymous/inline objects will be
+persisted (including schema and options) if possible or fail the compilation otherwise.
+For temporary objects, only the identifier is part of the plan and the object needs to be
+present in the session context during a restore.
+
+JSON encoding is assumed to be the default representation of a compiled plan in all API
+endpoints, and is the format used to persist the plan to files by default.
+For advanced use cases, :func:`~pyflink.table.CompiledPlan.as_smile_bytes` provides a binary
+format representation of the compiled plan.
+
+.. note::
+    Plan restores assume a stable session context. Configuration, loaded modules and
+    catalogs, and temporary objects must not change. Schema evolution and changes of function
+    signatures are not supported.
+
+.. currentmodule:: pyflink.table
+
+.. autosummary::
+    :toctree: api/
+
+    CompiledPlan.as_json_string
+    CompiledPlan.as_smile_bytes
+    CompiledPlan.write_to_file
+    CompiledPlan.get_flink_version
+    CompiledPlan.print_json_string
+    CompiledPlan.execute
+    CompiledPlan.explain
+    CompiledPlan.print_explain
+
+PlanReference
+-------------
+
+Unresolved pointer to a persisted plan.
+
+A plan represents a static, executable entity that has been compiled from a Table & SQL API
+pipeline definition.
+
+You can load the content of this reference into a :class:`~pyflink.table.CompiledPlan`
+using :func:`~pyflink.table.TableEnvironment.load_plan` with a
+:class:`~pyflink.table.PlanReference`, or you can directly load and execute it with
+:func:`~pyflink.table.TableEnvironment.execute_plan`.
+
+.. seealso:: :class:`~pyflink.table.CompiledPlan`
+
+.. currentmodule:: pyflink.table
+
+.. autosummary::
+    :toctree: api/
+
+    PlanReference.from_file
+    PlanReference.from_json_string
+    PlanReference.from_smile_bytes

flink-python/docs/reference/pyflink.table/descriptors.rst

@@ -111,7 +111,7 @@ TableSchema
 
 A table schema that represents a table's structure with field names and data types.
 
-.. currentmodule:: pyflink.table.table_schema
+.. currentmodule:: pyflink.table
 
 .. autosummary::
     :toctree: api/
@@ -132,7 +132,7 @@ ChangelogMode
 
 The set of changes contained in a changelog.
 
-.. currentmodule:: pyflink.table.changelog_mode
+.. currentmodule:: pyflink.table
 
 .. autosummary::
     :toctree: api/
@@ -147,11 +147,13 @@ TablePipeline
 
 Describes a complete pipeline from one or more source tables to a sink table.
 
-.. currentmodule:: pyflink.table.table_pipeline
+.. currentmodule:: pyflink.table
 
 .. autosummary::
     :toctree: api/
 
+    TablePipeline.compile_plan
     TablePipeline.execute
     TablePipeline.explain
+    TablePipeline.print_explain
     TablePipeline.get_sink_identifier

flink-python/docs/reference/pyflink.table/index.rst

@@ -34,3 +34,4 @@ This page gives an overview of all public PyFlink Table API.
     descriptors
     statement_set
     catalog
+    compiled_plans

flink-python/docs/reference/pyflink.table/statement_set.rst

@@ -28,15 +28,17 @@ The planner can optimize all added statements together and then submit them as o
 
 The added statements will be cleared when calling the :func:`~StatementSet.execute` method.
 
-.. currentmodule:: pyflink.table.statement_set
+.. currentmodule:: pyflink.table
 
 .. autosummary::
     :toctree: api/
 
     StatementSet.add_insert_sql
     StatementSet.attach_as_datastream
     StatementSet.add_insert
+    StatementSet.compile_plan
     StatementSet.explain
+    StatementSet.print_explain
     StatementSet.execute
 
 
@@ -45,7 +47,7 @@ TableResult
 
 A :class:`~pyflink.table.TableResult` is the representation of the statement execution result.
 
-.. currentmodule:: pyflink.table.table_result
+.. currentmodule:: pyflink.table
 
 .. autosummary::
     :toctree: api/
@@ -73,7 +75,7 @@ The statement (e.g. DDL, USE) executes successfully, and the result only contain
 The statement (e.g. DML, DQL, SHOW) executes successfully, and the result contains important
 content.
 
-.. currentmodule:: pyflink.table.table_result
+.. currentmodule:: pyflink.table
 
 .. autosummary::
     :toctree: api/

flink-python/docs/reference/pyflink.table/table.rst

@@ -88,6 +88,7 @@ Example:
     Table.execute
     Table.execute_insert
     Table.explain
+    Table.print_explain
     Table.fetch
     Table.filter
     Table.flat_aggregate

flink-python/docs/reference/pyflink.table/table_environment.rst

@@ -42,7 +42,7 @@ Example:
 :func:`~EnvironmentSettings.in_streaming_mode` or :func:`~EnvironmentSettings.in_batch_mode`
 might be convenient as shortcuts.
 
-.. currentmodule:: pyflink.table.environment_settings
+.. currentmodule:: pyflink.table
 
 .. autosummary::
     :toctree: api/
@@ -144,13 +144,14 @@ keyword, thus must be escaped) in a catalog named 'cat.1' and database named 'db
     other Flink APIs, it might be necessary to use one of the available language-specific table
     environments in the corresponding bridging modules.
 
-.. currentmodule:: pyflink.table.table_environment
+.. currentmodule:: pyflink.table
 
 .. autosummary::
     :toctree: api/
 
     TableEnvironment.add_python_archive
     TableEnvironment.add_python_file
+    TableEnvironment.compile_plan_sql
     TableEnvironment.create
     TableEnvironment.create_java_function
     TableEnvironment.create_java_temporary_function
@@ -169,6 +170,7 @@ keyword, thus must be escaped) in a catalog named 'cat.1' and database named 'db
     TableEnvironment.drop_temporary_table
     TableEnvironment.drop_temporary_view
     TableEnvironment.drop_view
+    TableEnvironment.execute_plan
     TableEnvironment.execute_sql
     TableEnvironment.explain_sql
     TableEnvironment.from_descriptor
@@ -190,6 +192,7 @@ keyword, thus must be escaped) in a catalog named 'cat.1' and database named 'db
     TableEnvironment.list_user_defined_functions
     TableEnvironment.list_views
     TableEnvironment.load_module
+    TableEnvironment.load_plan
     TableEnvironment.create_catalog
     TableEnvironment.register_catalog
     TableEnvironment.set_python_requirements
@@ -202,7 +205,7 @@ keyword, thus must be escaped) in a catalog named 'cat.1' and database named 'db
 StreamTableEnvironment
 ----------------------
 
-.. currentmodule:: pyflink.table.table_environment
+.. currentmodule:: pyflink.table
 
 .. autosummary::
     :toctree: api/

flink-python/pyflink/table/__init__.py

@@ -113,11 +113,13 @@
 from __future__ import absolute_import
 
 from pyflink.table.changelog_mode import ChangelogMode
+from pyflink.table.compiled_plan import CompiledPlan
 from pyflink.table.data_view import DataView, ListView, MapView
 from pyflink.table.environment_settings import EnvironmentSettings
 from pyflink.table.explain_detail import ExplainDetail
 from pyflink.table.expression import Expression
 from pyflink.table.module import Module, ModuleEntry
+from pyflink.table.plan_reference import PlanReference
 from pyflink.table.result_kind import ResultKind
 from pyflink.table.schema import Schema
 from pyflink.table.sql_dialect import SqlDialect
@@ -127,6 +129,7 @@
 from pyflink.table.table_config import TableConfig
 from pyflink.table.table_descriptor import TableDescriptor, FormatDescriptor
 from pyflink.table.table_environment import (TableEnvironment, StreamTableEnvironment)
+from pyflink.table.table_pipeline import TablePipeline
 from pyflink.table.table_result import TableResult
 from pyflink.table.table_schema import TableSchema
 from pyflink.table.types import DataTypes, UserDefinedType, Row, RowKind
@@ -168,4 +171,7 @@
     'ChangelogMode',
     'ExplainDetail',
     'ResultKind',
+    'CompiledPlan',
+    'PlanReference',
+    'TablePipeline'
 ]