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 d29738a6d8..ce58f2291c 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 @@ -135,6 +135,10 @@ public class ParquetFileReader implements Closeable { public static final long HADOOP_VECTORED_READ_TIMEOUT_SECONDS = 300; + // old parquet-mr versions (PARQUET-816) left the header of the dictionary page (13 to 19 bytes) out of the size of + // the column chunk + private static final int MAX_MISSING_DICTIONARY_HEADER_BYTES = 100; + private final ParquetMetadataConverter converter; private final CRC32 crc; @@ -2024,10 +2028,11 @@ private class ChunkListBuilder { private class ChunkData { final List buffers = new ArrayList<>(); OffsetIndex offsetIndex; + // end of the consecutive part list if the last part added ends it, otherwise -1 + long tailEndPos = -1; } private final Map map = new HashMap<>(); - private ChunkDescriptor lastDescriptor; private final long rowCount; private SeekableInputStream f; private final ByteBufferReleaser releaser = new ByteBufferReleaser(options.getAllocator()); @@ -2036,9 +2041,10 @@ public ChunkListBuilder(long rowCount) { this.rowCount = rowCount; } - void add(ChunkDescriptor descriptor, List buffers, SeekableInputStream f) { - map.computeIfAbsent(descriptor, d -> new ChunkData()).buffers.addAll(buffers); - lastDescriptor = descriptor; + void add(ChunkDescriptor descriptor, List buffers, SeekableInputStream f, long tailEndPos) { + ChunkData data = map.computeIfAbsent(descriptor, d -> new ChunkData()); + data.buffers.addAll(buffers); + data.tailEndPos = tailEndPos; this.f = f; } @@ -2056,9 +2062,10 @@ List build() { for (Entry entry : entries) { ChunkDescriptor descriptor = entry.getKey(); ChunkData data = entry.getValue(); - if (descriptor.equals(lastDescriptor)) { - // because of a bug, the last chunk might be larger than descriptor.size - chunks.add(new WorkaroundChunk(lastDescriptor, data.buffers, f, data.offsetIndex, rowCount)); + if (data.tailEndPos >= 0) { + // because of a bug, the last chunk of a consecutive part list might be larger than descriptor.size + chunks.add(new WorkaroundChunk( + descriptor, data.buffers, f, data.tailEndPos, data.offsetIndex, rowCount)); } else { chunks.add(new Chunk(descriptor, data.buffers, data.offsetIndex, rowCount)); } @@ -2311,19 +2318,24 @@ public BytesInput readAsBytesInput(int size) throws IOException { private class WorkaroundChunk extends Chunk { private final SeekableInputStream f; + // where the bytes still missing at the end of this chunk are in the file + private long missingBytesPos; /** * @param descriptor the descriptor of the chunk - * @param f the file stream positioned at the end of this chunk + * @param f the file stream, it is repositioned to read the missing bytes + * @param endPos the position following the last byte of this chunk according to the metadata */ private WorkaroundChunk( ChunkDescriptor descriptor, List buffers, SeekableInputStream f, + long endPos, OffsetIndex offsetIndex, long rowCount) { super(descriptor, buffers, offsetIndex, rowCount); this.f = f; + this.missingBytesPos = endPos; } protected PageHeader readPageHeader() throws IOException { @@ -2339,15 +2351,17 @@ protected PageHeader readPageHeader() throws IOException { // if the last page is smaller than this, the page header itself is truncated in the buffer. stream.reset(); // resetting the buffer to the position before we got the error LOG.info("completing the column chunk to read the page header"); + f.seek(missingBytesPos); pageHeader = Util.readPageHeader( new SequenceInputStream(stream, f)); // trying again from the buffer + remainder of the stream. + missingBytesPos = f.getPos(); } return pageHeader; } public BytesInput readAsBytesInput(int size) throws IOException { int available = stream.available(); - if (size > available) { + if (size > available && size - available <= MAX_MISSING_DICTIONARY_HEADER_BYTES) { // this is to workaround a bug where the compressedLength // of the chunk is missing the size of the header of the dictionary // to allow reading older files (using dictionary) we need this. @@ -2358,7 +2372,11 @@ public BytesInput readAsBytesInput(int size) throws IOException { List streamBuffers = stream.sliceBuffers(available); ByteBuffer lastBuffer = ByteBuffer.allocate(missingBytes); + // f is not at the end of this chunk after a vectored read or after reading the other part lists + f.seek(missingBytesPos); f.readFully(lastBuffer); + lastBuffer.flip(); + missingBytesPos += missingBytes; List buffers = new ArrayList<>(streamBuffers.size() + 1); buffers.addAll(streamBuffers); @@ -2474,7 +2492,7 @@ public void readAll(SeekableInputStream f, ChunkListBuilder builder) throws IOEx BenchmarkCounter.incrementBytesRead(length); ByteBufferInputStream stream = ByteBufferInputStream.wrap(buffers); for (final ChunkDescriptor descriptor : chunks) { - builder.add(descriptor, stream.sliceBuffers(descriptor.size), f); + builder.add(descriptor, stream.sliceBuffers(descriptor.size), f, tailEndPos(descriptor)); } } @@ -2533,10 +2551,18 @@ public void readFromVectoredRanges( } ByteBufferInputStream stream = ByteBufferInputStream.wrap(buffers); for (ChunkDescriptor descriptor : chunks) { - builder.add(descriptor, stream.sliceBuffers(descriptor.size), f); + builder.add(descriptor, stream.sliceBuffers(descriptor.size), f, tailEndPos(descriptor)); } } + /** + * @param descriptor a chunk of this list + * @return the position following the last byte of these chunks if the chunk is the last one, otherwise -1 + */ + private long tailEndPos(ChunkDescriptor descriptor) { + return descriptor == chunks.get(chunks.size() - 1) ? endPos() : -1; + } + /** * @return the position following the last byte of these chunks */ diff --git a/parquet-hadoop/src/test/java/org/apache/parquet/hadoop/TestOldWriterDictionaryChunks.java b/parquet-hadoop/src/test/java/org/apache/parquet/hadoop/TestOldWriterDictionaryChunks.java new file mode 100644 index 0000000000..f23debf5cf --- /dev/null +++ b/parquet-hadoop/src/test/java/org/apache/parquet/hadoop/TestOldWriterDictionaryChunks.java @@ -0,0 +1,165 @@ +/* + * 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 static org.assertj.core.api.Assertions.assertThat; + +import java.io.IOException; +import java.net.URISyntaxException; +import java.util.ArrayList; +import java.util.List; +import org.apache.hadoop.conf.Configuration; +import org.apache.hadoop.fs.Path; +import org.apache.parquet.example.data.Group; +import org.apache.parquet.hadoop.api.ReadSupport; +import org.apache.parquet.hadoop.example.GroupReadSupport; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.ValueSource; + +/** + * Reads nation.dict-malformed.parquet of apache/parquet-testing. An old parquet-mr wrote it without the size of the + * dictionary page header in the length of the column chunks, so the chunks of name and comment_col are 15 bytes + * shorter than their pages and the missing bytes have to be read from the file after the end of the chunk. + */ +public class TestOldWriterDictionaryChunks { + + private static final Path FILE = createPathFromCP("/nation.dict-malformed.parquet"); + + private static final int[] REGIONS = {0, 1, 1, 1, 4, 0, 3, 3, 2, 2, 4, 4, 2, 4, 0, 0, 0, 1, 2, 3, 4, 2, 3, 3, 1}; + + private static final String[] NAMES = { + "ALGERIA", + "ARGENTINA", + "BRAZIL", + "CANADA", + "EGYPT", + "ETHIOPIA", + "FRANCE", + "GERMANY", + "INDIA", + "INDONESIA", + "IRAN", + "IRAQ", + "JAPAN", + "JORDAN", + "KENYA", + "MOROCCO", + "MOZAMBIQUE", + "PERU", + "CHINA", + "ROMANIA", + "SAUDI ARABIA", + "VIETNAM", + "RUSSIA", + "UNITED KINGDOM", + "UNITED STATES", + }; + + private static final String[] COMMENTS = { + " haggle. carefully final deposits detect slyly agai", + "al foxes promise slyly according to the regular accounts. bold requests alon", + "y alongside of the pending deposits. carefully special packages are about the ironic forges. slyly special ", + "eas hang ironic, silent packages. slyly regular packages are furiously over the tithes. fluffily bold", + "y above the carefully unusual theodolites. final dugouts are quickly across the furiously regular d", + "ven packages wake quickly. regu", + "refully final requests. regular, ironi", + "l platelets. regular accounts x-ray: unusual, regular acco", + "ss excuses cajole slyly across the packages. deposits print aroun", + " slyly express asymptotes. regular deposits haggle slyly. carefully ironic hockey players sleep blithely. carefull", + "efully alongside of the slyly final dependencies. ", + "nic deposits boost atop the quickly final requests? quickly regula", + "ously. final, express gifts cajole a", + "ic deposits are blithely about the carefully regular pa", + " pending excuses haggle furiously deposits. pending, express pinto beans wake fluffily past t", + "rns. blithely bold courts among the closely regular packages use furiously bold platelets?", + "s. ironic, unusual asymptotes wake blithely r", + "platelets. blithely pending dependencies use fluffily across the even pinto beans. carefully silent accoun", + "c dependencies. furiously express notornis sleep slyly regular accounts. ideas sleep. depos", + "ular asymptotes are about the furious multipliers. express dependencies nag above the ironically ironic account", + "ts. silent requests haggle. closely express packages sleep across the blithely", + "hely enticingly express accounts. even, final ", + " requests against the platelets use never according to the quickly regular pint", + "eans boost carefully special requests. accounts are. carefull", + "y final packages. slow foxes cajole quickly. quickly silent platelets breach ironic accounts. unusual pinto be", + }; + + private static Path createPathFromCP(String path) { + try { + return new Path( + TestOldWriterDictionaryChunks.class.getResource(path).toURI()); + } catch (URISyntaxException e) { + throw new RuntimeException(e); + } + } + + private static List read(boolean vectoredIo, String requestedSchema) throws IOException { + Configuration conf = new Configuration(); + conf.setBoolean(ParquetInputFormat.HADOOP_VECTORED_IO_ENABLED, vectoredIo); + if (requestedSchema != null) { + conf.set(ReadSupport.PARQUET_READ_SCHEMA, requestedSchema); + } + List rows = new ArrayList<>(); + try (ParquetReader reader = ParquetReader.builder(new GroupReadSupport(), FILE) + .withConf(conf) + .build()) { + Group row; + while ((row = reader.read()) != null) { + rows.add(row); + } + } + return rows; + } + + @ParameterizedTest + @ValueSource(booleans = {true, false}) + public void testAllColumns(boolean vectoredIo) throws IOException { + List rows = read(vectoredIo, null); + + assertThat(rows).hasSize(25); + for (int i = 0; i < rows.size(); i++) { + Group row = rows.get(i); + assertThat(row.getInteger("nation_key", 0)).isEqualTo(i); + assertThat(row.getString("name", 0)).isEqualTo(NAMES[i]); + assertThat(row.getInteger("region_key", 0)).isEqualTo(REGIONS[i]); + assertThat(row.getString("comment_col", 0)).isEqualTo(COMMENTS[i]); + } + } + + @ParameterizedTest + @ValueSource(booleans = {true, false}) + public void testNameOnly(boolean vectoredIo) throws IOException { + List rows = read(vectoredIo, "message m { optional binary name; }"); + + assertThat(rows).hasSize(25); + for (int i = 0; i < rows.size(); i++) { + assertThat(rows.get(i).getString("name", 0)).isEqualTo(NAMES[i]); + } + } + + @ParameterizedTest + @ValueSource(booleans = {true, false}) + public void testCommentOnly(boolean vectoredIo) throws IOException { + List rows = read(vectoredIo, "message m { optional binary comment_col; }"); + + assertThat(rows).hasSize(25); + for (int i = 0; i < rows.size(); i++) { + assertThat(rows.get(i).getString("comment_col", 0)).isEqualTo(COMMENTS[i]); + } + } +} diff --git a/parquet-hadoop/src/test/resources/nation.dict-malformed.parquet b/parquet-hadoop/src/test/resources/nation.dict-malformed.parquet new file mode 100644 index 0000000000..5008ac0b22 Binary files /dev/null and b/parquet-hadoop/src/test/resources/nation.dict-malformed.parquet differ