Python UDF not work in sql mode

[0akarma]

I have tested these two ways setting python udf function in sql-client , they worked well.
But when I tried to set Dynamic Properties, Program Args, flink-conf.yaml and sql code in streampark, but they are not worked well :(
I have no idea about the differences between org.apache.streampark.flink.cli.SqlClientand the origin SqlClient .

# Way 1
bin/sql-client.sh \
-pyexec /opt/flink_data/requirements/py_env/jm_venv/bin/python3.7 \
-pyclientexec /opt/flink_data/requirements/py_env/jm_venv/bin/python3.7 \
-pyfs file:///opt/flink_data/requirements/udfs

# Way 2
Flink SQL> SET 'python.client.executable'='/opt/flink_data/requirements/py_env/jm_venv/bin/python3.7';
[INFO] Session property has been set.

Flink SQL> SET 'python.executable'='/opt/flink_data/requirements/py_env/jm_venv/bin/python3.7';
[INFO] Session property has been set.

Flink SQL> SET 'python.files'='file:///opt/flink_data/requirements/udfs';
[INFO] Session property has been set.


# the way to registry udf function
Flink SQL> CREATE TEMPORARY FUNCTION is_domain AS 'python_udf.is_domain' LANGUAGE PYTHON;
[INFO] Execute statement succeed.

# the way to use udf function
Flink SQL> select is_domain('www.google.com');

error log

2023-01-24 16:12:14,973 WARN  org.apache.flink.client.deployment.application.DetachedApplicationRunner [] - Could not execute application: 
org.apache.flink.client.program.ProgramInvocationException: The main method caused an error: SQL validation failed. Cannot instantiate user-defined function 'is_domain'.
	at org.apache.flink.client.program.PackagedProgram.callMainMethod(PackagedProgram.java:372) ~[flink-dist-1.15.0.jar:1.15.0]
	at org.apache.flink.client.program.PackagedProgram.invokeInteractiveModeForExecution(PackagedProgram.java:222) ~[flink-dist-1.15.0.jar:1.15.0]
	at org.apache.flink.client.ClientUtils.executeProgram(ClientUtils.java:114) ~[flink-dist-1.15.0.jar:1.15.0]
	at org.apache.flink.client.deployment.application.DetachedApplicationRunner.tryExecuteJobs(DetachedApplicationRunner.java:84) ~[flink-dist-1.15.0.jar:1.15.0]
	at org.apache.flink.client.deployment.application.DetachedApplicationRunner.run(DetachedApplicationRunner.java:70) ~[flink-dist-1.15.0.jar:1.15.0]
	at org.apache.flink.runtime.webmonitor.handlers.JarRunHandler.lambda$handleRequest$0(JarRunHandler.java:107) ~[flink-dist-1.15.0.jar:1.15.0]
	at java.util.concurrent.CompletableFuture$AsyncSupply.run(Unknown Source) [?:?]
	at java.lang.Thread.run(Unknown Source) [?:?]
Caused by: org.apache.flink.table.api.ValidationException: SQL validation failed. Cannot instantiate user-defined function 'is_domain'.
	at org.apache.flink.table.planner.calcite.FlinkPlannerImpl.org$apache$flink$table$planner$calcite$FlinkPlannerImpl$$validate(FlinkPlannerImpl.scala:185) ~[?:?]
	at org.apache.flink.table.planner.calcite.FlinkPlannerImpl.validate(FlinkPlannerImpl.scala:110) ~[?:?]
	at org.apache.flink.table.planner.operations.SqlToOperationConverter.convert(SqlToOperationConverter.java:237) ~[?:?]
	at org.apache.flink.table.planner.delegation.ParserImpl.parse(ParserImpl.java:105) ~[?:?]
	at org.apache.flink.table.api.internal.StatementSetImpl.addInsertSql(StatementSetImpl.java:62) ~[flink-table-api-java-uber-1.15.0.jar:1.15.0]
	at org.apache.streampark.flink.core.FlinkSqlExecutor$.$anonfun$executeSql$3(FlinkSqlExecutor.scala:106) ~[?:?]
	at org.apache.streampark.flink.core.FlinkSqlExecutor$.$anonfun$executeSql$3$adapted(FlinkSqlExecutor.scala:48) ~[?:?]
	at scala.collection.immutable.List.foreach(List.scala:388) ~[flink-scala_2.12-1.15.0.jar:1.15.0]
	at org.apache.streampark.flink.core.FlinkSqlExecutor$.executeSql(FlinkSqlExecutor.scala:48) ~[?:?]
	at org.apache.streampark.flink.core.FlinkTableTrait.sql(FlinkTableTrait.scala:47) ~[?:?]
	at org.apache.streampark.flink.cli.SqlClient$BatchSqlApp$.handle(SqlClient.scala:60) ~[?:?]
	at org.apache.streampark.flink.core.scala.FlinkTable.main(FlinkTable.scala:40) ~[?:?]
	at org.apache.streampark.flink.core.scala.FlinkTable.main$(FlinkTable.scala:37) ~[?:?]
	at org.apache.streampark.flink.cli.SqlClient$BatchSqlApp$.main(SqlClient.scala:59) ~[?:?]
	at org.apache.streampark.flink.cli.SqlClient$.delayedEndpoint$org$apache$streampark$flink$cli$SqlClient$1(SqlClient.scala:54) ~[?:?]
	at org.apache.streampark.flink.cli.SqlClient$delayedInit$body.apply(SqlClient.scala:30) ~[?:?]
	at scala.Function0.apply$mcV$sp(Function0.scala:34) ~[flink-scala_2.12-1.15.0.jar:1.15.0]
	at scala.Function0.apply$mcV$sp$(Function0.scala:34) ~[flink-scala_2.12-1.15.0.jar:1.15.0]
	at scala.runtime.AbstractFunction0.apply$mcV$sp(AbstractFunction0.scala:12) ~[flink-scala_2.12-1.15.0.jar:1.15.0]
	at scala.App.$anonfun$main$1$adapted(App.scala:76) ~[flink-scala_2.12-1.15.0.jar:1.15.0]
	at scala.collection.immutable.List.foreach(List.scala:388) ~[flink-scala_2.12-1.15.0.jar:1.15.0]
	at scala.App.main(App.scala:76) ~[flink-scala_2.12-1.15.0.jar:1.15.0]
	at scala.App.main$(App.scala:74) ~[flink-scala_2.12-1.15.0.jar:1.15.0]
	at org.apache.streampark.flink.cli.SqlClient$.main(SqlClient.scala:30) ~[?:?]
	at org.apache.streampark.flink.cli.SqlClient.main(SqlClient.scala) ~[?:?]
	at jdk.internal.reflect.NativeMethodAccessorImpl.invoke0(Native Method) ~[?:?]
	at jdk.internal.reflect.NativeMethodAccessorImpl.invoke(Unknown Source) ~[?:?]
	at jdk.internal.reflect.DelegatingMethodAccessorImpl.invoke(Unknown Source) ~[?:?]
	at java.lang.reflect.Method.invoke(Unknown Source) ~[?:?]
	at org.apache.flink.client.program.PackagedProgram.callMainMethod(PackagedProgram.java:355) ~[flink-dist-1.15.0.jar:1.15.0]
	... 7 more
Caused by: org.apache.flink.table.api.ValidationException: Cannot instantiate user-defined function 'is_domain'.
	at org.apache.flink.table.functions.UserDefinedFunctionHelper.instantiateFunction(UserDefinedFunctionHelper.java:219) ~[flink-table-api-java-uber-1.15.0.jar:1.15.0]
	at org.apache.flink.table.catalog.FunctionCatalog.getFunctionDefinition(FunctionCatalog.java:653) ~[flink-table-api-java-uber-1.15.0.jar:1.15.0]
	at org.apache.flink.table.catalog.FunctionCatalog.resolvePreciseFunctionReference(FunctionCatalog.java:557) ~[flink-table-api-java-uber-1.15.0.jar:1.15.0]
	at org.apache.flink.table.catalog.FunctionCatalog.lambda$resolveAmbiguousFunctionReference$3(FunctionCatalog.java:620) ~[flink-table-api-java-uber-1.15.0.jar:1.15.0]
	at java.util.Optional.orElseGet(Unknown Source) ~[?:?]
	at org.apache.flink.table.catalog.FunctionCatalog.resolveAmbiguousFunctionReference(FunctionCatalog.java:620) ~[flink-table-api-java-uber-1.15.0.jar:1.15.0]
	at org.apache.flink.table.catalog.FunctionCatalog.lookupFunction(FunctionCatalog.java:357) ~[flink-table-api-java-uber-1.15.0.jar:1.15.0]
	at org.apache.flink.table.planner.catalog.FunctionCatalogOperatorTable.lookupOperatorOverloads(FunctionCatalogOperatorTable.java:98) ~[?:?]
	at org.apache.calcite.sql.util.ChainedSqlOperatorTable.lookupOperatorOverloads(ChainedSqlOperatorTable.java:67) ~[?:?]
	at org.apache.calcite.sql.validate.SqlValidatorImpl.performUnconditionalRewrites(SqlValidatorImpl.java:1183) ~[?:?]
	at org.apache.calcite.sql.validate.SqlValidatorImpl.performUnconditionalRewrites(SqlValidatorImpl.java:1200) ~[?:?]
	at org.apache.calcite.sql.validate.SqlValidatorImpl.performUnconditionalRewrites(SqlValidatorImpl.java:1169) ~[?:?]
	at org.apache.calcite.sql.validate.SqlValidatorImpl.validateScopedExpression(SqlValidatorImpl.java:945) ~[?:?]
	at org.apache.calcite.sql.validate.SqlValidatorImpl.validate(SqlValidatorImpl.java:704) ~[?:?]
	at org.apache.flink.table.planner.calcite.FlinkPlannerImpl.org$apache$flink$table$planner$calcite$FlinkPlannerImpl$$validate(FlinkPlannerImpl.scala:170) ~[?:?]
	at org.apache.flink.table.planner.calcite.FlinkPlannerImpl.validate(FlinkPlannerImpl.scala:110) ~[?:?]
	at org.apache.flink.table.planner.operations.SqlToOperationConverter.convert(SqlToOperationConverter.java:237) ~[?:?]
	at org.apache.flink.table.planner.delegation.ParserImpl.parse(ParserImpl.java:105) ~[?:?]
	at org.apache.flink.table.api.internal.StatementSetImpl.addInsertSql(StatementSetImpl.java:62) ~[flink-table-api-java-uber-1.15.0.jar:1.15.0]
	at org.apache.streampark.flink.core.FlinkSqlExecutor$.$anonfun$executeSql$3(FlinkSqlExecutor.scala:106) ~[?:?]
	at org.apache.streampark.flink.core.FlinkSqlExecutor$.$anonfun$executeSql$3$adapted(FlinkSqlExecutor.scala:48) ~[?:?]
	at scala.collection.immutable.List.foreach(List.scala:388) ~[flink-scala_2.12-1.15.0.jar:1.15.0]
	at org.apache.streampark.flink.core.FlinkSqlExecutor$.executeSql(FlinkSqlExecutor.scala:48) ~[?:?]
	at org.apache.streampark.flink.core.FlinkTableTrait.sql(FlinkTableTrait.scala:47) ~[?:?]
	at org.apache.streampark.flink.cli.SqlClient$BatchSqlApp$.handle(SqlClient.scala:60) ~[?:?]
	at org.apache.streampark.flink.core.scala.FlinkTable.main(FlinkTable.scala:40) ~[?:?]
	at org.apache.streampark.flink.core.scala.FlinkTable.main$(FlinkTable.scala:37) ~[?:?]
	at org.apache.streampark.flink.cli.SqlClient$BatchSqlApp$.main(SqlClient.scala:59) ~[?:?]
	at org.apache.streampark.flink.cli.SqlClient$.delayedEndpoint$org$apache$streampark$flink$cli$SqlClient$1(SqlClient.scala:54) ~[?:?]
	at org.apache.streampark.flink.cli.SqlClient$delayedInit$body.apply(SqlClient.scala:30) ~[?:?]
	at scala.Function0.apply$mcV$sp(Function0.scala:34) ~[flink-scala_2.12-1.15.0.jar:1.15.0]
	at scala.Function0.apply$mcV$sp$(Function0.scala:34) ~[flink-scala_2.12-1.15.0.jar:1.15.0]
	at scala.runtime.AbstractFunction0.apply$mcV$sp(AbstractFunction0.scala:12) ~[flink-scala_2.12-1.15.0.jar:1.15.0]
	at scala.App.$anonfun$main$1$adapted(App.scala:76) ~[flink-scala_2.12-1.15.0.jar:1.15.0]
	at scala.collection.immutable.List.foreach(List.scala:388) ~[flink-scala_2.12-1.15.0.jar:1.15.0]
	at scala.App.main(App.scala:76) ~[flink-scala_2.12-1.15.0.jar:1.15.0]
	at scala.App.main$(App.scala:74) ~[flink-scala_2.12-1.15.0.jar:1.15.0]
	at org.apache.streampark.flink.cli.SqlClient$.main(SqlClient.scala:30) ~[?:?]
	at org.apache.streampark.flink.cli.SqlClient.main(SqlClient.scala) ~[?:?]
	at jdk.internal.reflect.NativeMethodAccessorImpl.invoke0(Native Method) ~[?:?]
	at jdk.internal.reflect.NativeMethodAccessorImpl.invoke(Unknown Source) ~[?:?]
	at jdk.internal.reflect.DelegatingMethodAccessorImpl.invoke(Unknown Source) ~[?:?]
	at java.lang.reflect.Method.invoke(Unknown Source) ~[?:?]
	at org.apache.flink.client.program.PackagedProgram.callMainMethod(PackagedProgram.java:355) ~[flink-dist-1.15.0.jar:1.15.0]
	... 7 more
Caused by: java.lang.IllegalStateException: Instantiating python function 'python_udf.is_domain' failed.
	at org.apache.flink.table.functions.python.utils.PythonFunctionUtils.getPythonFunction(PythonFunctionUtils.java:48) ~[flink-table-api-java-uber-1.15.0.jar:1.15.0]
	at org.apache.flink.table.functions.UserDefinedFunctionHelper.instantiateFunction(UserDefinedFunctionHelper.java:206) ~[flink-table-api-java-uber-1.15.0.jar:1.15.0]
	at org.apache.flink.table.catalog.FunctionCatalog.getFunctionDefinition(FunctionCatalog.java:653) ~[flink-table-api-java-uber-1.15.0.jar:1.15.0]
	at org.apache.flink.table.catalog.FunctionCatalog.resolvePreciseFunctionReference(FunctionCatalog.java:557) ~[flink-table-api-java-uber-1.15.0.jar:1.15.0]
	at org.apache.flink.table.catalog.FunctionCatalog.lambda$resolveAmbiguousFunctionReference$3(FunctionCatalog.java:620) ~[flink-table-api-java-uber-1.15.0.jar:1.15.0]
	at java.util.Optional.orElseGet(Unknown Source) ~[?:?]
	at org.apache.flink.table.catalog.FunctionCatalog.resolveAmbiguousFunctionReference(FunctionCatalog.java:620) ~[flink-table-api-java-uber-1.15.0.jar:1.15.0]
	at org.apache.flink.table.catalog.FunctionCatalog.lookupFunction(FunctionCatalog.java:357) ~[flink-table-api-java-uber-1.15.0.jar:1.15.0]
	at org.apache.flink.table.planner.catalog.FunctionCatalogOperatorTable.lookupOperatorOverloads(FunctionCatalogOperatorTable.java:98) ~[?:?]
	at org.apache.calcite.sql.util.ChainedSqlOperatorTable.lookupOperatorOverloads(ChainedSqlOperatorTable.java:67) ~[?:?]
	at org.apache.calcite.sql.validate.SqlValidatorImpl.performUnconditionalRewrites(SqlValidatorImpl.java:1183) ~[?:?]
	at org.apache.calcite.sql.validate.SqlValidatorImpl.performUnconditionalRewrites(SqlValidatorImpl.java:1200) ~[?:?]
	at org.apache.calcite.sql.validate.SqlValidatorImpl.performUnconditionalRewrites(SqlValidatorImpl.java:1169) ~[?:?]
	at org.apache.calcite.sql.validate.SqlValidatorImpl.validateScopedExpression(SqlValidatorImpl.java:945) ~[?:?]
	at org.apache.calcite.sql.validate.SqlValidatorImpl.validate(SqlValidatorImpl.java:704) ~[?:?]
	at org.apache.flink.table.planner.calcite.FlinkPlannerImpl.org$apache$flink$table$planner$calcite$FlinkPlannerImpl$$validate(FlinkPlannerImpl.scala:170) ~[?:?]
	at org.apache.flink.table.planner.calcite.FlinkPlannerImpl.validate(FlinkPlannerImpl.scala:110) ~[?:?]
	at org.apache.flink.table.planner.operations.SqlToOperationConverter.convert(SqlToOperationConverter.java:237) ~[?:?]
	at org.apache.flink.table.planner.delegation.ParserImpl.parse(ParserImpl.java:105) ~[?:?]
	at org.apache.flink.table.api.internal.StatementSetImpl.addInsertSql(StatementSetImpl.java:62) ~[flink-table-api-java-uber-1.15.0.jar:1.15.0]
	at org.apache.streampark.flink.core.FlinkSqlExecutor$.$anonfun$executeSql$3(FlinkSqlExecutor.scala:106) ~[?:?]
	at org.apache.streampark.flink.core.FlinkSqlExecutor$.$anonfun$executeSql$3$adapted(FlinkSqlExecutor.scala:48) ~[?:?]
	at scala.collection.immutable.List.foreach(List.scala:388) ~[flink-scala_2.12-1.15.0.jar:1.15.0]
	at org.apache.streampark.flink.core.FlinkSqlExecutor$.executeSql(FlinkSqlExecutor.scala:48) ~[?:?]
	at org.apache.streampark.flink.core.FlinkTableTrait.sql(FlinkTableTrait.scala:47) ~[?:?]
	at org.apache.streampark.flink.cli.SqlClient$BatchSqlApp$.handle(SqlClient.scala:60) ~[?:?]
	at org.apache.streampark.flink.core.scala.FlinkTable.main(FlinkTable.scala:40) ~[?:?]
	at org.apache.streampark.flink.core.scala.FlinkTable.main$(FlinkTable.scala:37) ~[?:?]
	at org.apache.streampark.flink.cli.SqlClient$BatchSqlApp$.main(SqlClient.scala:59) ~[?:?]
	at org.apache.streampark.flink.cli.SqlClient$.delayedEndpoint$org$apache$streampark$flink$cli$SqlClient$1(SqlClient.scala:54) ~[?:?]
	at org.apache.streampark.flink.cli.SqlClient$delayedInit$body.apply(SqlClient.scala:30) ~[?:?]
	at scala.Function0.apply$mcV$sp(Function0.scala:34) ~[flink-scala_2.12-1.15.0.jar:1.15.0]
	at scala.Function0.apply$mcV$sp$(Function0.scala:34) ~[flink-scala_2.12-1.15.0.jar:1.15.0]
	at scala.runtime.AbstractFunction0.apply$mcV$sp(AbstractFunction0.scala:12) ~[flink-scala_2.12-1.15.0.jar:1.15.0]
	at scala.App.$anonfun$main$1$adapted(App.scala:76) ~[flink-scala_2.12-1.15.0.jar:1.15.0]
	at scala.collection.immutable.List.foreach(List.scala:388) ~[flink-scala_2.12-1.15.0.jar:1.15.0]
	at scala.App.main(App.scala:76) ~[flink-scala_2.12-1.15.0.jar:1.15.0]
	at scala.App.main$(App.scala:74) ~[flink-scala_2.12-1.15.0.jar:1.15.0]
	at org.apache.streampark.flink.cli.SqlClient$.main(SqlClient.scala:30) ~[?:?]
	at org.apache.streampark.flink.cli.SqlClient.main(SqlClient.scala) ~[?:?]
	at jdk.internal.reflect.NativeMethodAccessorImpl.invoke0(Native Method) ~[?:?]
	at jdk.internal.reflect.NativeMethodAccessorImpl.invoke(Unknown Source) ~[?:?]
	at jdk.internal.reflect.DelegatingMethodAccessorImpl.invoke(Unknown Source) ~[?:?]
	at java.lang.reflect.Method.invoke(Unknown Source) ~[?:?]
	at org.apache.flink.client.program.PackagedProgram.callMainMethod(PackagedProgram.java:355) ~[flink-dist-1.15.0.jar:1.15.0]
	... 7 more
Caused by: java.lang.ClassNotFoundException: org.apache.flink.client.python.PythonFunctionFactory
	at java.net.URLClassLoader.findClass(Unknown Source) ~[?:?]
	at java.lang.ClassLoader.loadClass(Unknown Source) ~[?:?]
	at org.apache.flink.util.FlinkUserCodeClassLoader.loadClassWithoutExceptionHandling(FlinkUserCodeClassLoader.java:68) ~[flink-dist-1.15.0.jar:1.15.0]
	at org.apache.flink.util.ChildFirstClassLoader.loadClassWithoutExceptionHandling(ChildFirstClassLoader.java:65) ~[flink-dist-1.15.0.jar:1.15.0]
	at org.apache.flink.util.FlinkUserCodeClassLoader.loadClass(FlinkUserCodeClassLoader.java:52) ~[flink-dist-1.15.0.jar:1.15.0]
	at java.lang.ClassLoader.loadClass(Unknown Source) ~[?:?]
	at org.apache.flink.runtime.execution.librarycache.FlinkUserCodeClassLoaders$SafetyNetWrapperClassLoader.loadClass(FlinkUserCodeClassLoaders.java:172) ~[flink-dist-1.15.0.jar:1.15.0]
	at java.lang.Class.forName0(Native Method) ~[?:?]
	at java.lang.Class.forName(Unknown Source) ~[?:?]
	at org.apache.flink.table.functions.python.utils.PythonFunctionUtils.getPythonFunction(PythonFunctionUtils.java:34) ~[flink-table-api-java-uber-1.15.0.jar:1.15.0]
	at org.apache.flink.table.functions.UserDefinedFunctionHelper.instantiateFunction(UserDefinedFunctionHelper.java:206) ~[flink-table-api-java-uber-1.15.0.jar:1.15.0]
	at org.apache.flink.table.catalog.FunctionCatalog.getFunctionDefinition(FunctionCatalog.java:653) ~[flink-table-api-java-uber-1.15.0.jar:1.15.0]
	at org.apache.flink.table.catalog.FunctionCatalog.resolvePreciseFunctionReference(FunctionCatalog.java:557) ~[flink-table-api-java-uber-1.15.0.jar:1.15.0]
	at org.apache.flink.table.catalog.FunctionCatalog.lambda$resolveAmbiguousFunctionReference$3(FunctionCatalog.java:620) ~[flink-table-api-java-uber-1.15.0.jar:1.15.0]
	at java.util.Optional.orElseGet(Unknown Source) ~[?:?]
	at org.apache.flink.table.catalog.FunctionCatalog.resolveAmbiguousFunctionReference(FunctionCatalog.java:620) ~[flink-table-api-java-uber-1.15.0.jar:1.15.0]
	at org.apache.flink.table.catalog.FunctionCatalog.lookupFunction(FunctionCatalog.java:357) ~[flink-table-api-java-uber-1.15.0.jar:1.15.0]
	at org.apache.flink.table.planner.catalog.FunctionCatalogOperatorTable.lookupOperatorOverloads(FunctionCatalogOperatorTable.java:98) ~[?:?]
	at org.apache.calcite.sql.util.ChainedSqlOperatorTable.lookupOperatorOverloads(ChainedSqlOperatorTable.java:67) ~[?:?]
	at org.apache.calcite.sql.validate.SqlValidatorImpl.performUnconditionalRewrites(SqlValidatorImpl.java:1183) ~[?:?]
	at org.apache.calcite.sql.validate.SqlValidatorImpl.performUnconditionalRewrites(SqlValidatorImpl.java:1200) ~[?:?]
	at org.apache.calcite.sql.validate.SqlValidatorImpl.performUnconditionalRewrites(SqlValidatorImpl.java:1169) ~[?:?]
	at org.apache.calcite.sql.validate.SqlValidatorImpl.validateScopedExpression(SqlValidatorImpl.java:945) ~[?:?]
	at org.apache.calcite.sql.validate.SqlValidatorImpl.validate(SqlValidatorImpl.java:704) ~[?:?]
	at org.apache.flink.table.planner.calcite.FlinkPlannerImpl.org$apache$flink$table$planner$calcite$FlinkPlannerImpl$$validate(FlinkPlannerImpl.scala:170) ~[?:?]
	at org.apache.flink.table.planner.calcite.FlinkPlannerImpl.validate(FlinkPlannerImpl.scala:110) ~[?:?]
	at org.apache.flink.table.planner.operations.SqlToOperationConverter.convert(SqlToOperationConverter.java:237) ~[?:?]
	at org.apache.flink.table.planner.delegation.ParserImpl.parse(ParserImpl.java:105) ~[?:?]
	at org.apache.flink.table.api.internal.StatementSetImpl.addInsertSql(StatementSetImpl.java:62) ~[flink-table-api-java-uber-1.15.0.jar:1.15.0]
	at org.apache.streampark.flink.core.FlinkSqlExecutor$.$anonfun$executeSql$3(FlinkSqlExecutor.scala:106) ~[?:?]
	at org.apache.streampark.flink.core.FlinkSqlExecutor$.$anonfun$executeSql$3$adapted(FlinkSqlExecutor.scala:48) ~[?:?]
	at scala.collection.immutable.List.foreach(List.scala:388) ~[flink-scala_2.12-1.15.0.jar:1.15.0]
	at org.apache.streampark.flink.core.FlinkSqlExecutor$.executeSql(FlinkSqlExecutor.scala:48) ~[?:?]
	at org.apache.streampark.flink.core.FlinkTableTrait.sql(FlinkTableTrait.scala:47) ~[?:?]
	at org.apache.streampark.flink.cli.SqlClient$BatchSqlApp$.handle(SqlClient.scala:60) ~[?:?]
	at org.apache.streampark.flink.core.scala.FlinkTable.main(FlinkTable.scala:40) ~[?:?]
	at org.apache.streampark.flink.core.scala.FlinkTable.main$(FlinkTable.scala:37) ~[?:?]
	at org.apache.streampark.flink.cli.SqlClient$BatchSqlApp$.main(SqlClient.scala:59) ~[?:?]
	at org.apache.streampark.flink.cli.SqlClient$.delayedEndpoint$org$apache$streampark$flink$cli$SqlClient$1(SqlClient.scala:54) ~[?:?]
	at org.apache.streampark.flink.cli.SqlClient$delayedInit$body.apply(SqlClient.scala:30) ~[?:?]
	at scala.Function0.apply$mcV$sp(Function0.scala:34) ~[flink-scala_2.12-1.15.0.jar:1.15.0]
	at scala.Function0.apply$mcV$sp$(Function0.scala:34) ~[flink-scala_2.12-1.15.0.jar:1.15.0]
	at scala.runtime.AbstractFunction0.apply$mcV$sp(AbstractFunction0.scala:12) ~[flink-scala_2.12-1.15.0.jar:1.15.0]
	at scala.App.$anonfun$main$1$adapted(App.scala:76) ~[flink-scala_2.12-1.15.0.jar:1.15.0]
	at scala.collection.immutable.List.foreach(List.scala:388) ~[flink-scala_2.12-1.15.0.jar:1.15.0]
	at scala.App.main(App.scala:76) ~[flink-scala_2.12-1.15.0.jar:1.15.0]
	at scala.App.main$(App.scala:74) ~[flink-scala_2.12-1.15.0.jar:1.15.0]
	at org.apache.streampark.flink.cli.SqlClient$.main(SqlClient.scala:30) ~[?:?]
	at org.apache.streampark.flink.cli.SqlClient.main(SqlClient.scala) ~[?:?]
	at jdk.internal.reflect.NativeMethodAccessorImpl.invoke0(Native Method) ~[?:?]
	at jdk.internal.reflect.NativeMethodAccessorImpl.invoke(Unknown Source) ~[?:?]
	at jdk.internal.reflect.DelegatingMethodAccessorImpl.invoke(Unknown Source) ~[?:?]
	at java.lang.reflect.Method.invoke(Unknown Source) ~[?:?]
	at org.apache.flink.client.program.PackagedProgram.callMainMethod(PackagedProgram.java:355) ~[flink-dist-1.15.0.jar:1.15.0]
	... 7 more

[MonsterChenzhuo]

@0akarma hi,I think you can add my WeChat: shihong960429