[parquet-java 1.18.0] data-corruption for reading NUMERIC arrays

[yimingli-vmware]

Describe the bug, including details regarding any error messages, version, and platform.

We have an existing passing test case to write a parquet numeric array and read it back. After upgrading parquet-java from 1.17.1 to 1.18.0, the test failed because of the read mismatch.

For example:

# what we write via a parquet writer
{"12345678901234567890.123456789012345678", "12345678901234567890.123456789012345678"},
{"22.2345", "22.2345"},

# what we read from a parquet reader
row 0: 12345678901234567890.123456789012345678 12345678901234567890.123456789012345678
row 1: 12345678901234567890.123456789012345678 22.234500000000000000 <- the first record is wrong

I can reproduce this issue with the following test code:

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.parquet.example.data.Group;
import org.apache.parquet.example.data.simple.SimpleGroupFactory;
import org.apache.parquet.hadoop.ParquetWriter;
import org.apache.parquet.hadoop.example.ExampleParquetWriter;
import org.apache.parquet.hadoop.example.GroupWriteSupport;
import org.apache.parquet.hadoop.metadata.CompressionCodecName;
import org.apache.parquet.io.api.Binary;
import org.apache.parquet.schema.MessageType;
import org.apache.parquet.schema.MessageTypeParser;
import org.apache.parquet.hadoop.ParquetReader;
import org.apache.parquet.hadoop.example.GroupReadSupport;

import java.math.BigDecimal;
import java.math.BigInteger;
import java.util.Arrays;

public class Repro {
    public static void main(String[] args) throws Exception {
        String schemaStr = "message test {\n" +
                "  optional group numeric_array (LIST) {\n" +
                "    repeated group list {\n" +
                "      optional fixed_len_byte_array(16) element (DECIMAL(38,18));\n" +
                "    }\n" +
                "  }\n" +
                "}";
        MessageType schema = MessageTypeParser.parseMessageType(schemaStr);
        Configuration conf = new Configuration();
        GroupWriteSupport.setSchema(schema, conf);

        Path path = new Path("file:///tmp/parquet-repro/out.parquet");
        path.getFileSystem(conf).delete(path, false);

        // 2 rows, 2 elements per row -- this is the minimal trigger.
        // A single element per row does NOT reproduce the bug.
        String[][] rows = {
            {"12345678901234567890.123456789012345678", "12345678901234567890.123456789012345678"},
            {"22.2345", "22.2345"},
        };

        try (ParquetWriter<Group> writer = ExampleParquetWriter.builder(path)
                .withConf(conf)
                .withType(schema)
                .withCompressionCodec(CompressionCodecName.UNCOMPRESSED)
                .withDictionaryEncoding(false)
                .build()) {
            SimpleGroupFactory factory = new SimpleGroupFactory(schema);
            for (String[] row : rows) {
                Group group = factory.newGroup();
                Group listGroup = group.addGroup(0);
                for (String decVal : row) {
                    Group repeatedGroup = listGroup.addGroup(0);
                    byte[] bytes = toFixedLenBytes(decVal, 18, 16);
                    repeatedGroup.add(0, Binary.fromReusedByteArray(bytes));
                }
                writer.write(group);
            }
        }

        try (ParquetReader<Group> reader = ParquetReader.builder(new GroupReadSupport(), path)
                .withConf(conf)
                .build()) {
            Group g;
            int i = 0;
            while ((g = reader.read()) != null) {
                Group listGroup = g.getGroup(0, 0);
                int count = listGroup.getFieldRepetitionCount(0);
                StringBuilder sb = new StringBuilder("row " + i + ": ");
                for (int j = 0; j < count; j++) {
                    Group repeatedGroup = listGroup.getGroup(0, j);
                    byte[] readBytes = repeatedGroup.getBinary(0, 0).getBytes();
                    BigDecimal bd = new BigDecimal(new BigInteger(readBytes), 18);
                    sb.append(bd).append(" ");
                }
                System.out.println(sb);
                i++;
            }
        }
    }

    private static byte[] toFixedLenBytes(String decStr, int scale, int len) {
        BigDecimal bd = new BigDecimal(decStr).setScale(scale);
        BigInteger unscaled = bd.unscaledValue();
        byte[] full = unscaled.toByteArray();
        byte[] tgt = new byte[len];
        if (unscaled.signum() == -1) {
            Arrays.fill(tgt, (byte) 0xFF);
        }
        int srcStart = Math.max(0, full.length - len);
        int copyLen = full.length - srcStart;
        System.arraycopy(full, srcStart, tgt, len - copyLen, copyLen);
        return tgt;
    }
}

Component(s)

No response