diff --git a/x-pack/plugin/sql/jdbc/src/main/java/org/elasticsearch/xpack/sql/jdbc/jdbc/JdbcResultSet.java b/x-pack/plugin/sql/jdbc/src/main/java/org/elasticsearch/xpack/sql/jdbc/jdbc/JdbcResultSet.java
index 201ae251ca0..ebdeaef15ca 100644
--- a/x-pack/plugin/sql/jdbc/src/main/java/org/elasticsearch/xpack/sql/jdbc/jdbc/JdbcResultSet.java
+++ b/x-pack/plugin/sql/jdbc/src/main/java/org/elasticsearch/xpack/sql/jdbc/jdbc/JdbcResultSet.java
@@ -133,72 +133,37 @@ class JdbcResultSet implements ResultSet, JdbcWrapper {
 
     @Override
     public boolean getBoolean(int columnIndex) throws SQLException {
-        Object val = column(columnIndex);
-        try {
-            return val != null ? (Boolean) val : false;
-        } catch (ClassCastException cce) {
-            throw new SQLException("unable to convert column " + columnIndex + " to a boolean", cce);
-        }
+        return column(columnIndex) != null ? getObject(columnIndex, Boolean.class) : false; 
     }
 
     @Override
     public byte getByte(int columnIndex) throws SQLException {
-        Object val = column(columnIndex);
-        try {
-            return val != null ? ((Number) val).byteValue() : 0;
-        } catch (ClassCastException cce) {
-            throw new SQLException("unable to convert column " + columnIndex + " to a byte", cce);
-        }
+        return column(columnIndex) != null ? getObject(columnIndex, Byte.class) : 0;
     }
 
     @Override
     public short getShort(int columnIndex) throws SQLException {
-        Object val = column(columnIndex);
-        try {
-            return val != null ? ((Number) val).shortValue() : 0;
-        } catch (ClassCastException cce) {
-            throw new SQLException("unable to convert column " + columnIndex + " to a short", cce);
-        }
+        return column(columnIndex) != null ? getObject(columnIndex, Short.class) : 0;
     }
 
     @Override
     public int getInt(int columnIndex) throws SQLException {
-        Object val = column(columnIndex);
-        try {
-            return val != null ? ((Number) val).intValue() : 0;
-        } catch (ClassCastException cce) {
-            throw new SQLException("unable to convert column " + columnIndex + " to an int", cce);
-        }
+        return column(columnIndex) != null ? getObject(columnIndex, Integer.class) : 0;
     }
 
     @Override
     public long getLong(int columnIndex) throws SQLException {
-        Object val = column(columnIndex);
-        try {
-            return val != null ? ((Number) val).longValue() : 0;
-        } catch (ClassCastException cce) {
-            throw new SQLException("unable to convert column " + columnIndex + " to a long", cce);
-        }
+        return column(columnIndex) != null ? getObject(columnIndex, Long.class) : 0;
     }
 
     @Override
     public float getFloat(int columnIndex) throws SQLException {
-        Object val = column(columnIndex);
-        try {
-            return val != null ? ((Number) val).floatValue() : 0;
-        } catch (ClassCastException cce) {
-            throw new SQLException("unable to convert column " + columnIndex + " to a float", cce);
-        }
+        return column(columnIndex) != null ? getObject(columnIndex, Float.class) : 0;
     }
 
     @Override
     public double getDouble(int columnIndex) throws SQLException {
-        Object val = column(columnIndex);
-        try {
-            return val != null ? ((Number) val).doubleValue() : 0;
-        } catch (ClassCastException cce) {
-            throw new SQLException("unable to convert column " + columnIndex + " to a double", cce);
-        }
+        return column(columnIndex) != null ? getObject(columnIndex, Double.class) : 0;
     }
 
     @Override
@@ -272,15 +237,29 @@ class JdbcResultSet implements ResultSet, JdbcWrapper {
 
     @Override
     public Date getDate(String columnLabel) throws SQLException {
+        // TODO: the error message in case the value in the column cannot be converted to a Date refers to a column index
+        // (for example - "unable to convert column 4 to a long") and not to the column name, which is a bit confusing.
+        // Should we reconsider this? Maybe by catching the exception here and rethrowing it with the columnLabel instead.
         return getDate(column(columnLabel));
     }
 
     private Long dateTime(int columnIndex) throws SQLException {
         Object val = column(columnIndex);
+        JDBCType type = cursor.columns().get(columnIndex - 1).type;
         try {
+            // TODO: the B6 appendix of the jdbc spec does mention CHAR, VARCHAR, LONGVARCHAR, DATE, TIMESTAMP as supported
+            // jdbc types that should be handled by getDate and getTime methods. From all of those we support VARCHAR and
+            // TIMESTAMP. Should we consider the VARCHAR conversion as a later enhancement?
+            if (JDBCType.TIMESTAMP.equals(type)) {
+                // the cursor can return an Integer if the date-since-epoch is small enough, XContentParser (Jackson) will
+                // return the "smallest" data type for numbers when parsing
+                // TODO: this should probably be handled server side
+                return val == null ? null : ((Number) val).longValue();
+            };
             return val == null ? null : (Long) val;
         } catch (ClassCastException cce) {
-            throw new SQLException("unable to convert column " + columnIndex + " to a long", cce);
+            throw new SQLException(
+                    format(Locale.ROOT, "Unable to convert value [%.128s] of type [%s] to a Long", val, type.getName()), cce);
         }
     }
 
diff --git a/x-pack/plugin/sql/jdbc/src/main/java/org/elasticsearch/xpack/sql/jdbc/jdbc/TypeConverter.java b/x-pack/plugin/sql/jdbc/src/main/java/org/elasticsearch/xpack/sql/jdbc/jdbc/TypeConverter.java
index 3b5180b71f7..7b638d8bd09 100644
--- a/x-pack/plugin/sql/jdbc/src/main/java/org/elasticsearch/xpack/sql/jdbc/jdbc/TypeConverter.java
+++ b/x-pack/plugin/sql/jdbc/src/main/java/org/elasticsearch/xpack/sql/jdbc/jdbc/TypeConverter.java
@@ -10,7 +10,6 @@ import org.elasticsearch.xpack.sql.type.DataType;
 
 import java.sql.Date;
 import java.sql.JDBCType;
-import java.sql.SQLDataException;
 import java.sql.SQLException;
 import java.sql.SQLFeatureNotSupportedException;
 import java.sql.Time;
@@ -56,9 +55,10 @@ final class TypeConverter {
 
     }
 
-    private static final long DAY_IN_MILLIS = 60 * 60 * 24;
+    private static final long DAY_IN_MILLIS = 60 * 60 * 24 * 1000;
     private static final Map<Class<?>, JDBCType> javaToJDBC;
 
+
     static {
         Map<Class<?>, JDBCType> aMap = Arrays.stream(DataType.values())
                 .filter(dataType -> dataType.javaClass() != null
@@ -120,6 +120,7 @@ final class TypeConverter {
         }
     }
 
+
     static long convertFromCalendarToUTC(long value, Calendar cal) {
         if (cal == null) {
             return value;
@@ -143,11 +144,15 @@ final class TypeConverter {
             return (T) convert(val, columnType);
         }
 
-        if (type.isInstance(val)) {
+        // converting a Long to a Timestamp shouldn't be possible according to the spec,
+        // it feels a little brittle to check this scenario here and I don't particularly like it
+        // TODO: can we do any better or should we go over the spec and allow getLong(date) to be valid?
+        if (!(type == Long.class && columnType == JDBCType.TIMESTAMP) && type.isInstance(val)) {
             try {
                 return type.cast(val);
             } catch (ClassCastException cce) {
-                throw new SQLDataException("Unable to convert " + val.getClass().getName() + " to " + columnType, cce);
+                throw new SQLException(format(Locale.ROOT, "Unable to convert value [%.128s] of type [%s] to a %s", val, 
+                        columnType.getName(), type.getName()), cce);
             }
         }
 
@@ -205,7 +210,8 @@ final class TypeConverter {
         if (type == OffsetDateTime.class) {
             return (T) asOffsetDateTime(val, columnType);
         }
-        throw new SQLException("Conversion from type [" + columnType + "] to [" + type.getName() + "] not supported");
+        throw new SQLException(format(Locale.ROOT, "Unable to convert value [%.128s] of type [%s] to a %s", val, 
+                columnType.getName(), type.getName()));
     }
 
     /**
@@ -336,8 +342,11 @@ final class TypeConverter {
             case FLOAT:
             case DOUBLE:
                 return Boolean.valueOf(Integer.signum(((Number) val).intValue()) != 0);
+            case VARCHAR:
+                return Boolean.valueOf((String) val);
             default:
-                throw new SQLException("Conversion from type [" + columnType + "] to [Boolean] not supported");
+                throw new SQLException(
+                        format(Locale.ROOT, "Unable to convert value [%.128s] of type [%s] to a Boolean", val, columnType.getName()));
 
         }
     }
@@ -355,10 +364,16 @@ final class TypeConverter {
             case FLOAT:
             case DOUBLE:
                 return safeToByte(safeToLong(((Number) val).doubleValue()));
+            case VARCHAR:
+                try {
+                    return Byte.valueOf((String) val);
+                } catch (NumberFormatException e) {
+                    throw new SQLException(format(Locale.ROOT, "Unable to convert value [%.128s] of type [VARCHAR] to a Byte", val), e);
+                }
             default:
         }
 
-        throw new SQLException("Conversion from type [" + columnType + "] to [Byte] not supported");
+        throw new SQLException(format(Locale.ROOT, "Unable to convert value [%.128s] of type [%s] to a Byte", val, columnType.getName()));
     }
 
     private static Short asShort(Object val, JDBCType columnType) throws SQLException {
@@ -374,10 +389,16 @@ final class TypeConverter {
             case FLOAT:
             case DOUBLE:
                 return safeToShort(safeToLong(((Number) val).doubleValue()));
+            case VARCHAR:
+                try {
+                    return Short.valueOf((String) val);
+                } catch (NumberFormatException e) {
+                    throw new SQLException(format(Locale.ROOT, "Unable to convert value [%.128s] of type [VARCHAR] to a Short", val), e);
+                }
             default:
         }
 
-        throw new SQLException("Conversion from type [" + columnType + "] to [Short] not supported");
+        throw new SQLException(format(Locale.ROOT, "Unable to convert value [%.128s] of type [%s] to a Short", val, columnType.getName()));
     }
 
     private static Integer asInteger(Object val, JDBCType columnType) throws SQLException {
@@ -393,10 +414,18 @@ final class TypeConverter {
             case FLOAT:
             case DOUBLE:
                 return safeToInt(safeToLong(((Number) val).doubleValue()));
+            case VARCHAR:
+                try {
+                    return Integer.valueOf((String) val);
+                } catch (NumberFormatException e) {
+                    throw new SQLException(
+                            format(Locale.ROOT, "Unable to convert value [%.128s] of type [VARCHAR] to an Integer", val), e);
+                }
             default:
         }
 
-        throw new SQLException("Conversion from type [" + columnType + "] to [Integer] not supported");
+        throw new SQLException(
+                format(Locale.ROOT, "Unable to convert value [%.128s] of type [%s] to an Integer", val, columnType.getName()));
     }
 
     private static Long asLong(Object val, JDBCType columnType) throws SQLException {
@@ -412,12 +441,21 @@ final class TypeConverter {
             case FLOAT:
             case DOUBLE:
                 return safeToLong(((Number) val).doubleValue());
-            case TIMESTAMP:
-                return ((Number) val).longValue();
+            //TODO: should we support conversion to TIMESTAMP?
+            //The spec says that getLong() should support the following types conversions:
+            //TINYINT, SMALLINT, INTEGER, BIGINT, REAL, FLOAT, DOUBLE, DECIMAL, NUMERIC, BIT, BOOLEAN, CHAR, VARCHAR, LONGVARCHAR
+            //case TIMESTAMP:
+            //    return ((Number) val).longValue();
+            case VARCHAR:
+                try {
+                    return Long.valueOf((String) val);
+                } catch (NumberFormatException e) {
+                    throw new SQLException(format(Locale.ROOT, "Unable to convert value [%.128s] of type [VARCHAR] to a Long", val), e);
+                }
             default:
         }
 
-        throw new SQLException("Conversion from type [" + columnType + "] to [Long] not supported");
+        throw new SQLException(format(Locale.ROOT, "Unable to convert value [%.128s] of type [%s] to a Long", val, columnType.getName()));
     }
 
     private static Float asFloat(Object val, JDBCType columnType) throws SQLException {
@@ -433,10 +471,16 @@ final class TypeConverter {
             case FLOAT:
             case DOUBLE:
                 return Float.valueOf((((float) ((Number) val).doubleValue())));
+            case VARCHAR:
+                try {
+                    return Float.valueOf((String) val);
+                } catch (NumberFormatException e) {
+                    throw new SQLException(format(Locale.ROOT, "Unable to convert value [%.128s] of type [VARCHAR] to a Float", val), e);
+                }
             default:
         }
 
-        throw new SQLException("Conversion from type [" + columnType + "] to [Float] not supported");
+        throw new SQLException(format(Locale.ROOT, "Unable to convert value [%.128s] of type [%s] to a Float", val, columnType.getName()));
     }
 
     private static Double asDouble(Object val, JDBCType columnType) throws SQLException {
@@ -451,32 +495,41 @@ final class TypeConverter {
             case REAL:
             case FLOAT:
             case DOUBLE:
+
                 return Double.valueOf(((Number) val).doubleValue());
+            case VARCHAR:
+                try {
+                    return Double.valueOf((String) val);
+                } catch (NumberFormatException e) {
+                    throw new SQLException(format(Locale.ROOT, "Unable to convert value [%.128s] of type [VARCHAR] to a Double", val), e);
+                }
             default:
         }
 
-        throw new SQLException("Conversion from type [" + columnType + "] to [Double] not supported");
+        throw new SQLException(
+                format(Locale.ROOT, "Unable to convert value [%.128s] of type [%s] to a Double", val, columnType.getName()));
     }
 
     private static Date asDate(Object val, JDBCType columnType) throws SQLException {
         if (columnType == JDBCType.TIMESTAMP) {
             return new Date(utcMillisRemoveTime(((Number) val).longValue()));
         }
-        throw new SQLException("Conversion from type [" + columnType + "] to [Date] not supported");
+        throw new SQLException(format(Locale.ROOT, "Unable to convert value [%.128s] of type [%s] to a Date", val, columnType.getName()));
     }
 
     private static Time asTime(Object val, JDBCType columnType) throws SQLException {
         if (columnType == JDBCType.TIMESTAMP) {
             return new Time(utcMillisRemoveDate(((Number) val).longValue()));
         }
-        throw new SQLException("Conversion from type [" + columnType + "] to [Time] not supported");
+        throw new SQLException(format(Locale.ROOT, "Unable to convert value [%.128s] of type [%s] to a Time", val, columnType.getName()));
     }
 
     private static Timestamp asTimestamp(Object val, JDBCType columnType) throws SQLException {
         if (columnType == JDBCType.TIMESTAMP) {
             return new Timestamp(((Number) val).longValue());
         }
-        throw new SQLException("Conversion from type [" + columnType + "] to [Timestamp] not supported");
+        throw new SQLException(
+                format(Locale.ROOT, "Unable to convert value [%.128s] of type [%s] to a Timestamp", val, columnType.getName()));
     }
 
     private static byte[] asByteArray(Object val, JDBCType columnType) {
diff --git a/x-pack/plugin/sql/jdbc/src/test/java/org/elasticsearch/xpack/sql/jdbc/jdbc/JdbcPreparedStatementTests.java b/x-pack/plugin/sql/jdbc/src/test/java/org/elasticsearch/xpack/sql/jdbc/jdbc/JdbcPreparedStatementTests.java
index 9da06f6537c..35a3ec57487 100644
--- a/x-pack/plugin/sql/jdbc/src/test/java/org/elasticsearch/xpack/sql/jdbc/jdbc/JdbcPreparedStatementTests.java
+++ b/x-pack/plugin/sql/jdbc/src/test/java/org/elasticsearch/xpack/sql/jdbc/jdbc/JdbcPreparedStatementTests.java
@@ -25,6 +25,7 @@ import java.util.Date;
 import java.util.Locale;
 import java.util.Map;
 
+import static java.lang.String.format;
 import static java.sql.JDBCType.BIGINT;
 import static java.sql.JDBCType.BOOLEAN;
 import static java.sql.JDBCType.DOUBLE;
@@ -68,7 +69,7 @@ public class JdbcPreparedStatementTests extends ESTestCase {
         JdbcPreparedStatement jps = createJdbcPreparedStatement();
 
         SQLException sqle = expectThrows(SQLException.class, () -> jps.setObject(1, true, Types.TIMESTAMP));
-        assertEquals("Conversion from type [BOOLEAN] to [Timestamp] not supported", sqle.getMessage());
+        assertEquals("Unable to convert value [true] of type [BOOLEAN] to a Timestamp", sqle.getMessage());
     }
 
     public void testSettingStringValues() throws SQLException {
@@ -92,7 +93,7 @@ public class JdbcPreparedStatementTests extends ESTestCase {
         JdbcPreparedStatement jps = createJdbcPreparedStatement();
 
         SQLException sqle = expectThrows(SQLException.class, () -> jps.setObject(1, "foo bar", Types.INTEGER));
-        assertEquals("Conversion from type [VARCHAR] to [Integer] not supported", sqle.getMessage());
+        assertEquals("Unable to convert value [foo bar] of type [VARCHAR] to an Integer", sqle.getMessage());
     }
 
     public void testSettingByteTypeValues() throws SQLException {
@@ -128,7 +129,7 @@ public class JdbcPreparedStatementTests extends ESTestCase {
         JdbcPreparedStatement jps = createJdbcPreparedStatement();
 
         SQLException sqle = expectThrows(SQLException.class, () -> jps.setObject(1, (byte) 6, Types.TIMESTAMP));
-        assertEquals("Conversion from type [TINYINT] to [Timestamp] not supported", sqle.getMessage());
+        assertEquals("Unable to convert value [6] of type [TINYINT] to a Timestamp", sqle.getMessage());
     }
 
     public void testSettingShortTypeValues() throws SQLException {
@@ -161,7 +162,7 @@ public class JdbcPreparedStatementTests extends ESTestCase {
         JdbcPreparedStatement jps = createJdbcPreparedStatement();
 
         SQLException sqle = expectThrows(SQLException.class, () -> jps.setObject(1, (short) 6, Types.TIMESTAMP));
-        assertEquals("Conversion from type [SMALLINT] to [Timestamp] not supported", sqle.getMessage());
+        assertEquals("Unable to convert value [6] of type [SMALLINT] to a Timestamp", sqle.getMessage());
 
         sqle = expectThrows(SQLException.class, () -> jps.setObject(1, 256, Types.TINYINT));
         assertEquals("Numeric " + 256 + " out of range", sqle.getMessage());
@@ -195,7 +196,7 @@ public class JdbcPreparedStatementTests extends ESTestCase {
         int someInt = randomInt();
 
         SQLException sqle = expectThrows(SQLException.class, () -> jps.setObject(1, someInt, Types.TIMESTAMP));
-        assertEquals("Conversion from type [INTEGER] to [Timestamp] not supported", sqle.getMessage());
+        assertEquals(format(Locale.ROOT, "Unable to convert value [%.128s] of type [INTEGER] to a Timestamp", someInt), sqle.getMessage());
 
         Integer randomIntNotShort = randomIntBetween(32768, Integer.MAX_VALUE);
         sqle = expectThrows(SQLException.class, () -> jps.setObject(1, randomIntNotShort, Types.SMALLINT));
@@ -236,7 +237,7 @@ public class JdbcPreparedStatementTests extends ESTestCase {
         long someLong = randomLong();
 
         SQLException sqle = expectThrows(SQLException.class, () -> jps.setObject(1, someLong, Types.TIMESTAMP));
-        assertEquals("Conversion from type [BIGINT] to [Timestamp] not supported", sqle.getMessage());
+        assertEquals(format(Locale.ROOT, "Unable to convert value [%.128s] of type [BIGINT] to a Timestamp", someLong), sqle.getMessage());
 
         Long randomLongNotShort = randomLongBetween(Integer.MAX_VALUE + 1, Long.MAX_VALUE);
         sqle = expectThrows(SQLException.class, () -> jps.setObject(1, randomLongNotShort, Types.INTEGER));
@@ -277,7 +278,7 @@ public class JdbcPreparedStatementTests extends ESTestCase {
         float someFloat = randomFloat();
 
         SQLException sqle = expectThrows(SQLException.class, () -> jps.setObject(1, someFloat, Types.TIMESTAMP));
-        assertEquals("Conversion from type [REAL] to [Timestamp] not supported", sqle.getMessage());
+        assertEquals(format(Locale.ROOT, "Unable to convert value [%.128s] of type [REAL] to a Timestamp", someFloat), sqle.getMessage());
 
         Float floatNotInt =  5_155_000_000f;
         sqle = expectThrows(SQLException.class, () -> jps.setObject(1, floatNotInt, Types.INTEGER));
@@ -316,7 +317,8 @@ public class JdbcPreparedStatementTests extends ESTestCase {
         double someDouble = randomDouble();
 
         SQLException sqle = expectThrows(SQLException.class, () -> jps.setObject(1, someDouble, Types.TIMESTAMP));
-        assertEquals("Conversion from type [DOUBLE] to [Timestamp] not supported", sqle.getMessage());
+        assertEquals(
+                format(Locale.ROOT, "Unable to convert value [%.128s] of type [DOUBLE] to a Timestamp", someDouble), sqle.getMessage());
 
         Double doubleNotInt = 5_155_000_000d;
         sqle = expectThrows(SQLException.class, () -> jps.setObject(1, doubleNotInt, Types.INTEGER));
@@ -361,7 +363,7 @@ public class JdbcPreparedStatementTests extends ESTestCase {
     public void testSettingTimestampValues() throws SQLException {
         JdbcPreparedStatement jps = createJdbcPreparedStatement();
 
-        Timestamp someTimestamp = new Timestamp(randomMillisSinceEpoch());
+        Timestamp someTimestamp = new Timestamp(randomLong());
         jps.setTimestamp(1, someTimestamp);
         assertEquals(someTimestamp.getTime(), ((Date)value(jps)).getTime());
         assertEquals(TIMESTAMP, jdbcType(jps));
@@ -372,7 +374,7 @@ public class JdbcPreparedStatementTests extends ESTestCase {
         assertEquals(1456708675000L, convertFromUTCtoCalendar(((Date)value(jps)), nonDefaultCal));
         assertEquals(TIMESTAMP, jdbcType(jps));
 
-        long beforeEpochTime = -randomMillisSinceEpoch();
+        long beforeEpochTime = randomLongBetween(Long.MIN_VALUE, 0);
         jps.setTimestamp(1, new Timestamp(beforeEpochTime), nonDefaultCal);
         assertEquals(beforeEpochTime, convertFromUTCtoCalendar(((Date)value(jps)), nonDefaultCal));
         assertTrue(value(jps) instanceof java.util.Date);
@@ -384,7 +386,7 @@ public class JdbcPreparedStatementTests extends ESTestCase {
 
     public void testThrownExceptionsWhenSettingTimestampValues() throws SQLException {
         JdbcPreparedStatement jps = createJdbcPreparedStatement();
-        Timestamp someTimestamp = new Timestamp(randomMillisSinceEpoch());
+        Timestamp someTimestamp = new Timestamp(randomLong());
 
         SQLException sqle = expectThrows(SQLFeatureNotSupportedException.class, () -> jps.setObject(1, someTimestamp, Types.INTEGER));
         assertEquals("Conversion from type java.sql.Timestamp to INTEGER not supported", sqle.getMessage());
@@ -416,12 +418,12 @@ public class JdbcPreparedStatementTests extends ESTestCase {
     public void testSettingSqlDateValues() throws SQLException {
         JdbcPreparedStatement jps = createJdbcPreparedStatement();
 
-        java.sql.Date someSqlDate = new java.sql.Date(randomMillisSinceEpoch());
+        java.sql.Date someSqlDate = new java.sql.Date(randomLong());
         jps.setDate(1, someSqlDate);
         assertEquals(someSqlDate.getTime(), ((Date)value(jps)).getTime());
         assertEquals(TIMESTAMP, jdbcType(jps));
 
-        someSqlDate = new java.sql.Date(randomMillisSinceEpoch());
+        someSqlDate = new java.sql.Date(randomLong());
         Calendar nonDefaultCal = randomCalendar();
         jps.setDate(1, someSqlDate, nonDefaultCal);
         assertEquals(someSqlDate.getTime(), convertFromUTCtoCalendar(((Date)value(jps)), nonDefaultCal));
@@ -435,17 +437,17 @@ public class JdbcPreparedStatementTests extends ESTestCase {
 
     public void testThrownExceptionsWhenSettingSqlDateValues() throws SQLException {
         JdbcPreparedStatement jps = createJdbcPreparedStatement();
-        java.sql.Date someSqlDate = new java.sql.Date(randomMillisSinceEpoch());
+        java.sql.Date someSqlDate = new java.sql.Date(randomLong());
 
         SQLException sqle = expectThrows(SQLFeatureNotSupportedException.class,
-                () -> jps.setObject(1, new java.sql.Date(randomMillisSinceEpoch()), Types.DOUBLE));
+                () -> jps.setObject(1, new java.sql.Date(randomLong()), Types.DOUBLE));
         assertEquals("Conversion from type " + someSqlDate.getClass().getName() + " to DOUBLE not supported", sqle.getMessage());
     }
 
     public void testSettingCalendarValues() throws SQLException {
         JdbcPreparedStatement jps = createJdbcPreparedStatement();
         Calendar someCalendar = randomCalendar();
-        someCalendar.setTimeInMillis(randomMillisSinceEpoch());
+        someCalendar.setTimeInMillis(randomLong());
 
         jps.setObject(1, someCalendar);
         assertEquals(someCalendar.getTime(), (Date) value(jps));
@@ -472,7 +474,7 @@ public class JdbcPreparedStatementTests extends ESTestCase {
 
     public void testSettingDateValues() throws SQLException {
         JdbcPreparedStatement jps = createJdbcPreparedStatement();
-        Date someDate = new Date(randomMillisSinceEpoch());
+        Date someDate = new Date(randomLong());
 
         jps.setObject(1, someDate);
         assertEquals(someDate, (Date) value(jps));
@@ -486,7 +488,7 @@ public class JdbcPreparedStatementTests extends ESTestCase {
 
     public void testThrownExceptionsWhenSettingDateValues() throws SQLException {
         JdbcPreparedStatement jps = createJdbcPreparedStatement();
-        Date someDate = new Date(randomMillisSinceEpoch());
+        Date someDate = new Date(randomLong());
 
         SQLException sqle = expectThrows(SQLFeatureNotSupportedException.class, () -> jps.setObject(1, someDate, Types.BIGINT));
         assertEquals("Conversion from type " + someDate.getClass().getName() + " to BIGINT not supported", sqle.getMessage());
@@ -549,10 +551,6 @@ public class JdbcPreparedStatementTests extends ESTestCase {
         assertEquals("Conversion from type byte[] to DOUBLE not supported", sqle.getMessage());
     }
 
-    private long randomMillisSinceEpoch() {
-        return randomLongBetween(0, System.currentTimeMillis());
-    }
-
     private JdbcPreparedStatement createJdbcPreparedStatement() throws SQLException {
         return new JdbcPreparedStatement(null, JdbcConfiguration.create("jdbc:es://l:1", null, 0), "?");
     }
diff --git a/x-pack/qa/sql/no-security/src/test/java/org/elasticsearch/xpack/qa/sql/nosecurity/JdbcResultSetIT.java b/x-pack/qa/sql/no-security/src/test/java/org/elasticsearch/xpack/qa/sql/nosecurity/JdbcResultSetIT.java
new file mode 100644
index 00000000000..30756a11f62
--- /dev/null
+++ b/x-pack/qa/sql/no-security/src/test/java/org/elasticsearch/xpack/qa/sql/nosecurity/JdbcResultSetIT.java
@@ -0,0 +1,16 @@
+/*
+ * 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.qa.sql.nosecurity;
+
+import org.elasticsearch.xpack.qa.sql.jdbc.ResultSetTestCase;
+
+/*
+ * Integration testing class for "no security" (cluster running without the Security plugin,
+ * or the Security is disbled) scenario. Runs all tests in the base class.
+ */
+public class JdbcResultSetIT extends ResultSetTestCase {
+}
diff --git a/x-pack/qa/sql/src/main/java/org/elasticsearch/xpack/qa/sql/jdbc/ResultSetTestCase.java b/x-pack/qa/sql/src/main/java/org/elasticsearch/xpack/qa/sql/jdbc/ResultSetTestCase.java
index 861a6dccaba..447fc4f17e1 100644
--- a/x-pack/qa/sql/src/main/java/org/elasticsearch/xpack/qa/sql/jdbc/ResultSetTestCase.java
+++ b/x-pack/qa/sql/src/main/java/org/elasticsearch/xpack/qa/sql/jdbc/ResultSetTestCase.java
@@ -5,55 +5,1067 @@
  */
 package org.elasticsearch.xpack.qa.sql.jdbc;
 
+import org.elasticsearch.client.Request;
+import org.elasticsearch.common.CheckedBiFunction;
+import org.elasticsearch.common.CheckedConsumer;
+import org.elasticsearch.common.CheckedFunction;
+import org.elasticsearch.common.Strings;
+import org.elasticsearch.common.collect.Tuple;
+import org.elasticsearch.common.xcontent.XContentBuilder;
+import org.elasticsearch.common.xcontent.json.JsonXContent;
+import org.elasticsearch.xpack.sql.jdbc.jdbc.JdbcConfiguration;
+import org.elasticsearch.xpack.sql.jdbc.jdbcx.JdbcDataSource;
+import org.elasticsearch.xpack.sql.type.DataType;
+
 import java.io.IOException;
+import java.io.InputStream;
+import java.io.Reader;
+import java.sql.Blob;
+import java.sql.Clob;
 import java.sql.Connection;
+import java.sql.DriverManager;
+import java.sql.JDBCType;
+import java.sql.NClob;
 import java.sql.PreparedStatement;
 import java.sql.ResultSet;
 import java.sql.ResultSetMetaData;
 import java.sql.SQLException;
+import java.sql.SQLFeatureNotSupportedException;
 import java.sql.Timestamp;
+import java.sql.Types;
+import java.util.Arrays;
+import java.util.Calendar;
 import java.util.Date;
+import java.util.GregorianCalendar;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Locale;
+import java.util.Map;
+import java.util.Map.Entry;
+import java.util.Properties;
+import java.util.Set;
+import java.util.TimeZone;
+import java.util.function.Supplier;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+
+import static java.lang.String.format;
+import static java.util.Calendar.DAY_OF_MONTH;
+import static java.util.Calendar.ERA;
+import static java.util.Calendar.HOUR_OF_DAY;
+import static java.util.Calendar.MILLISECOND;
+import static java.util.Calendar.MINUTE;
+import static java.util.Calendar.MONTH;
+import static java.util.Calendar.SECOND;
+import static java.util.Calendar.YEAR;
 
 public class ResultSetTestCase extends JdbcIntegrationTestCase {
-    public void testGettingTimestamp() throws Exception {
-        long randomMillis = randomLongBetween(0, System.currentTimeMillis());
+    
+    static final Set<String> fieldsNames = Stream.of("test_byte", "test_integer", "test_long", "test_short", "test_double",
+            "test_float", "test_keyword")
+            .collect(Collectors.toCollection(HashSet::new));
+    static final Map<Tuple<String,Object>,JDBCType> dateTimeTestingFields = new HashMap<Tuple<String,Object>,JDBCType>();
+    static final String SELECT_ALL_FIELDS = "SELECT test_boolean, test_byte, test_integer,"
+            + "test_long, test_short, test_double, test_float, test_keyword, test_date FROM test";
+    static final String SELECT_WILDCARD = "SELECT * FROM test";
+    static {
+        dateTimeTestingFields.put(new Tuple<String,Object>("test_boolean", true), DataType.BOOLEAN.jdbcType);
+        dateTimeTestingFields.put(new Tuple<String,Object>("test_byte", 1), DataType.BYTE.jdbcType);
+        dateTimeTestingFields.put(new Tuple<String,Object>("test_integer", 1), DataType.INTEGER.jdbcType);
+        dateTimeTestingFields.put(new Tuple<String,Object>("test_long", 1L), DataType.LONG.jdbcType);
+        dateTimeTestingFields.put(new Tuple<String,Object>("test_short", 1), DataType.SHORT.jdbcType);
+        dateTimeTestingFields.put(new Tuple<String,Object>("test_double", 1d), DataType.DOUBLE.jdbcType);
+        dateTimeTestingFields.put(new Tuple<String,Object>("test_float", 1f), DataType.FLOAT.jdbcType);
+        dateTimeTestingFields.put(new Tuple<String,Object>("test_keyword", "true"), DataType.KEYWORD.jdbcType);
+    }
+    
+    // Byte values testing
+    public void testGettingValidByteWithoutCasting() throws Exception {
+        byte random1 = randomByte();
+        byte random2 = randomValueOtherThan(random1, () -> randomByte());
+        byte random3 = randomValueOtherThanMany(Arrays.asList(random1, random2)::contains, () -> randomByte());
+        
+        createTestDataForByteValueTests(random1, random2, random3);
+        
+        doWithQuery("SELECT test_byte, test_null_byte, test_keyword FROM test", (results) -> {
+            ResultSetMetaData resultSetMetaData = results.getMetaData();
+
+            results.next();
+            assertEquals(3, resultSetMetaData.getColumnCount());
+            assertEquals(Types.TINYINT, resultSetMetaData.getColumnType(1));
+            assertEquals(Types.TINYINT, resultSetMetaData.getColumnType(2));
+            assertEquals(random1, results.getByte(1));
+            assertEquals(random1, results.getByte("test_byte"));
+            assertEquals(random1, (byte) results.getObject("test_byte", Byte.class));
+            assertTrue(results.getObject(1) instanceof Byte);
+            
+            assertEquals(0, results.getByte(2));
+            assertTrue(results.wasNull());
+            assertEquals(null, results.getObject("test_null_byte"));
+            assertTrue(results.wasNull());
+            
+            assertTrue(results.next());
+            assertEquals(random2, results.getByte(1));
+            assertEquals(random2, results.getByte("test_byte"));
+            assertTrue(results.getObject(1) instanceof Byte);
+            assertEquals(random3, results.getByte("test_keyword"));
+            
+            assertFalse(results.next());
+        });
+    }
+    
+    public void testGettingValidByteWithCasting() throws Exception {
+        Map<String,Number> map = createTestDataForNumericValueTypes(() -> randomByte());
+        
+        doWithQuery(SELECT_WILDCARD, (results) -> {
+            results.next();
+            for(Entry<String, Number> e : map.entrySet()) {
+                byte actual = results.getObject(e.getKey(), Byte.class);
+                if (e.getValue() instanceof Double) {
+                    assertEquals("For field " + e.getKey(), Math.round(e.getValue().doubleValue()), results.getByte(e.getKey()));
+                    assertEquals("For field " + e.getKey(), Math.round(e.getValue().doubleValue()), actual);
+                } else if (e.getValue() instanceof Float) {
+                    assertEquals("For field " + e.getKey(), Math.round(e.getValue().floatValue()), results.getByte(e.getKey()));
+                    assertEquals("For field " + e.getKey(), Math.round(e.getValue().floatValue()), actual);
+                } else {
+                    assertEquals("For field " + e.getKey(), e.getValue().byteValue(), results.getByte(e.getKey()));
+                    assertEquals("For field " + e.getKey(), e.getValue().byteValue(), actual);
+                }
+            }
+        });
+    }
+
+    public void testGettingInvalidByte() throws Exception {
+        createIndex("test");
+        updateMappingForNumericValuesTests("test");
+        updateMapping("test", builder -> {
+            builder.startObject("test_keyword").field("type", "keyword").endObject();
+            builder.startObject("test_date").field("type", "date").endObject();
+        });
+        
+        int intNotByte = randomIntBetween(Byte.MAX_VALUE + 1, Integer.MAX_VALUE);
+        long longNotByte = randomLongBetween(Byte.MAX_VALUE + 1, Long.MAX_VALUE);
+        short shortNotByte = (short) randomIntBetween(Byte.MAX_VALUE + 1, Short.MAX_VALUE);
+        double doubleNotByte = randomDoubleBetween(Byte.MAX_VALUE + 1, Double.MAX_VALUE, true);
+        float floatNotByte = randomFloatBetween(Byte.MAX_VALUE + 1, Float.MAX_VALUE);
+        String randomString = randomUnicodeOfCodepointLengthBetween(128, 256);
+        long randomDate = randomLong();
+        
+        String doubleErrorMessage = (doubleNotByte > Long.MAX_VALUE || doubleNotByte < Long.MIN_VALUE) ?
+                Double.toString(doubleNotByte) : Long.toString(Math.round(doubleNotByte));
+                
+        index("test", "1", builder -> {
+            builder.field("test_integer", intNotByte);
+            builder.field("test_long", longNotByte);
+            builder.field("test_short", shortNotByte);
+            builder.field("test_double", doubleNotByte);
+            builder.field("test_float", floatNotByte);
+            builder.field("test_keyword", randomString);
+            builder.field("test_date", randomDate);
+        });
+        
+        doWithQuery(SELECT_WILDCARD, (results) -> {
+            results.next();
+            
+            SQLException sqle = expectThrows(SQLException.class, () -> results.getByte("test_integer"));
+            assertEquals(format(Locale.ROOT, "Numeric %s out of range", intNotByte), sqle.getMessage());
+            sqle = expectThrows(SQLException.class, () -> results.getObject("test_integer", Byte.class));
+            assertEquals(format(Locale.ROOT, "Numeric %s out of range", intNotByte), sqle.getMessage());
+            
+            sqle = expectThrows(SQLException.class, () -> results.getByte("test_short"));
+            assertEquals(format(Locale.ROOT, "Numeric %s out of range", shortNotByte), sqle.getMessage());
+            sqle = expectThrows(SQLException.class, () -> results.getObject("test_short", Byte.class));
+            assertEquals(format(Locale.ROOT, "Numeric %s out of range", shortNotByte), sqle.getMessage());
+            
+            sqle = expectThrows(SQLException.class, () -> results.getByte("test_long"));
+            assertEquals(format(Locale.ROOT, "Numeric %s out of range", Long.toString(longNotByte)), sqle.getMessage());
+            sqle = expectThrows(SQLException.class, () -> results.getObject("test_long", Byte.class));
+            assertEquals(format(Locale.ROOT, "Numeric %s out of range", Long.toString(longNotByte)), sqle.getMessage());
+            
+            sqle = expectThrows(SQLException.class, () -> results.getByte("test_double"));
+            assertEquals(format(Locale.ROOT, "Numeric %s out of range", doubleErrorMessage), sqle.getMessage());
+            sqle = expectThrows(SQLException.class, () -> results.getObject("test_double", Byte.class));
+            assertEquals(format(Locale.ROOT, "Numeric %s out of range", doubleErrorMessage), sqle.getMessage());
+            
+            sqle = expectThrows(SQLException.class, () -> results.getByte("test_float"));
+            assertEquals(format(Locale.ROOT, "Numeric %s out of range", Double.toString(floatNotByte)), sqle.getMessage());
+            sqle = expectThrows(SQLException.class, () -> results.getObject("test_float", Byte.class));
+            assertEquals(format(Locale.ROOT, "Numeric %s out of range", Double.toString(floatNotByte)), sqle.getMessage());
+            
+            sqle = expectThrows(SQLException.class, () -> results.getByte("test_keyword"));
+            assertEquals(format(Locale.ROOT, "Unable to convert value [%.128s] of type [VARCHAR] to a Byte", randomString), 
+                    sqle.getMessage());
+            sqle = expectThrows(SQLException.class, () -> results.getObject("test_keyword", Byte.class));
+            assertEquals(format(Locale.ROOT, "Unable to convert value [%.128s] of type [VARCHAR] to a Byte", randomString), 
+                    sqle.getMessage());
+            
+            sqle = expectThrows(SQLException.class, () -> results.getByte("test_date"));
+            assertEquals(format(Locale.ROOT, "Unable to convert value [%.128s] of type [TIMESTAMP] to a Byte", randomDate), 
+                    sqle.getMessage());
+            sqle = expectThrows(SQLException.class, () -> results.getObject("test_date", Byte.class));
+            assertEquals(format(Locale.ROOT, "Unable to convert value [%.128s] of type [TIMESTAMP] to a Byte", randomDate), 
+                    sqle.getMessage());
+        });
+    }
+    
+    // Short values testing
+    public void testGettingValidShortWithoutCasting() throws Exception {
+        short random1 = randomShort();
+        short random2 = randomValueOtherThan(random1, () -> randomShort());
+        short random3 = randomValueOtherThanMany(Arrays.asList(random1, random2)::contains, () -> randomShort());
+        
+        createTestDataForShortValueTests(random1, random2, random3);
+        
+        doWithQuery("SELECT test_short, test_null_short, test_keyword FROM test", (results) -> {
+            ResultSetMetaData resultSetMetaData = results.getMetaData();
+
+            results.next();
+            assertEquals(3, resultSetMetaData.getColumnCount());
+            assertEquals(Types.SMALLINT, resultSetMetaData.getColumnType(1));
+            assertEquals(Types.SMALLINT, resultSetMetaData.getColumnType(2));
+            assertEquals(random1, results.getShort(1));
+            assertEquals(random1, results.getShort("test_short"));
+            assertEquals(random1, results.getObject("test_short"));
+            assertTrue(results.getObject(1) instanceof Short);
+            
+            assertEquals(0, results.getShort(2));
+            assertTrue(results.wasNull());
+            assertEquals(null, results.getObject("test_null_short"));
+            assertTrue(results.wasNull());
+            
+            assertTrue(results.next());
+            assertEquals(random2, results.getShort(1));
+            assertEquals(random2, results.getShort("test_short"));
+            assertTrue(results.getObject(1) instanceof Short);
+            assertEquals(random3, results.getShort("test_keyword"));
+            
+            assertFalse(results.next());
+        });
+    }
+    
+    public void testGettingValidShortWithCasting() throws Exception {
+        Map<String,Number> map = createTestDataForNumericValueTypes(() -> randomShort());
+        
+        doWithQuery(SELECT_WILDCARD, (results) -> {
+            results.next();
+            for(Entry<String, Number> e : map.entrySet()) {
+                short actual = (short) results.getObject(e.getKey(), Short.class);
+                if (e.getValue() instanceof Double) {
+                    assertEquals("For field " + e.getKey(), Math.round(e.getValue().doubleValue()), results.getShort(e.getKey()));
+                    assertEquals("For field " + e.getKey(), Math.round(e.getValue().doubleValue()), actual);
+                } else if (e.getValue() instanceof Float) {
+                    assertEquals("For field " + e.getKey(), Math.round(e.getValue().floatValue()), results.getShort(e.getKey()));
+                    assertEquals("For field " + e.getKey(), Math.round(e.getValue().floatValue()), actual);
+                } else {
+                    assertEquals("For field " + e.getKey(),
+                            e.getValue().shortValue(), results.getShort(e.getKey()));
+                    assertEquals("For field " + e.getKey(), e.getValue().shortValue(), actual);
+                }
+            }
+        });
+    }
+
+    public void testGettingInvalidShort() throws Exception {
+        createIndex("test");
+        updateMappingForNumericValuesTests("test");
+        updateMapping("test", builder -> {
+            builder.startObject("test_keyword").field("type", "keyword").endObject();
+            builder.startObject("test_date").field("type", "date").endObject();
+        });
+        
+        int intNotShort = randomIntBetween(Short.MAX_VALUE + 1, Integer.MAX_VALUE);
+        long longNotShort = randomLongBetween(Short.MAX_VALUE + 1, Long.MAX_VALUE);
+        double doubleNotShort = randomDoubleBetween(Short.MAX_VALUE + 1, Double.MAX_VALUE, true);
+        float floatNotShort = randomFloatBetween(Short.MAX_VALUE + 1, Float.MAX_VALUE);
+        String randomString = randomUnicodeOfCodepointLengthBetween(128, 256);
+        long randomDate = randomLong();
+        
+        String doubleErrorMessage = (doubleNotShort > Long.MAX_VALUE || doubleNotShort < Long.MIN_VALUE) ?
+                Double.toString(doubleNotShort) : Long.toString(Math.round(doubleNotShort));
+
+        index("test", "1", builder -> {
+            builder.field("test_integer", intNotShort);
+            builder.field("test_long", longNotShort);
+            builder.field("test_double", doubleNotShort);
+            builder.field("test_float", floatNotShort);
+            builder.field("test_keyword", randomString);
+            builder.field("test_date", randomDate);
+        });
+        
+        doWithQuery(SELECT_WILDCARD, (results) -> {
+            results.next();
+            
+            SQLException sqle = expectThrows(SQLException.class, () -> results.getShort("test_integer"));
+            assertEquals(format(Locale.ROOT, "Numeric %s out of range", intNotShort), sqle.getMessage());
+            sqle = expectThrows(SQLException.class, () -> results.getObject("test_integer", Short.class));
+            assertEquals(format(Locale.ROOT, "Numeric %s out of range", intNotShort), sqle.getMessage());
+            
+            sqle = expectThrows(SQLException.class, () -> results.getShort("test_long"));
+            assertEquals(format(Locale.ROOT, "Numeric %s out of range", Long.toString(longNotShort)), sqle.getMessage());
+            sqle = expectThrows(SQLException.class, () -> results.getObject("test_long", Short.class));
+            assertEquals(format(Locale.ROOT, "Numeric %s out of range", Long.toString(longNotShort)), sqle.getMessage());
+            
+            sqle = expectThrows(SQLException.class, () -> results.getShort("test_double"));
+            assertEquals(format(Locale.ROOT, "Numeric %s out of range", doubleErrorMessage), sqle.getMessage());
+            sqle = expectThrows(SQLException.class, () -> results.getObject("test_double", Short.class));
+            assertEquals(format(Locale.ROOT, "Numeric %s out of range", doubleErrorMessage), sqle.getMessage());
+            
+            sqle = expectThrows(SQLException.class, () -> results.getShort("test_float"));
+            assertEquals(format(Locale.ROOT, "Numeric %s out of range", Double.toString(floatNotShort)), sqle.getMessage());
+            sqle = expectThrows(SQLException.class, () -> results.getObject("test_float", Short.class));
+            assertEquals(format(Locale.ROOT, "Numeric %s out of range", Double.toString(floatNotShort)), sqle.getMessage());
+            
+            sqle = expectThrows(SQLException.class, () -> results.getShort("test_keyword"));
+            assertEquals(format(Locale.ROOT, "Unable to convert value [%.128s] of type [VARCHAR] to a Short", randomString),
+                    sqle.getMessage());
+            sqle = expectThrows(SQLException.class, () -> results.getObject("test_keyword", Short.class));
+            assertEquals(format(Locale.ROOT, "Unable to convert value [%.128s] of type [VARCHAR] to a Short", randomString),
+                    sqle.getMessage());
+            
+            sqle = expectThrows(SQLException.class, () -> results.getShort("test_date"));
+            assertEquals(format(Locale.ROOT, "Unable to convert value [%.128s] of type [TIMESTAMP] to a Short", randomDate),
+                    sqle.getMessage());
+            sqle = expectThrows(SQLException.class, () -> results.getObject("test_date", Short.class));
+            assertEquals(format(Locale.ROOT, "Unable to convert value [%.128s] of type [TIMESTAMP] to a Short", randomDate),
+                    sqle.getMessage());
+        });
+    }
+    
+    // Integer values testing
+    public void testGettingValidIntegerWithoutCasting() throws Exception {
+        int random1 = randomInt();
+        int random2 = randomValueOtherThan(random1, () -> randomInt());
+        int random3 = randomValueOtherThanMany(Arrays.asList(random1, random2)::contains, () -> randomInt());
+        
+        createTestDataForIntegerValueTests(random1, random2, random3);
+        
+        doWithQuery("SELECT test_integer,test_null_integer,test_keyword FROM test", (results) -> {
+            ResultSetMetaData resultSetMetaData = results.getMetaData();
+
+            results.next();
+            assertEquals(3, resultSetMetaData.getColumnCount());
+            assertEquals(Types.INTEGER, resultSetMetaData.getColumnType(1));
+            assertEquals(Types.INTEGER, resultSetMetaData.getColumnType(2));
+            assertEquals(random1, results.getInt(1));
+            assertEquals(random1, results.getInt("test_integer"));
+            assertEquals(random1, (int) results.getObject("test_integer", Integer.class));
+            assertTrue(results.getObject(1) instanceof Integer);
+            
+            assertEquals(0, results.getInt(2));
+            assertTrue(results.wasNull());
+            assertEquals(null, results.getObject("test_null_integer"));
+            assertTrue(results.wasNull());
+            
+            assertTrue(results.next());
+            assertEquals(random2, results.getInt(1));
+            assertEquals(random2, results.getInt("test_integer"));
+            assertTrue(results.getObject(1) instanceof Integer);
+            assertEquals(random3, results.getInt("test_keyword"));
+            
+            assertFalse(results.next());
+        });
+    }
+    
+    public void testGettingValidIntegerWithCasting() throws Exception {
+        Map<String,Number> map = createTestDataForNumericValueTypes(() -> randomInt());
+        
+        doWithQuery(SELECT_WILDCARD, (results) -> {
+            results.next();
+            for(Entry<String, Number> e : map.entrySet()) {
+                int actual = results.getObject(e.getKey(), Integer.class);
+                if (e.getValue() instanceof Double) {
+                    assertEquals("For field " + e.getKey(), Math.round(e.getValue().doubleValue()), results.getInt(e.getKey()));
+                    assertEquals("For field " + e.getKey(), Math.round(e.getValue().doubleValue()), actual);
+                } else if (e.getValue() instanceof Float) {
+                    assertEquals("For field " + e.getKey(), Math.round(e.getValue().floatValue()), results.getInt(e.getKey()));
+                    assertEquals("For field " + e.getKey(), Math.round(e.getValue().floatValue()), actual);
+                } else {
+                    assertEquals("For field " + e.getKey(), e.getValue().intValue(), results.getInt(e.getKey()));
+                    assertEquals("For field " + e.getKey(), e.getValue().intValue(), actual);
+                }
+            }
+        });
+    }
+
+    public void testGettingInvalidInteger() throws Exception {
+        createIndex("test");
+        updateMappingForNumericValuesTests("test");
+        updateMapping("test", builder -> {
+            builder.startObject("test_keyword").field("type", "keyword").endObject();
+            builder.startObject("test_date").field("type", "date").endObject();
+        });
+        
+        long longNotInt = randomLongBetween(getMaxIntPlusOne(), Long.MAX_VALUE);
+        double doubleNotInt = randomDoubleBetween(getMaxIntPlusOne().doubleValue(), Double.MAX_VALUE, true);
+        float floatNotInt = randomFloatBetween(getMaxIntPlusOne().floatValue(), Float.MAX_VALUE);
+        String randomString = randomUnicodeOfCodepointLengthBetween(128, 256);
+        long randomDate = randomLong();
+        
+        String doubleErrorMessage = (doubleNotInt > Long.MAX_VALUE || doubleNotInt < Long.MIN_VALUE) ?
+                Double.toString(doubleNotInt) : Long.toString(Math.round(doubleNotInt));
+
+        index("test", "1", builder -> {
+            builder.field("test_long", longNotInt);
+            builder.field("test_double", doubleNotInt);
+            builder.field("test_float", floatNotInt);
+            builder.field("test_keyword", randomString);
+            builder.field("test_date", randomDate);
+        });
+        
+        doWithQuery(SELECT_WILDCARD, (results) -> {
+            results.next();
+            
+            SQLException sqle = expectThrows(SQLException.class, () -> results.getInt("test_long"));
+            assertEquals(format(Locale.ROOT, "Numeric %s out of range", Long.toString(longNotInt)), sqle.getMessage());
+            sqle = expectThrows(SQLException.class, () -> results.getObject("test_long", Integer.class));
+            assertEquals(format(Locale.ROOT, "Numeric %s out of range", Long.toString(longNotInt)), sqle.getMessage());
+            
+            sqle = expectThrows(SQLException.class, () -> results.getInt("test_double"));
+            assertEquals(format(Locale.ROOT, "Numeric %s out of range", doubleErrorMessage), sqle.getMessage());
+            sqle = expectThrows(SQLException.class, () -> results.getObject("test_double", Integer.class));
+            assertEquals(format(Locale.ROOT, "Numeric %s out of range", doubleErrorMessage), sqle.getMessage());
+            
+            sqle = expectThrows(SQLException.class, () -> results.getInt("test_float"));
+            assertEquals(format(Locale.ROOT, "Numeric %s out of range", Double.toString(floatNotInt)), sqle.getMessage());
+            sqle = expectThrows(SQLException.class, () -> results.getObject("test_float", Integer.class));
+            assertEquals(format(Locale.ROOT, "Numeric %s out of range", Double.toString(floatNotInt)), sqle.getMessage());
+            
+            sqle = expectThrows(SQLException.class, () -> results.getInt("test_keyword"));
+            assertEquals(format(Locale.ROOT, "Unable to convert value [%.128s] of type [VARCHAR] to an Integer", randomString),
+                    sqle.getMessage());
+            sqle = expectThrows(SQLException.class, () -> results.getObject("test_keyword", Integer.class));
+            assertEquals(format(Locale.ROOT, "Unable to convert value [%.128s] of type [VARCHAR] to an Integer", randomString),
+                    sqle.getMessage());
+            
+            sqle = expectThrows(SQLException.class, () -> results.getInt("test_date"));
+            assertEquals(format(Locale.ROOT, "Unable to convert value [%.128s] of type [TIMESTAMP] to an Integer", randomDate),
+                    sqle.getMessage());
+            sqle = expectThrows(SQLException.class, () -> results.getObject("test_date", Integer.class));
+            assertEquals(format(Locale.ROOT, "Unable to convert value [%.128s] of type [TIMESTAMP] to an Integer", randomDate),
+                    sqle.getMessage());
+        });
+    }
+    
+    // Long values testing
+    public void testGettingValidLongWithoutCasting() throws Exception {
+        long random1 = randomLong();
+        long random2 = randomValueOtherThan(random1, () -> randomLong());
+        long random3 = randomValueOtherThanMany(Arrays.asList(random1, random2)::contains, () -> randomLong());
+        
+        createTestDataForLongValueTests(random1, random2, random3);
+        
+        doWithQuery("SELECT test_long, test_null_long, test_keyword FROM test", (results) -> {
+            ResultSetMetaData resultSetMetaData = results.getMetaData();
+
+            results.next();
+            assertEquals(3, resultSetMetaData.getColumnCount());
+            assertEquals(Types.BIGINT, resultSetMetaData.getColumnType(1));
+            assertEquals(Types.BIGINT, resultSetMetaData.getColumnType(2));
+            assertEquals(random1, results.getLong(1));
+            assertEquals(random1, results.getLong("test_long"));
+            assertEquals(random1, (long) results.getObject("test_long", Long.class));
+            assertTrue(results.getObject(1) instanceof Long);
+            
+            assertEquals(0, results.getLong(2));
+            assertTrue(results.wasNull());
+            assertEquals(null, results.getObject("test_null_long"));
+            assertTrue(results.wasNull());
+            
+            assertTrue(results.next());
+            assertEquals(random2, results.getLong(1));
+            assertEquals(random2, results.getLong("test_long"));
+            assertTrue(results.getObject(1) instanceof Long);
+            assertEquals(random3, results.getLong("test_keyword"));
+            
+            assertFalse(results.next());
+        });
+    }
+    
+    public void testGettingValidLongWithCasting() throws Exception {
+        Map<String,Number> map = createTestDataForNumericValueTypes(() -> randomLong());
+        
+        doWithQuery(SELECT_WILDCARD, (results) -> {
+            results.next();
+            for(Entry<String, Number> e : map.entrySet()) {
+                long actual = results.getObject(e.getKey(), Long.class);
+                if (e.getValue() instanceof Double || e.getValue() instanceof Float) {
+                    assertEquals("For field " + e.getKey(), Math.round(e.getValue().doubleValue()), results.getLong(e.getKey()));
+                    assertEquals("For field " + e.getKey(), Math.round(e.getValue().doubleValue()), actual);
+                } else {
+                    assertEquals("For field " + e.getKey(), e.getValue().longValue(), results.getLong(e.getKey()));
+                    assertEquals("For field " + e.getKey(), e.getValue().longValue(), actual);
+                }
+            }
+        });
+    }
+
+    public void testGettingInvalidLong() throws Exception {
+        createIndex("test");
+        updateMappingForNumericValuesTests("test");
+        updateMapping("test", builder -> {
+            builder.startObject("test_keyword").field("type", "keyword").endObject();
+            builder.startObject("test_date").field("type", "date").endObject();
+        });
+        
+        double doubleNotLong = randomDoubleBetween(getMaxLongPlusOne().doubleValue(), Double.MAX_VALUE, true);
+        float floatNotLong = randomFloatBetween(getMaxLongPlusOne().floatValue(), Float.MAX_VALUE);
+        String randomString = randomUnicodeOfCodepointLengthBetween(128, 256);
+        long randomDate = randomLong();
+
+        index("test", "1", builder -> {
+            builder.field("test_double", doubleNotLong);
+            builder.field("test_float", floatNotLong);
+            builder.field("test_keyword", randomString);
+            builder.field("test_date", randomDate);
+        });
+        
+        doWithQuery(SELECT_WILDCARD, (results) -> {
+            results.next();
+            
+            SQLException sqle = expectThrows(SQLException.class, () -> results.getLong("test_double"));
+            assertEquals(format(Locale.ROOT, "Numeric %s out of range", Double.toString(doubleNotLong)), sqle.getMessage());
+            sqle = expectThrows(SQLException.class, () -> results.getObject("test_double", Long.class));
+            assertEquals(format(Locale.ROOT, "Numeric %s out of range", Double.toString(doubleNotLong)), sqle.getMessage());
+            
+            sqle = expectThrows(SQLException.class, () -> results.getLong("test_float"));
+            assertEquals(format(Locale.ROOT, "Numeric %s out of range", Double.toString(floatNotLong)), sqle.getMessage());
+            sqle = expectThrows(SQLException.class, () -> results.getObject("test_float", Long.class));
+            assertEquals(format(Locale.ROOT, "Numeric %s out of range", Double.toString(floatNotLong)), sqle.getMessage());
+            
+            sqle = expectThrows(SQLException.class, () -> results.getLong("test_keyword"));
+            assertEquals(format(Locale.ROOT, "Unable to convert value [%.128s] of type [VARCHAR] to a Long", randomString),
+                    sqle.getMessage());
+            sqle = expectThrows(SQLException.class, () -> results.getObject("test_keyword", Long.class));
+            assertEquals(format(Locale.ROOT, "Unable to convert value [%.128s] of type [VARCHAR] to a Long", randomString),
+                    sqle.getMessage());
+            
+            sqle = expectThrows(SQLException.class, () -> results.getLong("test_date"));
+            assertEquals(format(Locale.ROOT, "Unable to convert value [%.128s] of type [TIMESTAMP] to a Long", randomDate),
+                    sqle.getMessage());
+            sqle = expectThrows(SQLException.class, () -> results.getObject("test_date", Long.class));
+            assertEquals(format(Locale.ROOT, "Unable to convert value [%.128s] of type [TIMESTAMP] to a Long", randomDate),
+                    sqle.getMessage());
+        });
+    }
+    
+    // Double values testing
+    public void testGettingValidDoubleWithoutCasting() throws Exception {
+        double random1 = randomDouble();
+        double random2 = randomValueOtherThan(random1, () -> randomDouble());
+        double random3 = randomValueOtherThanMany(Arrays.asList(random1, random2)::contains, () -> randomDouble());
+        
+        createTestDataForDoubleValueTests(random1, random2, random3);
+        
+        doWithQuery("SELECT test_double, test_null_double, test_keyword FROM test", (results) -> {
+            ResultSetMetaData resultSetMetaData = results.getMetaData();
+
+            results.next();
+            assertEquals(3, resultSetMetaData.getColumnCount());
+            assertEquals(Types.DOUBLE, resultSetMetaData.getColumnType(1));
+            assertEquals(Types.DOUBLE, resultSetMetaData.getColumnType(2));
+            assertEquals(random1, results.getDouble(1), 0.0d);
+            assertEquals(random1, results.getDouble("test_double"), 0.0d);
+            assertEquals(random1, results.getObject("test_double", Double.class), 0.0d);
+            assertTrue(results.getObject(1) instanceof Double);
+            
+            assertEquals(0, results.getDouble(2), 0.0d);
+            assertTrue(results.wasNull());
+            assertEquals(null, results.getObject("test_null_double"));
+            assertTrue(results.wasNull());
+            
+            assertTrue(results.next());
+            assertEquals(random2, results.getDouble(1), 0.0d);
+            assertEquals(random2, results.getDouble("test_double"), 0.0d);
+            assertTrue(results.getObject(1) instanceof Double);
+            assertEquals(random3, results.getDouble("test_keyword"), 0.0d);
+            
+            assertFalse(results.next());
+        });
+    }
+    
+    public void testGettingValidDoubleWithCasting() throws Exception {
+        Map<String,Number> map = createTestDataForNumericValueTypes(() -> randomDouble());
+        
+        doWithQuery(SELECT_WILDCARD, (results) -> {
+            results.next();
+            for(Entry<String, Number> e : map.entrySet()) {
+                assertEquals("For field " + e.getKey(), e.getValue().doubleValue(), results.getDouble(e.getKey()), 0.0d);
+                assertEquals("For field " + e.getKey(), 
+                        e.getValue().doubleValue(), results.getObject(e.getKey(), Double.class), 0.0d);
+            }
+        });
+    }
+
+    public void testGettingInvalidDouble() throws Exception {
+        createIndex("test");
+        updateMappingForNumericValuesTests("test");
+        updateMapping("test", builder -> {
+            builder.startObject("test_keyword").field("type", "keyword").endObject();
+            builder.startObject("test_date").field("type", "date").endObject();
+        });
+        
+        String randomString = randomUnicodeOfCodepointLengthBetween(128, 256);
+        long randomDate = randomLong();
+
+        index("test", "1", builder -> {
+            builder.field("test_keyword", randomString);
+            builder.field("test_date", randomDate);
+        });
+        
+        doWithQuery(SELECT_WILDCARD, (results) -> {
+            results.next();
+            
+            SQLException sqle = expectThrows(SQLException.class, () -> results.getDouble("test_keyword"));
+            assertEquals(format(Locale.ROOT, "Unable to convert value [%.128s] of type [VARCHAR] to a Double", randomString),
+                    sqle.getMessage());
+            sqle = expectThrows(SQLException.class, () -> results.getObject("test_keyword", Double.class));
+            assertEquals(format(Locale.ROOT, "Unable to convert value [%.128s] of type [VARCHAR] to a Double", randomString),
+                    sqle.getMessage());
+            
+            sqle = expectThrows(SQLException.class, () -> results.getDouble("test_date"));
+            assertEquals(format(Locale.ROOT, "Unable to convert value [%.128s] of type [TIMESTAMP] to a Double", randomDate),
+                    sqle.getMessage());
+            sqle = expectThrows(SQLException.class, () -> results.getObject("test_date", Double.class));
+            assertEquals(format(Locale.ROOT, "Unable to convert value [%.128s] of type [TIMESTAMP] to a Double", randomDate),
+                    sqle.getMessage());
+        });
+    }
+    
+    // Float values testing
+    public void testGettingValidFloatWithoutCasting() throws Exception {
+        float random1 = randomFloat();
+        float random2 = randomValueOtherThan(random1, () -> randomFloat());
+        float random3 = randomValueOtherThanMany(Arrays.asList(random1, random2)::contains, () -> randomFloat());
+        
+        createTestDataForFloatValueTests(random1, random2, random3);
+        
+        doWithQuery("SELECT test_float, test_null_float, test_keyword FROM test", (results) -> {
+            ResultSetMetaData resultSetMetaData = results.getMetaData();
+    
+            results.next();
+            assertEquals(3, resultSetMetaData.getColumnCount());
+            assertEquals(Types.REAL, resultSetMetaData.getColumnType(1));
+            assertEquals(Types.REAL, resultSetMetaData.getColumnType(2));
+            assertEquals(random1, results.getFloat(1), 0.0f);
+            assertEquals(random1, results.getFloat("test_float"), 0.0f);
+            assertEquals(random1, results.getObject("test_float", Float.class), 0.0f);
+            assertTrue(results.getObject(1) instanceof Float);
+            
+            assertEquals(0, results.getFloat(2), 0.0d);
+            assertTrue(results.wasNull());
+            assertEquals(null, results.getObject("test_null_float"));
+            assertTrue(results.wasNull());
+            
+            assertTrue(results.next());
+            assertEquals(random2, results.getFloat(1), 0.0d);
+            assertEquals(random2, results.getFloat("test_float"), 0.0d);
+            assertTrue(results.getObject(1) instanceof Float);
+            assertEquals(random3, results.getFloat("test_keyword"), 0.0d);
+            
+            assertFalse(results.next());
+        });
+    }
+    
+    public void testGettingValidFloatWithCasting() throws Exception {
+        Map<String,Number> map = createTestDataForNumericValueTypes(() -> randomFloat());
+        
+        doWithQuery(SELECT_WILDCARD, (results) -> {
+            results.next();
+            for(Entry<String, Number> e : map.entrySet()) {
+                assertEquals("For field " + e.getKey(), e.getValue().floatValue(), results.getFloat(e.getKey()), 0.0f);
+                assertEquals("For field " + e.getKey(), 
+                        e.getValue().floatValue(), results.getObject(e.getKey(), Float.class), 0.0f);
+            }
+        });
+    }
+
+    public void testGettingInvalidFloat() throws Exception {
+        createIndex("test");
+        updateMappingForNumericValuesTests("test");
+        updateMapping("test", builder -> {
+            builder.startObject("test_keyword").field("type", "keyword").endObject();
+            builder.startObject("test_date").field("type", "date").endObject();
+        });
+        
+        String randomString = randomUnicodeOfCodepointLengthBetween(128, 256);
+        long randomDate = randomLong();
+
+        index("test", "1", builder -> {
+            builder.field("test_keyword", randomString);
+            builder.field("test_date", randomDate);
+        });
+        
+        doWithQuery(SELECT_WILDCARD, (results) -> {
+            results.next();
+
+            SQLException sqle = expectThrows(SQLException.class, () -> results.getFloat("test_keyword"));
+            assertEquals(format(Locale.ROOT, "Unable to convert value [%.128s] of type [VARCHAR] to a Float", randomString),
+                    sqle.getMessage());
+            sqle = expectThrows(SQLException.class, () -> results.getObject("test_keyword", Float.class));
+            assertEquals(format(Locale.ROOT, "Unable to convert value [%.128s] of type [VARCHAR] to a Float", randomString),
+                    sqle.getMessage());
+            
+            sqle = expectThrows(SQLException.class, () -> results.getFloat("test_date"));
+            assertEquals(format(Locale.ROOT, "Unable to convert value [%.128s] of type [TIMESTAMP] to a Float", randomDate),
+                    sqle.getMessage());
+            sqle = expectThrows(SQLException.class, () -> results.getObject("test_date", Float.class));
+            assertEquals(format(Locale.ROOT, "Unable to convert value [%.128s] of type [TIMESTAMP] to a Float", randomDate),
+                    sqle.getMessage());
+        });
+    }
+    
+    public void testGettingBooleanValues() throws Exception {
+        createIndex("test");
+        updateMappingForNumericValuesTests("test");
+        updateMapping("test", builder -> {
+            builder.startObject("test_boolean").field("type", "boolean").endObject();
+            builder.startObject("test_date").field("type", "date").endObject();
+        });
+        long randomDate1 = randomLong();
+        long randomDate2 = randomLong();
+        
+        // true values
+        indexSimpleDocumentWithTrueValues(randomDate1);
+        
+        // false values
+        index("test", "2", builder -> {
+            builder.field("test_boolean", false);
+            builder.field("test_byte", 0);
+            builder.field("test_integer", 0);
+            builder.field("test_long", 0L);
+            builder.field("test_short", 0);
+            builder.field("test_double", 0d);
+            builder.field("test_float", 0f);
+            builder.field("test_keyword", "false");
+            builder.field("test_date", randomDate2);
+        });
+        
+        // other (non 0 = true) values
+        index("test", "3", builder -> {
+            builder.field("test_byte", randomValueOtherThan((byte) 0, () -> randomByte()));
+            builder.field("test_integer", randomValueOtherThan(0, () -> randomInt()));
+            builder.field("test_long", randomValueOtherThan(0L, () -> randomLong()));
+            builder.field("test_short", randomValueOtherThan((short) 0, () -> randomShort()));
+            builder.field("test_double", randomValueOtherThanMany(i -> i < 1.0d && i > -1.0d && i < Double.MAX_VALUE 
+                    && i > Double.MIN_VALUE,
+                    () -> randomDouble() * randomInt()));
+            builder.field("test_float", randomValueOtherThanMany(i -> i < 1.0f && i > -1.0f && i < Float.MAX_VALUE && i > Float.MIN_VALUE,
+                    () -> randomFloat() * randomInt()));
+            builder.field("test_keyword", "1");
+        });
+        
+        // other false values
+        index("test", "4", builder -> {
+            builder.field("test_keyword", "0");
+        });
+        
+        doWithQuery(SELECT_WILDCARD, (results) -> {
+            results.next();
+            assertEquals(true, results.getBoolean("test_boolean"));
+            for(String fld : fieldsNames) {
+                assertEquals("Expected: <true> but was: <false> for field " + fld, true, results.getBoolean(fld));
+                assertEquals("Expected: <true> but was: <false> for field " + fld, true, results.getObject(fld, Boolean.class));
+            }
+            SQLException sqle = expectThrows(SQLException.class, () -> results.getBoolean("test_date"));
+            assertEquals(format(Locale.ROOT, "Unable to convert value [%.128s] of type [TIMESTAMP] to a Boolean", randomDate1),
+                    sqle.getMessage());
+            
+            results.next();
+            assertEquals(false, results.getBoolean("test_boolean"));
+            for(String fld : fieldsNames) {
+                assertEquals("Expected: <false> but was: <true> for field " + fld, false, results.getBoolean(fld));
+                assertEquals("Expected: <false> but was: <true> for field " + fld, false, results.getObject(fld, Boolean.class));
+            }
+            sqle = expectThrows(SQLException.class, () -> results.getBoolean("test_date"));
+            assertEquals(format(Locale.ROOT, "Unable to convert value [%.128s] of type [TIMESTAMP] to a Boolean", randomDate2),
+                    sqle.getMessage());
+            
+            sqle = expectThrows(SQLException.class, () -> results.getObject("test_date", Boolean.class));
+            assertEquals(format(Locale.ROOT, "Unable to convert value [%.128s] of type [TIMESTAMP] to a Boolean", randomDate2),
+                    sqle.getMessage());
+            
+            results.next();
+            for(String fld : fieldsNames.stream()
+                    .filter((f) -> !f.equals("test_keyword")).collect(Collectors.toCollection(HashSet::new))) {
+                assertEquals("Expected: <true> but was: <false> for field " + fld, true, results.getBoolean(fld));
+                assertEquals("Expected: <true> but was: <false> for field " + fld, true, results.getObject(fld, Boolean.class));
+            }
+            
+            results.next();
+            assertEquals(false, results.getBoolean("test_keyword"));
+            assertEquals(false, results.getObject("test_keyword", Boolean.class));
+        });
+    }
+    
+    public void testGettingDateWithoutCalendar() throws Exception {
+        createIndex("test");
+        updateMappingForNumericValuesTests("test");
+        updateMapping("test", builder -> {
+            builder.startObject("test_boolean").field("type", "boolean").endObject();
+            builder.startObject("test_date").field("type", "date").endObject();
+        });
+        Long randomLongDate = randomLong();
+        indexSimpleDocumentWithTrueValues(randomLongDate);
+        
+        String timeZoneId = randomKnownTimeZone();
+        Calendar connCalendar = Calendar.getInstance(TimeZone.getTimeZone(timeZoneId), Locale.ROOT);
+        
+        doWithQueryAndTimezone(SELECT_ALL_FIELDS, timeZoneId, (results) -> {
+            results.next();
+            connCalendar.setTimeInMillis(randomLongDate);
+            connCalendar.set(HOUR_OF_DAY, 0);
+            connCalendar.set(MINUTE, 0);
+            connCalendar.set(SECOND, 0);
+            connCalendar.set(MILLISECOND, 0);
+            
+            assertEquals(results.getDate("test_date"), new java.sql.Date(connCalendar.getTimeInMillis()));
+            assertEquals(results.getDate(9), new java.sql.Date(connCalendar.getTimeInMillis()));
+            assertEquals(results.getObject("test_date", java.sql.Date.class), 
+                    new java.sql.Date(randomLongDate - (randomLongDate % 86400000L)));
+            assertEquals(results.getObject(9, java.sql.Date.class), 
+                    new java.sql.Date(randomLongDate - (randomLongDate % 86400000L)));
+
+            // bulk validation for all fields which are not of type date
+            validateErrorsForDateTimeTestsWithoutCalendar(results::getDate);
+        });
+    }
+    
+    public void testGettingDateWithCalendar() throws Exception {
+        createIndex("test");
+        updateMappingForNumericValuesTests("test");
+        updateMapping("test", builder -> {
+            builder.startObject("test_boolean").field("type", "boolean").endObject();
+            builder.startObject("test_date").field("type", "date").endObject();
+        });
+        Long randomLongDate = randomLong();
+        indexSimpleDocumentWithTrueValues(randomLongDate);
+        index("test", "2", builder -> {
+            builder.timeField("test_date", null);
+        });
+        
+        String timeZoneId = randomKnownTimeZone();
+        String anotherTZId = randomValueOtherThan(timeZoneId, () -> randomKnownTimeZone());
+        Calendar c = Calendar.getInstance(TimeZone.getTimeZone(anotherTZId), Locale.ROOT);
+        
+        doWithQueryAndTimezone(SELECT_ALL_FIELDS, timeZoneId, (results) -> {
+            results.next();
+            c.setTimeInMillis(randomLongDate);
+            c.set(HOUR_OF_DAY, 0);
+            c.set(MINUTE, 0);
+            c.set(SECOND, 0);
+            c.set(MILLISECOND, 0);
+
+            assertEquals(results.getDate("test_date", c), new java.sql.Date(c.getTimeInMillis()));
+            assertEquals(results.getDate(9, c), new java.sql.Date(c.getTimeInMillis()));
+            
+            // bulk validation for all fields which are not of type date
+            validateErrorsForDateTimeTestsWithCalendar(c, results::getDate);
+            
+            results.next();
+            assertNull(results.getDate("test_date"));
+        });
+    }
+    
+    public void testGettingTimeWithoutCalendar() throws Exception {
+        createIndex("test");
+        updateMappingForNumericValuesTests("test");
+        updateMapping("test", builder -> {
+            builder.startObject("test_boolean").field("type", "boolean").endObject();
+            builder.startObject("test_date").field("type", "date").endObject();
+        });
+        Long randomLongDate = randomLong();
+        indexSimpleDocumentWithTrueValues(randomLongDate);
+        
+        String timeZoneId = randomKnownTimeZone();
+        Calendar c = Calendar.getInstance(TimeZone.getTimeZone(timeZoneId), Locale.ROOT);
+        
+        doWithQueryAndTimezone(SELECT_ALL_FIELDS, timeZoneId, (results) -> {
+            results.next();
+            c.setTimeInMillis(randomLongDate);
+            c.set(ERA, GregorianCalendar.AD);
+            c.set(YEAR, 1970);
+            c.set(MONTH, 0);
+            c.set(DAY_OF_MONTH, 1);
+            
+            assertEquals(results.getTime("test_date"), new java.sql.Time(c.getTimeInMillis()));
+            assertEquals(results.getTime(9), new java.sql.Time(c.getTimeInMillis()));
+            assertEquals(results.getObject("test_date", java.sql.Time.class), 
+                    new java.sql.Time(randomLongDate % 86400000L));
+            assertEquals(results.getObject(9, java.sql.Time.class), 
+                    new java.sql.Time(randomLongDate % 86400000L));
+            
+            validateErrorsForDateTimeTestsWithoutCalendar(results::getTime);
+        });
+    }
+    
+    public void testGettingTimeWithCalendar() throws Exception {
+        createIndex("test");
+        updateMappingForNumericValuesTests("test");
+        updateMapping("test", builder -> {
+            builder.startObject("test_boolean").field("type", "boolean").endObject();
+            builder.startObject("test_date").field("type", "date").endObject();
+        });
+        Long randomLongDate = randomLong();
+        indexSimpleDocumentWithTrueValues(randomLongDate);
+        index("test", "2", builder -> {
+            builder.timeField("test_date", null);
+        });
+        
+        String timeZoneId = randomKnownTimeZone();
+        String anotherTZId = randomValueOtherThan(timeZoneId, () -> randomKnownTimeZone());
+        Calendar c = Calendar.getInstance(TimeZone.getTimeZone(anotherTZId), Locale.ROOT);
+        
+        doWithQueryAndTimezone(SELECT_ALL_FIELDS, timeZoneId, (results) -> {
+            results.next();
+            c.setTimeInMillis(randomLongDate);
+            c.set(ERA, GregorianCalendar.AD);
+            c.set(YEAR, 1970);
+            c.set(MONTH, 0);
+            c.set(DAY_OF_MONTH, 1);
+            
+            assertEquals(results.getTime("test_date", c), new java.sql.Time(c.getTimeInMillis()));
+            assertEquals(results.getTime(9, c), new java.sql.Time(c.getTimeInMillis()));
+            
+            validateErrorsForDateTimeTestsWithCalendar(c, results::getTime);
+            
+            results.next();
+            assertNull(results.getTime("test_date"));
+        });
+    }
+    
+    public void testGettingTimestampWithoutCalendar() throws Exception {
+        createIndex("library");
+        updateMapping("library", builder -> {
+            builder.startObject("release_date").field("type", "date").endObject();
+            builder.startObject("republish_date").field("type", "date").endObject();
+        });
+        long randomMillis = randomLong();
 
         index("library", "1", builder -> {
             builder.field("name", "Don Quixote");
             builder.field("page_count", 1072);
-            builder.timeField("release_date", new Date(randomMillis));
+            builder.field("release_date", randomMillis);
             builder.timeField("republish_date", null);
         });
         index("library", "2", builder -> {
             builder.field("name", "1984");
             builder.field("page_count", 328);
-            builder.timeField("release_date", new Date(-649036800000L));
-            builder.timeField("republish_date", new Date(599616000000L));
+            builder.field("release_date", -649036800000L);
+            builder.field("republish_date", 599616000000L);
         });
 
-        try (Connection connection = esJdbc()) {
-            try (PreparedStatement statement = connection.prepareStatement("SELECT name, release_date, republish_date FROM library")) {
-                try (ResultSet results = statement.executeQuery()) {
-                    ResultSetMetaData resultSetMetaData = results.getMetaData();
+        doWithQuery("SELECT name, release_date, republish_date FROM library", (results) -> {
+            ResultSetMetaData resultSetMetaData = results.getMetaData();
 
-                    results.next();
-                    assertEquals(3, resultSetMetaData.getColumnCount());
-                    assertEquals(randomMillis, results.getTimestamp("release_date").getTime());
-                    assertEquals(randomMillis, results.getTimestamp(2).getTime());
-                    assertTrue(results.getObject(2) instanceof Timestamp);
-                    assertEquals(randomMillis, ((Timestamp) results.getObject("release_date")).getTime());
-                    
-                    assertNull(results.getTimestamp(3));
-                    assertNull(results.getObject("republish_date"));
+            results.next();
+            assertEquals(3, resultSetMetaData.getColumnCount());
+            assertEquals(randomMillis, results.getTimestamp("release_date").getTime());
+            assertEquals(randomMillis, results.getTimestamp(2).getTime());
+            assertTrue(results.getObject(2) instanceof Timestamp);
+            assertEquals(randomMillis, ((Timestamp) results.getObject("release_date")).getTime());
+            
+            assertNull(results.getTimestamp(3));
+            assertNull(results.getObject("republish_date"));
 
-                    assertTrue(results.next());
-                    assertEquals(599616000000L, results.getTimestamp("republish_date").getTime());
-                    assertEquals(-649036800000L, ((Timestamp) results.getObject(2)).getTime());
+            assertTrue(results.next());
+            assertEquals(599616000000L, results.getTimestamp("republish_date").getTime());
+            assertEquals(-649036800000L, ((Timestamp) results.getObject(2)).getTime());
 
-                    assertFalse(results.next());
-                }
-            }
-        }
+            assertFalse(results.next());
+        });
+    }
+    
+    public void testGettingTimestampWithCalendar() throws Exception {
+        createIndex("test");
+        updateMappingForNumericValuesTests("test");
+        updateMapping("test", builder -> {
+            builder.startObject("test_boolean").field("type", "boolean").endObject();
+            builder.startObject("test_date").field("type", "date").endObject();
+        });
+        Long randomLongDate = randomLong();
+        indexSimpleDocumentWithTrueValues(randomLongDate);
+        index("test", "2", builder -> {
+            builder.timeField("test_date", null);
+        });
+        
+        String timeZoneId = randomKnownTimeZone();
+        String anotherTZId = randomValueOtherThan(timeZoneId, () -> randomKnownTimeZone());
+        Calendar c = Calendar.getInstance(TimeZone.getTimeZone(anotherTZId), Locale.ROOT);
+        
+        doWithQueryAndTimezone(SELECT_ALL_FIELDS, timeZoneId, (results) -> {
+            results.next();
+            c.setTimeInMillis(randomLongDate);
+            
+            assertEquals(results.getTimestamp("test_date", c), new java.sql.Timestamp(c.getTimeInMillis()));
+            assertEquals(results.getTimestamp(9, c), new java.sql.Timestamp(c.getTimeInMillis()));
+            
+            validateErrorsForDateTimeTestsWithCalendar(c, results::getTimestamp);
+
+            results.next();
+            assertNull(results.getTimestamp("test_date"));
+        });
+    }
+    
+    public void testValidGetObjectCalls() throws Exception {
+        createIndex("test");
+        updateMappingForNumericValuesTests("test");
+        updateMapping("test", builder -> {
+            builder.startObject("test_boolean").field("type", "boolean").endObject();
+            builder.startObject("test_date").field("type", "date").endObject();
+        });
+        
+        byte b = randomByte();
+        int i = randomInt();
+        long l = randomLong();
+        short s = (short) randomIntBetween(Short.MIN_VALUE, Short.MAX_VALUE);
+        double d = randomDouble();
+        float f = randomFloat();
+        boolean randomBool = randomBoolean();
+        Long randomLongDate = randomLong();
+        String randomString = randomUnicodeOfCodepointLengthBetween(128, 256);
+                
+        index("test", "1", builder -> {
+            builder.field("test_byte", b);
+            builder.field("test_integer", i);
+            builder.field("test_long", l);
+            builder.field("test_short", s);
+            builder.field("test_double", d);
+            builder.field("test_float", f);
+            builder.field("test_keyword", randomString);
+            builder.field("test_date", randomLongDate);
+            builder.field("test_boolean", randomBool);
+        });
+        
+        doWithQuery(SELECT_WILDCARD, (results) -> {
+            results.next();
+            
+            assertEquals(b, results.getObject("test_byte"));
+            assertTrue(results.getObject("test_byte") instanceof Byte);
+            
+            assertEquals(i, results.getObject("test_integer"));
+            assertTrue(results.getObject("test_integer") instanceof Integer);
+            
+            assertEquals(l, results.getObject("test_long"));
+            assertTrue(results.getObject("test_long") instanceof Long);
+            
+            assertEquals(s, results.getObject("test_short"));
+            assertTrue(results.getObject("test_short") instanceof Short);
+            
+            assertEquals(d, results.getObject("test_double"));
+            assertTrue(results.getObject("test_double") instanceof Double);
+            
+            assertEquals(f, results.getObject("test_float"));
+            assertTrue(results.getObject("test_float") instanceof Float);
+            
+            assertEquals(randomString, results.getObject("test_keyword"));
+            assertTrue(results.getObject("test_keyword") instanceof String);
+            
+            assertEquals(new Date(randomLongDate), results.getObject("test_date"));
+            assertTrue(results.getObject("test_date") instanceof Timestamp);
+            
+            assertEquals(randomBool, results.getObject("test_boolean"));
+            assertTrue(results.getObject("test_boolean") instanceof Boolean);
+        });
     }
 
     /*
@@ -79,4 +1091,458 @@ public class ResultSetTestCase extends JdbcIntegrationTestCase {
             fail("Infinite recursive call on getObject() method");
         }
     }
+    
+    public void testUnsupportedGetMethods() throws IOException, SQLException {
+        index("test", "1", builder -> {
+            builder.field("test", "test");
+        });
+        Connection conn = esJdbc();
+        PreparedStatement statement = conn.prepareStatement("SELECT * FROM test");
+        ResultSet r = statement.executeQuery();
+        
+        r.next();
+        assertThrowsUnsupportedAndExpectErrorMessage(() -> r.getAsciiStream("test"), "AsciiStream not supported");
+        assertThrowsUnsupportedAndExpectErrorMessage(() -> r.getAsciiStream(1), "AsciiStream not supported");
+        assertThrowsUnsupportedAndExpectErrorMessage(() -> r.getArray("test"), "Array not supported");
+        assertThrowsUnsupportedAndExpectErrorMessage(() -> r.getArray(1), "Array not supported");
+        assertThrowsUnsupportedAndExpectErrorMessage(() -> r.getBigDecimal("test"), "BigDecimal not supported");
+        assertThrowsUnsupportedAndExpectErrorMessage(() -> r.getBigDecimal("test"), "BigDecimal not supported");
+        assertThrowsUnsupportedAndExpectErrorMessage(() -> r.getBinaryStream("test"), "BinaryStream not supported");
+        assertThrowsUnsupportedAndExpectErrorMessage(() -> r.getBinaryStream(1), "BinaryStream not supported");
+        assertThrowsUnsupportedAndExpectErrorMessage(() -> r.getBlob("test"), "Blob not supported");
+        assertThrowsUnsupportedAndExpectErrorMessage(() -> r.getBlob(1), "Blob not supported");
+        assertThrowsUnsupportedAndExpectErrorMessage(() -> r.getCharacterStream("test"), "CharacterStream not supported");
+        assertThrowsUnsupportedAndExpectErrorMessage(() -> r.getCharacterStream(1), "CharacterStream not supported");
+        assertThrowsUnsupportedAndExpectErrorMessage(() -> r.getClob("test"), "Clob not supported");
+        assertThrowsUnsupportedAndExpectErrorMessage(() -> r.getClob(1), "Clob not supported");
+        assertThrowsUnsupportedAndExpectErrorMessage(() -> r.getNCharacterStream("test"), "NCharacterStream not supported");
+        assertThrowsUnsupportedAndExpectErrorMessage(() -> r.getNCharacterStream(1), "NCharacterStream not supported");
+        assertThrowsUnsupportedAndExpectErrorMessage(() -> r.getNClob("test"), "NClob not supported");
+        assertThrowsUnsupportedAndExpectErrorMessage(() -> r.getNClob(1), "NClob not supported");
+        assertThrowsUnsupportedAndExpectErrorMessage(() -> r.getNString("test"), "NString not supported");
+        assertThrowsUnsupportedAndExpectErrorMessage(() -> r.getNString(1), "NString not supported");
+        assertThrowsUnsupportedAndExpectErrorMessage(() -> r.getRef("test"), "Ref not supported");
+        assertThrowsUnsupportedAndExpectErrorMessage(() -> r.getRef(1), "Ref not supported");
+        assertThrowsUnsupportedAndExpectErrorMessage(() -> r.getRowId("test"), "RowId not supported");
+        assertThrowsUnsupportedAndExpectErrorMessage(() -> r.getRowId(1), "RowId not supported");
+        assertThrowsUnsupportedAndExpectErrorMessage(() -> r.getSQLXML("test"), "SQLXML not supported");
+        assertThrowsUnsupportedAndExpectErrorMessage(() -> r.getSQLXML(1), "SQLXML not supported");        
+        assertThrowsUnsupportedAndExpectErrorMessage(() -> r.getURL("test"), "URL not supported");
+        assertThrowsUnsupportedAndExpectErrorMessage(() -> r.getURL(1), "URL not supported");
+    }
+    
+    public void testUnsupportedUpdateMethods() throws IOException, SQLException {
+        index("test", "1", builder -> {
+            builder.field("test", "test");
+        });
+        Connection conn = esJdbc();
+        PreparedStatement statement = conn.prepareStatement("SELECT * FROM test");
+        ResultSet r = statement.executeQuery();
+        
+        r.next();
+        Blob b = null;
+        InputStream i = null;
+        Clob c = null;
+        NClob nc = null;
+        Reader rd = null;
+        
+        assertThrowsWritesUnsupportedForUpdate(() -> r.updateBytes(1, null));
+        assertThrowsWritesUnsupportedForUpdate(() -> r.updateBytes("", null));
+        assertThrowsWritesUnsupportedForUpdate(() -> r.updateArray(1, null));
+        assertThrowsWritesUnsupportedForUpdate(() -> r.updateArray("", null));
+        assertThrowsWritesUnsupportedForUpdate(() -> r.updateAsciiStream(1, null));
+        assertThrowsWritesUnsupportedForUpdate(() -> r.updateAsciiStream("", null));
+        assertThrowsWritesUnsupportedForUpdate(() -> r.updateAsciiStream(1, null, 1));
+        assertThrowsWritesUnsupportedForUpdate(() -> r.updateAsciiStream(1, null, 1));
+        assertThrowsWritesUnsupportedForUpdate(() -> r.updateAsciiStream("", null, 1));
+        assertThrowsWritesUnsupportedForUpdate(() -> r.updateAsciiStream("", null, 1L));
+        assertThrowsWritesUnsupportedForUpdate(() -> r.updateBigDecimal(1, null));
+        assertThrowsWritesUnsupportedForUpdate(() -> r.updateBigDecimal("", null));
+        assertThrowsWritesUnsupportedForUpdate(() -> r.updateBinaryStream(1, null));
+        assertThrowsWritesUnsupportedForUpdate(() -> r.updateBinaryStream("", null));
+        assertThrowsWritesUnsupportedForUpdate(() -> r.updateBinaryStream(1, null, 1));
+        assertThrowsWritesUnsupportedForUpdate(() -> r.updateBinaryStream(1, null, 1));
+        assertThrowsWritesUnsupportedForUpdate(() -> r.updateBinaryStream("", null, 1));
+        assertThrowsWritesUnsupportedForUpdate(() -> r.updateBinaryStream("", null, 1L));
+        assertThrowsWritesUnsupportedForUpdate(() -> r.updateBlob(1, b));
+        assertThrowsWritesUnsupportedForUpdate(() -> r.updateBlob(1, i));
+        assertThrowsWritesUnsupportedForUpdate(() -> r.updateBlob("", b));
+        assertThrowsWritesUnsupportedForUpdate(() -> r.updateBlob("", i));
+        assertThrowsWritesUnsupportedForUpdate(() -> r.updateBlob(1, null, 1));
+        assertThrowsWritesUnsupportedForUpdate(() -> r.updateBlob("", null, 1));
+        assertThrowsWritesUnsupportedForUpdate(() -> r.updateBoolean(1, false));
+        assertThrowsWritesUnsupportedForUpdate(() -> r.updateBoolean("", false));
+        assertThrowsWritesUnsupportedForUpdate(() -> r.updateByte(1, (byte) 1));
+        assertThrowsWritesUnsupportedForUpdate(() -> r.updateByte("", (byte) 1));
+        assertThrowsWritesUnsupportedForUpdate(() -> r.updateCharacterStream(1, null));
+        assertThrowsWritesUnsupportedForUpdate(() -> r.updateCharacterStream("", null));
+        assertThrowsWritesUnsupportedForUpdate(() -> r.updateCharacterStream(1, null, 1));
+        assertThrowsWritesUnsupportedForUpdate(() -> r.updateCharacterStream(1, null, 1L));
+        assertThrowsWritesUnsupportedForUpdate(() -> r.updateCharacterStream("", null, 1));
+        assertThrowsWritesUnsupportedForUpdate(() -> r.updateCharacterStream("", null, 1L));
+        assertThrowsWritesUnsupportedForUpdate(() -> r.updateClob(1, c));
+        assertThrowsWritesUnsupportedForUpdate(() -> r.updateClob(1, rd));
+        assertThrowsWritesUnsupportedForUpdate(() -> r.updateClob("", c));
+        assertThrowsWritesUnsupportedForUpdate(() -> r.updateClob("", rd));
+        assertThrowsWritesUnsupportedForUpdate(() -> r.updateClob(1, null, 1L));
+        assertThrowsWritesUnsupportedForUpdate(() -> r.updateClob("", null, 1L));
+        assertThrowsWritesUnsupportedForUpdate(() -> r.updateDate(1, null));
+        assertThrowsWritesUnsupportedForUpdate(() -> r.updateDate("", null));
+        assertThrowsWritesUnsupportedForUpdate(() -> r.updateDouble(1, 0d));
+        assertThrowsWritesUnsupportedForUpdate(() -> r.updateDouble("", 0d));
+        assertThrowsWritesUnsupportedForUpdate(() -> r.updateFloat(1, 0f));
+        assertThrowsWritesUnsupportedForUpdate(() -> r.updateFloat("", 0f));
+        assertThrowsWritesUnsupportedForUpdate(() -> r.updateInt(1, 0));
+        assertThrowsWritesUnsupportedForUpdate(() -> r.updateInt("", 0));
+        assertThrowsWritesUnsupportedForUpdate(() -> r.updateLong(1, 0L));
+        assertThrowsWritesUnsupportedForUpdate(() -> r.updateLong("", 0L));
+        assertThrowsWritesUnsupportedForUpdate(() -> r.updateNCharacterStream(1, null));
+        assertThrowsWritesUnsupportedForUpdate(() -> r.updateNCharacterStream("", null));
+        assertThrowsWritesUnsupportedForUpdate(() -> r.updateNCharacterStream(1, null, 1L));
+        assertThrowsWritesUnsupportedForUpdate(() -> r.updateNCharacterStream("", null, 1L));
+        assertThrowsWritesUnsupportedForUpdate(() -> r.updateNClob(1, nc));
+        assertThrowsWritesUnsupportedForUpdate(() -> r.updateNClob(1, rd));
+        assertThrowsWritesUnsupportedForUpdate(() -> r.updateNClob("", nc));
+        assertThrowsWritesUnsupportedForUpdate(() -> r.updateNClob("", rd));
+        assertThrowsWritesUnsupportedForUpdate(() -> r.updateNClob(1, null, 1));
+        assertThrowsWritesUnsupportedForUpdate(() -> r.updateNClob("", null, 1));
+        assertThrowsWritesUnsupportedForUpdate(() -> r.updateNString(1, null));
+        assertThrowsWritesUnsupportedForUpdate(() -> r.updateNString("", null));
+        assertThrowsWritesUnsupportedForUpdate(() -> r.updateNull(1));
+        assertThrowsWritesUnsupportedForUpdate(() -> r.updateNull(""));
+        assertThrowsWritesUnsupportedForUpdate(() -> r.updateObject(1, null));
+        assertThrowsWritesUnsupportedForUpdate(() -> r.updateObject("", null));
+        assertThrowsWritesUnsupportedForUpdate(() -> r.updateObject(1, null, 1));
+        assertThrowsWritesUnsupportedForUpdate(() -> r.updateObject("", null, 1));
+        assertThrowsWritesUnsupportedForUpdate(() -> r.updateRef(1, null));
+        assertThrowsWritesUnsupportedForUpdate(() -> r.updateRef("", null));
+        assertThrowsWritesUnsupportedForUpdate(() -> r.updateRow());
+        assertThrowsWritesUnsupportedForUpdate(() -> r.updateRowId(1, null));
+        assertThrowsWritesUnsupportedForUpdate(() -> r.updateRowId("", null));
+        assertThrowsWritesUnsupportedForUpdate(() -> r.updateSQLXML(1, null));
+        assertThrowsWritesUnsupportedForUpdate(() -> r.updateSQLXML("", null));
+        assertThrowsWritesUnsupportedForUpdate(() -> r.updateShort(1, (short) 0));
+        assertThrowsWritesUnsupportedForUpdate(() -> r.updateShort("", (short) 0));
+        assertThrowsWritesUnsupportedForUpdate(() -> r.updateString(1, null));
+        assertThrowsWritesUnsupportedForUpdate(() -> r.updateString("", null));
+        assertThrowsWritesUnsupportedForUpdate(() -> r.updateTime(1, null));
+        assertThrowsWritesUnsupportedForUpdate(() -> r.updateTime("", null));
+        assertThrowsWritesUnsupportedForUpdate(() -> r.updateTimestamp(1, null));
+        assertThrowsWritesUnsupportedForUpdate(() -> r.updateTimestamp("", null));
+        assertThrowsWritesUnsupportedForUpdate(() -> r.insertRow());
+        assertThrowsWritesUnsupportedForUpdate(() -> r.updateRow());
+        assertThrowsWritesUnsupportedForUpdate(() -> r.deleteRow());
+        assertThrowsWritesUnsupportedForUpdate(() -> r.cancelRowUpdates());
+        assertThrowsWritesUnsupportedForUpdate(() -> r.moveToInsertRow());
+        assertThrowsWritesUnsupportedForUpdate(() -> r.refreshRow());
+        assertThrowsWritesUnsupportedForUpdate(() -> r.moveToCurrentRow());
+        assertThrowsWritesUnsupportedForUpdate(() -> r.rowUpdated());
+        assertThrowsWritesUnsupportedForUpdate(() -> r.rowInserted());
+        assertThrowsWritesUnsupportedForUpdate(() -> r.rowDeleted());
+    }
+    
+    private void doWithQuery(String query, CheckedConsumer<ResultSet, SQLException> consumer) throws SQLException {
+        try (Connection connection = esJdbc()) {
+            try (PreparedStatement statement = connection.prepareStatement(query)) {
+                try (ResultSet results = statement.executeQuery()) {
+                    consumer.accept(results);
+                }
+            }
+        }
+    }
+    
+    private void doWithQueryAndTimezone(String query, String tz, CheckedConsumer<ResultSet, SQLException> consumer) throws SQLException {
+        try (Connection connection = esJdbc(tz)) {
+            try (PreparedStatement statement = connection.prepareStatement(query)) {
+                try (ResultSet results = statement.executeQuery()) {
+                    consumer.accept(results);
+                }
+            }
+        }
+    }
+    
+    private void createIndex(String index) throws Exception {
+        Request request = new Request("PUT", "/" + index);
+        XContentBuilder createIndex = JsonXContent.contentBuilder().startObject();
+        createIndex.startObject("settings");
+        {
+            createIndex.field("number_of_shards", 1);
+            createIndex.field("number_of_replicas", 1);
+        }
+        createIndex.endObject();
+        createIndex.startObject("mappings");
+        {
+            createIndex.startObject("doc");
+            {
+                createIndex.startObject("properties");
+                {}
+                createIndex.endObject();
+            }
+            createIndex.endObject();
+        }
+        createIndex.endObject().endObject();
+        request.setJsonEntity(Strings.toString(createIndex));
+        client().performRequest(request);
+    }
+
+    private void updateMapping(String index, CheckedConsumer<XContentBuilder, IOException> body) throws Exception {
+        Request request = new Request("PUT", "/" + index + "/_mapping/doc");
+        XContentBuilder updateMapping = JsonXContent.contentBuilder().startObject();
+        updateMapping.startObject("properties");
+        {
+            body.accept(updateMapping);
+        }
+        updateMapping.endObject().endObject();
+        
+        request.setJsonEntity(Strings.toString(updateMapping));
+        client().performRequest(request);
+    }
+
+    private void createTestDataForByteValueTests(byte random1, byte random2, byte random3) throws Exception, IOException {
+        createIndex("test");
+        updateMapping("test", builder -> {
+            builder.startObject("test_byte").field("type", "byte").endObject();
+            builder.startObject("test_null_byte").field("type", "byte").endObject();
+            builder.startObject("test_keyword").field("type", "keyword").endObject();
+        });
+        
+        index("test", "1", builder -> {
+            builder.field("test_byte", random1);
+            builder.field("test_null_byte", (Byte) null);
+        });
+        index("test", "2", builder -> {
+            builder.field("test_byte", random2);
+            builder.field("test_keyword", random3);
+        });
+    }
+
+    private void createTestDataForShortValueTests(short random1, short random2, short random3) throws Exception, IOException {
+        createIndex("test");
+        updateMapping("test", builder -> {
+            builder.startObject("test_short").field("type", "short").endObject();
+            builder.startObject("test_null_short").field("type", "short").endObject();
+            builder.startObject("test_keyword").field("type", "keyword").endObject();
+        });
+        
+        index("test", "1", builder -> {
+            builder.field("test_short", random1);
+            builder.field("test_null_short", (Short) null);
+        });
+        index("test", "2", builder -> {
+            builder.field("test_short", random2);
+            builder.field("test_keyword", random3);
+        });
+    }
+
+    private void createTestDataForIntegerValueTests(int random1, int random2, int random3) throws Exception, IOException {
+        createIndex("test");
+        updateMapping("test", builder -> {
+            builder.startObject("test_integer").field("type", "integer").endObject();
+            builder.startObject("test_null_integer").field("type", "integer").endObject();
+            builder.startObject("test_keyword").field("type", "keyword").endObject();
+        });
+        
+        index("test", "1", builder -> {
+            builder.field("test_integer", random1);
+            builder.field("test_null_integer", (Integer) null);
+        });
+        index("test", "2", builder -> {
+            builder.field("test_integer", random2);
+            builder.field("test_keyword", random3);
+        });
+    }
+
+    private void createTestDataForLongValueTests(long random1, long random2, long random3) throws Exception, IOException {
+        createIndex("test");
+        updateMapping("test", builder -> {
+            builder.startObject("test_long").field("type", "long").endObject();
+            builder.startObject("test_null_long").field("type", "long").endObject();
+            builder.startObject("test_keyword").field("type", "keyword").endObject();
+        });
+        
+        index("test", "1", builder -> {
+            builder.field("test_long", random1);
+            builder.field("test_null_long", (Long) null);
+        });
+        index("test", "2", builder -> {
+            builder.field("test_long", random2);
+            builder.field("test_keyword", random3);
+        });
+    }
+
+    private void createTestDataForDoubleValueTests(double random1, double random2, double random3) throws Exception, IOException {
+        createIndex("test");
+        updateMapping("test", builder -> {
+            builder.startObject("test_double").field("type", "double").endObject();
+            builder.startObject("test_null_double").field("type", "double").endObject();
+            builder.startObject("test_keyword").field("type", "keyword").endObject();
+        });
+        
+        index("test", "1", builder -> {
+            builder.field("test_double", random1);
+            builder.field("test_null_double", (Double) null);
+        });
+        index("test", "2", builder -> {
+            builder.field("test_double", random2);
+            builder.field("test_keyword", random3);
+        });
+    }
+
+    private void createTestDataForFloatValueTests(float random1, float random2, float random3) throws Exception, IOException {
+        createIndex("test");
+        updateMapping("test", builder -> {
+            builder.startObject("test_float").field("type", "float").endObject();
+            builder.startObject("test_null_float").field("type", "float").endObject();
+            builder.startObject("test_keyword").field("type", "keyword").endObject();
+        });
+        
+        index("test", "1", builder -> {
+            builder.field("test_float", random1);
+            builder.field("test_null_float", (Double) null);
+        });
+        index("test", "2", builder -> {
+            builder.field("test_float", random2);
+            builder.field("test_keyword", random3);
+        });
+    }
+    
+    private void indexSimpleDocumentWithTrueValues(Long randomLongDate) throws IOException {
+        index("test", "1", builder -> {
+            builder.field("test_boolean", true);
+            builder.field("test_byte", 1);
+            builder.field("test_integer", 1);
+            builder.field("test_long", 1L);
+            builder.field("test_short", 1);
+            builder.field("test_double", 1d);
+            builder.field("test_float", 1f);
+            builder.field("test_keyword", "true");
+            builder.field("test_date", randomLongDate);
+        });
+    }
+
+    /**
+     * Creates test data for all numeric get* methods. All values random and different from the other numeric fields already generated.
+     * It returns a map containing the field name and its randomly generated value to be later used in checking the returned values.
+     */
+    private Map<String,Number> createTestDataForNumericValueTypes(Supplier<Number> randomGenerator) throws Exception, IOException {
+        Map<String,Number> map = new HashMap<String,Number>();
+        createIndex("test");
+        updateMappingForNumericValuesTests("test");
+    
+        index("test", "1", builder -> {
+            // random Byte
+            byte test_byte = randomValueOtherThanMany(map::containsValue, randomGenerator).byteValue();
+            builder.field("test_byte", test_byte);
+            map.put("test_byte", test_byte);
+            
+            // random Integer
+            int test_integer = randomValueOtherThanMany(map::containsValue, randomGenerator).intValue();
+            builder.field("test_integer", test_integer);
+            map.put("test_integer", test_integer);
+    
+            // random Short
+            int test_short = randomValueOtherThanMany(map::containsValue, randomGenerator).shortValue();
+            builder.field("test_short", test_short);
+            map.put("test_short", test_short);
+            
+            // random Long
+            long test_long = randomValueOtherThanMany(map::containsValue, randomGenerator).longValue();
+            builder.field("test_long", test_long);
+            map.put("test_long", test_long);
+            
+            // random Double
+            double test_double = randomValueOtherThanMany(map::containsValue, randomGenerator).doubleValue();
+            builder.field("test_double", test_double);
+            map.put("test_double", test_double);
+            
+            // random Float
+            float test_float = randomValueOtherThanMany(map::containsValue, randomGenerator).floatValue();
+            builder.field("test_float", test_float);
+            map.put("test_float", test_float);
+        });
+        return map;
+    }
+
+    private void updateMappingForNumericValuesTests(String indexName) throws Exception {
+        updateMapping(indexName, builder -> {
+            for(String field : fieldsNames) {
+                builder.startObject(field).field("type", field.substring(5)).endObject();
+            }
+        });
+    }
+    
+    private void assertThrowsUnsupportedAndExpectErrorMessage(ThrowingRunnable runnable, String message) {
+        SQLException sqle = expectThrows(SQLFeatureNotSupportedException.class, runnable);
+        assertEquals(message, sqle.getMessage());
+    }
+
+    private void assertThrowsWritesUnsupportedForUpdate(ThrowingRunnable r) {
+        assertThrowsUnsupportedAndExpectErrorMessage(r, "Writes not supported");
+    }
+    
+    private void validateErrorsForDateTimeTestsWithoutCalendar(CheckedFunction<String,Object,SQLException> method) {
+        SQLException sqle;
+        for(Entry<Tuple<String,Object>,JDBCType> field : dateTimeTestingFields.entrySet()) {
+            sqle = expectThrows(SQLException.class, () -> method.apply(field.getKey().v1()));
+            assertEquals(
+                    format(Locale.ROOT, "Unable to convert value [%.128s] of type [%s] to a Long", 
+                            field.getKey().v2(), field.getValue()), sqle.getMessage());
+        }
+    }
+    
+    private void validateErrorsForDateTimeTestsWithCalendar(Calendar c, CheckedBiFunction<String,Calendar,Object,SQLException> method) {
+        SQLException sqle;
+        for(Entry<Tuple<String,Object>,JDBCType> field : dateTimeTestingFields.entrySet()) {
+            sqle = expectThrows(SQLException.class, () -> method.apply(field.getKey().v1(), c));
+            assertEquals(
+                    format(Locale.ROOT, "Unable to convert value [%.128s] of type [%s] to a Long", 
+                            field.getKey().v2(), field.getValue()), sqle.getMessage());
+        }
+    }
+    
+    private float randomFloatBetween(float start, float end) {
+        float result = 0.0f;
+        while (result < start || result > end || Float.isNaN(result)) {
+            result = start + randomFloat() * (end - start);
+        }
+        
+        return result;
+    }
+
+    private Long getMaxIntPlusOne() {
+        return Long.valueOf(Integer.MAX_VALUE) + 1L;
+    }
+
+    private Double getMaxLongPlusOne() {
+        return Double.valueOf(Long.MAX_VALUE) + 1d;
+    }
+
+    private Connection esJdbc(String timeZoneId) throws SQLException {
+        return randomBoolean() ? useDriverManager(timeZoneId) : useDataSource(timeZoneId);
+    }
+
+    private Connection useDriverManager(String timeZoneId) throws SQLException {
+        String elasticsearchAddress = getProtocol() + "://" + elasticsearchAddress();
+        String address = "jdbc:es://" + elasticsearchAddress;
+        Properties connectionProperties = connectionProperties();
+        connectionProperties.put(JdbcConfiguration.TIME_ZONE, timeZoneId);
+        Connection connection = DriverManager.getConnection(address, connectionProperties);
+        
+        assertNotNull("The timezone should be specified", connectionProperties.getProperty(JdbcConfiguration.TIME_ZONE));
+        return connection;
+    }
+
+    private Connection useDataSource(String timeZoneId) throws SQLException {
+        String elasticsearchAddress = getProtocol() + "://" + elasticsearchAddress();
+        JdbcDataSource dataSource = new JdbcDataSource();
+        String address = "jdbc:es://" + elasticsearchAddress;
+        dataSource.setUrl(address);
+        Properties connectionProperties = connectionProperties();
+        connectionProperties.put(JdbcConfiguration.TIME_ZONE, timeZoneId);
+        dataSource.setProperties(connectionProperties);
+        Connection connection = dataSource.getConnection();
+        
+        assertNotNull("The timezone should be specified", connectionProperties.getProperty(JdbcConfiguration.TIME_ZONE));
+        return connection;
+    }
 }
diff --git a/x-pack/qa/sql/src/main/java/org/elasticsearch/xpack/qa/sql/jdbc/SimpleExampleTestCase.java b/x-pack/qa/sql/src/main/java/org/elasticsearch/xpack/qa/sql/jdbc/SimpleExampleTestCase.java
index 7621743481a..f5d559d9bf0 100644
--- a/x-pack/qa/sql/src/main/java/org/elasticsearch/xpack/qa/sql/jdbc/SimpleExampleTestCase.java
+++ b/x-pack/qa/sql/src/main/java/org/elasticsearch/xpack/qa/sql/jdbc/SimpleExampleTestCase.java
@@ -25,7 +25,8 @@ public class SimpleExampleTestCase extends JdbcIntegrationTestCase {
                 assertEquals("Don Quixote", results.getString(1));
                 assertEquals(1072, results.getInt(2));
                 SQLException e = expectThrows(SQLException.class, () -> results.getInt(1));
-                assertTrue(e.getMessage(), e.getMessage().contains("unable to convert column 1 to an int"));
+                assertTrue(e.getMessage(), 
+                        e.getMessage().contains("Unable to convert value [Don Quixote] of type [VARCHAR] to an Integer"));
                 assertFalse(results.next());
             }
             // end::simple_example