sankalpsthakur · Oct 21, 2026
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
@@ -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<ByteBuffer> 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<ChunkDescriptor, ChunkData> 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<ByteBuffer> buffers, SeekableInputStream f) {
-      map.computeIfAbsent(descriptor, d -> new ChunkData()).buffers.addAll(buffers);
-      lastDescriptor = descriptor;
+    void add(ChunkDescriptor descriptor, List<ByteBuffer> 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<Chunk> build() {
       for (Entry<ChunkDescriptor, ChunkData> 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<ByteBuffer> 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<ByteBuffer> 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<ByteBuffer> 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
@@ -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<Group> 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<Group> rows = new ArrayList<>();
+    try (ParquetReader<Group> 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<Group> 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<Group> 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<Group> 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