SQL: Fix issue regarding INTERVAL * number

Interval * integer number is a valid operation which previously was
only supported for foldables (literals) and not when a field was
involved. That was because:

1. There was no common type returned for that combination
2. The `BinaryArithmeticOperation` was permitting the multiplication
(called by fold()) but the BinaryArithmeticProcessor didn't allow it

Moreover the error message for invalid arithmetic operations was wrong
because of the issue with the overloading methods of
`LoggerMessageFormat.format`.

Fixes: #41239

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));