[jira] [Created] (FLINK-17804) Follow the spec when decoding Parquet logical DECIMAL type

["Sergii Mikhtoniuk (Jira)" <ji...@apache.org>]

Sergii Mikhtoniuk created FLINK-17804:
-----------------------------------------

             Summary: Follow the spec when decoding Parquet logical DECIMAL type
                 Key: FLINK-17804
                 URL: https://issues.apache.org/jira/browse/FLINK-17804
             Project: Flink
          Issue Type: Bug
          Components: Formats (JSON, Avro, Parquet, ORC, SequenceFile)
    Affects Versions: 1.10.1
            Reporter: Sergii Mikhtoniuk


When reading a Parquet file (produced by Spark 2.4.0 with default configuration) Flink's {{ParquetRowInputFormat}} fails with {{NumberFormatException}}.

After debugging this it seems that Flink doesn't follow the Parquet spec on [encoding DECIMAL logical type|https://github.com/apache/parquet-format/blob/master/LogicalTypes.md#decimal]

The Parquet schema for this field is:
{code}
optional fixed_len_byte_array(9) price_usd (DECIMAL(19,4));
{code}

If I understand the spec correctly, it says that the value should contain a binary representation of an unscaled decimal. Flink's [RowConverter|https://github.com/apache/flink/blob/release-1.10.1/flink-formats/flink-parquet/src/main/java/org/apache/flink/formats/parquet/utils/RowConverter.java#L202] however treats it as a base-10 UTF-8 string.

What Flink essentially is doing:
{code}
val binary = Array[Byte](0, 0, 0, 0, 0, 11, -97, 118, -64)
val decimal = new java.math.BigDecimal(new String(binary, "UTF-8").toCharArray)
{code}

What I think spec suggests:
{code}
val binary = Array[Byte](0, 0, 0, 0, 0, 11, -97, 118, -64)
val unscaled = new java.math.BigInteger(binary)
val decimal = new java.math.BigDecimal(unscaled)
{code}

Error stacktrace:
 {code}
java.lang.NumberFormatException
	at java.math.BigDecimal.<init>(BigDecimal.java:497)
	at java.math.BigDecimal.<init>(BigDecimal.java:383)
	at java.math.BigDecimal.<init>(BigDecimal.java:680)
	at org.apache.flink.formats.parquet.utils.RowConverter$RowPrimitiveConverter.addBinary(RowConverter.java:202)
	at org.apache.parquet.column.impl.ColumnReaderImpl$2$6.writeValue(ColumnReaderImpl.java:317)
	at org.apache.parquet.column.impl.ColumnReaderImpl.writeCurrentValueToConverter(ColumnReaderImpl.java:367)
	at org.apache.parquet.io.RecordReaderImplementation.read(RecordReaderImplementation.java:406)
	at org.apache.flink.formats.parquet.utils.ParquetRecordReader.readNextRecord(ParquetRecordReader.java:235)
	at org.apache.flink.formats.parquet.utils.ParquetRecordReader.reachEnd(ParquetRecordReader.java:207)
	at org.apache.flink.formats.parquet.ParquetInputFormat.reachedEnd(ParquetInputFormat.java:233)
	at org.apache.flink.streaming.api.functions.source.ContinuousFileReaderOperator$SplitReader.run(ContinuousFileReaderOperator.java:327)
{code}

Thanks!



--
This message was sent by Atlassian Jira
(v8.3.4#803005)