content: /*
   * Licensed to the Apache Software Foundation (ASF) under one or more
   * contributor license agreements.  See the NOTICE file distributed with
   * this work for additional information regarding copyright ownership.
   * The ASF licenses this file to you under the Apache License, Version 2.0
   * (the "License"); you may not use this file except in compliance with
   * the License.  You may obtain a copy of the License at
   *
   * http://www.apache.org/licenses/LICENSE-2.0
   *
   * Unless required by applicable law or agreed to in writing, software
   * distributed under the License is distributed on an "AS IS" BASIS,
   * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
   * See the License for the specific language governing permissions and
   * limitations under the License.
   */
  package org.apache.calcite.avatica.remote;
  
  import org.apache.calcite.avatica.AvaticaUtils;
  import org.apache.calcite.avatica.ColumnMetaData;
  import org.apache.calcite.avatica.ColumnMetaData.AvaticaType;
  import org.apache.calcite.avatica.ColumnMetaData.Rep;
  import org.apache.calcite.avatica.SqlType;
  import org.apache.calcite.avatica.proto.Common;
  import org.apache.calcite.avatica.util.ArrayFactoryImpl;
  import org.apache.calcite.avatica.util.Base64;
  import org.apache.calcite.avatica.util.ByteString;
  import org.apache.calcite.avatica.util.DateTimeUtils;
  
  import com.fasterxml.jackson.annotation.JsonCreator;
  import com.fasterxml.jackson.annotation.JsonProperty;
  import com.google.protobuf.Descriptors.FieldDescriptor;
  import com.google.protobuf.UnsafeByteOperations;
  
  import java.math.BigDecimal;
  import java.math.BigInteger;
  import java.sql.Array;
  import java.sql.ResultSet;
  import java.sql.SQLException;
  import java.sql.Time;
  import java.sql.Timestamp;
  import java.util.ArrayList;
  import java.util.Calendar;
  import java.util.Date;
  import java.util.List;
  import java.util.Objects;
  
  import static java.nio.charset.StandardCharsets.UTF_8;
  
  /** Value and type.
   *
   * <p>There are 3 representations:
   * <ul>
   *   <li>JDBC - the representation used by JDBC get and set methods
   *   <li>Serial - suitable for serializing using JSON
   *   <li>Local - used by Calcite for efficient computation
   * </ul>
   *
   * <p>The following table shows the Java type(s) that may represent each SQL
   * type in each representation.
   *
   * <table>
   *   <caption>SQL types and their representations</caption>
   *   <tr>
   *     <th>Type</th> <th>JDBC</th> <th>Serial</th> <th>Local</th>
   *   </tr>
   *   <tr>
   *     <td>BOOLEAN</td> <td>boolean</td> <td>boolean</td> <td>boolean</td>
   *   </tr>
   *   <tr>
   *     <td>BINARY, VARBINARY</td> <td>byte[]</td>
   *                    <td>String (base64)</td> <td>{@link ByteString}</td>
   *   </tr>
   *   <tr>
   *     <td>DATE</td> <td>{@link java.sql.Date}</td>
   *                                   <td>int</td> <td>int</td>
   *   </tr>
   *   <tr>
   *     <td>TIME</td> <td>{@link java.sql.Time}</td>
   *                                   <td>int</td> <td>int</td>
   *   </tr>
   *   <tr>
   *     <td>DATE</td> <td>{@link java.sql.Timestamp}</td>
   *                                   <td>long</td> <td>long</td>
   *   </tr>
   *   <tr>
   *     <td>CHAR, VARCHAR</td>
   *                   <td>String</td> <td>String</td> <td>String</td>
   *   </tr>
   *   <tr>
   *     <td>TINYINT</td> <td>byte</td> <td>Number</td> <td>byte</td>
   *   </tr>
   *   <tr>
   *     <td>SMALLINT</td> <td>short</td> <td>Number</td> <td>short</td>
   *   </tr>
   *   <tr>
   *     <td>INTEGER</td> <td>int</td> <td>Number</td> <td>int</td>
   *   </tr>
   *   <tr>
   *     <td>BIGINT</td> <td>long</td> <td>Number</td> <td>long</td>
   *   </tr>
   *   <tr>
   *     <td>REAL</td> <td>float</td> <td>Number</td> <td>float</td>
   *   </tr>
   *   <tr>
   *     <td>FLOAT, DOUBLE</td>
   *                   <td>double</td> <td>Number</td> <td>double</td>
   *   </tr>
   *   <tr>
   *     <td>DECIMAL</td>
   *                   <td>BigDecimal</td> <td>Number</td> <td>BigDecimal</td>
   *   </tr>
   *   <tr>
   *     <td>ARRAY</td>
   *                  <td>Array</td> <td>List&lt;Object&gt;</td> <td>List&lt;Object&gt;</td>
   *   </tr>
   * </table>
   *
   * <p>Note:
   *
   * <ul>
   *   <li>The various numeric types (TINYINT, SMALLINT, INTEGER, BIGINT, REAL,
   *   FLOAT, DOUBLE) are represented by {@link Number} in serial format because
   *   JSON numbers are not strongly typed. A {@code float} value {@code 3.0} is
   *   transmitted as {@code 3}, and is therefore decoded as an {@code int}.
   *
   *   <li>The date-time types (DATE, TIME, TIMESTAMP) are represented in JDBC as
   *   {@link java.sql.Date}, {@link java.sql.Time}, {@link java.sql.Timestamp},
   *   all sub-classes of {@link java.util.Date}. When they are passed to and
   *   from the server, they are interpreted in terms of a time zone, by default
   *   the current connection's time zone. Their serial and local representations
   *   as {@code int} (days since 1970-01-01 for DATE, milliseconds since
   *   00:00:00.000 for TIME), and long (milliseconds since 1970-01-01
   *   00:00:00.000 for TIMESTAMP) are easier to work with, because it is clear
   *   that time zone is not involved.
   *
   *   <li>BINARY and VARBINARY values are represented as base64-encoded strings
   *   for serialization over JSON.
   * </ul>
   */
  public class TypedValue {
    private static final FieldDescriptor NUMBER_DESCRIPTOR = Common.TypedValue.getDescriptor()
        .findFieldByNumber(Common.TypedValue.NUMBER_VALUE_FIELD_NUMBER);
    private static final FieldDescriptor STRING_DESCRIPTOR = Common.TypedValue.getDescriptor()
        .findFieldByNumber(Common.TypedValue.STRING_VALUE_FIELD_NUMBER);
    private static final FieldDescriptor BYTES_DESCRIPTOR = Common.TypedValue.getDescriptor()
        .findFieldByNumber(Common.TypedValue.BYTES_VALUE_FIELD_NUMBER);
  
    // If the user sets a `null` Object, it's explicitly null
    public static final TypedValue EXPLICIT_NULL =
        new TypedValue(ColumnMetaData.Rep.OBJECT, null);
    // The user might also implicitly not set a value for a parameter.
    public static final Common.TypedValue PROTO_IMPLICIT_NULL =
        Common.TypedValue.newBuilder().setImplicitlyNull(true).build();
  
    /** Type of the value. */
    public final ColumnMetaData.Rep type;
  
    /** Value.
     *
     * <p>Always in a form that can be serialized to JSON by Jackson.
     * For example, byte arrays are represented as String. */
    public final Object value;
  
    /** Non-null for ARRAYs, the type of the values stored in the ARRAY. Null for all other cases. */
    public final ColumnMetaData.Rep componentType;
  
    private TypedValue(ColumnMetaData.Rep rep, Object value) {
      this(rep, null, value);
    }
  
    private TypedValue(ColumnMetaData.Rep rep, ColumnMetaData.Rep componentType, Object value) {
      this.type = rep;
      this.componentType = componentType;
      this.value = value;
      assert isSerial(rep, value) : "rep: " + rep + ", value: " + value;
    }
  
    private boolean isSerial(ColumnMetaData.Rep rep, Object value) {
      if (value == null) {
        return true;
      }
      switch (rep) {
      case BYTE_STRING:
        return value instanceof String;
      case JAVA_SQL_DATE:
      case JAVA_SQL_TIME:
        return value instanceof Integer;
      case JAVA_SQL_TIMESTAMP:
      case JAVA_UTIL_DATE:
        return value instanceof Long;
      default:
        return true;
      }
    }
  
    @JsonCreator
    public static TypedValue create(@JsonProperty("type") String type,
        @JsonProperty("value") Object value) {
      if (value == null) {
        return EXPLICIT_NULL;
      }
      ColumnMetaData.Rep rep = ColumnMetaData.Rep.valueOf(type);
      return ofLocal(rep, serialToLocal(rep, value));
    }
  
    /** Creates a TypedValue from a value in local representation. */
    public static TypedValue ofLocal(ColumnMetaData.Rep rep, Object value) {
      return new TypedValue(rep, localToSerial(rep, value));
    }
  
    /** Creates a TypedValue from a value in serial representation. */
    public static TypedValue ofSerial(ColumnMetaData.Rep rep, Object value) {
      return new TypedValue(rep, value);
    }
  
    /** Creates a TypedValue from a value in JDBC representation. */
    public static TypedValue ofJdbc(ColumnMetaData.Rep rep, Object value,
        Calendar calendar) {
      if (value == null) {
        return EXPLICIT_NULL;
      }
      final Object serialValue;
      if (ColumnMetaData.Rep.ARRAY == rep) {
        // Sanity check that we were given an Array
        if (null != value && !(value instanceof Array)) {
          throw new IllegalArgumentException("Provided Rep was ARRAY, but the value was "
              + value.getClass());
        }
        final Array array = (Array) value;
        try {
          SqlType type = SqlType.valueOf(array.getBaseType());
          serialValue = jdbcToSerial(rep, array, calendar, type);
          // Because an Array may have null entries, we must always return the non-primitive type
          // variants of the array values.
          return new TypedValue(rep, Rep.nonPrimitiveRepOf(type), serialValue);
        } catch (SQLException e) {
          throw new RuntimeException("Could not extract Array component type", e);
        }
      } else {
        serialValue = jdbcToSerial(rep, value, calendar);
      }
      return new TypedValue(rep, serialValue);
    }
  
    /** Creates a TypedValue from a value in JDBC representation,
     * deducing its type. */
    public static TypedValue ofJdbc(Object value, Calendar calendar) {
      if (value == null) {
        return EXPLICIT_NULL;
      }
      final ColumnMetaData.Rep rep = ColumnMetaData.Rep.of(value.getClass());
      return new TypedValue(rep, jdbcToSerial(rep, value, calendar));
    }
  
    /** Converts the value into the local representation.
     *
     * <p>For example, a byte string is represented as a {@link ByteString};
     * a long is represented as a {@link Long} (not just some {@link Number}).
     */
    public Object toLocal() {
      if (value == null) {
        return null;
      }
      return serialToLocal(type, value);
    }
  
    /** Converts a value to the exact type required for the given
     * representation. */
    private static Object serialToLocal(ColumnMetaData.Rep rep, Object value) {
      assert value != null;
      if (rep.clazz.isInstance(value)) {
        return value;
      }
      switch (rep) {
      case BYTE:
        return ((Number) value).byteValue();
      case SHORT:
        return ((Number) value).shortValue();
      case INTEGER:
      case JAVA_SQL_DATE:
      case JAVA_SQL_TIME:
        return ((Number) value).intValue();
      case LONG:
      case JAVA_UTIL_DATE:
      case JAVA_SQL_TIMESTAMP:
        return ((Number) value).longValue();
      case FLOAT:
        return ((Number) value).floatValue();
      case DOUBLE:
        return ((Number) value).doubleValue();
      case NUMBER:
        return value instanceof BigDecimal ? value
            : value instanceof BigInteger ? new BigDecimal((BigInteger) value)
            : value instanceof Double ? new BigDecimal((Double) value)
            : value instanceof Float ? new BigDecimal((Float) value)
            : new BigDecimal(((Number) value).longValue());
      case BYTE_STRING:
        return ByteString.ofBase64((String) value);
      case ARRAY:
        //List<Object>
        return value;
      default:
        throw new IllegalArgumentException("cannot convert " + value + " ("
            + value.getClass() + ") to " + rep);
      }
    }
  
    /** Converts the value into the JDBC representation.
     *
     * <p>For example, a byte string is represented as a {@link ByteString};
     * a long is represented as a {@link Long} (not just some {@link Number}).
     */
    public Object toJdbc(Calendar calendar) {
      if (value == null) {
        return null;
      }
      return serialToJdbc(type, componentType, value, calendar);
    }
  
    /**
     * Converts the given value from serial form to JDBC form.
     *
     * @param type The type of the value
     * @param value The value
     * @param calendar A calendar instance
     * @return The JDBC representation of the value.
     */
    private static Object serialToJdbc(ColumnMetaData.Rep type, ColumnMetaData.Rep componentRep,
        Object value, Calendar calendar) {
      switch (type) {
      case BYTE_STRING:
        return ByteString.ofBase64((String) value).getBytes();
      case JAVA_UTIL_DATE:
        return DateTimeUtils.unixTimestampToUtilDate((Long) value, calendar);
      case JAVA_SQL_DATE:
        return DateTimeUtils.unixDateToSqlDate((Integer) value, calendar);
      case JAVA_SQL_TIME:
        return DateTimeUtils.unixTimeToSqlTime((Integer) value, calendar);
      case JAVA_SQL_TIMESTAMP:
        return DateTimeUtils.unixTimestampToSqlTimestamp((Long) value, calendar);
      case ARRAY:
        if (null == value) {
          return null;
        }
        final List<?> list = (List<?>) value;
        final List<Object> copy = new ArrayList<>(list.size());
        // Copy the list from the serial representation to a JDBC representation
        for (Object o : list) {
          if (null == o) {
            copy.add(null);
          } else if (o instanceof TypedValue) {
            // Protobuf can maintain the TypedValue hierarchy to simplify things
            copy.add(((TypedValue) o).toJdbc(calendar));
          } else {
            // We can't get the above recursion with the JSON serialization
            copy.add(serialToJdbc(componentRep, null, o, calendar));
          }
        }
        if (componentRep == null && list.size() > 0) {
          componentRep = ((TypedValue) list.get(0)).type;
          if (componentRep == null) {
            throw new RuntimeException("ComponentRep of element must not be null for ARRAYs");
          }
        }
        AvaticaType elementType = new AvaticaType(componentRep.typeId, componentRep.name(),
            componentRep);
        return new ArrayFactoryImpl(calendar.getTimeZone()).createArray(elementType, copy);
      default:
        return serialToLocal(type, value);
      }
    }
  
    private static Object jdbcToSerial(ColumnMetaData.Rep rep, Object value,
        Calendar calendar) {
      return jdbcToSerial(rep, value, calendar, null);
    }
  
    /** Converts a value from JDBC format to a type that can be serialized as
     * JSON. */
    private static Object jdbcToSerial(ColumnMetaData.Rep rep, Object value,
        Calendar calendar, SqlType componentType) {
      if (null == value) {
        return null;
      }
      switch (rep) {
      case BYTE_STRING:
        return new ByteString((byte[]) value).toBase64String();
      case JAVA_UTIL_DATE:
        return DateTimeUtils.utilDateToUnixTimestamp((Date) value, calendar);
      case JAVA_SQL_TIMESTAMP:
        return DateTimeUtils.sqlTimestampToUnixTimestamp((Timestamp) value, calendar);
      case JAVA_SQL_DATE:
        return DateTimeUtils.sqlDateToUnixDate((java.sql.Date) value, calendar);
      case JAVA_SQL_TIME:
        return DateTimeUtils.sqlTimeToUnixTime((Time) value, calendar);
      case ARRAY:
        Array array = (Array) value;
        Objects.requireNonNull(componentType, "Component Type must not be null for ARRAYs");
        try {
          switch (componentType) {
          case BINARY:
          case VARBINARY:
          case LONGVARBINARY:
            Object[] byteStrings = (Object[]) array.getArray();
            List<String> convertedStrings = new ArrayList<>(byteStrings.length);
            for (Object byteString : byteStrings) {
              convertedStrings.add(
                  (String) jdbcToSerial(Rep.BYTE_STRING, byteString, calendar, null));
            }
            return convertedStrings;
          case DATE:
          case TIME:
            Object[] dates = (Object[]) array.getArray();
            List<Integer> serializedDates = new ArrayList<>(dates.length);
            for (Object obj : dates) {
              Date date = (Date) obj;
              if (null == obj) {
                serializedDates.add(null);
              } else if (componentType == SqlType.DATE) {
                serializedDates.add((int) jdbcToSerial(Rep.JAVA_SQL_DATE, date, calendar, null));
              } else if (componentType == SqlType.TIME) {
                serializedDates.add((int) jdbcToSerial(Rep.JAVA_SQL_TIME, date, calendar, null));
              } else {
                throw new RuntimeException("Unexpected type: " + componentType);
              }
            }
            return serializedDates;
          case TIMESTAMP:
            Object[] timestamps = (Object[]) array.getArray();
            List<Long> serializedTimestamps = new ArrayList<>(timestamps.length);
            for (Object obj : timestamps) {
              Timestamp timestamp = (Timestamp) obj;
              if (null == obj) {
                serializedTimestamps.add(null);
              } else {
                serializedTimestamps.add(
                    (long) jdbcToSerial(Rep.JAVA_SQL_TIMESTAMP, timestamp, calendar, null));
              }
            }
            return serializedTimestamps;
          default:
            // Either a primitive array or Object[], converted into List<Object>
            return AvaticaUtils.primitiveList(array.getArray());
          }
        } catch (SQLException e) {
          throw new RuntimeException("Could not obtain base array object", e);
        }
      default:
        return value;
      }
    }
  
    /** Converts a value from internal format to a type that can be serialized
     * as JSON. */
    private static Object localToSerial(ColumnMetaData.Rep rep, Object value) {
      switch (rep) {
      case BYTE_STRING:
        return ((ByteString) value).toBase64String();
      default:
        return value;
      }
    }
  
    /** Converts a list of {@code TypedValue} to a list of values. */
    public static List<Object> values(List<TypedValue> typedValues) {
      final List<Object> list = new ArrayList<>();
      for (TypedValue typedValue : typedValues) {
        list.add(typedValue.toLocal());
      }
      return list;
    }
  
    /**
     * Creates a protocol buffer equivalent object for <code>this</code>.
     * @return A protobuf TypedValue equivalent for <code>this</code>
     */
    public Common.TypedValue toProto() {
      final Common.TypedValue.Builder builder = Common.TypedValue.newBuilder();
      // This isn't a static method, therefore we have a non-null TypedValue. Thus, this message
      // cannot be implicitly null
      builder.setImplicitlyNull(false);
  
      Common.Rep protoRep = type.toProto();
      // Protobuf has an explicit BIG_DECIMAL representation enum value.
      if (Common.Rep.NUMBER == protoRep && value instanceof BigDecimal) {
        protoRep = Common.Rep.BIG_DECIMAL;
      } else if (Common.Rep.ARRAY == protoRep) {
        // This column value is an Array (many TypedValue's)
        builder.setType(Common.Rep.ARRAY);
        // Get the array component's type
        Common.Rep protoComponentRep = componentType.toProto();
        // Set the array's component on the message
        builder.setComponentType(protoComponentRep);
        // Serialize that array into the builder
        @SuppressWarnings("unchecked")
        List<Object> list = (List<Object>) value;
        return serializeArray(list, builder, protoComponentRep);
      }
  
      // Serialize the type into the protobuf
      writeToProtoWithType(builder, value, protoRep);
  
      return builder.build();
    }
  
    Common.TypedValue serializeArray(List<Object> list, Common.TypedValue.Builder builder,
        Common.Rep protoArrayComponentRep) {
      for (Object element : list) {
        if (element instanceof List) {
          // We have a list of lists: recursively build up the protobuf
          @SuppressWarnings("unchecked")
          List<Object> subList = (List<Object>) element;
          Common.TypedValue.Builder subListBuilder = Common.TypedValue.newBuilder();
          // This is "technically" an array, but we just persist the underlying component type
          subListBuilder.setType(protoArrayComponentRep);
          Common.TypedValue protoSubList = serializeArray(subList, subListBuilder,
              protoArrayComponentRep);
          builder.addArrayValue(protoSubList);
        } else {
          // We have a list of "scalars", just serialize the value
          Common.TypedValue.Builder elementBuilder = Common.TypedValue.newBuilder();
          if (null == element) {
            writeToProtoWithType(elementBuilder, null, Common.Rep.NULL);
          } else {
            writeToProtoWithType(elementBuilder, element, protoArrayComponentRep);
          }
          builder.addArrayValue(elementBuilder.build());
        }
      }
      return builder.build();
    }
  
    private static void writeToProtoWithType(Common.TypedValue.Builder builder, Object o,
        Common.Rep type) {
      builder.setType(type);
  
      switch (type) {
      case BOOLEAN:
      case PRIMITIVE_BOOLEAN:
        builder.setBoolValue((boolean) o);
        return;
      case BYTE_STRING:
        byte[] bytes;
        // Serial representation is b64. We don't need to do that for protobuf
        if (o instanceof String) {
          // Backwards compatibility for client CALCITE-1209
          builder.setStringValue((String) o);
          // Assume strings are already b64 encoded
          bytes = ByteString.parseBase64((String) o);
        } else {
          // Backwards compatibility for client CALCITE-1209
          builder.setStringValue(Base64.encodeBytes((byte[]) o));
          // Use the byte array
          bytes = (byte[]) o;
        }
        builder.setBytesValue(UnsafeByteOperations.unsafeWrap(bytes));
        return;
      case STRING:
        String s;
        if (DateTimeUtils.isOffsetDateTime(o)) {
          s = DateTimeUtils.offsetDateTimeValue(o);
        } else {
          s = (String) o;
        }
        builder.setStringValueBytes(UnsafeByteOperations.unsafeWrap(s.getBytes(UTF_8)));
        return;
      case PRIMITIVE_CHAR:
      case CHARACTER:
        builder.setStringValue(Character.toString((char) o));
        return;
      case BYTE:
      case PRIMITIVE_BYTE:
        builder.setNumberValue(Byte.valueOf((byte) o).longValue());
        return;
      case DOUBLE:
      case PRIMITIVE_DOUBLE:
        builder.setDoubleValue((double) o);
        return;
      case FLOAT:
      case PRIMITIVE_FLOAT:
        builder.setNumberValue(Float.floatToIntBits((float) o));
        return;
      case INTEGER:
      case PRIMITIVE_INT:
        builder.setNumberValue(Integer.valueOf((int) o).longValue());
        return;
      case PRIMITIVE_SHORT:
      case SHORT:
        builder.setNumberValue(Short.valueOf((short) o).longValue());
        return;
      case LONG:
      case PRIMITIVE_LONG:
        builder.setNumberValue((long) o);
        return;
      case JAVA_SQL_DATE:
      case JAVA_SQL_TIME:
        long sqlDateOrTime;
        if (o instanceof java.sql.Date) {
          sqlDateOrTime = ((java.sql.Date) o).getTime();
        } else if (o instanceof java.sql.Time) {
          sqlDateOrTime = ((java.sql.Time) o).getTime();
        } else if (o instanceof Integer) {
          sqlDateOrTime = ((Integer) o).longValue();
        } else {
          sqlDateOrTime = (long) o;
        }
        // Persisted as numbers
        builder.setNumberValue(sqlDateOrTime);
        return;
      case JAVA_SQL_TIMESTAMP:
      case JAVA_UTIL_DATE:
        long sqlTimestampOrUtilDate;
        if (o instanceof java.sql.Timestamp) {
          sqlTimestampOrUtilDate = ((java.sql.Timestamp) o).getTime();
        } else if (o instanceof java.util.Date) {
          sqlTimestampOrUtilDate = ((java.util.Date) o).getTime();
        } else {
          sqlTimestampOrUtilDate = (long) o;
        }
        // Persisted as longs
        builder.setNumberValue(sqlTimestampOrUtilDate);
        return;
      case BIG_INTEGER:
        byte[] byteRep = ((BigInteger) o).toByteArray();
        builder.setBytesValue(com.google.protobuf.ByteString.copyFrom(byteRep));
        return;
      case BIG_DECIMAL:
        final BigDecimal bigDecimal = (BigDecimal) o;
        builder.setStringValue(bigDecimal.toString());
        return;
      case NUMBER:
        builder.setNumberValue(((Number) o).longValue());
        return;
      case NULL:
        builder.setNull(true);
        return;
      case OBJECT:
        if (null == o) {
          // We can persist a null value through easily
          builder.setNull(true);
          return;
        }
        // Intentional fall-through to RTE because we can't serialize something we have no type
        // insight into.
        // fall through
      case UNRECOGNIZED:
        // Fail?
        throw new RuntimeException("Unhandled value: " + type + " " + o.getClass());
      default:
        // Fail?
        throw new RuntimeException("Unknown serialized type: " + type);
      }
    }
  
    /**
     * Constructs a {@link TypedValue} from the protocol buffer representation.
     *
     * @param proto The protobuf Typedvalue
     * @return A {@link TypedValue} instance
     */
    public static TypedValue fromProto(Common.TypedValue proto) {
      ColumnMetaData.Rep rep = ColumnMetaData.Rep.fromProto(proto.getType());
      ColumnMetaData.Rep componentRep = ColumnMetaData.Rep.fromProto(proto.getComponentType());
      Object value = getSerialFromProto(proto);
  
      return new TypedValue(rep, componentRep, value);
    }
  
    /**
     * Converts the serialized value into the appropriate primitive/object.
     *
     * @param protoValue The serialized TypedValue.
     * @return The appropriate concrete type for the parameter value (as an Object).
     */
    public static Object getSerialFromProto(Common.TypedValue protoValue) {
      // Deserialize the value again
      switch (protoValue.getType()) {
      case BOOLEAN:
      case PRIMITIVE_BOOLEAN:
        return protoValue.getBoolValue();
      case BYTE_STRING:
        if (protoValue.hasField(STRING_DESCRIPTOR) && !protoValue.hasField(BYTES_DESCRIPTOR)) {
          // Prior to CALCITE-1103, clients would send b64 strings for bytes instead of using the
          // native byte format. The value we need to provide as the local format for TypedValue
          // is directly accessible via the Protobuf representation. Both fields are sent by the
          // server to support older clients, so only parse the string value when it is alone.
          return protoValue.getStringValue();
        }
        // TypedValue is still going to expect a b64string for BYTE_STRING even though we sent it
        // across the wire natively as bytes. Return it as b64.
        return (new ByteString(protoValue.getBytesValue().toByteArray())).toBase64String();
      case STRING:
        return protoValue.getStringValue();
      case PRIMITIVE_CHAR:
      case CHARACTER:
        return protoValue.getStringValue().charAt(0);
      case BYTE:
      case PRIMITIVE_BYTE:
        return Long.valueOf(protoValue.getNumberValue()).byteValue();
      case DOUBLE:
      case PRIMITIVE_DOUBLE:
        return protoValue.getDoubleValue();
      case FLOAT:
      case PRIMITIVE_FLOAT:
        return Float.intBitsToFloat((int) protoValue.getNumberValue());
      case INTEGER:
      case PRIMITIVE_INT:
        return Long.valueOf(protoValue.getNumberValue()).intValue();
      case PRIMITIVE_SHORT:
      case SHORT:
        return Long.valueOf(protoValue.getNumberValue()).shortValue();
      case LONG:
      case PRIMITIVE_LONG:
        return Long.valueOf(protoValue.getNumberValue());
      case JAVA_SQL_DATE:
      case JAVA_SQL_TIME:
        return Long.valueOf(protoValue.getNumberValue()).intValue();
      case JAVA_SQL_TIMESTAMP:
      case JAVA_UTIL_DATE:
        return protoValue.getNumberValue();
      case BIG_INTEGER:
        return new BigInteger(protoValue.getBytesValue().toByteArray());
      case BIG_DECIMAL:
        // CALCITE-1103 shifts BigDecimals to be serialized as strings.
        if (protoValue.hasField(NUMBER_DESCRIPTOR)) {
          // This is the old (broken) style.
          BigInteger bigInt = new BigInteger(protoValue.getBytesValue().toByteArray());
          return new BigDecimal(bigInt, (int) protoValue.getNumberValue());
        }
        return new BigDecimal(protoValue.getStringValueBytes().toStringUtf8());
      case NUMBER:
        return Long.valueOf(protoValue.getNumberValue());
      case NULL:
        return null;
      case ARRAY:
        final List<Common.TypedValue> protoList = protoValue.getArrayValueList();
        final List<Object> list = new ArrayList<>(protoList.size());
        for (Common.TypedValue protoElement : protoList) {
          // Deserialize the TypedValue protobuf into the JDBC type
          TypedValue listElement = TypedValue.fromProto(protoElement);
          // Must preserve the TypedValue so serial/jdbc/local conversion can work as expected
          list.add(listElement);
        }
        return list;
      case OBJECT:
        if (protoValue.getNull()) {
          return null;
        }
        // Intentional fall through to RTE. If we sent an object over the wire, it could only
        // possibly be null (at this point). Anything else has to be an error.
        // fall through
      case UNRECOGNIZED:
        // Fail?
        throw new RuntimeException("Unhandled type: " + protoValue.getType());
      default:
        // Fail?
        throw new RuntimeException("Unknown type: " + protoValue.getType());
      }
    }
  
    /**
     * Writes the given object into the Protobuf representation of a TypedValue. The object is
     * serialized given the type of that object, mapping it to the appropriate representation.
     *
     * @param builder The TypedValue protobuf builder
     * @param o The object (value)
     */
    public static Common.Rep toProto(Common.TypedValue.Builder builder, Object o) {
      // Numbers
      if (o instanceof Byte) {
        writeToProtoWithType(builder, o, Common.Rep.BYTE);
        return Common.Rep.BYTE;
      } else if (o instanceof Short) {
        writeToProtoWithType(builder, o, Common.Rep.SHORT);
        return Common.Rep.SHORT;
      } else if (o instanceof Integer) {
        writeToProtoWithType(builder, o, Common.Rep.INTEGER);
        return Common.Rep.INTEGER;
      } else if (o instanceof Long) {
        writeToProtoWithType(builder, o, Common.Rep.LONG);
        return Common.Rep.LONG;
      } else if (o instanceof Double) {
        writeToProtoWithType(builder, o, Common.Rep.DOUBLE);
        return Common.Rep.DOUBLE;
      } else if (o instanceof Float) {
        writeToProtoWithType(builder, o, Common.Rep.FLOAT);
        return Common.Rep.FLOAT;
      } else if (o instanceof BigDecimal) {
        writeToProtoWithType(builder, o, Common.Rep.BIG_DECIMAL);
        return Common.Rep.BIG_DECIMAL;
      // Strings
      } else if (o instanceof String) {
        writeToProtoWithType(builder, o, Common.Rep.STRING);
        return Common.Rep.STRING;
      } else if (o instanceof Character) {
        writeToProtoWithType(builder, o, Common.Rep.CHARACTER);
        return Common.Rep.CHARACTER;
      // Bytes
      } else if (o instanceof byte[]) {
        writeToProtoWithType(builder, o, Common.Rep.BYTE_STRING);
        return Common.Rep.BYTE_STRING;
      // Boolean
      } else if (o instanceof Boolean) {
        writeToProtoWithType(builder, o, Common.Rep.BOOLEAN);
        return Common.Rep.BOOLEAN;
      } else if (o instanceof Timestamp) {
        writeToProtoWithType(builder, o, Common.Rep.JAVA_SQL_TIMESTAMP);
        return Common.Rep.JAVA_SQL_TIMESTAMP;
      } else if (o instanceof java.sql.Date) {
        writeToProtoWithType(builder, o, Common.Rep.JAVA_SQL_DATE);
        return Common.Rep.JAVA_SQL_DATE;
      } else if (o instanceof Time) {
        writeToProtoWithType(builder, o, Common.Rep.JAVA_SQL_TIME);
        return Common.Rep.JAVA_SQL_TIME;
      } else if (DateTimeUtils.isOffsetDateTime(o)) {
        // SQL type is TIMESTAMP WITH TIMEZONE. Transmit as a string.
        writeToProtoWithType(builder, o, Common.Rep.STRING);
        return Common.Rep.STRING;
      } else if (o instanceof List) {
        // Treat a List as an Array
        builder.setType(Common.Rep.ARRAY);
        builder.setComponentType(Common.Rep.OBJECT);
        boolean setComponentType = false;
        for (Object listElement : (List<?>) o) {
          Common.TypedValue.Builder listElementBuilder = Common.TypedValue.newBuilder();
          // Recurse on each list element
          Common.Rep componentRep = toProto(listElementBuilder, listElement);
          if (!setComponentType) {
            if (Common.Rep.NULL != componentRep) {
              builder.setComponentType(componentRep);
            }
            setComponentType = true;
          }
          builder.addArrayValue(listElementBuilder.build());
        }
        return Common.Rep.ARRAY;
      } else if (o instanceof Array) {
        builder.setType(Common.Rep.ARRAY);
        Array a = (Array) o;
        try {
          ResultSet rs = a.getResultSet();
          builder.setComponentType(Common.Rep.OBJECT);
          boolean setComponentType = false;
          while (rs.next()) {
            Common.TypedValue.Builder listElementBuilder = Common.TypedValue.newBuilder();
            Object arrayValue = rs.getObject(2);
            Common.Rep componentRep = toProto(listElementBuilder, arrayValue);
            if (!setComponentType) {
              if (Common.Rep.NULL != componentRep) {
                builder.setComponentType(componentRep);
              }
              setComponentType = true;
            }
            builder.addArrayValue(listElementBuilder.build());
          }
        } catch (SQLException e) {
          throw new RuntimeException("Could not serialize ARRAY", e);
        }
        return Common.Rep.ARRAY;
      } else if (null == o) {
        writeToProtoWithType(builder, o, Common.Rep.NULL);
        return Common.Rep.NULL;
      // Unhandled
      } else {
        throw new RuntimeException("Unhandled type in Frame: " + o.getClass());
      }
    }
  
    /**
     * Extracts the JDBC value from protobuf-TypedValue representation.
     *
     * @param protoValue Protobuf TypedValue
     * @param calendar Instance of a calendar
     * @return The JDBC representation of this TypedValue
     */
    public static Object protoToJdbc(Common.TypedValue protoValue, Calendar calendar) {
      Object o = getSerialFromProto(Objects.requireNonNull(protoValue));
      // Shortcircuit the null
      if (null == o) {
        return o;
      }
      return serialToJdbc(Rep.fromProto(protoValue.getType()), null, o, calendar);
    }
  
    @Override public int hashCode() {
      final int prime = 31;
      int result = 1;
      result = prime * result + ((type == null) ? 0 : type.hashCode());
      result = prime * result + ((value == null) ? 0 : value.hashCode());
      return result;
    }
  
    @Override public boolean equals(Object o) {
      if (o == this) {
        return true;
      }
      if (o instanceof TypedValue) {
        TypedValue other = (TypedValue) o;
  
        if (type != other.type) {
          return false;
        }
  
        if (null == value) {
          return null == other.value;
        }
  
        return value.equals(other.value);
      }
  
      return false;
    }
  }
  
  // End TypedValue.java
  