UserDefinedPredicate throw NPE

[asfimport]

It throws NullPointerException after upgrade parquet to 1.11.0 when using UserDefinedPredicate.

The  UserDefinedPredicate is:

new UserDefinedPredicate[Binary] with Serializable {                                  
  private val strToBinary = Binary.fromReusedByteArray(v.getBytes)                    
  private val size = strToBinary.length                                               
                                                                                      
  override def canDrop(statistics: Statistics[Binary]): Boolean = {                   
    val comparator = PrimitiveComparator.UNSIGNED_LEXICOGRAPHICAL_BINARY_COMPARATOR   
    val max = statistics.getMax                                                       
    val min = statistics.getMin                                                       
    comparator.compare(max.slice(0, math.min(size, max.length)), strToBinary) < 0 ||  
      comparator.compare(min.slice(0, math.min(size, min.length)), strToBinary) > 0   
  }                                                                                   
                                                                                      
  override def inverseCanDrop(statistics: Statistics[Binary]): Boolean = {            
    val comparator = PrimitiveComparator.UNSIGNED_LEXICOGRAPHICAL_BINARY_COMPARATOR   
    val max = statistics.getMax                                                       
    val min = statistics.getMin                                                       
    comparator.compare(max.slice(0, math.min(size, max.length)), strToBinary) == 0 && 
      comparator.compare(min.slice(0, math.min(size, min.length)), strToBinary) == 0  
  }                                                                                   
                                                                                      
  override def keep(value: Binary): Boolean = {                                       
    UTF8String.fromBytes(value.getBytes).startsWith(                                  
      UTF8String.fromBytes(strToBinary.getBytes))                                     
  }                                                                                   
}                                                                                     

The stack trace is:

java.lang.NullPointerException
	at org.apache.spark.sql.execution.datasources.parquet.ParquetFilters$$anon$1.keep(ParquetFilters.scala:573)
	at org.apache.spark.sql.execution.datasources.parquet.ParquetFilters$$anon$1.keep(ParquetFilters.scala:552)
	at org.apache.parquet.internal.filter2.columnindex.ColumnIndexFilter.visit(ColumnIndexFilter.java:152)
	at org.apache.parquet.internal.filter2.columnindex.ColumnIndexFilter.visit(ColumnIndexFilter.java:56)
	at org.apache.parquet.filter2.predicate.Operators$UserDefined.accept(Operators.java:377)
	at org.apache.parquet.internal.filter2.columnindex.ColumnIndexFilter.visit(ColumnIndexFilter.java:181)
	at org.apache.parquet.internal.filter2.columnindex.ColumnIndexFilter.visit(ColumnIndexFilter.java:56)
	at org.apache.parquet.filter2.predicate.Operators$And.accept(Operators.java:309)
	at org.apache.parquet.internal.filter2.columnindex.ColumnIndexFilter$1.visit(ColumnIndexFilter.java:86)
	at org.apache.parquet.internal.filter2.columnindex.ColumnIndexFilter$1.visit(ColumnIndexFilter.java:81)

Reporter: Yuming Wang / @wangyum
Assignee: Gabor Szadovszky / @gszadovszky

Related issues:

• Release parquet-mr 1.11.0 (blocks)
• Insufficient documentation for UserDefinedPredicate.keep(T) (is related to)

PRs and other links:

• GitHub Pull Request #663

_{Note: This issue was originally created as PARQUET-1488. Please see the migration documentation for further details.}

[asfimport]

Yuming Wang / @wangyum:
cc @gszadovszky

[asfimport]

Gabor Szadovszky / @gszadovszky:
@wangyum,

UserDefinedPredicate.keep(T) shall handle null values because the parquet-mr library do invokes it with a null. This is the only way the filtering API can check if a null value fulfills the UDP.
It did not occur before the ColumnIndex feature because the statistics and the dictionary filters invokes keep(T) only in some special scenarios (e.g. if all the values are null in the column chunk).
So the problem is at the client side. Please, update your code so that keep(T) checks for null value and returns accordingly.

It is another issue that the described behavior is not properly documented. I'll create a JIRA about it.

Closing this issue as not a bug.

[asfimport]

Yuming Wang / @wangyum:
@gszadovszky Thanks a lot.

[asfimport]

Ryan Blue / @rdblue:
We discussed this on SPARK-28371.

Previously, Parquet did not fail if a UserDefinedPredicate did not handle null values, so I think that it is a regression that Parquet will cause previously working code to fail. I think that it is correct for Parquet to call a UDP the way that it is, but that Parquet should catch exceptions thrown by the predicate and should process the row group where there error was thrown. That way, Parquet can keep the optimization for columns that are all null, but it doesn't break existing code.

@wangyum, would you like to submit a PR for this?

[asfimport]

Gabor Szadovszky / @gszadovszky:
@rdblue,
Previously, parquet might fail in special cases for both dictionary and statistics filters. However, because of performance reasons, column indexes invokes it all the time. I agree, it is a regression and my previous "fix" of having it properly documented was not enough.

@wangyum,
I'm happy to help you with this or work on it if you don't have the time.

[asfimport]

Yuming Wang / @wangyum:
Thank you @gszadovszky Please work on this.