SQL: Fix issue regarding INTERVAL * number

x-pack/plugin/sql/qa/src/main/resources/datetime-interval.csv-spec

@@ -182,6 +182,19 @@ SELECT -2 * INTERVAL '1 23:45' DAY TO MINUTES AS result;
 -3 23:30:00.0  
 ;
 
+intervalAndFieldMultiply
+schema::languages:byte|result:string
+SELECT languages, CAST (languages * INTERVAL '1 10:30' DAY TO MINUTES AS string) AS result FROM test_emp ORDER BY emp_no LIMIT 5;
+
+   languages   |  result
+---------------+---------------------------------------------
+2              |  +2 21:00:00.0
+5              |  +7 04:30:00.0
+4              |  +5 18:00:00.0
+5              |  +7 04:30:00.0
+1              |  +1 10:30:00.0
+;
+
 dateMinusInterval
 SELECT CAST('2018-05-13T12:34:56' AS DATETIME) - INTERVAL '2-8' YEAR TO MONTH AS result;
 

x-pack/plugin/sql/src/main/java/org/elasticsearch/xpack/sql/expression/predicate/operator/arithmetic/BinaryArithmeticProcessor.java

@@ -164,7 +164,7 @@ protected Object doProcess(Object left, Object right) {
             return null;
         }
 
-        if (f == BinaryArithmeticOperation.MUL || f == BinaryArithmeticOperation.DIV || f == BinaryArithmeticOperation.MOD) {
+        if (f == BinaryArithmeticOperation.DIV || f == BinaryArithmeticOperation.MOD) {
             if (!(left instanceof Number)) {
                 throw new SqlIllegalArgumentException("A number is required; received {}", left);
             }
@@ -176,8 +176,8 @@ protected Object doProcess(Object left, Object right) {
             return f.apply(left, right);
         }
 
-        if (f == BinaryArithmeticOperation.ADD || f == BinaryArithmeticOperation.SUB) {
-                return f.apply(left, right);
+        if (f == BinaryArithmeticOperation.ADD || f == BinaryArithmeticOperation.SUB || f == BinaryArithmeticOperation.MUL) {
+            return f.apply(left, right);
         }
 
         // this should not occur

x-pack/plugin/sql/src/main/java/org/elasticsearch/xpack/sql/expression/predicate/operator/arithmetic/DateTimeArithmeticOperation.java

@@ -43,7 +43,7 @@ protected TypeResolution resolveType() {
         // 2. 3. 4. intervals
         if ((DataTypes.isInterval(l) || DataTypes.isInterval(r))) {
             if (DataTypeConversion.commonType(l, r) == null) {
-                return new TypeResolution(format("[{}] has arguments with incompatible types [{}] and [{}]", symbol(), l, r));
+                return new TypeResolution(format(null, "[{}] has arguments with incompatible types [{}] and [{}]", symbol(), l, r));
             } else {
                 return resolveWithIntervals();
             }
@@ -54,6 +54,15 @@ protected TypeResolution resolveType() {
     }
 
     protected TypeResolution resolveWithIntervals() {
+        DataType l = left().dataType();
+        DataType r = right().dataType();
+
+        if (!(r.isDateOrTimeBased() || DataTypes.isInterval(r))) {
+            return new TypeResolution(format(null, "[{}] has arguments with incompatible types [{}] and [{}]", symbol(), l, r));
+        }
+        if (!(l.isDateOrTimeBased() || DataTypes.isInterval(l))) {
+            return new TypeResolution(format(null, "[{}] has arguments with incompatible types [{}] and [{}]", symbol(), l, r));
+        }
         return TypeResolution.TYPE_RESOLVED;
     }
 }

x-pack/plugin/sql/src/main/java/org/elasticsearch/xpack/sql/expression/predicate/operator/arithmetic/Mul.java

@@ -47,7 +47,7 @@ protected TypeResolution resolveType() {
             return TypeResolution.TYPE_RESOLVED;
         }
 
-        return new TypeResolution(format("[{}] has arguments with incompatible types [{}] and [{}]", symbol(), l, r));
+        return new TypeResolution(format(null, "[{}] has arguments with incompatible types [{}] and [{}]", symbol(), l, r));
     }
 
     @Override

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

@@ -121,6 +121,17 @@ public static DataType commonType(DataType left, DataType right) {
                 return right;
             }
         }
+        // Interval * integer is a valid operation
+        if (DataTypes.isInterval(left)) {
+            if (right.isInteger()) {
+                return left;
+            }
+        }
+        if (DataTypes.isInterval(right)) {
+            if (left.isInteger()) {
+                return right;
+            }
+        }
         if (DataTypes.isInterval(left)) {
             // intervals widening
             if (DataTypes.isInterval(right)) {

x-pack/plugin/sql/src/test/java/org/elasticsearch/xpack/sql/analysis/analyzer/VerifierErrorMessagesTests.java

@@ -241,6 +241,21 @@ public void testSubtractFromInterval() {
             error("SELECT INTERVAL 1 MONTH - CAST('12:23:56.789' AS TIME)"));
     }
 
+    public void testAddIntervalAndNumberNotAllowed() {
+        assertEquals("1:8: [+] has arguments with incompatible types [INTERVAL_DAY] and [INTEGER]",
+            error("SELECT INTERVAL 1 DAY + 100"));
+    }
+
+    public void testSubtractIntervalAndNumberNotAllowed() {
+        assertEquals("1:8: [-] has arguments with incompatible types [INTERVAL_MINUTE] and [DOUBLE]",
+            error("SELECT INTERVAL 10 MINUTE - 100.0"));
+    }
+
+    public void testMultiplyIntervalWithDecimalNotAllowed() {
+        assertEquals("1:8: [*] has arguments with incompatible types [INTERVAL_MONTH] and [DOUBLE]",
+            error("SELECT INTERVAL 1 MONTH * 1.234"));
+    }
+
     public void testMultipleColumns() {
         assertEquals("1:43: Unknown column [xxx]\nline 1:8: Unknown column [xxx]",
                 error("SELECT xxx FROM test GROUP BY DAY_oF_YEAR(xxx)"));

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

@@ -628,6 +628,10 @@ public void testCommonType() {
         assertEquals(FLOAT, commonType(FLOAT, INTEGER));
         assertEquals(DOUBLE, commonType(DOUBLE, FLOAT));
 
+        // numeric and intervals
+        assertEquals(INTERVAL_YEAR_TO_MONTH, commonType(INTERVAL_YEAR_TO_MONTH, LONG));
+        assertEquals(INTERVAL_HOUR_TO_MINUTE, commonType(INTEGER, INTERVAL_HOUR_TO_MINUTE));
+
         // dates/datetimes and intervals
         assertEquals(DATETIME, commonType(DATE, DATETIME));
         assertEquals(DATETIME, commonType(DATETIME, DATE));