| ################################################################################ |
| # 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. |
| ################################################################################ |
| import glob |
| import os |
| import subprocess |
| |
| from pyflink.find_flink_home import _find_flink_source_root |
| from pyflink.java_gateway import get_gateway |
| from pyflink.table import ResultKind, ExplainDetail |
| from pyflink.table import expressions as expr |
| from pyflink.testing import source_sink_utils |
| from pyflink.testing.test_case_utils import PyFlinkStreamTableTestCase, \ |
| PyFlinkTestCase |
| |
| |
| class StreamSqlTests(PyFlinkStreamTableTestCase): |
| |
| def test_sql_ddl(self): |
| self.t_env.execute_sql("create temporary function func1 as " |
| "'pyflink.table.tests.test_udf.add' language python") |
| table = self.t_env.from_elements([(1, 2)]) \ |
| .alias("a", "b") \ |
| .select(expr.call("func1", expr.col("a"), expr.col("b"))) |
| plan = table.explain() |
| self.assertGreaterEqual(plan.find("== Optimized Physical Plan =="), 0) |
| self.assertGreaterEqual(plan.find("PythonCalc(select=[func1(f0, f1) AS _c0])"), 0) |
| |
| plan = table.explain(ExplainDetail.PLAN_ADVICE) |
| self.assertGreaterEqual(plan.find("== Optimized Physical Plan With Advice =="), 0) |
| self.assertGreaterEqual(plan.find("No available advice..."), 0) |
| |
| def test_sql_query(self): |
| t_env = self.t_env |
| source = t_env.from_elements([(1, "Hi", "Hello"), (2, "Hello", "Hello")], ["a", "b", "c"]) |
| sink_table_ddl = """ |
| CREATE TABLE sinks_sql_query(a BIGINT, b STRING, c STRING) WITH ('connector'='test-sink') |
| """ |
| t_env.execute_sql(sink_table_ddl) |
| |
| result = t_env.sql_query("select a + 1, b, c from %s" % source) |
| result.execute_insert("sinks_sql_query").wait() |
| actual = source_sink_utils.results() |
| |
| expected = ['+I[2, Hi, Hello]', '+I[3, Hello, Hello]'] |
| self.assert_equals(actual, expected) |
| |
| def test_execute_sql(self): |
| t_env = self.t_env |
| table_result = t_env.execute_sql("create table tbl" |
| "(" |
| " a bigint," |
| " b int," |
| " c varchar" |
| ") with (" |
| " 'connector' = 'COLLECTION'," |
| " 'is-bounded' = 'false'" |
| ")") |
| self.assertIsNone(table_result.get_job_client()) |
| self.assert_equals(table_result.get_resolved_schema().get_column_names(), ["result"]) |
| self.assertEqual(table_result.get_result_kind(), ResultKind.SUCCESS) |
| table_result.print() |
| |
| table_result = t_env.execute_sql("alter table tbl set ('k1' = 'a', 'k2' = 'b')") |
| self.assertIsNone(table_result.get_job_client()) |
| self.assert_equals(table_result.get_resolved_schema().get_column_names(), ["result"]) |
| self.assertEqual(table_result.get_result_kind(), ResultKind.SUCCESS) |
| table_result.print() |
| |
| sink_table_ddl = """ |
| CREATE TABLE sinks(k1 BIGINT, k2 INT, c STRING) WITH ('connector'='test-sink') |
| """ |
| t_env.execute_sql(sink_table_ddl) |
| table_result = t_env.execute_sql("insert into sinks select * from tbl") |
| |
| job_execution_result = table_result.get_job_client().get_job_execution_result().result() |
| self.assertIsNotNone(job_execution_result.get_job_id()) |
| self.assert_equals(table_result.get_resolved_schema().get_column_names(), |
| ["default_catalog.default_database.sinks"]) |
| self.assertEqual(table_result.get_result_kind(), ResultKind.SUCCESS_WITH_CONTENT) |
| table_result.print() |
| |
| table_result = t_env.execute_sql("drop table tbl") |
| self.assertIsNone(table_result.get_job_client()) |
| self.assert_equals(table_result.get_resolved_schema().get_column_names(), ["result"]) |
| self.assertEqual(table_result.get_result_kind(), ResultKind.SUCCESS) |
| table_result.print() |
| |
| |
| class JavaSqlTests(PyFlinkTestCase): |
| """ |
| We need to start these Java tests from python process to make sure that Python environment is |
| available when the tests are running. |
| """ |
| |
| @staticmethod |
| def get_classpath(): |
| return get_gateway().jvm.System.getProperties().get("java.class.path") |
| |
| @staticmethod |
| def get_java_executable(): |
| return get_gateway().jvm.System.getProperty("java.home") + "/bin/java" |
| |
| def get_jar_path(self, jar_path_pattern): |
| test_jar_path = glob.glob(os.path.join(_find_flink_source_root(), jar_path_pattern)) |
| if not test_jar_path: |
| self.fail("'%s' is not available. Please compile the test jars first." |
| % jar_path_pattern) |
| if len(test_jar_path) > 1: |
| self.fail("There are multiple jars matches the pattern: %s, the jars are: %s" |
| % (jar_path_pattern, test_jar_path)) |
| return test_jar_path[0] |
| |
| def test_java_sql_ddl(self): |
| test_class = "org.apache.flink.client.python.PythonFunctionFactoryTest" |
| test_jar_pattern = "flink-python/target/artifacts/testJavaDdl.jar" |
| test_jar_path = self.get_jar_path(test_jar_pattern) |
| test_classpath = self.get_classpath() + os.pathsep + test_jar_path |
| java_executable = self.get_java_executable() |
| subprocess.check_output([java_executable, |
| "-XX:+IgnoreUnrecognizedVMOptions", |
| "--add-opens=java.base/java.lang=ALL-UNNAMED", |
| "-cp", test_classpath, test_class], shell=False) |
