[jira] [Updated] (FLINK-6355) TableEnvironment support register TableFunction

[Kaibo Zhou (JIRA)]

     [ 
https://issues.apache.org/jira/browse/FLINK-6355?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
 ]

Kaibo Zhou updated FLINK-6355:
------------------------------
    Description: 
Currently {{TableEnvironment}} only supports register ScalarFunction, the Java 
users cannot use {{TableEnvironment}} to register {{TableFunction}}.

This is the test code:

TableFunc0.java

{quote}
public class TableFunc0 extends TableFunction<String> {
        public void eval(String str) {
           for (String s : str.split(" ")) {
             collect(s);
           }
        }
}
{quote}

WordCountExample.java

{quote}
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.table.api.java.StreamTableEnvironment;
import org.apache.flink.table.api.TableEnvironment;
import org.apache.flink.table.api.Types;
import org.apache.flink.table.sinks.CsvTableSink;
import org.apache.flink.table.sources.CsvTableSource;
import org.apache.flink.table.sources.TableSource;
import org.apache.flink.types.Row;

public class WordCountExample {

    public void build(TableEnvironment tableEnv, java.util.Properties 
properties) {
        TableSource<Row> tableSource = new CsvTableSource("/path/to/csv/", new 
String[]{"a", "b"}, new TypeInformation[]{Types.STRING(), Types.STRING()});
        tableEnv.registerTableSource("csv_source", tableSource);

        TableFunc0 tableFunc0 = new TableFunc0();

        tableEnv.registerFunction("func0", tableFunc0); // compile error here

        ((StreamTableEnvironment) tableEnv).registerFunction("func0", 
tableFunc0); // also can work, but not what we want

        tableEnv.scan("csv_source")
                .select("a, b")
                .join("func0(a) as c")
                .select("b, c")
                .writeToSink(new CsvTableSink("/path/to/result", ","));
    }
}
{quote}

There will be compilation errors in tableEnv.registerFunction： "Found 
xx.xxx.TableFunc0，required org.apache.flink.table.functions.ScalarFunction"。

I did some testing, only Java users have this problem.


  was:
Currently {{TableEnvironment}} only supports register ScalarFunction, the Java 
users cannot use {{TableEnvironment}} to register {{TableFunction}}.

This is the test code:

TableFunc0.scala

{quote}
class TableFunc0 extends TableFunction[String] {
  def eval(str: String): Unit = {
    // do somethings...
  }
}
{quote}

UserMainClass.java

{quote}
TableEnvironment tableEnv = ... 

DataStream<Tuple3<Integer, Long, String>> ds = 
StreamTestData.getSmall3TupleDataSet2(env);
Table in = ((StreamTableEnvironment)tableEnv).fromDataStream(ds, "a,b,c");

TableFunc0 tableFunc0 = new TableFunc0();
tableEnv.registerFunction("func0", tableFunc0);   // compile error here
Table result = in.join("func0(c) as d")
                 .select("c, d");
...
{quote}

There will be compilation errors in registerFunction： "Found 
xx.xxx.TableFunc0，required org.apache.flink.table.functions.ScalarFunction"。

I did some testing, only Java users have this problem.



> TableEnvironment support register TableFunction
> -----------------------------------------------
>
>                 Key: FLINK-6355
>                 URL: https://issues.apache.org/jira/browse/FLINK-6355
>             Project: Flink
>          Issue Type: Improvement
>          Components: Table API & SQL
>            Reporter: Kaibo Zhou
>            Assignee: Kaibo Zhou
>
> Currently {{TableEnvironment}} only supports register ScalarFunction, the 
> Java users cannot use {{TableEnvironment}} to register {{TableFunction}}.
> This is the test code:
> TableFunc0.java
> {quote}
> public class TableFunc0 extends TableFunction<String> {
>       public void eval(String str) {
>            for (String s : str.split(" ")) {
>              collect(s);
>            }
>         }
> }
> {quote}
> WordCountExample.java
> {quote}
> import org.apache.flink.api.common.typeinfo.TypeInformation;
> import org.apache.flink.table.api.java.StreamTableEnvironment;
> import org.apache.flink.table.api.TableEnvironment;
> import org.apache.flink.table.api.Types;
> import org.apache.flink.table.sinks.CsvTableSink;
> import org.apache.flink.table.sources.CsvTableSource;
> import org.apache.flink.table.sources.TableSource;
> import org.apache.flink.types.Row;
> public class WordCountExample {
>     public void build(TableEnvironment tableEnv, java.util.Properties 
> properties) {
>         TableSource<Row> tableSource = new CsvTableSource("/path/to/csv/", 
> new String[]{"a", "b"}, new TypeInformation[]{Types.STRING(), 
> Types.STRING()});
>         tableEnv.registerTableSource("csv_source", tableSource);
>         TableFunc0 tableFunc0 = new TableFunc0();
>         tableEnv.registerFunction("func0", tableFunc0); // compile error here
>         ((StreamTableEnvironment) tableEnv).registerFunction("func0", 
> tableFunc0); // also can work, but not what we want
>         tableEnv.scan("csv_source")
>                 .select("a, b")
>                 .join("func0(a) as c")
>                 .select("b, c")
>                 .writeToSink(new CsvTableSink("/path/to/result", ","));
>     }
> }
> {quote}
> There will be compilation errors in tableEnv.registerFunction： "Found 
> xx.xxx.TableFunc0，required org.apache.flink.table.functions.ScalarFunction"。
> I did some testing, only Java users have this problem.



--
This message was sent by Atlassian JIRA
(v6.3.15#6346)