[fix][sql] Fix the decimal type error convert in json schema (#15687)

shibd · web-flow · commit 0c6e2ca24fd3 · 2022-05-26T08:24:14.000+08:00
### Motivation

In the current sql implementation, If using `JSON` schema and querying for decimal type, there will be the following two errors:

1. The data type is displayed as varchar.
2. Loss of precision because scientific notation is used to display.

```
presto&gt; select bigdecimal, typeof(bigdecimal) as devimal_type from pulsar."public/default".test_avro2;
      bigdecimal       | devimal_type
-----------------------+--------------
 1.2345678912345678E36 | varchar
 1.2345678912345678E36 | varchar
(2 rows)
```
The original data is: `1234567891234567891234567891234567.89`

### Modifications
-  When getting jsonNode,  use `BIG_DECIMAL` instead of float and double.
- `PulsarJsonFieldDecoder` increases the processing of Decimal types
diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/schema/generic/GenericJsonReader.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/schema/generic/GenericJsonReader.java
@@ -19,8 +19,10 @@
 package org.apache.pulsar.client.impl.schema.generic;
 
 import static java.nio.charset.StandardCharsets.UTF_8;
+import com.fasterxml.jackson.databind.DeserializationFeature;
 import com.fasterxml.jackson.databind.JsonNode;
 import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.ObjectReader;
 import java.io.IOException;
 import java.io.InputStream;
 import java.util.List;
@@ -34,16 +36,13 @@
 
 public class GenericJsonReader implements SchemaReader<GenericRecord> {
 
-    private final ObjectMapper objectMapper;
+    private final ObjectReader objectReader;
     private final byte[] schemaVersion;
     private final List<Field> fields;
     private SchemaInfo schemaInfo;
 
     public GenericJsonReader(List<Field> fields, SchemaInfo schemaInfo){
-        this.fields = fields;
-        this.schemaVersion = null;
-        this.objectMapper = new ObjectMapper();
-        this.schemaInfo = schemaInfo;
+        this(null, fields, schemaInfo);
     }
 
     public GenericJsonReader(List<Field> fields){
@@ -55,16 +54,17 @@ public GenericJsonReader(byte[] schemaVersion, List<Field> fields){
     }
 
     public GenericJsonReader(byte[] schemaVersion, List<Field> fields, SchemaInfo schemaInfo){
-        this.objectMapper = new ObjectMapper();
         this.fields = fields;
         this.schemaVersion = schemaVersion;
         this.schemaInfo = schemaInfo;
+        ObjectMapper objectMapper = new ObjectMapper();
+        this.objectReader = objectMapper.reader().with(DeserializationFeature.USE_BIG_DECIMAL_FOR_FLOATS);
     }
 
     @Override
     public GenericJsonRecord read(byte[] bytes, int offset, int length) {
         try {
-            JsonNode jn = objectMapper.readTree(new String(bytes, offset, length, UTF_8));
+            JsonNode jn = objectReader.readTree(new String(bytes, offset, length, UTF_8));
             return new GenericJsonRecord(schemaVersion, fields, jn, schemaInfo);
         } catch (IOException ioe) {
             throw new SchemaSerializationException(ioe);
@@ -74,7 +74,7 @@ public GenericJsonRecord read(byte[] bytes, int offset, int length) {
     @Override
     public GenericRecord read(InputStream inputStream) {
         try {
-            JsonNode jn = objectMapper.readTree(inputStream);
+            JsonNode jn = objectReader.readTree(inputStream);
             return new GenericJsonRecord(schemaVersion, fields, jn, schemaInfo);
         } catch (IOException ioe) {
             throw new SchemaSerializationException(ioe);
diff --git a/pulsar-sql/presto-pulsar/src/main/java/org/apache/pulsar/sql/presto/decoder/json/PulsarJsonFieldDecoder.java b/pulsar-sql/presto-pulsar/src/main/java/org/apache/pulsar/sql/presto/decoder/json/PulsarJsonFieldDecoder.java
@@ -57,6 +57,8 @@
 import io.prestosql.spi.type.BigintType;
 import io.prestosql.spi.type.BooleanType;
 import io.prestosql.spi.type.DateType;
+import io.prestosql.spi.type.DecimalType;
+import io.prestosql.spi.type.Decimals;
 import io.prestosql.spi.type.DoubleType;
 import io.prestosql.spi.type.IntegerType;
 import io.prestosql.spi.type.MapType;
@@ -69,6 +71,7 @@
 import io.prestosql.spi.type.Type;
 import io.prestosql.spi.type.VarbinaryType;
 import io.prestosql.spi.type.VarcharType;
+import java.math.BigInteger;
 import java.util.Iterator;
 import java.util.List;
 import java.util.Map;
@@ -118,6 +121,9 @@ private static Pair<Long, Long> getNumRangeByType(Type type) {
     }
 
     private boolean isSupportedType(Type type) {
+        if (type instanceof DecimalType) {
+            return true;
+        }
         if (isVarcharType(type)) {
             return true;
         }
@@ -226,6 +232,13 @@ public static long getLong(JsonNode value, Type type, String columnName, long mi
                     return floatToIntBits((Float) parseFloat(value.asText()));
                 }
 
+                // If it is decimalType, need to eliminate the decimal point,
+                // and give it to presto to set the decimal point
+                if (type instanceof DecimalType) {
+                    String decimalLong = value.asText().replace(".", "");
+                    return Long.valueOf(decimalLong);
+                }
+
                 long longValue;
                 if (value.isIntegralNumber() && !value.isBigInteger()) {
                     longValue = value.longValue();
@@ -265,6 +278,15 @@ public static double getDouble(JsonNode value, Type type, String columnName) {
 
         private static Slice getSlice(JsonNode value, Type type, String columnName) {
             String textValue = value.isValueNode() ? value.asText() : value.toString();
+
+            // If it is decimalType, need to eliminate the decimal point,
+            // and give it to presto to set the decimal point
+            if (type instanceof DecimalType) {
+                textValue = textValue.replace(".", "");
+                BigInteger bigInteger = new BigInteger(textValue);
+                return Decimals.encodeUnscaledValue(bigInteger);
+            }
+
             Slice slice = utf8Slice(textValue);
             if (isVarcharType(type)) {
                 slice = truncateToLength(slice, type);
diff --git a/pulsar-sql/presto-pulsar/src/main/java/org/apache/pulsar/sql/presto/decoder/json/PulsarJsonRowDecoderFactory.java b/pulsar-sql/presto-pulsar/src/main/java/org/apache/pulsar/sql/presto/decoder/json/PulsarJsonRowDecoderFactory.java
@@ -33,6 +33,7 @@
 import io.prestosql.spi.type.ArrayType;
 import io.prestosql.spi.type.BigintType;
 import io.prestosql.spi.type.BooleanType;
+import io.prestosql.spi.type.DecimalType;
 import io.prestosql.spi.type.DoubleType;
 import io.prestosql.spi.type.IntegerType;
 import io.prestosql.spi.type.RealType;
@@ -128,11 +129,13 @@ private Type parseJsonPrestoType(String fieldname, Schema schema) {
                                 + "please check the schema or report the bug.", fieldname));
             case FIXED:
             case BYTES:
-                // In the current implementation, since JsonSchema is generated by Avro,
-                // there may exist LogicalTypes.Decimal.
-                // Mapping decimalType with varcharType in JsonSchema.
+                //  When the precision <= 0, throw Exception.
+                //  When the precision > 0 and <= 18, use ShortDecimalType. and mapping Long
+                //  When the precision > 18 and <= 36, use LongDecimalType. and mapping Slice
+                //  When the precision > 36, throw Exception.
                 if (logicalType instanceof LogicalTypes.Decimal) {
-                    return createUnboundedVarcharType();
+                    LogicalTypes.Decimal decimal = (LogicalTypes.Decimal) logicalType;
+                    return DecimalType.createDecimalType(decimal.getPrecision(), decimal.getScale());
                 }
                 return VarbinaryType.VARBINARY;
             case INT:
diff --git a/pulsar-sql/presto-pulsar/src/test/java/org/apache/pulsar/sql/presto/decoder/avro/TestAvroDecoder.java b/pulsar-sql/presto-pulsar/src/test/java/org/apache/pulsar/sql/presto/decoder/avro/TestAvroDecoder.java
@@ -89,8 +89,6 @@ public void testPrimitiveType() {
         message.longField = 222L;
         message.timestampField = System.currentTimeMillis();
         message.enumField = DecoderTestMessage.TestEnum.TEST_ENUM_1;
-        message.decimalField = BigDecimal.valueOf(2233, 2);
-        message.longDecimalField = new BigDecimal("1234567891234567891234567891.23");
 
         LocalTime now = LocalTime.now(ZoneId.systemDefault());
         message.timeField = now.toSecondOfDay() * 1000;
@@ -130,6 +128,17 @@ public void testPrimitiveType() {
         PulsarColumnHandle enumFieldColumnHandle = new PulsarColumnHandle(getPulsarConnectorId().toString(),
                 "enumField", VARCHAR, false, false, "enumField", null, null, PulsarColumnHandle.HandleKeyValueType.NONE);
         checkValue(decodedRow, enumFieldColumnHandle, message.enumField.toString());
+    }
+
+    @Test
+    public void testDecimal() {
+        DecoderTestMessage message = new DecoderTestMessage();
+        message.decimalField = BigDecimal.valueOf(2233, 2);
+        message.longDecimalField = new BigDecimal("1234567891234567891234567891.23");
+
+        ByteBuf payload = io.netty.buffer.Unpooled
+                .copiedBuffer(schema.encode(message));
+        Map<DecoderColumnHandle, FieldValueProvider> decodedRow = pulsarRowDecoder.decodeRow(payload).get();
 
         PulsarColumnHandle decimalFieldColumnHandle = new PulsarColumnHandle(getPulsarConnectorId().toString(),
                 "decimalField", DecimalType.createDecimalType(4, 2), false, false, "decimalField", null, null, PulsarColumnHandle.HandleKeyValueType.NONE);
diff --git a/pulsar-sql/presto-pulsar/src/test/java/org/apache/pulsar/sql/presto/decoder/json/TestJsonDecoder.java b/pulsar-sql/presto-pulsar/src/test/java/org/apache/pulsar/sql/presto/decoder/json/TestJsonDecoder.java
@@ -24,6 +24,7 @@
 import io.prestosql.decoder.FieldValueProvider;
 import io.prestosql.spi.PrestoException;
 import io.prestosql.spi.type.*;
+import java.math.BigDecimal;
 import org.apache.pulsar.client.impl.schema.JSONSchema;
 import org.apache.pulsar.client.impl.schema.generic.GenericJsonRecord;
 import org.apache.pulsar.client.impl.schema.generic.GenericJsonSchema;
@@ -119,6 +120,25 @@ public void testPrimitiveType() {
 
     }
 
+    @Test
+    public void testDecimal() {
+        DecoderTestMessage message = new DecoderTestMessage();
+        message.decimalField = BigDecimal.valueOf(2233, 2);
+        message.longDecimalField = new BigDecimal("1234567891234567891234567891.23");
+
+        ByteBuf payload = io.netty.buffer.Unpooled
+                .copiedBuffer(schema.encode(message));
+        Map<DecoderColumnHandle, FieldValueProvider> decodedRow = pulsarRowDecoder.decodeRow(payload).get();
+
+        PulsarColumnHandle decimalFieldColumnHandle = new PulsarColumnHandle(getPulsarConnectorId().toString(),
+                "decimalField", DecimalType.createDecimalType(4, 2), false, false, "decimalField", null, null, PulsarColumnHandle.HandleKeyValueType.NONE);
+        checkValue(decodedRow, decimalFieldColumnHandle, message.decimalField);
+
+        PulsarColumnHandle longDecimalFieldColumnHandle = new PulsarColumnHandle(getPulsarConnectorId().toString(),
+                "longDecimalField", DecimalType.createDecimalType(30, 2), false, false, "longDecimalField", null, null, PulsarColumnHandle.HandleKeyValueType.NONE);
+        checkValue(decodedRow, longDecimalFieldColumnHandle, message.longDecimalField);
+    }
+
     @Test
     public void testArray() {
         DecoderTestMessage message = new DecoderTestMessage();
