Grouping on arrays as arrays

core/src/main/java/org/apache/druid/math/expr/ExprEval.java

@@ -405,11 +405,26 @@ public static ExprEval ofType(@Nullable ExpressionType type, @Nullable Object va
         if (value instanceof List) {
           return bestEffortOf(value);
         }
-        return of((String) value);
+        if (value == null) {
+          return of(null);
+        }
+        return of(String.valueOf(value));
       case LONG:
-        return ofLong((Number) value);
+        if (value instanceof Number) {
+          return ofLong((Number) value);
+        }
+        if (value instanceof String) {
+          return ofLong(ExprEval.computeNumber((String) value));
+        }
+        return ofLong(null);
       case DOUBLE:
-        return ofDouble((Number) value);
+        if (value instanceof Number) {
+          return ofDouble((Number) value);
+        }
+        if (value instanceof String) {
+          return ofDouble(ExprEval.computeNumber((String) value));
+        }
+        return ofDouble(null);
       case COMPLEX:
         byte[] bytes = null;
         if (value instanceof String) {

core/src/main/java/org/apache/druid/math/expr/ExpressionProcessing.java

@@ -48,13 +48,13 @@ public class ExpressionProcessing
   @VisibleForTesting
   public static void initializeForTests(@Nullable Boolean allowNestedArrays)
   {
-    INSTANCE = new ExpressionProcessingConfig(allowNestedArrays, null);
+    INSTANCE = new ExpressionProcessingConfig(allowNestedArrays, null, null);
   }
 
   @VisibleForTesting
   public static void initializeForStrictBooleansTests(boolean useStrict)
   {
-    INSTANCE = new ExpressionProcessingConfig(null, useStrict);
+    INSTANCE = new ExpressionProcessingConfig(null, useStrict, null);
   }
 
   /**
@@ -81,4 +81,16 @@ public static boolean useStrictBooleans()
     }
     return INSTANCE.isUseStrictBooleans();
   }
+
+
+  public static boolean processArraysAsMultiValueStrings()
+  {
+    // this should only be null in a unit test context, in production this will be injected by the null handling module
+    if (INSTANCE == null) {
+      throw new IllegalStateException(
+          "ExpressionProcessing module not initialized, call ExpressionProcessing.initializeForTests()"
+      );
+    }
+    return INSTANCE.processArraysAsMultiValueStrings();
+  }
 }

core/src/main/java/org/apache/druid/math/expr/ExpressionProcessingConfig.java

@@ -28,17 +28,24 @@ public class ExpressionProcessingConfig
 {
   public static final String NESTED_ARRAYS_CONFIG_STRING = "druid.expressions.allowNestedArrays";
   public static final String NULL_HANDLING_LEGACY_LOGICAL_OPS_STRING = "druid.expressions.useStrictBooleans";
+  // Coerce arrays to multi value strings
+  public static final String
+      PROCESS_ARRAYS_AS_MULTIVALUE_STRINGS_CONFIG_STRING = "druid.expressions.processArraysAsMultiValueStrings";
 
   @JsonProperty("allowNestedArrays")
   private final boolean allowNestedArrays;
 
   @JsonProperty("useStrictBooleans")
   private final boolean useStrictBooleans;
 
+  @JsonProperty("processArraysAsMultiValueStrings")
+  private final boolean processArraysAsMultiValueStrings;
+
   @JsonCreator
   public ExpressionProcessingConfig(
       @JsonProperty("allowNestedArrays") @Nullable Boolean allowNestedArrays,
-      @JsonProperty("useStrictBooleans") @Nullable Boolean useStrictBooleans
+      @JsonProperty("useStrictBooleans") @Nullable Boolean useStrictBooleans,
+      @JsonProperty("processArraysAsMultiValueStrings") @Nullable Boolean processArraysAsMultiValueStrings
   )
   {
     this.allowNestedArrays = allowNestedArrays == null
@@ -51,6 +58,10 @@ public ExpressionProcessingConfig(
     } else {
       this.useStrictBooleans = useStrictBooleans;
     }
+    this.processArraysAsMultiValueStrings
+        = processArraysAsMultiValueStrings == null
+          ? Boolean.valueOf(System.getProperty(PROCESS_ARRAYS_AS_MULTIVALUE_STRINGS_CONFIG_STRING, "false"))
+          : processArraysAsMultiValueStrings;
   }
 
   public boolean allowNestedArrays()
@@ -62,4 +73,9 @@ public boolean isUseStrictBooleans()
   {
     return useStrictBooleans;
   }
+
+  public boolean processArraysAsMultiValueStrings()
+  {
+    return processArraysAsMultiValueStrings;
+  }
 }

core/src/main/java/org/apache/druid/math/expr/Function.java

@@ -2976,6 +2976,67 @@ public ExpressionType getOutputType(Expr.InputBindingInspector inspector, List<E
     }
   }
 
+  class MVToArrayFunction implements Function
+  {
+    @Override
+    public String name()
+    {
+      return "mv_to_array";
+    }
+
+    @Override
+    public ExprEval apply(List<Expr> args, Expr.ObjectBinding bindings)
+    {
+      return args.get(0).eval(bindings).castTo(ExpressionType.STRING_ARRAY);
+    }
+
+    @Override
+    public void validateArguments(List<Expr> args)
+    {
+      if (args.size() != 1) {
+        throw new IAE("Function[%s] needs exactly 1 argument of type String", name());
+      }
+      IdentifierExpr expr = args.get(0).getIdentifierExprIfIdentifierExpr();
+
+      if (expr == null) {
+        throw new IAE(
+            "Arg %s should be an identifier expression ie refer to columns directaly. Use array() instead",
+            args.get(0).toString()
+        );
+      }
+    }
+
+    @Nullable
+    @Override
+    public ExpressionType getOutputType(Expr.InputBindingInspector inspector, List<Expr> args)
+    {
+      return ExpressionType.STRING_ARRAY;
+    }
+
+    @Override
+    public boolean hasArrayInputs()
+    {
+      return true;
+    }
+
+    @Override
+    public boolean hasArrayOutput()
+    {
+      return true;
+    }
+
+    @Override
+    public Set<Expr> getScalarInputs(List<Expr> args)
+    {
+      return Collections.emptySet();
+    }
+
+    @Override
+    public Set<Expr> getArrayInputs(List<Expr> args)
+    {
+      return ImmutableSet.copyOf(args);
+    }
+  }
   class ArrayConstructorFunction implements Function
   {
     @Override
@@ -2993,6 +3054,7 @@ public ExprEval apply(List<Expr> args, Expr.ObjectBinding bindings)
       Object[] out = new Object[length];
 
       ExpressionType arrayType = null;
+
       for (int i = 0; i < length; i++) {
         ExprEval<?> evaluated = args.get(i).eval(bindings);
         arrayType = setArrayOutput(arrayType, out, i, evaluated);

core/src/test/java/org/apache/druid/math/expr/FunctionTest.java

@@ -934,6 +934,45 @@ public void testComplexDecodeBaseArg0Unknown()
     );
   }
 
+  @Test
+  public void testMVToArrayWithValidInputs()
+  {
+    assertArrayExpr("mv_to_array(x)", new String[]{"foo"});
+    assertArrayExpr("mv_to_array(a)", new String[]{"foo", "bar", "baz", "foobar"});
+  }
+
+  @Test
+  public void testMVToArrayWithConstantLiteral()
+  {
+    expectedException.expect(IAE.class);
+    expectedException.expectMessage("should be an identifier expression");
+    assertArrayExpr("mv_to_array('1')", null);
+  }
+
+  @Test
+  public void testMVToArrayWithFunction()
+  {
+    expectedException.expect(IAE.class);
+    expectedException.expectMessage("should be an identifier expression");
+    assertArrayExpr("mv_to_array(repeat('hello', 2))", null);
+  }
+
+  @Test
+  public void testMVToArrayWithMoreArgs()
+  {
+    expectedException.expect(IAE.class);
+    expectedException.expectMessage("needs exactly 1 argument of type String");
+    assertArrayExpr("mv_to_array(x,y)", null);
+  }
+
+  @Test
+  public void testMVToArrayWithNoArgs()
+  {
+    expectedException.expect(IAE.class);
+    expectedException.expectMessage("needs exactly 1 argument of type String");
+    assertArrayExpr("mv_to_array()", null);
+  }
+
   private void assertExpr(final String expression, @Nullable final Object expectedResult)
   {
     final Expr expr = Parser.parse(expression, ExprMacroTable.nil());

processing/src/main/java/org/apache/druid/query/aggregation/ExpressionLambdaAggregatorFactory.java

@@ -486,22 +486,18 @@ private class FactorizePlan
 
     FactorizePlan(ColumnSelectorFactory metricFactory)
     {
-      final List<String> columns;
-
       if (fields != null) {
         // if fields are set, we are accumulating from raw inputs, use fold expression
         plan = ExpressionPlanner.plan(inspectorWithAccumulator(metricFactory), foldExpression.get());
         seed = initialValue.get();
-        columns = plan.getAnalysis().getRequiredBindingsList();
       } else {
         // else we are merging intermediary results, use combine expression
         plan = ExpressionPlanner.plan(inspectorWithAccumulator(metricFactory), combineExpression.get());
         seed = initialCombineValue.get();
-        columns = plan.getAnalysis().getRequiredBindingsList();
       }
 
       bindings = new ExpressionLambdaAggregatorInputBindings(
-          ExpressionSelectors.createBindings(metricFactory, columns),
+          ExpressionSelectors.createBindings(metricFactory, plan),
           accumulatorId,
           seed
       );

processing/src/main/java/org/apache/druid/query/groupby/GroupByQuery.java

@@ -67,6 +67,8 @@
 import org.apache.druid.segment.column.ColumnHolder;
 import org.apache.druid.segment.column.ColumnType;
 import org.apache.druid.segment.column.RowSignature;
+import org.apache.druid.segment.data.ComparableList;
+import org.apache.druid.segment.data.ComparableStringArray;
 import org.joda.time.DateTime;
 import org.joda.time.Interval;
 
@@ -775,6 +777,15 @@ private static int compareDimsForLimitPushDown(
         } else {
           dimCompare = comparator.compare(String.valueOf(lhsObj), String.valueOf(rhsObj));
         }
+      } else if (dimensionType.equals(ColumnType.STRING_ARRAY)) {
+        final ComparableStringArray lhsArr = DimensionHandlerUtils.convertToComparableStringArray(lhsObj);
+        final ComparableStringArray rhsArr = DimensionHandlerUtils.convertToComparableStringArray(rhsObj);
+        dimCompare = Comparators.<Comparable>naturalNullsFirst().compare(lhsArr, rhsArr);
+      } else if (dimensionType.equals(ColumnType.LONG_ARRAY)
+                 || dimensionType.equals(ColumnType.DOUBLE_ARRAY)) {
+        final ComparableList lhsArr = DimensionHandlerUtils.convertToList(lhsObj);
+        final ComparableList rhsArr = DimensionHandlerUtils.convertToList(rhsObj);
+        dimCompare = Comparators.<Comparable>naturalNullsFirst().compare(lhsArr, rhsArr);
       } else {
         dimCompare = comparator.compare((String) lhsObj, (String) rhsObj);
       }

processing/src/main/java/org/apache/druid/query/groupby/epinephelinae/GroupByQueryEngineV2.java

@@ -42,6 +42,9 @@
 import org.apache.druid.query.groupby.GroupByQuery;
 import org.apache.druid.query.groupby.GroupByQueryConfig;
 import org.apache.druid.query.groupby.ResultRow;
+import org.apache.druid.query.groupby.epinephelinae.column.ArrayDoubleGroupByColumnSelectorStrategy;
+import org.apache.druid.query.groupby.epinephelinae.column.ArrayLongGroupByColumnSelectorStrategy;
+import org.apache.druid.query.groupby.epinephelinae.column.ArrayStringGroupByColumnSelectorStrategy;
 import org.apache.druid.query.groupby.epinephelinae.column.DictionaryBuildingStringGroupByColumnSelectorStrategy;
 import org.apache.druid.query.groupby.epinephelinae.column.DoubleGroupByColumnSelectorStrategy;
 import org.apache.druid.query.groupby.epinephelinae.column.FloatGroupByColumnSelectorStrategy;
@@ -233,7 +236,7 @@ public GroupByEngineIterator make()
                       processingBuffer,
                       fudgeTimestamp,
                       dims,
-                      isAllSingleValueDims(columnSelectorFactory, query.getDimensions()),
+                      hasNoExplodingDimensions(columnSelectorFactory, query.getDimensions()),
                       cardinalityForArrayAggregation
                   );
                 } else {
@@ -244,7 +247,7 @@ public GroupByEngineIterator make()
                       processingBuffer,
                       fudgeTimestamp,
                       dims,
-                      isAllSingleValueDims(columnSelectorFactory, query.getDimensions())
+                      hasNoExplodingDimensions(columnSelectorFactory, query.getDimensions())
                   );
                 }
               }
@@ -290,6 +293,11 @@ public static int getCardinalityForArrayAggregation(
       if (query.getVirtualColumns().exists(Iterables.getOnlyElement(dimensions).getDimension())) {
         return -1;
       }
+      // We cannot support array-based aggregation on array based grouping as we we donot have all the indexes up front
+      // to allocate appropriate values
+      if (dimensions.get(0).getOutputType().equals(ColumnType.STRING_ARRAY)) {
+        return -1;
+      }
 
       final String columnName = Iterables.getOnlyElement(dimensions).getDimension();
       columnCapabilities = storageAdapter.getColumnCapabilities(columnName);
@@ -319,11 +327,12 @@ public static int getCardinalityForArrayAggregation(
   }
 
   /**
-   * Checks whether all "dimensions" are either single-valued, or if allowed, nonexistent. Since non-existent column
-   * selectors will show up as full of nulls they are effectively single valued, however they can also be null during
-   * broker merge, for example with an 'inline' datasource subquery.
+   * Checks whether all "dimensions" are either single-valued,
+   * or STRING_ARRAY, in case we don't want to explode the underline multi value column,
+   * or if allowed, nonexistent. Since non-existent columnselectors will show up as full of nulls they are effectively
+   * single valued, however they can also be null during broker merge, for example with an 'inline' datasource subquery.
    */
-  public static boolean isAllSingleValueDims(
+  public static boolean hasNoExplodingDimensions(
       final ColumnInspector inspector,
       final List<DimensionSpec> dimensions
   )
@@ -340,7 +349,8 @@ public static boolean isAllSingleValueDims(
 
               // Now check column capabilities, which must be present and explicitly not multi-valued
               final ColumnCapabilities columnCapabilities = inspector.getColumnCapabilities(dimension.getDimension());
-              return columnCapabilities != null && columnCapabilities.hasMultipleValues().isFalse();
+              return dimension.getOutputType().equals(ColumnType.STRING_ARRAY)
+                     || (columnCapabilities != null && columnCapabilities.hasMultipleValues().isFalse());
             });
   }
 
@@ -403,6 +413,20 @@ public GroupByColumnSelectorStrategy makeColumnSelectorStrategy(
           return makeNullableNumericStrategy(new FloatGroupByColumnSelectorStrategy());
         case DOUBLE:
           return makeNullableNumericStrategy(new DoubleGroupByColumnSelectorStrategy());
+        case ARRAY:
+          switch (capabilities.getElementType().getType()) {
+            case LONG:
+              return new ArrayLongGroupByColumnSelectorStrategy();
+            case STRING:
+              return new ArrayStringGroupByColumnSelectorStrategy();
+            case DOUBLE:
+              return new ArrayDoubleGroupByColumnSelectorStrategy();
+            case FLOAT:
+              // Array<Float> not supported in expressions, ingestion
+            default:
+              throw new IAE("Cannot create query type helper from invalid type [%s]", capabilities.asTypeString());
+
+          }
         default:
           throw new IAE("Cannot create query type helper from invalid type [%s]", capabilities.asTypeString());
       }