[
https://issues.apache.org/jira/browse/FLINK-5571?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=15831514#comment-15831514
]
ASF GitHub Bot commented on FLINK-5571:
---------------------------------------
Github user wuchong commented on a diff in the pull request:
https://github.com/apache/flink/pull/3176#discussion_r97033852
--- Diff:
flink-libraries/flink-table/src/main/scala/org/apache/flink/table/codegen/CodeGenerator.scala
---
@@ -220,56 +247,105 @@ class CodeGenerator(
// manual casting here
val samHeader =
// FlatMapFunction
- if (clazz == classOf[FlatMapFunction[_,_]]) {
+ if (clazz == classOf[FlatMapFunction[_, _]]) {
+ val baseClass = if (generatedRichFunctions) {
+ classOf[RichFlatMapFunction[_, _]]
+ } else {
+ classOf[FlatMapFunction[_, _]]
+ }
val inputTypeTerm = boxedTypeTermForTypeInfo(input1)
- (s"void flatMap(Object _in1, org.apache.flink.util.Collector
$collectorTerm)",
+ (baseClass,
+ s"void flatMap(Object _in1, org.apache.flink.util.Collector
$collectorTerm)",
List(s"$inputTypeTerm $input1Term = ($inputTypeTerm) _in1;"))
}
// MapFunction
- else if (clazz == classOf[MapFunction[_,_]]) {
+ else if (clazz == classOf[MapFunction[_, _]]) {
+ val baseClass = if (generatedRichFunctions) {
+ classOf[RichMapFunction[_, _]]
+ } else {
+ classOf[MapFunction[_, _]]
+ }
val inputTypeTerm = boxedTypeTermForTypeInfo(input1)
- ("Object map(Object _in1)",
+ (baseClass,
+ "Object map(Object _in1)",
List(s"$inputTypeTerm $input1Term = ($inputTypeTerm) _in1;"))
}
// FlatJoinFunction
- else if (clazz == classOf[FlatJoinFunction[_,_,_]]) {
+ else if (clazz == classOf[FlatJoinFunction[_, _, _]]) {
+ val baseClass = if (generatedRichFunctions) {
+ classOf[RichFlatJoinFunction[_, _, _]]
+ } else {
+ classOf[FlatJoinFunction[_, _, _]]
+ }
val inputTypeTerm1 = boxedTypeTermForTypeInfo(input1)
val inputTypeTerm2 = boxedTypeTermForTypeInfo(input2.getOrElse(
- throw new CodeGenException("Input 2 for FlatJoinFunction
should not be null")))
- (s"void join(Object _in1, Object _in2,
org.apache.flink.util.Collector $collectorTerm)",
+ throw new CodeGenException("Input 2 for FlatJoinFunction should
not be null")))
+ (baseClass,
+ s"void join(Object _in1, Object _in2,
org.apache.flink.util.Collector $collectorTerm)",
List(s"$inputTypeTerm1 $input1Term = ($inputTypeTerm1) _in1;",
- s"$inputTypeTerm2 $input2Term = ($inputTypeTerm2) _in2;"))
+ s"$inputTypeTerm2 $input2Term = ($inputTypeTerm2) _in2;"))
}
else {
// TODO more functions
throw new CodeGenException("Unsupported Function.")
}
- val funcCode = j"""
- public class $funcName
- implements ${clazz.getCanonicalName} {
+ val funcCode = if (generatedRichFunctions) {
--- End diff --
There is a lot of duplicate code between RichFunction codegen and
non-RichFunction codegen. The only difference between them is the open close
method code, so I think it would be better to **insert** open close code when
it is a RichFunction.
> add open and close methods for UserDefinedFunction in TableAPI & SQL
> --------------------------------------------------------------------
>
> Key: FLINK-5571
> URL: https://issues.apache.org/jira/browse/FLINK-5571
> Project: Flink
> Issue Type: New Feature
> Components: Table API & SQL
> Reporter: godfrey he
> Assignee: godfrey he
>
> Currently, a User Defined Function (UDF) in table API & SQL works on zero,
> one, or multiple values in custom evaluation method. Many UDFs need more
> complex features, e.g. report metrics, get parameters from job configuration,
> or get extra data from distribute cache file, etc. Adding open and close
> methods in UserDefinedFunction class can solve this problem. The code cloud
> look like:
> {code}
> trait UserDefinedFunction {
> def open(context: UDFContext): Unit = {}
> def close(): Unit = {}
> }
> {code}
> UDFContext contains the information about metric reporters, job parameters,
> distribute cache, etc. The code cloud look like:
> {code}
> class UDFContext(context: RuntimeContext) {
> def getMetricGroup: MetricGroup = ???
> def getDistributedCacheFile(name: String): File = ???
> def getJobParameter(key: String, default: String): String = ???
> }
> {code}
--
This message was sent by Atlassian JIRA
(v6.3.4#6332)
