DeltaByteArrayReader fails with ArrayIndexOutOfBoundsException when moving across pages

[asfimport]

DeltaByteArrayReader.readBytes() fails with ArrayIndexOutOfBoundsException soon after it has processed a new page via initFromPage(). This issue can be reproduced by trying to read a Binary column that is encoded using delta byte array and spans multiple pages.

This is happening because ColumnReaderImpl.initDataReader() creates a new ValueReader every time a new page is processed (see this.dataColumn = dataEncoding.getValuesReader(path, VALUES)). The DeltaByteArrayReader is stateful and needs to remember the previous Binary value that was read across pages. When a new DeltaByteArrayReader is created, this information is lost.

Reporter: Alosh Bennett

Related issues:

• ArrayIndexOutOfBoundsException with Parquet write version v2 (is duplicated by)

_{Note: This issue was originally created as PARQUET-244. Please see the migration documentation for further details.}

[asfimport]

Alosh Bennett:
Stacktrace:

Exception in thread "main" parquet.io.ParquetDecodingException: Can not read value at 94400 in block 0 in file file:/home/abennett/parquet/bug/sample.par
	at parquet.hadoop.InternalParquetRecordReader.nextKeyValue(InternalParquetRecordReader.java:228)
	at parquet.hadoop.ParquetReader.read(ParquetReader.java:129)
	at in.alosh.garage.parquet.SimplestParquetRead$.main(SimplestParquetRead.scala:26)
	at in.alosh.garage.parquet.SimplestParquetRead.main(SimplestParquetRead.scala)
	at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
	at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
	at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
	at java.lang.reflect.Method.invoke(Method.java:483)
	at com.intellij.rt.execution.application.AppMain.main(AppMain.java:134)
Caused by: parquet.io.ParquetDecodingException: Can't read value in column [message] BINARY at value 94401 out of 100000, 1 out of 5600 in currentPage. repetition level: 0, definition level: 0
	at parquet.column.impl.ColumnReaderImpl.readValue(ColumnReaderImpl.java:464)
	at parquet.column.impl.ColumnReaderImpl.writeCurrentValueToConverter(ColumnReaderImpl.java:365)
	at parquet.io.RecordReaderImplementation.read(RecordReaderImplementation.java:418)
	at parquet.hadoop.InternalParquetRecordReader.nextKeyValue(InternalParquetRecordReader.java:209)
	... 8 more
Caused by: java.lang.ArrayIndexOutOfBoundsException
	at parquet.column.values.deltastrings.DeltaByteArrayReader.readBytes(DeltaByteArrayReader.java:70)
	at parquet.column.impl.ColumnReaderImpl$2$6.read(ColumnReaderImpl.java:308)
	at parquet.column.impl.ColumnReaderImpl.readValue(ColumnReaderImpl.java:459)
	... 11 more

[asfimport]

Alosh Bennett:
Code to reproduce the bug. This will create a Parquet file that should contain a UTF-8 column split across multiple pages. The bug can be reproduced by reading this file from a sample program or via spark-shell.

    public static void main(String[] args) throws IOException {
        String parquetFile = "file:///home/abennett/parquet/bug/sample.par";
        String schema = "message Document { required binary message (UTF8); }";
        WriteSupport<Document> writeSupport = new WriteSupport<Document>() {
            RecordConsumer rec;
            @Override
            public WriteContext init(Configuration configuration) {
                return new WriteContext(MessageTypeParser.parseMessageType(schema), new HashMap<>());
            }

            @Override
            public void prepareForWrite(RecordConsumer recordConsumer) {
                rec = recordConsumer;
            }

            @Override
            public void write(Document document) {
                rec.startMessage();
                rec.startField("message", 0);
                rec.addBinary(Binary.fromString(document.message));
                rec.endField("message", 0);
                rec.endMessage();

            }
        };
        ParquetWriter<Document> writer = new ParquetWriter<Document>(new Path(parquetFile), writeSupport, CompressionCodecName.SNAPPY,
                ParquetWriter.DEFAULT_BLOCK_SIZE, ParquetWriter.DEFAULT_PAGE_SIZE,
                ParquetWriter.DEFAULT_PAGE_SIZE, true, false, ParquetProperties.WriterVersion.PARQUET_2_0);
        Document doc = new Document();
        for(int i = 0; i < 100000; i++) {
            doc.message = UUID.randomUUID().toString();
            writer.write(doc);
        }
        writer.close();
    }

    private static class Document {
        String message;
    }