Hi, I think the problem is that the Scala compiler derives a wrong type for this statement:
On Sun, 14 Jun 2015 at 18:28 Shiti Saxena <ssaxena....@gmail.com> wrote: > Hi Aljoscha, > > I created the issue FLINK-2210 > <https://issues.apache.org/jira/browse/FLINK-2210> for aggregate on null. > I made changes to ExpressionAggregateFunction to handle ignore null values. > But I am unable to create a Table with null values in tests. > > The code I used is, > > def testAggregationWithNull(): Unit = { > > val env = ExecutionEnvironment.getExecutionEnvironment > val table = env.fromElements((123, "a"), (234, "b"), (345, "c"), > (null, "d")).toTable > > val total = table.select('_1.sum).collect().head.productElement(0) > assertEquals(total, 702) > } > > and the error i get is, > > org.apache.flink.api.table.ExpressionException: Invalid expression > "('_1).sum": Unsupported type GenericType<java.lang.Object> for aggregation > ('_1).sum. Only numeric data types supported. > at > org.apache.flink.api.table.expressions.analysis.TypeCheck.apply(TypeCheck.scala:50) > at > org.apache.flink.api.table.expressions.analysis.TypeCheck.apply(TypeCheck.scala:31) > at > org.apache.flink.api.table.trees.Analyzer$$anonfun$analyze$1.apply(Analyzer.scala:34) > at > org.apache.flink.api.table.trees.Analyzer$$anonfun$analyze$1.apply(Analyzer.scala:31) > at scala.collection.immutable.List.foreach(List.scala:318) > at org.apache.flink.api.table.trees.Analyzer.analyze(Analyzer.scala:31) > at org.apache.flink.api.table.Table$$anonfun$1.apply(Table.scala:59) > at org.apache.flink.api.table.Table$$anonfun$1.apply(Table.scala:59) > at > scala.collection.TraversableLike$$anonfun$map$1.apply(TraversableLike.scala:244) > at > scala.collection.TraversableLike$$anonfun$map$1.apply(TraversableLike.scala:244) > at > scala.collection.IndexedSeqOptimized$class.foreach(IndexedSeqOptimized.scala:33) > at scala.collection.mutable.WrappedArray.foreach(WrappedArray.scala:34) > at scala.collection.TraversableLike$class.map(TraversableLike.scala:244) > at scala.collection.AbstractTraversable.map(Traversable.scala:105) > at org.apache.flink.api.table.Table.select(Table.scala:59) > at > org.apache.flink.api.scala.table.test.AggregationsITCase.testAggregationWithNull(AggregationsITCase.scala:135) > at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method) > at > sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:57) > at > sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43) > at > org.junit.runners.model.FrameworkMethod$1.runReflectiveCall(FrameworkMethod.java:47) > at > org.junit.internal.runners.model.ReflectiveCallable.run(ReflectiveCallable.java:12) > at > org.junit.runners.model.FrameworkMethod.invokeExplosively(FrameworkMethod.java:44) > at > org.junit.internal.runners.statements.InvokeMethod.evaluate(InvokeMethod.java:17) > at > org.junit.internal.runners.statements.RunBefores.evaluate(RunBefores.java:26) > at > org.junit.internal.runners.statements.RunAfters.evaluate(RunAfters.java:27) > at org.junit.rules.ExternalResource$1.evaluate(ExternalResource.java:48) > at org.junit.rules.RunRules.evaluate(RunRules.java:20) > at org.junit.runners.ParentRunner.runLeaf(ParentRunner.java:271) > at > org.junit.runners.BlockJUnit4ClassRunner.runChild(BlockJUnit4ClassRunner.java:70) > at > org.junit.runners.BlockJUnit4ClassRunner.runChild(BlockJUnit4ClassRunner.java:50) > at org.junit.runners.ParentRunner$3.run(ParentRunner.java:238) > at org.junit.runners.ParentRunner$1.schedule(ParentRunner.java:63) > at org.junit.runners.ParentRunner.runChildren(ParentRunner.java:236) > at org.junit.runners.ParentRunner.access$000(ParentRunner.java:53) > at org.junit.runners.ParentRunner$2.evaluate(ParentRunner.java:229) > at org.junit.runners.ParentRunner.run(ParentRunner.java:309) > at org.junit.runners.Suite.runChild(Suite.java:127) > at org.junit.runners.Suite.runChild(Suite.java:26) > at org.junit.runners.ParentRunner$3.run(ParentRunner.java:238) > at org.junit.runners.ParentRunner$1.schedule(ParentRunner.java:63) > at org.junit.runners.ParentRunner.runChildren(ParentRunner.java:236) > at org.junit.runners.ParentRunner.access$000(ParentRunner.java:53) > at org.junit.runners.ParentRunner$2.evaluate(ParentRunner.java:229) > at > org.junit.internal.runners.statements.RunBefores.evaluate(RunBefores.java:26) > at > org.junit.internal.runners.statements.RunAfters.evaluate(RunAfters.java:27) > at org.junit.runners.ParentRunner.run(ParentRunner.java:309) > at org.junit.runner.JUnitCore.run(JUnitCore.java:160) > at > com.intellij.junit4.JUnit4IdeaTestRunner.startRunnerWithArgs(JUnit4IdeaTestRunner.java:78) > at > com.intellij.rt.execution.junit.JUnitStarter.prepareStreamsAndStart(JUnitStarter.java:212) > at com.intellij.rt.execution.junit.JUnitStarter.main(JUnitStarter.java:68) > at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method) > at > sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:57) > at com.intellij.rt.execution.application.AppMain.main(AppMain.java:140) > > > The ExecutionEnvironment.fromCollection method also throws an error when > the collection contains a null. > > Could you please point out what I am doing wrong? How do we create a Table > with null values? > > In our application, we load a file and transform each line into a Row > resulting in a DataSet[Row]. This DataSet[Row] is then converted into > Table. Should I use the same approach for the test case? > > > Thanks, > Shiti > > > > > > > > > > On Sun, Jun 14, 2015 at 4:10 PM, Shiti Saxena <ssaxena....@gmail.com> > wrote: > >> I'll do the fix >> >> On Sun, Jun 14, 2015 at 12:42 AM, Aljoscha Krettek <aljos...@apache.org> >> wrote: >> >>> I merged your PR for the RowSerializer. Teaching the aggregators to deal >>> with null values should be a very simple fix in >>> ExpressionAggregateFunction.scala. There it is simply always aggregating >>> the values without checking whether they are null. If you want you can also >>> fix that or I can quickly fix it. >>> >>> On Thu, 11 Jun 2015 at 10:40 Aljoscha Krettek <aljos...@apache.org> >>> wrote: >>> >>>> Cool, good to hear. >>>> >>>> The PojoSerializer already handles null fields. The RowSerializer can >>>> be modified in pretty much the same way. So you should start by looking at >>>> the copy()/serialize()/deserialize() methods of PojoSerializer and then >>>> modify RowSerializer in a similar way. >>>> >>>> You can also send me a private mail if you want more in-depth >>>> explanations. >>>> >>>> On Thu, 11 Jun 2015 at 09:33 Till Rohrmann <trohrm...@apache.org> >>>> wrote: >>>> >>>>> Hi Shiti, >>>>> >>>>> here is the issue [1]. >>>>> >>>>> Cheers, >>>>> Till >>>>> >>>>> [1] https://issues.apache.org/jira/browse/FLINK-2203 >>>>> >>>>> On Thu, Jun 11, 2015 at 8:42 AM Shiti Saxena <ssaxena....@gmail.com> >>>>> wrote: >>>>> >>>>>> Hi Aljoscha, >>>>>> >>>>>> Could you please point me to the JIRA tickets? If you could provide >>>>>> some guidance on how to resolve these, I will work on them and raise a >>>>>> pull-request. >>>>>> >>>>>> Thanks, >>>>>> Shiti >>>>>> >>>>>> On Thu, Jun 11, 2015 at 11:31 AM, Aljoscha Krettek < >>>>>> aljos...@apache.org> wrote: >>>>>> >>>>>>> Hi, >>>>>>> yes, I think the problem is that the RowSerializer does not support >>>>>>> null-values. I think we can add support for this, I will open a Jira >>>>>>> issue. >>>>>>> >>>>>>> Another problem I then see is that the aggregations can not properly >>>>>>> deal with null-values. This would need separate support. >>>>>>> >>>>>>> Regards, >>>>>>> Aljoscha >>>>>>> >>>>>>> On Thu, 11 Jun 2015 at 06:41 Shiti Saxena <ssaxena....@gmail.com> >>>>>>> wrote: >>>>>>> >>>>>>>> Hi, >>>>>>>> >>>>>>>> In our project, we are using the Flink Table API and are facing the >>>>>>>> following issues, >>>>>>>> >>>>>>>> We load data from a CSV file and create a DataSet[Row]. The CSV >>>>>>>> file can also have invalid entries in some of the fields which we >>>>>>>> replace >>>>>>>> with null when building the DataSet[Row]. >>>>>>>> >>>>>>>> This DataSet[Row] is later on transformed to Table whenever >>>>>>>> required and specific operation such as select or aggregate, etc are >>>>>>>> performed. >>>>>>>> >>>>>>>> When a null value is encountered, we get a null pointer exception >>>>>>>> and the whole job fails. (We can see this by calling collect on the >>>>>>>> resulting DataSet). >>>>>>>> >>>>>>>> The error message is similar to, >>>>>>>> >>>>>>>> Job execution failed. >>>>>>>> org.apache.flink.runtime.client.JobExecutionException: Job >>>>>>>> execution failed. >>>>>>>> at >>>>>>>> org.apache.flink.runtime.jobmanager.JobManager$$anonfun$receiveWithLogMessages$1.applyOrElse(JobManager.scala:315) >>>>>>>> at >>>>>>>> scala.runtime.AbstractPartialFunction$mcVL$sp.apply$mcVL$sp(AbstractPartialFunction.scala:33) >>>>>>>> at >>>>>>>> scala.runtime.AbstractPartialFunction$mcVL$sp.apply(AbstractPartialFunction.scala:33) >>>>>>>> at >>>>>>>> scala.runtime.AbstractPartialFunction$mcVL$sp.apply(AbstractPartialFunction.scala:25) >>>>>>>> at >>>>>>>> org.apache.flink.runtime.ActorLogMessages$$anon$1.apply(ActorLogMessages.scala:43) >>>>>>>> at >>>>>>>> org.apache.flink.runtime.ActorLogMessages$$anon$1.apply(ActorLogMessages.scala:29) >>>>>>>> at >>>>>>>> scala.PartialFunction$class.applyOrElse(PartialFunction.scala:118) >>>>>>>> at >>>>>>>> org.apache.flink.runtime.ActorLogMessages$$anon$1.applyOrElse(ActorLogMessages.scala:29) >>>>>>>> at akka.actor.Actor$class.aroundReceive(Actor.scala:465) >>>>>>>> at >>>>>>>> org.apache.flink.runtime.jobmanager.JobManager.aroundReceive(JobManager.scala:94) >>>>>>>> at akka.actor.ActorCell.receiveMessage(ActorCell.scala:516) >>>>>>>> at akka.actor.ActorCell.invoke(ActorCell.scala:487) >>>>>>>> at akka.dispatch.Mailbox.processMailbox(Mailbox.scala:254) >>>>>>>> at akka.dispatch.Mailbox.run(Mailbox.scala:221) >>>>>>>> at akka.dispatch.Mailbox.exec(Mailbox.scala:231) >>>>>>>> at >>>>>>>> scala.concurrent.forkjoin.ForkJoinTask.doExec(ForkJoinTask.java:260) >>>>>>>> at >>>>>>>> scala.concurrent.forkjoin.ForkJoinPool$WorkQueue.runTask(ForkJoinPool.java:1339) >>>>>>>> at >>>>>>>> scala.concurrent.forkjoin.ForkJoinPool.runWorker(ForkJoinPool.java:1979) >>>>>>>> at >>>>>>>> scala.concurrent.forkjoin.ForkJoinWorkerThread.run(ForkJoinWorkerThread.java:107) >>>>>>>> Caused by: java.lang.NullPointerException >>>>>>>> at >>>>>>>> org.apache.flink.api.common.typeutils.base.IntSerializer.serialize(IntSerializer.java:63) >>>>>>>> at >>>>>>>> org.apache.flink.api.common.typeutils.base.IntSerializer.serialize(IntSerializer.java:27) >>>>>>>> at >>>>>>>> org.apache.flink.api.table.typeinfo.RowSerializer.serialize(RowSerializer.scala:80) >>>>>>>> at >>>>>>>> org.apache.flink.api.table.typeinfo.RowSerializer.serialize(RowSerializer.scala:28) >>>>>>>> at >>>>>>>> org.apache.flink.runtime.plugable.SerializationDelegate.write(SerializationDelegate.java:51) >>>>>>>> at org.apache.flink.runtime.io >>>>>>>> .network.api.serialization.SpanningRecordSerializer.addRecord(SpanningRecordSerializer.java:76) >>>>>>>> at org.apache.flink.runtime.io >>>>>>>> .network.api.writer.RecordWriter.emit(RecordWriter.java:83) >>>>>>>> at >>>>>>>> org.apache.flink.runtime.operators.shipping.OutputCollector.collect(OutputCollector.java:65) >>>>>>>> at >>>>>>>> org.apache.flink.runtime.operators.chaining.ChainedMapDriver.collect(ChainedMapDriver.java:78) >>>>>>>> at >>>>>>>> org.apache.flink.runtime.operators.chaining.ChainedMapDriver.collect(ChainedMapDriver.java:78) >>>>>>>> at >>>>>>>> org.apache.flink.runtime.operators.DataSourceTask.invoke(DataSourceTask.java:177) >>>>>>>> at org.apache.flink.runtime.taskmanager.Task.run(Task.java:559) >>>>>>>> at java.lang.Thread.run(Thread.java:724) >>>>>>>> >>>>>>>> Could this be because the RowSerializer does not support null >>>>>>>> values? (Similar to Flink-629 >>>>>>>> <https://issues.apache.org/jira/browse/FLINK-629> ) >>>>>>>> >>>>>>>> Currently, to overcome this issue, we are ignoring all the rows >>>>>>>> which may have null values. For example, we have a method cleanData >>>>>>>> defined >>>>>>>> as, >>>>>>>> >>>>>>>> def cleanData(table:Table, relevantColumns:Seq[String]):Table = { >>>>>>>> val whereClause: String = relevantColumns.map{ >>>>>>>> cName=> >>>>>>>> s"$cName.isNotNull" >>>>>>>> }.mkString(" && ") >>>>>>>> >>>>>>>> val result :Table = >>>>>>>> table.select(relevantColumns.mkString(",")).where(whereClause) >>>>>>>> result >>>>>>>> } >>>>>>>> >>>>>>>> Before operating on any Table, we use this method and then continue >>>>>>>> with task. >>>>>>>> >>>>>>>> Is this the right way to handle this? If not please let me know how >>>>>>>> to go about it. >>>>>>>> >>>>>>>> >>>>>>>> Thanks, >>>>>>>> Shiti >>>>>>>> >>>>>>>> >>>>>>>> >>>>>>>> >>>>>> >> >
