Mailing-List: contact issues-help@flink.apache.org; run by ezmlm
Precedence: bulk
Reply-To: dev@flink.apache.org
Date: Sun, 11 Jun 2017 11:54:18 +0000 (UTC)
From: "Jark Wu (JIRA)" <jira@apache.org>
To: issues@flink.apache.org
Message-ID: <JIRA.13078938.1497181441000.48134.1497182058250@Atlassian.JIRA>
In-Reply-To: <JIRA.13078938.1497181441000@Atlassian.JIRA>
References: <JIRA.13078938.1497181441000@Atlassian.JIRA> <JIRA.13078938.1497181441450@jira-lw-us.apache.org>
Subject: [jira] [Updated] (FLINK-6888) Can not determine TypeInformation of
 ACC type of AggregateFunction when ACC is a Scala case/tuple class
MIME-Version: 1.0
Content-Type: text/plain; charset=utf-8
Content-Transfer-Encoding: 7bit
archived-at: Sun, 11 Jun 2017 11:54:24 -0000


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

Jark Wu updated FLINK-6888:
---------------------------
    Description: 
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>}}.

  was:
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()}}.


> 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.3.15#6346)