[ 
https://issues.apache.org/jira/browse/FLINK-6888?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16067649#comment-16067649
 ] 

ASF GitHub Bot commented on FLINK-6888:
---------------------------------------

Github user wuchong commented on a diff in the pull request:

    https://github.com/apache/flink/pull/4105#discussion_r124704209
  
    --- Diff: 
flink-libraries/flink-table/src/main/scala/org/apache/flink/table/functions/utils/UserDefinedFunctionUtils.scala
 ---
    @@ -320,12 +348,20 @@ object UserDefinedFunctionUtils {
         } else if(extractedType != null) {
           extractedType
         } else {
    -      TypeExtractor
    +      try {
    +        TypeExtractor
    --- End diff --
    
    I think it is more safe than `TypeInformation.of(Any)`, because 
`TypeInformation.of(Any)` can only works for non-generic types. 
    
    For example: 
    
    ```
    public class UDAF extends AggregateFunction[Tuple2<String, Long>, Long] {
        public Long createAccumulator() {...}
        public Tuple2<String, Long> getValue(Long acc) {...}
    }
    ```
    For the given UDAF, the return type is a generic type `Tuple2<String, 
Long>`. We can't extract the return type  from an instance of Tuple2, because 
of Java type erasure. But we can extract it from the type hierarchy of UDAF 
class. 
    
    Am I right, @twalthr ?
    
    
    
    ```
    
    
    



> Can not determine TypeInformation of ACC type of AggregateFunction when ACC 
> is a Scala case/tuple class
> -------------------------------------------------------------------------------------------------------
>
>                 Key: FLINK-6888
>                 URL: https://issues.apache.org/jira/browse/FLINK-6888
>             Project: Flink
>          Issue Type: Bug
>          Components: Table API & SQL
>            Reporter: Jark Wu
>            Assignee: Jark Wu
>             Fix For: 1.4.0
>
>
> Currently the {{ACC}} TypeInformation of 
> {{org.apache.flink.table.functions.AggregateFunction[T, ACC]}} is extracted 
> using {{TypeInformation.of(Class)}}. When {{ACC}} is a Scala case class or 
> tuple class, the TypeInformation will fall back to {{GenericType}} which 
> result in bad performance when state de/serialization. 
> I suggest to extract the ACC TypeInformation when called 
> {{TableEnvironment.registerFunction()}}.
> Here is an example:
> {code}
> case class Accumulator(sum: Long, count: Long)
> class MyAgg extends AggregateFunction[Long, Accumulator] {
>   //Overloaded accumulate method
>   def accumulate(acc: Accumulator, value: Long): Unit = {
>   }
>   override def createAccumulator(): Accumulator = Accumulator(0, 0)
>   override def getValue(accumulator: Accumulator): Long = 1
> }
> {code}
> The {{Accumulator}} will be recognized as {{GenericType<Accumulator>}}.



--
This message was sent by Atlassian JIRA
(v6.4.14#64029)

Reply via email to