Window Functions : Reject MVDs during window processing (#17002)

This commit aims to reject MVDs in window processing as we do not support them.
Earlier to this commit, query running a window aggregate partitioned by an MVD column would fail with ClassCastException

processing/src/main/java/org/apache/druid/query/rowsandcols/ArrayListRowsAndColumns.java

@@ -25,6 +25,7 @@ import it.unimi.dsi.fastutil.ints.IntComparator;
 import it.unimi.dsi.fastutil.ints.IntList;
 import org.apache.druid.common.semantic.SemanticCreator;
 import org.apache.druid.common.semantic.SemanticUtils;
+import org.apache.druid.error.InvalidInput;
 import org.apache.druid.java.util.common.ISE;
 import org.apache.druid.java.util.common.guava.Sequences;
 import org.apache.druid.query.operator.ColumnWithDirection;
@@ -153,11 +154,28 @@ public class ArrayListRowsAndColumns<RowType> implements AppendableRowsAndColumn
       return new LimitedColumn(retVal, startOffset, endOffset);
     }
 
-    final Function<RowType, Object> adapterForValue = rowAdapter.columnFunction(name);
     final Optional<ColumnType> maybeColumnType = rowSignature.getColumnType(name);
     final ColumnType columnType = maybeColumnType.orElse(ColumnType.UNKNOWN_COMPLEX);
     final Comparator<Object> comparator = Comparator.nullsFirst(columnType.getStrategy());
 
+    final Function<RowType, Object> adapterForValue;
+    if (columnType.equals(ColumnType.STRING)) {
+      // special handling to reject MVDs
+      adapterForValue = f -> {
+        Object value = rowAdapter.columnFunction(name).apply(f);
+        if (value instanceof List) {
+          throw InvalidInput.exception(
+              "Encountered a multi value column [%s]. Window processing does not support MVDs. "
+              + "Consider using UNNEST or MV_TO_ARRAY.",
+              name
+          );
+        }
+        return value;
+      };
+    } else {
+      adapterForValue = rowAdapter.columnFunction(name);
+    }
+
     return new Column()
     {
       @Nonnull

sql/src/test/java/org/apache/druid/sql/calcite/CalciteWindowQueryTest.java

@@ -24,6 +24,7 @@ import com.fasterxml.jackson.databind.ObjectMapper;
 import com.fasterxml.jackson.dataformat.yaml.YAMLFactory;
 import com.google.common.collect.ImmutableList;
 import com.google.common.collect.ImmutableMap;
+import org.apache.druid.error.DruidException;
 import org.apache.druid.jackson.DefaultObjectMapper;
 import org.apache.druid.java.util.common.ISE;
 import org.apache.druid.java.util.common.RE;
@@ -249,6 +250,28 @@ public class CalciteWindowQueryTest extends BaseCalciteQueryTest
         .run();
   }
 
+  @Test
+  public void testFailure_partitionByMVD()
+  {
+    final DruidException e = Assert.assertThrows(
+        DruidException.class,
+        () -> testBuilder()
+            .sql("select cityName, countryName, array_to_mv(array[1,length(cityName)]),\n"
+                 + "row_number() over (partition by  array_to_mv(array[1,length(cityName)]) order by countryName, cityName)\n"
+                 + "from wikipedia\n"
+                 + "where countryName in ('Austria', 'Republic of Korea') and cityName is not null\n"
+                 + "order by 1, 2, 3")
+            .queryContext(DEFAULT_QUERY_CONTEXT)
+            .run()
+    );
+
+    assertEquals(
+        "Encountered a multi value column [v0]. Window processing does not support MVDs. "
+        + "Consider using UNNEST or MV_TO_ARRAY.",
+        e.getMessage()
+    );
+  }
+
   private WindowOperatorQuery getWindowOperatorQuery(List<Query<?>> queries)
   {
     assertEquals(1, queries.size());