flink-issues mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From "ASF GitHub Bot (JIRA)" <j...@apache.org>
Subject [jira] [Commented] (FLINK-6888) Can not determine TypeInformation of ACC type of AggregateFunction when ACC is a Scala case/tuple class
Date Mon, 12 Jun 2017 08:17:00 GMT

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

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

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

    https://github.com/apache/flink/pull/4105#discussion_r121325300
  
    --- Diff: flink-libraries/flink-table/src/test/scala/org/apache/flink/table/api/scala/stream/sql/AggregationsTest.scala
---
    @@ -39,4 +44,34 @@ class AggregationsTest extends TableTestBase {
     
         streamUtil.tEnv.sql(sqlQuery)
       }
    +
    +  @Test
    +  def testUserDefinedAggregateFunctionWithScalaAccumulator(): Unit = {
    +    streamUtil.addFunction("udag", new MyAgg)
    +    val call = streamUtil
    +      .tEnv
    +      .functionCatalog
    +      .lookupFunction("udag", Seq())
    +      .asInstanceOf[AggFunctionCall]
    +
    +    val typeInfo = call.accTypeInfo
    +    assertTrue(typeInfo.isInstanceOf[CaseClassTypeInfo[_]])
    +    assertEquals(2, typeInfo.getTotalFields)
    +    val caseTypeInfo = typeInfo.asInstanceOf[CaseClassTypeInfo[_]]
    +    assertEquals(Types.LONG, caseTypeInfo.getTypeAt(0))
    +    assertEquals(Types.LONG, caseTypeInfo.getTypeAt(1))
    +  }
    +}
    +
    +case class Accumulator(sum: Long, count: Long)
    --- End diff --
    
    The members of `Accumulator` must be modified. 


> 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)

Mime
View raw message