SQL: Consolidate JDBC type handling (elastic/x-pack-elasticsearch#3761)

Consolidates handling of JDBC types conversion into a single file that should simplify maintaining consistency between type handling. Also separates the types that are handled as part of Elasticsearch output and types that are handled as user-supplied parameters.

relates elastic/x-pack-elasticsearch#3556

Original commit: elastic/x-pack-elasticsearch@d251fce66b

plugin/sql/jdbc/src/main/java/org/elasticsearch/xpack/sql/jdbc/jdbc/JdbcDatabaseMetaData.java

@@ -22,6 +22,9 @@ import java.sql.SQLFeatureNotSupportedException;
 import java.util.ArrayList;
 import java.util.List;
 
+import static java.sql.JDBCType.INTEGER;
+import static java.sql.JDBCType.SMALLINT;
+
 /**
  * Implementation of {@link DatabaseMetaData} for Elasticsearch. Draws inspiration
  * from <a href="https://www.postgresql.org/docs/9.0/static/information-schema.html">
@@ -640,11 +643,11 @@ class JdbcDatabaseMetaData implements DatabaseMetaData, JdbcWrapper {
                      "PROCEDURE_CAT",
                      "PROCEDURE_SCHEM",
                      "PROCEDURE_NAME",
-                     "NUM_INPUT_PARAMS", int.class,
+                     "NUM_INPUT_PARAMS", INTEGER,
-                     "NUM_OUTPUT_PARAMS", int.class,
+                     "NUM_OUTPUT_PARAMS", INTEGER,
-                     "NUM_RESULT_SETS", int.class,
+                     "NUM_RESULT_SETS", INTEGER,
                      "REMARKS",
-                     "PROCEDURE_TYPE", short.class,
+                     "PROCEDURE_TYPE", SMALLINT,
                      "SPECIFIC_NAME");
     }
 
@@ -657,20 +660,20 @@ class JdbcDatabaseMetaData implements DatabaseMetaData, JdbcWrapper {
                      "PROCEDURE_SCHEM",
                      "PROCEDURE_NAME",
                      "COLUMN_NAME",
-                     "COLUMN_TYPE", short.class,
+                     "COLUMN_TYPE", SMALLINT,
-                     "DATA_TYPE", int.class,
+                     "DATA_TYPE", INTEGER,
                      "TYPE_NAME",
-                     "PRECISION", int.class,
+                     "PRECISION", INTEGER,
-                     "LENGTH", int.class,
+                     "LENGTH", INTEGER,
-                     "SCALE", short.class,
+                     "SCALE", SMALLINT,
-                     "RADIX", short.class,
+                     "RADIX", SMALLINT,
-                     "NULLABLE", short.class,
+                     "NULLABLE", SMALLINT,
                      "REMARKS",
                      "COLUMN_DEF",
-                     "SQL_DATA_TYPE", int.class,
+                     "SQL_DATA_TYPE", INTEGER,
-                     "SQL_DATETIME_SUB", int.class,
+                     "SQL_DATETIME_SUB", INTEGER,
-                     "CHAR_OCTET_LENGTH", int.class,
+                     "CHAR_OCTET_LENGTH", INTEGER,
-                     "ORDINAL_POSITION", int.class,
+                     "ORDINAL_POSITION", INTEGER,
                      "IS_NULLABLE",
                      "SPECIFIC_NAME");
     }
@@ -739,7 +742,6 @@ class JdbcDatabaseMetaData implements DatabaseMetaData, JdbcWrapper {
     @Override
     public ResultSet getColumns(String catalog, String schemaPattern, String tableNamePattern, String columnNamePattern)
             throws SQLException {
-
         PreparedStatement ps = con.prepareStatement("SYS COLUMNS TABLES LIKE ? LIKE ?");
         ps.setString(1, tableNamePattern != null ? tableNamePattern.trim() : "%");
         ps.setString(2, columnNamePattern != null ? columnNamePattern.trim() : "%");
@@ -865,9 +867,9 @@ class JdbcDatabaseMetaData implements DatabaseMetaData, JdbcWrapper {
                     "TYPE_SCHEM",
                     "TYPE_NAME",
                     "CLASS_NAME",
-                    "DATA_TYPE", int.class,
+                    "DATA_TYPE", INTEGER,
                     "REMARKS",
-                    "BASE_TYPE", short.class);
+                    "BASE_TYPE", SMALLINT);
     }
 
     @Override
@@ -926,23 +928,23 @@ class JdbcDatabaseMetaData implements DatabaseMetaData, JdbcWrapper {
                      "TYPE_SCHEM",
                      "TYPE_NAME",
                      "ATTR_NAME",
-                     "DATA_TYPE", int.class,
+                     "DATA_TYPE", INTEGER,
                      "ATTR_TYPE_NAME",
-                     "ATTR_SIZE", int.class,
+                     "ATTR_SIZE", INTEGER,
-                     "DECIMAL_DIGITS", int.class,
+                     "DECIMAL_DIGITS", INTEGER,
-                     "NUM_PREC_RADIX", int.class,
+                     "NUM_PREC_RADIX", INTEGER,
-                     "NULLABLE", int.class,
+                     "NULLABLE", INTEGER,
                      "REMARKS",
                      "ATTR_DEF",
-                     "SQL_DATA_TYPE", int.class,
+                     "SQL_DATA_TYPE", INTEGER,
-                     "SQL_DATETIME_SUB", int.class,
+                     "SQL_DATETIME_SUB", INTEGER,
-                     "CHAR_OCTET_LENGTH", int.class,
+                     "CHAR_OCTET_LENGTH", INTEGER,
-                     "ORDINAL_POSITION", int.class,
+                     "ORDINAL_POSITION", INTEGER,
                      "IS_NULLABLE",
                      "SCOPE_CATALOG",
                      "SCOPE_SCHEMA",
                      "SCOPE_TABLE",
-                     "SOURCE_DATA_TYPE", short.class);
+                     "SOURCE_DATA_TYPE", SMALLINT);
     }
 
     @Override
@@ -1018,7 +1020,7 @@ class JdbcDatabaseMetaData implements DatabaseMetaData, JdbcWrapper {
                      "FUNCTION_SCHEM",
                      "FUNCTION_NAME",
                      "REMARKS",
-                     "FUNCTION_TYPE", short.class,
+                     "FUNCTION_TYPE", SMALLINT,
                      "SPECIFIC_NAME");
     }
 
@@ -1031,16 +1033,16 @@ class JdbcDatabaseMetaData implements DatabaseMetaData, JdbcWrapper {
                      "FUNCTION_SCHEM",
                      "FUNCTION_NAME",
                      "COLUMN_NAME",
-                     "DATA_TYPE", int.class,
+                     "DATA_TYPE", INTEGER,
                      "TYPE_NAME",
-                     "PRECISION", int.class,
+                     "PRECISION", INTEGER,
-                     "LENGTH", int.class,
+                     "LENGTH", INTEGER,
-                     "SCALE", short.class,
+                     "SCALE", SMALLINT,
-                     "RADIX", short.class,
+                     "RADIX", SMALLINT,
-                     "NULLABLE", short.class,
+                     "NULLABLE", SMALLINT,
                      "REMARKS",
-                     "CHAR_OCTET_LENGTH", int.class,
+                     "CHAR_OCTET_LENGTH", INTEGER,
-                     "ORDINAL_POSITION", int.class,
+                     "ORDINAL_POSITION", INTEGER,
                      "IS_NULLABLE",
                      "SPECIFIC_NAME");
     }
@@ -1054,10 +1056,10 @@ class JdbcDatabaseMetaData implements DatabaseMetaData, JdbcWrapper {
                      "TABLE_SCHEM",
                      "TABLE_NAME",
                      "COLUMN_NAME",
-                     "DATA_TYPE", int.class,
+                     "DATA_TYPE", INTEGER,
-                     "COLUMN_SIZE", int.class,
+                     "COLUMN_SIZE", INTEGER,
-                     "DECIMAL_DIGITS", int.class,
+                     "DECIMAL_DIGITS", INTEGER,
-                     "NUM_PREC_RADIX", int.class,
+                     "NUM_PREC_RADIX", INTEGER,
                      "REMARKS",
                      "COLUMN_USAGE",
                      "IS_NULLABLE");
@@ -1078,8 +1080,8 @@ class JdbcDatabaseMetaData implements DatabaseMetaData, JdbcWrapper {
                 JDBCType type = JDBCType.VARCHAR;
                 if (i + 1 < cols.length) {
                     // check if the next item it's a type
-                    if (cols[i + 1] instanceof Class) {
+                    if (cols[i + 1] instanceof JDBCType) {
-                        type = JDBCType.valueOf(JdbcUtils.fromClass((Class<?>) cols[i + 1]));
+                        type = (JDBCType) cols[i + 1];
                         i++;
                     }
                     // it's not, use the default and move on

plugin/sql/jdbc/src/main/java/org/elasticsearch/xpack/sql/jdbc/jdbc/JdbcParameterMetaData.java

@@ -32,7 +32,7 @@ class JdbcParameterMetaData implements ParameterMetaData, JdbcWrapper {
 
     @Override
     public boolean isSigned(int param) throws SQLException {
-        return JdbcUtils.isSigned(paramInfo(param).type.getVendorTypeNumber().intValue());
+        return TypeConverter.isSigned(paramInfo(param).type);
     }
 
     @Override
@@ -49,7 +49,7 @@ class JdbcParameterMetaData implements ParameterMetaData, JdbcWrapper {
 
     @Override
     public int getParameterType(int param) throws SQLException {
-        return paramInfo(param).type.getVendorTypeNumber().intValue();
+        return paramInfo(param).type.getVendorTypeNumber();
     }
 
     @Override
@@ -59,7 +59,7 @@ class JdbcParameterMetaData implements ParameterMetaData, JdbcWrapper {
 
     @Override
     public String getParameterClassName(int param) throws SQLException {
-        return JdbcUtils.classOf(paramInfo(param).type.getVendorTypeNumber()).getName();
+        return TypeConverter.classNameOf(paramInfo(param).type);
     }
 
     @Override

plugin/sql/jdbc/src/main/java/org/elasticsearch/xpack/sql/jdbc/jdbc/JdbcResultSet.java

@@ -369,12 +369,7 @@ class JdbcResultSet implements ResultSet, JdbcWrapper {
 
         JDBCType columnType = cursor.columns().get(columnIndex - 1).type;
         
-        T t = TypeConverter.convert(val, columnType, type);
+        return TypeConverter.convert(val, columnType, type);
-        
-        if (t != null || type == null) {
-            return t;
-        }
-        throw new SQLException("Conversion from type [" + columnType + "] to [" + type.getName() + "] not supported");
     }
 
     @Override

plugin/sql/jdbc/src/main/java/org/elasticsearch/xpack/sql/jdbc/jdbc/JdbcResultSetMetaData.java

@@ -62,7 +62,7 @@ class JdbcResultSetMetaData implements ResultSetMetaData, JdbcWrapper {
 
     @Override
     public boolean isSigned(int column) throws SQLException {
-        return JdbcUtils.isSigned(getColumnType(column));
+        return TypeConverter.isSigned(column(column).type);
     }
 
     @Override
@@ -137,7 +137,7 @@ class JdbcResultSetMetaData implements ResultSetMetaData, JdbcWrapper {
 
     @Override
     public String getColumnClassName(int column) throws SQLException {
-        return JdbcUtils.classOf(column(column).type.getVendorTypeNumber()).getName();
+        return TypeConverter.classNameOf(column(column).type);
     }
 
     private void checkOpen() throws SQLException {

plugin/sql/jdbc/src/main/java/org/elasticsearch/xpack/sql/jdbc/jdbc/JdbcUtils.java

@@ -1,169 +0,0 @@
-/*
- * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
- * or more contributor license agreements. Licensed under the Elastic License;
- * you may not use this file except in compliance with the Elastic License.
- */
-package org.elasticsearch.xpack.sql.jdbc.jdbc;
-
-import org.elasticsearch.xpack.sql.jdbc.JdbcSQLException;
-
-import java.math.BigDecimal;
-import java.sql.Blob;
-import java.sql.Clob;
-import java.sql.Date;
-import java.sql.JDBCType;
-import java.sql.Time;
-import java.sql.Timestamp;
-
-import static java.sql.Types.BIGINT;
-import static java.sql.Types.BINARY;
-import static java.sql.Types.BIT;
-import static java.sql.Types.BLOB;
-import static java.sql.Types.BOOLEAN;
-import static java.sql.Types.CHAR;
-import static java.sql.Types.CLOB;
-import static java.sql.Types.DATE;
-import static java.sql.Types.DECIMAL;
-import static java.sql.Types.DOUBLE;
-import static java.sql.Types.FLOAT;
-import static java.sql.Types.INTEGER;
-import static java.sql.Types.LONGVARBINARY;
-import static java.sql.Types.LONGVARCHAR;
-import static java.sql.Types.NULL;
-import static java.sql.Types.NUMERIC;
-import static java.sql.Types.REAL;
-import static java.sql.Types.SMALLINT;
-import static java.sql.Types.TIME;
-import static java.sql.Types.TIMESTAMP;
-import static java.sql.Types.TIMESTAMP_WITH_TIMEZONE;
-import static java.sql.Types.TINYINT;
-import static java.sql.Types.VARBINARY;
-import static java.sql.Types.VARCHAR;
-
-public abstract class JdbcUtils {
-
-    public static int fromClass(Class<?> clazz) throws JdbcSQLException {
-        if (clazz == null) {
-            return NULL;
-        }
-        if (clazz == String.class) {
-            return VARCHAR;
-        }
-        if (clazz == Boolean.class || clazz == boolean.class) {
-            return BOOLEAN;
-        }
-        if (clazz == Byte.class || clazz == byte.class) {
-            return TINYINT;
-        }
-        if (clazz == Short.class || clazz == short.class) {
-            return SMALLINT;
-        }
-        if (clazz == Integer.class || clazz == int.class) {
-            return INTEGER;
-        }
-        if (clazz == Long.class || clazz == long.class) {
-            return BIGINT;
-        }
-        if (clazz == Float.class || clazz == float.class) {
-            return REAL;
-        }
-        if (clazz == Double.class || clazz == double.class) {
-            return DOUBLE;
-        }
-        if (clazz == Void.class || clazz == void.class) {
-            return NULL;
-        }
-        if (clazz == byte[].class) {
-            return VARBINARY;
-        }
-        if (clazz == Date.class) {
-            return DATE;
-        }
-        if (clazz == Time.class) {
-            return TIME;
-        }
-        if (clazz == Timestamp.class) {
-            return TIMESTAMP;
-        }
-        if (clazz == Blob.class) {
-            return BLOB;
-        }
-        if (clazz == Clob.class) {
-            return CLOB;
-        }
-        if (clazz == BigDecimal.class) {
-            return DECIMAL;
-        }
-
-        throw new JdbcSQLException("Unrecognized class [" + clazz + "]");
-    }
-
-    // see javax.sql.rowset.RowSetMetaDataImpl
-    // and https://db.apache.org/derby/docs/10.5/ref/rrefjdbc20377.html
-    public static Class<?> classOf(int jdbcType) throws JdbcSQLException {
-
-        switch (jdbcType) {
-            case NUMERIC:
-            case DECIMAL:
-                return BigDecimal.class;
-            case BOOLEAN:
-            case BIT:
-                return Boolean.class;
-            case TINYINT:
-                return Byte.class;
-            case SMALLINT:
-                return Short.class;
-            case INTEGER:
-                return Integer.class;
-            case BIGINT:
-                return Long.class;
-            case REAL:
-                return Float.class;
-            case FLOAT:
-            case DOUBLE:
-                return Double.class;
-            case BINARY:
-            case VARBINARY:
-            case LONGVARBINARY:
-                return byte[].class;
-            case CHAR:
-            case VARCHAR:
-            case LONGVARCHAR:
-                return String.class;
-            case DATE:
-                return Date.class;
-            case TIME:
-                return Time.class;
-            case TIMESTAMP:
-                return Timestamp.class;
-            case BLOB:
-                return Blob.class;
-            case CLOB:
-                return Clob.class;
-            case TIMESTAMP_WITH_TIMEZONE:
-                return Long.class;
-            default:
-                throw new JdbcSQLException("Unsupported JDBC type " + jdbcType + ", " + type(jdbcType).getName() + "");
-        }
-    }
-
-    static boolean isSigned(int type) {
-        switch (type) {
-            case BIGINT:
-            case DECIMAL:
-            case DOUBLE:
-            case FLOAT:
-            case INTEGER:
-            case SMALLINT:
-            case REAL:
-            case NUMERIC:
-                return true;
-            default:
-                return false;
-        }
-    }
-
-    static JDBCType type(int jdbcType) {
-        return JDBCType.valueOf(jdbcType);
-    }
-}

plugin/sql/jdbc/src/main/java/org/elasticsearch/xpack/sql/jdbc/jdbc/TypeConverter.java

@@ -5,6 +5,11 @@
  */
 package org.elasticsearch.xpack.sql.jdbc.jdbc;
 
+import org.elasticsearch.xpack.sql.jdbc.JdbcSQLException;
+
+import java.math.BigDecimal;
+import java.sql.Blob;
+import java.sql.Clob;
 import java.sql.Date;
 import java.sql.JDBCType;
 import java.sql.SQLException;
@@ -30,10 +35,28 @@ import static java.util.Calendar.MONTH;
 import static java.util.Calendar.SECOND;
 import static java.util.Calendar.YEAR;
 
-abstract class TypeConverter {
+/**
+ * Conversion utilities for conversion of JDBC types to Java type and back
+ * <p>
+ * The following JDBC types are supported as part of Elasticsearch Response. See org.elasticsearch.xpack.sql.type.DataType for details.
+ * <p>
+ * NULL, BOOLEAN, TINYINT, SMALLINT, INTEGER, BIGINT, DOUBLE, REAL, FLOAT, VARCHAR, VARBINARY and TIMESTAMP
+ * <p>
+ * The following additional types are also supported as parameters:
+ * <p>
+ * NUMERIC, DECIMAL, BIT, BINARY, LONGVARBINARY, CHAR, LONGVARCHAR, DATE, TIME, BLOB, CLOB, TIMESTAMP_WITH_TIMEZONE
+ */
+final class TypeConverter {
+
+    private TypeConverter() {
+
+    }
 
     private static final long DAY_IN_MILLIS = 60 * 60 * 24;
 
+    /**
+     * Converts millisecond after epoc to date
+     */
     static Date convertDate(Long millis, Calendar cal) {
         return dateTimeConvert(millis, cal, c -> {
             c.set(HOUR_OF_DAY, 0);
@@ -44,6 +67,9 @@ abstract class TypeConverter {
         });
     }
 
+    /**
+     * Converts millisecond after epoc to time
+     */
     static Time convertTime(Long millis, Calendar cal) {
         return dateTimeConvert(millis, cal, c -> {
             c.set(ERA, GregorianCalendar.AD);
@@ -54,10 +80,11 @@ abstract class TypeConverter {
         });
     }
 
+    /**
+     * Converts millisecond after epoc to timestamp
+     */
     static Timestamp convertTimestamp(Long millis, Calendar cal) {
-        return dateTimeConvert(millis, cal, c -> {
+        return dateTimeConvert(millis, cal, c -> new Timestamp(c.getTimeInMillis()));
-            return new Timestamp(c.getTimeInMillis());
-        });
     }
 
     private static <T> T dateTimeConvert(Long millis, Calendar c, Function<Calendar, T> creator) {
@@ -66,20 +93,23 @@ abstract class TypeConverter {
         }
         long initial = c.getTimeInMillis();
         try {
-            c.setTimeInMillis(millis.longValue());
+            c.setTimeInMillis(millis);
             return creator.apply(c);
         } finally {
             c.setTimeInMillis(initial);
         }
     }
 
+    /**
+     * Converts object val from columnType to type
+     */
     @SuppressWarnings("unchecked")
     static <T> T convert(Object val, JDBCType columnType, Class<T> type) throws SQLException {
         if (type == null) {
-            return (T) asNative(val, columnType);
+            return (T) convert(val, columnType);
         }
         if (type == String.class) {
-            return (T) asString(asNative(val, columnType));
+            return (T) asString(convert(val, columnType));
         }
         if (type == Boolean.class) {
             return (T) asBoolean(val, columnType);
@@ -132,43 +162,145 @@ abstract class TypeConverter {
         if (type == OffsetDateTime.class) {
             return (T) asOffsetDateTime(val, columnType);
         }
-        return null;
+        throw new SQLException("Conversion from type [" + columnType + "] to [" + type.getName() + "] not supported");
     }
 
-    // keep in check with JdbcUtils#columnType
+    /**
-    static Object asNative(Object v, JDBCType columnType) {
+     * Translates numeric JDBC type into corresponding Java class
-        switch (columnType) {
+     * <p>
+     * See {@link javax.sql.rowset.RowSetMetaDataImpl#getColumnClassName} and
+     * https://db.apache.org/derby/docs/10.5/ref/rrefjdbc20377.html
+     */
+    public static String classNameOf(JDBCType jdbcType) throws JdbcSQLException {
+        switch (jdbcType) {
+
+            // ES - supported types
+            case BOOLEAN:
+                return Boolean.class.getName();
+            case TINYINT: // BYTE DataType
+                return Byte.class.getName();
+            case SMALLINT: // SHORT DataType
+                return Short.class.getName();
+            case INTEGER:
+                return Integer.class.getName();
+            case BIGINT: // LONG DataType
+                return Long.class.getName();
+            case DOUBLE:
+                return Double.class.getName();
+            case REAL: // FLOAT DataType
+                return Float.class.getName();
+            case FLOAT: // HALF_FLOAT DataType
+                return Double.class.getName(); // TODO: Is this correct?
+            case VARCHAR: // KEYWORD or TEXT DataType
+                return String.class.getName();
+            case VARBINARY: // BINARY DataType
+                return byte[].class.getName();
+            case TIMESTAMP: // DATE DataType
+                return Timestamp.class.getName();
+
+            // Parameters data types that cannot be returned by ES but can appear in client - supplied parameters
+            case NUMERIC:
+            case DECIMAL:
+                return BigDecimal.class.getName();
             case BIT:
+                return Boolean.class.getName();
+            case BINARY:
+            case LONGVARBINARY:
+                return byte[].class.getName();
+            case CHAR:
+            case LONGVARCHAR:
+                return String.class.getName();
+            case DATE:
+                return Date.class.getName();
+            case TIME:
+                return Time.class.getName();
+            case BLOB:
+                return Blob.class.getName();
+            case CLOB:
+                return Clob.class.getName();
+            case TIMESTAMP_WITH_TIMEZONE:
+                return Long.class.getName();
+            default:
+                throw new JdbcSQLException("Unsupported JDBC type [" + jdbcType + "]");
+        }
+    }
+
+    /**
+     * Converts the object from JSON representation to the specified JDBCType
+     * <p>
+     * The returned types needs to correspond to ES-portion of classes returned by {@link TypeConverter#classNameOf}
+     */
+    static Object convert(Object v, JDBCType columnType) throws SQLException {
+        switch (columnType) {
+            case NULL:
+                return null;
             case BOOLEAN:
             case BINARY:
             case VARBINARY:
-            case LONGVARBINARY:
-            case CHAR:
             case VARCHAR:
-            case LONGVARCHAR:
+                return v;  // These types are already represented correctly in JSON
-                return v;
             case TINYINT:
-                return ((Number) v).byteValue();
+                return ((Number) v).byteValue();  // Parser might return it as integer or long - need to update to the correct type
             case SMALLINT:
-                return ((Number) v).shortValue();
+                return ((Number) v).shortValue(); // Parser might return it as integer or long - need to update to the correct type
             case INTEGER:
                 return ((Number) v).intValue();
             case BIGINT:
                 return ((Number) v).longValue();
             case FLOAT:
             case DOUBLE:
-                return doubleValue(v);
+                return doubleValue(v); // Double might be represented as string for infinity and NaN values
             case REAL:
-                return floatValue(v);
+                return floatValue(v);  // Float might be represented as string for infinity and NaN values
             case TIMESTAMP:
                 return ((Number) v).longValue();
-            // since the date is already in UTC_CALENDAR just do calendar math
-            case DATE:
-                return new Date(utcMillisRemoveTime(((Number) v).longValue()));
-            case TIME:
-                return new Time(utcMillisRemoveDate(((Number) v).longValue()));
             default:
-                return null;
+                throw new SQLException("Unexpected column type [" + columnType.getName() + "]");
+
+        }
+    }
+
+    /**
+     * Returns true if the type represents a signed number, false otherwise
+     * <p>
+     * It needs to support both params and column types
+     */
+    static boolean isSigned(JDBCType type) throws SQLException {
+        switch (type) {
+            // ES Supported types
+            case BIGINT:
+            case DOUBLE:
+            case FLOAT:
+            case INTEGER:
+            case TINYINT:
+            case SMALLINT:
+                return true;
+            case NULL:
+            case BOOLEAN:
+            case VARCHAR:
+            case VARBINARY:
+            case TIMESTAMP:
+                return false;
+
+            // Parameter types
+            case REAL:
+            case DECIMAL:
+            case NUMERIC:
+                return true;
+            case BIT:
+            case BINARY:
+            case LONGVARBINARY:
+            case CHAR:
+            case LONGVARCHAR:
+            case DATE:
+            case TIME:
+            case BLOB:
+            case CLOB:
+            case TIMESTAMP_WITH_TIMEZONE:
+                return false;
+
+            default:
+                throw new SQLException("Unexpected column or parameter type [" + type + "]");
         }
     }
 
@@ -208,9 +340,8 @@ abstract class TypeConverter {
         return nativeValue == null ? null : String.valueOf(nativeValue);
     }
 
-    private static Boolean asBoolean(Object val, JDBCType columnType) {
+    private static Boolean asBoolean(Object val, JDBCType columnType) throws SQLException {
         switch (columnType) {
-            case BIT:
             case BOOLEAN:
             case TINYINT:
             case SMALLINT:
@@ -220,14 +351,14 @@ abstract class TypeConverter {
             case FLOAT:
             case DOUBLE:
                 return Boolean.valueOf(Integer.signum(((Number) val).intValue()) == 0);
-             default:
+            default:
-                 return null;
+                throw new SQLException("Conversion from type [" + columnType + "] to [Boolean] not supported");
+
         }
     }
 
     private static Byte asByte(Object val, JDBCType columnType) throws SQLException {
         switch (columnType) {
-            case BIT:
             case BOOLEAN:
                 return Byte.valueOf(((Boolean) val).booleanValue() ? (byte) 1 : (byte) 0);
             case TINYINT:
@@ -242,12 +373,11 @@ abstract class TypeConverter {
             default:
         }
 
-        return null;
+        throw new SQLException("Conversion from type [" + columnType + "] to [Byte] not supported");
     }
 
     private static Short asShort(Object val, JDBCType columnType) throws SQLException {
         switch (columnType) {
-            case BIT:
             case BOOLEAN:
                 return Short.valueOf(((Boolean) val).booleanValue() ? (short) 1 : (short) 0);
             case TINYINT:
@@ -262,12 +392,11 @@ abstract class TypeConverter {
             default:
         }
 
-        return null;
+        throw new SQLException("Conversion from type [" + columnType + "] to [Short] not supported");
     }
 
     private static Integer asInteger(Object val, JDBCType columnType) throws SQLException {
         switch (columnType) {
-            case BIT:
             case BOOLEAN:
                 return Integer.valueOf(((Boolean) val).booleanValue() ? 1 : 0);
             case TINYINT:
@@ -282,12 +411,11 @@ abstract class TypeConverter {
             default:
         }
 
-        return null;
+        throw new SQLException("Conversion from type [" + columnType + "] to [Integer] not supported");
     }
 
     private static Long asLong(Object val, JDBCType columnType) throws SQLException {
         switch (columnType) {
-            case BIT:
             case BOOLEAN:
                 return Long.valueOf(((Boolean) val).booleanValue() ? 1 : 0);
             case TINYINT:
@@ -309,12 +437,11 @@ abstract class TypeConverter {
             default:
         }
 
-        return null;
+        throw new SQLException("Conversion from type [" + columnType + "] to [Long] not supported");
     }
 
     private static Float asFloat(Object val, JDBCType columnType) throws SQLException {
         switch (columnType) {
-            case BIT:
             case BOOLEAN:
                 return Float.valueOf(((Boolean) val).booleanValue() ? 1 : 0);
             case TINYINT:
@@ -329,12 +456,11 @@ abstract class TypeConverter {
             default:
         }
 
-        return null;
+        throw new SQLException("Conversion from type [" + columnType + "] to [Float] not supported");
     }
 
     private static Double asDouble(Object val, JDBCType columnType) throws SQLException {
         switch (columnType) {
-            case BIT:
             case BOOLEAN:
                 return Double.valueOf(((Boolean) val).booleanValue() ? 1 : 0);
             case TINYINT:
@@ -349,7 +475,7 @@ abstract class TypeConverter {
             default:
         }
 
-        return null;
+        throw new SQLException("Conversion from type [" + columnType + "] to [Double] not supported");
     }
 
     private static Date asDate(Object val, JDBCType columnType) throws SQLException {
@@ -364,7 +490,7 @@ abstract class TypeConverter {
             default:
         }
 
-        return null;
+        throw new SQLException("Conversion from type [" + columnType + "] to [Date] not supported");
     }
 
     private static Time asTime(Object val, JDBCType columnType) throws SQLException {
@@ -379,7 +505,7 @@ abstract class TypeConverter {
             default:
         }
 
-        return null;
+        throw new SQLException("Conversion from type [" + columnType + "] to [Time] not supported");
     }
 
     private static Timestamp asTimestamp(Object val, JDBCType columnType) throws SQLException {
@@ -394,12 +520,13 @@ abstract class TypeConverter {
             default:
         }
 
-        return null;
+        throw new SQLException("Conversion from type [" + columnType + "] to [Timestamp] not supported");
     }
 
     private static byte[] asByteArray(Object val, JDBCType columnType) {
         throw new UnsupportedOperationException();
     }
+
     private static LocalDate asLocalDate(Object val, JDBCType columnType) {
         throw new UnsupportedOperationException();
     }

plugin/sql/jdbc/src/test/java/org/elasticsearch/xpack/sql/jdbc/jdbc/TypeConverterTests.java

@@ -13,7 +13,6 @@ import org.elasticsearch.xpack.sql.plugin.AbstractSqlRequest;
 import org.elasticsearch.xpack.sql.plugin.SqlQueryResponse;
 import org.joda.time.DateTime;
 
-import java.io.IOException;
 import java.sql.JDBCType;
 
 import static org.hamcrest.Matchers.instanceOf;
@@ -22,7 +21,7 @@ import static org.hamcrest.Matchers.instanceOf;
 public class TypeConverterTests extends ESTestCase {
 
 
-    public void testFloatAsNative() throws IOException {
+    public void testFloatAsNative() throws Exception {
         assertThat(convertAsNative(42.0f, JDBCType.REAL), instanceOf(Float.class));
         assertThat(convertAsNative(42.0, JDBCType.REAL), instanceOf(Float.class));
         assertEquals(42.0f, (float) convertAsNative(42.0, JDBCType.REAL), 0.001f);
@@ -31,7 +30,7 @@ public class TypeConverterTests extends ESTestCase {
         assertEquals(Float.POSITIVE_INFINITY, convertAsNative(Float.POSITIVE_INFINITY, JDBCType.REAL));
     }
 
-    public void testDoubleAsNative() throws IOException {
+    public void testDoubleAsNative() throws Exception {
         JDBCType type = randomFrom(JDBCType.FLOAT, JDBCType.DOUBLE);
         assertThat(convertAsNative(42.0, type), instanceOf(Double.class));
         assertEquals(42.0f, (double) convertAsNative(42.0, type), 0.001f);
@@ -40,13 +39,13 @@ public class TypeConverterTests extends ESTestCase {
         assertEquals(Double.POSITIVE_INFINITY, convertAsNative(Double.POSITIVE_INFINITY, type));
     }
 
-    public void testTimestampAsNative() throws IOException {
+    public void testTimestampAsNative() throws Exception {
         DateTime now = DateTime.now();
         assertThat(convertAsNative(now, JDBCType.TIMESTAMP), instanceOf(Long.class));
         assertEquals(now.getMillis(), convertAsNative(now, JDBCType.TIMESTAMP));
     }
 
-    private Object convertAsNative(Object value, JDBCType type) throws IOException {
+    private Object convertAsNative(Object value, JDBCType type) throws Exception {
         // Simulate sending over XContent
         XContentBuilder builder = JsonXContent.contentBuilder();
         builder.startObject();
@@ -55,7 +54,7 @@ public class TypeConverterTests extends ESTestCase {
         builder.endObject();
         builder.close();
         Object copy = XContentHelper.convertToMap(builder.bytes(), false, builder.contentType()).v2().get("value");
-        return TypeConverter.asNative(copy, type);
+        return TypeConverter.convert(copy, type);
     }
 
 }

plugin/sql/src/main/java/org/elasticsearch/xpack/sql/plan/logical/command/sys/SysColumns.java

@@ -132,7 +132,10 @@ public class SysColumns extends Command {
                         type.size,
                         // no DECIMAL support
                         null,
-                        // RADIX
+                        // RADIX  - Determines how numbers returned by COLUMN_SIZE and DECIMAL_DIGITS should be interpreted.
+                        // 10 means they represent the number of decimal digits allowed for the column.
+                        // 2 means they represent the number of bits allowed for the column.
+                        // null means radix is not applicable for the given type.
                         type.isInteger ? Integer.valueOf(10) : type.isRational ? Integer.valueOf(2) : null,
                         // everything is nullable
                         DatabaseMetaData.columnNullable,

plugin/sql/src/main/java/org/elasticsearch/xpack/sql/plan/logical/command/sys/SysTypes.java

@@ -81,7 +81,7 @@ public class SysTypes extends Command {
                         // everything is searchable,
                         DatabaseMetaData.typeSearchable,
                         // only numerics are signed
-                        t.isNumeric() ? !t.isSigned : null,
+                        t.isSigned(),
                         //no fixed precision scale SQL_FALSE
                         0,
                         null,

plugin/sql/src/main/java/org/elasticsearch/xpack/sql/type/DataType.java

@@ -13,23 +13,23 @@ import java.util.Locale;
  */
 public enum DataType {
     // @formatter:off
-    //           jdbc type,          size,              defPrecision, dispSize, sig, int, rat, docvals
+    //           jdbc type,          size,              defPrecision, dispSize, int, rat, docvals
     NULL(        JDBCType.NULL,      0,                 0,                 0),
     UNSUPPORTED( JDBCType.OTHER,     0,                 0,                 0),
     BOOLEAN(     JDBCType.BOOLEAN,   1,                 1,                 1),
-    BYTE(        JDBCType.TINYINT,   Byte.BYTES,        3,                 5,  true, true, false, true),
+    BYTE(        JDBCType.TINYINT,   Byte.BYTES,        3,                 5, true, false, true),
-    SHORT(       JDBCType.SMALLINT,  Short.BYTES,       5,                 6,  true, true, false, true),
+    SHORT(       JDBCType.SMALLINT,  Short.BYTES,       5,                 6, true, false, true),
-    INTEGER(     JDBCType.INTEGER,   Integer.BYTES,     10,                11, true, true, false, true),
+    INTEGER(     JDBCType.INTEGER,   Integer.BYTES,     10,                11, true, false, true),
-    LONG(        JDBCType.BIGINT,    Long.BYTES,        19,                20, true, true, false, true),
+    LONG(        JDBCType.BIGINT,    Long.BYTES,        19,                20, true, false, true),
     // 53 bits defaultPrecision ~ 16(15.95) decimal digits (53log10(2)),
-    DOUBLE(      JDBCType.DOUBLE,    Double.BYTES,      16,                25, true, false, true, true),
+    DOUBLE(      JDBCType.DOUBLE,    Double.BYTES,      16,                25, false, true, true),
     // 24 bits defaultPrecision - 24*log10(2) =~ 7 (7.22)
-    FLOAT(       JDBCType.REAL,      Float.BYTES,       7,                 15, true, false, true, true),
+    FLOAT(       JDBCType.REAL,      Float.BYTES,       7,                 15, false, true, true),
-    HALF_FLOAT(  JDBCType.FLOAT,     Double.BYTES,      16,                25, true, false, true, true),
+    HALF_FLOAT(  JDBCType.FLOAT,     Double.BYTES,      16,                25, false, true, true),
     // precision is based on long
-    SCALED_FLOAT(JDBCType.FLOAT,     Double.BYTES,      19,                25, true, false, true, true),
+    SCALED_FLOAT(JDBCType.FLOAT,     Double.BYTES,      19,                25, false, true, true),
     KEYWORD(     JDBCType.VARCHAR,   Integer.MAX_VALUE, 256,               0),
-    TEXT(        JDBCType.VARCHAR,   Integer.MAX_VALUE, Integer.MAX_VALUE, 0,  false, false, false, false),
+    TEXT(        JDBCType.VARCHAR,   Integer.MAX_VALUE, Integer.MAX_VALUE, 0, false, false, false),
     OBJECT(      JDBCType.STRUCT,    -1,                0,                 0),
     NESTED(      JDBCType.STRUCT,    -1,                0,                 0),
     BINARY(      JDBCType.VARBINARY, -1,                Integer.MAX_VALUE, 0),
@@ -71,11 +71,6 @@ public enum DataType {
      */
     public final int displaySize;
 
-    /**
-     * True if the type represents a signed number
-     */
-    public final boolean isSigned;
-
     /**
      * True if the type represents an integer number
      */
@@ -91,21 +86,20 @@ public enum DataType {
      */
     public final boolean defaultDocValues;
 
-    DataType(JDBCType jdbcType, int size, int defaultPrecision, int displaySize, boolean isSigned, boolean isInteger, boolean isRational,
+    DataType(JDBCType jdbcType, int size, int defaultPrecision, int displaySize, boolean isInteger, boolean isRational,
              boolean defaultDocValues) {
         this.esType = name().toLowerCase(Locale.ROOT);
         this.jdbcType = jdbcType;
         this.size = size;
         this.defaultPrecision = defaultPrecision;
         this.displaySize = displaySize;
-        this.isSigned = isSigned;
         this.isInteger = isInteger;
         this.isRational = isRational;
         this.defaultDocValues = defaultDocValues;
     }
 
     DataType(JDBCType jdbcType, int size, int defaultPrecision, int displaySize) {
-        this(jdbcType, size, defaultPrecision, displaySize, false, false, false, true);
+        this(jdbcType, size, defaultPrecision, displaySize, false, false, true);
     }
 
     public String sqlName() {
@@ -116,6 +110,14 @@ public enum DataType {
         return isInteger || isRational;
     }
 
+    /**
+     * Returns true if value is signed, false if it is unsigned or null if the type doesn't represent a number
+     */
+    public Boolean isSigned() {
+        // For now all numeric values that es supports are signed
+        return isNumeric() ? true : null;
+    }
+
     public boolean isString() {
         return this == KEYWORD || this == TEXT;
     }

plugin/sql/src/main/java/org/elasticsearch/xpack/sql/type/DataTypeConversion.java

@@ -16,7 +16,8 @@ import java.util.function.Function;
 import java.util.function.LongFunction;
 
 /**
- * Conversions from one data type to another.
+ * Conversions from one Elasticsearch data type to another Elasticsearch data types.
+ * <p>
  * This class throws {@link SqlIllegalArgumentException} to differentiate between validation
  * errors inside SQL as oppose to the rest of ES.
  */
@@ -24,6 +25,13 @@ public abstract class DataTypeConversion {
 
     private static final DateTimeFormatter UTC_DATE_FORMATTER = ISODateTimeFormat.dateTimeNoMillis().withZoneUTC();
 
+    /**
+     * Returns the type compatible with both left and right types
+     * <p>
+     * If one of the types is null - returns another type
+     * If both types are numeric - returns type with the highest precision int < long < float < double
+     * If one of the types is string and another numeric - returns numeric
+     */
     public static DataType commonType(DataType left, DataType right) {
         if (left == right) {
             return left;
@@ -65,46 +73,30 @@ public abstract class DataTypeConversion {
         return null;
     }
 
-    public static boolean canConvert(DataType from, DataType to) { // TODO it'd be cleaner and more right to fetch the conversion
+    /**
+     * Returns true if the from type can be converted to the to type, false - otherwise
+     */
+    public static boolean canConvert(DataType from, DataType to) {
+        // Special handling for nulls and if conversion is not requires
+        if (from == to || from == DataType.NULL) {
+            return true;
+        }
         // only primitives are supported so far
-        if (!from.isPrimitive() || !to.isPrimitive()) {
+        return from.isPrimitive() && to.isPrimitive() && conversion(from, to) != null;
-            return false;
-        }
-
-        if (from.getClass() == to.getClass()) {
-            return true;
-        }
-        if (from == DataType.NULL) {
-            return true;
-        }
-
-        // anything can be converted to String
-        if (to.isString()) {
-            return true;
-        }
-
-        // also anything can be converted into a bool
-        if (to == DataType.BOOLEAN) {
-            return true;
-        }
-
-        // numeric conversion
-        if ((from.isString() || from == DataType.BOOLEAN || from == DataType.DATE || from.isNumeric()) && to.isNumeric()) {
-            return true;
-        }
-
-        // date conversion
-        if ((from == DataType.DATE || from.isString() || from.isNumeric()) && to == DataType.DATE) {
-            return true;
-        }
-
-        return false;
     }
 
     /**
      * Get the conversion from one type to another.
      */
     public static Conversion conversionFor(DataType from, DataType to) {
+        Conversion conversion = conversion(from, to);
+        if (conversion == null) {
+            throw new SqlIllegalArgumentException("cannot convert from [" + from + "] to [" + to + "]");
+        }
+        return conversion;
+    }
+
+    private static Conversion conversion(DataType from, DataType to) {
         switch (to) {
             case KEYWORD:
             case TEXT:
@@ -126,8 +118,9 @@ public abstract class DataTypeConversion {
             case BOOLEAN:
                 return conversionToBoolean(from);
             default:
-                throw new SqlIllegalArgumentException("cannot convert from [" + from + "] to [" + to + "]");
+                return null;
         }
+
     }
 
     private static Conversion conversionToString(DataType from) {
@@ -150,7 +143,7 @@ public abstract class DataTypeConversion {
         if (from.isString()) {
             return Conversion.STRING_TO_LONG;
         }
-        throw new SqlIllegalArgumentException("cannot convert from [" + from + "] to [Long]");
+        return null;
     }
 
     private static Conversion conversionToInt(DataType from) {
@@ -166,7 +159,7 @@ public abstract class DataTypeConversion {
         if (from.isString()) {
             return Conversion.STRING_TO_INT;
         }
-        throw new SqlIllegalArgumentException("cannot convert from [" + from + "] to [Integer]");
+        return null;
     }
 
     private static Conversion conversionToShort(DataType from) {
@@ -182,7 +175,7 @@ public abstract class DataTypeConversion {
         if (from.isString()) {
             return Conversion.STRING_TO_SHORT;
         }
-        throw new SqlIllegalArgumentException("cannot convert [" + from + "] to [Short]");
+        return null;
     }
 
     private static Conversion conversionToByte(DataType from) {
@@ -198,7 +191,7 @@ public abstract class DataTypeConversion {
         if (from.isString()) {
             return Conversion.STRING_TO_BYTE;
         }
-        throw new SqlIllegalArgumentException("cannot convert [" + from + "] to [Byte]");
+        return null;
     }
 
     private static Conversion conversionToFloat(DataType from) {
@@ -214,7 +207,7 @@ public abstract class DataTypeConversion {
         if (from.isString()) {
             return Conversion.STRING_TO_FLOAT;
         }
-        throw new SqlIllegalArgumentException("cannot convert [" + from + "] to [Float]");
+        return null;
     }
 
     private static Conversion conversionToDouble(DataType from) {
@@ -230,7 +223,7 @@ public abstract class DataTypeConversion {
         if (from.isString()) {
             return Conversion.STRING_TO_DOUBLE;
         }
-        throw new SqlIllegalArgumentException("cannot convert [" + from + "] to [Double]");
+        return null;
     }
 
     private static Conversion conversionToDate(DataType from) {
@@ -246,7 +239,7 @@ public abstract class DataTypeConversion {
         if (from.isString()) {
             return Conversion.STRING_TO_DATE;
         }
-        throw new SqlIllegalArgumentException("cannot convert [" + from + "] to [Date]");
+        return null;
     }
 
     private static Conversion conversionToBoolean(DataType from) {
@@ -256,7 +249,7 @@ public abstract class DataTypeConversion {
         if (from.isString()) {
             return Conversion.STRING_TO_BOOLEAN;
         }
-        throw new SqlIllegalArgumentException("cannot convert [" + from + "] to [Boolean]");
+        return null;
     }
 
     public static byte safeToByte(long x) {
@@ -295,9 +288,14 @@ public abstract class DataTypeConversion {
         return Booleans.parseBoolean(lowVal);
     }
 
+    /**
+     * Converts arbitrary object to the desired data type.
+     * <p>
+     * Throws SqlIllegalArgumentException if such conversion is not possible
+     */
     public static Object convert(Object value, DataType dataType) {
         DataType detectedType = DataTypes.fromJava(value);
-        if (detectedType.equals(dataType) || value == null) {
+        if (detectedType == dataType || value == null) {
             return value;
         }
         return conversionFor(detectedType, dataType).convert(value);

plugin/sql/src/test/java/org/elasticsearch/xpack/sql/type/DataTypeConversionTests.java

@@ -177,4 +177,18 @@ public class DataTypeConversionTests extends ESTestCase {
             assertEquals("[" + Short.MAX_VALUE + "] out of [Byte] range", e.getMessage());
         }
     }
+
+    public void testCommonType() {
+        assertEquals(DataType.BOOLEAN, DataTypeConversion.commonType(DataType.BOOLEAN, DataType.NULL));
+        assertEquals(DataType.BOOLEAN, DataTypeConversion.commonType(DataType.NULL, DataType.BOOLEAN));
+        assertEquals(DataType.BOOLEAN, DataTypeConversion.commonType(DataType.BOOLEAN, DataType.BOOLEAN));
+        assertEquals(DataType.NULL, DataTypeConversion.commonType(DataType.NULL, DataType.NULL));
+        assertEquals(DataType.INTEGER, DataTypeConversion.commonType(DataType.INTEGER, DataType.KEYWORD));
+        assertEquals(DataType.LONG, DataTypeConversion.commonType(DataType.TEXT, DataType.LONG));
+        assertEquals(null, DataTypeConversion.commonType(DataType.TEXT, DataType.KEYWORD));
+        assertEquals(DataType.SHORT, DataTypeConversion.commonType(DataType.SHORT, DataType.BYTE));
+        assertEquals(DataType.FLOAT, DataTypeConversion.commonType(DataType.BYTE, DataType.FLOAT));
+        assertEquals(DataType.FLOAT, DataTypeConversion.commonType(DataType.FLOAT, DataType.INTEGER));
+        assertEquals(DataType.DOUBLE, DataTypeConversion.commonType(DataType.DOUBLE, DataType.FLOAT));
+    }
 }