diff --git a/parquet-column/src/main/java/org/apache/parquet/column/page/DictionaryPageReadStore.java b/parquet-column/src/main/java/org/apache/parquet/column/page/DictionaryPageReadStore.java new file mode 100644 index 0000000000..e401bff704 --- /dev/null +++ b/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); +} diff --git a/parquet-column/src/main/java/org/apache/parquet/column/page/DictionaryPageReader.java b/parquet-column/src/main/java/org/apache/parquet/column/page/DictionaryPageReader.java new file mode 100644 index 0000000000..81cc412194 --- /dev/null +++ b/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(); +} diff --git a/parquet-hadoop/pom.xml b/parquet-hadoop/pom.xml index a7f9d2c424..1d299715f8 100644 --- a/parquet-hadoop/pom.xml +++ b/parquet-hadoop/pom.xml @@ -77,7 +77,6 @@ com.google.guava guava 11.0 - test org.xerial.snappy diff --git a/parquet-hadoop/src/main/java/org/apache/parquet/filter2/compat/RowGroupFilter.java b/parquet-hadoop/src/main/java/org/apache/parquet/filter2/compat/RowGroupFilter.java index d85a231a31..d55f3cb8f0 100644 --- a/parquet-hadoop/src/main/java/org/apache/parquet/filter2/compat/RowGroupFilter.java +++ b/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> { private final List blocks; private final MessageType schema; + private final List levels; + private final Supplier streamSupplier; + + public enum FilterLevel { + STATISTICS, + DICTIONARY + } public static List filterRowGroups(Filter filter, List blocks, MessageType schema) { checkNotNull(filter, "filter"); return filter.accept(new RowGroupFilter(blocks, schema)); } + public static List filterRowGroups(List levels, Filter filter, List blocks, MessageType schema, Supplier streamSupplier) { + checkNotNull(filter, "filter"); + return filter.accept(new RowGroupFilter(levels, blocks, schema, streamSupplier)); + } + + @Deprecated private RowGroupFilter(List blocks, MessageType schema) { + this(Collections.singletonList(FilterLevel.STATISTICS), blocks, schema, null); + } + + private RowGroupFilter(List levels, List blocks, MessageType schema, Supplier streamSupplier) { this.blocks = checkNotNull(blocks, "blocks"); this.schema = checkNotNull(schema, "schema"); + this.levels = levels; + this.streamSupplier = streamSupplier; } @Override @@ -61,7 +85,18 @@ public List visit(FilterCompat.FilterPredicateCompat filterPredic List filteredBlocks = new ArrayList(); 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); } } diff --git a/parquet-hadoop/src/main/java/org/apache/parquet/filter2/dictionarylevel/DictionaryFilter.java b/parquet-hadoop/src/main/java/org/apache/parquet/filter2/dictionarylevel/DictionaryFilter.java new file mode 100644 index 0000000000..6cfa42d8b7 --- /dev/null +++ b/parquet-hadoop/src/main/java/org/apache/parquet/filter2/dictionarylevel/DictionaryFilter.java @@ -0,0 +1,263 @@ +/* + * 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.filter2.dictionarylevel; + +import org.apache.hadoop.conf.Configuration; +import org.apache.hadoop.fs.FSDataInputStream; +import org.apache.parquet.Log; +import org.apache.parquet.ParquetRuntimeException; +import org.apache.parquet.column.ColumnDescriptor; +import org.apache.parquet.column.Dictionary; +import org.apache.parquet.column.page.DictionaryPage; +import org.apache.parquet.filter2.predicate.FilterPredicate; +import org.apache.parquet.filter2.predicate.Operators.*; +import org.apache.parquet.filter2.predicate.UserDefinedPredicate; +import org.apache.parquet.hadoop.ParquetFileReader; +import org.apache.parquet.hadoop.metadata.ColumnChunkMetaData; +import org.apache.parquet.hadoop.metadata.ColumnPath; + +import java.io.IOException; +import java.util.*; + +import static org.apache.parquet.Preconditions.checkArgument; +import static org.apache.parquet.Preconditions.checkNotNull; + + +/** + * Applies filters based on the contents of column dictionaries. + */ +public class DictionaryFilter implements FilterPredicate.Visitor { + + private static final Log LOG = Log.getLog(DictionaryFilter.class); + + public static boolean canDrop(FilterPredicate pred, List columns, FSDataInputStream s) { + checkNotNull(pred, "pred"); + checkNotNull(columns, "columns"); + return pred.accept(new DictionaryFilter(columns, s)); + } + + private Configuration conf = null; + private final Map columns = new HashMap(); + private FSDataInputStream f; + + private DictionaryFilter(List columnsList, FSDataInputStream s) { + for (ColumnChunkMetaData chunk : columnsList) { + columns.put(chunk.getPath(), chunk); + } + + this.conf = new Configuration(); + this.f = s; + } + + private ColumnChunkMetaData getColumnChunk(ColumnPath columnPath) { + ColumnChunkMetaData c = columns.get(columnPath); + checkArgument(c != null, "Column " + columnPath.toDotString() + " not found in schema!"); + return c; + } + + private > Set expandDictionary(ColumnChunkMetaData meta) throws IOException, ParquetRuntimeException { + DictionaryPage page = ParquetFileReader.readDictionary(conf, meta, f); + + Dictionary dict = page.getEncoding().initDictionary(new ColumnDescriptor(null, meta.getType(), -1, -1), page); + + Set dictSet = new HashSet(); + + for(int i=0; i<=dict.getMaxId(); i++) { + switch(meta.getType()) { + case BINARY: dictSet.add(dict.decodeToBinary(i)); + break; + case INT32: dictSet.add(dict.decodeToInt(i)); + break; + case INT64: dictSet.add(dict.decodeToLong(i)); + break; + case FLOAT: dictSet.add(dict.decodeToFloat(i)); + break; + case DOUBLE: dictSet.add(dict.decodeToDouble(i)); + break; + default: + LOG.warn("Unknown dictionary type" + meta.getType()); + } + } + + return dictSet; + } + + @Override + public > Boolean visit(Eq eq) { + Column filterColumn = eq.getColumn(); + ColumnChunkMetaData meta = getColumnChunk(filterColumn.getColumnPath()); + T value = eq.getValue(); + + filterColumn.getColumnPath(); + + try { + Set dictSet = expandDictionary(meta); + return !dictSet.contains(value); + } catch (IOException e) { + LOG.warn("Failed to process dictionary for filter evaluation.", e); + } + + return false; + } + + @Override + public > Boolean visit(NotEq notEq) { + Column filterColumn = notEq.getColumn(); + ColumnChunkMetaData meta = getColumnChunk(filterColumn.getColumnPath()); + T value = notEq.getValue(); + + filterColumn.getColumnPath(); + + try { + Set dictSet = expandDictionary(meta); + return dictSet.size() == 1 && dictSet.contains(value); + } catch (IOException e) { + LOG.warn("Failed to process dictionary for filter evaluation.", e); + } + + return false; + } + + @Override + public > Boolean visit(Lt lt) { + Column filterColumn = lt.getColumn(); + ColumnChunkMetaData meta = getColumnChunk(filterColumn.getColumnPath()); + T value = lt.getValue(); + + filterColumn.getColumnPath(); + + try { + Set dictSet = expandDictionary(meta); + + for(T entry : dictSet) { + if(value.compareTo(entry) > 0) { + return false; + } + } + + return true; + } catch (IOException e) { + LOG.warn("Failed to process dictionary for filter evaluation.", e); + } + + return false; + } + + @Override + public > Boolean visit(LtEq ltEq) { + Column filterColumn = ltEq.getColumn(); + ColumnChunkMetaData meta = getColumnChunk(filterColumn.getColumnPath()); + T value = ltEq.getValue(); + + filterColumn.getColumnPath(); + + try { + Set dictSet = expandDictionary(meta); + + for(T entry : dictSet) { + if(value.compareTo(entry) >= 0) { + return false; + } + } + + return true; + } catch (IOException e) { + LOG.warn("Failed to process dictionary for filter evaluation.", e); + } + + return false; + } + + @Override + public > Boolean visit(Gt gt) { + Column filterColumn = gt.getColumn(); + ColumnChunkMetaData meta = getColumnChunk(filterColumn.getColumnPath()); + T value = gt.getValue(); + + filterColumn.getColumnPath(); + + try { + Set dictSet = expandDictionary(meta); + + for(T entry : dictSet) { + if(value.compareTo(entry) < 0) { + return false; + } + } + + return true; + } catch (IOException e) { + LOG.warn("Failed to process dictionary for filter evaluation.", e); + } + + return false; + } + + @Override + public > Boolean visit(GtEq gtEq) { + Column filterColumn = gtEq.getColumn(); + ColumnChunkMetaData meta = getColumnChunk(filterColumn.getColumnPath()); + T value = gtEq.getValue(); + + filterColumn.getColumnPath(); + + try { + Set dictSet = expandDictionary(meta); + + for(T entry : dictSet) { + if(value.compareTo(entry) <= 0) { + return false; + } + } + + return true; + } catch (IOException e) { + LOG.warn("Failed to process dictionary for filter evaluation.", e); + } + + return false; + } + + @Override + public Boolean visit(And and) { + return and.getLeft().accept(this) || and.getRight().accept(this); + } + + @Override + public Boolean visit(Or or) { + return or.getLeft().accept(this) && or.getRight().accept(this); + } + + @Override + public Boolean visit(Not not) { + throw new IllegalArgumentException( + "This predicate contains a not! Did you forget to run this predicate through LogicalInverseRewriter? " + not); + } + + @Override + public , U extends UserDefinedPredicate> Boolean visit(UserDefined udp) { + throw new UnsupportedOperationException("UDP not supported with dictionary evaluation."); + } + + @Override + public , U extends UserDefinedPredicate> Boolean visit(LogicalNotUserDefined udp) { + throw new UnsupportedOperationException("UDP not supported with dictionary evaluation."); + } + +} diff --git a/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ColumnChunkDictionaryPageReadStore.java b/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ColumnChunkDictionaryPageReadStore.java new file mode 100644 index 0000000000..4cd34d0688 --- /dev/null +++ b/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ColumnChunkDictionaryPageReadStore.java @@ -0,0 +1,89 @@ +/* + * 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.hadoop; + +import java.io.IOException; +import java.util.HashMap; +import java.util.Map; + +import org.apache.parquet.Ints; +import org.apache.parquet.Log; +import org.apache.parquet.column.ColumnDescriptor; +import org.apache.parquet.column.page.DictionaryPage; +import org.apache.parquet.column.page.DictionaryPageReadStore; +import org.apache.parquet.column.page.DictionaryPageReader; +import org.apache.parquet.hadoop.CodecFactory.BytesDecompressor; + +class ColumnChunkDictionaryPageReadStore implements DictionaryPageReadStore { + /** + * DictionaryPageReader for a single column chunk. A column chunk contains + * several pages, first of which could be a dictionary page. + * + * This implementation is provided with compressed dictionary page + */ + static final class ColumnChunkDictionaryPageReader implements DictionaryPageReader { + + private final BytesDecompressor decompressor; + private final DictionaryPage compressedDictionaryPage; + + ColumnChunkDictionaryPageReader(BytesDecompressor decompressor, DictionaryPage compressedDictionaryPage) { + this.decompressor = decompressor; + this.compressedDictionaryPage = compressedDictionaryPage; + } + + @Override + public int getDictionarySize() { + return compressedDictionaryPage.getDictionarySize(); + } + + @Override + public DictionaryPage readDictionaryPage() { + if (compressedDictionaryPage == null) { + return null; + } + try { + return new DictionaryPage( + decompressor.decompress(compressedDictionaryPage.getBytes(), compressedDictionaryPage.getUncompressedSize()), + compressedDictionaryPage.getDictionarySize(), + compressedDictionaryPage.getEncoding()); + } catch (IOException e) { + throw new RuntimeException(e); + } + } + } + + private final Map readers = new HashMap(); + + public ColumnChunkDictionaryPageReadStore() { + } + + @Override + public DictionaryPageReader getDictionaryPageReader(ColumnDescriptor column) { + if (!readers.containsKey(column)) { + throw new IllegalArgumentException(column + " is not in the store: " + readers.keySet()); + } + return readers.get(column); + } + + void addColumn(ColumnDescriptor column, ColumnChunkDictionaryPageReader reader) { + if (readers.put(column, reader) != null) { + throw new RuntimeException(column + " was added twice"); + } + } +} diff --git a/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ParquetFileReader.java b/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ParquetFileReader.java index f43e6924f2..7f592b60d6 100644 --- a/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ParquetFileReader.java +++ b/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ParquetFileReader.java @@ -59,6 +59,7 @@ import org.apache.parquet.column.page.DataPage; import org.apache.parquet.column.page.DataPageV1; import org.apache.parquet.column.page.DataPageV2; +import org.apache.parquet.column.page.DictionaryPageReader; import org.apache.parquet.column.page.DictionaryPage; import org.apache.parquet.column.page.PageReadStore; import org.apache.parquet.hadoop.metadata.ColumnPath; @@ -71,6 +72,7 @@ import org.apache.parquet.format.converter.ParquetMetadataConverter.MetadataFilter; import org.apache.parquet.hadoop.CodecFactory.BytesDecompressor; import org.apache.parquet.hadoop.ColumnChunkPageReadStore.ColumnChunkPageReader; +import org.apache.parquet.hadoop.ColumnChunkDictionaryPageReadStore.ColumnChunkDictionaryPageReader; import org.apache.parquet.hadoop.metadata.BlockMetaData; import org.apache.parquet.hadoop.metadata.ColumnChunkMetaData; import org.apache.parquet.hadoop.metadata.FileMetaData; @@ -445,6 +447,7 @@ public static final ParquetMetadata readFooter(Configuration configuration, File private final CodecFactory codecFactory; private final List blocks; + private final Configuration configuration; private final FSDataInputStream f; private final Path filePath; private final Map paths = new HashMap(); @@ -468,9 +471,8 @@ public ParquetFileReader(Configuration configuration, Path filePath, List blocks, List columns) throws IOException { + public ParquetFileReader(Configuration configuration, FileMetaData fileMetaData, Path filePath, List blocks, List columns) throws IOException { + this.configuration = configuration; this.filePath = filePath; this.fileMetaData = fileMetaData; this.createdBy = fileMetaData == null ? null : fileMetaData.getCreatedBy(); @@ -526,7 +528,35 @@ public PageReadStore readNextRowGroup() throws IOException { return columnChunkPageReadStore; } + public static DictionaryPage readDictionary(Configuration conf, ColumnChunkMetaData meta, FSDataInputStream fin) throws IOException { + if(fin.getPos() != meta.getStartingPos()) { + fin.seek(meta.getStartingPos()); + } + + PageHeader pageHeader = Util.readPageHeader(fin); + + return readDictionary(conf, pageHeader, meta, fin); + } + + public static DictionaryPage readDictionary(Configuration conf, PageHeader pageHeader, ColumnChunkMetaData meta, FSDataInputStream fin) throws IOException { + DictionaryPageHeader dictHeader = pageHeader.getDictionary_page_header(); + + int uncompressedPageSize = pageHeader.getUncompressed_page_size(); + int compressedPageSize = pageHeader.getCompressed_page_size(); + + byte [] dictPageBytes = new byte[compressedPageSize]; + + fin.readFully(dictPageBytes); + BytesInput bin = BytesInput.from(dictPageBytes); + + DictionaryPage compressedPage = new DictionaryPage(bin, uncompressedPageSize, dictHeader.getNum_values(), converter.getEncoding(dictHeader.getEncoding())); + + BytesDecompressor decompressor = new CodecFactory(conf).getDecompressor(meta.getCodec()); + DictionaryPageReader dictionaryPageReader = new ColumnChunkDictionaryPageReader(decompressor, compressedPage); + + return dictionaryPageReader.readDictionaryPage(); + } @Override public void close() throws IOException { @@ -578,14 +608,8 @@ public ColumnChunkPageReader readAllPages() throws IOException { if (dictionaryPage != null) { throw new ParquetDecodingException("more than one dictionary page in column " + descriptor.col); } - DictionaryPageHeader dicHeader = pageHeader.getDictionary_page_header(); - dictionaryPage = - new DictionaryPage( - this.readAsBytesInput(compressedPageSize), - uncompressedPageSize, - dicHeader.getNum_values(), - converter.getEncoding(dicHeader.getEncoding()) - ); + + dictionaryPage = readDictionary(configuration, pageHeader, descriptor.metadata, f); break; case DATA_PAGE: DataPageHeader dataHeaderV1 = pageHeader.getData_page_header(); diff --git a/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ParquetRecordReader.java b/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ParquetRecordReader.java index 1558fc03bb..0d2cd6aa48 100644 --- a/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ParquetRecordReader.java +++ b/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ParquetRecordReader.java @@ -19,6 +19,7 @@ package org.apache.parquet.hadoop; import static org.apache.parquet.filter2.compat.RowGroupFilter.filterRowGroups; +import static org.apache.parquet.filter2.compat.RowGroupFilter.FilterLevel.*; import static org.apache.parquet.format.converter.ParquetMetadataConverter.NO_FILTER; import static org.apache.parquet.format.converter.ParquetMetadataConverter.range; import static org.apache.parquet.hadoop.ParquetFileReader.readFooter; @@ -32,8 +33,13 @@ import java.util.List; import java.util.Set; +import com.google.common.base.Supplier; +import com.google.common.base.Suppliers; import org.apache.hadoop.conf.Configuration; +import org.apache.hadoop.fs.FSDataInputStream; +import org.apache.hadoop.fs.FileSystem; import org.apache.hadoop.fs.Path; +import org.apache.hadoop.io.IOUtils; import org.apache.hadoop.mapred.Reporter; import org.apache.hadoop.mapreduce.InputSplit; import org.apache.hadoop.mapreduce.RecordReader; @@ -47,6 +53,7 @@ import org.apache.parquet.filter.UnboundRecordFilter; import org.apache.parquet.filter2.compat.FilterCompat; import org.apache.parquet.filter2.compat.FilterCompat.Filter; +import org.apache.parquet.filter2.compat.RowGroupFilter.FilterLevel; import org.apache.parquet.hadoop.api.ReadSupport; import org.apache.parquet.hadoop.metadata.BlockMetaData; import org.apache.parquet.hadoop.metadata.ColumnChunkMetaData; @@ -57,6 +64,8 @@ import org.apache.parquet.io.ParquetDecodingException; import org.apache.parquet.schema.MessageType; + + /** * Reads the records from a block of a Parquet file * @@ -152,7 +161,8 @@ public void initialize(InputSplit inputSplit, Configuration configuration, Repor } private void initializeInternalReader(ParquetInputSplit split, Configuration configuration) throws IOException { - Path path = split.getPath(); + final Path path = split.getPath(); + final Configuration conf = configuration; long[] rowGroupOffsets = split.getRowGroupOffsets(); List filteredBlocks; ParquetMetadata footer; @@ -162,7 +172,34 @@ private void initializeInternalReader(ParquetInputSplit split, Configuration con footer = readFooter(configuration, path, range(split.getStart(), split.getEnd())); MessageType fileSchema = footer.getFileMetaData().getSchema(); Filter filter = getFilter(configuration); - filteredBlocks = filterRowGroups(filter, footer.getBlocks(), fileSchema); + + List levels = new ArrayList(); + + if(configuration.getBoolean("parquet.filter.statistics.enabled", true)) { + levels.add(STATISTICS); + } + + //This is for lazy evaluation so that if stats level can provide the + //result, we don't need to open a new file stream. + Supplier streamSupplier = null; + if(configuration.getBoolean("parquet.filter.dictionary.enabled", false)) { + levels.add(DICTIONARY); + + streamSupplier = Suppliers.memoize(new Supplier() { + @Override + public FSDataInputStream get() { + try { + FileSystem fs = path.getFileSystem(conf); + return fs.open(path); + } catch (IOException e) { + LOG.warn("Failed to open path for dictionary predicate evaluation: " + path, e); + } + return null; + } + }); + } + + filteredBlocks = filterRowGroups(levels, filter, footer.getBlocks(), fileSchema, streamSupplier); } else { // otherwise we find the row groups that were selected on the client footer = readFooter(configuration, path, NO_FILTER); diff --git a/parquet-hadoop/src/test/java/org/apache/parquet/filter2/dictionarylevel/DictionaryFilterTest.java b/parquet-hadoop/src/test/java/org/apache/parquet/filter2/dictionarylevel/DictionaryFilterTest.java new file mode 100644 index 0000000000..ca7577a104 --- /dev/null +++ b/parquet-hadoop/src/test/java/org/apache/parquet/filter2/dictionarylevel/DictionaryFilterTest.java @@ -0,0 +1,171 @@ +/* + * 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.filter2.dictionarylevel; + +import org.apache.hadoop.conf.Configuration; +import org.apache.hadoop.fs.FSDataInputStream; +import org.apache.hadoop.fs.FileSystem; +import org.apache.hadoop.fs.Path; +import org.apache.parquet.example.data.Group; +import org.apache.parquet.example.data.simple.SimpleGroupFactory; +import org.apache.parquet.filter2.predicate.FilterPredicate; +import org.apache.parquet.filter2.predicate.Operators.BinaryColumn; +import org.apache.parquet.filter2.predicate.Operators.IntColumn; +import org.apache.parquet.format.converter.ParquetMetadataConverter; +import org.apache.parquet.hadoop.ParquetFileReader; +import org.apache.parquet.hadoop.ParquetWriter; +import org.apache.parquet.hadoop.example.GroupWriteSupport; +import org.apache.parquet.hadoop.metadata.ColumnChunkMetaData; +import org.apache.parquet.hadoop.metadata.ParquetMetadata; +import org.apache.parquet.io.api.Binary; +import org.apache.parquet.schema.MessageType; +import org.junit.*; + +import java.io.IOException; +import java.util.List; + +import static org.apache.parquet.column.ParquetProperties.WriterVersion.PARQUET_1_0; +import static org.apache.parquet.filter2.dictionarylevel.DictionaryFilter.canDrop; +import static org.apache.parquet.filter2.predicate.FilterApi.*; +import static org.apache.parquet.hadoop.metadata.CompressionCodecName.UNCOMPRESSED; +import static org.apache.parquet.schema.MessageTypeParser.parseMessageType; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertTrue; + +/** + * Created by dweeks on 10/19/15. + */ +public class DictionaryFilterTest { + + private static final int nElements = 1000; + private static final Configuration conf = new Configuration(); + private static Path file = new Path("target/test/TestDictionaryFilter/testParquetFile"); + private static final MessageType schema = parseMessageType( + "message test { " + + "required binary binary_field; " + + "required int32 int32_field; " + + "required int64 int64_field; " + + "required double double_field; " + + "required float float_field; " + + "} "); + + private static final int ENGLISH_CHARACTER_NUMBER = 26; + private static final int[] intValues = new int[] {-100, 302, 3333333, 7654321, 1234567, -2000, -77775, 0, 75, 22223, + 77, 22221, -444443, 205, 12, 44444, 889, 66665, -777889, -7, + 52, 33, -257, 1111, 775, 26}; + private static final long[] longValues = new long[] {-100L, 302L, 3333333L, 7654321L, 1234567L, -2000L, -77775L, 0L, + 75L, 22223L, 77L, 22221L, -444443L, 205L, 12L, 44444L, 889L, 66665L, + -777889L, -7L, 52L, 33L, -257L, 1111L, 775L, 26L}; + + private static void writeData(SimpleGroupFactory f, ParquetWriter writer) throws IOException { + for (int i = 0; i < nElements; i++) { + int index = i % ENGLISH_CHARACTER_NUMBER; + char c = (char) ((index) + 'a'); + String b = String.valueOf(c); + + Group group = f.newGroup() + .append("binary_field", b) + .append("int32_field", intValues[index]) + .append("int64_field", longValues[index]) + .append("double_field", intValues[index] * 1.0) + .append("float_field", ((float) (intValues[index] * 2.0))); + + writer.write(group); + } + writer.close(); + } + + @BeforeClass + public static void prepareFile() throws IOException { + cleanup(); + + boolean dictionaryEnabled = true; + boolean validating = false; + GroupWriteSupport.setSchema(schema, conf); + SimpleGroupFactory f = new SimpleGroupFactory(schema); + ParquetWriter writer = new ParquetWriter( + file, + new GroupWriteSupport(), + UNCOMPRESSED, 1024*1024, 1024, 1024*1024, + dictionaryEnabled, validating, PARQUET_1_0, conf); + writeData(f, writer); + } + + @AfterClass + public static void cleanup() throws IOException { + FileSystem fs = file.getFileSystem(conf); + if (fs.exists(file)) { + fs.delete(file, true); + } + } + + + List ccmd; + FSDataInputStream stream; + + @Before + public void setUp() throws Exception { + FileSystem fs = FileSystem.getLocal(conf); + ParquetMetadata meta = ParquetFileReader.readFooter(conf, fs.getFileStatus(file), ParquetMetadataConverter.NO_FILTER); + + ccmd = meta.getBlocks().get(0).getColumns(); + stream = fs.open(file); + } + + @After + public void tearDown() throws Exception { + stream.close(); + } + + @Test + public void testCanDropEq() throws Exception { + BinaryColumn b = binaryColumn("binary_field"); + FilterPredicate pred = eq(b, Binary.fromString("c")); + + assertFalse(canDrop(pred, ccmd, stream)); + } + + @Test + public void testCanDropLt() throws Exception { + IntColumn i32 = intColumn("int32_field"); + FilterPredicate predDrop = lt(i32, -777889); + assertTrue(canDrop(predDrop, ccmd, stream)); + + FilterPredicate predKeep = lt(i32, -1); + assertFalse(canDrop(predKeep, ccmd, stream)); + } + + @Test + public void testCanDropLtEq() throws Exception { + IntColumn i32 = intColumn("int32_field"); + FilterPredicate pred = ltEq(i32, -777890); + + assertTrue(canDrop(pred, ccmd, stream)); + } + + @Test + public void testCanDropAnd() throws Exception { + IntColumn i32 = intColumn("int32_field"); + FilterPredicate pred = and(lt(i32, -777889), gt(i32, 3333332)); + + assertTrue(canDrop(pred, ccmd, stream)); + } + +} \ No newline at end of file