[SPARK-28371] Parquet "starts with" filter is not null-safe - ASF JIRA

Attach files

Attach Screenshot

Voters

Watch issue

Watchers

Create sub-task

Link

Clone

Update Comment Author

Replace String in Comment

Update Comment Visibility

Delete Comments

XML

Word

Printable

JSON

Details

Type: Bug
Status: Resolved
Priority: Major
Resolution: Fixed
Affects Version/s: 3.0.0
Fix Version/s: 2.4.4, 3.0.0
Component/s: SQL
Labels:
None

Description

I ran into this when running unit tests with Parquet 1.11. It seems that 1.10 has the same behavior in a few places but Spark somehow doesn't trigger those code paths.

Basically, UserDefinedPredicate.keep should be null-safe, and Spark's implementation is not. This was clarified in Parquet's documentation in ~~PARQUET-1489~~.

Failure I was getting:

Job aborted due to stage failure: Task 0 in stage 1304.0 failed 1 times, most recent failure: Lost task 0.0 in stage 1304.0 (TID 2528, localhost, executor driver): java.lang.NullPointerException&#010;
  at org.apache.spark.sql.execution.datasources.parquet.ParquetFilters$$anonfun$createFilter$16$$anon$1.keep(ParquetFilters.scala:544)&#010;
  at org.apache.spark.sql.execution.datasources.parquet.ParquetFilters$$anonfun$createFilter$16$$anon$1.keep(ParquetFilters.scala:523)&#010;
  at org.apache.parquet.internal.filter2.columnindex.ColumnIndexFilter.visit(ColumnIndexFilter.java:152)&#010;
  at org.apache.parquet.internal.filter2.columnindex.ColumnIndexFilter.visit(ColumnIndexFilter.java:56)&#010;
  at org.apache.parquet.filter2.predicate.Operators$UserDefined.accept(Operators.java:377)&#010;
  at org.apache.parquet.internal.filter2.columnindex.ColumnIndexFilter.visit(ColumnIndexFilter.java:181)&#010;
  at org.apache.parquet.internal.filter2.columnindex.ColumnIndexFilter.visit(ColumnIndexFilter.java:56)&#010;
  at org.apache.parquet.filter2.predicate.Operators$And.accept(Operators.java:309)&#010;
  at org.apache.parquet.internal.filter2.columnindex.ColumnIndexFilter$1.visit(ColumnIndexFilter.java:86)&#010;
  at org.apache.parquet.internal.filter2.columnindex.ColumnIndexFilter$1.visit(ColumnIndexFilter.java:81)&#010;
  at org.apache.parquet.filter2.compat.FilterCompat$FilterPredicateCompat.accept(FilterCompat.java:137)&#010;
  at org.apache.parquet.internal.filter2.columnindex.ColumnIndexFilter.calculateRowRanges(ColumnIndexFilter.java:81)&#010;
  at org.apache.parquet.hadoop.ParquetFileReader.getRowRanges(ParquetFileReader.java:954)&#010;
  at org.apache.parquet.hadoop.ParquetFileReader.getFilteredRecordCount(ParquetFileReader.java:759)&#010;
  at org.apache.parquet.hadoop.InternalParquetRecordReader.initialize(InternalParquetRecordReader.java:207)&#010;
  at org.apache.parquet.hadoop.ParquetRecordReader.initializeInternalReader(ParquetRecordReader.java:182)&#010;
  at org.apache.parquet.hadoop.ParquetRecordReader.initialize(ParquetRecordReader.java:140)&#010;
  at org.apache.spark.sql.execution.datasources.parquet.ParquetFileFormat$$anonfun$buildReaderWithPartitionValues$1.apply(ParquetFileFormat.scala:439)&#010;
  ...