From 83ca778966033f59039824d5c9db2d341b872149 Mon Sep 17 00:00:00 2001 From: Sankalp Thakur Date: Wed, 21 Oct 2026 08:01:08 +0400 Subject: [PATCH] GH-3336: Read dictionary pages of old parquet-mr files with default settings parquet-mr up to 1.2.8 left the dictionary page header out of the size of the column chunk (PARQUET-816), so the last bytes of such a chunk are not in the buffer that was read for it. ParquetFileReader completes the chunk from the file stream, but only for the last chunk of the row group, and only assuming that the stream is still positioned at the end of that chunk. All part lists are read before the pages are decoded and a vectored read does not move the stream at all, so the read failed with an EOFException or read the wrong bytes. The completed bytes were also never flipped. Every chunk that ends a consecutive part list is now completed from the file at the end offset of its part list, in the sequential and in the vectored path. The completion is limited to the size of a page header. --- .../parquet/hadoop/ParquetFileReader.java | 48 +++-- .../hadoop/TestOldWriterDictionaryChunks.java | 165 ++++++++++++++++++ .../resources/nation.dict-malformed.parquet | Bin 0 -> 2850 bytes 3 files changed, 202 insertions(+), 11 deletions(-) create mode 100644 parquet-hadoop/src/test/java/org/apache/parquet/hadoop/TestOldWriterDictionaryChunks.java create mode 100644 parquet-hadoop/src/test/resources/nation.dict-malformed.parquet 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 0000000000000000000000000000000000000000..5008ac0b22b622cbcf37d5e286c749324fc26535 GIT binary patch literal 2850 zcmb7G&2Af26n0F~Hfi|Brja6Y7Z7#fSP&MJN~pStO)_mfb{z*P0aRU&ubpYebEosu z#$t!MKs*8qR@ng0z>-bZJOz)yito(KnX$tHVfi>WbIxG@;>Aa{l8rMmA*#^4*%p<9v!s%mL$)zf6%rEmMtmdUSFPBN3`0PwrojCrG0SP z8&DqY!J*adS*hZeeQEEv$rT)RqI~1aSa*QmfStqtNXt%^8%RXpzJJiDb$?-rA+Q&fUUhN*ivpf&a%BuK5 z{`mRbU;g;%yN_R4zx_`AM1%hWO+u7}GN%tAOF}4<&~p-AV_AHzvw@jG_OyotUwwdY zj^?L$B+k`j;_0SvRH(;^=gq{}^_3?^IxvwNM_9z#iAChiu%{-FJxy?N7zR*W?u2S~U*sh;78E3`k*L0Ok_ezU>Z$QcY7|9qH=8BN zg3v|#RS1%VaFRo&FN09Sp^9eHAU0{6I3cD*PTtgucrh%vi+`@qD4k5uHvcPxApWaG#x`#T;Ba@bX|{ns~7rcrKXUEJ(Ca ztFIKieadgbAGMu=s0tm2k!s9T~s%| zIb53J9kEvZsCsA3SFvmSCpZ8rwRJpL_SID9d+5u;-fhQBrvT@PW4uazcVlzoX#3V? zt-7(YG4H5`z53rv_-^CHZu|S{W^J8RQrGF`%a!`e6>MPl>LxoVqs=+E!rR#P&#SqI k`2

tF8Lg7WV0tw7y%pyB4SrY>0QJVdXCX(Zk#EKmWZ<*#H0l literal 0 HcmV?d00001