[Pulsar SQL] support protobuf/timestamp (apache#13287)

Author: tjiuming

pulsar-sql/presto-pulsar/src/main/java/org/apache/pulsar/sql/presto/decoder/protobufnative/PulsarProtobufNativeColumnDecoder.java

@@ -50,6 +50,7 @@
 import io.prestosql.spi.type.RowType;
 import io.prestosql.spi.type.RowType.Field;
 import io.prestosql.spi.type.SmallintType;
+import io.prestosql.spi.type.TimestampType;
 import io.prestosql.spi.type.TinyintType;
 import io.prestosql.spi.type.Type;
 import io.prestosql.spi.type.VarbinaryType;
@@ -69,7 +70,8 @@ public class PulsarProtobufNativeColumnDecoder {
             BigintType.BIGINT,
             RealType.REAL,
             DoubleType.DOUBLE,
-            VarbinaryType.VARBINARY);
+            VarbinaryType.VARBINARY,
+            TimestampType.TIMESTAMP);
 
     private final Type columnType;
     private final String columnMapping;
@@ -192,6 +194,15 @@ public long getLong() {
                 return floatToIntBits((Float) value);
             }
 
+            //return millisecond which parsed from protobuf/timestamp
+            if (columnType instanceof TimestampType && value instanceof DynamicMessage) {
+                DynamicMessage message = (DynamicMessage) value;
+                int nanos = (int) message.getField(message.getDescriptorForType().findFieldByName("nanos"));
+                long seconds = (long) message.getField(message.getDescriptorForType().findFieldByName("seconds"));
+                //maybe an exception here, but seems will never happen in hundred years.
+                return seconds * MILLIS_PER_SECOND + nanos / NANOS_PER_MILLISECOND;
+            }
+
             throw new PrestoException(DECODER_CONVERSION_NOT_SUPPORTED,
                     format("cannot decode object of '%s' as '%s' for column '%s'",
                             value.getClass(), columnType, columnName));
@@ -376,5 +387,6 @@ private static Block serializeRow(BlockBuilder parentBlockBuilder, Object value,
 
     protected static final String PROTOBUF_MAP_KEY_NAME = "key";
     protected static final String PROTOBUF_MAP_VALUE_NAME = "value";
-
+    private static final long MILLIS_PER_SECOND = 1000;
+    private static final long NANOS_PER_MILLISECOND = 1000000;
 }

pulsar-sql/presto-pulsar/src/main/java/org/apache/pulsar/sql/presto/decoder/protobufnative/PulsarProtobufNativeRowDecoderFactory.java

@@ -25,6 +25,7 @@
 
 import com.google.common.collect.ImmutableList;
 import com.google.protobuf.Descriptors;
+import com.google.protobuf.TimestampProto;
 import io.airlift.log.Logger;
 import io.prestosql.decoder.DecoderColumnHandle;
 import io.prestosql.spi.PrestoException;
@@ -37,11 +38,13 @@
 import io.prestosql.spi.type.RealType;
 import io.prestosql.spi.type.RowType;
 import io.prestosql.spi.type.StandardTypes;
+import io.prestosql.spi.type.TimestampType;
 import io.prestosql.spi.type.Type;
 import io.prestosql.spi.type.TypeManager;
 import io.prestosql.spi.type.TypeSignature;
 import io.prestosql.spi.type.TypeSignatureParameter;
 import io.prestosql.spi.type.VarbinaryType;
+
 import java.util.List;
 import java.util.Optional;
 import java.util.Set;
@@ -142,11 +145,16 @@ private Type parseProtobufPrestoType(Descriptors.FieldDescriptor field) {
                             ImmutableList.of(TypeSignatureParameter.typeParameter(keyType),
                                     TypeSignatureParameter.typeParameter(valueType)));
                 } else {
-                    //row
-                    dataType = RowType.from(msg.getFields().stream()
-                            .map(rowField -> new RowType.Field(Optional.of(rowField.getName()),
-                                    parseProtobufPrestoType(rowField)))
-                            .collect(toImmutableList()));
+                    if (TimestampProto.getDescriptor().toProto().getName().equals(msg.getFile().toProto().getName())) {
+                        //if msg type is protobuf/timestamp
+                        dataType = TimestampType.TIMESTAMP;
+                    } else {
+                        //row
+                        dataType = RowType.from(msg.getFields().stream()
+                                .map(rowField -> new RowType.Field(Optional.of(rowField.getName()),
+                                        parseProtobufPrestoType(rowField)))
+                                .collect(toImmutableList()));
+                    }
                 }
                 break;
             default: