[fix](planner) fix hll_union plan: Invalid Aggregate Operator: hll_union

fe/fe-core/src/main/java/org/apache/doris/analysis/AnalyticExpr.java

@@ -24,6 +24,7 @@
 import org.apache.doris.analysis.AnalyticWindow.BoundaryType;
 import org.apache.doris.catalog.AggregateFunction;
 import org.apache.doris.catalog.Function;
+import org.apache.doris.catalog.FunctionSet;
 import org.apache.doris.catalog.PrimitiveType;
 import org.apache.doris.catalog.Type;
 import org.apache.doris.common.AnalysisException;
@@ -96,9 +97,6 @@ public class AnalyticExpr extends Expr {
     // additional null handling in the backend.
     public static String FIRST_VALUE_REWRITE = "FIRST_VALUE_REWRITE";
 
-    // The function of HLL_UNION_AGG can't be used with a window by now.
-    public static String HLL_UNION_AGG = "HLL_UNION_AGG";
-
     public AnalyticExpr(FunctionCallExpr fnCall, List<Expr> partitionExprs,
                         List<OrderByElement> orderByElements, AnalyticWindow window) {
         Preconditions.checkNotNull(fnCall);
@@ -248,7 +246,7 @@ private static boolean isHllAggFn(Function fn) {
             return false;
         }
 
-        return fn.functionName().equalsIgnoreCase(HLL_UNION_AGG);
+        return fn.functionName().equalsIgnoreCase(FunctionSet.HLL_UNION_AGG);
     }
 
     private static boolean isNTileFn(Function fn) {

fe/fe-core/src/main/java/org/apache/doris/analysis/FunctionCallExpr.java

@@ -785,9 +785,9 @@ private void analyzeBuiltinAggFunction(Analyzer analyzer) throws AnalysisExcepti
             }
         }
 
-        if ((fnName.getFunction().equalsIgnoreCase("HLL_UNION_AGG")
-                || fnName.getFunction().equalsIgnoreCase("HLL_CARDINALITY")
-                || fnName.getFunction().equalsIgnoreCase("HLL_RAW_AGG"))
+        if ((fnName.getFunction().equalsIgnoreCase(FunctionSet.HLL_UNION_AGG)
+                || fnName.getFunction().equalsIgnoreCase(FunctionSet.HLL_CARDINALITY)
+                || fnName.getFunction().equalsIgnoreCase(FunctionSet.HLL_RAW_AGG))
                 && !arg.type.isHllType()) {
             throw new AnalysisException(
                     "HLL_UNION_AGG, HLL_RAW_AGG and HLL_CARDINALITY's params must be hll column");
@@ -799,7 +799,7 @@ private void analyzeBuiltinAggFunction(Analyzer analyzer) throws AnalysisExcepti
         } else if (fnName.getFunction().equalsIgnoreCase("DISTINCT_PC")
                 || fnName.getFunction().equalsIgnoreCase("DISTINCT_PCSA")
                 || fnName.getFunction().equalsIgnoreCase("NDV")
-                || fnName.getFunction().equalsIgnoreCase("HLL_UNION_AGG")) {
+                || fnName.getFunction().equalsIgnoreCase(FunctionSet.HLL_UNION_AGG)) {
             fnParams.setIsDistinct(false);
         }
 

fe/fe-core/src/main/java/org/apache/doris/catalog/AggregateFunction.java

@@ -47,11 +47,13 @@ public class AggregateFunction extends Function {
     private static final Logger LOG = LogManager.getLogger(AggregateFunction.class);
 
     public static ImmutableSet<String> NOT_NULLABLE_AGGREGATE_FUNCTION_NAME_SET = ImmutableSet.of("row_number", "rank",
-            "dense_rank", "multi_distinct_count", "multi_distinct_sum", "hll_union_agg", "hll_union", "bitmap_union",
-            "bitmap_intersect", "orthogonal_bitmap_intersect", "orthogonal_bitmap_intersect_count", "intersect_count",
-            "orthogonal_bitmap_union_count", FunctionSet.COUNT, "approx_count_distinct", "ndv",
-            FunctionSet.BITMAP_UNION_INT, FunctionSet.BITMAP_UNION_COUNT, "ndv_no_finalize", FunctionSet.WINDOW_FUNNEL,
-            FunctionSet.RETENTION, FunctionSet.SEQUENCE_MATCH, FunctionSet.SEQUENCE_COUNT);
+            "dense_rank", "multi_distinct_count", "multi_distinct_sum", FunctionSet.HLL_UNION_AGG,
+            FunctionSet.HLL_UNION, FunctionSet.BITMAP_UNION, FunctionSet.BITMAP_INTERSECT,
+            FunctionSet.ORTHOGONAL_BITMAP_INTERSECT, FunctionSet.ORTHOGONAL_BITMAP_INTERSECT_COUNT,
+            FunctionSet.INTERSECT_COUNT, FunctionSet.ORTHOGONAL_BITMAP_UNION_COUNT,
+            FunctionSet.COUNT, "approx_count_distinct", "ndv", FunctionSet.BITMAP_UNION_INT,
+            FunctionSet.BITMAP_UNION_COUNT, "ndv_no_finalize", FunctionSet.WINDOW_FUNNEL, FunctionSet.RETENTION,
+            FunctionSet.SEQUENCE_MATCH, FunctionSet.SEQUENCE_COUNT);
 
     public static ImmutableSet<String> ALWAYS_NULLABLE_AGGREGATE_FUNCTION_NAME_SET =
             ImmutableSet.of("stddev_samp", "variance_samp", "var_samp", "percentile_approx");

fe/fe-core/src/main/java/org/apache/doris/catalog/FunctionSet.java

@@ -653,8 +653,12 @@ public boolean isNullResultWithOneNullParamFunctions(String funcName) {
                 .put(Type.MAX_DECIMALV2_TYPE,
                         "33decimalv2_knuth_var_pop_get_valueEPN9doris_udf15FunctionContextERKNS1_9StringValE")
                 .build();
+
     public static final String HLL_HASH = "hll_hash";
     public static final String HLL_UNION = "hll_union";
+    public static final String HLL_UNION_AGG = "hll_union_agg";
+    public static final String HLL_RAW_AGG = "hll_raw_agg";
+    public static final String HLL_CARDINALITY = "hll_cardinality";
 
     private static final Map<Type, String> HLL_UPDATE_SYMBOL =
             ImmutableMap.<Type, String>builder()

fe/fe-core/src/main/java/org/apache/doris/planner/SingleNodePlanner.java

@@ -738,10 +738,15 @@ private void turnOffPreAgg(AggregateInfo aggInfo, SelectStmt selectStmt, Analyze
                             returnColumnValidate = false;
                             break;
                         }
-                    } else if (functionName.equalsIgnoreCase("HLL_UNION_AGG")) {
-                        // do nothing
-                    } else if (functionName.equalsIgnoreCase("HLL_RAW_AGG")) {
-                        // do nothing
+                    } else if (functionName.equalsIgnoreCase(FunctionSet.HLL_UNION_AGG)
+                            || functionName.equalsIgnoreCase(FunctionSet.HLL_RAW_AGG)
+                            || functionName.equalsIgnoreCase(FunctionSet.HLL_UNION)) {
+                        if (col.getAggregationType() != AggregateType.HLL_UNION) {
+                            turnOffReason =
+                                    "Aggregate Operator not match: HLL_UNION <--> " + col.getAggregationType();
+                            returnColumnValidate = false;
+                            break;
+                        }
                     } else if (functionName.equalsIgnoreCase("NDV")) {
                         if ((!col.isKey())) {
                             turnOffReason = "NDV function with non-key column: " + col.getName();

fe/fe-core/src/main/java/org/apache/doris/rewrite/mvrewrite/HLLHashToSlotRefRule.java

@@ -26,6 +26,7 @@
 import org.apache.doris.analysis.TableName;
 import org.apache.doris.catalog.AggregateType;
 import org.apache.doris.catalog.Column;
+import org.apache.doris.catalog.FunctionSet;
 import org.apache.doris.catalog.OlapTable;
 import org.apache.doris.catalog.TableIf;
 import org.apache.doris.common.AnalysisException;
@@ -57,16 +58,16 @@ public Expr apply(Expr expr, Analyzer analyzer, ExprRewriter.ClauseType clauseTy
         }
         FunctionCallExpr fnExpr = (FunctionCallExpr) expr;
         String fnNameString = fnExpr.getFnName().getFunction();
-        if (!fnNameString.equalsIgnoreCase("hll_union")
-                && !fnNameString.equalsIgnoreCase("hll_raw_agg")
-                && !fnNameString.equalsIgnoreCase("hll_union_agg")) {
+        if (!fnNameString.equalsIgnoreCase(FunctionSet.HLL_UNION)
+                && !fnNameString.equalsIgnoreCase(FunctionSet.HLL_RAW_AGG)
+                && !fnNameString.equalsIgnoreCase(FunctionSet.HLL_UNION_AGG)) {
             return expr;
         }
         if (!(fnExpr.getChild(0) instanceof FunctionCallExpr)) {
             return expr;
         }
         FunctionCallExpr child0FnExpr = (FunctionCallExpr) fnExpr.getChild(0);
-        if (!child0FnExpr.getFnName().getFunction().equalsIgnoreCase("hll_hash")) {
+        if (!child0FnExpr.getFnName().getFunction().equalsIgnoreCase(FunctionSet.HLL_HASH)) {
             return expr;
         }
         if (child0FnExpr.getChild(0) instanceof SlotRef) {

fe/fe-core/src/main/java/org/apache/doris/rewrite/mvrewrite/NDVToHll.java

@@ -25,6 +25,7 @@
 import org.apache.doris.analysis.TableName;
 import org.apache.doris.catalog.AggregateType;
 import org.apache.doris.catalog.Column;
+import org.apache.doris.catalog.FunctionSet;
 import org.apache.doris.catalog.OlapTable;
 import org.apache.doris.catalog.TableIf;
 import org.apache.doris.common.AnalysisException;
@@ -93,7 +94,7 @@ private Expr rewriteExpr(SlotRef queryColumnSlotRef, Column mvColumn, Analyzer a
         SlotRef mvSlotRef = new SlotRef(tableName, mvColumn.getName());
         List<Expr> newFnParams = Lists.newArrayList();
         newFnParams.add(mvSlotRef);
-        FunctionCallExpr result = new FunctionCallExpr("hll_union_agg", newFnParams);
+        FunctionCallExpr result = new FunctionCallExpr(FunctionSet.HLL_UNION_AGG, newFnParams);
         result.analyzeNoThrow(analyzer);
         return result;
     }

fe/fe-core/src/test/java/org/apache/doris/planner/QueryPlanTest.java

@@ -2196,4 +2196,27 @@ public void testPreaggregationOfOrthogonalBitmapUDAF() throws Exception {
         String explainString2 = getSQLPlanOrErrorMsg(queryTableStr);
         Assert.assertTrue(explainString2.contains("PREAGGREGATION: ON"));
     }
+
+    @Test
+    public void testPreaggregationOfHllUnion() throws Exception {
+        connectContext.setDatabase("default_cluster:test");
+        createTable("create table test.test_hll(\n"
+                + "    dt date,\n"
+                + "    id int,\n"
+                + "    name char(10),\n"
+                + "    province char(10),\n"
+                + "    os char(10),\n"
+                + "    pv hll hll_union\n"
+                + ")\n"
+                + "Aggregate KEY (dt,id,name,province,os)\n"
+                + "distributed by hash(id) buckets 10\n"
+                + "PROPERTIES(\n"
+                + "    \"replication_num\" = \"1\",\n"
+                + "    \"in_memory\"=\"false\"\n"
+                + ");");
+
+        String queryBaseTableStr = "explain select dt, hll_union(pv) from test.test_hll group by dt";
+        String explainString = getSQLPlanOrErrorMsg(queryBaseTableStr);
+        Assert.assertTrue(explainString.contains("PREAGGREGATION: ON"));
+    }
 }