SqlResults: Coerce arrays to lists for VARCHAR. (apache#14260)

* SqlResults: Coerce arrays to lists for VARCHAR.

Useful for STRING_TO_MV, which returns VARCHAR at the SQL layer and an
ExprEval with String[] at the native layer.

* Fix style.

* Improve test coverage.

* Remove unnecessary throws.

sql/src/main/java/org/apache/druid/sql/calcite/run/SqlResults.java

@@ -39,7 +39,7 @@
 import org.joda.time.DateTime;
 import org.joda.time.DateTimeZone;
 
-import java.io.IOException;
+import javax.annotation.Nullable;
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Collection;
@@ -70,21 +70,23 @@ public static Object coerce(
         coercedValue = String.valueOf(value);
       } else if (value instanceof Boolean) {
         coercedValue = String.valueOf(value);
-      } else if (value instanceof Collection) {
-        // Iterate through the collection, coercing each value. Useful for handling selects of multi-value dimensions.
-        final List<String> valueStrings =
-            ((Collection<?>) value).stream()
-                                   .map(v -> (String) coerce(jsonMapper, context, v, sqlTypeName))
-                                   .collect(Collectors.toList());
+      } else {
+        final Object maybeList = maybeCoerceArrayToList(value, false);
 
-        try {
-          coercedValue = jsonMapper.writeValueAsString(valueStrings);
-        }
-        catch (IOException e) {
-          throw new RuntimeException(e);
+        // Check if "maybeList" was originally a Collection of some kind, or was able to be coerced to one.
+        // Then Iterate through the collection, coercing each value. Useful for handling multi-value dimensions.
+        if (maybeList instanceof Collection) {
+          final List<String> valueStrings =
+              ((Collection<?>) maybeList)
+                  .stream()
+                  .map(v -> (String) coerce(jsonMapper, context, v, sqlTypeName))
+                  .collect(Collectors.toList());
+
+          // Must stringify since the caller is expecting CHAR_TYPES.
+          coercedValue = coerceUsingObjectMapper(jsonMapper, valueStrings, sqlTypeName);
+        } else {
+          throw cannotCoerce(value, sqlTypeName);
         }
-      } else {
-        throw new ISE("Cannot coerce [%s] to %s", value.getClass().getName(), sqlTypeName);
       }
     } else if (value == null) {
       coercedValue = null;
@@ -93,51 +95,48 @@ public static Object coerce(
     } else if (sqlTypeName == SqlTypeName.TIMESTAMP) {
       return Calcites.jodaToCalciteTimestamp(coerceDateTime(value, sqlTypeName), context.getTimeZone());
     } else if (sqlTypeName == SqlTypeName.BOOLEAN) {
-      if (value instanceof String) {
+      if (value instanceof Boolean) {
+        coercedValue = value;
+      } else if (value instanceof String) {
         coercedValue = Evals.asBoolean(((String) value));
       } else if (value instanceof Number) {
         coercedValue = Evals.asBoolean(((Number) value).longValue());
       } else {
-        throw new ISE("Cannot coerce [%s] to %s", value.getClass().getName(), sqlTypeName);
+        throw cannotCoerce(value, sqlTypeName);
       }
     } else if (sqlTypeName == SqlTypeName.INTEGER) {
       if (value instanceof String) {
         coercedValue = Ints.tryParse((String) value);
       } else if (value instanceof Number) {
         coercedValue = ((Number) value).intValue();
       } else {
-        throw new ISE("Cannot coerce [%s] to %s", value.getClass().getName(), sqlTypeName);
+        throw cannotCoerce(value, sqlTypeName);
       }
     } else if (sqlTypeName == SqlTypeName.BIGINT) {
       try {
         coercedValue = DimensionHandlerUtils.convertObjectToLong(value);
       }
       catch (Exception e) {
-        throw new ISE("Cannot coerce [%s] to %s", value.getClass().getName(), sqlTypeName);
+        throw cannotCoerce(value, sqlTypeName);
       }
     } else if (sqlTypeName == SqlTypeName.FLOAT) {
       try {
         coercedValue = DimensionHandlerUtils.convertObjectToFloat(value);
       }
       catch (Exception e) {
-        throw new ISE("Cannot coerce [%s] to %s", value.getClass().getName(), sqlTypeName);
+        throw cannotCoerce(value, sqlTypeName);
       }
     } else if (SqlTypeName.FRACTIONAL_TYPES.contains(sqlTypeName)) {
       try {
         coercedValue = DimensionHandlerUtils.convertObjectToDouble(value);
       }
       catch (Exception e) {
-        throw new ISE("Cannot coerce [%s] to %s", value.getClass().getName(), sqlTypeName);
+        throw cannotCoerce(value, sqlTypeName);
       }
     } else if (sqlTypeName == SqlTypeName.OTHER) {
       // Complex type, try to serialize if we should, else print class name
       if (context.isSerializeComplexValues()) {
-        try {
-          coercedValue = jsonMapper.writeValueAsString(value);
-        }
-        catch (JsonProcessingException jex) {
-          throw new ISE(jex, "Cannot coerce [%s] to %s", value.getClass().getName(), sqlTypeName);
-        }
+        coercedValue = coerceUsingObjectMapper(jsonMapper, value, sqlTypeName);
       } else {
         coercedValue = value.getClass().getName();
       }
@@ -148,31 +147,30 @@ public static Object coerce(
         } else if (value instanceof NlsString) {
           coercedValue = ((NlsString) value).getValue();
         } else {
-          try {
-            coercedValue = jsonMapper.writeValueAsString(value);
-          }
-          catch (IOException e) {
-            throw new RuntimeException(e);
-          }
+          coercedValue = coerceUsingObjectMapper(jsonMapper, value, sqlTypeName);
         }
       } else {
         // the protobuf jdbc handler prefers lists (it actually can't handle java arrays as sql arrays, only java lists)
         // the json handler could handle this just fine, but it handles lists as sql arrays as well so just convert
         // here if needed
         coercedValue = maybeCoerceArrayToList(value, true);
         if (coercedValue == null) {
-          throw new ISE("Cannot coerce [%s] to %s", value.getClass().getName(), sqlTypeName);
+          throw cannotCoerce(value, sqlTypeName);
         }
       }
     } else {
-      throw new ISE("Cannot coerce [%s] to %s", value.getClass().getName(), sqlTypeName);
+      throw cannotCoerce(value, sqlTypeName);
     }
 
     return coercedValue;
   }
 
-
+  /**
+   * Attempt to coerce a value to {@link List}. If it cannot be coerced, either return the original value (if mustCoerce
+   * is false) or return null (if mustCoerce is true).
+   */
   @VisibleForTesting
+  @Nullable
   static Object maybeCoerceArrayToList(Object value, boolean mustCoerce)
   {
     if (value instanceof List) {
@@ -222,11 +220,39 @@ private static DateTime coerceDateTime(Object value, SqlTypeName sqlType)
     } else if (value instanceof DateTime) {
       dateTime = (DateTime) value;
     } else {
-      throw new ISE("Cannot coerce[%s] to %s", value.getClass().getName(), sqlType);
+      throw cannotCoerce(value, sqlType);
     }
     return dateTime;
   }
 
+  private static String coerceUsingObjectMapper(
+      final ObjectMapper jsonMapper,
+      final Object value,
+      final SqlTypeName sqlTypeName
+  )
+  {
+    try {
+      return jsonMapper.writeValueAsString(value);
+    }
+    catch (JsonProcessingException e) {
+      throw cannotCoerce(e, value, sqlTypeName);
+    }
+  }
+
+  private static IllegalStateException cannotCoerce(
+      final Throwable t,
+      final Object value,
+      final SqlTypeName sqlTypeName
+  )
+  {
+    return new ISE(t, "Cannot coerce [%s] to [%s]", value == null ? "null" : value.getClass().getName(), sqlTypeName);
+  }
+
+  private static IllegalStateException cannotCoerce(final Object value, final SqlTypeName sqlTypeName)
+  {
+    return cannotCoerce(null, value, sqlTypeName);
+  }
+
   /**
    * Context for {@link #coerce(ObjectMapper, Context, Object, SqlTypeName)}
    */