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-column/src/main/java/org/apache/parquet/column/page/DictionaryReader.java b/parquet-column/src/main/java/org/apache/parquet/column/page/DictionaryReader.java new file mode 100644 index 0000000000..844d800668 --- /dev/null +++ b/parquet-column/src/main/java/org/apache/parquet/column/page/DictionaryReader.java @@ -0,0 +1,46 @@ +/* + * 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; +import org.apache.parquet.column.Dictionary; + +import java.io.IOException; + +/** + * Read a dictionary from dictionary page + */ +public class DictionaryReader { + private final ColumnDescriptor column; + private final DictionaryPageReader pageReader; + + public DictionaryReader(ColumnDescriptor column, DictionaryPageReadStore dictionaryPageReadStore) { + this.column = column; + this.pageReader = dictionaryPageReadStore.getDictionaryPageReader(column); + } + + public Dictionary readDictionary() throws IOException { + DictionaryPage dictionaryPage = pageReader.readDictionaryPage(); + return dictionaryPage.getEncoding().initDictionary(column, dictionaryPage); + } + + public int getDictionarySize() { + return pageReader.getDictionarySize(); + } +} 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..e144546d8a --- /dev/null +++ b/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ColumnChunkDictionaryPageReadStore.java @@ -0,0 +1,73 @@ +/* + * 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; +import org.apache.parquet.hadoop.ColumnChunkPageReadStore.ColumnChunkPageReader; + +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 a ColumnChunkPageReader delegate + */ + static final class ColumnChunkDictionaryPageReader implements DictionaryPageReader { + + private final ColumnChunkPageReader columnChunkPageReader; + + ColumnChunkDictionaryPageReader(ColumnChunkPageReader columnChunkPageReader) { + this.columnChunkPageReader = columnChunkPageReader; + } + + @Override + public int getDictionarySize() { + return columnChunkPageReader.getCompressedDictionaryPage().getDictionarySize(); + } + + @Override + public DictionaryPage readDictionaryPage() { + return columnChunkPageReader.readDictionaryPage(); + } + } + + private final Map readers; + + ColumnChunkDictionaryPageReadStore(Map readers) { + this.readers = readers; + } + + @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); + } +} diff --git a/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ColumnChunkPageReadStore.java b/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ColumnChunkPageReadStore.java index b6934c2fb6..ed56fd3d7b 100644 --- a/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ColumnChunkPageReadStore.java +++ b/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ColumnChunkPageReadStore.java @@ -139,6 +139,10 @@ public DictionaryPage readDictionaryPage() { throw new RuntimeException(e); // TODO: cleanup } } + + public DictionaryPage getCompressedDictionaryPage() { + return compressedDictionaryPage; + } } private final Map readers = new HashMap(); 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 ea7a6723fa..e985c62050 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 @@ -56,9 +56,13 @@ import org.apache.parquet.Log; import org.apache.parquet.bytes.BytesInput; import org.apache.parquet.column.ColumnDescriptor; +import org.apache.parquet.column.Dictionary; +import org.apache.parquet.column.Encoding; 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.DictionaryPageReadStore; +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; @@ -66,10 +70,12 @@ import org.apache.parquet.format.DataPageHeaderV2; import org.apache.parquet.format.DictionaryPageHeader; import org.apache.parquet.format.PageHeader; +import org.apache.parquet.format.PageType; import org.apache.parquet.format.Util; import org.apache.parquet.format.converter.ParquetMetadataConverter; import org.apache.parquet.format.converter.ParquetMetadataConverter.MetadataFilter; import org.apache.parquet.hadoop.CodecFactory.BytesDecompressor; +import org.apache.parquet.hadoop.ColumnChunkDictionaryPageReadStore.ColumnChunkDictionaryPageReader; import org.apache.parquet.hadoop.ColumnChunkPageReadStore.ColumnChunkPageReader; import org.apache.parquet.hadoop.metadata.BlockMetaData; import org.apache.parquet.hadoop.metadata.ColumnChunkMetaData; @@ -78,6 +84,7 @@ import org.apache.parquet.hadoop.util.HiddenFileFilter; import org.apache.parquet.hadoop.util.counters.BenchmarkCounter; import org.apache.parquet.io.ParquetDecodingException; +import org.apache.parquet.schema.MessageType; /** * Internal implementation of the Parquet file reader as a block container @@ -517,7 +524,74 @@ public PageReadStore readNextRowGroup() throws IOException { return columnChunkPageReadStore; } + /** + * Reads the requested predicate columns dictionaries from the row group at the current file position. + * Not reading all the requested columns, only read predicate columns. + * eg. select columnA, columnB, columnC from table where columnC > 10 + * only reads columnC's dictionary + * @param requestedSchema the predicate columns to read + * @throws IOException if an error occurs while reading + * @return the DictionaryPageReadStore which can provide DictionaryPageReaders for each column. + */ + public DictionaryPageReadStore getCurrentRowGroupDictionaries(MessageType requestedSchema) throws IOException { + if (currentBlock == blocks.size()) { + return null; + } + BlockMetaData block = blocks.get(currentBlock); + if (block.getRowCount() == 0) { + throw new RuntimeException("Illegal row group of 0 rows"); + } + Map dictionaryReaders = new HashMap(); + for (ColumnChunkMetaData columnChunkMetaData : block.getColumns()) { + for (ColumnDescriptor columnDescriptor : requestedSchema.getColumns()) { + if (columnChunkMetaData.getPath().equals(ColumnPath.get(columnDescriptor.getPath())) && + isOnlyDictionaryEncodingPages(columnChunkMetaData.getEncodings())) { + long startingPosition = columnChunkMetaData.getStartingPos(); + this.f.seek(startingPosition); + + int totalSize = (int) columnChunkMetaData.getTotalSize(); + byte[] buffer = new byte[totalSize]; + // TODO: set read size to be (page header size + compressed dictionary page size) + this.f.readFully(buffer); + + ChunkDescriptor chunkDescriptor = new ChunkDescriptor(columnDescriptor, + columnChunkMetaData, + startingPosition, + totalSize); + Chunk chunk = new Chunk(chunkDescriptor, buffer, 0); + DictionaryPage compressedPage = chunk.readDictionaryPage(); + if (compressedPage == null) { + dictionaryReaders.put(columnDescriptor, null); + break; + } + + BytesDecompressor decompressor = codecFactory.getDecompressor(columnChunkMetaData.getCodec()); + ColumnChunkPageReader columnChunkPageReader = new ColumnChunkPageReader(decompressor, new ArrayList(), compressedPage); + ColumnChunkDictionaryPageReader dictionaryPageReader = new ColumnChunkDictionaryPageReader(columnChunkPageReader); + dictionaryReaders.put(columnDescriptor, dictionaryPageReader); + } + } + } + return new ColumnChunkDictionaryPageReadStore(dictionaryReaders); + } + + // are all the value pages dictionary encoded + private boolean isOnlyDictionaryEncodingPages(Set encodings) { + // if values have more than one encodings, definitely not dictionary encoding only + if (encodings.size() > 3) { + return false; + } + // heuristics: definition level, repetition level are not dictionary encoded + // if there is dictionary encoding, it must be values + // TODO: use PageEncodingStats to determine whether all value pages are dictionary encoded + for (Encoding encoding : encodings) { + if (encoding.usesDictionary()) { + return true; + } + } + return false; + } @Override public void close() throws IOException { @@ -535,6 +609,8 @@ private class Chunk extends ByteArrayInputStream { private final ChunkDescriptor descriptor; + private PageHeader pageHeader; + /** * * @param descriptor descriptor for the chunk @@ -551,6 +627,26 @@ protected PageHeader readPageHeader() throws IOException { return Util.readPageHeader(this); } + /** + * Read only dictionary page in a given column chunk. + * readDictionaryPage should not be used interleaving with readAllPages + * either read only dictionary page, or read all pages + * @return dictionary page + */ + public DictionaryPage readDictionaryPage() throws IOException { + this.pageHeader = readPageHeader(); + int uncompressedPageSize = pageHeader.getUncompressed_page_size(); + int compressedPageSize = pageHeader.getCompressed_page_size(); + if (pageHeader.type == PageType.DICTIONARY_PAGE) { + DictionaryPageHeader dicHeader = pageHeader.getDictionary_page_header(); + return new DictionaryPage(this.readAsBytesInput(compressedPageSize), + uncompressedPageSize, + dicHeader.getNum_values(), + converter.getEncoding(dicHeader.getEncoding())); + } + return null; + } + /** * Read all of the pages in a given column chunk. * @return the list of pages @@ -560,24 +656,14 @@ public ColumnChunkPageReader readAllPages() throws IOException { DictionaryPage dictionaryPage = null; long valuesCountReadSoFar = 0; while (valuesCountReadSoFar < descriptor.metadata.getValueCount()) { - PageHeader pageHeader = readPageHeader(); - int uncompressedPageSize = pageHeader.getUncompressed_page_size(); - int compressedPageSize = pageHeader.getCompressed_page_size(); + DictionaryPage nextDictionaryPage = readDictionaryPage(); + if (dictionaryPage != null && nextDictionaryPage != null) { + throw new ParquetDecodingException("more than one dictionary page in column " + descriptor.col); + } + dictionaryPage = nextDictionaryPage; + int uncompressedPageSize = this.pageHeader.getUncompressed_page_size(); + int compressedPageSize = this.pageHeader.getCompressed_page_size(); switch (pageHeader.type) { - case DICTIONARY_PAGE: - // there is only one dictionary page per column chunk - 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()) - ); - break; case DATA_PAGE: DataPageHeader dataHeaderV1 = pageHeader.getData_page_header(); pagesInChunk.add( @@ -616,6 +702,8 @@ public ColumnChunkPageReader readAllPages() throws IOException { )); valuesCountReadSoFar += dataHeaderV2.getNum_values(); break; + case DICTIONARY_PAGE: + throw new ParquetDecodingException("more than one dictionary page in column " + descriptor.col); default: if (DEBUG) LOG.debug("skipping page of type " + pageHeader.getType() + " of size " + compressedPageSize); this.skip(compressedPageSize); @@ -801,5 +889,4 @@ public long endPos() { } } - } diff --git a/parquet-hadoop/src/test/java/org/apache/parquet/hadoop/TestDictionaryReader.java b/parquet-hadoop/src/test/java/org/apache/parquet/hadoop/TestDictionaryReader.java new file mode 100644 index 0000000000..f6d95ba167 --- /dev/null +++ b/parquet-hadoop/src/test/java/org/apache/parquet/hadoop/TestDictionaryReader.java @@ -0,0 +1,201 @@ +/* + * 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 org.apache.hadoop.conf.Configuration; +import org.apache.hadoop.fs.FileSystem; +import org.apache.hadoop.fs.Path; +import org.apache.parquet.column.ColumnDescriptor; +import org.apache.parquet.column.Dictionary; +import org.apache.parquet.column.page.DictionaryPageReadStore; +import org.apache.parquet.column.page.DictionaryPage; +import org.apache.parquet.column.page.DictionaryReader; +import org.apache.parquet.example.data.Group; +import org.apache.parquet.example.data.simple.SimpleGroupFactory; +import org.apache.parquet.hadoop.ParquetReader; +import org.apache.parquet.hadoop.ParquetWriter; +import org.apache.parquet.hadoop.example.GroupWriteSupport; +import org.apache.parquet.hadoop.example.GroupReadSupport; +import org.apache.parquet.hadoop.metadata.BlockMetaData; +import org.apache.parquet.hadoop.metadata.ParquetMetadata; +import org.apache.parquet.io.api.Binary; +import org.apache.parquet.schema.MessageType; +import org.junit.AfterClass; +import org.junit.Test; + +import java.io.IOException; +import java.nio.CharBuffer; +import java.util.Arrays; +import java.util.Map; + +import static com.google.common.base.Charsets.UTF_8; +import static java.lang.Boolean.FALSE; +import static java.lang.Boolean.TRUE; +import static org.apache.parquet.column.ParquetProperties.WriterVersion.PARQUET_1_0; +import static org.apache.parquet.hadoop.api.ReadSupport.PARQUET_READ_SCHEMA; +import static org.apache.parquet.hadoop.metadata.CompressionCodecName.GZIP; +import static org.apache.parquet.schema.MessageTypeParser.parseMessageType; +import static org.junit.Assert.*; +import org.junit.BeforeClass; + +public class TestDictionaryReader { + + private static final int nElements = 1000; + private static final Configuration conf = new Configuration(); + private static final Path file = new Path("target/test/TestDictionaryReader/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(), + GZIP, 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); + } + } + + @Test + public void testBinaryRead() throws Exception { + MessageType requestedSchema = parseMessageType("message test { required binary binary_field;}"); + ParquetMetadata metadata = ParquetFileReader.readFooter(conf, file); + ParquetFileReader fileReader = new ParquetFileReader(conf, metadata.getFileMetaData(), file, metadata.getBlocks(), requestedSchema.getColumns()); + DictionaryPageReadStore dictionaryReadStore = fileReader.getCurrentRowGroupDictionaries(requestedSchema); + ColumnDescriptor column = requestedSchema.getColumns().get(0); + DictionaryReader dictionaryReader = new DictionaryReader(column, dictionaryReadStore); + Dictionary dictionary = dictionaryReader.readDictionary(); + + for (int i = 0; i < intValues.length; i++) { + assertEquals(String.valueOf((char) ((i % intValues.length) + 'a')), dictionary.decodeToBinary(i).toStringUsingUTF8()); + } + fileReader.close(); + } + + @Test + public void testIntRead() throws Exception { + MessageType requestedSchema = parseMessageType("message test { required int32 int32_field;}"); + ParquetMetadata metadata = ParquetFileReader.readFooter(conf, file); + ParquetFileReader fileReader = new ParquetFileReader(conf, metadata.getFileMetaData(), file, metadata.getBlocks(), requestedSchema.getColumns()); + DictionaryPageReadStore dictionaryReadStore = fileReader.getCurrentRowGroupDictionaries(requestedSchema); + ColumnDescriptor column = requestedSchema.getColumns().get(0); + DictionaryReader dictionaryReader = new DictionaryReader(column, dictionaryReadStore); + Dictionary dictionary = dictionaryReader.readDictionary(); + + for (int i = 0; i < intValues.length; i++) { + assertEquals(intValues[i % intValues.length], dictionary.decodeToInt(i)); + } + fileReader.close(); + } + + @Test + public void testLongRead() throws Exception { + MessageType requestedSchema = parseMessageType("message test { required int64 int64_field;}"); + ParquetMetadata metadata = ParquetFileReader.readFooter(conf, file); + ParquetFileReader fileReader = new ParquetFileReader(conf, metadata.getFileMetaData(), file, metadata.getBlocks(), requestedSchema.getColumns()); + DictionaryPageReadStore dictionaryReadStore = fileReader.getCurrentRowGroupDictionaries(requestedSchema); + ColumnDescriptor column = requestedSchema.getColumns().get(0); + DictionaryReader dictionaryReader = new DictionaryReader(column, dictionaryReadStore); + Dictionary dictionary = dictionaryReader.readDictionary(); + + for (int i = 0; i < intValues.length; i++) { + assertEquals(longValues[i % intValues.length], dictionary.decodeToLong(i)); + } + fileReader.close(); + } + + @Test + public void testDoubleRead() throws Exception { + MessageType requestedSchema = parseMessageType("message test { required double double_field;}"); + ParquetMetadata metadata = ParquetFileReader.readFooter(conf, file); + ParquetFileReader fileReader = new ParquetFileReader(conf, metadata.getFileMetaData(), file, metadata.getBlocks(), requestedSchema.getColumns()); + DictionaryPageReadStore dictionaryReadStore = fileReader.getCurrentRowGroupDictionaries(requestedSchema); + ColumnDescriptor column = requestedSchema.getColumns().get(0); + DictionaryReader dictionaryReader = new DictionaryReader(column, dictionaryReadStore); + Dictionary dictionary = dictionaryReader.readDictionary(); + + for (int i = 0; i < intValues.length; i++) { + assertEquals(intValues[i % intValues.length] * 1.0, dictionary.decodeToDouble(i), 1e-15); + } + fileReader.close(); + } + + @Test + public void testFloatRead() throws Exception { + MessageType requestedSchema = parseMessageType("message test { required float float_field;}"); + ParquetMetadata metadata = ParquetFileReader.readFooter(conf, file); + ParquetFileReader fileReader = new ParquetFileReader(conf, metadata.getFileMetaData(), file, metadata.getBlocks(), requestedSchema.getColumns()); + DictionaryPageReadStore dictionaryReadStore = fileReader.getCurrentRowGroupDictionaries(requestedSchema); + ColumnDescriptor column = requestedSchema.getColumns().get(0); + DictionaryReader dictionaryReader = new DictionaryReader(column, dictionaryReadStore); + Dictionary dictionary = dictionaryReader.readDictionary(); + + for (int i = 0; i < intValues.length; i++) { + assertEquals((float) intValues[i % intValues.length] * 2.0, dictionary.decodeToFloat(i), 1e-15); + } + fileReader.close(); + } +} +