godfrey he created FLINK-6436: --------------------------------- Summary: Bug in CommonCorrelate's generateCollector method when using udtf with udf Key: FLINK-6436 URL: https://issues.apache.org/jira/browse/FLINK-6436 Project: Flink Issue Type: Bug Components: Table API & SQL Reporter: godfrey he Assignee: godfrey he
When running table api code (using udtf with udf) like: {code} val env = ExecutionEnvironment.getExecutionEnvironment val tableEnv = TableEnvironment.getTableEnvironment(env, config) val in = testData(env).toTable(tableEnv).as('a, 'b, 'c) val func0 = new TableFunc0 val result = in .join(func0('c)) .where(Func18('name, "J") && ('age > 20)) .select('c, 'name, 'age) .toDataSet[Row] {code} An exception will be thrown, as below: {code} Caused by: java.lang.Exception: The user defined 'open(Configuration)' method in class org.apache.flink.table.runtime.CorrelateFlatMapRunner caused an exception: Table program cannot be compiled. This is a bug. Please file an issue. at org.apache.flink.runtime.operators.BatchTask.openUserCode(BatchTask.java:1335) at org.apache.flink.runtime.operators.chaining.ChainedFlatMapDriver.openTask(ChainedFlatMapDriver.java:47) at org.apache.flink.runtime.operators.BatchTask.openChainedTasks(BatchTask.java:1375) at org.apache.flink.runtime.operators.DataSourceTask.invoke(DataSourceTask.java:129) at org.apache.flink.runtime.taskmanager.Task.run(Task.java:700) at java.lang.Thread.run(Thread.java:745) Caused by: org.apache.flink.api.common.InvalidProgramException: Table program cannot be compiled. This is a bug. Please file an issue. at org.apache.flink.table.codegen.Compiler$class.compile(Compiler.scala:36) at org.apache.flink.table.runtime.CorrelateFlatMapRunner.compile(CorrelateFlatMapRunner.scala:30) at org.apache.flink.table.runtime.CorrelateFlatMapRunner.open(CorrelateFlatMapRunner.scala:47) at org.apache.flink.api.common.functions.util.FunctionUtils.openFunction(FunctionUtils.java:36) at org.apache.flink.runtime.operators.BatchTask.openUserCode(BatchTask.java:1333) ... 5 more Caused by: org.codehaus.commons.compiler.CompileException: Line 72, Column 121: Unknown variable or type "function_org$apache$flink$table$expressions$utils$Func18$$752fa7801f5195571801442d482dcbc8" at org.codehaus.janino.UnitCompiler.compileError(UnitCompiler.java:11523) {code} -- This message was sent by Atlassian JIRA (v6.3.15#6346)