Parquet dictionary filter

parquet-column/src/main/java/org/apache/parquet/column/page/DictionaryPageReadStore.java

@@ -0,0 +1,34 @@
+/* 
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ * 
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ * 
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.parquet.column.page;
+
+import org.apache.parquet.column.ColumnDescriptor;
+
+/**
+ * contains all the dictionary readers for all the columns of the corresponding row group
+ */
+public interface DictionaryPageReadStore {
+
+  /**
+   *
+   * @param descriptor the descriptor of the column
+   * @return the dictionary page reader for that column
+   */
+  DictionaryPageReader getDictionaryPageReader(ColumnDescriptor descriptor);
+}

parquet-column/src/main/java/org/apache/parquet/column/page/DictionaryPageReader.java

@@ -0,0 +1,35 @@
+/* 
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ * 
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ * 
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.parquet.column.page;
+
+/**
+ * Reader for a dictionary page from a given column chunk
+ */
+public interface DictionaryPageReader {
+
+ /**
+  * @return the dictionary page in that chunk or null if none
+  */
+  DictionaryPage readDictionaryPage();
+
+  /**
+   * @return the dictionary size
+   */
+  int getDictionarySize();
+}

parquet-hadoop/pom.xml

@@ -77,7 +77,6 @@
       <groupId>com.google.guava</groupId>
       <artifactId>guava</artifactId>
       <version>11.0</version>
-      <scope>test</scope>
     </dependency>
     <dependency>
       <groupId>org.xerial.snappy</groupId>

parquet-hadoop/src/main/java/org/apache/parquet/filter2/compat/RowGroupFilter.java

@@ -19,11 +19,16 @@
 package org.apache.parquet.filter2.compat;
 
 import java.util.ArrayList;
+import java.util.Collections;
 import java.util.List;
 
+import com.google.common.base.Supplier;
+import org.apache.hadoop.fs.FSDataInputStream;
+import org.apache.hadoop.io.IOUtils;
 import org.apache.parquet.filter2.compat.FilterCompat.Filter;
 import org.apache.parquet.filter2.compat.FilterCompat.NoOpFilter;
 import org.apache.parquet.filter2.compat.FilterCompat.Visitor;
+import org.apache.parquet.filter2.dictionarylevel.DictionaryFilter;
 import org.apache.parquet.filter2.predicate.FilterPredicate;
 import org.apache.parquet.filter2.predicate.SchemaCompatibilityValidator;
 import org.apache.parquet.filter2.statisticslevel.StatisticsFilter;
@@ -40,15 +45,34 @@
 public class RowGroupFilter implements Visitor<List<BlockMetaData>> {
   private final List<BlockMetaData> blocks;
   private final MessageType schema;
+  private final List<FilterLevel> levels;
+  private final Supplier<FSDataInputStream> streamSupplier;
+
+  public enum FilterLevel {
+    STATISTICS,
+    DICTIONARY
+  }
 
   public static List<BlockMetaData> filterRowGroups(Filter filter, List<BlockMetaData> blocks, MessageType schema) {
     checkNotNull(filter, "filter");
     return filter.accept(new RowGroupFilter(blocks, schema));
   }
 
+  public static List<BlockMetaData> filterRowGroups(List<FilterLevel> levels, Filter filter, List<BlockMetaData> blocks, MessageType schema, Supplier<FSDataInputStream> streamSupplier) {
+    checkNotNull(filter, "filter");
+    return filter.accept(new RowGroupFilter(levels, blocks, schema, streamSupplier));
+  }
+
+  @Deprecated
   private RowGroupFilter(List<BlockMetaData> blocks, MessageType schema) {
+    this(Collections.singletonList(FilterLevel.STATISTICS), blocks, schema, null);
+  }
+
+  private RowGroupFilter(List<FilterLevel> levels, List<BlockMetaData> blocks, MessageType schema, Supplier<FSDataInputStream> streamSupplier) {
     this.blocks = checkNotNull(blocks, "blocks");
     this.schema = checkNotNull(schema, "schema");
+    this.levels = levels;
+    this.streamSupplier = streamSupplier;
   }
 
   @Override
@@ -61,7 +85,18 @@ public List<BlockMetaData> visit(FilterCompat.FilterPredicateCompat filterPredic
     List<BlockMetaData> filteredBlocks = new ArrayList<BlockMetaData>();
 
     for (BlockMetaData block : blocks) {
-      if (!StatisticsFilter.canDrop(filterPredicate, block.getColumns())) {
+      boolean drop = false;
+
+      if(levels.contains(FilterLevel.STATISTICS)) {
+        drop = StatisticsFilter.canDrop(filterPredicate, block.getColumns());
+      }
+
+      if(!drop && levels.contains(FilterLevel.DICTIONARY)) {
+        drop = DictionaryFilter.canDrop(filterPredicate, block.getColumns(), streamSupplier.get());
+        IOUtils.closeStream(streamSupplier.get());
+      }
+
+      if(!drop) {
         filteredBlocks.add(block);
       }
     }