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.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at [email protected] or file a JIRA ticket
with INFRA.
---
