Fix cast statement in aggregate function return 0

integ-test/src/test/java/com/amazon/opendistroforelasticsearch/sql/legacy/AggregationExpressionIT.java

@@ -252,6 +252,16 @@ public void groupByDateWithAliasShouldPass() {
         rows("2018-06-23 00:00:00.000", 1));
   }
 
+  @Test
+  public void aggregateCastStatementShouldNotReturnZero() {
+    JSONObject response = executeJdbcRequest(String.format(
+        "SELECT SUM(CAST(male AS INT)) AS male_sum FROM %s",
+        Index.BANK.getName()));
+
+    verifySchema(response, schema("male_sum", "male_sum", "double"));
+    verifyDataRows(response, rows(4));
+  }
+
   private JSONObject executeJdbcRequest(String query) {
     return new JSONObject(executeQuery(query, "jdbc"));
   }

legacy/src/main/java/com/amazon/opendistroforelasticsearch/sql/legacy/parser/FieldMaker.java

@@ -347,6 +347,14 @@ public MethodField makeMethodField(String name, List<SQLExpr> arguments, SQLAggr
                 methodParameters.add(new KVValue(((SQLCastExpr) object).getExpr().toString()));
                 String castType = ((SQLCastExpr) object).getDataType().getName();
                 String scriptCode = sqlFunctions.getCastScriptStatement(castName, castType, methodParameters);
+
+                // Parameter "first" indicates if return statement is required. Take CAST statement nested in
+                // aggregate function SUM(CAST...) for example, return statement is required in this case.
+                // Otherwise DSL with metric aggregation always returns 0 as result. And this works also because
+                // the caller makeFieldImpl(SQLExpr("SUM...")) does pass first=true to here.
+                if (first) {
+                    scriptCode += "; return " + castName;
+                }
                 methodParameters.add(new KVValue(scriptCode));
                 paramers.add(new KVValue("script", new SQLCharExpr(scriptCode)));
             } else if (object instanceof SQLAggregateExpr) {