[jira] [Updated] (FLINK-14642) Flink TupleSerializer and CaseClassSerializer shoud support copy NULL values

["Victor Wong (Jira)" <ji...@apache.org>]

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

Victor Wong updated FLINK-14642:
--------------------------------
    Description: 
Currently, TupleSerializer and CaseCassSerializer do not support serialize NULL values, which I think is acceptable. But not supporting copy NULL values will cause the following codes to throw an exception, which I think is not matched with users' expectations and prone to error.

*codes:*
{code:java}
stream.map(xxx).filter(_ != null).xxx //the return type of the map function is Tuple and it may return null{code}
 

*exception info:*

 
{code:java}
Caused by: java.lang.NullPointerException 
  at org.apache.flink.api.scala.typeutils.CaseClassSerializer.copy(CaseClassSerializer.scala:92) 
  at org.apache.flink.api.scala.typeutils.CaseClassSerializer.copy(CaseClassSerializer.scala:32) 
  at org.apache.flink.streaming.runtime.tasks.OperatorChain$CopyingChainingOutput.pushToOperator(OperatorChain.java:635)
{code}
 

*suggestion:*

Can we make the `copy` method of TupleSerializer/CaseClassSerializer to handle NULL values? e.g.
{code:java}
// org.apache.flink.api.scala.typeutils.CaseClassSerializer#copy
def copy(from: T): T = {
  // handle NULL values.
  if(from == null) {
    return from
  }
  initArray()
  var i = 0
  while (i < arity) {
    fields(i) = fieldSerializers(i).copy(from.productElement(i).asInstanceOf[AnyRef])
    i += 1
  }
  createInstance(fields)
}
{code}
 

 

 

 

 

  was:
Currently, TupleSerializer and CaseCassSerializer do not support serialize NULL values, which I think is acceptable. But not supporting to copy NULL values will cause the following codes to throw an exception, which I think is not matched with users' expectations.

*codes:*

 
{code:java}
stream.map(xxx).filter(_ != null).xxx //the return type of the map function is Tuple and it may return null{code}
 

*exception info:*

 
{code:java}
Caused by: java.lang.NullPointerExceptionCaused by: java.lang.NullPointerException at org.apache.flink.api.scala.typeutils.CaseClassSerializer.copy(CaseClassSerializer.scala:92) at org.apache.flink.api.scala.typeutils.CaseClassSerializer.copy(CaseClassSerializer.scala:32) at org.apache.flink.streaming.runtime.tasks.OperatorChain$CopyingChainingOutput.pushToOperator(OperatorChain.java:635)
{code}
 

*suggestion:*

Can we make the `copy` method of TupleSerializer/CaseClassSerializer to handle NULL values? e.g.
{code:java}
// org.apache.flink.api.scala.typeutils.CaseClassSerializer#copy
def copy(from: T): T = {
  // handle NULL values.
  if(from == null) {
    return from
  }
  initArray()
  var i = 0
  while (i < arity) {
    fields(i) = fieldSerializers(i).copy(from.productElement(i).asInstanceOf[AnyRef])
    i += 1
  }
  createInstance(fields)
}
{code}
 

 

 

 

 


> Flink TupleSerializer and CaseClassSerializer shoud support copy NULL values
> ----------------------------------------------------------------------------
>
>                 Key: FLINK-14642
>                 URL: https://issues.apache.org/jira/browse/FLINK-14642
>             Project: Flink
>          Issue Type: Bug
>          Components: API / Type Serialization System
>    Affects Versions: 1.9.1
>            Reporter: Victor Wong
>            Priority: Major
>
> Currently, TupleSerializer and CaseCassSerializer do not support serialize NULL values, which I think is acceptable. But not supporting copy NULL values will cause the following codes to throw an exception, which I think is not matched with users' expectations and prone to error.
> *codes:*
> {code:java}
> stream.map(xxx).filter(_ != null).xxx //the return type of the map function is Tuple and it may return null{code}
>  
> *exception info:*
>  
> {code:java}
> Caused by: java.lang.NullPointerException 
>   at org.apache.flink.api.scala.typeutils.CaseClassSerializer.copy(CaseClassSerializer.scala:92) 
>   at org.apache.flink.api.scala.typeutils.CaseClassSerializer.copy(CaseClassSerializer.scala:32) 
>   at org.apache.flink.streaming.runtime.tasks.OperatorChain$CopyingChainingOutput.pushToOperator(OperatorChain.java:635)
> {code}
>  
> *suggestion:*
> Can we make the `copy` method of TupleSerializer/CaseClassSerializer to handle NULL values? e.g.
> {code:java}
> // org.apache.flink.api.scala.typeutils.CaseClassSerializer#copy
> def copy(from: T): T = {
>   // handle NULL values.
>   if(from == null) {
>     return from
>   }
>   initArray()
>   var i = 0
>   while (i < arity) {
>     fields(i) = fieldSerializers(i).copy(from.productElement(i).asInstanceOf[AnyRef])
>     i += 1
>   }
>   createInstance(fields)
> }
> {code}
>  
>  
>  
>  
>  



--
This message was sent by Atlassian Jira
(v8.3.4#803005)