Support Cast() function

src/main/antlr/OpenDistroSqlLexer.g4

@@ -49,24 +49,30 @@ ASC:                                'ASC';
 BETWEEN:                            'BETWEEN';
 BY:                                 'BY';
 CASE:                               'CASE';
+CAST:                               'CAST';
 CROSS:                              'CROSS';
+DATETIME:                           'DATETIME';
 DELETE:                             'DELETE';
 DESC:                               'DESC';
 DESCRIBE:                           'DESCRIBE';
 DISTINCT:                           'DISTINCT';
+DOUBLE:                             'DOUBLE';
 ELSE:                               'ELSE';
 EXISTS:                             'EXISTS';
 FALSE:                              'FALSE';
+FLOAT:                              'FLOAT';
 FROM:                               'FROM';
 GROUP:                              'GROUP';
 HAVING:                             'HAVING';
 IN:                                 'IN';
 INNER:                              'INNER';
+INT:                                'INT';
 IS:                                 'IS';
 JOIN:                               'JOIN';
 LEFT:                               'LEFT';
 LIKE:                               'LIKE';
 LIMIT:                              'LIMIT';
+LONG:                               'LONG';
 MATCH:                              'MATCH';
 NATURAL:                            'NATURAL';
 NOT:                                'NOT';
@@ -79,6 +85,7 @@ REGEXP:                             'REGEXP';
 RIGHT:                              'RIGHT';
 SELECT:                             'SELECT';
 SHOW:                               'SHOW';
+STRING:                             'STRING';
 THEN:                               'THEN';
 TRUE:                               'TRUE';
 UNION:                              'UNION';

src/main/antlr/OpenDistroSqlParser.g4

@@ -241,7 +241,6 @@ dottedId
     | '.' uid
     ;
 
-
 //    Literals
 
 decimalLiteral
@@ -297,7 +296,8 @@ functionCall
     ;
 
 specificFunction
-    : CASE expression caseFuncAlternative+
+    : CAST '(' expression AS convertedDataType ')'                  #dataTypeFunctionCall
+    | CASE expression caseFuncAlternative+
       (ELSE elseArg=functionArg)? END                               #caseFunctionCall
     | CASE caseFuncAlternative+
       (ELSE elseArg=functionArg)? END                               #caseFunctionCall
@@ -308,6 +308,15 @@ caseFuncAlternative
       THEN consequent=functionArg
     ;
 
+convertedDataType
+    : typeName=DATETIME
+    | typeName=INT
+    | typeName=DOUBLE
+    | typeName=LONG
+    | typeName=FLOAT
+    | typeName=STRING
+    ;
+
 aggregateWindowedFunction
     : (AVG | MAX | MIN | SUM)
       '(' aggregator=(ALL | DISTINCT)? functionArg ')'

src/main/java/com/amazon/opendistroforelasticsearch/sql/executor/format/SelectResultSet.java

@@ -15,6 +15,7 @@
 
 package com.amazon.opendistroforelasticsearch.sql.executor.format;
 
+import com.alibaba.druid.sql.ast.expr.SQLCastExpr;
 import com.amazon.opendistroforelasticsearch.sql.domain.Field;
 import com.amazon.opendistroforelasticsearch.sql.domain.JoinSelect;
 import com.amazon.opendistroforelasticsearch.sql.domain.MethodField;
@@ -315,7 +316,11 @@ private Schema.Type fetchMethodReturnType(Field field) {
                 // TODO: return type information is disconnected from the function definitions in SQLFunctions.
                 // Refactor SQLFunctions to have functions self-explanatory (types, scripts) and pluggable
                 // (similar to Strategy pattern)
-
+                MethodField methodField = (MethodField) field;
+                if (field.getExpression() instanceof SQLCastExpr) {
+                    return SQLFunctions.getCastFunctionReturnType(
+                            ((SQLCastExpr) field.getExpression()).getDataType().getName());
+                }
                 return SQLFunctions.getScriptFunctionReturnType(
                         ((ScriptMethodField) field).getFunctionName());
             }

src/main/java/com/amazon/opendistroforelasticsearch/sql/query/DefaultQueryAction.java

@@ -18,6 +18,7 @@
 import com.alibaba.druid.sql.ast.SQLExpr;
 import com.alibaba.druid.sql.ast.expr.SQLBinaryOpExpr;
 import com.alibaba.druid.sql.ast.expr.SQLBinaryOperator;
+import com.alibaba.druid.sql.ast.expr.SQLCastExpr;
 import com.amazon.opendistroforelasticsearch.sql.domain.Field;
 import com.amazon.opendistroforelasticsearch.sql.domain.KVValue;
 import com.amazon.opendistroforelasticsearch.sql.domain.MethodField;
@@ -95,7 +96,6 @@ public SqlElasticSearchRequestBuilder explain() throws SqlParseException {
         setIndicesAndTypes();
 
         setFields(select.getFields());
-
         setWhere(select.getWhere());
         setSorts(select.getOrderBys());
         setLimit(select.getOffset(), select.getRowCount());
@@ -152,6 +152,9 @@ public void setFields(List<Field> fields) throws SqlParseException {
                     MethodField method = (MethodField) field;
                     if (method.getName().toLowerCase().equals("script")) {
                         handleScriptField(method);
+                        if (method.getExpression() instanceof SQLCastExpr) {
+                            includeFields.add(method.getParams().get(0).toString());
+                        }
                     } else if (method.getName().equalsIgnoreCase("include")) {
                         for (KVValue kvValue : method.getParams()) {
                             includeFields.add(kvValue.value.toString());

src/main/java/com/amazon/opendistroforelasticsearch/sql/utils/SQLFunctions.java

@@ -614,4 +614,24 @@ public static Schema.Type getScriptFunctionReturnType(String functionName) {
                         "The following method is not supported in Schema: %s",
                         functionName));
     }
+
+    public static Schema.Type getCastFunctionReturnType(String castType) {
+        if (castType.toUpperCase().equals("FLOAT")) {
+            return Schema.Type.FLOAT;
+        } else if (castType.toUpperCase().equals("DOUBLE")) {
+            return Schema.Type.DOUBLE;
+        } else if (castType.toUpperCase().equals("INT")) {
+            return Schema.Type.INTEGER;
+        } else if (castType.toUpperCase().equals("STRING")) {
+            return Schema.Type.TEXT;
+        } else if (castType.toUpperCase().equals("DATETIME")) {
+            return Schema.Type.DATE;
+        } else if (castType.toUpperCase().equals("LONG")) {
+            return Schema.Type.LONG;
+        }
+
+        throw new UnsupportedOperationException(
+                StringUtils.format("The following type is not supported by cast(): %s", castType)
+        );
+    }
 }

src/test/java/com/amazon/opendistroforelasticsearch/sql/esintgtest/SQLFunctionsIT.java

@@ -50,6 +50,7 @@
 import static org.hamcrest.Matchers.hasEntry;
 import static org.hamcrest.Matchers.hasItems;
 import static org.hamcrest.Matchers.hasValue;
+import static org.hamcrest.Matchers.instanceOf;
 import static org.hamcrest.Matchers.is;
 import static org.hamcrest.Matchers.isEmptyOrNullString;
 import static org.hamcrest.Matchers.not;
@@ -159,6 +160,74 @@ public void caseChangeWithAggregationTest() throws IOException {
                 hitAny("/aggregations/UPPER_1/buckets", kvString("/key", equalTo("AMBER"))));
     }
 
+    @Test
+    public void castIntFieldToDoubleWithoutAliasTest() throws IOException {
+        String query = "SELECT CAST(age AS DOUBLE) FROM " + TestsConstants.TEST_INDEX_ACCOUNT + " /account limit 1";
+
+        final SearchHit[] hits = query(query).getHits();
+        assertTrue(hits[0].getFields().containsKey("cast_age"));
+        assertTrue(hits[0].getFields().get("cast_age").getValue() instanceof Double);
+    }
+
+    @Test
+    public void castIntFieldToDoubleWithAliasTest() throws IOException {
+        String query = "SELECT CAST(age AS DOUBLE) AS test_alias " +
+                       "FROM " + TestsConstants.TEST_INDEX_ACCOUNT + " /account limit 1";
+
+        final SearchHit[] hits = query(query).getHits();
+        assertTrue(hits[0].getFields().containsKey("test_alias"));
+        assertTrue(hits[0].getFields().get("test_alias").getValue() instanceof Double);
+    }
+
+    @Test
+    public void castIntFieldToStringWithoutAliasTest() throws IOException {
+        String query = "SELECT CAST(balance AS STRING) FROM " + TestsConstants.TEST_INDEX_ACCOUNT + " /account limit 1";
+
+        final SearchHit[] hits = query(query).getHits();
+        assertTrue(hits[0].getFields().containsKey("cast_balance"));
+        assertTrue(hits[0].getFields().get("cast_balance").getValue() instanceof String);
+    }
+
+    @Test
+    public void castIntFieldToStringWithAliasTest() throws IOException {
+        String query = "SELECT CAST(balance AS STRING) AS cast_string_alias " +
+                       "FROM " + TestsConstants.TEST_INDEX_ACCOUNT + " /account limit 1";
+
+        final SearchHit[] hits = query(query).getHits();
+        assertTrue(hits[0].getFields().containsKey("cast_string_alias"));
+        assertTrue(hits[0].getFields().get("cast_string_alias").getValue() instanceof String);
+    }
+
+    @Ignore
+    @Test
+    public void castIntFieldToFloatWithoutAliasTest() throws IOException {
+        String query = "SELECT CAST(balance AS FLOAT) FROM " + TestsConstants.TEST_INDEX_ACCOUNT + " /account limit 1";
+
+        final SearchHit[] hits = query(query).getHits();
+        System.out.println(hits[0].getFields().toString());
+        assertTrue(hits[0].getFields().containsKey("cast_balance"));
+        assertTrue(hits[0].getFields().get("cast_balance").getValue() instanceof Double);
+    }
+
+    @Test
+    public void castIntFieldToFloatWithoutAliasJdbcFormatTest() {
+        JSONObject response = executeJdbcRequest(
+                "SELECT CAST(balance AS FLOAT) FROM " + TestsConstants.TEST_INDEX_ACCOUNT + " limit 1");
+
+        String float_type_cast = "{\"name\":\"cast_balance\",\"type\":\"float\"}";
+        assertEquals(response.getJSONArray("schema").get(0).toString(), float_type_cast);
+    }
+
+    @Test
+    public void castIntFieldToFloatWithAliasJdbcFormatTest() {
+        JSONObject response = executeJdbcRequest(
+                "SELECT CAST(balance AS FLOAT) AS jdbc_float_alias " +
+                        "FROM " + TestsConstants.TEST_INDEX_ACCOUNT + " limit 1");
+
+        String float_type_cast = "{\"name\":\"jdbc_float_alias\",\"type\":\"float\"}";
+        assertEquals(response.getJSONArray("schema").get(0).toString(), float_type_cast);
+    }
+
     @Test
     public void concat_ws_field_and_string() throws Exception {
         //here is a bug,csv field with spa
@@ -302,4 +371,8 @@ private SearchHits query(String query) throws IOException {
                 rsp);
         return SearchResponse.fromXContent(parser).getHits();
     }
+
+    private JSONObject executeJdbcRequest(String query) {
+        return new JSONObject(executeQuery(query, "jdbc"));
+    }
 }