PARQUET-1510: Fix notEq for optional columns with null values. (#603)

Dictionaries cannot contain null values, so notEq filters cannot
conclude that a block cannot match using only the dictionary. Instead,
it must also check whether the block may have at least one null value.
If there are no null values, then the existing check is correct.

Author: rdblue

parquet-hadoop/src/main/java/org/apache/parquet/filter2/dictionarylevel/DictionaryFilter.java

@@ -178,7 +178,10 @@ public <T extends Comparable<T>> Boolean visit(NotEq<T> notEq) {
 
     try {
       Set<T> dictSet = expandDictionary(meta);
-      if (dictSet != null && dictSet.size() == 1 && dictSet.contains(value)) {
+      boolean mayContainNull = (meta.getStatistics() == null
+          || !meta.getStatistics().isNumNullsSet()
+          || meta.getStatistics().getNumNulls() > 0);
+      if (dictSet != null && dictSet.size() == 1 && dictSet.contains(value) && !mayContainNull) {
         return BLOCK_CANNOT_MATCH;
       }
     } catch (IOException e) {

parquet-hadoop/src/test/java/org/apache/parquet/filter2/dictionarylevel/DictionaryFilterTest.java

@@ -79,6 +79,7 @@ public class DictionaryFilterTest {
       "message test { "
           + "required binary binary_field; "
           + "required binary single_value_field; "
+          + "optional binary optional_single_value_field; "
           + "required int32 int32_field; "
           + "required int64 int64_field; "
           + "required double double_field; "
@@ -112,6 +113,11 @@ private static void writeData(SimpleGroupFactory f, ParquetWriter<Group> writer)
           .append("fallback_binary_field", i < (nElements / 2) ?
               ALPHABET.substring(index, index+1) : UUID.randomUUID().toString());
 
+      // 10% of the time, leave the field null
+      if (index % 10 > 0) {
+        group.append("optional_single_value_field", "sharp");
+      }
+
       writer.write(group);
     }
     writer.close();
@@ -165,7 +171,7 @@ public void tearDown() throws Exception {
   @SuppressWarnings("deprecation")
   public void testDictionaryEncodedColumns() throws Exception {
     Set<String> dictionaryEncodedColumns = new HashSet<String>(Arrays.asList(
-        "binary_field", "single_value_field", "int32_field", "int64_field",
+        "binary_field", "single_value_field", "optional_single_value_field", "int32_field", "int64_field",
         "double_field", "float_field"));
     for (ColumnChunkMetaData column : ccmd) {
       String name = column.getPath().toDotString();
@@ -208,6 +214,7 @@ public void testEqBinary() throws Exception {
   @Test
   public void testNotEqBinary() throws Exception {
     BinaryColumn sharp = binaryColumn("single_value_field");
+    BinaryColumn sharpAndNull = binaryColumn("optional_single_value_field");
     BinaryColumn b = binaryColumn("binary_field");
 
     assertTrue("Should drop block with only the excluded value",
@@ -216,6 +223,12 @@ public void testNotEqBinary() throws Exception {
     assertFalse("Should not drop block with any other value",
         canDrop(notEq(sharp, Binary.fromString("applause")), ccmd, dictionaries));
 
+    assertFalse("Should not drop block with only the excluded value and null",
+        canDrop(notEq(sharpAndNull, Binary.fromString("sharp")), ccmd, dictionaries));
+
+    assertFalse("Should not drop block with any other value",
+        canDrop(notEq(sharpAndNull, Binary.fromString("applause")), ccmd, dictionaries));
+
     assertFalse("Should not drop block with a known value",
         canDrop(notEq(b, Binary.fromString("x")), ccmd, dictionaries));