[feat][sql] Bump Trino version to 368 and fix Decimal breaking change (#20016)

Signed-off-by: tison <wander4096@gmail.com>

Author: tisonkun

pom.xml

@@ -184,7 +184,7 @@ flexible messaging model and an intuitive client API.</description>
     <json-smart.version>2.4.10</json-smart.version>
     <opensearch.version>1.2.4</opensearch.version>
     <elasticsearch-java.version>8.5.2</elasticsearch-java.version>
-    <trino.version>363</trino.version>
+    <trino.version>368</trino.version>
     <debezium.version>1.9.7.Final</debezium.version>
     <debezium.postgresql.version>42.5.0</debezium.postgresql.version>
     <debezium.mysql.version>8.0.30</debezium.mysql.version>

pulsar-sql/pom.xml

@@ -36,7 +36,7 @@
         <okhttp3.version>3.14.9</okhttp3.version>
         <!-- use okio version that matches the okhttp3 version -->
         <okio.version>1.17.2</okio.version>
-        <airlift.version>208</airlift.version>
+        <airlift.version>213</airlift.version>
     </properties>
 
     <dependencyManagement>

pulsar-sql/presto-distribution/LICENSE

@@ -303,28 +303,28 @@ The Apache Software License, Version 2.0
     - bytecode-1.2.jar
   * Airlift
     - aircompressor-0.20.jar
-    - bootstrap-208.jar
-    - concurrent-208.jar
-    - configuration-208.jar
-    - discovery-208.jar
+    - bootstrap-213.jar
+    - concurrent-213.jar
+    - configuration-213.jar
+    - discovery-213.jar
     - discovery-server-1.30.jar
-    - event-208.jar
-    - event-http-208.jar
-    - http-client-208.jar
-    - http-server-208.jar
-    - jmx-208.jar
-    - jmx-http-208.jar
-    - jmx-http-rpc-208.jar
+    - event-213.jar
+    - event-http-213.jar
+    - http-client-213.jar
+    - http-server-213.jar
+    - jmx-213.jar
+    - jmx-http-213.jar
+    - jmx-http-rpc-213.jar
     - joni-2.1.5.3.jar
-    - json-208.jar
-    - log-208.jar
-    - log-manager-208.jar
-    - node-208.jar
+    - json-213.jar
+    - log-213.jar
+    - log-manager-213.jar
+    - node-213.jar
     - parameternames-1.4.jar
-    - security-208.jar
-    - slice-0.39.jar
-    - stats-208.jar
-    - trace-token-208.jar
+    - security-213.jar
+    - slice-0.41.jar
+    - stats-213.jar
+    - trace-token-213.jar
     - units-1.6.jar
    * Apache HTTP Client
     - httpclient-4.5.13.jar
@@ -340,7 +340,9 @@ The Apache Software License, Version 2.0
   * J2ObjC Annotations
     - j2objc-annotations-1.3.jar
   * JSON Web Token Support For The JVM
-    - jjwt-0.9.0.jar
+    - jjwt-api-0.11.1.jar
+    - jjwt-impl-0.11.1.jar
+    - jjwt-jackson-0.11.1.jar
   * Jmxutils
     - jmxutils-1.21.jar
   * LevelDB
@@ -384,18 +386,18 @@ The Apache Software License, Version 2.0
   * Okio
     - okio-1.17.2.jar
   * Trino
-    - trino-array-363.jar
-    - trino-cli-363.jar
-    - trino-client-363.jar
-    - trino-geospatial-toolkit-363.jar
-    - trino-main-363.jar
-    - trino-matching-363.jar
-    - trino-memory-context-363.jar
-    - trino-parser-363.jar
-    - trino-plugin-toolkit-363.jar
-    - trino-server-main-363.jar
-    - trino-spi-363.jar
-    - trino-record-decoder-363.jar
+    - trino-array-368.jar
+    - trino-cli-368.jar
+    - trino-client-368.jar
+    - trino-geospatial-toolkit-368.jar
+    - trino-main-368.jar
+    - trino-matching-368.jar
+    - trino-memory-context-368.jar
+    - trino-parser-368.jar
+    - trino-plugin-toolkit-368.jar
+    - trino-server-main-368.jar
+    - trino-spi-368.jar
+    - trino-record-decoder-368.jar
   * RocksDB JNI
     - rocksdbjni-6.29.4.1.jar
   * SnakeYAML
@@ -456,6 +458,7 @@ The Apache Software License, Version 2.0
     - javassist-3.25.0-GA.jar
   * Java Native Access
     - jna-5.12.1.jar
+    - jna-platform-5.10.0.jar
   * Java Object Layout: Core
     - jol-core-0.2.jar
   * Yahoo Datasketches
@@ -493,7 +496,7 @@ BSD License
  * ANTLR 4 Runtime
     - antlr4-runtime-4.9.2.jar
  * ASM, a very small and fast Java bytecode manipulation framework
-    - asm-6.2.1.jar
+    - asm-9.1.jar
     - asm-analysis-6.2.1.jar
     - asm-tree-6.2.1.jar
     - asm-util-6.2.1.jar
@@ -518,6 +521,8 @@ MIT License
  * ScribeJava
    - scribejava-apis-6.9.0.jar
    - scribejava-core-6.9.0.jar
+ * OSHI
+   - oshi-core-5.8.5.jar
 
 CDDL - 1.0
  * OSGi Resource Locator
@@ -533,16 +538,16 @@ CDDL-1.1 -- licenses/LICENSE-CDDL-1.1.txt
    - hk2-utils-2.6.1.jar
    - aopalliance-repackaged-2.6.1.jar
  * Jersey
-    - jaxrs-208.jar
+    - jaxrs-213.jar
     - jersey-client-2.34.jar
+    - jersey-common-2.34.jar
     - jersey-container-servlet-2.34.jar
     - jersey-container-servlet-core-2.34.jar
     - jersey-entity-filtering-2.34.jar
     - jersey-hk2-2.34.jar
     - jersey-media-json-jackson-2.34.jar
     - jersey-media-multipart-2.34.jar
     - jersey-server-2.34.jar
-    - jersey-common-2.34.jar
  * JAXB
     - jaxb-api-2.3.1.jar
     - jaxb-runtime-2.3.4.jar

pulsar-sql/presto-distribution/pom.xml

@@ -97,6 +97,10 @@
           <groupId>com.google.inject.extensions</groupId>
           <artifactId>guice-multibindings</artifactId>
         </exclusion>
+        <exclusion>
+          <groupId>org.apache.logging.log4j</groupId>
+          <artifactId>log4j-to-slf4j</artifactId>
+        </exclusion>
       </exclusions>
     </dependency>
 

pulsar-sql/presto-pulsar/pom.xml

@@ -41,8 +41,13 @@
         <dependency>
             <groupId>io.airlift</groupId>
             <artifactId>bootstrap</artifactId>
+            <exclusions>
+                <exclusion>
+                    <groupId>org.apache.logging.log4j</groupId>
+                    <artifactId>log4j-to-slf4j</artifactId>
+                </exclusion>
+            </exclusions>
         </dependency>
-
         <dependency>
             <groupId>io.airlift</groupId>
             <artifactId>json</artifactId>

pulsar-sql/presto-pulsar/src/main/java/org/apache/pulsar/sql/presto/decoder/avro/PulsarAvroColumnDecoder.java

@@ -41,8 +41,8 @@
 import io.trino.spi.type.BooleanType;
 import io.trino.spi.type.DateType;
 import io.trino.spi.type.DecimalType;
-import io.trino.spi.type.Decimals;
 import io.trino.spi.type.DoubleType;
+import io.trino.spi.type.Int128;
 import io.trino.spi.type.IntegerType;
 import io.trino.spi.type.MapType;
 import io.trino.spi.type.RealType;
@@ -66,11 +66,14 @@
 import org.apache.avro.generic.GenericRecord;
 
 /**
- * Copy from {@link io.trino.decoder.avro.AvroColumnDecoder} (presto-record-decoder-345)
- * with A little bit pulsar's extensions.
- * 1) support {@link io.trino.spi.type.TimestampType},{@link io.trino.spi.type.DateType}DATE,
- *  * {@link io.trino.spi.type.TimeType}.
+ * Copy from {@link io.trino.decoder.avro.AvroColumnDecoder}
+ * with A little pulsar's extensions.
+ * 1) support date and time types.
+ *  {@link io.trino.spi.type.TimestampType}
+ *  {@link io.trino.spi.type.DateType}
+ *  {@link io.trino.spi.type.TimeType}
  * 2) support {@link io.trino.spi.type.RealType}.
+ * 3) support {@link io.trino.spi.type.DecimalType}.
  */
 public class PulsarAvroColumnDecoder {
     private static final Set<Type> SUPPORTED_PRIMITIVE_TYPES = ImmutableSet.of(
@@ -252,13 +255,6 @@ private static Slice getSlice(Object value, Type type, String columnName) {
             }
         }
 
-        // The returned Slice size must be equals to 18 Byte
-        if (type instanceof DecimalType) {
-            ByteBuffer buffer = (ByteBuffer) value;
-            BigInteger bigInteger = new BigInteger(buffer.array());
-            return Decimals.encodeUnscaledValue(bigInteger);
-        }
-
         throw new TrinoException(DECODER_CONVERSION_NOT_SUPPORTED,
                 format("cannot decode object of '%s' as '%s' for column '%s'",
                         value.getClass(), type, columnName));
@@ -274,6 +270,9 @@ private static Block serializeObject(BlockBuilder builder, Object value, Type ty
         if (type instanceof RowType) {
             return serializeRow(builder, value, type, columnName);
         }
+        if (type instanceof DecimalType && !((DecimalType) type).isShort()) {
+            return serializeLongDecimal(builder, value, type, columnName);
+        }
         serializePrimitive(builder, value, type, columnName);
         return null;
     }
@@ -299,6 +298,22 @@ private static Block serializeList(BlockBuilder parentBlockBuilder, Object value
         return blockBuilder.build();
     }
 
+    private static Block serializeLongDecimal(
+            BlockBuilder parentBlockBuilder, Object value, Type type, String columnName) {
+        final BlockBuilder blockBuilder;
+        if (parentBlockBuilder != null) {
+            blockBuilder = parentBlockBuilder;
+        } else {
+            blockBuilder = type.createBlockBuilder(null, 1);
+        }
+        final ByteBuffer buffer = (ByteBuffer) value;
+        type.writeObject(blockBuilder, Int128.fromBigEndian(buffer.array()));
+        if (parentBlockBuilder == null) {
+            return blockBuilder.getSingleValueBlock(0);
+        }
+        return null;
+    }
+
     private static void serializePrimitive(BlockBuilder blockBuilder, Object value, Type type, String columnName) {
         requireNonNull(blockBuilder, "parent blockBuilder is null");
 

pulsar-sql/presto-pulsar/src/main/java/org/apache/pulsar/sql/presto/decoder/json/PulsarJsonFieldDecoder.java

@@ -29,6 +29,7 @@
 import static java.util.Objects.requireNonNull;
 import com.fasterxml.jackson.databind.JsonNode;
 import com.fasterxml.jackson.databind.node.ArrayNode;
+import com.fasterxml.jackson.databind.node.DecimalNode;
 import com.fasterxml.jackson.databind.node.ObjectNode;
 import com.google.common.collect.ImmutableList;
 import io.airlift.log.Logger;
@@ -45,8 +46,8 @@
 import io.trino.spi.type.BooleanType;
 import io.trino.spi.type.DateType;
 import io.trino.spi.type.DecimalType;
-import io.trino.spi.type.Decimals;
 import io.trino.spi.type.DoubleType;
+import io.trino.spi.type.Int128;
 import io.trino.spi.type.IntegerType;
 import io.trino.spi.type.MapType;
 import io.trino.spi.type.RealType;
@@ -59,21 +60,22 @@
 import io.trino.spi.type.Type;
 import io.trino.spi.type.VarbinaryType;
 import io.trino.spi.type.VarcharType;
-import java.math.BigInteger;
 import java.util.Iterator;
 import java.util.List;
 import java.util.Map;
 import org.apache.commons.lang3.tuple.Pair;
 
 /**
- * Copy from {@link io.trino.decoder.json.DefaultJsonFieldDecoder} (presto-record-decoder-345)
- * with some pulsar's extensions.
+ * Copy from {@link io.trino.decoder.json.DefaultJsonFieldDecoder} with some pulsar's extensions.
  * 1) support {@link io.trino.spi.type.ArrayType}.
  * 2) support {@link io.trino.spi.type.MapType}.
  * 3) support {@link io.trino.spi.type.RowType}.
- * 4) support {@link io.trino.spi.type.TimestampType},{@link io.trino.spi.type.DateType},
- * {@link io.trino.spi.type.TimeType}.
+ * 4) support date and time types.
+ *  {@link io.trino.spi.type.TimestampType}
+ *  {@link io.trino.spi.type.DateType}
+ *  {@link io.trino.spi.type.TimeType}
  * 5) support {@link io.trino.spi.type.RealType}.
+ * 6) support {@link io.trino.spi.type.DecimalType}.
  */
 public class PulsarJsonFieldDecoder
         implements JsonFieldDecoder {
@@ -90,7 +92,6 @@ public PulsarJsonFieldDecoder(DecoderColumnHandle columnHandle) {
         Pair<Long, Long> range = getNumRangeByType(columnHandle.getType());
         minValue = range.getKey();
         maxValue = range.getValue();
-
     }
 
     private static Pair<Long, Long> getNumRangeByType(Type type) {
@@ -221,7 +222,7 @@ public static long getLong(JsonNode value, Type type, String columnName, long mi
                 }
 
                 // If it is decimalType, need to eliminate the decimal point,
-                // and give it to presto to set the decimal point
+                // and give it to trino to set the decimal point
                 if (type instanceof DecimalType) {
                     String decimalLong = value.asText().replace(".", "");
                     return Long.parseLong(decimalLong);
@@ -273,14 +274,6 @@ 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 (type instanceof VarcharType) {
                 slice = truncateToLength(slice, type);
@@ -298,6 +291,9 @@ private Block serializeObject(BlockBuilder builder, Object value, Type type, Str
             if (type instanceof RowType) {
                 return serializeRow(builder, value, type, columnName);
             }
+            if (type instanceof DecimalType && !((DecimalType) type).isShort()) {
+                return serializeLongDecimal(builder, value, type, columnName);
+            }
             serializePrimitive(builder, value, type, columnName);
             return null;
         }
@@ -330,6 +326,27 @@ private Block serializeList(BlockBuilder parentBlockBuilder, Object value, Type
             return blockBuilder.build();
         }
 
+        private static Block serializeLongDecimal(
+                BlockBuilder parentBlockBuilder, Object value, Type type, String columnName) {
+            final BlockBuilder blockBuilder;
+            if (parentBlockBuilder != null) {
+                blockBuilder = parentBlockBuilder;
+            } else {
+                blockBuilder = type.createBlockBuilder(null, 1);
+            }
+
+            assert value instanceof DecimalNode;
+            final DecimalNode node = (DecimalNode) value;
+            // For decimalType, need to eliminate the decimal point,
+            // and give it to trino to set the decimal point
+            type.writeObject(blockBuilder, Int128.valueOf(node.asText().replace(".", "")));
+
+            if (parentBlockBuilder == null) {
+                return blockBuilder.getSingleValueBlock(0);
+            }
+            return null;
+        }
+
         private void serializePrimitive(BlockBuilder blockBuilder, Object node, Type type, String columnName) {
             requireNonNull(blockBuilder, "parent blockBuilder is null");