FUNNEL_COUNT Aggregation Function

pinot-common/src/main/java/org/apache/pinot/common/utils/DataSchema.java

@@ -24,6 +24,7 @@
 import com.fasterxml.jackson.annotation.JsonPropertyOrder;
 import com.google.common.collect.Ordering;
 import it.unimi.dsi.fastutil.doubles.DoubleArrayList;
+import it.unimi.dsi.fastutil.longs.LongArrayList;
 import java.io.ByteArrayOutputStream;
 import java.io.DataOutputStream;
 import java.io.IOException;
@@ -469,6 +470,8 @@ private static double[] toDoubleArray(Object value) {
     private static long[] toLongArray(Object value) {
       if (value instanceof long[]) {
         return (long[]) value;
+      } else if (value instanceof LongArrayList) {
+        return ((LongArrayList) value).elements();
       } else {
         int[] intValues = (int[]) value;
         int length = intValues.length;

pinot-core/src/main/java/org/apache/pinot/core/operator/blocks/results/AggregationResultsBlock.java

@@ -19,6 +19,7 @@
 package org.apache.pinot.core.operator.blocks.results;
 
 import it.unimi.dsi.fastutil.doubles.DoubleArrayList;
+import it.unimi.dsi.fastutil.longs.LongArrayList;
 import java.io.IOException;
 import java.math.BigDecimal;
 import java.util.Collections;
@@ -182,6 +183,9 @@ private void setFinalResult(DataTableBuilder dataTableBuilder, ColumnDataType[]
       case DOUBLE_ARRAY:
         dataTableBuilder.setColumn(index, ((DoubleArrayList) result).elements());
         break;
+      case LONG_ARRAY:
+        dataTableBuilder.setColumn(index, ((LongArrayList) result).elements());
+        break;
       default:
         throw new IllegalStateException("Illegal column data type in final result: " + columnDataType);
     }

pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/AggregationFunctionFactory.java

@@ -358,6 +358,9 @@ public static AggregationFunction getAggregationFunction(FunctionContext functio
           case ARGMIN:
             throw new IllegalArgumentException(
                 "Aggregation function: " + function + " is only supported in selection without alias.");
+          case FUNNELCOUNT:
+            return new FunnelCountAggregationFunction(arguments);
+
           default:
             throw new IllegalArgumentException();
         }