rajatbhatta
diff --git a/‎src/main/java/com/google/cloud/spanner/jdbc/AbstractJdbcWrapper.java
+3 b/‎src/main/java/com/google/cloud/spanner/jdbc/AbstractJdbcWrapper.java
+3
diff --git a/‎src/main/java/com/google/cloud/spanner/jdbc/JdbcArray.java
+4 b/‎src/main/java/com/google/cloud/spanner/jdbc/JdbcArray.java
+4
diff --git a/‎src/main/java/com/google/cloud/spanner/jdbc/JdbcDataType.java
+26 b/‎src/main/java/com/google/cloud/spanner/jdbc/JdbcDataType.java
+26
diff --git a/‎src/main/java/com/google/cloud/spanner/jdbc/JdbcParameterStore.java
+58-22 b/‎src/main/java/com/google/cloud/spanner/jdbc/JdbcParameterStore.java
+58-22
diff --git a/‎src/main/java/com/google/cloud/spanner/jdbc/JdbcResultSet.java
+17 b/‎src/main/java/com/google/cloud/spanner/jdbc/JdbcResultSet.java
+17
diff --git a/‎src/main/java/com/google/cloud/spanner/jdbc/JdbcTypeConverter.java
+6-1 b/‎src/main/java/com/google/cloud/spanner/jdbc/JdbcTypeConverter.java
+6-1
@@ -48,6 +48,7 @@ static int extractColumnType(Type type) {
     if (type.equals(Type.int64())) return Types.BIGINT;
     if (type.equals(Type.numeric())) return Types.NUMERIC;
     if (type.equals(Type.string())) return Types.NVARCHAR;
+    if (type.equals(Type.json())) return Types.NVARCHAR;
     if (type.equals(Type.timestamp())) return Types.TIMESTAMP;
     if (type.getCode() == Code.ARRAY) return Types.ARRAY;
     return Types.OTHER;
@@ -106,6 +107,7 @@ static String getClassName(Type type) {
     if (type == Type.int64()) return Long.class.getName();
     if (type == Type.numeric()) return BigDecimal.class.getName();
     if (type == Type.string()) return String.class.getName();
+    if (type == Type.json()) return String.class.getName();
     if (type == Type.timestamp()) return Timestamp.class.getName();
     if (type.getCode() == Code.ARRAY) {
       if (type.getArrayElementType() == Type.bool()) return Boolean[].class.getName();
@@ -115,6 +117,7 @@ static String getClassName(Type type) {
       if (type.getArrayElementType() == Type.int64()) return Long[].class.getName();
       if (type.getArrayElementType() == Type.numeric()) return BigDecimal[].class.getName();
       if (type.getArrayElementType() == Type.string()) return String[].class.getName();
+      if (type.getArrayElementType() == Type.json()) return String[].class.getName();
       if (type.getArrayElementType() == Type.timestamp()) return Timestamp[].class.getName();
     }
     return null;
 
@@ -21,6 +21,7 @@
 import com.google.cloud.spanner.Struct;
 import com.google.cloud.spanner.Type;
 import com.google.cloud.spanner.Type.StructField;
+import com.google.cloud.spanner.Value;
 import com.google.cloud.spanner.ValueBinder;
 import com.google.common.collect.ImmutableList;
 import com.google.rpc.Code;
@@ -201,6 +202,9 @@ public ResultSet getResultSet(long startIndex, int count) throws SQLException {
           case STRING:
             builder = binder.to((String) value);
             break;
+          case JSON:
+            builder = binder.to(Value.json((String) value));
+            break;
           case TIMESTAMP:
             builder = binder.to(JdbcTypeConverter.toGoogleTimestamp((Timestamp) value));
             break;
 
@@ -229,6 +229,32 @@ public Type getSpannerType() {
       return Type.string();
     }
   },
+  JSON {
+    @Override
+    public int getSqlType() {
+      return JsonType.VENDOR_TYPE_NUMBER;
+    }
+
+    @Override
+    public Class<String> getJavaClass() {
+      return String.class;
+    }
+
+    @Override
+    public Code getCode() {
+      return Code.JSON;
+    }
+
+    @Override
+    public List<String> getArrayElements(ResultSet rs, int columnIndex) {
+      return rs.getJsonList(columnIndex);
+    }
+
+    @Override
+    public Type getSpannerType() {
+      return Type.json();
+    }
+  },
   TIMESTAMP {
     @Override
     public int getSqlType() {
 
@@ -19,6 +19,7 @@
 import com.google.cloud.ByteArray;
 import com.google.cloud.spanner.Statement;
 import com.google.cloud.spanner.Statement.Builder;
+import com.google.cloud.spanner.Type;
 import com.google.cloud.spanner.Value;
 import com.google.cloud.spanner.ValueBinder;
 import com.google.cloud.spanner.jdbc.JdbcSqlExceptionFactory.JdbcSqlExceptionImpl;
@@ -264,6 +265,7 @@ private boolean isTypeSupported(int sqlType) {
       case Types.NCLOB:
       case Types.NUMERIC:
       case Types.DECIMAL:
+      case JsonType.VENDOR_TYPE_NUMBER:
         return true;
     }
     return false;
@@ -315,6 +317,11 @@ private boolean isValidTypeAndValue(Object value, int sqlType) {
         return value instanceof Clob || value instanceof Reader;
       case Types.NCLOB:
         return value instanceof NClob || value instanceof Reader;
+      case JsonType.VENDOR_TYPE_NUMBER:
+        return value instanceof String
+            || value instanceof InputStream
+            || value instanceof Reader
+            || (value instanceof Value && ((Value) value).getType().getCode() == Type.Code.JSON);
     }
     return false;
   }
@@ -544,30 +551,34 @@ private Builder setParamWithKnownType(ValueBinder<Builder> binder, Object value,
       case Types.NCHAR:
       case Types.NVARCHAR:
       case Types.LONGNVARCHAR:
+        String stringValue;
         if (value instanceof String) {
-          return binder.to((String) value);
+          stringValue = (String) value;
         } else if (value instanceof InputStream) {
-          InputStreamReader reader =
-              new InputStreamReader((InputStream) value, StandardCharsets.US_ASCII);
-          try {
-            return binder.to(CharStreams.toString(reader));
-          } catch (IOException e) {
-            throw JdbcSqlExceptionFactory.of(
-                "could not set string from input stream", Code.INVALID_ARGUMENT, e);
-          }
+          stringValue = getStringFromInputStream((InputStream) value);
         } else if (value instanceof Reader) {
-          try {
-            return binder.to(CharStreams.toString((Reader) value));
-          } catch (IOException e) {
-            throw JdbcSqlExceptionFactory.of(
-                "could not set string from reader", Code.INVALID_ARGUMENT, e);
-          }
+          stringValue = getStringFromReader((Reader) value);
         } else if (value instanceof URL) {
-          return binder.to(((URL) value).toString());
+          stringValue = ((URL) value).toString();
         } else if (value instanceof UUID) {
-          return binder.to(((UUID) value).toString());
+          stringValue = ((UUID) value).toString();
+        } else {
+          throw JdbcSqlExceptionFactory.of(value + " is not a valid string", Code.INVALID_ARGUMENT);
+        }
+        return binder.to(stringValue);
+      case JsonType.VENDOR_TYPE_NUMBER:
+        String jsonValue;
+        if (value instanceof String) {
+          jsonValue = (String) value;
+        } else if (value instanceof InputStream) {
+          jsonValue = getStringFromInputStream((InputStream) value);
+        } else if (value instanceof Reader) {
+          jsonValue = getStringFromReader((Reader) value);
+        } else {
+          throw JdbcSqlExceptionFactory.of(
+              value + " is not a valid JSON value", Code.INVALID_ARGUMENT);
         }
-        throw JdbcSqlExceptionFactory.of(value + " is not a valid string", Code.INVALID_ARGUMENT);
+        return binder.to(Value.json(jsonValue));
       case Types.DATE:
         if (value instanceof Date) {
           return binder.to(JdbcTypeConverter.toGoogleDate((Date) value));
@@ -652,6 +663,25 @@ private Builder setParamWithKnownType(ValueBinder<Builder> binder, Object value,
     return null;
   }
 
+  private String getStringFromInputStream(InputStream inputStream) throws SQLException {
+    InputStreamReader reader = new InputStreamReader(inputStream, StandardCharsets.US_ASCII);
+    try {
+      return CharStreams.toString(reader);
+    } catch (IOException e) {
+      throw JdbcSqlExceptionFactory.of(
+          "could not set string from input stream", Code.INVALID_ARGUMENT, e);
+    }
+  }
+
+  private String getStringFromReader(Reader reader) throws SQLException {
+    try {
+      return CharStreams.toString(reader);
+    } catch (IOException e) {
+      throw JdbcSqlExceptionFactory.of(
+          "could not set string from reader", Code.INVALID_ARGUMENT, e);
+    }
+  }
+
   /** Set the parameter value based purely on the type of the value. */
   private Builder setParamWithUnknownType(ValueBinder<Builder> binder, Object value)
       throws SQLException {
@@ -769,14 +799,16 @@ private Builder setArrayValue(ValueBinder<Builder> binder, int type, Object valu
         case Types.LONGNVARCHAR:
         case Types.CLOB:
         case Types.NCLOB:
-          return binder.toStringArray((Iterable<String>) null);
+          return binder.toStringArray(null);
+        case JsonType.VENDOR_TYPE_NUMBER:
+          return binder.toJsonArray(null);
         case Types.DATE:
-          return binder.toDateArray((Iterable<com.google.cloud.Date>) null);
+          return binder.toDateArray(null);
         case Types.TIME:
         case Types.TIME_WITH_TIMEZONE:
         case Types.TIMESTAMP:
         case Types.TIMESTAMP_WITH_TIMEZONE:
-          return binder.toTimestampArray((Iterable<com.google.cloud.Timestamp>) null);
+          return binder.toTimestampArray(null);
         case Types.BINARY:
         case Types.VARBINARY:
         case Types.LONGVARBINARY:
@@ -829,7 +861,11 @@ private Builder setArrayValue(ValueBinder<Builder> binder, int type, Object valu
     } else if (Timestamp[].class.isAssignableFrom(value.getClass())) {
       return binder.toTimestampArray(JdbcTypeConverter.toGoogleTimestamps((Timestamp[]) value));
     } else if (String[].class.isAssignableFrom(value.getClass())) {
-      return binder.toStringArray(Arrays.asList((String[]) value));
+      if (type == JsonType.VENDOR_TYPE_NUMBER) {
+        return binder.toJsonArray(Arrays.asList((String[]) value));
+      } else {
+        return binder.toStringArray(Arrays.asList((String[]) value));
+      }
     } else if (byte[][].class.isAssignableFrom(value.getClass())) {
       return binder.toBytesArray(JdbcTypeConverter.toGoogleBytes((byte[][]) value));
     }
 
@@ -142,6 +142,8 @@ public String getString(int columnIndex) throws SQLException {
         return isNull ? null : spanner.getBigDecimal(spannerIndex).toString();
       case STRING:
         return isNull ? null : spanner.getString(spannerIndex);
+      case JSON:
+        return isNull ? null : spanner.getJson(spannerIndex);
       case TIMESTAMP:
         return isNull ? null : spanner.getTimestamp(spannerIndex).toString();
       case STRUCT:
@@ -169,6 +171,7 @@ public boolean getBoolean(int columnIndex) throws SQLException {
       case STRING:
         return isNull ? false : Boolean.valueOf(spanner.getString(spannerIndex));
       case BYTES:
+      case JSON:
       case DATE:
       case STRUCT:
       case TIMESTAMP:
@@ -198,6 +201,7 @@ public byte getByte(int columnIndex) throws SQLException {
       case STRING:
         return isNull ? (byte) 0 : checkedCastToByte(parseLong(spanner.getString(spannerIndex)));
       case BYTES:
+      case JSON:
       case DATE:
       case STRUCT:
       case TIMESTAMP:
@@ -227,6 +231,7 @@ public short getShort(int columnIndex) throws SQLException {
       case STRING:
         return isNull ? 0 : checkedCastToShort(parseLong(spanner.getString(spannerIndex)));
       case BYTES:
+      case JSON:
       case DATE:
       case STRUCT:
       case TIMESTAMP:
@@ -256,6 +261,7 @@ public int getInt(int columnIndex) throws SQLException {
       case STRING:
         return isNull ? 0 : checkedCastToInt(parseLong(spanner.getString(spannerIndex)));
       case BYTES:
+      case JSON:
       case DATE:
       case STRUCT:
       case TIMESTAMP:
@@ -283,6 +289,7 @@ public long getLong(int columnIndex) throws SQLException {
       case STRING:
         return isNull ? 0L : parseLong(spanner.getString(spannerIndex));
       case BYTES:
+      case JSON:
       case DATE:
       case STRUCT:
       case TIMESTAMP:
@@ -310,6 +317,7 @@ public float getFloat(int columnIndex) throws SQLException {
       case STRING:
         return isNull ? 0 : checkedCastToFloat(parseDouble(spanner.getString(spannerIndex)));
       case BYTES:
+      case JSON:
       case DATE:
       case STRUCT:
       case TIMESTAMP:
@@ -337,6 +345,7 @@ public double getDouble(int columnIndex) throws SQLException {
       case STRING:
         return isNull ? 0 : parseDouble(spanner.getString(spannerIndex));
       case BYTES:
+      case JSON:
       case DATE:
       case STRUCT:
       case TIMESTAMP:
@@ -372,6 +381,7 @@ public Date getDate(int columnIndex) throws SQLException {
       case INT64:
       case NUMERIC:
       case BYTES:
+      case JSON:
       case STRUCT:
       case ARRAY:
       default:
@@ -396,6 +406,7 @@ public Time getTime(int columnIndex) throws SQLException {
       case INT64:
       case NUMERIC:
       case BYTES:
+      case JSON:
       case STRUCT:
       case ARRAY:
       default:
@@ -421,6 +432,7 @@ public Timestamp getTimestamp(int columnIndex) throws SQLException {
       case INT64:
       case NUMERIC:
       case BYTES:
+      case JSON:
       case STRUCT:
       case ARRAY:
       default:
@@ -576,6 +588,7 @@ private Object getObject(Type type, int columnIndex) throws SQLException {
     if (type == Type.int64()) return getLong(columnIndex);
     if (type == Type.numeric()) return getBigDecimal(columnIndex);
     if (type == Type.string()) return getString(columnIndex);
+    if (type == Type.json()) return getString(columnIndex);
     if (type == Type.timestamp()) return getTimestamp(columnIndex);
     if (type.getCode() == Code.ARRAY) return getArray(columnIndex);
     throw JdbcSqlExceptionFactory.of(
@@ -664,6 +677,7 @@ private BigDecimal getBigDecimal(int columnIndex, boolean fixedScale, int scale)
               e);
         }
       case BYTES:
+      case JSON:
       case DATE:
       case TIMESTAMP:
       case STRUCT:
@@ -749,6 +763,7 @@ public Date getDate(int columnIndex, Calendar cal) throws SQLException {
       case INT64:
       case NUMERIC:
       case BYTES:
+      case JSON:
       case STRUCT:
       case ARRAY:
       default:
@@ -778,6 +793,7 @@ public Time getTime(int columnIndex, Calendar cal) throws SQLException {
       case INT64:
       case NUMERIC:
       case BYTES:
+      case JSON:
       case STRUCT:
       case ARRAY:
       default:
@@ -810,6 +826,7 @@ public Timestamp getTimestamp(int columnIndex, Calendar cal) throws SQLException
       case INT64:
       case NUMERIC:
       case BYTES:
+      case JSON:
       case STRUCT:
       case ARRAY:
       default:
 
@@ -81,7 +81,8 @@ static Object convert(Object value, Type type, Class<?> targetType) throws SQLEx
       }
       if (targetType.equals(byte[].class)) {
         if (type.getCode() == Code.BYTES) return value;
-        if (type.getCode() == Code.STRING) return ((String) value).getBytes(UTF8);
+        if (type.getCode() == Code.STRING || type.getCode() == Code.JSON)
+          return ((String) value).getBytes(UTF8);
       }
       if (targetType.equals(Boolean.class)) {
         if (type.getCode() == Code.BOOL) return value;
@@ -186,6 +187,8 @@ private static Value convertToSpannerValue(Object value, Type type) throws SQLEx
           case TIMESTAMP:
             return Value.timestampArray(
                 toGoogleTimestamps((java.sql.Timestamp[]) ((java.sql.Array) value).getArray()));
+          case JSON:
+            return Value.jsonArray(Arrays.asList((String[]) ((java.sql.Array) value).getArray()));
           case STRUCT:
           default:
             throw JdbcSqlExceptionFactory.of(
@@ -207,6 +210,8 @@ private static Value convertToSpannerValue(Object value, Type type) throws SQLEx
         return Value.string((String) value);
       case TIMESTAMP:
         return Value.timestamp(toGoogleTimestamp((java.sql.Timestamp) value));
+      case JSON:
+        return Value.json((String) value);
       case STRUCT:
       default:
         throw JdbcSqlExceptionFactory.of(