[flink] branch master updated: [FLINK-27901][python] Support TableEnvironment.create(configuration)

[di...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

dianfu pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git


The following commit(s) were added to refs/heads/master by this push:
     new f011e52ac99 [FLINK-27901][python] Support TableEnvironment.create(configuration)
f011e52ac99 is described below

commit f011e52ac9943639c62f68b44bddcbacfa96e51c
Author: Juntao Hu <ma...@gmail.com>
AuthorDate: Mon Jun 6 15:18:58 2022 +0800

    [FLINK-27901][python] Support TableEnvironment.create(configuration)
    
    This closes #19883.
---
 docs/content.zh/docs/dev/python/python_config.md |  3 +++
 docs/content/docs/dev/python/python_config.md    |  3 +++
 flink-python/pyflink/table/table_environment.py  | 17 ++++++++++++-----
 3 files changed, 18 insertions(+), 5 deletions(-)

diff --git a/docs/content.zh/docs/dev/python/python_config.md b/docs/content.zh/docs/dev/python/python_config.md
index 269594c4877..138969d7863 100644
--- a/docs/content.zh/docs/dev/python/python_config.md
+++ b/docs/content.zh/docs/dev/python/python_config.md
@@ -65,6 +65,9 @@ env_settings = EnvironmentSettings \
     .with_configuration(config) \
     .build()
 table_env = TableEnvironment.create(env_settings)
+
+# or directly pass config into create method
+table_env = TableEnvironment.create(config)
 ```
 
 ## Python Options
diff --git a/docs/content/docs/dev/python/python_config.md b/docs/content/docs/dev/python/python_config.md
index 8b44eafd659..e388aac49a4 100644
--- a/docs/content/docs/dev/python/python_config.md
+++ b/docs/content/docs/dev/python/python_config.md
@@ -65,6 +65,9 @@ env_settings = EnvironmentSettings \
     .with_configuration(config) \
     .build()
 table_env = TableEnvironment.create(env_settings)
+
+# or directly pass config into create method
+table_env = TableEnvironment.create(config)
 ```
 
 ## Python Options
diff --git a/flink-python/pyflink/table/table_environment.py b/flink-python/pyflink/table/table_environment.py
index 8c3047fa6a9..544b796bd0c 100644
--- a/flink-python/pyflink/table/table_environment.py
+++ b/flink-python/pyflink/table/table_environment.py
@@ -23,6 +23,7 @@ from typing import Union, List, Tuple, Iterable
 
 from py4j.java_gateway import get_java_class, get_method
 
+from pyflink.common.configuration import Configuration
 from pyflink.datastream import StreamExecutionEnvironment
 from pyflink.table.sources import TableSource
 
@@ -98,18 +99,24 @@ class TableEnvironment(object):
         self._open()
 
     @staticmethod
-    def create(environment_settings: EnvironmentSettings) -> 'TableEnvironment':
+    def create(environment_settings: Union[EnvironmentSettings, Configuration]) \
+            -> 'TableEnvironment':
         """
         Creates a table environment that is the entry point and central context for creating Table
         and SQL API programs.
 
-        :param environment_settings: The environment settings used to instantiate the
-                                     :class:`~pyflink.table.TableEnvironment`.
+        :param environment_settings: The configuration or environment settings used to instantiate
+            the :class:`~pyflink.table.TableEnvironment`, the name is for backward compatibility.
         :return: The :class:`~pyflink.table.TableEnvironment`.
         """
         gateway = get_gateway()
-        j_tenv = gateway.jvm.TableEnvironment.create(
-            environment_settings._j_environment_settings)
+        if isinstance(environment_settings, Configuration):
+            environment_settings = EnvironmentSettings.new_instance() \
+                .with_configuration(environment_settings).build()
+        elif not isinstance(environment_settings, EnvironmentSettings):
+            raise TypeError("argument should be EnvironmentSettings or Configuration")
+
+        j_tenv = gateway.jvm.TableEnvironment.create(environment_settings._j_environment_settings)
         return TableEnvironment(j_tenv)
 
     def from_table_source(self, table_source: 'TableSource') -> 'Table':